crypto: add managed secrets API

authentik/blueprints/api.py

@@ -3,8 +3,10 @@
 from json import JSONDecodeError, loads
 from typing import cast
 
+from django.apps import apps
 from django.conf import settings
 from django.core.files.uploadedfile import InMemoryUploadedFile
+from django.db.models import Model
 from django.utils.translation import gettext_lazy as _
 from drf_spectacular.utils import extend_schema, inline_serializer
 from rest_framework.decorators import action
@@ -23,7 +25,7 @@
 
 from authentik.api.validation import validate
 from authentik.blueprints.models import BlueprintInstance
-from authentik.blueprints.v1.common import Blueprint, EntryInvalidError
+from authentik.blueprints.v1.common import Blueprint, BlueprintEntry, EntryInvalidError, YAMLTag
 from authentik.blueprints.v1.importer import Importer
 from authentik.blueprints.v1.oci import OCI_PREFIX
 from authentik.blueprints.v1.tasks import apply_blueprint, blueprints_find_dict
@@ -159,6 +161,28 @@ def check_blueprint_perms(blueprint: Blueprint, user: User, explicit_action: str
                         )
                     }
                 )
+        check_secret_references(entry, apps.get_model(full_model), user)
+
+
+def check_secret_references(entry: BlueprintEntry, model: type[Model], user: User):
+    """Attaching a secret can disclose it through the object, like in the API.
+
+    Serializers only check this with a request, and blueprints are imported without one.
+    """
+    from authentik.crypto.secrets.models import Secret
+
+    for field in model._meta.fields:
+        if field.related_model is not Secret:
+            continue
+        value = entry.attrs.get(field.name)
+        # References to secrets created by the same blueprint need permission to create them.
+        if value is None or isinstance(value, YAMLTag):
+            continue
+        secret = Secret.objects.filter(pk=getattr(value, "pk", value)).first()
+        if secret and not user.has_perm("authentik_crypto_secrets.view_secret_value", secret):
+            raise PermissionDenied(
+                {entry.id: _("User lacks permission to use secret {secret}").format(secret=secret)}
+            )
 
 
 class BlueprintInstanceViewSet(UsedByMixin, ModelViewSet):

authentik/core/api/used_by.py

@@ -4,7 +4,7 @@
 from inspect import getmembers
 
 from django.db.models.base import Model
-from django.db.models.deletion import SET_DEFAULT, SET_NULL
+from django.db.models.deletion import PROTECT, SET_DEFAULT, SET_NULL
 from django.db.models.manager import Manager
 from drf_spectacular.utils import extend_schema
 from guardian.shortcuts import get_objects_for_user
@@ -24,6 +24,7 @@ class DeleteAction(Enum):
     CASCADE_MANY = "cascade_many"
     SET_NULL = "set_null"
     SET_DEFAULT = "set_default"
+    PROTECT = "protect"
     LEFT_DANGLING = "left_dangling"
 
 
@@ -44,6 +45,8 @@ def get_delete_action(manager: Manager) -> str:
             return DeleteAction.SET_NULL.value
         if manager.field.remote_field.on_delete.__name__ == SET_DEFAULT.__name__:
             return DeleteAction.SET_DEFAULT.value
+        if manager.field.remote_field.on_delete.__name__ == PROTECT.__name__:
+            return DeleteAction.PROTECT.value
     if hasattr(manager, "source_field"):
         return DeleteAction.CASCADE_MANY.value
     return DeleteAction.CASCADE.value

authentik/core/api/utils.py

@@ -4,6 +4,7 @@
 
 from django.db import models
 from django.db.models import Model
+from django.utils.translation import gettext_lazy as _
 from drf_spectacular.extensions import OpenApiSerializerFieldExtension
 from drf_spectacular.plumbing import build_basic_type
 from drf_spectacular.types import OpenApiTypes
@@ -66,6 +67,23 @@ def to_representation(self, instance: Model):
             data.pop(field_name, None)
         return data
 
+    def to_internal_value(self, data):
+        from authentik.crypto.secrets.api import SecretReferenceField
+
+        # Credentials moved from their own fields to secret references. Reject the old field
+        # names instead of ignoring them, so existing blueprints and API clients fail loudly.
+        # Remove with the legacy credential columns in 2027.2.
+        if hasattr(data, "keys"):
+            legacy = {
+                name: [_("Use %(field)s to reference a secret instead.") % {"field": f"{name}_ref"}]
+                for name in data.keys()
+                if name not in self.fields
+                and isinstance(self.fields.get(f"{name}_ref"), SecretReferenceField)
+            }
+            if legacy:
+                raise ValidationError(legacy)
+        return super().to_internal_value(data)
+
     def update(self, instance: Model, validated_data):
         raise_errors_on_nested_writes("update", self, validated_data)
         info = model_meta.get_field_info(instance)
@@ -107,7 +125,18 @@ def build_relational_field(self, field_name, relation_info):
         ):
             relation_info = relation_info._replace(has_through_model=False)
 
-        return super().build_relational_field(field_name, relation_info)
+        field_class, field_kwargs = super().build_relational_field(field_name, relation_info)
+        if relation_info.related_model._meta.label_lower == "authentik_crypto_secrets.secret":
+            from authentik.crypto.secrets.api import SecretReferenceField
+
+            # The model field says whether a secret is required, optional, or generated when
+            # the object is created: generated references are blank, but not null.
+            model_field = relation_info.model_field
+            field_class = SecretReferenceField
+            field_kwargs["required"] = not model_field.blank
+            field_kwargs["allow_null"] = model_field.blank
+            field_kwargs["required_on_update"] = not model_field.null
+        return field_class, field_kwargs
 
 
 class PassiveSerializer(Serializer):

authentik/core/tests/test_used_by.py

@@ -0,0 +1,27 @@
+"""Test used_by delete actions"""
+
+from django.urls import reverse
+from rest_framework.test import APITestCase
+
+from authentik.core.api.used_by import DeleteAction
+from authentik.core.tests.utils import create_test_admin_user
+from authentik.lib.generators import generate_id
+from authentik.stages.authenticator_duo.models import AuthenticatorDuoStage, DuoDevice
+
+
+class TestUsedBy(APITestCase):
+    def test_protected_relation(self):
+        """Objects that block the deletion aren't reported as deleted with it"""
+        user = create_test_admin_user()
+        stage = AuthenticatorDuoStage.objects.create(
+            name=generate_id(), client_id=generate_id(), api_hostname=generate_id()
+        )
+        DuoDevice.objects.create(user=user, stage=stage, duo_user_id=generate_id())
+        self.client.force_login(user)
+        response = self.client.get(
+            reverse("authentik_api:authenticatorduostage-used-by", kwargs={"pk": stage.pk})
+        )
+        self.assertEqual(response.status_code, 200, response.content)
+        self.assertEqual(
+            [entry["action"] for entry in response.json()], [DeleteAction.PROTECT.value]
+        )

authentik/crypto/secrets/__init__.py

@@ -0,0 +1 @@
+"""Managed secrets."""

authentik/crypto/secrets/api.py

@@ -0,0 +1,202 @@
+"""Managed secrets API."""
+
+from django.core.exceptions import ValidationError as DjangoValidationError
+from django.db import transaction
+from django.db.models.deletion import ProtectedError
+from django.utils.translation import gettext_lazy as _
+from drf_spectacular.utils import extend_schema
+from rest_framework.decorators import action
+from rest_framework.exceptions import PermissionDenied, ValidationError
+from rest_framework.fields import CharField, IntegerField, SkipField, empty
+from rest_framework.permissions import IsAuthenticated
+from rest_framework.relations import PrimaryKeyRelatedField
+from rest_framework.request import Request
+from rest_framework.response import Response
+from rest_framework.viewsets import ModelViewSet
+
+from authentik.blueprints.api import ManagedSerializer
+from authentik.core.api.used_by import UsedByMixin
+from authentik.core.api.utils import ModelSerializer, PassiveSerializer
+from authentik.crypto.secrets.models import Secret, SecretType, generate_secret_value
+from authentik.events.models import Event, EventAction
+from authentik.rbac.decorators import permission_required
+
+# Long enough for any credential, short enough not to be abused to store large values.
+MAX_GENERATED_LENGTH = 1024
+
+
+class SecretReferenceField(PrimaryKeyRelatedField):
+    """Attaching a credential can disclose it through the consumer."""
+
+    def __init__(self, allowed_types=(SecretType.TEXT,), required_on_update=False, **kwargs):
+        self.allowed_types = allowed_types
+        self.required_on_update = required_on_update
+        super().__init__(**kwargs)
+
+    def run_validation(self, data=empty):
+        # Objects that generate their secret may omit it when created, but must keep one.
+        if data in (None, "") and self.required_on_update and self.parent.instance:
+            raise ValidationError(_("A secret is required."))
+        return super().run_validation(data)
+
+    def to_internal_value(self, data):
+        secret = super().to_internal_value(data)
+        request = self.context.get("request")
+        instance = self.parent.instance
+        unchanged = instance and getattr(instance, f"{self.source}_id") == secret.pk
+        if (
+            request
+            and not unchanged
+            and not request.user.has_perm("authentik_crypto_secrets.view_secret_value", secret)
+        ):
+            raise PermissionDenied(_("You do not have permission to use this secret."))
+        if secret.type not in self.allowed_types:
+            raise ValidationError(_("This secret type is not supported by this field."))
+        return secret
+
+
+class SecretSerializer(ManagedSerializer, ModelSerializer):
+    """Create and configure a secret without exposing its value."""
+
+    value = CharField(
+        source="secret_value",
+        write_only=True,
+        required=False,
+        allow_blank=True,
+        trim_whitespace=False,
+    )
+    length = IntegerField(
+        write_only=True,
+        required=False,
+        min_value=1,
+        max_value=MAX_GENERATED_LENGTH,
+        help_text=_(
+            "Length of the generated value when creating a text secret without a value. "
+            "Defaults to the default token length."
+        ),
+    )
+
+    def validate_value(self, value: str) -> str:
+        if value == "":
+            raise SkipField
+        instance = self.instance
+        if not instance:
+            return value
+        request = self.context.get("request")
+        if request and not request.user.has_perm(
+            "authentik_crypto_secrets.rotate_secret", instance
+        ):
+            raise PermissionDenied(_("You do not have permission to replace this value."))
+        return value
+
+    def validate(self, attrs: dict) -> dict:
+        instance = self.instance
+        if instance and attrs.get("type", instance.type) != instance.type:
+            raise ValidationError({"type": _("Type cannot be changed after creation.")})
+        secret_type = attrs.get("type", instance.type if instance else SecretType.TEXT)
+        if not instance and secret_type != SecretType.TEXT and not attrs.get("secret_value"):
+            raise ValidationError({"value": _("A value is required for this type.")})
+        if instance:
+            # Re-applying a blueprint passes the length again, but it only applies when
+            # the secret is created.
+            attrs.pop("length", None)
+        elif "length" in attrs and (secret_type != SecretType.TEXT or attrs.get("secret_value")):
+            raise ValidationError(
+                {"length": _("Length only applies when generating a new text secret.")}
+            )
+        if "secret_value" in attrs:
+            try:
+                (instance or Secret(type=secret_type)).validate_value(attrs["secret_value"])
+            except DjangoValidationError as exc:
+                raise ValidationError({"value": exc.messages}) from exc
+        return attrs
+
+    def create(self, validated_data: dict) -> Secret:
+        if length := validated_data.pop("length", None):
+            validated_data["secret_value"] = generate_secret_value(length)
+        return super().create(validated_data)
+
+    def update(self, instance: Secret, validated_data: dict) -> Secret:
+        value = validated_data.pop("secret_value", None)
+        with transaction.atomic():
+            if validated_data:
+                for field, field_value in validated_data.items():
+                    setattr(instance, field, field_value)
+                instance.save(update_fields=[*validated_data, "last_updated"])
+            if value is not None:
+                instance.replace_value(value, self.context.get("request"))
+        return instance
+
+    class Meta:
+        model = Secret
+        fields = ["pk", "name", "type", "managed", "value", "length", "created", "last_updated"]
+        extra_kwargs = {
+            "managed": {"read_only": True},
+            "created": {"read_only": True},
+            "last_updated": {"read_only": True},
+        }
+
+
+class SecretValueSerializer(PassiveSerializer):
+    """A secret value."""
+
+    value = CharField(read_only=True)
+
+
+class RotatedSecretSerializer(PassiveSerializer):
+    """A rotated value, hidden when the caller cannot view it."""
+
+    value = CharField(read_only=True, allow_null=True)
+
+
+class SecretViewSet(UsedByMixin, ModelViewSet):
+    """Create and manage named credentials referenced by authentik configuration objects.
+
+    List and detail responses contain metadata only. Reading a value requires the
+    view_value action and its separate permission. Replacing or manually rotating
+    a value requires rotate_secret permission and records an audit event.
+    Rotation changes the value stored in authentik; it does not update external systems.
+    """
+
+    queryset = Secret.objects.all()
+    serializer_class = SecretSerializer
+    ordering = ["name"]
+    search_fields = ["name"]
+    filterset_fields = {"name": ["exact"], "type": ["exact", "in"], "managed": ["exact"]}
+
+    def destroy(self, request: Request, *args, **kwargs) -> Response:
+        try:
+            return super().destroy(request, *args, **kwargs)
+        except ProtectedError:
+            raise ValidationError(
+                _("Secret is in use. Remove it from the objects referencing it first.")
+            ) from None
+
+    @permission_required("authentik_crypto_secrets.view_secret_value")
+    @extend_schema(responses={200: SecretValueSerializer})
+    @action(detail=True, pagination_class=None, permission_classes=[IsAuthenticated])
+    def view_value(self, request: Request, pk: str) -> Response:
+        """Return and audit a secret value."""
+        secret = self.get_object()
+        Event.new(EventAction.SECRET_VIEW, secret=secret).from_http(request)
+        return Response(SecretValueSerializer({"value": secret.secret_value}).data)
+
+    @permission_required("authentik_crypto_secrets.rotate_secret")
+    @extend_schema(request=None, responses={200: RotatedSecretSerializer})
+    @action(
+        detail=True,
+        methods=["POST"],
+        pagination_class=None,
+        permission_classes=[IsAuthenticated],
+    )
+    def rotate(self, request: Request, pk: str) -> Response:
+        """Replace a text secret with a generated value."""
+        secret = self.get_object()
+        if secret.type != SecretType.TEXT:
+            raise ValidationError({"non_field_errors": [_("Only text secrets can be rotated.")]})
+        try:
+            value = secret.rotate(request)
+        except DjangoValidationError as exc:
+            raise ValidationError({"non_field_errors": exc.messages}) from exc
+        can_view = request.user.has_perm("authentik_crypto_secrets.view_secret_value", secret)
+        return Response(RotatedSecretSerializer({"value": value if can_view else None}).data)

authentik/crypto/secrets/apps.py

@@ -0,0 +1,12 @@
+"""Managed secrets app configuration."""
+
+from authentik.blueprints.apps import ManagedAppConfig
+
+
+class AuthentikSecretsConfig(ManagedAppConfig):
+    """Managed secrets app configuration."""
+
+    name = "authentik.crypto.secrets"
+    label = "authentik_crypto_secrets"
+    verbose_name = "authentik Secrets"
+    default = True