From 866a627f57ebcf64f461d7bf1654ffb6952c968a Mon Sep 17 00:00:00 2001
From: Serghei Mihai <smihai@entrouvert.com>
Date: Thu, 25 Feb 2021 18:37:59 +0100
Subject: [PATCH 2/2] ldap: add method to deactivate orphaned users (#6379)

---
 src/authentic2/backends/ldap_backend.py | 38 +++++++++++++++++++++++--
 tests/test_ldap.py                      | 22 ++++++++++++++
 2 files changed, 58 insertions(+), 2 deletions(-)

diff --git a/src/authentic2/backends/ldap_backend.py b/src/authentic2/backends/ldap_backend.py
index 2be05a15..bfa194f5 100644
--- a/src/authentic2/backends/ldap_backend.py
+++ b/src/authentic2/backends/ldap_backend.py
@@ -1015,6 +1015,12 @@ class LDAPBackend(object):
                 attribute = attribute.split(':', 1)[0]
             yield attribute
 
+    @classmethod
+    def get_user_filter(cls, block):
+        user_filter = force_text(block['sync_ldap_users_filter'] or block['user_filter'])
+        user_filter = user_filter.replace('%s', '*')
+        return user_filter
+
     @classmethod
     def get_ldap_attributes_names(cls, block):
         attributes = set()
@@ -1309,8 +1315,7 @@ class LDAPBackend(object):
                 logger.warning(u'unable to synchronize with LDAP servers %s', force_text(block['url']))
                 continue
             user_basedn = force_text(block.get('user_basedn') or block['basedn'])
-            user_filter = force_text(block['sync_ldap_users_filter'] or block['user_filter'])
-            user_filter = user_filter.replace('%s', '*')
+            user_filter = cls.get_user_filter(block)
             attribute_names = cls.get_ldap_attributes_names(block)
             results = cls.paged_search(conn, user_basedn, ldap.SCOPE_SUBTREE, user_filter, attrlist=attribute_names)
             backend = cls()
@@ -1318,6 +1323,35 @@ class LDAPBackend(object):
                 yield backend._return_user(dn, None, conn, block, attrs)
 
 
+    @classmethod
+    def deactivate_orphaned_users(cls):
+        for block in cls.get_config():
+            conn = cls.get_connection(block)
+            if conn is None:
+                continue
+            eids = list(UserExternalId.objects.filter(user__is_active=True,
+                                                      source=block['realm']).values_list('external_id', flat=True))
+            basedn = force_text(block.get('user_basedn') or block['basedn'])
+            attribute_names = cls.get_ldap_attributes_names(block)
+            user_filter = cls.get_user_filter(block)
+            results = cls.paged_search(conn, basedn, ldap.SCOPE_SUBTREE,
+                                       user_filter,
+                                       attrlist=attribute_names)
+            for dn, attrs in results:
+                data = attrs.copy()
+                data['dn'] = dn
+                for eid_tuple in map_text(block['external_id_tuples']):
+                    backend = cls()
+                    external_id = backend.build_external_id(eid_tuple, data)
+                    if external_id:
+                        try:
+                            eids.remove(external_id)
+                        except ValueError:
+                            pass
+            for eid in UserExternalId.objects.filter(external_id__in=eids):
+                eid.user.mark_as_inactive()
+
+
     @classmethod
     def ad_encoding(cls, s):
         '''Encode a string for AD consumption as a password'''
diff --git a/tests/test_ldap.py b/tests/test_ldap.py
index c02898a3..018a9689 100644
--- a/tests/test_ldap.py
+++ b/tests/test_ldap.py
@@ -218,6 +218,28 @@ def test_simple(slapd, settings, client, db):
     assert 'password' not in client.session['ldap-data']
 
 
+def test_deactivate_orphaned_users(slapd, settings, client, db):
+    settings.LDAP_AUTH_SETTINGS = [{
+        'url': [slapd.ldap_url],
+        'basedn': u'o=ôrga',
+        'use_tls': False,
+    }]
+
+    # create users as a side effect
+    list(ldap_backend.LDAPBackend.get_users())
+    block = settings.LDAP_AUTH_SETTINGS[0]
+    assert ldap_backend.UserExternalId.objects.filter(
+        user__is_active=False, source=block['realm']).count() == 0
+
+    conn = slapd.get_connection_admin()
+    conn.delete_s(DN)
+
+    ldap_backend.LDAPBackend.deactivate_orphaned_users()
+
+    assert ldap_backend.UserExternalId.objects.filter(
+        user__is_active=False, source=block['realm']).count() == 1
+
+
 @pytest.mark.django_db
 def test_simple_with_binddn(slapd, settings, client):
     settings.LDAP_AUTH_SETTINGS = [{
-- 
2.31.0

