authentik.admin.files.api

  1from django.db.models import Q
  2from django.utils.translation import gettext as _
  3from drf_spectacular.utils import extend_schema
  4from guardian.shortcuts import get_objects_for_user
  5from rest_framework.exceptions import ValidationError
  6from rest_framework.fields import BooleanField, CharField, ChoiceField, FileField
  7from rest_framework.parsers import MultiPartParser
  8from rest_framework.permissions import SAFE_METHODS
  9from rest_framework.request import Request
 10from rest_framework.response import Response
 11from rest_framework.views import APIView
 12
 13from authentik.admin.files.backends.base import get_content_type
 14from authentik.admin.files.fields import FileField as AkFileField
 15from authentik.admin.files.manager import get_file_manager
 16from authentik.admin.files.usage import FileApiUsage
 17from authentik.admin.files.validation import validate_upload_file_name
 18from authentik.api.validation import validate
 19from authentik.core.api.used_by import DeleteAction, UsedBySerializer
 20from authentik.core.api.utils import PassiveSerializer, ThemedUrlsSerializer
 21from authentik.events.models import Event, EventAction
 22from authentik.lib.utils.reflection import get_apps
 23from authentik.rbac.permissions import HasPermission
 24
 25MAX_FILE_SIZE_BYTES = 25 * 1024 * 1024  # 25MB
 26
 27
 28class FileView(APIView):
 29    pagination_class = None
 30    parser_classes = [MultiPartParser]
 31
 32    def get_permissions(self):
 33        return [
 34            HasPermission(
 35                "authentik_rbac.view_media_files"
 36                if self.request.method in SAFE_METHODS
 37                else "authentik_rbac.manage_media_files"
 38            )()
 39        ]
 40
 41    class FileListParameters(PassiveSerializer):
 42        usage = ChoiceField(choices=list(FileApiUsage), default=FileApiUsage.MEDIA.value)
 43        search = CharField(required=False)
 44        manageable_only = BooleanField(required=False, default=False)
 45
 46    class FileListSerializer(PassiveSerializer):
 47        name = CharField()
 48        mime_type = CharField()
 49        url = CharField()
 50        themed_urls = ThemedUrlsSerializer(required=False, allow_null=True)
 51
 52    @extend_schema(
 53        parameters=[FileListParameters],
 54        responses={200: FileListSerializer(many=True)},
 55    )
 56    @validate(FileListParameters, location="query")
 57    def get(self, request: Request, query: FileListParameters) -> Response:
 58        """List files from storage backend."""
 59        params = query.validated_data
 60
 61        try:
 62            usage = FileApiUsage(params.get("usage", FileApiUsage.MEDIA.value))
 63        except ValueError as exc:
 64            raise ValidationError(
 65                f"Invalid usage parameter provided: {params.get('usage')}"
 66            ) from exc
 67
 68        # Backend is source of truth - list all files from storage
 69        manager = get_file_manager(usage)
 70        files = manager.list_files(manageable_only=params.get("manageable_only", False))
 71        search_query = params.get("search", "")
 72        if search_query:
 73            files = filter(lambda file: search_query in file.lower(), files)
 74        files = [
 75            FileView.FileListSerializer(
 76                data={
 77                    "name": file,
 78                    "url": manager.file_url(file, request),
 79                    "mime_type": get_content_type(file),
 80                    "themed_urls": manager.themed_urls(file, request),
 81                }
 82            )
 83            for file in files
 84        ]
 85        for file in files:
 86            file.is_valid(raise_exception=True)
 87
 88        return Response([file.data for file in files])
 89
 90    class FileUploadSerializer(PassiveSerializer):
 91        file = FileField(required=True)
 92        name = CharField(required=False, allow_blank=True)
 93        usage = CharField(required=False, default=FileApiUsage.MEDIA.value)
 94
 95    @extend_schema(
 96        request=FileUploadSerializer,
 97        responses={200: None},
 98    )
 99    @validate(FileUploadSerializer)
100    def post(self, request: Request, body: FileUploadSerializer) -> Response:
101        """Upload file to storage backend."""
102        file = body.validated_data["file"]
103        name = body.validated_data.get("name", "").strip()
104        usage_value = body.validated_data.get("usage", FileApiUsage.MEDIA.value)
105
106        # Validate file size and type
107        if file.size > MAX_FILE_SIZE_BYTES:
108            raise ValidationError(
109                {
110                    "file": [
111                        _(
112                            f"File size ({file.size}B) exceeds maximum allowed "
113                            f"size ({MAX_FILE_SIZE_BYTES}B)."
114                        )
115                    ]
116                }
117            )
118
119        try:
120            usage = FileApiUsage(usage_value)
121        except ValueError as exc:
122            raise ValidationError(f"Invalid usage parameter provided: {usage_value}") from exc
123
124        # Use original filename
125        if not name:
126            name = file.name
127
128        # Sanitize path to prevent directory traversal
129        validate_upload_file_name(name, ValidationError)
130
131        manager = get_file_manager(usage)
132
133        # Check if file already exists
134        if manager.file_exists(name):
135            raise ValidationError({"name": ["A file with this name already exists."]})
136
137        # Save to backend
138        with manager.save_file_stream(name) as f:
139            f.write(file.read())
140
141        Event.new(
142            EventAction.MODEL_CREATED,
143            model={
144                "app": "authentik_admin_files",
145                "model_name": "File",
146                "pk": name,
147                "name": name,
148                "usage": usage.value,
149                "mime_type": get_content_type(name),
150            },
151        ).from_http(request)
152
153        return Response()
154
155    class FileDeleteParameters(PassiveSerializer):
156        name = CharField()
157        usage = ChoiceField(choices=list(FileApiUsage), default=FileApiUsage.MEDIA.value)
158
159    @extend_schema(
160        parameters=[FileDeleteParameters],
161        responses={200: None},
162    )
163    @validate(FileDeleteParameters, location="query")
164    def delete(self, request: Request, query: FileDeleteParameters) -> Response:
165        """Delete file from storage backend."""
166        params = query.validated_data
167
168        validate_upload_file_name(params.get("name", ""), ValidationError)
169
170        try:
171            usage = FileApiUsage(params.get("usage", FileApiUsage.MEDIA.value))
172        except ValueError as exc:
173            raise ValidationError(
174                f"Invalid usage parameter provided: {params.get('usage')}"
175            ) from exc
176
177        manager = get_file_manager(usage)
178
179        # Delete from backend
180        manager.delete_file(params.get("name"))
181
182        # Audit log for file deletion
183        Event.new(
184            EventAction.MODEL_DELETED,
185            model={
186                "app": "authentik_admin_files",
187                "model_name": "File",
188                "pk": params.get("name"),
189                "name": params.get("name"),
190                "usage": usage.value,
191            },
192        ).from_http(request)
193
194        return Response()
195
196
197class FileUsedByView(APIView):
198    pagination_class = None
199
200    def get_permissions(self):
201        return [
202            HasPermission(
203                "authentik_rbac.view_media_files"
204                if self.request.method in SAFE_METHODS
205                else "authentik_rbac.manage_media_files"
206            )()
207        ]
208
209    class FileUsedByParameters(PassiveSerializer):
210        name = CharField()
211
212    @extend_schema(
213        parameters=[FileUsedByParameters],
214        responses={200: UsedBySerializer(many=True)},
215    )
216    @validate(FileUsedByParameters, location="query")
217    def get(self, request: Request, query: FileUsedByParameters) -> Response:
218        params = query.validated_data
219
220        models_and_fields = {}
221        for app in get_apps():
222            for model in app.get_models():
223                if model._meta.abstract:
224                    continue
225                for field in model._meta.get_fields():
226                    if isinstance(field, AkFileField):
227                        models_and_fields.setdefault(model, []).append(field.name)
228
229        used_by = []
230
231        for model, fields in models_and_fields.items():
232            app = model._meta.app_label
233            model_name = model._meta.model_name
234
235            q = Q()
236            for field in fields:
237                q |= Q(**{field: params.get("name")})
238
239            objs = get_objects_for_user(
240                request.user, f"{app}.view_{model_name}", model.objects.all()
241            )
242            objs = objs.filter(q)
243            for obj in objs:
244                serializer = UsedBySerializer(
245                    data={
246                        "app": model._meta.app_label,
247                        "model_name": model._meta.model_name,
248                        "pk": str(obj.pk),
249                        "name": str(obj),
250                        "action": DeleteAction.LEFT_DANGLING,
251                    }
252                )
253                serializer.is_valid()
254                used_by.append(serializer.data)
255
256        return Response(used_by)