authentik.core.api.users

   1"""User API Views"""
   2
   3from datetime import timedelta
   4from json import loads
   5from typing import Any
   6
   7from django.contrib.auth import update_session_auth_hash
   8from django.contrib.auth.models import AnonymousUser, Permission
   9from django.db.models import Exists, OuterRef, Prefetch, Q
  10from django.db.transaction import atomic
  11from django.db.utils import IntegrityError
  12from django.urls import reverse_lazy
  13from django.utils.http import urlencode
  14from django.utils.text import slugify
  15from django.utils.timezone import now
  16from django.utils.translation import gettext as _
  17from django.utils.translation import gettext_lazy
  18from django_filters.filters import (
  19    BooleanFilter,
  20    CharFilter,
  21    IsoDateTimeFilter,
  22    ModelMultipleChoiceFilter,
  23    MultipleChoiceFilter,
  24    UUIDFilter,
  25)
  26from django_filters.filterset import FilterSet
  27from djangoql.schema import BoolField, StrField
  28from drf_spectacular.types import OpenApiTypes
  29from drf_spectacular.utils import (
  30    OpenApiParameter,
  31    OpenApiResponse,
  32    extend_schema,
  33    extend_schema_field,
  34    inline_serializer,
  35)
  36from rest_framework.authentication import SessionAuthentication
  37from rest_framework.decorators import action
  38from rest_framework.exceptions import ValidationError
  39from rest_framework.fields import (
  40    BooleanField,
  41    CharField,
  42    ChoiceField,
  43    DateTimeField,
  44    IntegerField,
  45    ListField,
  46    SerializerMethodField,
  47    UUIDField,
  48)
  49from rest_framework.permissions import IsAuthenticated
  50from rest_framework.request import Request
  51from rest_framework.response import Response
  52from rest_framework.serializers import (
  53    ListSerializer,
  54    PrimaryKeyRelatedField,
  55)
  56from rest_framework.validators import UniqueValidator
  57from rest_framework.viewsets import ModelViewSet
  58from structlog.stdlib import get_logger
  59
  60from authentik.api.authentication import TokenAuthentication
  61from authentik.api.search.fields import (
  62    ChoiceSearchField,
  63    JSONSearchField,
  64)
  65from authentik.api.validation import validate
  66from authentik.blueprints.v1.importer import SERIALIZER_CONTEXT_BLUEPRINT
  67from authentik.brands.models import Brand
  68from authentik.core.api.used_by import UsedByMixin
  69from authentik.core.api.utils import (
  70    JSONDictField,
  71    LinkSerializer,
  72    ModelSerializer,
  73    PassiveSerializer,
  74)
  75from authentik.core.middleware import (
  76    SESSION_KEY_IMPERSONATE_ORIGINAL_USER,
  77    SESSION_KEY_IMPERSONATE_USER,
  78)
  79from authentik.core.models import (
  80    USER_ATTRIBUTE_TOKEN_EXPIRING,
  81    USER_PATH_SERVICE_ACCOUNT,
  82    USERNAME_MAX_LENGTH,
  83    Group,
  84    Session,
  85    Token,
  86    TokenIntents,
  87    User,
  88    UserTypes,
  89    default_token_duration,
  90)
  91from authentik.endpoints.connectors.agent.auth import AgentAuth
  92from authentik.events.models import Event, EventAction
  93from authentik.flows.exceptions import FlowNonApplicableException
  94from authentik.flows.models import FlowToken
  95from authentik.flows.planner import PLAN_CONTEXT_PENDING_USER, FlowPlanner
  96from authentik.flows.views.executor import QS_KEY_TOKEN
  97from authentik.lib.avatars import get_avatar
  98from authentik.lib.utils.reflection import ConditionalInheritance
  99from authentik.lib.utils.time import timedelta_from_string, timedelta_string_validator
 100from authentik.rbac.api.roles import RoleSerializer
 101from authentik.rbac.decorators import permission_required
 102from authentik.rbac.models import Role, get_permission_choices
 103from authentik.stages.email.flow import pickle_flow_token_for_email
 104from authentik.stages.email.models import EmailStage
 105from authentik.stages.email.tasks import send_mails
 106from authentik.stages.email.utils import TemplateEmailMessage
 107
 108LOGGER = get_logger()
 109
 110INVALID_PASSWORD_HASH_MESSAGE = gettext_lazy(
 111    "Invalid password hash format. Must be a valid Django password hash."
 112)
 113
 114
 115class ParamUserSerializer(PassiveSerializer):
 116    """Partial serializer for query parameters to select a user"""
 117
 118    user = PrimaryKeyRelatedField(queryset=User.objects.all().exclude_anonymous(), required=False)
 119
 120
 121class PartialGroupSerializer(ModelSerializer):
 122    """Partial Group Serializer, does not include child relations."""
 123
 124    attributes = JSONDictField(required=False)
 125
 126    class Meta:
 127        model = Group
 128        fields = [
 129            "pk",
 130            "num_pk",
 131            "name",
 132            "is_superuser",
 133            "attributes",
 134        ]
 135
 136
 137class UserSerializer(ModelSerializer):
 138    """User Serializer"""
 139
 140    is_superuser = SerializerMethodField()
 141    avatar = SerializerMethodField()
 142    attributes = JSONDictField(required=False)
 143    groups = PrimaryKeyRelatedField(
 144        allow_empty=True,
 145        many=True,
 146        queryset=Group.objects.all().order_by("name"),
 147        default=list,
 148    )
 149    groups_obj = SerializerMethodField(allow_null=True)
 150    roles = PrimaryKeyRelatedField(
 151        allow_empty=True,
 152        many=True,
 153        queryset=Role.objects.all().order_by("name"),
 154        default=list,
 155    )
 156    roles_obj = SerializerMethodField(allow_null=True)
 157    uid = CharField(read_only=True)
 158    username = CharField(
 159        max_length=USERNAME_MAX_LENGTH,
 160        validators=[UniqueValidator(queryset=User.objects.all().order_by("username"))],
 161    )
 162
 163    @property
 164    def _should_include_groups(self) -> bool:
 165        request: Request = self.context.get("request", None)
 166        if not request:
 167            return True
 168        return str(request.query_params.get("include_groups", "true")).lower() == "true"
 169
 170    @property
 171    def _should_include_roles(self) -> bool:
 172        request: Request = self.context.get("request", None)
 173        if not request:
 174            return True
 175        return str(request.query_params.get("include_roles", "true")).lower() == "true"
 176
 177    @extend_schema_field(BooleanField)
 178    def get_is_superuser(self, instance: User) -> bool:
 179        """Use annotation if available to avoid N+1 query"""
 180        ann = getattr(instance, "_annotated_is_superuser", None)
 181        if ann is not None:
 182            return ann
 183        return instance.is_superuser
 184
 185    @extend_schema_field(PartialGroupSerializer(many=True))
 186    def get_groups_obj(self, instance: User) -> list[PartialGroupSerializer] | None:
 187        if not self._should_include_groups:
 188            return None
 189        return PartialGroupSerializer(instance.groups, many=True).data
 190
 191    @extend_schema_field(RoleSerializer(many=True))
 192    def get_roles_obj(self, instance: User) -> list[RoleSerializer] | None:
 193        if not self._should_include_roles:
 194            return None
 195        return RoleSerializer(instance.roles, many=True).data
 196
 197    def __init__(self, *args, **kwargs):
 198        """Setting password and permissions directly is allowed only in blueprints."""
 199        super().__init__(*args, **kwargs)
 200        if SERIALIZER_CONTEXT_BLUEPRINT in self.context:
 201            self.fields["password"] = CharField(required=False, allow_null=True)
 202            self.fields["password_hash"] = CharField(required=False, allow_null=True)
 203            self.fields["permissions"] = ListField(
 204                required=False,
 205                child=ChoiceField(choices=get_permission_choices()),
 206            )
 207
 208    def create(self, validated_data: dict) -> User:
 209        """Create a user, with blueprint-only password and permission writes."""
 210        is_blueprint = SERIALIZER_CONTEXT_BLUEPRINT in self.context
 211        if is_blueprint:
 212            password = validated_data.pop("password", None)
 213            password_hash = validated_data.pop("password_hash", None)
 214            permissions = validated_data.pop("permissions", [])
 215            self._validate_password_inputs(password, password_hash)
 216
 217        instance: User = super().create(validated_data)
 218        if is_blueprint:
 219            self._set_password(instance, password, password_hash)
 220            perms_qs = Permission.objects.filter(
 221                codename__in=[permission.split(".")[1] for permission in permissions]
 222            ).values_list("content_type__app_label", "codename")
 223            perms_list = [f"{ct}.{name}" for ct, name in perms_qs]
 224            instance.assign_perms_to_managed_role(perms_list)
 225        self._ensure_password_not_empty(instance)
 226        return instance
 227
 228    def update(self, instance: User, validated_data: dict) -> User:
 229        """Update a user, with blueprint-only password and permission writes."""
 230        is_blueprint = SERIALIZER_CONTEXT_BLUEPRINT in self.context
 231        if is_blueprint:
 232            password = validated_data.pop("password", None)
 233            password_hash = validated_data.pop("password_hash", None)
 234            permissions = validated_data.pop("permissions", [])
 235            self._validate_password_inputs(password, password_hash)
 236
 237        instance = super().update(instance, validated_data)
 238        if is_blueprint:
 239            self._set_password(instance, password, password_hash)
 240            perms_qs = Permission.objects.filter(
 241                codename__in=[permission.split(".")[1] for permission in permissions]
 242            ).values_list("content_type__app_label", "codename")
 243            perms_list = [f"{ct}.{name}" for ct, name in perms_qs]
 244            instance.assign_perms_to_managed_role(perms_list)
 245        self._ensure_password_not_empty(instance)
 246        return instance
 247
 248    def _validate_password_inputs(self, password: str | None, password_hash: str | None):
 249        """Validate mutually-exclusive password inputs before any model mutation."""
 250        if password is not None and password_hash is not None:
 251            raise ValidationError(_("Cannot set both password and password_hash. Use only one."))
 252        if password_hash is None:
 253            return
 254        try:
 255            User.validate_password_hash(password_hash)
 256        except ValueError as exc:
 257            LOGGER.warning("Failed to identify password hash format", exc_info=exc)
 258            raise ValidationError(INVALID_PASSWORD_HASH_MESSAGE) from exc
 259
 260    def _set_password(self, instance: User, password: str | None, password_hash: str | None = None):
 261        """Set password from plain text or hash."""
 262        if password_hash is not None:
 263            instance.set_password_from_hash(password_hash)
 264            instance.save()
 265        elif password:
 266            instance.set_password(password)
 267            instance.save()
 268
 269    def _ensure_password_not_empty(self, instance: User):
 270        """Store an explicit unusable password instead of an empty password field."""
 271        if len(instance.password) == 0:
 272            instance.set_unusable_password()
 273            instance.save()
 274
 275    def get_avatar(self, user: User) -> str:
 276        """User's avatar, either a http/https URL or a data URI"""
 277        return get_avatar(user, self.context.get("request"))
 278
 279    def validate_path(self, path: str) -> str:
 280        """Validate path"""
 281        if path[:1] == "/" or path[-1] == "/":
 282            raise ValidationError(_("No leading or trailing slashes allowed."))
 283        for segment in path.split("/"):
 284            if segment == "":
 285                raise ValidationError(_("No empty segments in user path allowed."))
 286        return path
 287
 288    def validate_type(self, user_type: str) -> str:
 289        """Validate user type, internal_service_account is an internal value"""
 290        if (
 291            self.instance
 292            and self.instance.type == UserTypes.INTERNAL_SERVICE_ACCOUNT
 293            and user_type != UserTypes.INTERNAL_SERVICE_ACCOUNT.value
 294        ):
 295            raise ValidationError(_("Can't change internal service account to other user type."))
 296        if not self.instance and user_type == UserTypes.INTERNAL_SERVICE_ACCOUNT.value:
 297            raise ValidationError(_("Setting a user to internal service account is not allowed."))
 298        return user_type
 299
 300    def validate_groups(self, groups: list) -> list:
 301        """Require enable_group_superuser permission when adding a user to a superuser group."""
 302        request: Request = self.context.get("request", None)
 303        if not request:
 304            return groups
 305        current_groups = set(self.instance.groups.all()) if self.instance else set()
 306        for group in groups:
 307            if not group.is_superuser:
 308                continue
 309            if group in current_groups:
 310                continue
 311            if not request.user.has_perm("authentik_core.enable_group_superuser"):
 312                raise ValidationError(
 313                    _("User does not have permission to add members to a superuser group.")
 314                )
 315        return groups
 316
 317    def validate_roles(self, roles: list) -> list:
 318        """Require change_role permission when assigning new roles to a user."""
 319        request: Request = self.context.get("request", None)
 320        if not request:
 321            return roles
 322        current_roles = set(self.instance.roles.all()) if self.instance else set()
 323        new_roles = [r for r in roles if r not in current_roles]
 324        if not new_roles:
 325            return roles
 326        if not request.user.has_perm("authentik_rbac.change_role"):
 327            raise ValidationError(_("User does not have permission to assign roles."))
 328        return roles
 329
 330    def validate(self, attrs: dict) -> dict:
 331        if self.instance and self.instance.type == UserTypes.INTERNAL_SERVICE_ACCOUNT:
 332            raise ValidationError(_("Can't modify internal service account users"))
 333        return super().validate(attrs)
 334
 335    class Meta:
 336        model = User
 337        fields = [
 338            "pk",
 339            "username",
 340            "name",
 341            "is_active",
 342            "last_login",
 343            "date_joined",
 344            "is_superuser",
 345            "groups",
 346            "groups_obj",
 347            "roles",
 348            "roles_obj",
 349            "email",
 350            "avatar",
 351            "attributes",
 352            "uid",
 353            "path",
 354            "type",
 355            "uuid",
 356            "password_change_date",
 357            "last_updated",
 358        ]
 359        extra_kwargs = {
 360            "name": {"allow_blank": True},
 361            "date_joined": {"read_only": True},
 362            "password_change_date": {"read_only": True},
 363        }
 364
 365
 366class UserSelfSerializer(ModelSerializer):
 367    """User Serializer for information a user can retrieve about themselves"""
 368
 369    is_superuser = BooleanField(read_only=True)
 370    avatar = SerializerMethodField()
 371    groups = SerializerMethodField()
 372    roles = SerializerMethodField()
 373    uid = CharField(read_only=True)
 374    settings = SerializerMethodField()
 375    system_permissions = SerializerMethodField()
 376
 377    def get_avatar(self, user: User) -> str:
 378        """User's avatar, either a http/https URL or a data URI"""
 379        return get_avatar(user, self.context.get("request"))
 380
 381    @extend_schema_field(
 382        ListSerializer(
 383            child=inline_serializer(
 384                "UserSelfGroups",
 385                {
 386                    "name": CharField(read_only=True),
 387                    "pk": CharField(read_only=True),
 388                },
 389            )
 390        )
 391    )
 392    def get_groups(self, _: User):
 393        """Return only the group names a user is member of"""
 394        for group in self.instance.all_groups().order_by("name"):
 395            yield {
 396                "name": group.name,
 397                "pk": group.pk,
 398            }
 399
 400    @extend_schema_field(
 401        ListSerializer(
 402            child=inline_serializer(
 403                "UserSelfRoles",
 404                {
 405                    "name": CharField(read_only=True),
 406                    "pk": CharField(read_only=True),
 407                },
 408            )
 409        )
 410    )
 411    def get_roles(self, _: User):
 412        """Return only the roles a user is member of"""
 413        for role in self.instance.all_roles().order_by("name"):
 414            yield {
 415                "name": role.name,
 416                "pk": role.pk,
 417            }
 418
 419    def get_settings(self, user: User) -> dict[str, Any]:
 420        """Get user settings with brand and group settings applied"""
 421        return user.group_attributes(self._context["request"]).get("settings", {})
 422
 423    def get_system_permissions(self, user: User) -> list[str]:
 424        """Get all system permissions assigned to the user"""
 425        return list(
 426            x.split(".", maxsplit=1)[1]
 427            for x in user.get_all_permissions()
 428            if x.startswith("authentik_rbac")
 429        )
 430
 431    class Meta:
 432        model = User
 433        fields = [
 434            "pk",
 435            "username",
 436            "name",
 437            "is_active",
 438            "is_superuser",
 439            "groups",
 440            "roles",
 441            "email",
 442            "avatar",
 443            "uid",
 444            "settings",
 445            "type",
 446            "system_permissions",
 447        ]
 448        extra_kwargs = {
 449            "is_active": {"read_only": True},
 450            "name": {"allow_blank": True},
 451        }
 452
 453
 454class SessionUserSerializer(PassiveSerializer):
 455    """Response for the /user/me endpoint, returns the currently active user (as `user` property)
 456    and, if this user is being impersonated, the original user in the `original` property.
 457    """
 458
 459    user = UserSelfSerializer()
 460    original = UserSelfSerializer(required=False)
 461
 462
 463class UserPasswordSetSerializer(PassiveSerializer):
 464    """Payload to set a users' password directly"""
 465
 466    password = CharField(required=True)
 467
 468
 469class UserPasswordHashSetSerializer(PassiveSerializer):
 470    """Payload to set a users' password hash directly"""
 471
 472    password = CharField(required=True)
 473
 474
 475class UserServiceAccountSerializer(PassiveSerializer):
 476    """Payload to create a service account"""
 477
 478    name = CharField(
 479        required=True,
 480        validators=[UniqueValidator(queryset=User.objects.all().order_by("username"))],
 481    )
 482    create_group = BooleanField(default=False)
 483    expiring = BooleanField(default=True)
 484    expires = DateTimeField(
 485        required=False,
 486        help_text="If not provided, valid for 360 days",
 487    )
 488
 489
 490class UserRecoveryLinkSerializer(PassiveSerializer):
 491    """Payload to create a recovery link"""
 492
 493    token_duration = CharField(required=False)
 494
 495
 496class UserRecoveryEmailSerializer(UserRecoveryLinkSerializer):
 497    """Payload to create and email a recovery link"""
 498
 499    email_stage = UUIDField()
 500
 501
 502class UsersFilter(FilterSet):
 503    """Filter for users"""
 504
 505    attributes = CharFilter(
 506        field_name="attributes",
 507        lookup_expr="",
 508        label="Attributes",
 509        method="filter_attributes",
 510    )
 511
 512    date_joined__lt = IsoDateTimeFilter(field_name="date_joined", lookup_expr="lt")
 513    date_joined = IsoDateTimeFilter(field_name="date_joined")
 514    date_joined__gt = IsoDateTimeFilter(field_name="date_joined", lookup_expr="gt")
 515
 516    last_updated__lt = IsoDateTimeFilter(field_name="last_updated", lookup_expr="lt")
 517    last_updated = IsoDateTimeFilter(field_name="last_updated")
 518    last_updated__gt = IsoDateTimeFilter(field_name="last_updated", lookup_expr="gt")
 519
 520    last_login__lt = IsoDateTimeFilter(field_name="last_login", lookup_expr="lt")
 521    last_login = IsoDateTimeFilter(field_name="last_login")
 522    last_login__gt = IsoDateTimeFilter(field_name="last_login", lookup_expr="gt")
 523    last_login__isnull = BooleanFilter(field_name="last_login", lookup_expr="isnull")
 524
 525    is_superuser = BooleanFilter(field_name="groups", method="filter_is_superuser")
 526    uuid = UUIDFilter(field_name="uuid")
 527
 528    path = CharFilter(field_name="path")
 529    path_startswith = CharFilter(field_name="path", lookup_expr="startswith")
 530
 531    type = MultipleChoiceFilter(choices=UserTypes.choices, field_name="type")
 532
 533    groups_by_name = ModelMultipleChoiceFilter(
 534        field_name="groups__name",
 535        to_field_name="name",
 536        queryset=Group.objects.all().order_by("name"),
 537    )
 538    groups_by_pk = ModelMultipleChoiceFilter(
 539        field_name="groups",
 540        queryset=Group.objects.all().order_by("name"),
 541    )
 542
 543    roles_by_name = ModelMultipleChoiceFilter(
 544        field_name="roles__name",
 545        to_field_name="name",
 546        queryset=Role.objects.all().order_by("name"),
 547    )
 548    roles_by_pk = ModelMultipleChoiceFilter(
 549        field_name="roles",
 550        queryset=Role.objects.all().order_by("name"),
 551    )
 552
 553    def filter_is_superuser(self, queryset, name, value):
 554        if value:
 555            return queryset.filter(groups__is_superuser=True).distinct()
 556        return queryset.exclude(groups__is_superuser=True).distinct()
 557
 558    def filter_attributes(self, queryset, name, value):
 559        """Filter attributes by query args"""
 560        try:
 561            value = loads(value)
 562        except ValueError:
 563            raise ValidationError(_("filter: failed to parse JSON")) from None
 564        if not isinstance(value, dict):
 565            raise ValidationError(_("filter: value must be key:value mapping"))
 566        qs = {}
 567        for key, _value in value.items():
 568            qs[f"attributes__{key}"] = _value
 569        try:
 570            __ = len(queryset.filter(**qs))
 571            return queryset.filter(**qs)
 572        except ValueError:
 573            return queryset
 574
 575    class Meta:
 576        model = User
 577        fields = [
 578            "username",
 579            "email",
 580            "date_joined",
 581            "last_updated",
 582            "last_login",
 583            "name",
 584            "is_active",
 585            "is_superuser",
 586            "attributes",
 587            "groups_by_name",
 588            "groups_by_pk",
 589            "roles_by_name",
 590            "roles_by_pk",
 591            "type",
 592        ]
 593
 594
 595class UserViewSet(
 596    ConditionalInheritance(
 597        "authentik.enterprise.stages.account_lockdown.api.UserAccountLockdownMixin"
 598    ),
 599    ConditionalInheritance("authentik.enterprise.reports.api.reports.ExportMixin"),
 600    UsedByMixin,
 601    ModelViewSet,
 602):
 603    """User Viewset"""
 604
 605    queryset = User.objects.none()
 606    ordering = ["username", "date_joined", "last_updated", "last_login"]
 607    serializer_class = UserSerializer
 608    filterset_class = UsersFilter
 609    search_fields = ["email", "name", "uuid", "username"]
 610    authentication_classes = [
 611        TokenAuthentication,
 612        SessionAuthentication,
 613        AgentAuth,
 614    ]
 615
 616    def get_ql_fields(self):
 617        return [
 618            StrField(User, "username"),
 619            StrField(User, "name"),
 620            StrField(User, "email"),
 621            StrField(User, "path"),
 622            BoolField(User, "is_active", nullable=True),
 623            ChoiceSearchField(User, "type"),
 624            JSONSearchField(User, "attributes"),
 625        ]
 626
 627    def get_queryset(self):
 628        base_qs = User.objects.all().exclude_anonymous()
 629        # Always prefetch groups since group PKs are always serialized.
 630        # Use full prefetch when include_groups=true (for groups_obj), ID-only otherwise.
 631        if self.serializer_class(context={"request": self.request})._should_include_groups:
 632            base_qs = base_qs.prefetch_related("groups")
 633        else:
 634            base_qs = base_qs.prefetch_related(
 635                Prefetch("groups", queryset=Group.objects.all().only("group_uuid"))
 636            )
 637        if self.serializer_class(context={"request": self.request})._should_include_roles:
 638            base_qs = base_qs.prefetch_related("roles")
 639        else:
 640            base_qs = base_qs.prefetch_related(
 641                Prefetch("roles", queryset=Role.objects.all().only("uuid"))
 642            )
 643        # Annotate is_superuser to avoid N+1 query per user
 644        base_qs = base_qs.annotate(
 645            _annotated_is_superuser=Exists(
 646                Group.objects.filter(
 647                    is_superuser=True,
 648                ).filter(
 649                    Q(users=OuterRef("pk")) | Q(descendant_nodes__descendant__users=OuterRef("pk"))
 650                )
 651            )
 652        )
 653        return base_qs
 654
 655    @extend_schema(
 656        parameters=[
 657            OpenApiParameter("include_groups", bool, default=True),
 658            OpenApiParameter("include_roles", bool, default=True),
 659        ]
 660    )
 661    def list(self, request, *args, **kwargs):
 662        return super().list(request, *args, **kwargs)
 663
 664    def _create_recovery_link(
 665        self, token_duration: str | None, for_email=False
 666    ) -> tuple[str, Token]:
 667        """Create a recovery link (when the current brand has a recovery flow set),
 668        that can either be shown to an admin or sent to the user directly"""
 669        brand: Brand = self.request.brand
 670        # Check that there is a recovery flow, if not return an error
 671        flow = brand.flow_recovery
 672        if not flow:
 673            raise ValidationError({"non_field_errors": _("No recovery flow set.")})
 674        user: User = self.get_object()
 675        planner = FlowPlanner(flow)
 676        planner.allow_empty_flows = True
 677        self.request._request.user = AnonymousUser()
 678        try:
 679            plan = planner.plan(
 680                self.request._request,
 681                {
 682                    PLAN_CONTEXT_PENDING_USER: user,
 683                },
 684            )
 685        except FlowNonApplicableException:
 686            raise ValidationError(
 687                {"non_field_errors": _("Recovery flow not applicable to user")}
 688            ) from None
 689        _plan = FlowToken.pickle(plan)
 690        if for_email:
 691            _plan = pickle_flow_token_for_email(plan)
 692        expires = default_token_duration()
 693        if token_duration:
 694            timedelta_string_validator(token_duration)
 695            expires = now() + timedelta_from_string(token_duration)
 696        token, __ = FlowToken.objects.update_or_create(
 697            identifier=f"{user.uid}-password-reset",
 698            defaults={
 699                "user": user,
 700                "flow": flow,
 701                "_plan": _plan,
 702                "revoke_on_execution": not for_email,
 703                "expires": expires,
 704            },
 705        )
 706        querystring = urlencode({QS_KEY_TOKEN: token.key})
 707        link = self.request.build_absolute_uri(
 708            reverse_lazy("authentik_core:if-flow", kwargs={"flow_slug": flow.slug})
 709            + f"?{querystring}"
 710        )
 711        return link, token
 712
 713    @permission_required(None, ["authentik_core.add_user", "authentik_core.add_token"])
 714    @extend_schema(
 715        request=UserServiceAccountSerializer,
 716        responses={
 717            200: inline_serializer(
 718                "UserServiceAccountResponse",
 719                {
 720                    "username": CharField(required=True),
 721                    "token": CharField(required=True),
 722                    "user_uid": CharField(required=True),
 723                    "user_pk": IntegerField(required=True),
 724                    "group_pk": CharField(required=False),
 725                },
 726            )
 727        },
 728    )
 729    @action(
 730        detail=False,
 731        methods=["POST"],
 732        pagination_class=None,
 733        filter_backends=[],
 734    )
 735    @validate(UserServiceAccountSerializer)
 736    def service_account(self, request: Request, body: UserServiceAccountSerializer) -> Response:
 737        """Create a new user account that is marked as a service account"""
 738        expires = body.validated_data.get("expires", now() + timedelta(days=360))
 739
 740        username = body.validated_data["name"]
 741        expiring = body.validated_data["expiring"]
 742        with atomic():
 743            try:
 744                user: User = User.objects.create(
 745                    username=username,
 746                    name=username,
 747                    type=UserTypes.SERVICE_ACCOUNT,
 748                    attributes={USER_ATTRIBUTE_TOKEN_EXPIRING: expiring},
 749                    path=USER_PATH_SERVICE_ACCOUNT,
 750                )
 751                user.set_unusable_password()
 752                user.save()
 753
 754                response = {
 755                    "username": user.username,
 756                    "user_uid": user.uid,
 757                    "user_pk": user.pk,
 758                }
 759                if body.validated_data["create_group"] and self.request.user.has_perm(
 760                    "authentik_core.add_group"
 761                ):
 762                    group = Group.objects.create(name=username)
 763                    group.users.add(user)
 764                    response["group_pk"] = str(group.pk)
 765                token = Token.objects.create(
 766                    identifier=slugify(f"service-account-{username}-password"),
 767                    intent=TokenIntents.INTENT_APP_PASSWORD,
 768                    user=user,
 769                    expires=expires,
 770                    expiring=expiring,
 771                )
 772                response["token"] = token.key
 773                return Response(response)
 774            except IntegrityError as exc:
 775                error_msg = str(exc).lower()
 776
 777                if "unique" in error_msg:
 778                    return Response(
 779                        data={
 780                            "non_field_errors": [
 781                                _("A user/group with these details already exists")
 782                            ]
 783                        },
 784                        status=400,
 785                    )
 786                else:
 787                    LOGGER.warning("Service account creation failed", exc=exc)
 788                    return Response(
 789                        data={"non_field_errors": [_("Unable to create user")]},
 790                        status=400,
 791                    )
 792            except (ValueError, TypeError) as exc:
 793                LOGGER.error("Unexpected error during service account creation", exc=exc)
 794                return Response(
 795                    data={"non_field_errors": [_("Unknown error occurred")]},
 796                    status=500,
 797                )
 798
 799    @extend_schema(responses={200: SessionUserSerializer(many=False)})
 800    @action(
 801        url_path="me",
 802        url_name="me",
 803        detail=False,
 804        pagination_class=None,
 805        filter_backends=[],
 806    )
 807    def user_me(self, request: Request) -> Response:
 808        """Get information about current user"""
 809        context = {"request": request}
 810        serializer = SessionUserSerializer(
 811            data={"user": UserSelfSerializer(instance=request.user, context=context).data}
 812        )
 813        if SESSION_KEY_IMPERSONATE_USER in request._request.session:
 814            serializer.initial_data["original"] = UserSelfSerializer(
 815                instance=request._request.session[SESSION_KEY_IMPERSONATE_ORIGINAL_USER],
 816                context=context,
 817            ).data
 818        self.request.session.modified = True
 819        return Response(serializer.initial_data)
 820
 821    def _update_session_hash_after_password_change(self, request: Request, user: User):
 822        if user.pk == request.user.pk and SESSION_KEY_IMPERSONATE_USER not in self.request.session:
 823            LOGGER.debug("Updating session hash after password change")
 824            update_session_auth_hash(self.request, user)
 825
 826    @permission_required("authentik_core.reset_user_password")
 827    @extend_schema(
 828        request=UserPasswordSetSerializer,
 829        responses={
 830            204: OpenApiResponse(description="Successfully changed password"),
 831            400: OpenApiResponse(description="Bad request"),
 832        },
 833    )
 834    @action(
 835        detail=True,
 836        methods=["POST"],
 837        permission_classes=[IsAuthenticated],
 838    )
 839    @validate(UserPasswordSetSerializer)
 840    def set_password(self, request: Request, pk: int, body: UserPasswordSetSerializer) -> Response:
 841        """Set password for user"""
 842        user: User = self.get_object()
 843        try:
 844            user.set_password(body.validated_data["password"], request=request)
 845            user.save()
 846        except (ValidationError, IntegrityError) as exc:
 847            LOGGER.debug("Failed to set password", exc=exc)
 848            return Response(status=400)
 849        self._update_session_hash_after_password_change(request, user)
 850        return Response(status=204)
 851
 852    @permission_required("authentik_core.reset_user_password")
 853    @extend_schema(
 854        request=UserPasswordHashSetSerializer,
 855        responses={
 856            204: OpenApiResponse(description="Successfully changed password"),
 857            400: OpenApiResponse(description="Bad request"),
 858        },
 859    )
 860    @action(
 861        detail=True,
 862        methods=["POST"],
 863        permission_classes=[IsAuthenticated],
 864    )
 865    @validate(UserPasswordHashSetSerializer)
 866    def set_password_hash(
 867        self, request: Request, pk: int, body: UserPasswordHashSetSerializer
 868    ) -> Response:
 869        """Set a user's password from a pre-hashed Django password value.
 870
 871        Submit the Django password hash in the shared ``password`` request field.
 872
 873        This updates authentik's local password verifier only. It does not attempt
 874        to propagate the password change to LDAP or Kerberos because no raw password
 875        is available from the request payload.
 876        """
 877        user: User = self.get_object()
 878        try:
 879            user.set_password_from_hash(body.validated_data["password"], request=request)
 880            user.save()
 881        except ValueError as exc:
 882            LOGGER.debug("Failed to set password hash", exc=exc)
 883            return Response(data={"password": [INVALID_PASSWORD_HASH_MESSAGE]}, status=400)
 884        except (ValidationError, IntegrityError) as exc:
 885            LOGGER.debug("Failed to set password hash", exc=exc)
 886            return Response(status=400)
 887        self._update_session_hash_after_password_change(request, user)
 888        return Response(status=204)
 889
 890    @permission_required("authentik_core.reset_user_password")
 891    @extend_schema(
 892        request=UserRecoveryLinkSerializer,
 893        responses={
 894            "200": LinkSerializer(many=False),
 895        },
 896    )
 897    @action(detail=True, pagination_class=None, filter_backends=[], methods=["POST"])
 898    @validate(UserRecoveryLinkSerializer)
 899    def recovery(self, request: Request, pk: int, body: UserRecoveryLinkSerializer) -> Response:
 900        """Create a temporary link that a user can use to recover their account"""
 901        link, _ = self._create_recovery_link(
 902            token_duration=body.validated_data.get("token_duration")
 903        )
 904        return Response({"link": link})
 905
 906    @permission_required("authentik_core.reset_user_password")
 907    @extend_schema(
 908        request=UserRecoveryEmailSerializer,
 909        responses={
 910            "204": OpenApiResponse(description="Successfully sent recover email"),
 911        },
 912    )
 913    @action(detail=True, pagination_class=None, filter_backends=[], methods=["POST"])
 914    @validate(UserRecoveryEmailSerializer)
 915    def recovery_email(
 916        self, request: Request, pk: int, body: UserRecoveryEmailSerializer
 917    ) -> Response:
 918        """Send an email with a temporary link that a user can use to recover their account"""
 919        email_error_message = _("User does not have an email address set.")
 920        stage_error_message = _("Email stage not found.")
 921        user: User = self.get_object()
 922        if not user.email:
 923            LOGGER.debug("User doesn't have an email address")
 924            raise ValidationError({"non_field_errors": email_error_message})
 925        if not (stage := EmailStage.objects.filter(pk=body.validated_data["email_stage"]).first()):
 926            LOGGER.debug("Email stage does not exist")
 927            raise ValidationError({"non_field_errors": stage_error_message})
 928        if not request.user.has_perm("authentik_stages_email.view_emailstage", stage):
 929            LOGGER.debug("User has no view access to email stage")
 930            raise ValidationError({"non_field_errors": stage_error_message})
 931        link, token = self._create_recovery_link(
 932            token_duration=body.validated_data.get("token_duration"), for_email=True
 933        )
 934        message = TemplateEmailMessage(
 935            subject=_(stage.subject),
 936            to=[(user.name, user.email)],
 937            template_name=stage.template,
 938            language=user.locale(request),
 939            template_context={
 940                "url": link,
 941                "user": user,
 942                "expires": token.expires,
 943            },
 944        )
 945        send_mails(stage, message)
 946        return Response(status=204)
 947
 948    @permission_required("authentik_core.impersonate")
 949    @extend_schema(
 950        request=inline_serializer(
 951            "ImpersonationSerializer",
 952            {
 953                "reason": CharField(required=True),
 954            },
 955        ),
 956        responses={
 957            204: OpenApiResponse(description="Successfully started impersonation"),
 958        },
 959    )
 960    @action(detail=True, methods=["POST"], permission_classes=[IsAuthenticated])
 961    def impersonate(self, request: Request, pk: int) -> Response:
 962        """Impersonate a user"""
 963        if not request.tenant.impersonation:
 964            LOGGER.debug("User attempted to impersonate", user=request.user)
 965            return Response(status=401)
 966        user_to_be = self.get_object()
 967        reason = request.data.get("reason", "")
 968        # Check both object-level perms and global perms
 969        if not request.user.has_perm(
 970            "authentik_core.impersonate", user_to_be
 971        ) and not request.user.has_perm("authentik_core.impersonate"):
 972            LOGGER.debug(
 973                "User attempted to impersonate without permissions",
 974                user=request.user,
 975            )
 976            return Response(status=403)
 977        if user_to_be.pk == self.request.user.pk:
 978            LOGGER.debug("User attempted to impersonate themselves", user=request.user)
 979            return Response(status=401)
 980        if not reason and request.tenant.impersonation_require_reason:
 981            LOGGER.debug(
 982                "User attempted to impersonate without providing a reason",
 983                user=request.user,
 984            )
 985            raise ValidationError({"reason": _("This field is required.")})
 986
 987        request.session[SESSION_KEY_IMPERSONATE_ORIGINAL_USER] = request.user
 988        request.session[SESSION_KEY_IMPERSONATE_USER] = user_to_be
 989
 990        Event.new(EventAction.IMPERSONATION_STARTED, reason=reason).from_http(request, user_to_be)
 991
 992        return Response(status=204)
 993
 994    @extend_schema(
 995        request=None,
 996        responses={
 997            "204": OpenApiResponse(description="Successfully ended impersonation"),
 998        },
 999    )
1000    @action(detail=False, methods=["GET"])
1001    def impersonate_end(self, request: Request) -> Response:
1002        """End Impersonation a user"""
1003        if (
1004            SESSION_KEY_IMPERSONATE_USER not in request.session
1005            or SESSION_KEY_IMPERSONATE_ORIGINAL_USER not in request.session
1006        ):
1007            LOGGER.debug("Can't end impersonation", user=request.user)
1008            return Response(status=204)
1009
1010        original_user = request.session[SESSION_KEY_IMPERSONATE_ORIGINAL_USER]
1011
1012        del request.session[SESSION_KEY_IMPERSONATE_USER]
1013        del request.session[SESSION_KEY_IMPERSONATE_ORIGINAL_USER]
1014
1015        Event.new(EventAction.IMPERSONATION_ENDED).from_http(request, original_user)
1016
1017        return Response(status=204)
1018
1019    @extend_schema(
1020        responses={
1021            200: inline_serializer(
1022                "UserPathSerializer",
1023                {"paths": ListField(child=CharField(), read_only=True)},
1024            )
1025        },
1026        parameters=[
1027            OpenApiParameter(
1028                name="search",
1029                location=OpenApiParameter.QUERY,
1030                type=OpenApiTypes.STR,
1031            )
1032        ],
1033    )
1034    @action(detail=False, pagination_class=None)
1035    def paths(self, request: Request) -> Response:
1036        """Get all user paths"""
1037        return Response(
1038            data={
1039                "paths": list(
1040                    self.filter_queryset(self.get_queryset())
1041                    .values("path")
1042                    .distinct()
1043                    .order_by("path")
1044                    .values_list("path", flat=True)
1045                )
1046            }
1047        )
1048
1049    def partial_update(self, request: Request, *args, **kwargs) -> Response:
1050        response = super().partial_update(request, *args, **kwargs)
1051        instance: User = self.get_object()
1052        if not instance.is_active:
1053            Session.objects.filter(authenticatedsession__user=instance).delete()
1054            LOGGER.debug("Deleted user's sessions", user=instance.username)
1055        return response