def _bind()

in lemur/auth/ldap.py [0:0]
54 lines of code
13 McCabe index (conditional complexity)

    def _bind(self):
        """
        authenticate an ldap user.
        list groups for a user.
        raise an exception on error.
        """
        if "@" not in self.ldap_principal:
            self.ldap_principal = "%s@%s" % (
                self.ldap_principal,
                self.ldap_email_domain,
            )
        ldap_filter = "userPrincipalName=%s" % self.ldap_principal

        # query ldap for auth
        try:
            # build a client
            if not self.ldap_client:
                self.ldap_client = ldap.initialize(self.ldap_server)
            # perform a synchronous bind
            self.ldap_client.set_option(ldap.OPT_REFERRALS, 0)
            if self.ldap_use_tls:
                ldap.set_option(ldap.OPT_X_TLS_REQUIRE_CERT, ldap.OPT_X_TLS_NEVER)
                self.ldap_client.set_option(ldap.OPT_PROTOCOL_VERSION, 3)
                self.ldap_client.set_option(ldap.OPT_X_TLS, ldap.OPT_X_TLS_DEMAND)
                self.ldap_client.set_option(ldap.OPT_X_TLS_DEMAND, True)
                self.ldap_client.set_option(ldap.OPT_DEBUG_LEVEL, 255)
            if self.ldap_cacert_file:
                self.ldap_client.set_option(
                    ldap.OPT_X_TLS_CACERTFILE, self.ldap_cacert_file
                )
            self.ldap_client.simple_bind_s(self.ldap_principal, self.ldap_password)
        except ldap.INVALID_CREDENTIALS:
            self.ldap_client.unbind()
            raise Exception("The supplied ldap credentials are invalid")
        except ldap.SERVER_DOWN:
            raise Exception("ldap server unavailable")
        except ldap.LDAPError as e:
            raise Exception("ldap error: {0}".format(e))

        if self.ldap_is_active_directory:
            # Lookup user DN, needed to search for group membership
            userdn = self.ldap_client.search_s(
                self.ldap_base_dn,
                ldap.SCOPE_SUBTREE,
                ldap_filter,
                ["distinguishedName"],
            )[0][1]["distinguishedName"][0]
            userdn = userdn.decode("utf-8")
            # Search all groups that have the userDN as a member
            groupfilter = "(&(objectclass=group)(member:1.2.840.113556.1.4.1941:={0}))".format(
                userdn
            )
            lgroups = self.ldap_client.search_s(
                self.ldap_base_dn, ldap.SCOPE_SUBTREE, groupfilter, ["cn"]
            )

            # Create a list of group CN's from the result
            self.ldap_groups = []
            for group in lgroups:
                (dn, values) = group
                if type(values) == dict:
                    self.ldap_groups.append(values["cn"][0].decode("utf-8"))
        else:
            lgroups = self.ldap_client.search_s(
                self.ldap_base_dn, ldap.SCOPE_SUBTREE, ldap_filter, self.ldap_attrs
            )[0][1]["memberOf"]
            # lgroups is a list of utf-8 encoded strings
            # convert to a single string of groups to allow matching
            self.ldap_groups = b"".join(lgroups).decode("ascii")

        self.ldap_client.unbind()