Switch to ldap3

issue5324
review19251002

CHANGELOG

@@ -1,3 +1,5 @@
+* Switch to ldap3
+
 Version 4.0.0 - 2016-05-02
 * Bug fixes (see mercurial logs for details)
 

res.py

@@ -3,7 +3,7 @@
 import logging
 import urlparse
 
-import ldap
+import ldap3
 from trytond.transaction import Transaction
 from trytond.pool import PoolMeta
 from trytond.config import config, parse_uri
@@ -34,7 +34,7 @@ def parse_ldap_url(uri):
         extensions)
 
 
-def ldap_connection():
+def ldap_server():
     uri = config.get(section, 'uri')
     if not uri:
         return
@@ -43,27 +43,8 @@ def ldap_connection():
         scheme, port = 'ldaps', 636
     else:
         scheme, port = 'ldap', 389
-    conn = ldap.initialize('%s://%s:%s/' % (
+    return ldap3.Server('%s://%s:%s' % (
             scheme, uri.hostname, uri.port or port))
-    if config.getboolean(section, 'active_directory', default=False):
-        conn.set_option(ldap.OPT_REFERRALS, 0)
-    if 'tls' in uri.scheme:
-        conn.start_tls_s()
-
-    bindname, = extensions.get('bindname', [None])
-    if not bindname:
-        bindname, = extensions.get('!bindname', [None])
-    if bindname:
-        # XXX find better way to get the password
-        conn.simple_bind_s(bindname, config.get(section, 'bind_pass'))
-    return conn
-
-
-# python-ldap works only with str
-def unicode2str(param):
-    if isinstance(param, unicode):
-        param = param.encode('utf-8')
-    return param
 
 
 class User:
@@ -78,44 +59,53 @@ class User:
                 })
 
     @staticmethod
-    def ldap_search_user(login, con, attrs=None):
+    def ldap_search_user(login, server, attrs=None):
         '''
         Return the result of a ldap search for the login using the ldap
-        connection con based on connection.
+        server.
         The attributes values defined in attrs will be return.
         '''
-        _, dn, _, scope, filter_, _ = parse_ldap_url(
+        _, dn, _, scope, filter_, extensions = parse_ldap_url(
             config.get(section, 'uri'))
         scope = {
-            'base': ldap.SCOPE_BASE,
-            'onelevel': ldap.SCOPE_ONELEVEL,
-            'subtree': ldap.SCOPE_SUBTREE,
+            'base': ldap3.BASE,
+            'onelevel': ldap3.LEVEL,
+            'subtree': ldap3.SUBTREE,
             }.get(scope)
         uid = config.get(section, 'uid', default='uid')
         if filter_:
-            filter_ = '(&(%s=%s)%s)' % (uid, unicode2str(login), filter_)
+            filter_ = '(&(%s=%s)%s)' % (uid, login, filter_)
         else:
-            filter_ = '(%s=%s)' % (uid, unicode2str(login))
+            filter_ = '(%s=%s)' % (uid, login)
 
-        result = con.search_s(dn, scope, filter_, attrs)
-        if config.get(section, 'active_directory'):
-            result = [x for x in result if x[0]]
-        if result and len(result) > 1:
-            logger.info('ldap_search_user found more than 1 user')
-        return result
+        bindpass = None
+        bindname, = extensions.get('bindname', [None])
+        if not bindname:
+            bindname, = extensions.get('!bindname', [None])
+        if bindname:
+            # XXX find better way to get the password
+            bindpass = config.get(section, 'bind_pass')
+
+        with ldap3.Connection(server, bindname, bindpass) as con:
+            con.search(dn, filter_, search_scope=scope, attributes=attrs)
+            result = con.entries
+            if result and len(result) > 1:
+                logger.info('ldap_search_user found more than 1 user')
+            return [(e.entry_get_dn(), e.entry_get_attributes_dict())
+                for e in result]
 
     @classmethod
     def _check_passwd_ldap_user(cls, logins):
         find = False
         try:
-            con = ldap_connection()
-            if not con:
+            server = ldap_server()
+            if not server:
                 return
             for login in logins:
-                if cls.ldap_search_user(login, con, attrs=[]):
+                if cls.ldap_search_user(login, server, attrs=[]):
                     find = True
                     break
-        except ldap.LDAPError:
+        except ldap3.LDAPException:
             logger.error('LDAPError when checking password', exc_info=True)
         if find:
             cls.raise_user_error('set_passwd_ldap_user', (login,))
@@ -148,22 +138,23 @@ class User:
                 raise LoginException('password', msg, type='password')
             old_password = parameters['password']
             try:
-                con = ldap_connection()
-                if con:
+                server = ldap_server()
+                if server:
                     user = cls(Transaction().user)
                     uid = config.get(section, 'uid', default='uid')
-                    users = cls.ldap_search_user(user.login, con, attrs=[uid])
+                    users = cls.ldap_search_user(
+                        user.login, server, attrs=[uid])
                     if users and len(users) == 1:
                         [(dn, attrs)] = users
-                        if con.simple_bind_s(dn, unicode2str(old_password)):
-                            con.passwd_s(
-                                dn, unicode2str(old_password),
-                                unicode2str(values['password']))
+                        con = ldap3.Connection(server, dn, old_password)
+                        if con.bind():
+                            con.extend.standard.modify_password(
+                                dn, old_password, values['password'])
                             values = values.copy()
                             del values['password']
                         else:
                             cls.raise_user_error('wrong_password')
-            except ldap.LDAPError:
+            except ldap3.LDAPException:
                 logger.error('LDAPError when setting preferences',
                     exc_info=True)
         super(User, cls).set_preferences(values, old_password=old_password)
@@ -175,14 +166,14 @@ class User:
             raise LoginException('password', msg, type='password')
         password = parameters['password']
         try:
-            con = ldap_connection()
-            if con:
+            server = ldap_server()
+            if server:
                 uid = config.get(section, 'uid', default='uid')
-                users = cls.ldap_search_user(login, con, attrs=[uid])
+                users = cls.ldap_search_user(login, server, attrs=[uid])
                 if users and len(users) == 1:
                     [(dn, attrs)] = users
-                    if (password
-                            and con.simple_bind_s(dn, unicode2str(password))):
+                    con = ldap3.Connection(server, dn, password)
+                    if (password and con.bind()):
                         # Use ldap uid so we always get the right case
                         login = attrs.get(uid, [login])[0]
                         user_id, _ = cls._get_login(login)
@@ -194,5 +185,5 @@ class User:
                                         'login': login,
                                         }])
                             return user.id
-        except ldap.LDAPError:
+        except ldap3.LDAPException:
             logger.error('LDAPError when login', exc_info=True)

setup.py

@@ -41,7 +41,7 @@ if minor_version % 2:
         'hg+http://hg.tryton.org/modules/%s#egg=%s-%s' % (
             name[8:], name, version))
 
-requires = ['python-ldap']
+requires = ['ldap3']
 for dep in info.get('depends', []):
     if not re.match(r'(ir|res)(\W|$)', dep):
         requires.append(get_require_version('trytond_%s' % dep))

tests/test_ldap_authentication.py

@@ -2,8 +2,8 @@
 # this repository contains the full copyright notices and license terms.
 import unittest
 
-from mock import patch
-import ldap
+from mock import patch, ANY
+import ldap3
 
 import trytond.tests.test_tryton
 from trytond.tests.test_tryton import ModuleTestCase, with_transaction
@@ -34,18 +34,21 @@ class LDAPAuthenticationTestCase(ModuleTestCase):
         pool = Pool()
         User = pool.get('res.user')
 
-        @patch.object(ldap, 'initialize')
+        @patch.object(ldap3, 'Connection')
         @patch.object(User, 'ldap_search_user')
-        def get_login(login, password, find, ldap_search_user, initialize):
-            con = initialize.return_value
-            con.simple_bind_s.return_value = True
+        def get_login(login, password, find, ldap_search_user, Connection):
+            con = Connection.return_value
+            con.bind.return_value = bool(find)
             if find:
                 ldap_search_user.return_value = [('dn', {'uid': [find]})]
             else:
                 ldap_search_user.return_value = None
-            return User.get_login(login, {
+            user_id = User.get_login(login, {
                     'password': password,
                     })
+            if find:
+                Connection.assert_called_with(ANY, ANY, password)
+            return user_id
 
         # Test existing user
         user, = User.search([('login', '=', 'admin')])