changeset a87fed70983b in modules/ldap_authentication:default
details:
https://hg.tryton.org/modules/ldap_authentication?cmd=changeset&node=a87fed70983b
description:
Add support for STARTTLS with LDAP server
issue11561
review441181003
diffstat:
CHANGELOG | 1 +
doc/index.rst | 10 +++
res.py | 15 ++++-
tests/test_module.py | 144 ++++++++++++++++++++++++++++++++++++++++++++++----
4 files changed, 156 insertions(+), 14 deletions(-)
diffs (277 lines):
diff -r cc05d028e137 -r a87fed70983b CHANGELOG
--- a/CHANGELOG Sun Oct 02 00:23:53 2022 +0200
+++ b/CHANGELOG Mon Oct 10 21:18:17 2022 +0200
@@ -1,3 +1,4 @@
+* Add support for STARTTLS
* Enforce certificate validation for LDAP connection (issue11564)
Version 6.4.0 - 2022-05-02
diff -r cc05d028e137 -r a87fed70983b doc/index.rst
--- a/doc/index.rst Sun Oct 02 00:23:53 2022 +0200
+++ b/doc/index.rst Mon Oct 10 21:18:17 2022 +0200
@@ -16,6 +16,13 @@
---
The LDAP URL to use to connect to the server following the :rfc:`2255`.
+It is extended to support SSL_ and STARTTLS_.
+The available protocols are:
+
+ - ``ldap``: simple LDAP
+ - ``ldap+tls``: LDAP with STARTTLS
+ - ``ldaps``: LDAP with SSL
+
bind_pass
---------
@@ -31,3 +38,6 @@
-----------
A boolean to create user if not in the database.
+
++.. _SSL: https://en.wikipedia.org/wiki/Secure_Sockets_Layer
++.. _STARTTLS: https://en.wikipedia.org/wiki/STARTTLS
diff -r cc05d028e137 -r a87fed70983b res.py
--- a/res.py Sun Oct 02 00:23:53 2022 +0200
+++ b/res.py Mon Oct 10 21:18:17 2022 +0200
@@ -47,6 +47,8 @@
else:
scheme, port = 'ldap', 389
tls = None
+ if 'tls' in uri.scheme:
+ tls = ldap3.Tls(validate=ssl.CERT_REQUIRED)
return ldap3.Server('%s://%s:%s' % (
scheme, uri.hostname, uri.port or port), tls=tls)
@@ -82,7 +84,12 @@
# XXX find better way to get the password
bindpass = config.get(section, 'bind_pass')
- with ldap3.Connection(server, bindname, bindpass) as con:
+ bind_method = ldap3.AUTO_BIND_DEFAULT
+ if server.ssl is False and server.tls is not None:
+ bind_method = ldap3.AUTO_BIND_TLS_BEFORE_BIND
+
+ with ldap3.Connection(
+ server, bindname, bindpass, auto_bind=bind_method) as con:
con.search(dn, filter_, search_scope=scope, attributes=attrs)
result = con.entries
if result and len(result) > 1:
@@ -141,7 +148,11 @@
users = cls.ldap_search_user(login, server, attrs=[uid])
if users and len(users) == 1:
[(dn, attrs)] = users
- with ldap3.Connection(server, dn, password) as con:
+ with ldap3.Connection(
+ server, dn, password,
+ auto_bind=ldap3.AUTO_BIND_NONE) as con:
+ if server.ssl is False and server.tls is not None:
+ con.start_tls()
if (password and con.bind()):
# Use ldap uid so we always get the right case
login = attrs.get(uid, [login])[0]
diff -r cc05d028e137 -r a87fed70983b tests/test_module.py
--- a/tests/test_module.py Sun Oct 02 00:23:53 2022 +0200
+++ b/tests/test_module.py Mon Oct 10 21:18:17 2022 +0200
@@ -1,12 +1,12 @@
# This file is part of Tryton. The COPYRIGHT file at the top level of
# this repository contains the full copyright notices and license terms.
-from unittest.mock import ANY, patch
+from unittest.mock import ANY, MagicMock, Mock, patch
import ldap3
from trytond.config import config
-from trytond.modules.ldap_authentication.res import parse_ldap_url
+from trytond.modules.ldap_authentication.res import ldap_server, parse_ldap_url
from trytond.pool import Pool
from trytond.tests.test_tryton import ModuleTestCase, with_transaction
@@ -23,19 +23,19 @@
config.set('session', 'authentications', 'ldap')
self.addCleanup(config.set, 'session', 'authentications', methods)
config.add_section(section)
- config.set(section, 'uri', 'ldap://localhost/dc=tryton,dc=org')
self.addCleanup(config.remove_section, section)
- @with_transaction()
- def test_user_get_login(self):
- 'Test User.get_login'
+ def _get_login(
+ self, uri='ldap://localhost/dc=tryton,dc=org', start_tls=False):
pool = Pool()
User = pool.get('res.user')
+ config.set(section, 'uri', uri)
@patch.object(ldap3, 'Connection')
@patch.object(User, 'ldap_search_user')
def get_login(login, password, find, ldap_search_user, Connection):
- con = Connection.return_value
+ con = Connection.return_value = MagicMock()
+ con.__enter__.return_value = con
con.bind.return_value = bool(find)
if find:
ldap_search_user.return_value = [('dn', {'uid': [find]})]
@@ -45,31 +45,87 @@
'password': password,
})
if find:
- Connection.assert_called_with(ANY, ANY, password)
+ Connection.assert_called_with(
+ ANY, ANY, password, auto_bind=ldap3.AUTO_BIND_NONE)
+ if start_tls:
+ con.start_tls.assert_called()
+ else:
+ con.start_tls.assert_not_called()
+ con.bind.assert_called()
return user_id
+ return get_login
- # Test existing user
+ @with_transaction()
+ def test_user_get_login_existing_user(self):
+ "Test User.get_login with existing user"
+ pool = Pool()
+ User = pool.get('res.user')
user, = User.search([('login', '=', 'admin')])
+
+ get_login = self._get_login()
+
self.assertEqual(get_login('admin', 'admin', 'admin'), user.id)
self.assertEqual(get_login('AdMiN', 'admin', 'admin'), user.id)
- # Test new user
+ @with_transaction()
+ def test_user_get_login_unknown_user(self):
+ "test User.get_login with unknown user"
+ get_login = self._get_login()
+
self.assertFalse(get_login('foo', 'bar', None))
self.assertFalse(get_login('foo', 'bar', 'foo'))
- # Test create new user
+ @with_transaction()
+ def test_user_get_login_create_user(self):
+ "Test User.get_login with user to create"
+ pool = Pool()
+ User = pool.get('res.user')
config.set(section, 'create_user', 'True')
+ get_login = self._get_login()
+
user_id = get_login('foo', 'bar', 'foo')
foo, = User.search([('login', '=', 'foo')])
+
self.assertEqual(user_id, foo.id)
self.assertEqual(foo.name, 'foo')
- # Test create new user with different case
+ @with_transaction()
+ def test_user_get_login_create_user_case(self):
+ "Test User.get_login with user to create with different case"
+ pool = Pool()
+ User = pool.get('res.user')
+ config.set(section, 'create_user', 'True')
+ get_login = self._get_login()
+
user_id = get_login('BaR', 'foo', 'bar')
bar, = User.search([('login', '=', 'bar')])
+
self.assertEqual(user_id, bar.id)
self.assertEqual(bar.name, 'bar')
+ @with_transaction()
+ def test_user_get_login_with_tls(self):
+ "Test User.get_login with TLS"
+ pool = Pool()
+ User = pool.get('res.user')
+ user, = User.search([('login', '=', 'admin')])
+
+ get_login = self._get_login(
+ 'ldap+tls://localhost/dc=tryton,dc=org', start_tls=True)
+
+ self.assertEqual(get_login('admin', 'admin', 'admin'), user.id)
+
+ @with_transaction()
+ def test_user_get_login_with_ssl(self):
+ "Test User.get_login with SSL"
+ pool = Pool()
+ User = pool.get('res.user')
+ user, = User.search([('login', '=', 'admin')])
+
+ get_login = self._get_login('ldaps://localhost/dc=tryton,dc=org')
+
+ self.assertEqual(get_login('admin', 'admin', 'admin'), user.id)
+
def test_parse_ldap_url(self):
'Test parse_ldap_url'
self.assertEqual(
@@ -99,5 +155,69 @@
'ldap:///??sub??!bindname=cn=Manager%2co=Foo')[5],
{'!bindname': ['cn=Manager,o=Foo']})
+ def test_ldap_server(self):
+ "Test ldap_server"
+ for uri, (host, tls) in [
+ ('ldap://localhost/dc=tryton,dc=org',
+ ('ldap://localhost:389', None)),
+ ('ldaps://localhost/dc=tryton,dc=org',
+ ('ldaps://localhost:636', True)),
+ ('ldap+tls://localhost/dc=tryton,dc=org',
+ ('ldap://localhost:389', True)),
+ ]:
+ config.set(section, 'uri', uri)
+
+ with patch('ldap3.Server') as Server:
+ ldap_server()
+
+ self.assertEqual(Server.call_args.args, (host,))
+ if tls:
+ self.assertTrue(Server.call_args.kwargs.get('tls'))
+ else:
+ self.assertFalse(Server.call_args.kwargs.get('tls'))
+
+ def _ldap_search_user(
+ self, uri='ldap://localhost/dc=tryton,dc=org',
+ auto_bind=ldap3.AUTO_BIND_DEFAULT):
+ pool = Pool()
+ User = pool.get('res.user')
+ config.set(section, 'uri', uri)
+
+ @patch.object(ldap3, 'Connection')
+ def ldap_search_user(login, attrs, Connection):
+ con = Connection.return_value = MagicMock()
+ con.__enter__.return_value = con
+ con.entries = [Mock()]
+ server = ldap_server()
+ User.ldap_search_user(login, server, attrs=attrs)
+ Connection.assert_called_with(
+ ANY, ANY, ANY, auto_bind=auto_bind)
+ con.search.assert_called()
+ return ldap_search_user
+
+ @with_transaction()
+ def test_ldap_search_user(self):
+ "Test User.ldap_search_user"
+ ldap_search_user = self._ldap_search_user()
+
+ ldap_search_user('admin', None)
+
+ @with_transaction()
+ def test_ldap_search_user_with_tls(self):
+ "Test User.ldap_search_user with TSL"
+ ldap_search_user = self._ldap_search_user(
+ 'ldap+tls://localhost/dc=tryton,dc=org',
+ ldap3.AUTO_BIND_TLS_BEFORE_BIND)
+
+ ldap_search_user('admin', None)
+
+ @with_transaction()
+ def test_ldap_search_user_with_ssl(self):
+ "Test User.ldap_search_user with SSL"
+ ldap_search_user = self._ldap_search_user(
+ 'ldaps://localhost/dc=tryton,dc=org')
+
+ ldap_search_user('admin', None)
+
del ModuleTestCase
