0002-backends-add-SAML-LDAP-authentication-backend-30125.patch

Benjamin Dauvergne, 12 avril 2019 18:45

Voir les différences: en ligne côte à côte

Subject: [PATCH 2/2] backends: add SAML LDAP authentication backend (#30125)

 src/authentic2/backends/ldap_backend.py | 147 +++++++++++-------------
 src/authentic2_auth_saml/__init__.py    |   3 +-
 src/authentic2_auth_saml/backends.py    |  44 +++++++
 tests/test_auth_saml.py                 |  64 +++++++++++
 tests/test_ldap.py                      |  26 +++++
 5 files changed, 204 insertions(+), 80 deletions(-)

             'can_reset_password': False,
             # mapping from LDAP attributes to User attributes
             'user_attributes': [],
             # entity id of the IDP based on the same LDAP
             'saml_entity_id': '',
             # template for SAML username
             'saml_username_template': '{uid[0]}'
+        }
         _REQUIRED = ('url', 'basedn')
         _TO_ITERABLE = ('url', 'groupsu', 'groupstaff', 'groupactive')
-...
                 if user is not None:
                     return user
         def authenticate_block(self, block, username, password):
         def authenticate_block(self, block, username, password=None):
             for conn in self.get_connections(block):
                 authz_ids = []
                 user_basedn = force_text(block.get('user_basedn') or block['basedn'])
                 try:
                     if block['user_dn_template']:
                         template = force_text(block['user_dn_template'])
                         escaped_username = escape_dn_chars(username)
                         authz_ids.append(template.format(username=escaped_username))
                     else:
                         try:
                             if block.get('bind_with_username'):
                                 authz_ids.append(username)
                             elif block['user_filter']:
                                 # allow multiple occurences of the username in the filter
                                 user_filter = force_text(block['user_filter'])
                                 n = len(user_filter.split('%s')) - 1
                                 try:
                                     query = filter_format(user_filter, (username,) * n)
                                 except TypeError as e:
                                     log.error('user_filter syntax error %r: %s', block['user_filter'],
                                               e)
                                     return
                                 log.debug('looking up dn for username %r using query %r', username,
                                           query)
                                 results = conn.search_s(user_basedn, ldap.SCOPE_SUBTREE, query, [u'1.1'])
                                 # remove search references
                                 results = [result for result in results if result[0] is not None]
                                 log.debug('found dns %r', results)
                                 if len(results) == 0:
                                     log.debug('user lookup failed: no entry found, %s' % query)
                                 elif not block['multimatch'] and len(results) > 1:
                                     log.error('user lookup failed: too many (%d) entries found: %s',
                                               len(results), query)
                                 else:
                                     authz_ids.extend(result[0] for result in results)
                             else:
                                 raise NotImplementedError
                         except ldap.NO_SUCH_OBJECT:
                             log.error('user lookup failed: basedn %s not found', user_basedn)
                             if block['replicas']:
                                 break
                             continue
                         except ldap.LDAPError as e:
                             log.error('user lookup failed: with query %r got error %s: %s', username,
                                       query, e)
                             continue
                     if not authz_ids:
                         continue
                     try:
                         failed = False
                         for authz_id in authz_ids:
                             if failed:
                                 continue
                             try:
                                 conn.simple_bind_s(authz_id, password)
                                 user_login_success(authz_id)
                                 if not block['connect_with_user_credentials']:
                                     try:
                                         self.bind(block, conn)
                                     except Exception:
                                         log.exception(u'rebind failure after login bind')
                                         raise ldap.SERVER_DOWN
                                 break
                             except ldap.INVALID_CREDENTIALS:
                                 user_login_failure(authz_id)
                                 pass
                     for dn in self.get_dn_from_username(conn, block, username):
                         if password is None or self.try_bind(conn, block, username, dn, password):
                             user_login_success(dn)
                             return self._return_user(dn, password, conn, block)
                         else:
                             log.debug('user bind failed: invalid credentials')
                             if block['replicas']:
                                 break
                             continue
                     except ldap.NO_SUCH_OBJECT:
                         # should not happen as we just searched for this object !
                         log.error('user bind failed: authz_id not found %r', ', '.join(authz_ids))
                         if block['replicas']:
                             break
                     return self._return_user(authz_id, password, conn, block)
                             user_login_failure(dn)
                 except ldap.CONNECT_ERROR:
                     log.error('connection to %r failed, did you forget to declare the TLS certificate '
                               'in /etc/ldap/ldap.conf ?', block['url'])
-...
                     log.error('connection to %r timed out', block['url'])
                 except ldap.SERVER_DOWN:
                     log.error('ldap authentication error: %r is down', block['url'])
                 except ldap.LDAPError as e:
                     log.error('ldap error: %s', e)
                     pass
                 finally:
                     del conn
             return None
         def get_dn_from_username(self, conn, block, username):
             user_basedn = block.get('user_basedn') or block['basedn']
             if block['user_dn_template']:
                 template = force_text(block['user_dn_template'])
                 escaped_username = escape_dn_chars(username)
                 username = template.format(username=escaped_username)
             # allow multiple occurences of the username in the filter
             user_filter = force_text(block['user_filter'])
             n = len(user_filter.split('%s')) - 1
             try:
                 query = filter_format(user_filter, (username,) * n)
             except TypeError as e:
                 log.error('user_filter syntax error %r: %s', block['user_filter'], e)
                 return []
             log.debug('looking up dn for username %r using query %r', username,
                       query)
             try:
                 results = conn.search_s(user_basedn, ldap.SCOPE_SUBTREE, query)
                 results = [result for result in results if result[0] is not None]
                 log.debug('found dns %r', results)
                 if len(results) == 0:
                     log.debug('user lookup failed: no entry found, %s' % query)
                     return []
                 elif not block['multimatch'] and len(results) > 1:
                     log.error('user lookup failed: too many (%d) entries found: %s',
                               len(results), query)
                     return []
                 else:
                     return [result[0] for result in results]
             except ldap.NO_SUCH_OBJECT:
                 log.error('user dn lookup failed: basedn %s not found', user_basedn)
                 raise
             except ldap.LDAPError:
                 log.error('user dn lookup failed: with query %r got error %s', username, query)
                 raise
         def try_bind(self, conn, block, username, dn, password):
             try:
                 conn.simple_bind_s(dn, password)
                 if not block['connect_with_user_credentials']:
                     try:
                         self.bind(block, conn)
                     except Exception:
                         log.exception(u'rebind failure after login bind')
                         return False
                 return True
             except ldap.INVALID_CREDENTIALS:
                 return False
             except ldap.NO_SUCH_OBJECT:
                 log.error('user bind failed: username not found %s', username)
                 return False
         def get_user(self, user_id, session=None):
             try:
                 try:
-...
         @classmethod
         def get_users(cls):
             logger = logging.getLogger(__name__)
             for block in cls.get_config():
                 conn = cls.get_connection(block)
                 if conn is None:
-...
                     if not isinstance(cls._DEFAULTS[d], bool) and d in cls._REQUIRED and not block[d]:
                         raise ImproperlyConfigured(
                             'LDAP_AUTH_SETTINGS: attribute %r is required but is empty')
                     # force_bytes all strings in iterable or dict
                     # force_text all strings in iterable or dict
                     if isinstance(block[d], (list, tuple, dict)):
                         block[d] = map_text(block[d])
             # lowercase LDAP attribute names

             return ['mellon', __name__]
         def get_authentication_backends(self):
             return ['authentic2_auth_saml.backends.SAMLBackend']
             return ['authentic2_auth_saml.backends.SAMLLdapBackend',
                     'authentic2_auth_saml.backends.SAMLBackend']
         def get_authenticators(self):
             return ['authentic2_auth_saml.authenticators.SAMLAuthenticator']

     # authentic2_auth_saml - Authentic2 SAML Auth plugin
     # Copyright (C) 2010-2019 Entr'ouvert
+    #
     # This program is free software: you can redistribute it and/or modify it
     # under the terms of the GNU Affero General Public License as published
     # by the Free Software Foundation, either version 3 of the License, or
     # (at your option) any later version.
+    #
     # This program is distributed in the hope that it will be useful,
     # but WITHOUT ANY WARRANTY; without even the implied warranty of
     # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
     # GNU Affero General Public License for more details.
+    #
     # You should have received a copy of the GNU Affero General Public License
     # along with this program.  If not, see <http://www.gnu.org/licenses/>.
     from django.conf import settings
     from mellon.backends import SAMLBackend
     from authentic2.middleware import StoreRequestMiddleware
     from authentic2.backends.ldap_backend import LDAPBackend
     from . import app_settings
-...
             import lasso
             return lasso.SAML2_AUTHN_CONTEXT_PREVIOUS_SESSION
     class SAMLLdapBackend(SAMLBackend):
         def get_username_from_saml_attributes(self, block, saml_attributes):
             username_template = block.get('saml_username_template', '')
             return username_template.format(**saml_attributes)
         def authenticate(self, saml_attributes, request=None):
             if not getattr(settings, 'LDAP_AUTH_SETTINGS', []):
                 return None
             no_ldap_for_entity_id = True
             for block in settings.LDAP_AUTH_SETTINGS:
                 if block.get('saml_entity_id', '') == saml_attributes['issuer']:
                     no_ldap_for_entity_id = False
                     break
             if no_ldap_for_entity_id:
                 return None
             username = self.get_username_from_saml_attributes(block, saml_attributes)
             backend = LDAPBackend()
             return backend.authenticate_block(block, username)

+    #
     # Copyright (C) 2010-2019 Entr'ouvert
+    #
     # This program is free software: you can redistribute it and/or modify it
     # under the terms of the GNU Affero General Public License as published
     # by the Free Software Foundation, either version 3 of the License, or
     # (at your option) any later version.
+    #
     # This program is distributed in the hope that it will be useful,
     # but WITHOUT ANY WARRANTY; without even the implied warranty of
     # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
     # GNU Affero General Public License for more details.
+    #
     # You should have received a copy of the GNU Affero General Public License
     # along with this program.  If not, see <http://www.gnu.org/licenses/>.
     import pytest
     import mock
     from django.contrib.auth import get_user_model
     from django.utils.timezone import datetime
     from django.test.utils import override_settings
     from authentic2.models import Attribute
     pytestmark = pytest.mark.django_db
-...
         del saml_attributes['mail']
         with pytest.raises(ValueError):
             adapter.finish_create_user(idp, saml_attributes, user)
     @mock.patch('authentic2.backends.ldap_backend.LDAPBackend.get_dn_from_username')
     @mock.patch('authentic2.backends.ldap_backend.LDAPBackend.authenticate_block')
     def test_disabled_saml_ldap_backend(ldap_authenticate_block, ldap_user_dn):
         from authentic2_auth_saml.backends import SAMLLdapBackend
         backend = SAMLLdapBackend()
         saml_attributes = {}
         assert backend.authenticate(saml_attributes) == None
         with override_settings(LDAP_AUTH_SETTINGS=[]):
             assert backend.authenticate(saml_attributes) == None
         LDAP_SETTINGS = {
             "realm": "example.com",
             "url": "ldaps://ldap.example.com/",
             "basedn": "ou=people,o=example,o=com",
             "user_filter": "(|(mail=%s)(uid=%s))",
             "sync_ldap_users_filter": "",
             "username_template": "{uid[0]}",
             "attributes": [ "uid" ],
             "set_mandatory_groups": ["LDAP Entrouvert"],
             "timeout": 3,
             "use_tls": False,
             'saml_entity_id': 'https://some-idp.com/idp/saml2/metadata',
             'global_ldap_options': {},
+        }
         saml_attributes = {'username': [u'foo'], 'first_name': [u'Foo'],
                     'last_name': [u'Bar'], 'uid': [u'foobar'],
                     'name_id_name_qualifier': u'https://some-idp.com/idp/saml2/metadata',
                     'authn_instant': datetime(2019, 1, 30, 11, 12, 40),
                     'name_id_format': u'urn:oasis:names:tc:SAML:1.1:nameid-format:unspecified',
                     'name_id_content': u'9f160fa0eb6045e5a5257a34fb4651e0',
                     'mail': [u'foo@example.com'],
                     'issuer': 'https://some-idp.com/idp/saml2/metadata'
+        }
         with override_settings(A2_AUTH_SAML_LDAP_ENABLE=True, LDAP_AUTH_SETTINGS=[LDAP_SETTINGS]):
             ldap_user_dn.return_value = []
             ldap_authenticate_block.return_value = None
             backend.authenticate(saml_attributes)
             assert ldap_authenticate_block.call_args[0][0] == LDAP_SETTINGS

         assert 'Étienne Michu' in response.body
     def test_authenticate_block(slapd_strict_acl, db, settings, app):
         slapd = slapd_strict_acl
         settings.LDAP_AUTH_SETTINGS = [{
             'url': [slapd.ldap_url],
             'binddn': force_text(slapd.root_bind_dn),
             'bindpw': force_text(slapd.root_bind_password),
             'basedn': u'o=ôrga',
             'use_tls': False,
         }]
         backend = ldap_backend.LDAPBackend()
         block = backend.get_config()[0]
         user = backend.authenticate_block(block, USERNAME, PASS.decode('utf-8'))
         assert user is None
         # authenticate user with no password
         block = backend.get_config()[0]
         user = backend.authenticate_block(block, USERNAME)
         assert user.username == 'etienne.michu@ldap'
         # do not authenticate with user credentials
         settings.LDAP_AUTH_SETTINGS[0]['connect_with_user_credentials'] = False
         block = backend.get_config()[0]
         user = backend.authenticate_block(block, USERNAME, PASS.decode('utf-8'))
         assert user.username == 'etienne.michu@ldap'
     def test_reset_password_ldap_user(slapd, settings, app, db):
         settings.LDAP_AUTH_SETTINGS = [{
             'url': [slapd.ldap_url],
+    -

Projet

Général

Profil

Produits Entr'ouvert » Authentic 2

0002-backends-add-SAML-LDAP-authentication-backend-30125.patch