From 7fccc5bb9ef355223d542994b57eb51fd1272f66 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Nicolas=20=C3=89vrard?= Date: Fri, 3 Jun 2016 16:40:43 +0200 Subject: [PATCH] Switch to ldap3 issue5324 review19251002 --- CHANGELOG | 2 + res.py | 99 ++++++++++++++----------------- setup.py | 2 +- tests/test_ldap_authentication.py | 17 +++--- 4 files changed, 58 insertions(+), 62 deletions(-) diff --git a/CHANGELOG b/CHANGELOG index 28a2d98bc6..b808476d5c 100644 --- a/CHANGELOG +++ b/CHANGELOG @@ -1,3 +1,5 @@ +* Switch to ldap3 + Version 4.0.0 - 2016-05-02 * Bug fixes (see mercurial logs for details) diff --git a/res.py b/res.py index 0d20b8be25..e0c88d8000 100644 --- a/res.py +++ b/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) diff --git a/setup.py b/setup.py index c09cd96fd8..60e1901a85 100644 --- a/setup.py +++ b/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)) diff --git a/tests/test_ldap_authentication.py b/tests/test_ldap_authentication.py index 9ab2a7bda8..cce1459682 100644 --- a/tests/test_ldap_authentication.py +++ b/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')])