authentik.policies.api.bindings

  1"""policy binding API Views"""
  2
  3from collections import OrderedDict
  4
  5from django.core.exceptions import ObjectDoesNotExist
  6from django_filters.filters import BooleanFilter, ModelMultipleChoiceFilter
  7from django_filters.filterset import FilterSet
  8from rest_framework.exceptions import ValidationError
  9from rest_framework.serializers import PrimaryKeyRelatedField
 10from rest_framework.viewsets import ModelViewSet
 11
 12from authentik.core.api.groups import PartialUserSerializer
 13from authentik.core.api.used_by import UsedByMixin
 14from authentik.core.api.users import PartialGroupSerializer
 15from authentik.core.api.utils import ModelSerializer
 16from authentik.policies.api.policies import PolicySerializer
 17from authentik.policies.models import PolicyBinding, PolicyBindingModel
 18
 19
 20class PolicyBindingModelForeignKey(PrimaryKeyRelatedField):
 21    """rest_framework PrimaryKeyRelatedField which resolves
 22    model_manager's InheritanceQuerySet"""
 23
 24    def use_pk_only_optimization(self):
 25        return False
 26
 27    def to_internal_value(self, data):
 28        if self.pk_field is not None:
 29            data = self.pk_field.to_internal_value(data)
 30        try:
 31            # Due to inheritance, a direct DB lookup for the primary key
 32            # won't return anything. This is because the direct lookup
 33            # checks the PK of PolicyBindingModel (for example),
 34            # but we get given the Primary Key of the inheriting class
 35            for model in self.get_queryset().select_subclasses().all():
 36                if str(model.pk) == str(data):
 37                    return model
 38            # as a fallback we still try a direct lookup
 39            return self.get_queryset().get_subclass(pk=data)
 40        except ObjectDoesNotExist:
 41            self.fail("does_not_exist", pk_value=data)
 42        except TypeError, ValueError:
 43            self.fail("incorrect_type", data_type=type(data).__name__)
 44
 45    def to_representation(self, value):
 46        correct_model = PolicyBindingModel.objects.get_subclass(pbm_uuid=value.pbm_uuid)
 47        return correct_model.pk
 48
 49
 50class PolicyBindingSerializer(ModelSerializer):
 51    """PolicyBinding Serializer"""
 52
 53    # Because we're not interested in the PolicyBindingModel's PK but rather the subclasses PK,
 54    # we have to manually declare this field
 55    target = PolicyBindingModelForeignKey(
 56        queryset=PolicyBindingModel.objects.select_subclasses(),
 57        required=True,
 58    )
 59
 60    policy_obj = PolicySerializer(required=False, allow_null=True, read_only=True, source="policy")
 61    group_obj = PartialGroupSerializer(
 62        required=False, allow_null=True, read_only=True, source="group"
 63    )
 64    user_obj = PartialUserSerializer(required=False, allow_null=True, read_only=True, source="user")
 65
 66    class Meta:
 67        model = PolicyBinding
 68        fields = [
 69            "pk",
 70            "policy",
 71            "group",
 72            "user",
 73            "policy_obj",
 74            "group_obj",
 75            "user_obj",
 76            "target",
 77            "negate",
 78            "enabled",
 79            "order",
 80            "timeout",
 81            "failure_result",
 82        ]
 83
 84    def validate(self, attrs: OrderedDict) -> OrderedDict:
 85        """Check that either policy, group or user is set."""
 86        target: PolicyBindingModel = attrs.get("target")
 87        supported = target.supported_policy_binding_targets()
 88        supported.sort()
 89        count = sum([bool(attrs.get(x, None)) for x in supported])
 90        invalid = count > 1
 91        empty = count < 1
 92        warning = ", ".join(f"'{x}'" for x in supported)
 93        if invalid:
 94            raise ValidationError(f"Only one of {warning} can be set.")
 95        if empty:
 96            raise ValidationError(f"One of {warning} must be set.")
 97        return attrs
 98
 99
100class PolicyBindingFilter(FilterSet):
101    """Filter for PolicyBindings"""
102
103    target_in = ModelMultipleChoiceFilter(
104        field_name="target__pbm_uuid",
105        to_field_name="pbm_uuid",
106        queryset=PolicyBindingModel.objects.select_subclasses(),
107    )
108    policy__isnull = BooleanFilter("policy", "isnull")
109
110    class Meta:
111        model = PolicyBinding
112        fields = ["policy", "policy__isnull", "target", "target_in", "enabled", "order", "timeout"]
113
114
115class PolicyBindingViewSet(UsedByMixin, ModelViewSet):
116    """PolicyBinding Viewset"""
117
118    queryset = (
119        PolicyBinding.objects.all()
120        .select_related("target", "group", "user")
121        .prefetch_related("policy")
122    )  # prefetching policy so we resolve the subclass
123    serializer_class = PolicyBindingSerializer
124    search_fields = ["policy__name"]
125    filterset_class = PolicyBindingFilter
126    ordering = ["order", "pk"]
127    ordering_fields = ["order", "target__uuid", "pk"]