Python中ldap3模块在AD域中分页查询用户修改用户信息

╰+哭是因爲堅強的太久メ 2022-03-16 00:54 597阅读 0赞

> 最近找ldap信息，找的比较艰辛，原文也没有使用代码模式，粘过来重新共享

原文链接：[https://www.jianshu.com/p/fabb04fd93e3][https_www.jianshu.com_p_fabb04fd93e3]

# -*- coding:utf8 -*-
    
    import ldap3
    import logging
    from ldap3 import Connection,Server,ALL,SUBTREE,MODIFY_REPLACE
    from app import app
    
    
    class AdApi(object):
        server =None
        connect =None
        @staticmethod
        def init_connection():
            try:
                # AdApi.server = Server(app.config['ADSERVER'], app.config['ADSERVERPORT'], get_info=ALL) 
                AdApi.server = Server(app.config['ADSERVER'],
                                      app.config['ADSERVERPORT'],
                                      use_ssl=True)
                AdApi.connect = Connection(AdApi.server,
                                           user=app.config['ADACCOUNT'],
                                           password=app.config['ADPASSWORD'],
                                           auto_bind=True)
                AdApi.connect.start_tls()
            except Exception as e:
                logging.exception("init_connection error: %s", e)
    
        @staticmethod
        def list_ad_user(adconfig):
            """ 从AD域服务器拉取用户列表，每次1000 :param adconfig: :return: """
            if AdApi.server is None or AdApi.connect is None:
                AdApi.init_connection()
            try:
            	# attributes = ['cn', 'givenName', 'mail', 'sAMAccountName'], 自定义过滤属性
                AdApi.connect.search(adconfig,
                                     '(objectclass=person)',
                                     attributes=['cn', 'description','userAccountControl'],
                                     paged_size=1000,
                                     search_scope=SUBTREE)
                ad_users_list =list()
                ad_users_list.extend(AdApi.connect.entries)
                cookie = AdApi.connect.result['controls']['1.2.840.113556.1.4.319']['value']['cookie']
                while cookie:
                    AdApi.connect.search(adconfig,
                                         '(objectclass=person)',
                                         attributes=['cn', 'description','userAccountControl'],
                                         paged_size=1000,
                                         search_scope=SUBTREE, paged_cookie=cookie)
                    ad_users_list.extend(AdApi.connect.entries)
                    cookie = AdApi.connect.result['controls']['1.2.840.113556.1.4.319']['value']['cookie']
                    logging.debug(ad_users_list)
                return ad_users_list
            except Exception as e:
                logging.exception("init_connection error: %s", e)
                AdApi.server =None
                AdApi.connect =None
    
        @staticmethod
        def delete_ad_user(username,adconfig):
            """ 删除ad用户 :param username: :param adconfig: :return: """
            if AdApi.server is None or AdApi.connect is None:
                AdApi.init_connection()
            try:
                logging.info("delete_ad_user :"+username)
                res = AdApi.connect.delete('CN=' + username +',' + adconfig)
                logging.info(res)
                return res
            except Exception as e:
                logging.exception("delete_ad_user error: %s", e)
                AdApi.server =None
                AdApi.connect =None
                return False
    
        @staticmethod
        def add_ad_user(username, password, description,adflag):
            """ 增加ad用户 :param username: :param password: :param description: :param adflag: :return: """
            if AdApi.server is None or AdApi.connect is None:
                AdApi.init_connection()
            try:
                logging.info("add_ad_user :"+username)
                AdApi.connect.add('CN=' + username +',' + adflag,
                                  ['User'],
                                  { 'displayName': username,
                                   'description': description,
                                   'userPrincipalName':' %s@%s' % (username, app.config['DN']),
                                   'userAccountControl': '544',
                                   'sAMAccountName': username,
                                   'pwdLastSet': -1})
                #增加用户之后，修改密码
                USER_DN ='cn=%s,%s' % (username, adflag)
                CURREENTPWD =''
                NEWPWD = password
                ldap3.extend.microsoft.modifyPassword.ad_modify_password(AdApi.connect, USER_DN, NEWPWD,
                                                                         CURREENTPWD, controls=None)
                res = AdApi.connect.result
                if res['result'] ==0 and res['description'] =='success':
                    return True
                else:
                    return False
            except Exception as e:
                logging.exception("add_ad_user error: %s", e)
                AdApi.server =None
                AdApi.connect =None
                return False
    
        @staticmethod
        def disable_ad_user(username,adconfig):
            """ 禁用ad用户 :param username: :param adconfig: :return: """
            if AdApi.server is None or AdApi.connect is None:
                AdApi.init_connection()
            try:
                logging.info("disable_ad_user :"+username)
                AdApi.connect.modify('CN=' + username +',' + adconfig, { 'userAccountControl': [(MODIFY_REPLACE, ['514'])]})
                res = AdApi.connect.result
                if res['result'] ==0 and res['description'] =='success':
                    return True
                else:
                    return False
            except Exception as e:
                logging.exception("disable_ad_user error: %s", e)
                AdApi.server =None
                AdApi.connect =None
                return False
    
        @staticmethod
        def enable_ad_user(username,adconfig):
            """ 启用ad用户 :param username: :param adconfig: :return: """
            if AdApi.server is None or AdApi.connect is None:
                AdApi.init_connection()
            try:
                logging.info("enable_ad_user :" + username)
                AdApi.connect.modify('CN=' + username +',' + adconfig,
                                     { 'userAccountControl': [(MODIFY_REPLACE, ['544'])]})
                res = AdApi.connect.result
                if res['result'] ==0 and res['description'] =='success':
                    return True
                else:return False
            except Exception as e:
                logging.exception("enable_ad_user error: %s", e)
                AdApi.server =None
                AdApi.connect =None
                return False
    
        @staticmethod
        def get_user_pwd_last_set(username):
            if AdApi.server is None or AdApi.connect is None:
                AdApi.init_connection()
            try:
                AdApi.connect.search(app.config['BASEDN'], '(&(objectclass=User)(CN=%s))' % username, attributes=['pwdLastSet'])
                ens = AdApi.connect.entries
                for e in ens:
                    pwd_last_set = e['pwdLastSet']
                    pwd_last_set = pwd_last_set.value
                return pwd_last_set
            except Exception as e:
                logging.exception("get_user_pwd_last_set error: %s", e)
                AdApi.server =None
                AdApi.connect =None

[https_www.jianshu.com_p_fabb04fd93e3]: https://www.jianshu.com/p/fabb04fd93e3