authentik.enterprise.lifecycle.api.iterations

  1from datetime import datetime
  2
  3from django.db.models import BooleanField as ModelBooleanField
  4from django.db.models import Case, Q, Value, When
  5from django_filters.rest_framework import BooleanFilter, FilterSet
  6from drf_spectacular.utils import extend_schema
  7from rest_framework.decorators import action
  8from rest_framework.fields import IntegerField, SerializerMethodField
  9from rest_framework.mixins import CreateModelMixin
 10from rest_framework.request import Request
 11from rest_framework.response import Response
 12from rest_framework.viewsets import GenericViewSet
 13
 14from authentik.core.api.utils import ModelSerializer
 15from authentik.enterprise.api import EnterpriseRequiredMixin
 16from authentik.enterprise.lifecycle.api.reviews import ReviewSerializer
 17from authentik.enterprise.lifecycle.models import LifecycleIteration, ReviewState
 18from authentik.enterprise.lifecycle.utils import (
 19    ContentTypeField,
 20    ReviewerGroupSerializer,
 21    ReviewerUserSerializer,
 22    admin_link_for_model,
 23    parse_content_type,
 24    start_of_day,
 25)
 26from authentik.lib.utils.time import timedelta_from_string
 27
 28
 29class LifecycleIterationSerializer(EnterpriseRequiredMixin, ModelSerializer):
 30    content_type = ContentTypeField()
 31    object_verbose = SerializerMethodField()
 32    object_admin_url = SerializerMethodField(read_only=True)
 33    grace_period_end = SerializerMethodField(read_only=True)
 34    reviews = ReviewSerializer(many=True, read_only=True, source="review_set.all")
 35    user_can_review = SerializerMethodField(read_only=True)
 36
 37    reviewer_groups = ReviewerGroupSerializer(
 38        many=True, read_only=True, source="rule.reviewer_groups"
 39    )
 40    min_reviewers = IntegerField(read_only=True, source="rule.min_reviewers")
 41    reviewers = ReviewerUserSerializer(many=True, read_only=True, source="rule.reviewers")
 42
 43    next_review_date = SerializerMethodField(read_only=True)
 44
 45    class Meta:
 46        model = LifecycleIteration
 47        fields = [
 48            "id",
 49            "content_type",
 50            "object_id",
 51            "object_verbose",
 52            "object_admin_url",
 53            "state",
 54            "opened_on",
 55            "grace_period_end",
 56            "next_review_date",
 57            "reviews",
 58            "user_can_review",
 59            "reviewer_groups",
 60            "min_reviewers",
 61            "reviewers",
 62        ]
 63        read_only_fields = fields
 64
 65    def get_object_verbose(self, iteration: LifecycleIteration) -> str:
 66        return str(iteration.object)
 67
 68    def get_object_admin_url(self, iteration: LifecycleIteration) -> str:
 69        return admin_link_for_model(iteration.object)
 70
 71    def get_grace_period_end(self, iteration: LifecycleIteration) -> datetime:
 72        return start_of_day(
 73            iteration.opened_on + timedelta_from_string(iteration.rule.grace_period)
 74        )
 75
 76    def get_next_review_date(self, iteration: LifecycleIteration) -> datetime:
 77        return start_of_day(iteration.opened_on + timedelta_from_string(iteration.rule.interval))
 78
 79    def get_user_can_review(self, iteration: LifecycleIteration) -> bool:
 80        return iteration.user_can_review(self.context["request"].user)
 81
 82
 83class LifecycleIterationFilterSet(FilterSet):
 84    user_is_reviewer = BooleanFilter(field_name="user_is_reviewer", lookup_expr="exact")
 85
 86
 87class IterationViewSet(EnterpriseRequiredMixin, CreateModelMixin, GenericViewSet):
 88    queryset = LifecycleIteration.objects.all()
 89    serializer_class = LifecycleIterationSerializer
 90    ordering = ["-opened_on"]
 91    ordering_fields = ["state", "content_type__model", "opened_on", "grace_period_end"]
 92    filterset_class = LifecycleIterationFilterSet
 93
 94    def get_queryset(self):
 95        user = self.request.user
 96        return self.queryset.annotate(
 97            user_is_reviewer=Case(
 98                When(
 99                    Q(rule__reviewers=user)
100                    | Q(rule__reviewer_groups__in=user.groups.all().with_ancestors()),
101                    then=Value(True),
102                ),
103                default=Value(False),
104                output_field=ModelBooleanField(),
105            )
106        ).distinct()
107
108    @action(
109        detail=False,
110        methods=["get"],
111        url_path=r"latest/(?P<content_type>[^/]+)/(?P<object_id>[^/]+)",
112    )
113    def latest_iteration(self, request: Request, content_type: str, object_id: str) -> Response:
114        ct = parse_content_type(content_type)
115        try:
116            obj = (
117                self.get_queryset()
118                .filter(
119                    content_type__app_label=ct["app_label"],
120                    content_type__model=ct["model"],
121                    object_id=object_id,
122                )
123                .latest("opened_on")
124            )
125        except LifecycleIteration.DoesNotExist:
126            return Response(status=404)
127        serializer = self.get_serializer(obj)
128        return Response(serializer.data)
129
130    @extend_schema(
131        operation_id="lifecycle_iterations_list_open",
132        responses={200: LifecycleIterationSerializer(many=True)},
133    )
134    @action(
135        detail=False,
136        methods=["get"],
137        url_path=r"open",
138    )
139    def open_iterations(self, request: Request):
140        iterations = self.get_queryset().filter(
141            Q(state=ReviewState.PENDING) | Q(state=ReviewState.OVERDUE)
142        )
143        iterations = self.filter_queryset(iterations)
144        page = self.paginate_queryset(iterations)
145        if page is not None:
146            serializer = self.get_serializer(page, many=True)
147            return self.get_paginated_response(serializer.data)
148
149        serializer = self.get_serializer(iterations, many=True)
150        return Response(serializer.data)