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