restructure UserSerializer.create and UserSerializer.update

Co-authored-by: Codex <codex@openai.com>

authentik/core/api/users.py

@@ -5,6 +5,7 @@ from json import loads
 from typing import Any
 
 from django.contrib.auth import update_session_auth_hash
+from django.contrib.auth.hashers import identify_hasher
 from django.contrib.auth.models import AnonymousUser, Permission
 from django.db.transaction import atomic
 from django.db.utils import IntegrityError
@@ -181,47 +182,86 @@ class UserSerializer(ModelSerializer):
         return RoleSerializer(instance.roles, many=True).data
 
     def __init__(self, *args, **kwargs):
+        """Setting password and permissions directly are allowed only in blueprints. The `create`
+        and `update` methods are adjusted accordingly."""
         super().__init__(*args, **kwargs)
         if SERIALIZER_CONTEXT_BLUEPRINT in self.context:
             self.fields["password"] = CharField(required=False, allow_null=True)
+            self.fields["password_hash"] = CharField(required=False, allow_null=True)
             self.fields["permissions"] = ListField(
                 required=False,
                 child=ChoiceField(choices=get_permission_choices()),
             )
 
     def create(self, validated_data: dict) -> User:
-        """If this serializer is used in the blueprint context, we allow for
-        directly setting a password. However should be done via the `set_password`
-        method instead of directly setting it like rest_framework."""
         password = validated_data.pop("password", None)
-        perms_qs = Permission.objects.filter(
-            codename__in=[x.split(".")[1] for x in validated_data.pop("permissions", [])]
-        ).values_list("content_type__app_label", "codename")
-        perms_list = [f"{ct}.{name}" for ct, name in list(perms_qs)]
+        password_hash = validated_data.pop("password_hash", None)
+        permissions = validated_data.pop("permissions", [])
+        if SERIALIZER_CONTEXT_BLUEPRINT in self.context:
+            self._validate_password_inputs(password, password_hash)
+
         instance: User = super().create(validated_data)
-        self._set_password(instance, password)
-        instance.assign_perms_to_managed_role(perms_list)
+
+        if SERIALIZER_CONTEXT_BLUEPRINT in self.context:
+            self._set_password(instance, password, password_hash)
+            perms_qs = Permission.objects.filter(
+                codename__in=[x.split(".")[1] for x in permissions]
+            ).values_list("content_type__app_label", "codename")
+            perms_list = [f"{ct}.{name}" for ct, name in list(perms_qs)]
+            instance.assign_perms_to_managed_role(perms_list)
+
+        self._ensure_password_not_empty(instance)
         return instance
 
     def update(self, instance: User, validated_data: dict) -> User:
-        """Same as `create` above, set the password directly if we're in a blueprint
-        context"""
         password = validated_data.pop("password", None)
-        perms_qs = Permission.objects.filter(
-            codename__in=[x.split(".")[1] for x in validated_data.pop("permissions", [])]
-        ).values_list("content_type__app_label", "codename")
-        perms_list = [f"{ct}.{name}" for ct, name in list(perms_qs)]
+        password_hash = validated_data.pop("password_hash", None)
+        permissions = validated_data.pop("permissions", [])
+        if SERIALIZER_CONTEXT_BLUEPRINT in self.context:
+            self._validate_password_inputs(password, password_hash)
+
         instance = super().update(instance, validated_data)
-        self._set_password(instance, password)
-        instance.assign_perms_to_managed_role(perms_list)
+
+        if SERIALIZER_CONTEXT_BLUEPRINT in self.context:
+            self._set_password(instance, password, password_hash)
+            perms_qs = Permission.objects.filter(
+                codename__in=[x.split(".")[1] for x in permissions]
+            ).values_list("content_type__app_label", "codename")
+            perms_list = [f"{ct}.{name}" for ct, name in list(perms_qs)]
+            instance.assign_perms_to_managed_role(perms_list)
+
+        self._ensure_password_not_empty(instance)
         return instance
 
-    def _set_password(self, instance: User, password: str | None):
-        """Set password of user if we're in a blueprint context, and if it's an empty
-        string then use an unusable password"""
-        if SERIALIZER_CONTEXT_BLUEPRINT in self.context and password:
+    def _validate_password_inputs(self, password: str | None, password_hash: str | None):
+        """Validate mutually-exclusive password inputs before any model mutation."""
+        if password is not None and password_hash is not None:
+            raise ValidationError(_("Cannot set both password and password_hash. Use only one."))
+        if password_hash is None:
+            return
+        try:
+            identify_hasher(password_hash)
+        except ValueError as exc:
+            LOGGER.warning("Failed to identify password hash format", exc_info=exc)
+            raise ValidationError(
+                _("Invalid password hash format. Must be a valid Django password hash.")
+            ) from exc
+
+    def _set_password(self, instance: User, password: str | None, password_hash: str | None = None):
+        """Set password from plain text or hash"""
+        if password_hash is not None:
+            instance.set_password_from_hash(password_hash)
+            instance.save()
+            return
+        elif password:
             instance.set_password(password)
             instance.save()
+            return
+
+    def _ensure_password_not_empty(self, instance: User):
+        """In principle there should be no issue with storing an empty string in the password field,
+        since Django will treat that as an unusable password. However, let's make it explicit just
+        to be extra sure."""
         if len(instance.password) == 0:
             instance.set_unusable_password()
             instance.save()
@@ -390,6 +430,16 @@ class UserPasswordSetSerializer(PassiveSerializer):
     password = CharField(required=True)
 
 
+class UserPasswordHashSetSerializer(PassiveSerializer):
+    """Payload to set a user's password from a pre-hashed Django password value.
+
+    This only updates authentik's stored password verifier and does not propagate
+    the change to LDAP or Kerberos password-sync integrations.
+    """
+
+    password_hash = CharField(required=True)
+
+
 class UserServiceAccountSerializer(PassiveSerializer):
     """Payload to create a service account"""
 
@@ -741,6 +791,51 @@ class UserViewSet(
             update_session_auth_hash(self.request, user)
         return Response(status=204)
 
+    @permission_required("authentik_core.reset_user_password")
+    @extend_schema(
+        request=UserPasswordHashSetSerializer,
+        responses={
+            204: OpenApiResponse(description="Successfully changed password"),
+            400: OpenApiResponse(description="Bad request"),
+        },
+    )
+    @action(
+        detail=True,
+        methods=["POST"],
+        permission_classes=[IsAuthenticated],
+    )
+    @validate(UserPasswordHashSetSerializer)
+    def set_password_hash(
+        self, request: Request, pk: int, body: UserPasswordHashSetSerializer
+    ) -> Response:
+        """Set a user's password from a pre-hashed Django password value.
+
+        This updates authentik's local password verifier only. It does not attempt
+        to propagate the password change to LDAP or Kerberos because no raw password
+        is available from the request payload.
+        """
+        user: User = self.get_object()
+        try:
+            user.set_password_from_hash(body.validated_data["password_hash"], request=request)
+            user.save()
+        except ValueError as exc:
+            LOGGER.debug("Failed to set password hash", exc=exc)
+            return Response(
+                data={
+                    "password_hash": [
+                        _("Invalid password hash format. Must be a valid Django password hash.")
+                    ]
+                },
+                status=400,
+            )
+        except (ValidationError, IntegrityError) as exc:
+            LOGGER.debug("Failed to set password hash", exc=exc)
+            return Response(status=400)
+        if user.pk == request.user.pk and SESSION_KEY_IMPERSONATE_USER not in self.request.session:
+            LOGGER.debug("Updating session hash after password change")
+            update_session_auth_hash(self.request, user)
+        return Response(status=204)
+
     @permission_required("authentik_core.reset_user_password")
     @extend_schema(
         request=UserRecoveryLinkSerializer,

authentik/core/management/commands/hash_password.py

@@ -0,0 +1,28 @@
+"""Hash password using Django's password hashers"""
+
+from django.contrib.auth.hashers import make_password
+from django.core.management.base import BaseCommand, CommandError
+
+
+class Command(BaseCommand):
+    """Hash a password using Django's password hashers"""
+
+    help = "Hash a password for use with AUTHENTIK_BOOTSTRAP_PASSWORD_HASH"
+
+    def add_arguments(self, parser):
+        parser.add_argument(
+            "password",
+            type=str,
+            help="Password to hash",
+        )
+
+    def handle(self, *args, **options):
+        password = options["password"]
+
+        if not password:
+            raise CommandError("Password cannot be empty")
+        try:
+            hashed = make_password(password)
+            self.stdout.write(hashed)
+        except ValueError as exc:
+            raise CommandError(f"Error hashing password: {exc}") from exc

authentik/core/models.py

@@ -10,7 +10,7 @@ from uuid import uuid4
 
 import pgtrigger
 from deepmerge import always_merger
-from django.contrib.auth.hashers import check_password
+from django.contrib.auth.hashers import check_password, identify_hasher
 from django.contrib.auth.models import AbstractUser, Permission
 from django.contrib.auth.models import UserManager as DjangoUserManager
 from django.contrib.sessions.base_session import AbstractBaseSession
@@ -556,10 +556,43 @@ class User(SerializerModel, AttributesMixin, AbstractUser):
 
             if not sender:
                 sender = self
-            password_changed.send(sender=sender, user=self, password=raw_password, request=request)
+            password_changed.send(
+                sender=sender,
+                user=self,
+                password=raw_password,
+                request=request,
+            )
         self.password_change_date = now()
         return super().set_password(raw_password)
 
+    def set_password_from_hash(self, password_hash: str, signal=True, sender=None, request=None):
+        """Set password directly from a pre-hashed value.
+
+        Unlike set_password(), this does not hash the input again. The provided value
+        must already be a valid Django password hash, and it is stored directly on the
+        user after validation.
+
+        Because no raw password is available, downstream password sync integrations
+        such as LDAP and Kerberos cannot be updated from this code path.
+
+        Raises ValueError if the hash format is not recognized.
+        """
+        identify_hasher(password_hash)  # Raises ValueError if invalid
+        if self.pk and signal:
+            from authentik.core.signals import password_changed
+
+            if not sender:
+                sender = self
+            password_changed.send(
+                sender=sender,
+                user=self,
+                password=None,
+                password_source="hash",
+                request=request,
+            )
+        self.password = password_hash
+        self.password_change_date = now()
+
     def check_password(self, raw_password: str) -> bool:
         """
         Return a boolean of whether the raw_password was correct. Handles

authentik/core/signals.py

@@ -22,7 +22,8 @@ from authentik.core.models import (
 from authentik.flows.apps import RefreshOtherFlowsAfterAuthentication
 from authentik.root.ws.consumer import build_device_group
 
-# Arguments: user: User, password: str
+# Arguments: user: User, password: str | None, password_source: str | None
+# request: HttpRequest | None
 password_changed = Signal()
 # Arguments: credentials: dict[str, any], request: HttpRequest,
 #            stage: Stage, context: dict[str, any]

authentik/core/tests/test_hash_password_command.py

@@ -0,0 +1,383 @@
+"""Tests for hash_password management command and password_hash serializer functionality"""
+
+from io import StringIO
+from unittest.mock import patch
+
+from django.contrib.auth.hashers import check_password, identify_hasher, make_password
+from django.core.management import call_command
+from django.core.management.base import CommandError
+from django.test import TestCase
+from rest_framework.exceptions import ValidationError
+
+from authentik.blueprints.v1.importer import SERIALIZER_CONTEXT_BLUEPRINT
+from authentik.core.api.users import UserSerializer
+from authentik.core.models import User
+from authentik.core.signals import password_changed
+from authentik.lib.generators import generate_id
+
+
+class TestHashPasswordCommand(TestCase):
+    """Test hash_password management command"""
+
+    def test_hash_password_basic(self):
+        """Test basic password hashing"""
+        out = StringIO()
+        call_command("hash_password", "test123", stdout=out)
+        hashed = out.getvalue().strip()
+
+        # Verify it's a valid hash
+        self.assertTrue(hashed.startswith("pbkdf2_sha256$"))
+        # Verify the hash can be validated
+        self.assertTrue(check_password("test123", hashed))
+
+    def test_hash_password_special_chars(self):
+        """Test hashing password with special characters"""
+        out = StringIO()
+        password = "P@ssw0rd!#$%^&*(){}[]"  # nosec
+        call_command("hash_password", password, stdout=out)
+        hashed = out.getvalue().strip()
+
+        self.assertTrue(hashed.startswith("pbkdf2_sha256$"))
+        self.assertTrue(check_password(password, hashed))
+
+    def test_hash_password_unicode(self):
+        """Test hashing password with unicode characters"""
+        out = StringIO()
+        password = "пароль123"  # nosec
+        call_command("hash_password", password, stdout=out)
+        hashed = out.getvalue().strip()
+
+        self.assertTrue(hashed.startswith("pbkdf2_sha256$"))
+        self.assertTrue(check_password(password, hashed))
+
+    def test_hash_password_long(self):
+        """Test hashing a very long password"""
+        out = StringIO()
+        password = "a" * 1000
+        call_command("hash_password", password, stdout=out)
+        hashed = out.getvalue().strip()
+
+        self.assertTrue(hashed.startswith("pbkdf2_sha256$"))
+        self.assertTrue(check_password(password, hashed))
+
+    def test_hash_password_spaces(self):
+        """Test hashing password with spaces"""
+        out = StringIO()
+        password = "my super secret password"  # nosec
+        call_command("hash_password", password, stdout=out)
+        hashed = out.getvalue().strip()
+
+        self.assertTrue(hashed.startswith("pbkdf2_sha256$"))
+        self.assertTrue(check_password(password, hashed))
+
+    def test_hash_password_empty_fails(self):
+        """Test that empty password raises error"""
+        with self.assertRaises(CommandError) as ctx:
+            call_command("hash_password", "")
+        self.assertIn("Password cannot be empty", str(ctx.exception))
+
+    def test_hash_different_passwords_different_hashes(self):
+        """Test that different passwords produce different hashes"""
+        out1 = StringIO()
+        out2 = StringIO()
+
+        call_command("hash_password", "password1", stdout=out1)
+        call_command("hash_password", "password2", stdout=out2)
+
+        hash1 = out1.getvalue().strip()
+        hash2 = out2.getvalue().strip()
+
+        self.assertNotEqual(hash1, hash2)
+        self.assertTrue(check_password("password1", hash1))
+        self.assertTrue(check_password("password2", hash2))
+        self.assertFalse(check_password("password1", hash2))
+        self.assertFalse(check_password("password2", hash1))
+
+    def test_hash_same_password_different_hashes(self):
+        """Test that same password produces different hashes (due to salt)"""
+        out1 = StringIO()
+        out2 = StringIO()
+
+        call_command("hash_password", "samepassword", stdout=out1)
+        call_command("hash_password", "samepassword", stdout=out2)
+
+        hash1 = out1.getvalue().strip()
+        hash2 = out2.getvalue().strip()
+
+        # Hashes should be different due to random salt
+        self.assertNotEqual(hash1, hash2)
+        # But both should validate the same password
+        self.assertTrue(check_password("samepassword", hash1))
+        self.assertTrue(check_password("samepassword", hash2))
+
+
+class TestUserSerializerPasswordHash(TestCase):
+    """Test UserSerializer password_hash functionality in blueprint context"""
+
+    def test_password_hash_sets_password_correctly(self):
+        """Test that a valid password_hash sets the password directly without re-hashing"""
+        password = "test-password-123"  # nosec
+        password_hash = make_password(password)
+
+        username = generate_id()
+        data = {
+            "username": username,
+            "name": "Test User",
+            "password_hash": password_hash,
+        }
+
+        serializer = UserSerializer(data=data, context={SERIALIZER_CONTEXT_BLUEPRINT: True})
+        self.assertTrue(serializer.is_valid(), serializer.errors)
+        user = serializer.save()
+
+        # Verify password was set correctly
+        self.assertTrue(user.check_password(password))
+        # Verify the hash was set directly (not re-hashed)
+        self.assertEqual(user.password, password_hash)
+
+    def test_password_hash_invalid_format_raises_error(self):
+        """Test that an invalid password_hash raises ValidationError"""
+        username = generate_id()
+        data = {
+            "username": username,
+            "name": "Test User",
+            "password_hash": "not-a-valid-hash",
+        }
+
+        serializer = UserSerializer(data=data, context={SERIALIZER_CONTEXT_BLUEPRINT: True})
+        self.assertTrue(serializer.is_valid(), serializer.errors)
+
+        with self.assertRaises(ValidationError) as ctx:
+            serializer.save()
+
+        self.assertIn("Invalid password hash format", str(ctx.exception))
+
+    def test_password_and_password_hash_both_set_raises_error(self):
+        """Test that setting both password and password_hash raises an error"""
+        plaintext_password = "plaintext-password"  # nosec
+        password_hash = make_password("hash-password")  # nosec
+
+        username = generate_id()
+        data = {
+            "username": username,
+            "name": "Test User",
+            "password": plaintext_password,
+            "password_hash": password_hash,
+        }
+
+        serializer = UserSerializer(data=data, context={SERIALIZER_CONTEXT_BLUEPRINT: True})
+        self.assertTrue(serializer.is_valid(), serializer.errors)
+
+        with self.assertRaises(ValidationError) as ctx:
+            serializer.save()
+
+        self.assertIn("Cannot set both password and password_hash", str(ctx.exception))
+
+    def test_password_change_date_updated_with_password_hash(self):
+        """Test that password_change_date is updated when using password_hash"""
+        password_hash = make_password("test-password")  # nosec
+
+        username = generate_id()
+        data = {
+            "username": username,
+            "name": "Test User",
+            "password_hash": password_hash,
+        }
+
+        serializer = UserSerializer(data=data, context={SERIALIZER_CONTEXT_BLUEPRINT: True})
+        self.assertTrue(serializer.is_valid(), serializer.errors)
+        user = serializer.save()
+
+        # Verify password_change_date is set
+        self.assertIsNotNone(user.password_change_date)
+
+    def test_password_hash_update_existing_user(self):
+        """Test that password_hash works when updating an existing user"""
+        # Create user first
+        user = User.objects.create(username=generate_id(), name="Test User")
+        user.set_password("old-password")  # nosec
+        user.save()
+
+        new_password = "new-password-123"  # nosec
+        password_hash = make_password(new_password)
+
+        data = {
+            "password_hash": password_hash,
+        }
+
+        serializer = UserSerializer(
+            instance=user,
+            data=data,
+            context={SERIALIZER_CONTEXT_BLUEPRINT: True},
+            partial=True,
+        )
+        self.assertTrue(serializer.is_valid(), serializer.errors)
+        updated_user = serializer.save()
+
+        # Verify password was updated
+        self.assertTrue(updated_user.check_password(new_password))
+        self.assertFalse(updated_user.check_password("old-password"))
+
+    def test_password_hash_whitespace_only_rejected(self):
+        """Test that whitespace-only password_hash is rejected by serializer validation"""
+        username = generate_id()
+        data = {
+            "username": username,
+            "name": "Test User",
+            "password_hash": "   ",  # whitespace only
+        }
+
+        serializer = UserSerializer(data=data, context={SERIALIZER_CONTEXT_BLUEPRINT: True})
+        # Serializer should reject blank values
+        self.assertFalse(serializer.is_valid())
+        self.assertIn("password_hash", serializer.errors)
+
+    def test_password_hash_empty_string_rejected(self):
+        """Test that empty string password_hash is rejected by serializer validation"""
+        username = generate_id()
+        data = {
+            "username": username,
+            "name": "Test User",
+            "password_hash": "",
+        }
+
+        serializer = UserSerializer(data=data, context={SERIALIZER_CONTEXT_BLUEPRINT: True})
+        # Serializer should reject blank values
+        self.assertFalse(serializer.is_valid())
+        self.assertIn("password_hash", serializer.errors)
+
+    def test_password_hash_null_creates_unusable_password(self):
+        """Test that null password_hash results in unusable password"""
+        username = generate_id()
+        data = {
+            "username": username,
+            "name": "Test User",
+            "password_hash": None,
+        }
+
+        serializer = UserSerializer(data=data, context={SERIALIZER_CONTEXT_BLUEPRINT: True})
+        self.assertTrue(serializer.is_valid(), serializer.errors)
+        user = serializer.save()
+
+        # User should have an unusable password since no valid password was provided
+        self.assertFalse(user.has_usable_password())
+
+    def test_both_password_and_password_hash_null_creates_unusable_password(self):
+        """Test that explicitly null password and password_hash results in unusable password"""
+        username = generate_id()
+        data = {
+            "username": username,
+            "name": "Test User",
+            "password": None,
+            "password_hash": None,
+        }
+
+        serializer = UserSerializer(data=data, context={SERIALIZER_CONTEXT_BLUEPRINT: True})
+        self.assertTrue(serializer.is_valid(), serializer.errors)
+        user = serializer.save()
+
+        # User should have an unusable password since no valid password was provided
+        self.assertFalse(user.has_usable_password())
+
+    def test_password_hash_not_available_outside_blueprint_context(self):
+        """Test that password_hash field is not available outside blueprint context"""
+        username = generate_id()
+        data = {
+            "username": username,
+            "name": "Test User",
+            "password_hash": make_password("test"),  # nosec
+        }
+
+        # Without blueprint context
+        serializer = UserSerializer(data=data)
+        self.assertTrue(serializer.is_valid(), serializer.errors)
+
+        # password_hash should not be in validated_data
+        self.assertNotIn("password_hash", serializer.validated_data)
+
+    def test_password_hash_identifies_various_hashers(self):
+        """Test that valid Django password hashes are accepted"""
+        password = "test-password"  # nosec
+        password_hash = make_password(password)
+
+        # Verify the hash is a valid Django hash format
+        hasher = identify_hasher(password_hash)
+        self.assertIsNotNone(hasher)
+
+        username = generate_id()
+        data = {
+            "username": username,
+            "name": "Test User",
+            "password_hash": password_hash,
+        }
+
+        serializer = UserSerializer(data=data, context={SERIALIZER_CONTEXT_BLUEPRINT: True})
+        self.assertTrue(serializer.is_valid(), serializer.errors)
+        user = serializer.save()
+
+        self.assertTrue(user.check_password(password))
+
+    def test_set_password_from_hash_signal_does_not_expose_raw_password(self):
+        """Test password_changed signal payload when password hash is set directly."""
+        user = User.objects.create(username=generate_id(), name="Test User")
+        user.set_password("old-password")  # nosec
+        user.save()
+        captured = []
+        dispatch_uid = generate_id()
+
+        def receiver(sender, **kwargs):
+            captured.append(kwargs)
+
+        password_changed.connect(receiver, dispatch_uid=dispatch_uid)
+        try:
+            user.set_password_from_hash(make_password("new-password"))  # nosec
+            user.save()
+        finally:
+            password_changed.disconnect(dispatch_uid=dispatch_uid)
+
+        self.assertEqual(len(captured), 1)
+        self.assertIsNone(captured[0]["password"])
+        self.assertEqual(captured[0]["password_source"], "hash")
+
+    def test_set_password_signal_marks_non_hash_password(self):
+        """Test password_changed signal payload for a regular password set."""
+        user = User.objects.create(username=generate_id(), name="Test User")
+        captured = []
+        dispatch_uid = generate_id()
+
+        def receiver(sender, **kwargs):
+            captured.append(kwargs)
+
+        password_changed.connect(receiver, dispatch_uid=dispatch_uid)
+        try:
+            user.set_password("new-password")  # nosec
+            user.save()
+        finally:
+            password_changed.disconnect(dispatch_uid=dispatch_uid)
+
+        self.assertEqual(len(captured), 1)
+        self.assertEqual(captured[0]["password"], "new-password")
+        self.assertNotIn("password_source", captured[0])
+
+    def test_set_password_from_hash_skips_ldap_and_kerberos_sync_receivers(self):
+        """Hash-based password sets should not hit LDAP/Kerberos sync paths."""
+        user = User.objects.create(
+            username=generate_id(),
+            name="Test User",
+            attributes={"distinguishedName": "cn=test,ou=users,dc=example,dc=com"},
+        )
+
+        with (
+            patch(
+                "authentik.sources.ldap.signals.LDAPSource.objects.filter"
+            ) as ldap_sources_filter,
+            patch(
+                "authentik.sources.kerberos.signals."
+                "UserKerberosSourceConnection.objects.select_related"
+            ) as kerberos_connections_select,
+        ):
+            user.set_password_from_hash(make_password("new-password"))  # nosec
+            user.save()
+
+        ldap_sources_filter.assert_not_called()
+        kerberos_connections_select.assert_not_called()

authentik/enterprise/providers/ssf/signals.py

@@ -85,13 +85,20 @@ def ssf_user_session_delete_session_revoked(sender, instance: AuthenticatedSessi
 
 
 @receiver(password_changed)
-def ssf_password_changed_cred_change(sender, user: User, password: str | None, **_):
+def ssf_password_changed_cred_change(
+    sender, user: User, password: str | None, password_source: str | None = None, **_
+):
     """Credential change trigger (password changed)"""
+    # set_password_from_hash() has no raw password but still represents an update.
+    if password_source == "hash":
+        change_type = "update"
+    else:
+        change_type = "revoke" if password is None else "update"
     send_ssf_events(
         EventTypes.CAEP_CREDENTIAL_CHANGE,
         {
             "credential_type": "password",
-            "change_type": "revoke" if password is None else "update",
+            "change_type": change_type,
         },
         sub_id={
             "format": "complex",

authentik/enterprise/providers/ssf/tests/test_signals.py

@@ -1,5 +1,6 @@
 from uuid import uuid4
 
+from django.contrib.auth.hashers import make_password
 from django.urls import reverse
 from rest_framework.test import APITestCase
 
@@ -93,6 +94,48 @@ class TestSignals(APITestCase):
         self.assertEqual(event.payload["sub_id"]["user"]["format"], "email")
         self.assertEqual(event.payload["sub_id"]["user"]["email"], user.email)
 
+    def test_signal_password_change_from_hash(self):
+        """Test user password change from a pre-hashed password."""
+        user = create_test_user()
+        self.client.force_login(user)
+        user.set_password_from_hash(make_password(generate_id()))
+        user.save()
+
+        stream = Stream.objects.filter(provider=self.provider).first()
+        self.assertIsNotNone(stream)
+        event = StreamEvent.objects.filter(stream=stream).first()
+        self.assertIsNotNone(event)
+        self.assertEqual(event.status, SSFEventStatus.PENDING_FAILED)
+        event_payload = event.payload["events"][
+            "https://schemas.openid.net/secevent/caep/event-type/credential-change"
+        ]
+        self.assertEqual(event_payload["change_type"], "update")
+        self.assertEqual(event_payload["credential_type"], "password")
+        self.assertEqual(event.payload["sub_id"]["format"], "complex")
+        self.assertEqual(event.payload["sub_id"]["user"]["format"], "email")
+        self.assertEqual(event.payload["sub_id"]["user"]["email"], user.email)
+
+    def test_signal_password_revoke(self):
+        """Test explicit password revoke."""
+        user = create_test_user()
+        self.client.force_login(user)
+        user.set_password(None)
+        user.save()
+
+        stream = Stream.objects.filter(provider=self.provider).first()
+        self.assertIsNotNone(stream)
+        event = StreamEvent.objects.filter(stream=stream).first()
+        self.assertIsNotNone(event)
+        self.assertEqual(event.status, SSFEventStatus.PENDING_FAILED)
+        event_payload = event.payload["events"][
+            "https://schemas.openid.net/secevent/caep/event-type/credential-change"
+        ]
+        self.assertEqual(event_payload["change_type"], "revoke")
+        self.assertEqual(event_payload["credential_type"], "password")
+        self.assertEqual(event.payload["sub_id"]["format"], "complex")
+        self.assertEqual(event.payload["sub_id"]["user"]["format"], "email")
+        self.assertEqual(event.payload["sub_id"]["user"]["email"], user.email)
+
     def test_signal_authenticator_added(self):
         """Test authenticator creation signal"""
         user = create_test_user()

authentik/events/signals.py

@@ -113,7 +113,7 @@ def on_invitation_used(sender, request: HttpRequest, invitation: Invitation, **_
 
 
 @receiver(password_changed)
-def on_password_changed(sender, user: User, password: str, request: HttpRequest | None, **_):
+def on_password_changed(sender, user: User, password: str | None, request: HttpRequest | None, **_):
     """Log password change"""
     Event.new(EventAction.PASSWORD_SET).from_http(request, user=user)
 

authentik/events/tests/test_event.py

@@ -2,6 +2,7 @@
 
 from urllib.parse import urlencode
 
+from django.contrib.auth.hashers import make_password
 from django.contrib.contenttypes.models import ContentType
 from django.test import RequestFactory, TestCase
 from django.views.debug import SafeExceptionReporterFilter
@@ -10,7 +11,7 @@ from guardian.shortcuts import get_anonymous_user
 from authentik.brands.models import Brand
 from authentik.core.models import Group, User
 from authentik.core.tests.utils import create_test_user
-from authentik.events.models import Event
+from authentik.events.models import Event, EventAction
 from authentik.flows.planner import PLAN_CONTEXT_PENDING_USER, FlowPlan
 from authentik.flows.views.executor import QS_QUERY, SESSION_KEY_PLAN
 from authentik.lib.generators import generate_id
@@ -213,3 +214,14 @@ class TestEvents(TestCase):
         event = Event.new("unittest", foo="foo bar \u0000 baz")
         event.save()
         self.assertEqual(event.context["foo"], "foo bar  baz")
+
+    def test_password_set_signal_on_set_password_from_hash(self):
+        """Changing password from hash should still emit an audit event."""
+        user = create_test_user()
+        old_count = Event.objects.filter(action=EventAction.PASSWORD_SET, user__pk=user.pk).count()
+
+        user.set_password_from_hash(make_password(generate_id()))
+        user.save()
+
+        new_count = Event.objects.filter(action=EventAction.PASSWORD_SET, user__pk=user.pk).count()
+        self.assertEqual(new_count, old_count + 1)

authentik/sources/kerberos/signals.py

@@ -17,8 +17,10 @@ LOGGER = get_logger()
 
 
 @receiver(password_changed)
-def kerberos_sync_password(sender, user: User, password: str, **_):
+def kerberos_sync_password(sender, user: User, password: str | None, **_):
     """Connect to kerberos and update password."""
+    if password is None:  # No raw password available (e.g. set via hash)
+        return
     user_source_connections = UserKerberosSourceConnection.objects.select_related(
         "source__kerberossource"
     ).filter(

authentik/sources/ldap/signals.py

@@ -37,8 +37,10 @@ def ldap_password_validate(sender, password: str, plan_context: dict[str, Any],
 
 
 @receiver(password_changed)
-def ldap_sync_password(sender, user: User, password: str, **_):
+def ldap_sync_password(sender, user: User, password: str | None, **_):
     """Connect to ldap and update password."""
+    if password is None:  # No raw password available (e.g. set via hash)
+        return
     sources = LDAPSource.objects.filter(sync_users_password=True, enabled=True)
     if not sources.exists():
         return

blueprints/schema.json

@@ -5532,6 +5532,14 @@
                     "minLength": 1,
                     "title": "Password"
                 },
+                "password_hash": {
+                    "type": [
+                        "string",
+                        "null"
+                    ],
+                    "minLength": 1,
+                    "title": "Password hash"
+                },
                 "permissions": {
                     "type": "array",
                     "items": {

blueprints/system/bootstrap.yaml

@@ -11,6 +11,7 @@ context:
   group_name: authentik Admins
   email: !Env [AUTHENTIK_BOOTSTRAP_EMAIL, "root@example.com"]
   password: !Env [AUTHENTIK_BOOTSTRAP_PASSWORD, null]
+  password_hash: !Env [AUTHENTIK_BOOTSTRAP_PASSWORD_HASH, null]
   token: !Env [AUTHENTIK_BOOTSTRAP_TOKEN, null]
 entries:
   - model: authentik_core.group
@@ -31,6 +32,7 @@ entries:
       groups:
         - !KeyOf admin-group
       password: !Context password
+      password_hash: !Context password_hash
   - model: authentik_core.token
     state: created
     conditions:

lifecycle/ak

@@ -87,7 +87,7 @@ elif [[ "$1" == "worker" ]]; then
     # If we have bootstrap credentials set, run bootstrap tasks outside of main server
     # sync, so that we can sure the first start actually has working bootstrap
     # credentials
-    if [[ -n "${AUTHENTIK_BOOTSTRAP_PASSWORD}" || -n "${AUTHENTIK_BOOTSTRAP_TOKEN}" ]]; then
+    if [[ -n "${AUTHENTIK_BOOTSTRAP_PASSWORD}" || -n "${AUTHENTIK_BOOTSTRAP_PASSWORD_HASH}" || -n "${AUTHENTIK_BOOTSTRAP_TOKEN}" ]]; then
         python -m manage apply_blueprint system/bootstrap.yaml || true
     fi
     check_if_root "python -m manage worker --pid-file ${TMPDIR}/authentik-worker.pid $@"

packages/client-ts/src/apis/CoreApi.ts

@@ -54,6 +54,7 @@ import type {
     User,
     UserAccountRequest,
     UserConsent,
+    UserPasswordHashSetRequest,
     UserPasswordSetRequest,
     UserPath,
     UserRecoveryEmailRequest,
@@ -104,6 +105,7 @@ import {
     UserAccountRequestToJSON,
     UserConsentFromJSON,
     UserFromJSON,
+    UserPasswordHashSetRequestToJSON,
     UserPasswordSetRequestToJSON,
     UserPathFromJSON,
     UserRecoveryEmailRequestToJSON,
@@ -508,6 +510,11 @@ export interface CoreUsersSetPasswordCreateRequest {
     userPasswordSetRequest: UserPasswordSetRequest;
 }
 
+export interface CoreUsersSetPasswordHashCreateRequest {
+    id: number;
+    userPasswordHashSetRequest: UserPasswordHashSetRequest;
+}
+
 export interface CoreUsersUpdateRequest {
     id: number;
     userRequest: UserRequest;
@@ -5288,6 +5295,77 @@ export class CoreApi extends runtime.BaseAPI {
         await this.coreUsersSetPasswordCreateRaw(requestParameters, initOverrides);
     }
 
+    /**
+     * Creates request options for coreUsersSetPasswordHashCreate without sending the request
+     */
+    async coreUsersSetPasswordHashCreateRequestOpts(
+        requestParameters: CoreUsersSetPasswordHashCreateRequest,
+    ): Promise<runtime.RequestOpts> {
+        if (requestParameters["id"] == null) {
+            throw new runtime.RequiredError(
+                "id",
+                'Required parameter "id" was null or undefined when calling coreUsersSetPasswordHashCreate().',
+            );
+        }
+
+        if (requestParameters["userPasswordHashSetRequest"] == null) {
+            throw new runtime.RequiredError(
+                "userPasswordHashSetRequest",
+                'Required parameter "userPasswordHashSetRequest" was null or undefined when calling coreUsersSetPasswordHashCreate().',
+            );
+        }
+
+        const queryParameters: any = {};
+
+        const headerParameters: runtime.HTTPHeaders = {};
+
+        headerParameters["Content-Type"] = "application/json";
+
+        if (this.configuration && this.configuration.accessToken) {
+            const token = this.configuration.accessToken;
+            const tokenString = await token("authentik", []);
+
+            if (tokenString) {
+                headerParameters["Authorization"] = `Bearer ${tokenString}`;
+            }
+        }
+
+        let urlPath = `/core/users/{id}/set_password_hash/`;
+        urlPath = urlPath.replace(`{${"id"}}`, encodeURIComponent(String(requestParameters["id"])));
+
+        return {
+            path: urlPath,
+            method: "POST",
+            headers: headerParameters,
+            query: queryParameters,
+            body: UserPasswordHashSetRequestToJSON(requestParameters["userPasswordHashSetRequest"]),
+        };
+    }
+
+    /**
+     * Set a user\'s password from a pre-hashed Django password value.  This updates authentik\'s local password verifier only. It does not attempt to propagate the password change to LDAP or Kerberos because no raw password is available from the request payload.
+     */
+    async coreUsersSetPasswordHashCreateRaw(
+        requestParameters: CoreUsersSetPasswordHashCreateRequest,
+        initOverrides?: RequestInit | runtime.InitOverrideFunction,
+    ): Promise<runtime.ApiResponse<void>> {
+        const requestOptions =
+            await this.coreUsersSetPasswordHashCreateRequestOpts(requestParameters);
+        const response = await this.request(requestOptions, initOverrides);
+
+        return new runtime.VoidApiResponse(response);
+    }
+
+    /**
+     * Set a user\'s password from a pre-hashed Django password value.  This updates authentik\'s local password verifier only. It does not attempt to propagate the password change to LDAP or Kerberos because no raw password is available from the request payload.
+     */
+    async coreUsersSetPasswordHashCreate(
+        requestParameters: CoreUsersSetPasswordHashCreateRequest,
+        initOverrides?: RequestInit | runtime.InitOverrideFunction,
+    ): Promise<void> {
+        await this.coreUsersSetPasswordHashCreateRaw(requestParameters, initOverrides);
+    }
+
     /**
      * Creates request options for coreUsersUpdate without sending the request
      */

packages/client-ts/src/models/UserPasswordHashSetRequest.ts

@@ -0,0 +1,73 @@
+/* tslint:disable */
+/* eslint-disable */
+/**
+ * authentik
+ * Making authentication simple.
+ *
+ * The version of the OpenAPI document: 2026.5.0-rc1
+ * Contact: hello@goauthentik.io
+ *
+ * NOTE: This class is auto generated by OpenAPI Generator (https://openapi-generator.tech).
+ * https://openapi-generator.tech
+ * Do not edit the class manually.
+ */
+
+/**
+ * Payload to set a user's password from a pre-hashed Django password value.
+ *
+ * This only updates authentik's stored password verifier and does not propagate
+ * the change to LDAP or Kerberos password-sync integrations.
+ * @export
+ * @interface UserPasswordHashSetRequest
+ */
+export interface UserPasswordHashSetRequest {
+    /**
+     *
+     * @type {string}
+     * @memberof UserPasswordHashSetRequest
+     */
+    passwordHash: string;
+}
+
+/**
+ * Check if a given object implements the UserPasswordHashSetRequest interface.
+ */
+export function instanceOfUserPasswordHashSetRequest(
+    value: object,
+): value is UserPasswordHashSetRequest {
+    if (!("passwordHash" in value) || value["passwordHash"] === undefined) return false;
+    return true;
+}
+
+export function UserPasswordHashSetRequestFromJSON(json: any): UserPasswordHashSetRequest {
+    return UserPasswordHashSetRequestFromJSONTyped(json, false);
+}
+
+export function UserPasswordHashSetRequestFromJSONTyped(
+    json: any,
+    ignoreDiscriminator: boolean,
+): UserPasswordHashSetRequest {
+    if (json == null) {
+        return json;
+    }
+    return {
+        passwordHash: json["password_hash"],
+    };
+}
+
+export function UserPasswordHashSetRequestToJSON(json: any): UserPasswordHashSetRequest {
+    return UserPasswordHashSetRequestToJSONTyped(json, false);
+}
+
+export function UserPasswordHashSetRequestToJSONTyped(
+    value?: UserPasswordHashSetRequest | null,
+    ignoreDiscriminator: boolean = false,
+): any {
+    if (value == null) {
+        return value;
+    }
+
+    return {
+        password_hash: value["passwordHash"],
+    };
+}

packages/client-ts/src/models/index.ts

@@ -837,6 +837,7 @@ export * from "./UserLogoutStageRequest";
 export * from "./UserMatchingModeEnum";
 export * from "./UserOAuthSourceConnection";
 export * from "./UserOAuthSourceConnectionRequest";
+export * from "./UserPasswordHashSetRequest";
 export * from "./UserPasswordSetRequest";
 export * from "./UserPath";
 export * from "./UserPlexSourceConnection";

schema.yml

@@ -4522,6 +4522,39 @@ paths:
           description: Bad request
         '403':
           $ref: '#/components/responses/GenericErrorResponse'
+  /core/users/{id}/set_password_hash/:
+    post:
+      operationId: core_users_set_password_hash_create
+      description: |-
+        Set a user's password from a pre-hashed Django password value.
+
+        This updates authentik's local password verifier only. It does not attempt
+        to propagate the password change to LDAP or Kerberos because no raw password
+        is available from the request payload.
+      parameters:
+      - in: path
+        name: id
+        schema:
+          type: integer
+        description: A unique integer value identifying this User.
+        required: true
+      tags:
+      - core
+      requestBody:
+        content:
+          application/json:
+            schema:
+              $ref: '#/components/schemas/UserPasswordHashSetRequest'
+        required: true
+      security:
+      - authentik: []
+      responses:
+        '204':
+          description: Successfully changed password
+        '400':
+          description: Bad request
+        '403':
+          $ref: '#/components/responses/GenericErrorResponse'
   /core/users/{id}/used_by/:
     get:
       operationId: core_users_used_by_list
@@ -57270,6 +57303,19 @@ components:
       - identifier
       - source
       - user
+    UserPasswordHashSetRequest:
+      type: object
+      description: |-
+        Payload to set a user's password from a pre-hashed Django password value.
+
+        This only updates authentik's stored password verifier and does not propagate
+        the change to LDAP or Kerberos password-sync integrations.
+      properties:
+        password_hash:
+          type: string
+          minLength: 1
+      required:
+      - password_hash
     UserPasswordSetRequest:
       type: object
       description: Payload to set a users' password directly

web/src/admin/users/UserPasswordHashForm.ts

@@ -0,0 +1,59 @@
+import "#elements/forms/HorizontalFormElement";
+
+import { DEFAULT_CONFIG } from "#common/api/config";
+
+import { Form } from "#elements/forms/Form";
+
+import { CoreApi, UserPasswordHashSetRequest } from "@goauthentik/api";
+
+import { msg } from "@lit/localize";
+import { html, TemplateResult } from "lit";
+import { customElement, property } from "lit/decorators.js";
+
+@customElement("ak-user-password-hash-form")
+export class UserPasswordHashForm extends Form<UserPasswordHashSetRequest> {
+    public override submitLabel = msg("Set Password");
+
+    @property({ type: Number })
+    public instancePk?: number;
+
+    public override getSuccessMessage(): string {
+        return msg("Successfully updated password.");
+    }
+
+    protected override async send(data: UserPasswordHashSetRequest): Promise<void> {
+        return new CoreApi(DEFAULT_CONFIG).coreUsersSetPasswordHashCreate({
+            id: this.instancePk || 0,
+            userPasswordHashSetRequest: data,
+        });
+    }
+
+    protected override renderForm(): TemplateResult {
+        return html`
+            <ak-form-element-horizontal label=${msg("Password hash")} required name="passwordHash">
+                <input
+                    type="text"
+                    value=""
+                    class="pf-c-form-control"
+                    required
+                    placeholder=${msg("pbkdf2_sha256$...")}
+                    aria-label=${msg("Password hash")}
+                />
+                <p class="pf-c-form__helper-text">
+                    ${msg("Enter a pre-hashed password (e.g. pbkdf2_sha256$iterations$salt$hash).")}
+                </p>
+                <p class="pf-c-form__helper-text">
+                    ${msg(
+                        "Warning: Password hashes set here are not synced back to LDAP or Kerberos.",
+                    )}
+                </p>
+            </ak-form-element-horizontal>
+        `;
+    }
+}
+
+declare global {
+    interface HTMLElementTagNameMap {
+        "ak-user-password-hash-form": UserPasswordHashForm;
+    }
+}

web/src/admin/users/recovery.ts

@@ -1,7 +1,9 @@
 import { modalInvoker } from "#elements/dialogs";
 import { LitFC } from "#elements/types";
+import { getUserDisplayName } from "#elements/user/utils";
 
 import { UserPasswordForm } from "#admin/users/UserPasswordForm";
+import { UserPasswordHashForm } from "#admin/users/UserPasswordHashForm";
 import { UserRecoveryLinkForm } from "#admin/users/UserRecoveryLinkForm";
 import { UserResetEmailForm } from "#admin/users/UserResetEmailForm";
 
@@ -52,7 +54,7 @@ export const RecoveryButtons: LitFC<RecoveryButtonsProps> = ({
             class="pf-c-button pf-m-secondary ${buttonClasses || ""}"
             type="button"
             ${modalInvoker(UserPasswordForm, {
-                headline: msg(str`Update ${user.name || user.username}'s password`),
+                headline: msg(str`Update ${getUserDisplayName(user)}'s password`),
                 username: user.username,
                 email: user.email,
                 instancePk: user.pk,
@@ -60,6 +62,16 @@ export const RecoveryButtons: LitFC<RecoveryButtonsProps> = ({
         >
             ${msg("Set password")}
         </button>`,
+        html`<button
+            class="pf-c-button pf-m-secondary ${buttonClasses || ""}"
+            type="button"
+            ${modalInvoker(UserPasswordHashForm, {
+                headline: msg(str`Update ${getUserDisplayName(user)}'s password`),
+                instancePk: user.pk,
+            })}
+        >
+            ${msg("Set password hash")}
+        </button>`,
         ...recoveryModals,
     ];
 };

website/docs/install-config/automated-install.mdx

@@ -8,9 +8,48 @@ To install authentik automatically (skipping the Out-of-box experience), you can
 These can't be defined using the file-based syntax (`file://`), so you can't pass them in as secrets in a Docker Compose installation.
 :::
 
+### `AUTHENTIK_BOOTSTRAP_PASSWORD_HASH`
+
+Configure the default password for the `akadmin` user using a pre-hashed password. Only read on the first startup.
+
+This updates authentik's stored password hash directly. It does not propagate the password to LDAP or Kerberos integrations.
+
+To generate a hash, run this command before your initial deployment:
+
+```bash
+docker compose run --rm server hash_password 'your-password'
+```
+
+:::warning Escaping `$` in Docker Compose
+Password hashes contain `$` characters which Docker Compose interprets as variable references.
+
+**In `.env` files**, use single quotes to prevent interpolation:
+
+```bash
+AUTHENTIK_BOOTSTRAP_PASSWORD_HASH='pbkdf2_sha256$1000000$xKKFuYtJEE27km09BD49x2$4+Z6j3utmouPF5mik0Z21L2P0og5IlmMmIJ46Tj3zCM='
+```
+
+**In `docker-compose.yml`** (inline environment), escape each `$` with `$$`:
+
+```yaml
+services:
+    worker:
+        environment:
+            AUTHENTIK_BOOTSTRAP_PASSWORD_HASH: "pbkdf2_sha256$$1000000$$xKKFuYtJEE27km09BD49x2$$4+Z6j3utmouPF5mik0Z21L2P0og5IlmMmIJ46Tj3zCM="
+```
+
+See the Docker Compose documentation on [`.env` file interpolation](https://docs.docker.com/compose/how-tos/environment-variables/variable-interpolation/) and [Compose file interpolation](https://docs.docker.com/reference/compose-file/interpolation/) for details.
+:::
+
 ### `AUTHENTIK_BOOTSTRAP_PASSWORD`
 
-Configure the default password for the `akadmin` user. Only read on the first startup. Can be used for any flow executor.
+:::warning
+This option stores plaintext passwords in environment variables. Use [`AUTHENTIK_BOOTSTRAP_PASSWORD_HASH`](#authentik_bootstrap_password_hash) instead.
+:::
+
+Configure the default password for the `akadmin` user. Only read on the first startup.
+
+Setting both `AUTHENTIK_BOOTSTRAP_PASSWORD` and `AUTHENTIK_BOOTSTRAP_PASSWORD_HASH` will result in an error.
 
 ### `AUTHENTIK_BOOTSTRAP_TOKEN`
 
@@ -22,15 +61,23 @@ Set the email address for the default `akadmin` user.
 
 ## Kubernetes
 
-In the Helm values, set the `akadmin` user password and token:
+In the Helm values, set the `akadmin` user password hash and token:
 
 ```yaml
 authentik:
-    bootstrap_token: test
-    bootstrap_password: test
+    bootstrap_password_hash: "pbkdf2_sha256$1000000$xKKFuYtJEE27km09BD49x2$4+Z6j3utmouPF5mik0Z21L2P0og5IlmMmIJ46Tj3zCM="
+    bootstrap_token: "your-token-here"
+    bootstrap_email: "admin@authentik.company"
 ```
 
-To store the password and token in a secret, use:
+:::note Helm escaping
+When using password hashes in quoted YAML strings as shown above, no escaping of `$` characters is required. The `$` character only needs escaping when:
+
+- Using Helm templating syntax (e.g., `{{ .Values.something }}`) where `$` has special meaning
+- Referencing values from environment variable substitution in your values file
+  :::
+
+Or store the password hash in a secret and reference it via `envFrom`:
 
 ```yaml
 global:
@@ -39,4 +86,4 @@ global:
               name: _some-secret_
 ```
 
-where _some-secret_ contains the environment variables as in the documentation above.
+where _some-secret_ contains the environment variables as documented above.