拿来即用的 Python LDAP 实现类

Posted 2022-11-28 何小有

一个拿来即用的 Python LDAP 实现类，相关配置通过读取环境变量获取：

• LDAP_SERVER_HOST = [LDAP服务器IP: 127.0.0.1]
• LDAP_SEARCH_BASE = [LDAP搜索配置: OU=OU,DC=DC,DC=LOCAL]
• LDAP_USERNAME = [LDAP连接账户: [e-mail]]
• LDAP_PASSWORD = [LDAP连接密码]

快速调用方法如下：

def ldap_user_auth(username, password):
    v = LDAPVerify()
    state, result = v.main(username, password)

具体的 LDAP 通用类方法代码如下：

import logging
from ldap3 import Connection, SUBTREE

logger = logging.getLogger(__name__)

class LDAPVerify:
    def __init__(self):
        self.ldap_host = get_env('LDAP_SERVER_HOST', '127.0.0.1')
        self.ldap_search_base = get_env('LDAP_SEARCH_BASE', 'OU=OU,DC=DC,DC=DC')
        self.ldap_user = get_env('LDAP_USERNAME', '')
        self.ldap_pwd = get_env('LDAP_PASSWORD', '')
        self.response = None

    def __connect_ldap(self):
        # 与 LDAP 建立连接
        try:
            self.ldap_conn = Connection(
                self.ldap_host,
                self.ldap_user,
                self.ldap_pwd,
                auto_bind=True,
                raise_exceptions=False
            )
            return True
        except Exception as e:
            logger.error('LDAP Connection: ' + str(e))
            return False

    def __search_user(self, search_name):
        # 查询 LDAP 用户信息
        try:
            search_result = self.ldap_conn.search(
                search_base=self.ldap_search_base,
                search_filter='(sAMAccountName=)'.format(search_name),
                search_scope=SUBTREE,
                paged_size=5,
                attributes=['cn', 'mail', 'sAMAccountName', 'givenName']
            )
            if not search_result:
                return True
            self.response = self.ldap_conn.response[0]
            # 字符串, CN=员工姓名-员工编号,OU=直属组织,OU=上层组织,OU=上上层组织,,OU=企业名称,OU=行政组织,OU=OU,DC=DC,DC=DC
            self.dn = self.response.get('dn', '')
            # 字典, 'cn': '员工姓名-员工编号', 'givenName': '员工名称', 'sAMAccountName': '员工账户名', 'mail': '员工邮箱'
            self.attributes = self.response.get('attributes', )
            return True
        except Exception as e:
            logger.error('LDAP Search: ' + str(e))
            return False

    def __check_user_pwd(self, password):
        # 验证 LDAP 用户密码
        try:
            ldap_conn_check = Connection(
                self.ldap_host,
                user=self.dn,
                password=password,
                check_names=True,
                lazy=False,
                raise_exceptions=False
            )
            ldap_conn_check.bind()
            self.check_description = ldap_conn_check.result['description']
            return True
        except Exception as e:
            logger.error('LDAP Check: ' + str(e))
            return False

    def main(self, search_name, password):
        # LDAP 验证主方法
        if not self.__connect_ldap():
            return False, 'message': 'Failed to establish connection with LDAP'
        if not self.__search_user(search_name):
            return False, 'message': 'Failed to query LDAP user information'
        if not self.response:
            return False, 'message': 'No LDAP user information found'
        if not self.__check_user_pwd(password):
            return False, 'message': 'Unable to verify LDAP user password'
        if not self.check_description == 'success':
            return False, 'message': 'User name and password do not match'
        return True, 'dn': self.dn, 'attributes': self.attributes