Enterprise/agent service accounts

authentik/core/api/tokens.py

@@ -1,5 +1,6 @@
 """Tokens API Viewset"""
 
+from datetime import timedelta
 from typing import Any
 
 from django.utils.timezone import now
@@ -18,12 +19,15 @@
 from authentik.core.api.users import UserSerializer
 from authentik.core.api.utils import ModelSerializer, PassiveSerializer
 from authentik.core.models import (
+    USER_ATTRIBUTE_AGENT_OWNER_PK,
+    USER_ATTRIBUTE_IS_AGENT,
     USER_ATTRIBUTE_TOKEN_EXPIRING,
     USER_ATTRIBUTE_TOKEN_MAXIMUM_LIFETIME,
     Token,
     TokenIntents,
     User,
     default_token_duration,
+    default_token_key,
 )
 from authentik.events.models import Event, EventAction
 from authentik.events.utils import model_to_dict
@@ -171,6 +175,33 @@ def view_key(self, request: Request, identifier: str) -> Response:
         Event.new(EventAction.SECRET_VIEW, secret=token).from_http(request)  # noqa # nosec
         return Response(TokenViewSerializer({"key": token.key}).data)
 
+    @extend_schema(
+        request=None,
+        responses={
+            200: TokenViewSerializer(many=False),
+            403: OpenApiResponse(description="Not the token owner, agent owner, or superuser"),
+        },
+    )
+    @action(detail=True, pagination_class=None, filter_backends=[], methods=["POST"])
+    def rotate(self, request: Request, identifier: str) -> Response:
+        """Rotate the token key and reset the expiry to 24 hours. Only callable by the token
+        owner, the owning agent's human owner, or a superuser."""
+        token: Token = self.get_object()
+
+        if not request.user.is_superuser:
+            is_token_owner = token.user_id == request.user.pk
+            is_agent_owner = token.user.attributes.get(USER_ATTRIBUTE_IS_AGENT) and str(
+                request.user.pk
+            ) == token.user.attributes.get(USER_ATTRIBUTE_AGENT_OWNER_PK)
+            if not is_token_owner and not is_agent_owner:
+                return Response(status=403)
+
+        token.key = default_token_key()
+        token.expires = now() + timedelta(hours=24)
+        token.save()
+        Event.new(EventAction.SECRET_ROTATE, secret=token).from_http(request)  # noqa # nosec
+        return Response(TokenViewSerializer({"key": token.key}).data)
+
     @permission_required("authentik_core.set_token_key")
     @extend_schema(
         request=TokenSetKeySerializer(),

authentik/core/api/users.py

@@ -75,9 +75,14 @@
     SESSION_KEY_IMPERSONATE_USER,
 )
 from authentik.core.models import (
+    USER_ATTRIBUTE_AGENT_ALLOWED_APPS,
+    USER_ATTRIBUTE_AGENT_OWNER_PK,
+    USER_ATTRIBUTE_IS_AGENT,
     USER_ATTRIBUTE_TOKEN_EXPIRING,
+    USER_PATH_AGENT,
     USER_PATH_SERVICE_ACCOUNT,
     USERNAME_MAX_LENGTH,
+    Application,
     Group,
     Session,
     Token,
@@ -86,6 +91,7 @@
     UserTypes,
     default_token_duration,
 )
+from authentik.core.apps import AppAccessWithoutBindings
 from authentik.endpoints.connectors.agent.auth import AgentAuth
 from authentik.events.models import Event, EventAction
 from authentik.flows.exceptions import FlowNonApplicableException
@@ -95,6 +101,7 @@
 from authentik.lib.avatars import get_avatar
 from authentik.lib.utils.reflection import ConditionalInheritance
 from authentik.lib.utils.time import timedelta_from_string, timedelta_string_validator
+from authentik.policies.engine import PolicyEngine
 from authentik.rbac.api.roles import RoleSerializer
 from authentik.rbac.decorators import permission_required
 from authentik.rbac.models import Role, get_permission_choices
@@ -249,8 +256,27 @@ def validate_type(self, user_type: str) -> str:
             raise ValidationError(_("Can't change internal service account to other user type."))
         if not self.instance and user_type == UserTypes.INTERNAL_SERVICE_ACCOUNT.value:
             raise ValidationError(_("Setting a user to internal service account is not allowed."))
+        if (
+            self.instance
+            and self.instance.attributes.get(USER_ATTRIBUTE_IS_AGENT)
+            and user_type != UserTypes.SERVICE_ACCOUNT.value
+        ):
+            raise ValidationError(_("Can't change agent user to other user type."))
         return user_type
 
+    def validate_attributes(self, attrs: dict) -> dict:
+        """Prevent removal of agent marker or change of agent owner"""
+        if not self.instance:
+            return attrs
+        if self.instance.attributes.get(USER_ATTRIBUTE_IS_AGENT):
+            if not attrs.get(USER_ATTRIBUTE_IS_AGENT):
+                raise ValidationError(_("Can't remove agent marker from agent user."))
+            existing_owner = self.instance.attributes.get(USER_ATTRIBUTE_AGENT_OWNER_PK)
+            new_owner = attrs.get(USER_ATTRIBUTE_AGENT_OWNER_PK)
+            if existing_owner is not None and new_owner != existing_owner:
+                raise ValidationError(_("Can't change owner of agent user."))
+        return attrs
+
     def validate(self, attrs: dict) -> dict:
         if self.instance and self.instance.type == UserTypes.INTERNAL_SERVICE_ACCOUNT:
             raise ValidationError(_("Can't modify internal service account users"))
@@ -405,6 +431,24 @@ class UserServiceAccountSerializer(PassiveSerializer):
     )
 
 
+class UserAgentSerializer(PassiveSerializer):
+    """Payload to create an agent user"""
+
+    name = CharField(
+        required=True,
+        validators=[UniqueValidator(queryset=User.objects.all().order_by("username"))],
+    )
+
+
+class UserAgentAllowedAppsSerializer(PassiveSerializer):
+    """Payload to replace the allowed application list for an agent user"""
+
+    allowed_apps = ListField(
+        child=UUIDField(),
+        help_text="List of application UUIDs the agent is permitted to access.",
+    )
+
+
 class UserRecoveryLinkSerializer(PassiveSerializer):
     """Payload to create a recovery link"""
 
@@ -691,6 +735,175 @@ def service_account(self, request: Request, body: UserServiceAccountSerializer)
                     status=500,
                 )
 
+    @permission_required(None, ["authentik_core.add_agent_user"])
+    @extend_schema(
+        request=UserAgentSerializer,
+        responses={
+            200: inline_serializer(
+                "UserAgentResponse",
+                {
+                    "username": CharField(required=True),
+                    "token": CharField(required=True),
+                    "user_uid": CharField(required=True),
+                    "user_pk": IntegerField(required=True),
+                },
+            )
+        },
+    )
+    @action(
+        detail=False,
+        methods=["POST"],
+        pagination_class=None,
+        filter_backends=[],
+    )
+    @validate(UserAgentSerializer)
+    def agent(self, request: Request, body: UserAgentSerializer) -> Response:
+        """Create a new agent user. Enterprise only. Caller must be an internal user."""
+        from authentik.enterprise.license import LicenseKey
+
+        if not LicenseKey.cached_summary().status.is_valid:
+            raise ValidationError(_("Enterprise is required to use this endpoint."))
+
+        if request.user.type != UserTypes.INTERNAL:
+            raise ValidationError(_("Only internal users can create agent users."))
+
+        username = body.validated_data["name"]
+        with atomic():
+            try:
+                user: User = User.objects.create(
+                    username=username,
+                    name=username,
+                    type=UserTypes.SERVICE_ACCOUNT,
+                    attributes={
+                        USER_ATTRIBUTE_IS_AGENT: True,
+                        USER_ATTRIBUTE_AGENT_OWNER_PK: str(request.user.pk),
+                        USER_ATTRIBUTE_AGENT_ALLOWED_APPS: [],
+                    },
+                    path=USER_PATH_AGENT,
+                )
+                user.set_unusable_password()
+                user.save()
+
+                token = Token.objects.create(
+                    identifier=slugify(f"agent-{username}-token"),
+                    intent=TokenIntents.INTENT_API,
+                    user=user,
+                    expires=now() + timedelta(hours=24),
+                    expiring=True,
+                )
+                # Allow the agent to read its own rotated token key
+                user.assign_perms_to_managed_role("authentik_core.view_token_key", token)
+
+                return Response(
+                    {
+                        "username": user.username,
+                        "user_uid": user.uid,
+                        "user_pk": user.pk,
+                        "token": token.key,
+                    }
+                )
+            except IntegrityError as exc:
+                error_msg = str(exc).lower()
+                if "unique" in error_msg:
+                    return Response(
+                        data={
+                            "non_field_errors": [
+                                _("A user with this username already exists")
+                            ]
+                        },
+                        status=400,
+                    )
+                else:
+                    LOGGER.warning("Agent user creation failed", exc=exc)
+                    return Response(
+                        data={"non_field_errors": [_("Unable to create user")]},
+                        status=400,
+                    )
+            except (ValueError, TypeError) as exc:
+                LOGGER.error("Unexpected error during agent user creation", exc=exc)
+                return Response(
+                    data={"non_field_errors": [_("Unknown error occurred")]},
+                    status=500,
+                )
+
+    @extend_schema(
+        request=UserAgentAllowedAppsSerializer,
+        responses={
+            200: UserAgentAllowedAppsSerializer,
+            400: OpenApiResponse(description="Invalid app UUIDs or owner lacks access"),
+            403: OpenApiResponse(description="Not the agent's owner or superuser"),
+        },
+    )
+    @action(
+        detail=True,
+        methods=["PUT"],
+        url_path="agent_allowed_apps",
+        url_name="agent-allowed-apps",
+        pagination_class=None,
+        filter_backends=[],
+    )
+    @validate(UserAgentAllowedAppsSerializer)
+    def agent_allowed_apps(
+        self, request: Request, pk: int, body: UserAgentAllowedAppsSerializer
+    ) -> Response:
+        """Replace the allowed application list for an agent user.
+        Caller must be the agent's owner or a superuser. Each supplied application UUID
+        is validated against the owner's current access."""
+        agent: User = self.get_object()
+
+        if not agent.attributes.get(USER_ATTRIBUTE_IS_AGENT):
+            return Response(
+                data={"non_field_errors": [_("User is not an agent user.")]},
+                status=400,
+            )
+
+        owner_pk = agent.attributes.get(USER_ATTRIBUTE_AGENT_OWNER_PK)
+        is_owner = str(request.user.pk) == owner_pk
+        if not request.user.is_superuser and not is_owner:
+            return Response(status=403)
+
+        try:
+            owner = User.objects.get(pk=owner_pk)
+        except User.DoesNotExist:
+            return Response(
+                data={"non_field_errors": [_("Agent owner not found.")]},
+                status=400,
+            )
+
+        app_uuids = body.validated_data["allowed_apps"]
+        errors = []
+        for app_uuid in app_uuids:
+            try:
+                app = Application.objects.get(pk=app_uuid)
+            except Application.DoesNotExist:
+                errors.append(str(app_uuid))
+                continue
+            engine = PolicyEngine(app, owner, request)
+            engine.empty_result = AppAccessWithoutBindings.get()
+            engine.use_cache = False
+            engine.build()
+            if not engine.passing:
+                errors.append(str(app_uuid))
+
+        if errors:
+            return Response(
+                data={
+                    "allowed_apps": [
+                        _(
+                            "Owner does not have access to application %(uuid)s "
+                            "or application does not exist."
+                        )
+                        % {"uuid": uuid}
+                        for uuid in errors
+                    ]
+                },
+                status=400,
+            )
+
+        agent.attributes[USER_ATTRIBUTE_AGENT_ALLOWED_APPS] = [str(u) for u in app_uuids]
+        agent.save(update_fields=["attributes"])
+        return Response({"allowed_apps": [str(u) for u in app_uuids]})
+
     @extend_schema(responses={200: SessionUserSerializer(many=False)})
     @action(
         url_path="me",

authentik/core/migrations/0058_user_add_agent_user_permission.py

@@ -0,0 +1,27 @@
+"""Add add_agent_user permission to User model"""
+
+from django.db import migrations
+
+
+class Migration(migrations.Migration):
+
+    dependencies = [
+        ("authentik_core", "0057_remove_user_groups_remove_user_user_permissions_and_more"),
+    ]
+
+    operations = [
+        migrations.AlterModelOptions(
+            name="user",
+            options={
+                "permissions": [
+                    ("reset_user_password", "Reset Password"),
+                    ("impersonate", "Can impersonate other users"),
+                    ("preview_user", "Can preview user data sent to providers"),
+                    ("view_user_applications", "View applications the user has access to"),
+                    ("add_agent_user", "Can create agent users"),
+                ],
+                "verbose_name": "User",
+                "verbose_name_plural": "Users",
+            },
+        ),
+    ]

authentik/core/models.py

@@ -66,7 +66,12 @@
 USER_ATTRIBUTE_CHANGE_USERNAME = f"{_USER_ATTR_PREFIX}/can-change-username"
 USER_ATTRIBUTE_CHANGE_NAME = f"{_USER_ATTR_PREFIX}/can-change-name"
 USER_ATTRIBUTE_CHANGE_EMAIL = f"{_USER_ATTR_PREFIX}/can-change-email"
+_USER_ATTR_AGENT_PREFIX = f"{USER_PATH_SYSTEM_PREFIX}/agent"
+USER_ATTRIBUTE_IS_AGENT = f"{_USER_ATTR_AGENT_PREFIX}/is-agent"
+USER_ATTRIBUTE_AGENT_OWNER_PK = f"{_USER_ATTR_AGENT_PREFIX}/owner-pk"
+USER_ATTRIBUTE_AGENT_ALLOWED_APPS = f"{_USER_ATTR_AGENT_PREFIX}/allowed-apps"
 USER_PATH_SERVICE_ACCOUNT = f"{USER_PATH_SYSTEM_PREFIX}/service-accounts"
+USER_PATH_AGENT = f"{USER_PATH_SYSTEM_PREFIX}/agents"
 
 options.DEFAULT_NAMES = options.DEFAULT_NAMES + (
     # used_by API that allows models to specify if they shadow an object
@@ -385,6 +390,7 @@ class Meta:
             ("impersonate", _("Can impersonate other users")),
             ("preview_user", _("Can preview user data sent to providers")),
             ("view_user_applications", _("View applications the user has access to")),
+            ("add_agent_user", _("Can create agent users")),
         ]
         indexes = [
             models.Index(fields=["last_login"]),