authentik.enterprise.lifecycle.api.rules

  1from django.utils.translation import gettext as _
  2from rest_framework.exceptions import ValidationError
  3from rest_framework.fields import SerializerMethodField
  4from rest_framework.relations import SlugRelatedField
  5from rest_framework.viewsets import ModelViewSet
  6
  7from authentik.core.api.utils import ModelSerializer
  8from authentik.core.models import User
  9from authentik.enterprise.api import EnterpriseRequiredMixin
 10from authentik.enterprise.lifecycle.models import LifecycleRule
 11from authentik.enterprise.lifecycle.utils import (
 12    ContentTypeField,
 13    ReviewerGroupSerializer,
 14    ReviewerUserSerializer,
 15)
 16from authentik.lib.utils.time import timedelta_from_string
 17
 18
 19class LifecycleRuleSerializer(EnterpriseRequiredMixin, ModelSerializer):
 20    content_type = ContentTypeField()
 21    target_verbose = SerializerMethodField()
 22    reviewer_groups_obj = ReviewerGroupSerializer(
 23        many=True, read_only=True, source="reviewer_groups"
 24    )
 25    reviewers = SlugRelatedField(slug_field="uuid", many=True, queryset=User.objects.all())
 26    reviewers_obj = ReviewerUserSerializer(many=True, read_only=True, source="reviewers")
 27
 28    class Meta:
 29        model = LifecycleRule
 30        fields = [
 31            "id",
 32            "name",
 33            "content_type",
 34            "object_id",
 35            "interval",
 36            "grace_period",
 37            "reviewer_groups",
 38            "reviewer_groups_obj",
 39            "min_reviewers",
 40            "min_reviewers_is_per_group",
 41            "reviewers",
 42            "reviewers_obj",
 43            "notification_transports",
 44            "target_verbose",
 45        ]
 46        read_only_fields = ["id", "reviewers_obj", "reviewer_groups_obj", "target_verbose"]
 47
 48    def get_target_verbose(self, rule: LifecycleRule) -> str:
 49        if rule.object_id is None:
 50            return rule.content_type.model_class()._meta.verbose_name_plural
 51        else:
 52            return f"{rule.content_type.model_class()._meta.verbose_name}: {rule.object}"
 53
 54    def validate_object_id(self, value: str) -> str | None:
 55        if value == "":
 56            return None
 57        return value
 58
 59    def validate(self, attrs: dict) -> dict:
 60        if (
 61            attrs.get("object_id") is not None
 62            and not attrs["content_type"]
 63            .get_all_objects_for_this_type(pk=attrs["object_id"])
 64            .exists()
 65        ):
 66            raise ValidationError({"object_id": _("Object does not exist")})
 67        if "reviewer_groups" in attrs or "reviewers" in attrs:
 68            reviewer_groups = attrs.get(
 69                "reviewer_groups", self.instance.reviewer_groups.all() if self.instance else []
 70            )
 71            reviewers = attrs.get(
 72                "reviewers", self.instance.reviewers.all() if self.instance else []
 73            )
 74            if len(reviewer_groups) == 0 and len(reviewers) == 0:
 75                raise ValidationError(_("Either a reviewer group or a reviewer must be set."))
 76        if "grace_period" in attrs or "interval" in attrs:
 77            grace_period = attrs.get("grace_period", getattr(self.instance, "grace_period", None))
 78            interval = attrs.get("interval", getattr(self.instance, "interval", None))
 79            if (
 80                grace_period is not None
 81                and interval is not None
 82                and (timedelta_from_string(grace_period) > timedelta_from_string(interval))
 83            ):
 84                raise ValidationError(
 85                    {"grace_period": _("Grace period must be shorter than the interval.")}
 86                )
 87        if "content_type" in attrs or "object_id" in attrs:
 88            content_type = attrs.get("content_type", getattr(self.instance, "content_type", None))
 89            object_id = attrs.get("object_id", getattr(self.instance, "object_id", None))
 90            if content_type is not None and object_id is None:
 91                existing = LifecycleRule.objects.filter(
 92                    content_type=content_type, object_id__isnull=True
 93                )
 94                if self.instance:
 95                    existing = existing.exclude(pk=self.instance.pk)
 96                if existing.exists():
 97                    raise ValidationError(
 98                        {
 99                            "content_type": _(
100                                "Only one type-wide rule for each object type is allowed."
101                            )
102                        }
103                    )
104        return attrs
105
106
107class LifecycleRuleViewSet(ModelViewSet):
108    queryset = LifecycleRule.objects.all()
109    serializer_class = LifecycleRuleSerializer
110    search_fields = ["content_type__model", "reviewer_groups__name", "reviewers__username"]
111    ordering = ["name"]
112    ordering_fields = ["name", "content_type__model"]
113    filterset_fields = ["content_type__model"]