providers/ldap: set password_change_date in check_pwd_last_set to avoid loop ending user sessions

authentik/core/models.py

@@ -356,6 +356,16 @@ def setter(raw_password):
 
         return check_password(raw_password, self.password, setter)
 
+    def set_unusable_password(self, change_datetime: datetime = None):
+        """
+        In addition to the base version this also updates the password change date.
+        @param change_datetime: Use this value for the change time instead of now()
+        """
+        if change_datetime is None:
+            change_datetime = now()
+        self.password_change_date = change_datetime
+        super().set_unusable_password()
+
     @property
     def uid(self) -> str:
         """Generate a globally unique UID, based on the user ID and the hashed secret key"""

authentik/sources/ldap/sync/base.py

@@ -1,11 +1,14 @@
 """Sync LDAP Users and groups into authentik"""
 
 from collections.abc import Generator
+from datetime import UTC, datetime
+from typing import Any
 
 from django.conf import settings
 from ldap3 import DEREF_ALWAYS, SUBTREE, Connection
 from structlog.stdlib import BoundLogger, get_logger
 
+from authentik.core.models import User
 from authentik.core.sources.mapper import SourceMapper
 from authentik.lib.config import CONFIG
 from authentik.lib.sync.mapper import PropertyMappingManager
@@ -65,6 +68,45 @@
             return f"{self._source.additional_group_dn},{self._source.base_dn}"
         return self._source.base_dn
 
+    def check_pwd_last_set(
+        self, attribute_name: str, attributes: dict[str, Any], user: User, created: bool
+    ):
+        """
+        Test if the ldap password is newer than the authentik password.
+        If the ldap password is newer set the user password to an unusable password.
+        This ends all users sessions and forces the user to relogin.
+        During next user login the used authentication backend MAY choose to write a new usable user
+        password.
+
+        @param attribute_name: The name of the ldap attribute holding the information when
+                               the password was changed
+        @param attributes: All ldap attributes
+        @param user: The user object we are currently syncing
+        @param created: True, if the user is newly created
+        @return:
+        """
+        if attribute_name not in attributes:
+            self._logger.debug(
+                f"Missing attribute {attribute_name}. Can not test if a newer ldap password is set."
+                f"Ldap and authentik passwords may be out of sync.",
+                user=user.username,
+                created=created,
+            )
+            return
+
+        pwd_last_set: datetime = attributes.get(attribute_name, datetime.now())
+        pwd_last_set = pwd_last_set.replace(tzinfo=UTC)
+        if created or pwd_last_set > user.password_change_date:
+            self.message(f"'{user.username}': Reset user's password")
+            self._logger.debug(
+                "Reset user's password",
+                user=user.username,
+                created=created,
+                pwd_last_set=pwd_last_set,
+            )
+            user.set_unusable_password(change_datetime=pwd_last_set)
+            user.save()
+
     def message(self, *args, **kwargs):
         """Add message that is later added to the System Task and shown to the user"""
         formatted_message = " ".join(args)

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

@@ -1,7 +1,6 @@
 """FreeIPA specific"""
 
 from collections.abc import Generator
-from datetime import UTC, datetime
 from typing import Any
 
 from authentik.core.models import User
@@ -20,26 +19,9 @@
         yield None
 
     def sync(self, attributes: dict[str, Any], user: User, created: bool):
-        self.check_pwd_last_set(attributes, user, created)
+        self.check_pwd_last_set("krbLastPwdChange", attributes, user, created)
         self.check_nsaccountlock(attributes, user)
 
-    def check_pwd_last_set(self, attributes: dict[str, Any], user: User, created: bool):
-        """Check krbLastPwdChange"""
-        if "krbLastPwdChange" not in attributes:
-            return
-        pwd_last_set: datetime = attributes.get("krbLastPwdChange", datetime.now())
-        pwd_last_set = pwd_last_set.replace(tzinfo=UTC)
-        if created or pwd_last_set >= user.password_change_date:
-            self.message(f"'{user.username}': Reset user's password")
-            self._logger.debug(
-                "Reset user's password",
-                user=user.username,
-                created=created,
-                pwd_last_set=pwd_last_set,
-            )
-            user.set_unusable_password()
-            user.save()
-
     def check_nsaccountlock(self, attributes: dict[str, Any], user: User):
         """https://www.port389.org/docs/389ds/howto/howto-account-inactivation.html"""
         # This is more of a 389-ds quirk rather than FreeIPA, but FreeIPA uses

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

@@ -1,7 +1,6 @@
 """Active Directory specific"""
 
 from collections.abc import Generator
-from datetime import UTC, datetime
 from enum import IntFlag
 from typing import Any
 
@@ -50,26 +49,9 @@
         yield None
 
     def sync(self, attributes: dict[str, Any], user: User, created: bool):
-        self.ms_check_pwd_last_set(attributes, user, created)
+        self.check_pwd_last_set("pwdLastSet", attributes, user, created)
         self.ms_check_uac(attributes, user)
 
-    def ms_check_pwd_last_set(self, attributes: dict[str, Any], user: User, created: bool):
-        """Check pwdLastSet"""
-        if "pwdLastSet" not in attributes:
-            return
-        pwd_last_set: datetime = attributes.get("pwdLastSet", datetime.now())
-        pwd_last_set = pwd_last_set.replace(tzinfo=UTC)
-        if created or pwd_last_set >= user.password_change_date:
-            self.message(f"'{user.username}': Reset user's password")
-            self._logger.debug(
-                "Reset user's password",
-                user=user.username,
-                created=created,
-                pwd_last_set=pwd_last_set,
-            )
-            user.set_unusable_password()
-            user.save()
-
     def ms_check_uac(self, attributes: dict[str, Any], user: User):
         """Check userAccountControl"""
         if "userAccountControl" not in attributes: