kallithea Changeset - a00a58323729

Changeset - a00a58323729

Parent rev.

Child rev.

[Not reviewed]

default

0 1 0

Mads Kiilerich - 9 years ago 2016-10-24 15:18:51
madski@unity3d.com

auth: refactor LDAP authentication - make it more clear in program flow when authentication is accepted

1 file changed with 7 insertions and 6 deletions:

kallithea/lib/auth_modules/auth_ldap.py

0 comments (0 inline, 0 general)

kallithea/lib/auth_modules/auth_ldap.py

➞

Show inline comments

@@ @@ -55,213 +55,214 @@ class AuthLdap(object): @@
         if ldap is None:
             raise LdapImportError
         self.ldap_version = ldap_version
         ldap_server_type = 'ldap'
         self.TLS_KIND = tls_kind
         if self.TLS_KIND == 'LDAPS':
             port = port or 689
             ldap_server_type = ldap_server_type + 's'
         OPT_X_TLS_DEMAND = 2
         self.TLS_REQCERT = getattr(ldap, 'OPT_X_TLS_%s' % tls_reqcert,
                                    OPT_X_TLS_DEMAND)
         # split server into list
         self.LDAP_SERVER_ADDRESS = server.split(',')
         self.LDAP_SERVER_PORT = port
         # USE FOR READ ONLY BIND TO LDAP SERVER
         self.LDAP_BIND_DN = safe_str(bind_dn)
         self.LDAP_BIND_PASS = safe_str(bind_pass)
         _LDAP_SERVERS = []
         for host in self.LDAP_SERVER_ADDRESS:
             _LDAP_SERVERS.append("%s://%s:%s" % (ldap_server_type,
                                                      host.replace(' ', ''),
                                                      self.LDAP_SERVER_PORT))
         self.LDAP_SERVER = str(', '.join(s for s in _LDAP_SERVERS))
         self.BASE_DN = safe_str(base_dn)
         self.LDAP_FILTER = safe_str(ldap_filter)
         self.SEARCH_SCOPE = getattr(ldap, 'SCOPE_%s' % search_scope)
         self.attr_login = attr_login
     def authenticate_ldap(self, username, password):
         """
         Authenticate a user via LDAP and return his/her LDAP properties.
         Raises AuthenticationError if the credentials are rejected, or
         EnvironmentError if the LDAP server can't be reached.
         :param username: username
         :param password: password
         """
         from kallithea.lib.helpers import chop_at
         uid = chop_at(username, "@%s" % self.LDAP_SERVER_ADDRESS)
         if not password:
             log.debug("Attempt to authenticate LDAP user "
                       "with blank password rejected.")
             raise LdapPasswordError()
         if "," in username:
             raise LdapUsernameError("invalid character in username: ,")
         try:
             if hasattr(ldap, 'OPT_X_TLS_CACERTDIR'):
                 ldap.set_option(ldap.OPT_X_TLS_CACERTDIR,
                                 '/etc/openldap/cacerts')
             ldap.set_option(ldap.OPT_REFERRALS, ldap.OPT_OFF)
             ldap.set_option(ldap.OPT_RESTART, ldap.OPT_ON)
             ldap.set_option(ldap.OPT_TIMEOUT, 20)
             ldap.set_option(ldap.OPT_NETWORK_TIMEOUT, 10)
             ldap.set_option(ldap.OPT_TIMELIMIT, 15)
             if self.TLS_KIND != 'PLAIN':
                 ldap.set_option(ldap.OPT_X_TLS_REQUIRE_CERT, self.TLS_REQCERT)
             server = ldap.initialize(self.LDAP_SERVER)
             if self.ldap_version == 2:
                 server.protocol = ldap.VERSION2
             else:
                 server.protocol = ldap.VERSION3
             if self.TLS_KIND == 'START_TLS':
                 server.start_tls_s()
             if self.LDAP_BIND_DN and self.LDAP_BIND_PASS:
                 log.debug('Trying simple_bind with password and given DN: %s',
                           self.LDAP_BIND_DN)
                 server.simple_bind_s(self.LDAP_BIND_DN, self.LDAP_BIND_PASS)
             filter_ = '(&%s(%s=%s))' % (self.LDAP_FILTER, self.attr_login,
                                         username)
             log.debug("Authenticating %r filter %s at %s", self.BASE_DN,
                       filter_, self.LDAP_SERVER)
             lobjects = server.search_ext_s(self.BASE_DN, self.SEARCH_SCOPE,
                                            filter_)
             if not lobjects:
                 raise ldap.NO_SUCH_OBJECT()
             for (dn, _attrs) in lobjects:
                 if dn is None:
                     continue
                 try:
                     log.debug('Trying simple bind with %s', dn)
                     server.simple_bind_s(dn, safe_str(password))
                     attrs = server.search_ext_s(dn, ldap.SCOPE_BASE,
                                                 '(objectClass=*)')[0][1]
                     break
                     results = server.search_ext_s(dn, ldap.SCOPE_BASE,
                                                   '(objectClass=*)')
                     if len(results) == 1:
                         dn_, attrs = results[0]
                         assert dn_ == dn
                         return dn, attrs
                 except ldap.INVALID_CREDENTIALS:
                     log.debug("LDAP rejected password for user '%s' (%s): %s",
                               uid, username, dn)
                     continue # accept authentication as another ldap user with same username
             else:
                 log.debug("No matching LDAP objects for authentication "
                           "of '%s' (%s)", uid, username)
                 raise LdapPasswordError()
         except ldap.NO_SUCH_OBJECT:
             log.debug("LDAP says no such user '%s' (%s)", uid, username)
             raise LdapUsernameError()
         except ldap.SERVER_DOWN:
             raise LdapConnectionError("LDAP can't access authentication server")
         return dn, attrs
 class KallitheaAuthPlugin(auth_modules.KallitheaExternalAuthPlugin):
     def __init__(self):
         self._logger = logging.getLogger(__name__)
         self._tls_kind_values = ["PLAIN", "LDAPS", "START_TLS"]
         self._tls_reqcert_values = ["NEVER", "ALLOW", "TRY", "DEMAND", "HARD"]
         self._search_scopes = ["BASE", "ONELEVEL", "SUBTREE"]
     @hybrid_property
     def name(self):
         return "ldap"
     def settings(self):
         settings = [
+            {
                 "name": "host",
                 "validator": self.validators.UnicodeString(strip=True),
                 "type": "string",
                 "description": "Host of the LDAP Server",
                 "formname": "LDAP Host"
             },
+            {
                 "name": "port",
                 "validator": self.validators.Number(strip=True, not_empty=True),
                 "type": "string",
                 "description": "Port that the LDAP server is listening on",
                 "default": 389,
                 "formname": "Port"
             },
+            {
                 "name": "dn_user",
                 "validator": self.validators.UnicodeString(strip=True),
                 "type": "string",
                 "description": "User to connect to LDAP",
                 "formname": "Account"
             },
+            {
                 "name": "dn_pass",
                 "validator": self.validators.UnicodeString(strip=True),
                 "type": "password",
                 "description": "Password to connect to LDAP",
                 "formname": "Password"
             },
+            {
                 "name": "tls_kind",
                 "validator": self.validators.OneOf(self._tls_kind_values),
                 "type": "select",
                 "values": self._tls_kind_values,
                 "description": "TLS Type",
                 "default": 'PLAIN',
                 "formname": "Connection Security"
             },
+            {
                 "name": "tls_reqcert",
                 "validator": self.validators.OneOf(self._tls_reqcert_values),
                 "type": "select",
                 "values": self._tls_reqcert_values,
                 "description": "Require Cert over TLS?",
                 "formname": "Certificate Checks"
             },
+            {
                 "name": "base_dn",
                 "validator": self.validators.UnicodeString(strip=True),
                 "type": "string",
                 "description": "Base DN to search (e.g., dc=mydomain,dc=com)",
                 "formname": "Base DN"
             },
+            {
                 "name": "filter",
                 "validator": self.validators.UnicodeString(strip=True),
                 "type": "string",
                 "description": "Filter to narrow results (e.g., ou=Users, etc)",
                 "formname": "LDAP Search Filter"
             },
+            {
                 "name": "search_scope",
                 "validator": self.validators.OneOf(self._search_scopes),
                 "type": "select",
                 "values": self._search_scopes,
                 "description": "How deep to search LDAP",
                 "formname": "LDAP Search Scope"
             },
+            {
                 "name": "attr_login",
                 "validator": self.validators.AttrLoginValidator(not_empty=True, strip=True),
                 "type": "string",
                 "description": "LDAP Attribute to map to user name",
                 "formname": "Login Attribute"
             },
+            {
                 "name": "attr_firstname",
                 "validator": self.validators.UnicodeString(strip=True),
                 "type": "string",
                 "description": "LDAP Attribute to map to first name",
                 "formname": "First Name Attribute"
             },

0 comments (0 inline, 0 general)