authentik.blueprints.api

  1"""Serializer mixin for managed models"""
  2
  3from json import JSONDecodeError, loads
  4from typing import cast
  5
  6from django.conf import settings
  7from django.core.files.uploadedfile import InMemoryUploadedFile
  8from django.utils.translation import gettext_lazy as _
  9from drf_spectacular.utils import extend_schema, inline_serializer
 10from rest_framework.decorators import action
 11from rest_framework.exceptions import PermissionDenied, ValidationError
 12from rest_framework.fields import (
 13    BooleanField,
 14    CharField,
 15    DateTimeField,
 16    FileField,
 17)
 18from rest_framework.parsers import MultiPartParser
 19from rest_framework.request import Request
 20from rest_framework.response import Response
 21from rest_framework.serializers import ListSerializer
 22from rest_framework.viewsets import ModelViewSet
 23
 24from authentik.api.validation import validate
 25from authentik.blueprints.models import BlueprintInstance
 26from authentik.blueprints.v1.common import Blueprint
 27from authentik.blueprints.v1.importer import Importer
 28from authentik.blueprints.v1.oci import OCI_PREFIX
 29from authentik.blueprints.v1.tasks import apply_blueprint, blueprints_find_dict
 30from authentik.core.api.used_by import UsedByMixin
 31from authentik.core.api.utils import JSONDictField, ModelSerializer, PassiveSerializer
 32from authentik.core.models import User
 33from authentik.events.logs import LogEventSerializer
 34from authentik.rbac.decorators import permission_required
 35
 36
 37def get_blueprints():
 38    if settings.DEBUG:
 39        return blueprints_find_dict()
 40    return blueprints_find_dict.send().get_result(block=True)
 41
 42
 43class BlueprintUploadSerializer(PassiveSerializer):
 44    """Serializer to upload file"""
 45
 46    file = FileField(required=False)
 47    path = CharField(required=False)
 48    context = CharField(required=False, allow_blank=True)
 49
 50    def validate_path(self, path: str) -> str:
 51        """Ensure the path (if set) specified is retrievable"""
 52        if path == "":
 53            return path
 54        files: list[dict] = get_blueprints()
 55        if path not in [file["path"] for file in files]:
 56            raise ValidationError(_("Blueprint file does not exist"))
 57        return path
 58
 59    def validate_context(self, context: str) -> dict:
 60        """Parse context as a JSON object"""
 61        if not context:
 62            return {}
 63        try:
 64            parsed = loads(context)
 65        except JSONDecodeError as exc:
 66            raise ValidationError(_("Context must be valid JSON")) from exc
 67        if not isinstance(parsed, dict):
 68            raise ValidationError(_("Context must be a JSON object"))
 69        return parsed
 70
 71
 72class ManagedSerializer:
 73    """Managed Serializer"""
 74
 75    managed = CharField(read_only=True, allow_null=True)
 76
 77
 78class MetadataSerializer(PassiveSerializer):
 79    """Serializer for blueprint metadata"""
 80
 81    name = CharField()
 82    labels = JSONDictField()
 83
 84
 85class BlueprintInstanceSerializer(ModelSerializer):
 86    """Info about a single blueprint instance file"""
 87
 88    def validate_path(self, path: str) -> str:
 89        """Ensure the path (if set) specified is retrievable"""
 90        if path == "" or path.startswith(OCI_PREFIX):
 91            return path
 92        files: list[dict] = get_blueprints()
 93        if path not in [file["path"] for file in files]:
 94            raise ValidationError(_("Blueprint file does not exist"))
 95        return path
 96
 97    def validate_content(self, content: str) -> str:
 98        """Ensure content (if set) is a valid blueprint"""
 99        if content == "":
100            return content
101        context = self.instance.context if self.instance else {}
102        valid, logs = Importer.from_string(content, context).validate()
103        if not valid:
104            raise ValidationError(
105                [
106                    _("Failed to validate blueprint"),
107                    *[f"- {x.event}" for x in logs],
108                ]
109            )
110        return content
111
112    def validate(self, attrs: dict) -> dict:
113        if attrs.get("path", "") == "" and attrs.get("content", "") == "":
114            raise ValidationError(_("Either path or content must be set."))
115        return super().validate(attrs)
116
117    class Meta:
118        model = BlueprintInstance
119        fields = [
120            "pk",
121            "name",
122            "path",
123            "context",
124            "last_applied",
125            "last_applied_hash",
126            "status",
127            "enabled",
128            "managed_models",
129            "metadata",
130            "content",
131        ]
132        extra_kwargs = {
133            "status": {"read_only": True},
134            "last_applied": {"read_only": True},
135            "last_applied_hash": {"read_only": True},
136            "managed_models": {"read_only": True},
137            "metadata": {"read_only": True},
138        }
139
140
141def check_blueprint_perms(blueprint: Blueprint, user: User, explicit_action: str | None = None):
142    """Check for individual permissions for each model in a blueprint"""
143    for entry in blueprint.iter_entries():
144        full_model = entry.get_model(blueprint)
145        app, __, model = full_model.partition(".")
146        perms = [
147            f"{app}.add_{model}",
148            f"{app}.change_{model}",
149            f"{app}.delete_{model}",
150        ]
151        if explicit_action:
152            perms = [f"{app}.{explicit_action}_{model}"]
153        for perm in perms:
154            if not user.has_perm(perm):
155                raise PermissionDenied(
156                    {
157                        entry.id: _(
158                            "User lacks permission to create {model}".format_map(
159                                {
160                                    "model": full_model,
161                                }
162                            )
163                        )
164                    }
165                )
166
167
168class BlueprintInstanceViewSet(UsedByMixin, ModelViewSet):
169    """Blueprint instances"""
170
171    serializer_class = BlueprintInstanceSerializer
172    queryset = BlueprintInstance.objects.all()
173    search_fields = ["name", "path"]
174    filterset_fields = ["name", "path"]
175    ordering = ["name"]
176
177    class BlueprintImportResultSerializer(PassiveSerializer):
178        """Logs of an attempted blueprint import"""
179
180        logs = LogEventSerializer(many=True, read_only=True)
181        success = BooleanField(read_only=True)
182
183    @extend_schema(
184        responses={
185            200: ListSerializer(
186                child=inline_serializer(
187                    "BlueprintFile",
188                    fields={
189                        "path": CharField(),
190                        "last_m": DateTimeField(),
191                        "hash": CharField(),
192                        "meta": MetadataSerializer(required=False, read_only=True),
193                    },
194                )
195            )
196        }
197    )
198    @action(detail=False, pagination_class=None, filter_backends=[])
199    def available(self, request: Request) -> Response:
200        """Get blueprints"""
201        files: list[dict] = get_blueprints()
202        return Response(files)
203
204    @permission_required("authentik_blueprints.view_blueprintinstance")
205    @extend_schema(
206        request=None,
207        responses={
208            200: BlueprintInstanceSerializer(),
209        },
210    )
211    @action(detail=True, pagination_class=None, filter_backends=[], methods=["POST"])
212    def apply(self, request: Request, *args, **kwargs) -> Response:
213        """Apply a blueprint"""
214        blueprint = self.get_object()
215        apply_blueprint.send_with_options(args=(blueprint.pk,), rel_obj=blueprint)
216        return self.retrieve(request, *args, **kwargs)
217
218    @extend_schema(
219        request={"multipart/form-data": BlueprintUploadSerializer},
220        responses={200: BlueprintImportResultSerializer},
221    )
222    @action(url_path="import", detail=False, methods=["POST"], parser_classes=(MultiPartParser,))
223    @validate(
224        BlueprintUploadSerializer,
225    )
226    def import_(self, request: Request, body: BlueprintUploadSerializer) -> Response:
227        """Import blueprint from .yaml file and apply it once, without creating an instance"""
228        string_contents = ""
229        if body.validated_data.get("file"):
230            file = cast(InMemoryUploadedFile, body.validated_data["file"])
231            string_contents = file.read().decode()
232        elif body.validated_data.get("path"):
233            string_contents = BlueprintInstance(
234                path=body.validated_data.get("path")
235            ).retrieve_file()
236        else:
237            raise ValidationError("Either path or file must be set")
238        context = body.validated_data.get("context") or {}
239        importer = Importer.from_string(string_contents, context)
240
241        check_blueprint_perms(importer.blueprint, request.user)
242
243        valid, logs = importer.validate()
244
245        import_response = self.BlueprintImportResultSerializer(
246            data={
247                "logs": [LogEventSerializer(log).data for log in logs],
248                "success": valid,
249            }
250        )
251        import_response.is_valid(raise_exception=True)
252
253        if valid:
254            import_response.initial_data["success"] = importer.apply()
255            import_response.is_valid()
256        return Response(data=import_response.initial_data, status=200)