authentik.enterprise.lifecycle.api.iterations

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