From b75d681f602f3ef765ae617660f553bac791ea95 Mon Sep 17 00:00:00 2001
From: Serghei Mihai <smihai@entrouvert.com>
Date: Tue, 18 Dec 2018 15:10:31 +0100
Subject: [PATCH] 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, 205 insertions(+), 79 deletions(-)

diff --git a/src/authentic2/backends/ldap_backend.py b/src/authentic2/backends/ldap_backend.py
index dc084407..94e17849 100644
--- a/src/authentic2/backends/ldap_backend.py
+++ b/src/authentic2/backends/ldap_backend.py
@@ -463,6 +463,10 @@ class LDAPBackend(object):
         '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')
@@ -520,86 +524,16 @@ class LDAPBackend(object):
             if user is not None:
                 return user
 
-    def authenticate_block(self, block, username, password):
+    def authenticate_block(self, block, username, password=None):
+        utf8_username = force_bytes(username)
         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 as e:
-                                    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, 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'])
@@ -607,10 +541,68 @@ class LDAPBackend(object):
                 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, 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_bytes(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 = block['user_filter']
+        n = len(user_filter.split('%s')) - 1
+
+        try:
+            query = filter_format(user_filter, (username,) * n)
+        except TypeError, 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, e:
+            log.error('user dn lookup failed: with query %r got error %s', username,
+                      query)
+            raise
+
+    def try_bind(self, conn, block, dn, password):
+        try:
+            conn.simple_bind_s(dn, password)
+            if not block['connect_with_user_credentials']:
+                try:
+                    self.bind(block, conn)
+                except Exception as e:
+                    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:
@@ -1119,7 +1111,6 @@ class LDAPBackend(object):
 
     @classmethod
     def get_users(cls):
-        logger = logging.getLogger(__name__)
         for block in cls.get_config():
             conn = cls.get_connection(block)
             if conn is None:
diff --git a/src/authentic2_auth_saml/__init__.py b/src/authentic2_auth_saml/__init__.py
index eeeee9f5..0153973e 100644
--- a/src/authentic2_auth_saml/__init__.py
+++ b/src/authentic2_auth_saml/__init__.py
@@ -7,7 +7,8 @@ class Plugin(object):
         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']
diff --git a/src/authentic2_auth_saml/backends.py b/src/authentic2_auth_saml/backends.py
index ca448712..63e8323e 100644
--- a/src/authentic2_auth_saml/backends.py
+++ b/src/authentic2_auth_saml/backends.py
@@ -1,6 +1,26 @@
+# 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
 
@@ -22,3 +42,27 @@ class SAMLBackend(SAMLBackend):
 
         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)
diff --git a/tests/test_auth_saml.py b/tests/test_auth_saml.py
index cd9618e8..c3aed994 100644
--- a/tests/test_auth_saml.py
+++ b/tests/test_auth_saml.py
@@ -1,6 +1,27 @@
+#
+# 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
@@ -38,3 +59,46 @@ def test_provision_attributes():
     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
diff --git a/tests/test_ldap.py b/tests/test_ldap.py
index 7a352119..b59ae215 100644
--- a/tests/test_ldap.py
+++ b/tests/test_ldap.py
@@ -542,6 +542,32 @@ def test_no_connect_with_user_credentials(slapd_strict_acl, db, settings, app):
     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],
-- 
2.20.1

