import logging import os from collections import namedtuple import ldap from ldap.controls import SimplePagedResultsControl from ldap.filter import escape_filter_chars, filter_format from data.model import InvalidRobotException from data.model.user import ( find_user_by_email, get_nonrobot_user, get_username, lookup_robot, ) from data.users.federated import FederatedUsers, UserInformation from util.itertoolrecipes import take logger = logging.getLogger(__name__) _DEFAULT_NETWORK_TIMEOUT = 10.0 # seconds _DEFAULT_TIMEOUT = 10.0 # seconds _DEFAULT_PAGE_SIZE = 1000 # setting config LDAP_FOLLOW_REFERRALS: 0 to disable referral lookups _DEFAULT_REFERRALS = True _DEFAULT_KEEPALIVE_IDLE = 10 _DEFAULT_KEEPALIVE_INTERVAL = 5 _DEFAULT_KEEPALIVE_PROBES = 3 class LDAPConnectionBuilder(object): def __init__( self, ldap_uri, user_dn, user_pw, allow_tls_fallback=False, timeout=None, network_timeout=None, referrals=_DEFAULT_REFERRALS, ): self._ldap_uri = ldap_uri self._user_dn = user_dn self._user_pw = user_pw self._allow_tls_fallback = allow_tls_fallback self._timeout = timeout self._network_timeout = network_timeout self._referrals = int(referrals) def get_connection(self): return LDAPConnection( self._ldap_uri, self._user_dn, self._user_pw, self._allow_tls_fallback, self._timeout, self._network_timeout, self._referrals, ) class LDAPConnection(object): def __init__( self, ldap_uri, user_dn, user_pw, allow_tls_fallback=False, timeout=None, network_timeout=None, referrals=_DEFAULT_REFERRALS, ): self._ldap_uri = ldap_uri self._user_dn = user_dn self._user_pw = user_pw self._allow_tls_fallback = allow_tls_fallback self._timeout = timeout self._network_timeout = network_timeout self._referrals = int(referrals) self._conn = None def __enter__(self): trace_level = 2 if os.environ.get("USERS_DEBUG") == "1" else 0 self._conn = ldap.initialize(self._ldap_uri, trace_level=trace_level) self._conn.set_option(ldap.OPT_REFERRALS, self._referrals) self._conn.set_option( ldap.OPT_NETWORK_TIMEOUT, self._network_timeout or _DEFAULT_NETWORK_TIMEOUT ) self._conn.set_option(ldap.OPT_TIMEOUT, self._timeout or _DEFAULT_TIMEOUT) self._conn.set_option(ldap.OPT_X_KEEPALIVE_IDLE, _DEFAULT_KEEPALIVE_IDLE) self._conn.set_option(ldap.OPT_X_KEEPALIVE_INTERVAL, _DEFAULT_KEEPALIVE_INTERVAL) self._conn.set_option(ldap.OPT_X_KEEPALIVE_PROBES, _DEFAULT_KEEPALIVE_PROBES) self._conn.set_option(ldap.OPT_RESTART, ldap.OPT_ON) if self._allow_tls_fallback: logger.debug("TLS Fallback enabled in LDAP") self._conn.set_option(ldap.OPT_X_TLS_TRY, 1) # Must come _after_ all other TLS options self._conn.set_option(ldap.OPT_X_TLS_NEWCTX, ldap.OPT_OFF) self._conn.simple_bind_s(self._user_dn, self._user_pw) return self._conn def __exit__(self, exc_type, value, tb): self._conn.unbind_s() class LDAPUsers(FederatedUsers): _LDAPResult = namedtuple("_LDAPResult", ["dn", "attrs"]) def __init__( self, ldap_uri, base_dn, admin_dn, admin_passwd, user_rdn, uid_attr, email_attr, memberof_attr="memberOf", allow_tls_fallback=False, secondary_user_rdns=None, requires_email=True, timeout=None, network_timeout=None, force_no_pagination=False, ldap_user_filter=None, ldap_superuser_filter=None, ldap_global_readonly_superuser_filter=None, ldap_restricted_user_filter=None, ldap_referrals=_DEFAULT_REFERRALS, ): super(LDAPUsers, self).__init__("ldap", requires_email) self._ldap = LDAPConnectionBuilder( ldap_uri, admin_dn, admin_passwd, allow_tls_fallback, timeout, network_timeout, referrals=ldap_referrals, ) self._ldap_uri = ldap_uri self._uid_attr = uid_attr self._email_attr = email_attr self._memberof_attr = memberof_attr self._allow_tls_fallback = allow_tls_fallback self._requires_email = requires_email self._force_no_pagination = force_no_pagination self._ldap_user_filter = ldap_user_filter self._ldap_superuser_filter = ldap_superuser_filter self._ldap_global_readonly_superuser_filter = ldap_global_readonly_superuser_filter self._ldap_restricted_user_filter = ldap_restricted_user_filter self._ldap_referrals = int(ldap_referrals) # Note: user_rdn is a list of RDN pieces (for historical reasons), and secondary_user_rds # is a list of RDN strings. relative_user_dns = [",".join(user_rdn)] + (secondary_user_rdns or []) def get_full_rdn(relative_dn): prefix = relative_dn.split(",") if relative_dn else [] return ",".join(prefix + base_dn) # Create the set of full DN paths. self._user_dns = [get_full_rdn(relative_dn) for relative_dn in relative_user_dns] self._base_dn = ",".join(base_dn) def _get_ldap_referral_dn(self, referral_exception): logger.debug("Got referral: %s", referral_exception.args[0]) if not referral_exception.args[0] or not referral_exception.args[0].get("info"): logger.debug("LDAP referral missing info block") return None referral_info = referral_exception.args[0]["info"] if not referral_info.startswith("Referral:\n"): logger.debug("LDAP referral missing Referral header") return None referral_uri = referral_info[len("Referral:\n") :] if not referral_uri.startswith("ldap:///"): logger.debug("LDAP referral URI does not start with ldap:///") return None referral_dn = referral_uri[len("ldap:///") :] return referral_dn def _add_filter(self, query, user_filter): if not query.startswith("(") or not query.endswith(")"): query = "(%s)" % query if not user_filter.startswith("(") or not user_filter.endswith(")"): user_filter = "(%s)" % user_filter assert user_filter.startswith("(") and user_filter.endswith(")") assert query.startswith("(") and query.endswith(")") return "(&{0}{1})".format(query, user_filter) def _add_user_filter(self, query): if not self._ldap_user_filter: return query user_filter = self._ldap_user_filter return self._add_filter(query, self._ldap_user_filter) def _add_superuser_filter(self, query): assert self._ldap_superuser_filter return self._add_filter(query, self._ldap_superuser_filter) def _add_global_readonly_superuser_filter(self, query): assert self._ldap_global_readonly_superuser_filter return self._add_filter(query, self._ldap_global_readonly_superuser_filter) def _add_restricted_user_filter(self, query): assert self._ldap_restricted_user_filter return self._add_filter(query, self._ldap_restricted_user_filter) def _ldap_user_search_with_rdn( self, conn, username_or_email, user_search_dn, suffix="", filter_superusers=False, filter_restricted_users=False, filter_global_readonly_superusers=False, ): query = "(|({0}={2}{3})({1}={2}{3}))".format( self._uid_attr, self._email_attr, escape_filter_chars(username_or_email), suffix ) query = self._add_user_filter(query) if filter_restricted_users: if not self._ldap_restricted_user_filter: return (None, "Username not found") query = self._add_restricted_user_filter(query) elif filter_superusers: if not self._ldap_superuser_filter: return (None, "Superuser username not found") query = self._add_superuser_filter(query) elif filter_global_readonly_superusers: if not self._ldap_global_readonly_superuser_filter: return (None, "Global readonly superuser username not found") query = self._add_global_readonly_superuser_filter(query) logger.debug("Conducting user search: %s under %s", query, user_search_dn) try: return (conn.search_s(user_search_dn, ldap.SCOPE_SUBTREE, query), None) except ldap.REFERRAL as re: referral_dn = self._get_ldap_referral_dn(re) if not referral_dn: return (None, "Failed to follow referral when looking up username") try: subquery = "(%s=%s)" % (self._uid_attr, username_or_email) subquery = self._add_user_filter(subquery) return (conn.search_s(referral_dn, ldap.SCOPE_BASE, subquery), None) except ldap.LDAPError: logger.debug("LDAP referral search exception") return (None, "Username not found") except ldap.LDAPError: logger.debug("LDAP search exception") return (None, "Username not found") def _ldap_user_search( self, username_or_email, limit=20, suffix="", filter_superusers=False, filter_restricted_users=False, filter_global_readonly_superusers=False, ): if not username_or_email: return (None, "Empty username/email") try: lookup_robot(username_or_email) return (None, f"LDAP lookup for robots disabled {username_or_email}") except InvalidRobotException: # continue with LDAP lookup pass try: with self._ldap.get_connection() as conn: logger.debug("Incoming username or email param: %s", username_or_email.__repr__()) pairs = [] err_msg = None for user_search_dn in self._user_dns: (pairs, err_msg) = self._ldap_user_search_with_rdn( conn, username_or_email, user_search_dn, suffix=suffix, filter_superusers=filter_superusers, filter_restricted_users=filter_restricted_users, filter_global_readonly_superusers=filter_global_readonly_superusers, ) if pairs is not None and len(pairs) > 0: break if err_msg is not None: return (None, err_msg) dn_lst = [pair[0] for pair in pairs] logger.debug("Found matching DNs: %s" % dn_lst) results = [LDAPUsers._LDAPResult(*pair) for pair in take(limit, pairs)] # Filter out pairs without DNs. Some LDAP impls will return such pairs. with_dns = [result for result in results if result.dn] return (with_dns, None) except ldap.INVALID_CREDENTIALS: return (None, "LDAP Admin dn or password is invalid") def _ldap_single_user_search( self, username_or_email, filter_superusers=False, filter_restricted_users=False, filter_global_readonly_superusers=False, ): try: lookup_robot(username_or_email) return (None, f"LDAP lookup for robots disabled {username_or_email}") except InvalidRobotException: # continue with LDAP lookup pass with_dns, err_msg = self._ldap_user_search( username_or_email, filter_superusers=filter_superusers, filter_restricted_users=filter_restricted_users, filter_global_readonly_superusers=filter_global_readonly_superusers, ) if err_msg is not None: return (None, err_msg) # Make sure we have at least one result. if len(with_dns) < 1: return (None, "Invalid username or password.") # If we have found a single pair, then return it. if len(with_dns) == 1: return (with_dns[0], None) # Otherwise, there are multiple pairs with DNs, so find the one with the mail # attribute (if any). with_mail = [result for result in with_dns if result.attrs.get(self._email_attr)] return (with_mail[0] if with_mail else with_dns[0], None) def _build_user_information(self, response): if not response.get(self._uid_attr): return (None, 'Missing uid field "%s" in user record' % self._uid_attr) if self._requires_email and not response.get(self._email_attr): return (None, 'Missing mail field "%s" in user record' % self._email_attr) try: username = response[self._uid_attr][0].decode("utf-8") except (UnicodeDecodeError, AttributeError): username = response[self._uid_attr][0] try: email = response.get(self._email_attr, [None])[0].decode("utf-8") except (UnicodeDecodeError, AttributeError): email = response.get(self._email_attr, [None])[0] return (UserInformation(username=username, email=email, id=username), None) def ping(self): try: with self._ldap.get_connection(): pass except ldap.INVALID_CREDENTIALS: return (False, "LDAP Admin dn or password is invalid") except ldap.LDAPError as lde: logger.exception("Exception when trying to health check LDAP") return (False, str(lde)) return (True, None) def at_least_one_user_exists(self, filter_superusers=False, filter_restricted_users=False): logger.debug("Checking if any users exist in LDAP") has_pagination = not self._force_no_pagination try: with self._ldap.get_connection() as conn: for user_search_dn in self._user_dns: search_flt = "(objectClass=*)" search_flt = self._add_user_filter(search_flt) if filter_restricted_users: if self._ldap_restricted_user_filter: search_flt = self._add_restricted_user_filter(search_flt) else: return (False, "Restricted user filter not set") elif filter_superusers: if self._ldap_superuser_filter: search_flt = self._add_superuser_filter(search_flt) else: return (False, "Superuser filter not set") lc = ldap.controls.libldap.SimplePagedResultsControl( criticality=True, size=1, cookie="" ) try: if has_pagination: msgid = conn.search_ext( user_search_dn, ldap.SCOPE_SUBTREE, search_flt, serverctrls=[lc] ) _, rdata, _, _serverctrls = conn.result3(msgid) else: msgid = conn.search(user_search_dn, ldap.SCOPE_SUBTREE, search_flt) _, rdata = conn.result(msgid) for _entry in rdata: # Handles both lists and iterators. return (True, None) except ldap.LDAPError as lde: return (False, str(lde) or "Could not find DN %s" % user_search_dn) return (False, None) except ldap.INVALID_CREDENTIALS: return (None, "LDAP Admin dn or password is invalid") def get_user(self, username_or_email): """ Looks up a username or email in LDAP. """ logger.debug("Looking up LDAP username or email %s", username_or_email) (found_user, err_msg) = self._ldap_single_user_search(username_or_email) if err_msg is not None: return (None, err_msg) logger.debug("Found user for LDAP username or email %s", username_or_email) _, found_response = found_user return self._build_user_information(found_response) def query_users(self, query, limit=20): """ Queries LDAP for matching users. """ if not query: return (None, self.federated_service, "Empty query") logger.debug("Got query %s with limit %s", query, limit) (results, err_msg) = self._ldap_user_search(query, limit=limit, suffix="*") if err_msg is not None: return (None, self.federated_service, err_msg) final_results = [] for result in results[0:limit]: credentials, err_msg = self._build_user_information(result.attrs) if err_msg is not None: continue final_results.append(credentials) logger.debug("For query %s found results %s", query, final_results) return (final_results, self.federated_service, None) def verify_credentials(self, username_or_email, password): """ Verify the credentials with LDAP. """ # Make sure that even if the server supports anonymous binds, we don't allow it if not password: return (None, "Anonymous binding not allowed.") (found_user, err_msg) = self._ldap_single_user_search(username_or_email) if found_user is None: return (None, err_msg) found_dn, found_response = found_user logger.debug("Found user for LDAP username %s; validating password", username_or_email) logger.debug("DN %s found: %s", found_dn, found_response) # First validate the password by binding as the user try: with LDAPConnection(self._ldap_uri, found_dn, password, self._allow_tls_fallback): pass except ldap.REFERRAL as re: referral_dn = self._get_ldap_referral_dn(re) if not referral_dn: return (None, "Invalid username or password.") try: with LDAPConnection( self._ldap_uri, referral_dn, password, self._allow_tls_fallback ): pass except ldap.INVALID_CREDENTIALS: logger.debug("Invalid LDAP credentials") return (None, "Invalid username or password.") except ldap.INVALID_CREDENTIALS: logger.debug("Invalid LDAP credentials") return (None, "Invalid username or password.") return self._build_user_information(found_response) def service_metadata(self): return { "base_dn": self._base_dn, } def check_group_lookup_args(self, group_lookup_args, disable_pagination=False): if not group_lookup_args.get("group_dn"): return (False, "Missing group_dn") (it, err) = self.iterate_group_members( group_lookup_args, page_size=1, disable_pagination=disable_pagination ) if err is not None: return (False, err) if not next(it, False): return (False, "Group does not exist or is empty") return (True, None) def iterate_group_members(self, group_lookup_args, page_size=None, disable_pagination=False): try: with self._ldap.get_connection(): pass except ldap.INVALID_CREDENTIALS: return (None, "LDAP Admin dn or password is invalid") group_dn = group_lookup_args["group_dn"] memberof_attr = self._memberof_attr page_size = page_size or _DEFAULT_PAGE_SIZE return (self._iterate_members(group_dn, memberof_attr, page_size, disable_pagination), None) def is_superuser(self, username_or_email: str) -> bool: if not username_or_email: return False try: lookup_robot(username_or_email) return False # Robots are not in LDAP so return False as not being a superuser except InvalidRobotException: # continue with LDAP lookup pass logger.debug("Looking up LDAP superuser username or email %s", username_or_email) (found_user, err_msg) = self._ldap_single_user_search( username_or_email, filter_superusers=True ) if found_user is None: logger.debug("LDAP superuser %s not found: %s", username_or_email, err_msg) return False logger.debug("Found superuser for LDAP username or email %s", username_or_email) return True def has_superusers(self) -> bool: has_superusers, _ = self.at_least_one_user_exists(filter_superusers=True) return has_superusers def is_global_readonly_superuser(self, username_or_email: str) -> bool: if not username_or_email: return False try: lookup_robot(username_or_email) return False # Robots are not in LDAP so return False as not being a superuser except InvalidRobotException: # continue with LDAP lookup pass logger.debug( "Looking up LDAP global readonly superuser username or email %s", username_or_email ) (found_user, err_msg) = self._ldap_single_user_search( username_or_email, filter_global_readonly_superusers=True ) if found_user is None: logger.debug( "LDAP global readonly superuser %s not found: %s", username_or_email, err_msg ) return False logger.debug( "Found global readonly superuser for LDAP username or email %s", username_or_email ) return True def is_restricted_user(self, username_or_email: str) -> bool: if not username_or_email: return False if self._ldap_restricted_user_filter is None: return True try: lookup_robot(username_or_email) return False # Robots are not in LDAP so return False as not being a superuser except InvalidRobotException: # continue with LDAP lookup pass logger.debug("Looking up LDAP restricted user username or email %s", username_or_email) (found_user, err_msg) = self._ldap_single_user_search( username_or_email, filter_restricted_users=True ) if found_user is None: logger.debug("LDAP user %s not found: %s", username_or_email, err_msg) return False logger.debug("Found restricted user for LDAP username or email %s", username_or_email) return True def has_restricted_users(self) -> bool: if self._ldap_restricted_user_filter is None and self.at_least_one_user_exists(): return True has_restricted_users, _ = self.at_least_one_user_exists(filter_restricted_users=True) return has_restricted_users def _iterate_members(self, group_dn, memberof_attr, page_size, disable_pagination): has_pagination = not (self._force_no_pagination or disable_pagination) with self._ldap.get_connection() as conn: search_flt = filter_format("(%s=%s,%s)", (memberof_attr, group_dn, self._base_dn)) search_flt = self._add_user_filter(search_flt) attributes = [self._uid_attr, self._email_attr] for user_search_dn in self._user_dns: lc = ldap.controls.libldap.SimplePagedResultsControl( criticality=True, size=page_size, cookie="" ) # Conduct the initial search for users that are a member of the group. logger.debug( "Conducting LDAP search of DN: %s and filter %s", user_search_dn, search_flt ) try: if has_pagination: msgid = conn.search_ext( user_search_dn, ldap.SCOPE_SUBTREE, search_flt, serverctrls=[lc], attrlist=attributes, ) else: msgid = conn.search( user_search_dn, ldap.SCOPE_SUBTREE, search_flt, attrlist=attributes ) except ldap.LDAPError as lde: logger.exception( "Got error when trying to search %s with filter %s: %s", user_search_dn, search_flt, str(lde), ) break while True: found_results = 0 try: if has_pagination: _, rdata, _, serverctrls = conn.result3(msgid) else: _, rdata = conn.result(msgid) # Yield any users found. for userdata in rdata: found_results = found_results + 1 if isinstance(userdata[1], list): # we do not follow referrals here continue yield self._build_user_information(userdata[1]) logger.debug( "Found %s users in group %s; %s", found_results, user_search_dn, search_flt, ) except ldap.NO_SUCH_OBJECT as nsoe: logger.debug( "NSO when trying to lookup results of search %s with filter %s: %s", user_search_dn, search_flt, str(nsoe), ) except ldap.LDAPError as lde: logger.exception( "Error when trying to lookup results of search %s with filter %s: %s", user_search_dn, search_flt, str(lde), ) break # If no additional results, nothing more to do. if not found_results: break # If pagination is disabled, nothing more to do. if not has_pagination: logger.debug("Pagination is disabled, no further queries") break # Filter down the controls with which the server responded, looking for the paging # control type. If not found, then the server does not support pagination and we already # got all of the results. pctrls = [ control for control in serverctrls if control.controlType == ldap.controls.SimplePagedResultsControl.controlType ] if pctrls: # Server supports pagination. Update the cookie so the next search finds the next page, # then conduct the next search. cookie = lc.cookie = pctrls[0].cookie if cookie: logger.debug( "Pagination is supported for this LDAP server; trying next page" ) msgid = conn.search_ext( user_search_dn, ldap.SCOPE_SUBTREE, search_flt, serverctrls=[lc], attrlist=attributes, ) continue else: # No additional results. logger.debug( "Pagination is supported for this LDAP server but on last page" ) break else: # Pagination is not supported. logger.debug("Pagination is not supported for this LDAP server") break