sources/ldap: fix FreeIPA nsaccountlock sync (#6745)

Signed-off-by: Jens Langhammer <[email protected]>

authentik/sources/ldap/sync/vendor/freeipa.py

@@ -45,7 +45,11 @@ def check_nsaccountlock(self, attributes: dict[str, Any], user: User):
         # 389-ds and this will trigger regardless
         if "nsaccountlock" not in attributes:
             return
-        is_active = attributes.get("nsaccountlock", False)
+        # For some reason, nsaccountlock is not defined properly in the schema as bool
+        # hence we get it as a list of strings
+        _is_active = str(self._flatten(attributes.get("nsaccountlock", ["FALSE"])))
+        # So we have to attempt to convert it to a bool
+        is_active = _is_active.lower() == "true"
         if is_active != user.is_active:
             user.is_active = is_active
             user.save()

authentik/sources/ldap/tests/mock_freeipa.py

@@ -0,0 +1,111 @@
+"""ldap testing utils"""
+
+from ldap3 import MOCK_SYNC, OFFLINE_DS389_1_3_3, Connection, Server
+
+
+def mock_freeipa_connection(password: str) -> Connection:
+    """Create mock FreeIPA-ish connection"""
+    server = Server("my_fake_server", get_info=OFFLINE_DS389_1_3_3)
+    _pass = "foo"  # noqa # nosec
+    connection = Connection(
+        server,
+        user="cn=my_user,dc=goauthentik,dc=io",
+        password=_pass,
+        client_strategy=MOCK_SYNC,
+    )
+    # Entry for password checking
+    connection.strategy.add_entry(
+        "cn=user,ou=users,dc=goauthentik,dc=io",
+        {
+            "name": "test-user",
+            "uid": "unique-test-group",
+            "objectClass": "person",
+            "displayName": "Erin M. Hagens",
+        },
+    )
+    connection.strategy.add_entry(
+        "cn=group1,ou=groups,dc=goauthentik,dc=io",
+        {
+            "cn": "group1",
+            "uid": "unique-test-group",
+            "objectClass": "groupOfNames",
+            "member": ["cn=user0,ou=users,dc=goauthentik,dc=io"],
+        },
+    )
+    # Group without SID
+    connection.strategy.add_entry(
+        "cn=group2,ou=groups,dc=goauthentik,dc=io",
+        {
+            "cn": "group2",
+            "objectClass": "groupOfNames",
+        },
+    )
+    connection.strategy.add_entry(
+        "cn=user0,ou=users,dc=goauthentik,dc=io",
+        {
+            "userPassword": password,
+            "name": "user0_sn",
+            "uid": "user0_sn",
+            "objectClass": "person",
+        },
+    )
+    # User without SID
+    connection.strategy.add_entry(
+        "cn=user1,ou=users,dc=goauthentik,dc=io",
+        {
+            "userPassword": "test1111",
+            "name": "user1_sn",
+            "objectClass": "person",
+        },
+    )
+    # Duplicate users
+    connection.strategy.add_entry(
+        "cn=user2,ou=users,dc=goauthentik,dc=io",
+        {
+            "userPassword": "test2222",
+            "name": "user2_sn",
+            "uid": "unique-test2222",
+            "objectClass": "person",
+        },
+    )
+    connection.strategy.add_entry(
+        "cn=user3,ou=users,dc=goauthentik,dc=io",
+        {
+            "userPassword": "test2222",
+            "name": "user2_sn",
+            "uid": "unique-test2222",
+            "objectClass": "person",
+        },
+    )
+    # Group with posixGroup and memberUid
+    connection.strategy.add_entry(
+        "cn=group-posix,ou=groups,dc=goauthentik,dc=io",
+        {
+            "cn": "group-posix",
+            "objectClass": "posixGroup",
+            "memberUid": ["user-posix"],
+        },
+    )
+    # User with posixAccount
+    connection.strategy.add_entry(
+        "cn=user-posix,ou=users,dc=goauthentik,dc=io",
+        {
+            "userPassword": password,
+            "uid": "user-posix",
+            "cn": "user-posix",
+            "objectClass": "posixAccount",
+        },
+    )
+    # Locked out user
+    connection.strategy.add_entry(
+        "cn=user-nsaccountlock,ou=users,dc=goauthentik,dc=io",
+        {
+            "userPassword": password,
+            "uid": "user-nsaccountlock",
+            "cn": "user-nsaccountlock",
+            "objectClass": "person",
+            "nsaccountlock": ["TRUE"],
+        },
+    )
+    connection.bind()
+    return connection

authentik/sources/ldap/tests/mock_slapd.py

@@ -4,7 +4,7 @@
 
 
 def mock_slapd_connection(password: str) -> Connection:
-    """Create mock AD connection"""
+    """Create mock SLAPD connection"""
     server = Server("my_fake_server", get_info=OFFLINE_SLAPD_2_4)
     _pass = "foo"  # noqa # nosec
     connection = Connection(

authentik/sources/ldap/tests/test_sync.py

@@ -17,6 +17,7 @@
 from authentik.sources.ldap.sync.users import UserLDAPSynchronizer
 from authentik.sources.ldap.tasks import ldap_sync, ldap_sync_all
 from authentik.sources.ldap.tests.mock_ad import mock_ad_connection
+from authentik.sources.ldap.tests.mock_freeipa import mock_freeipa_connection
 from authentik.sources.ldap.tests.mock_slapd import mock_slapd_connection
 
 LDAP_PASSWORD = generate_key()
@@ -120,6 +121,23 @@ def test_sync_users_openldap(self):
             self.assertTrue(User.objects.filter(username="user0_sn").exists())
             self.assertFalse(User.objects.filter(username="user1_sn").exists())
 
+    def test_sync_users_freeipa_ish(self):
+        """Test user sync (FreeIPA-ish), mainly testing vendor quirks"""
+        self.source.object_uniqueness_field = "uid"
+        self.source.property_mappings.set(
+            LDAPPropertyMapping.objects.filter(
+                Q(managed__startswith="goauthentik.io/sources/ldap/default")
+                | Q(managed__startswith="goauthentik.io/sources/ldap/openldap")
+            )
+        )
+        self.source.save()
+        connection = MagicMock(return_value=mock_freeipa_connection(LDAP_PASSWORD))
+        with patch("authentik.sources.ldap.models.LDAPSource.connection", connection):
+            user_sync = UserLDAPSynchronizer(self.source)
+            user_sync.sync_full()
+            self.assertTrue(User.objects.filter(username="user0_sn").exists())
+            self.assertFalse(User.objects.filter(username="user1_sn").exists())
+
     def test_sync_groups_ad(self):
         """Test group sync"""
         self.source.property_mappings.set(