Delete all UserPasswordHistory records when UniquePasswordPolicy not …

…bound to anything

authentik/core/tasks.py

@@ -15,6 +15,7 @@
     AuthenticatedSession,
     ExpiringModel,
     User,
+    UserPasswordHistory,
 )
 from authentik.events.system_tasks import SystemTask, TaskStatus, prefill_task
 from authentik.lib.config import CONFIG
@@ -92,3 +93,20 @@ def clean_temporary_users(self: SystemTask):
             deleted_users += 1
     messages.append(f"Successfully deleted {deleted_users} users.")
     self.set_status(TaskStatus.SUCCESSFUL, *messages)
+
+
+@CELERY_APP.task(bind=True, base=SystemTask)
+@prefill_task
+def purge_password_history_table(self: SystemTask):
+    """Remove all entries from the core.models.UserPasswordHistory table"""
+    messages = []
+    try:
+        # n.b. a performance optimization to execute TRUNCATE
+        # instead of all().delete() would eliminate any FK checks.
+        UserPasswordHistory.objects.all().delete()
+    except Exception as err:
+        LOGGER.debug("Failed to purge core.models.UserPasswordHistory table.")
+        self.set_error(err)
+        return
+    messages.append("Successfully purged core.models.UserPasswordHistory")
+    self.set_status(TaskStatus.SUCCESSFUL, *messages)

authentik/core/tests/test_tasks.py

@@ -12,8 +12,13 @@
     Token,
     TokenIntents,
     User,
+    UserPasswordHistory,
+)
+from authentik.core.tasks import (
+    clean_expired_models,
+    clean_temporary_users,
+    purge_password_history_table,
 )
-from authentik.core.tasks import clean_expired_models, clean_temporary_users
 from authentik.core.tests.utils import create_test_admin_user
 from authentik.lib.generators import generate_id
 
@@ -49,3 +54,14 @@ def test_clean_temporary_users(self):
         )
         clean_temporary_users.delay().get()
         self.assertFalse(User.objects.filter(username=username))
+
+    def test_purge_password_history_table(self):
+        """Tests the task empties the core.models.UserPasswordHistory table"""
+        UserPasswordHistory.objects.bulk_create(
+            [
+                UserPasswordHistory(user=self.user, change={"old_password": "hunter1"}),
+                UserPasswordHistory(user=self.user, change={"old_password": "hunter2"}),
+            ]
+        )
+        purge_password_history_table.delay().get()
+        self.assertFalse(UserPasswordHistory.objects.all())

authentik/policies/signals.py

@@ -2,12 +2,13 @@
 
 from django.core.cache import cache
 from django.db import connection
-from django.db.models.signals import post_save
+from django.db.models.signals import post_delete, post_save
 from django.dispatch import receiver
 from structlog.stdlib import get_logger
 
 from authentik.core.api.applications import user_app_cache_key
 from authentik.core.models import Group, User
+from authentik.core.tasks import purge_password_history_table
 from authentik.policies.apps import GAUGE_POLICIES_CACHED
 from authentik.policies.models import Policy, PolicyBinding, PolicyBindingModel
 from authentik.policies.types import CACHE_PREFIX
@@ -42,3 +43,22 @@ def invalidate_policy_cache(sender, instance, **_):
     # Also delete user application cache
     keys = cache.keys(user_app_cache_key("*")) or []
     cache.delete_many(keys)
+
+
+@receiver(post_delete, sender=PolicyBinding)
+def purge_password_history(sender, instance, **_):
+    from authentik.policies.password.models import UniquePasswordPolicy
+
+    if not isinstance(instance.policy, UniquePasswordPolicy):
+        return
+
+    unique_password_policies = UniquePasswordPolicy.objects.all()
+
+    policy_binding_qs = PolicyBinding.objects.filter(policy__in=unique_password_policies).filter(
+        enabled=True
+    )
+
+    if policy_binding_qs.exists():
+        # No-op; At least one UniquePasswordPolicy binding still exists
+        return
+    purge_password_history_table.delay()