authentik.core.api.groups

  1"""Groups API Viewset"""
  2
  3from json import loads
  4
  5from django.db.models import Prefetch
  6from django.http import Http404
  7from django.utils.translation import gettext as _
  8from django_filters.filters import CharFilter, ModelMultipleChoiceFilter
  9from django_filters.filterset import FilterSet
 10from djangoql.schema import BoolField, StrField
 11from drf_spectacular.utils import (
 12    OpenApiParameter,
 13    OpenApiResponse,
 14    extend_schema,
 15    extend_schema_field,
 16)
 17from guardian.shortcuts import get_objects_for_user
 18from rest_framework.authentication import SessionAuthentication
 19from rest_framework.decorators import action
 20from rest_framework.fields import CharField, IntegerField, SerializerMethodField
 21from rest_framework.permissions import IsAuthenticated
 22from rest_framework.relations import ManyRelatedField, PrimaryKeyRelatedField
 23from rest_framework.request import Request
 24from rest_framework.response import Response
 25from rest_framework.serializers import ListSerializer, ValidationError
 26from rest_framework.viewsets import ModelViewSet
 27
 28from authentik.api.authentication import TokenAuthentication
 29from authentik.api.search.fields import (
 30    JSONSearchField,
 31)
 32from authentik.api.validation import validate
 33from authentik.core.api.used_by import UsedByMixin
 34from authentik.core.api.utils import JSONDictField, ModelSerializer, PassiveSerializer
 35from authentik.core.models import Group, User
 36from authentik.endpoints.connectors.agent.auth import AgentAuth
 37from authentik.rbac.api.roles import RoleSerializer
 38from authentik.rbac.decorators import permission_required
 39
 40
 41class BulkManyRelatedField(ManyRelatedField):
 42    """ManyRelatedField that validates all PKs in a single query instead of one per PK."""
 43
 44    def to_internal_value(self, data):
 45        if isinstance(data, str) or not hasattr(data, "__iter__"):
 46            self.fail("not_a_list", input_type=type(data).__name__)
 47        if not self.allow_empty and len(data) == 0:
 48            self.fail("empty")
 49
 50        child = self.child_relation
 51        pk_field = child.pk_field
 52        # Coerce PKs through pk_field if defined
 53        pk_map = {}
 54        for item in data:
 55            if isinstance(item, bool):
 56                self.fail("incorrect_type", data_type=type(item).__name__)
 57            pk = pk_field.to_internal_value(item) if pk_field else item
 58            pk_map[pk] = item  # map coerced PK -> original value for error reporting
 59
 60        queryset = child.get_queryset()
 61        # Use count to validate all PKs exist in a single query
 62        found_count = queryset.filter(pk__in=pk_map.keys()).count()
 63        if found_count < len(pk_map):
 64            # Some PKs not found — fall back to per-PK checks for error reporting.
 65            # This only runs when there's an actual validation error (rare path).
 66            for pk, original in pk_map.items():
 67                if not queryset.filter(pk=pk).exists():
 68                    child.fail("does_not_exist", pk_value=original)
 69
 70        # Return raw PKs — Django's M2M set() accepts both objects and PKs,
 71        # using get_prep_value() for type coercion. This avoids loading all
 72        # objects into memory and avoids triggering post_init signals.
 73        return list(pk_map.keys())
 74
 75    def to_representation(self, iterable):
 76        # For non-prefetched querysets, get PKs directly without loading model instances.
 77        # When prefetched, _result_cache is a list (possibly empty); when not, it's None.
 78        if hasattr(iterable, "values_list") and getattr(iterable, "_result_cache", None) is None:
 79            return list(iterable.values_list("pk", flat=True))
 80        return super().to_representation(iterable)
 81
 82
 83class BulkPrimaryKeyRelatedField(PrimaryKeyRelatedField):
 84    """PrimaryKeyRelatedField that uses bulk validation when many=True."""
 85
 86    @classmethod
 87    def many_init(cls, *args, **kwargs):
 88        allow_empty = kwargs.pop("allow_empty", None)
 89        max_length = kwargs.pop("max_length", None)
 90        min_length = kwargs.pop("min_length", None)
 91        child_relation = cls(*args, **kwargs)
 92        list_kwargs = {
 93            "child_relation": child_relation,
 94        }
 95        if allow_empty is not None:
 96            list_kwargs["allow_empty"] = allow_empty
 97        if max_length is not None:
 98            list_kwargs["max_length"] = max_length
 99        if min_length is not None:
100            list_kwargs["min_length"] = min_length
101        list_kwargs.update(
102            {
103                key: value
104                for key, value in kwargs.items()
105                if key in ("required", "default", "source")
106            }
107        )
108        return BulkManyRelatedField(**list_kwargs)
109
110
111PARTIAL_USER_SERIALIZER_MODEL_FIELDS = [
112    "pk",
113    "username",
114    "name",
115    "is_active",
116    "last_login",
117    "email",
118    "attributes",
119]
120
121
122class PartialUserSerializer(ModelSerializer):
123    """Partial User Serializer, does not include child relations."""
124
125    attributes = JSONDictField(required=False)
126    uid = CharField(read_only=True)
127
128    class Meta:
129        model = User
130        fields = PARTIAL_USER_SERIALIZER_MODEL_FIELDS + ["uid"]
131
132
133class RelatedGroupSerializer(ModelSerializer):
134    """Stripped down group serializer to show relevant children/parents for groups"""
135
136    attributes = JSONDictField(required=False)
137
138    class Meta:
139        model = Group
140        fields = [
141            "pk",
142            "name",
143            "is_superuser",
144            "attributes",
145            "group_uuid",
146        ]
147
148
149class GroupSerializer(ModelSerializer):
150    """Group Serializer"""
151
152    attributes = JSONDictField(required=False)
153    users = BulkPrimaryKeyRelatedField(queryset=User.objects.all(), many=True, default=list)
154    parents = PrimaryKeyRelatedField(queryset=Group.objects.all(), many=True, required=False)
155    parents_obj = SerializerMethodField(allow_null=True)
156    children_obj = SerializerMethodField(allow_null=True)
157    users_obj = SerializerMethodField(allow_null=True)
158    roles_obj = ListSerializer(
159        child=RoleSerializer(),
160        read_only=True,
161        source="roles",
162        required=False,
163    )
164    inherited_roles_obj = SerializerMethodField(allow_null=True)
165    num_pk = IntegerField(read_only=True)
166
167    @property
168    def _should_include_users(self) -> bool:
169        request: Request = self.context.get("request", None)
170        if not request:
171            return True
172        return str(request.query_params.get("include_users", "true")).lower() == "true"
173
174    @property
175    def _should_include_children(self) -> bool:
176        request: Request = self.context.get("request", None)
177        if not request:
178            return True
179        return str(request.query_params.get("include_children", "false")).lower() == "true"
180
181    @property
182    def _should_include_parents(self) -> bool:
183        request: Request = self.context.get("request", None)
184        if not request:
185            return True
186        return str(request.query_params.get("include_parents", "false")).lower() == "true"
187
188    @property
189    def _should_include_inherited_roles(self) -> bool:
190        request: Request = self.context.get("request", None)
191        if not request:
192            return True
193        return str(request.query_params.get("include_inherited_roles", "false")).lower() == "true"
194
195    @extend_schema_field(PartialUserSerializer(many=True))
196    def get_users_obj(self, instance: Group) -> list[PartialUserSerializer] | None:
197        if not self._should_include_users:
198            return None
199        return PartialUserSerializer(instance.users, many=True).data
200
201    @extend_schema_field(RelatedGroupSerializer(many=True))
202    def get_children_obj(self, instance: Group) -> list[RelatedGroupSerializer] | None:
203        if not self._should_include_children:
204            return None
205        return RelatedGroupSerializer(instance.children, many=True).data
206
207    @extend_schema_field(RelatedGroupSerializer(many=True))
208    def get_parents_obj(self, instance: Group) -> list[RelatedGroupSerializer] | None:
209        if not self._should_include_parents:
210            return None
211        return RelatedGroupSerializer(instance.parents, many=True).data
212
213    @extend_schema_field(RoleSerializer(many=True))
214    def get_inherited_roles_obj(self, instance: Group) -> list | None:
215        """Return only inherited roles from ancestor groups (excludes direct roles)"""
216        if not self._should_include_inherited_roles:
217            return None
218        direct_role_pks = instance.roles.values_list("pk", flat=True)
219        inherited_roles = instance.all_roles().exclude(pk__in=direct_role_pks)
220        return RoleSerializer(inherited_roles, many=True).data
221
222    def validate_is_superuser(self, superuser: bool):
223        """Ensure that the user creating this group has permissions to set the superuser flag"""
224        request: Request = self.context.get("request", None)
225        if not request:
226            return superuser
227        # If we're updating an instance, and the state hasn't changed, we don't need to check perms
228        if self.instance and superuser == self.instance.is_superuser:
229            return superuser
230        user: User = request.user
231        perm = (
232            "authentik_core.enable_group_superuser"
233            if superuser
234            else "authentik_core.disable_group_superuser"
235        )
236        if self.instance or superuser:
237            has_perm = user.has_perm(perm) or user.has_perm(perm, self.instance)
238            if not has_perm:
239                raise ValidationError(
240                    _(
241                        (
242                            "User does not have permission to set "
243                            "superuser status to {superuser_status}."
244                        ).format_map({"superuser_status": superuser})
245                    )
246                )
247        return superuser
248
249    def validate_users(self, users: list) -> list:
250        """Require add_user_to_group permission when adding new members via group PATCH."""
251        request: Request = self.context.get("request", None)
252        if not request:
253            return users
254        if not self.instance:
255            return users
256        # BulkManyRelatedField returns raw PKs, not model instances
257        current_user_pks = set(self.instance.users.values_list("pk", flat=True))
258        new_users = [u for u in users if u not in current_user_pks]
259        if not new_users:
260            return users
261        has_perm = request.user.has_perm(
262            "authentik_core.add_user_to_group"
263        ) or request.user.has_perm("authentik_core.add_user_to_group", self.instance)
264        if not has_perm:
265            raise ValidationError(_("User does not have permission to add members to this group."))
266        return users
267
268    class Meta:
269        model = Group
270        fields = [
271            "pk",
272            "num_pk",
273            "name",
274            "is_superuser",
275            "parents",
276            "parents_obj",
277            "users",
278            "users_obj",
279            "attributes",
280            "roles",
281            "roles_obj",
282            "inherited_roles_obj",
283            "children",
284            "children_obj",
285        ]
286        extra_kwargs = {
287            "children": {
288                "required": False,
289                "default": list,
290            },
291            "parents": {
292                "required": False,
293                "default": list,
294            },
295        }
296
297
298class GroupFilter(FilterSet):
299    """Filter for groups"""
300
301    attributes = CharFilter(
302        field_name="attributes",
303        lookup_expr="",
304        label="Attributes",
305        method="filter_attributes",
306    )
307
308    members_by_username = ModelMultipleChoiceFilter(
309        field_name="users__username",
310        to_field_name="username",
311        queryset=User.objects.all(),
312    )
313    members_by_pk = ModelMultipleChoiceFilter(
314        field_name="users",
315        queryset=User.objects.all(),
316        distinct=False,
317    )
318
319    def filter_attributes(self, queryset, name, value):
320        """Filter attributes by query args"""
321        try:
322            value = loads(value)
323        except ValueError:
324            raise ValidationError(detail="filter: failed to parse JSON") from None
325        if not isinstance(value, dict):
326            raise ValidationError(detail="filter: value must be key:value mapping")
327        qs = {}
328        for key, _value in value.items():
329            qs[f"attributes__{key}"] = _value
330        try:
331            _ = len(queryset.filter(**qs))
332            return queryset.filter(**qs)
333        except ValueError:
334            return queryset
335
336    class Meta:
337        model = Group
338        fields = ["name", "is_superuser", "members_by_pk", "attributes", "members_by_username"]
339
340
341class GroupViewSet(UsedByMixin, ModelViewSet):
342    """Group Viewset"""
343
344    class UserAccountSerializer(PassiveSerializer):
345        """Account adding/removing operations"""
346
347        pk = IntegerField(required=True)
348
349    queryset = Group.objects.none()
350    serializer_class = GroupSerializer
351    search_fields = ["name", "is_superuser"]
352    filterset_class = GroupFilter
353    ordering = ["name"]
354    authentication_classes = [
355        TokenAuthentication,
356        SessionAuthentication,
357        AgentAuth,
358    ]
359
360    def get_ql_fields(self):
361        return [
362            StrField(Group, "name"),
363            BoolField(Group, "is_superuser", nullable=True),
364            JSONSearchField(Group, "attributes"),
365        ]
366
367    def get_queryset(self):
368        # Always prefetch parents and children since their PKs are always serialized
369        base_qs = Group.objects.all().prefetch_related("roles", "parents", "children")
370
371        if self.serializer_class(context={"request": self.request})._should_include_users:
372            # Only fetch fields needed by PartialUserSerializer to reduce DB load and instantiation
373            # time
374            base_qs = base_qs.prefetch_related(
375                Prefetch(
376                    "users",
377                    queryset=User.objects.all().only(*PARTIAL_USER_SERIALIZER_MODEL_FIELDS),
378                )
379            )
380        # When include_users=false, skip users prefetch entirely.
381        # BulkManyRelatedField.to_representation will use values_list to get PKs
382        # directly without loading User instances into memory.
383
384        return base_qs
385
386    @extend_schema(
387        parameters=[
388            OpenApiParameter("include_users", bool, default=True),
389            OpenApiParameter("include_children", bool, default=False),
390            OpenApiParameter("include_parents", bool, default=False),
391            OpenApiParameter("include_inherited_roles", bool, default=False),
392        ]
393    )
394    def list(self, request, *args, **kwargs):
395        return super().list(request, *args, **kwargs)
396
397    @extend_schema(
398        parameters=[
399            OpenApiParameter("include_users", bool, default=True),
400            OpenApiParameter("include_children", bool, default=False),
401            OpenApiParameter("include_parents", bool, default=False),
402            OpenApiParameter("include_inherited_roles", bool, default=False),
403        ]
404    )
405    def retrieve(self, request, *args, **kwargs):
406        return super().retrieve(request, *args, **kwargs)
407
408    @permission_required("authentik_core.add_user_to_group")
409    @extend_schema(
410        request=UserAccountSerializer,
411        responses={
412            204: OpenApiResponse(description="User added"),
413            404: OpenApiResponse(description="User not found"),
414        },
415    )
416    @action(
417        detail=True,
418        methods=["POST"],
419        pagination_class=None,
420        filter_backends=[],
421        permission_classes=[IsAuthenticated],
422    )
423    @validate(UserAccountSerializer)
424    def add_user(self, request: Request, body: UserAccountSerializer, pk: str) -> Response:
425        """Add user to group"""
426        group: Group = self.get_object()
427        user: User = (
428            get_objects_for_user(request.user, "authentik_core.view_user")
429            .filter(
430                pk=body.validated_data.get("pk"),
431            )
432            .first()
433        )
434        if not user:
435            raise Http404
436        group.users.add(user)
437        return Response(status=204)
438
439    @permission_required("authentik_core.remove_user_from_group")
440    @extend_schema(
441        request=UserAccountSerializer,
442        responses={
443            204: OpenApiResponse(description="User removed"),
444            404: OpenApiResponse(description="User not found"),
445        },
446    )
447    @action(
448        detail=True,
449        methods=["POST"],
450        pagination_class=None,
451        filter_backends=[],
452        permission_classes=[IsAuthenticated],
453    )
454    @validate(UserAccountSerializer)
455    def remove_user(self, request: Request, body: UserAccountSerializer, pk: str) -> Response:
456        """Remove user from group"""
457        group: Group = self.get_object()
458        user: User = (
459            get_objects_for_user(request.user, "authentik_core.view_user")
460            .filter(
461                pk=body.validated_data.get("pk"),
462            )
463            .first()
464        )
465        if not user:
466            raise Http404
467        group.users.remove(user)
468        return Response(status=204)