2022年 11月 8日

使用Python实现与AD域进行对接

上上周接到需求，要将软件实现与客户AD域的信息进行对接。

需求如下：

    1、在软件注册用户的页面新建一个功能，能够查询该用户名是否在AD域内存在。

    2、用户登陆软件的时候对输入的账号密码与AD域内的账号密码进行校验，能够实时以AD域的账号密码登陆。

公司的考勤软件使用Django框架开发，用户登陆这一块用的是Django本身‘密码MD5加密’—‘保存数据库’—‘登陆时进行比对’的逻辑。

看到需求，直接否认了这种Django本身的逻辑，开玩笑，几千人的密码要进行实时更新，对服务器的压力也太大了，而且windows的官方文档写的清清楚楚：AD域不会提供密码。

Python对接AD域自然要用到ldap库，安装ldap库的时候最开始使用pip安装，后来软件报错才发现安装错误，应该安装python-ldap，而不是ldap。后来找到64位的安装包才安装成功。

（windows使用python有时候会有很多坑，因为有些库是依赖于VS，比较麻烦。）

ldap提供了接口可以供我们对AD域内的信息进行比对，只用修改Django的登陆逻辑，每次登陆的时候调用接口，和AD域内的信息进行比对就行了。

代码如下：

  1. # -*- coding: UTF-8 -*-
  2. import sys
  3. reload(sys)
  4. sys.setdefaultencoding('utf-8')
  5.  
  6. import ldap,logging,time
  7. logfile = 'e:\\a.txt'
  8. # logging.basicConfig(filename=logfile,level=logging.INFO)
  9. # logging.basicConfig(format='%(time.asctime)s %(message)s', datefmt='%m/%d/%Y %I:%M:%S %p')
  10. logging.basicConfig(level=logging.INFO,
  11.                     #format='%(asctime)s %(filename)s[line:%(lineno)d] %(levelname)s %(message)s', #返回值：Thu, 26 May 2016 15:09:31 t11.py[line:92] INFO
  12.                     format='%(asctime)s %(levelname)s %(message)s',
  13.                     #datefmt='%a, %d %b %Y %H:%M:%S',
  14.                     #datefmt='%Y/%m/%d %I:%M:%S %p', #返回2016/05/26 03:12:56 PM
  15.                     datefmt='%Y-%m-%d %H:%M:%S', #返回2016/05/26 03:12:56 PM
  16.                     filename=logfile#,
  17.                     #filemode='a' #默认为a
  18.                    )
  19. #logging输出结果：
  20. #2016-05-26 15:22:29 INFO liu1 valid passed.
  21. #2016-05-26 15:22:37 INFO liu1 valid passed.
  22.  
  23.  
  24. class ldapc:
  25.     def __init__(self,ldap_path,baseDN,ldap_authuser,ldap_authpass):
  26.         self.baseDN = baseDN
  27.         self.ldap_error = None
  28.         self.l=ldap.initialize(ldap_path)
  29.         self.l.protocol_version = ldap.VERSION3
  30.         print 'url:',ldap_path
  31.         print 'username:',ldap_authuser
  32.         print 'password:',ldap_authpass
  33.         try:
  34.             res=self.l.simple_bind_s(ldap_authuser,ldap_authpass)
  35.             print res
  36.         except ldap.LDAPError,err:
  37.             self.ldap_error = 'Connect to %s failed, Error:%s.' %(ldap_path,err.message['desc'])
  38.             print self.ldap_error
  39.         # finally:
  40.         #     self.l.unbind_s()
  41.         #     del self.l
  42.  
  43.     def search_users(self,username): #模糊查找，返回一个list，使用search_s()
  44.         if self.ldap_error is None:
  45.             try:
  46.                 searchScope = ldap.SCOPE_SUBTREE
  47.                 searchFiltername = "sAMAccountName" #通过samaccountname查找用户
  48.                 retrieveAttributes = None
  49.                 searchFilter = '(' + searchFiltername + "=" + username +'*)'
  50.                 ldap_result =self.l.search_s(self.baseDN, searchScope, searchFilter, retrieveAttributes)
  51.                 if len(ldap_result) == 0: #ldap_result is a list.
  52.                     return "%s doesn't exist." %username
  53.                 else:
  54.                     # result_type, result_data = self.l.result(ldap_result, 0)
  55.                     # return result_type, ldap_result
  56.                     return ldap_result
  57.             except ldap.LDAPError,err:
  58.                 return err
  59.  
  60.     def search_user(self,username): #精确查找，返回值为list，使用search()
  61.         if self.ldap_error is None:
  62.             try:
  63.                 searchScope = ldap.SCOPE_SUBTREE
  64.                 searchFiltername = "sAMAccountName" #通过samaccountname查找用户
  65.                 retrieveAttributes = None
  66.                 searchFilter = '(' + searchFiltername + "=" + username +')'
  67.                 ldap_result_id =self.l.search(self.baseDN, searchScope, searchFilter, retrieveAttributes)
  68.                 result_type, result_data = self.l.result(ldap_result_id, 0)
  69.                 if result_type == ldap.RES_SEARCH_ENTRY:
  70.                     return result_data
  71.                 else:
  72.                     return "%s doesn't exist." %username
  73.             except ldap.LDAPError,err:
  74.                 return err
  75.  
  76.     def search_userDN(self,username): #精确查找，最后返回该用户的DN值
  77.         if self.ldap_error is None:
  78.             try:
  79.                 searchScope = ldap.SCOPE_SUBTREE
  80.                 searchFiltername = "sAMAccountName" #通过samaccountname查找用户
  81.                 retrieveAttributes = None
  82.                 searchFilter = '(' + searchFiltername + "=" + username +')'
  83.                 ldap_result_id =self.l.search(self.baseDN, searchScope, searchFilter, retrieveAttributes)
  84.                 result_type, result_data = self.l.result(ldap_result_id, 0)
  85.                 if result_type == ldap.RES_SEARCH_ENTRY:
  86.                     print 'flag:',1
  87.                     return result_data[0][0] #list第一个值为用户的DN，第二个值是一个dict，包含了用户属性信息
  88.                 else:
  89.                     print 'flag:',0
  90.                     return "%s doesn't exist." %username
  91.             except ldap.LDAPError,err:
  92.                 return err
  93.  
  94.     def valid_user(self,username,userpassword): #验证用户密码是否正确
  95.         if self.ldap_error is None:
  96.             target_user = self.search_userDN(username) #使用前面定义的search_userDN函数获取用户的DN
  97.             if target_user.find("doesn't exist") == -1:
  98.                 try:
  99.                     self.l.simple_bind_s(target_user,userpassword)
  100.                     logging.info('%s valid passed.\r'%(username)) #logging会自动在每行log后面添加"\000"换行，windows下未自动换行
  101.                     return True
  102.                 except ldap.LDAPError,err:
  103.                     return err
  104.             else:
  105.                 return target_user
  106.  
  107.     def update_pass(self,username,oldpassword,newpassword): #####未测试#########
  108.         if self.ldap_error is None:
  109.             target_user = self.search_userDN(username)
  110.             if target_user.find("doesn't exist") == -1:
  111.                 try:
  112.                     self.l.simple_bind_s(target_user,oldpassword)
  113.                     self.l.passwd_s(target_user,oldpassword,newpassword)
  114.                     return 'Change password success.'
  115.                 except ldap.LDAPError,err:
  116.                     return err
  117.             else:
  118.                 return target_user
  119.  
  120.  
  121.  
  122. ldap_authuser='xxx@xxxx.com'
  123. ldap_authpass='xxx'
  124. domainname='xxx'
  125. ldappath='ldap://xxx.xxx.xxx.xxx:xxx'
  126.  
  127. baseDN='DC=xxx,DC=com' #ldap_authuser在连接到LDAP的时候不会用到baseDN，在验证其他用户的时候才需要使用
  128. username = 'xxx' #要查找/验证的用户
  129. p=ldapc(ldappath,baseDN,ldap_authuser,ldap_authpass)
  130. # print 'list--search:',p.search_users(username)
  131. print 'DN----search',p.search_userDN(username)
  132. print 'DN-ty',type(p.search_userDN(username))
  133. print 'user--valid',p.valid_user(ldap_authuser,ldap_authpass) #调用valid_user()方法验证用户是否为合法用户

在这里只要配置好AD域的域名、URL地址、管理员权限的账号密码，就直接调用接口了。

PS：如果调用模糊查找里面的ldap.search_s()方法，一直在报错。但是用ldap.search()方法就没有问题。