core: improve storage mgmt

authentik/api/tests/test_schema.py

@@ -2,7 +2,21 @@
 
 from django.urls import reverse
 from rest_framework.test import APITestCase
-from yaml import safe_load
+from yaml import add_representer, safe_load
+
+
+def represent_type(dumper, data):
+    """Custom representer for type objects"""
+    return dumper.represent_scalar("tag:yaml.org,2002:str", str(data))
+
+
+def represent_str_class(dumper, data):
+    """Custom representer for str class object (not string instances)"""
+    return dumper.represent_scalar("tag:yaml.org,2002:str", str(data))
+
+
+add_representer(type, represent_type)
+add_representer(str, represent_str_class)
 
 
 class TestSchemaGeneration(APITestCase):

authentik/core/api/applications.py

@@ -281,42 +281,48 @@
         serializer = self.get_serializer(allowed_applications, many=True)
         return self.get_paginated_response(serializer.data)
 
+    @action(
+        detail=True,
+        pagination_class=None,
+        filter_backends=[],
+        methods=["POST"],
+        parser_classes=(MultiPartParser,),
+    )
     @permission_required("authentik_core.change_application")
     @extend_schema(
         request={
             "multipart/form-data": FileUploadSerializer,
         },
         responses={
             200: OpenApiResponse(description="Success"),
-            400: OpenApiResponse(description="Bad request"),
+            400: OpenApiResponse(description="Bad request", response={"error": str}),
+            403: OpenApiResponse(description="Permission denied", response={"error": str}),
+            415: OpenApiResponse(description="Unsupported Media Type", response={"error": str}),
+            500: OpenApiResponse(description="Internal server error", response={"error": str}),
         },
     )
+    def set_icon(self, request: Request, slug: str):
+        """Set application icon"""
+        app: Application = self.get_object()
+        return set_file(request, app, "meta_icon")
+
     @action(
         detail=True,
         pagination_class=None,
         filter_backends=[],
         methods=["POST"],
-        parser_classes=(MultiPartParser,),
     )
-    def set_icon(self, request: Request, slug: str):
-        """Set application icon"""
-        app: Application = self.get_object()
-        return set_file(request, app, "meta_icon")
-
     @permission_required("authentik_core.change_application")
     @extend_schema(
         request=FilePathSerializer,
         responses={
             200: OpenApiResponse(description="Success"),
-            400: OpenApiResponse(description="Bad request"),
+            400: OpenApiResponse(description="Bad request", response={"error": str}),
+            403: OpenApiResponse(description="Permission denied", response={"error": str}),
+            415: OpenApiResponse(description="Unsupported Media Type", response={"error": str}),
+            500: OpenApiResponse(description="Internal server error", response={"error": str}),
         },
     )
-    @action(
-        detail=True,
-        pagination_class=None,
-        filter_backends=[],
-        methods=["POST"],
-    )
     def set_icon_url(self, request: Request, slug: str):
         """Set application icon (as URL)"""
         app: Application = self.get_object()

authentik/core/api/sources.py

@@ -98,7 +98,10 @@ def get_queryset(self):  # pragma: no cover
         },
         responses={
             200: OpenApiResponse(description="Success"),
-            400: OpenApiResponse(description="Bad request"),
+            400: OpenApiResponse(description="Bad request", response={"error": str}),
+            403: OpenApiResponse(description="Permission denied", response={"error": str}),
+            415: OpenApiResponse(description="Unsupported Media Type", response={"error": str}),
+            500: OpenApiResponse(description="Internal server error", response={"error": str}),
         },
     )
     @action(
@@ -118,7 +121,10 @@ def set_icon(self, request: Request, slug: str):
         request=FilePathSerializer,
         responses={
             200: OpenApiResponse(description="Success"),
-            400: OpenApiResponse(description="Bad request"),
+            400: OpenApiResponse(description="Bad request", response={"error": str}),
+            403: OpenApiResponse(description="Permission denied", response={"error": str}),
+            415: OpenApiResponse(description="Unsupported Media Type", response={"error": str}),
+            500: OpenApiResponse(description="Internal server error", response={"error": str}),
         },
     )
     @action(

authentik/core/models.py

@@ -553,9 +553,21 @@
         """Get the URL to the App Icon image. If the name is /static or starts with http
         it is returned as-is"""
         if not self.meta_icon:
+            LOGGER.debug("No meta_icon set")
             return None
+
+        LOGGER.debug(
+            "Getting meta_icon URL",
+            name=self.meta_icon.name,
+            url=self.meta_icon.url if hasattr(self.meta_icon, "url") else None,
+            storage_backend=self.meta_icon.storage.__class__.__name__,
+        )
+
         if "://" in self.meta_icon.name or self.meta_icon.name.startswith("/static"):
+            LOGGER.debug("Using direct meta_icon name", name=self.meta_icon.name)
             return self.meta_icon.name
+
+        LOGGER.debug("Using storage URL", url=self.meta_icon.url)
         return self.meta_icon.url
 
     def get_launch_url(self, user: Optional["User"] = None) -> str | None:

authentik/core/tests/test_applications_api.py

@@ -1,11 +1,14 @@
 """Test Applications API"""
 
+import io
 from json import loads
 
 from django.core.files.base import ContentFile
+from django.core.files.uploadedfile import InMemoryUploadedFile
 from django.test.client import BOUNDARY, MULTIPART_CONTENT, encode_multipart
 from django.urls import reverse
-from rest_framework.test import APITestCase
+from PIL import Image
+from rest_framework.test import APITransactionTestCase
 
 from authentik.core.models import Application
 from authentik.core.tests.utils import create_test_admin_user, create_test_flow
@@ -17,7 +20,7 @@
 from authentik.providers.saml.models import SAMLProvider
 
 
-class TestApplicationsAPI(APITestCase):
+class TestApplicationsAPI(APITransactionTestCase):
     """Test applications API"""
 
     def setUp(self) -> None:
@@ -40,6 +43,30 @@
             policy=DummyPolicy.objects.create(name="deny", result=False, wait_min=1, wait_max=2),
             order=0,
         )
+        self.test_files = []
+
+    def tearDown(self) -> None:
+        # Clean up any test files
+        for app in [self.allowed, self.denied]:
+            if app.meta_icon:
+                app.meta_icon.delete()
+        super().tearDown()
+
+    def create_test_image(self, name="test.png") -> ContentFile:
+        """Create a valid test PNG image file.
+
+        Args:
+            name: The name to give the test file
+
+        Returns:
+            ContentFile: A ContentFile containing a valid PNG image
+        """
+        # Create a small test image
+        image = Image.new("RGB", (1, 1), color="red")
+        img_io = io.BytesIO()
+        image.save(img_io, format="PNG")
+        img_io.seek(0)
+        return ContentFile(img_io.getvalue(), name=name)
 
     def test_formatted_launch_url(self):
         """Test formatted launch URL"""
@@ -58,19 +85,38 @@
         )
 
     def test_set_icon(self):
-        """Test set_icon"""
-        file = ContentFile(b"text", "name")
+        """Test set_icon and cleanup"""
+        # Create a test image file with a valid image
+        image = Image.new("RGB", (100, 100), color="red")
+        img_io = io.BytesIO()
+        image.save(img_io, format="PNG")
+        img_io.seek(0)
+        file = InMemoryUploadedFile(
+            img_io,
+            "file",
+            "test_icon.png",
+            "image/png",
+            len(img_io.getvalue()),
+            None,
+        )
         self.client.force_login(self.user)
+
+        # Test setting icon
         response = self.client.post(
             reverse(
                 "authentik_api:application-set-icon",
                 kwargs={"slug": self.allowed.slug},
             ),
-            data=encode_multipart(data={"file": file}, boundary=BOUNDARY),
+            data=encode_multipart(BOUNDARY, {"file": file}),
             content_type=MULTIPART_CONTENT,
         )
-        self.assertEqual(response.status_code, 200)
+        self.assertEqual(
+            response.status_code,
+            200,
+            msg=f"Unexpected status code: {response.status_code}, Response: {response.content}",
+        )
 
+        # Verify icon was set correctly
         app_raw = self.client.get(
             reverse(
                 "authentik_api:application-detail",
@@ -80,7 +126,40 @@
         app = loads(app_raw.content)
         self.allowed.refresh_from_db()
         self.assertEqual(self.allowed.get_meta_icon, app["meta_icon"])
-        self.assertEqual(self.allowed.meta_icon.read(), b"text")
+        file.seek(0)
+        self.assertEqual(self.allowed.meta_icon.read(), file.read())
+
+        # Test icon replacement
+        new_image = Image.new("RGB", (100, 100), color="blue")
+        new_img_io = io.BytesIO()
+        new_image.save(new_img_io, format="PNG")
+        new_img_io.seek(0)
+        new_file = InMemoryUploadedFile(
+            new_img_io,
+            "file",
+            "new_icon.png",
+            "image/png",
+            len(new_img_io.getvalue()),
+            None,
+        )
+        response = self.client.post(
+            reverse(
+                "authentik_api:application-set-icon",
+                kwargs={"slug": self.allowed.slug},
+            ),
+            data=encode_multipart(BOUNDARY, {"file": new_file}),
+            content_type=MULTIPART_CONTENT,
+        )
+        self.assertEqual(
+            response.status_code,
+            200,
+            msg=f"Unexpected status code: {response.status_code}, Response: {response.content}",
+        )
+
+        # Verify new icon was set and old one was cleaned up
+        self.allowed.refresh_from_db()
+        new_file.seek(0)
+        self.assertEqual(self.allowed.meta_icon.read(), new_file.read())
 
     def test_check_access(self):
         """Test check_access operation"""

authentik/flows/api/flows.py

@@ -242,7 +242,10 @@ def diagram(self, request: Request, slug: str) -> Response:
         },
         responses={
             200: OpenApiResponse(description="Success"),
-            400: OpenApiResponse(description="Bad request"),
+            400: OpenApiResponse(description="Bad request", response={"error": str}),
+            403: OpenApiResponse(description="Permission denied", response={"error": str}),
+            415: OpenApiResponse(description="Unsupported Media Type", response={"error": str}),
+            500: OpenApiResponse(description="Internal server error", response={"error": str}),
         },
     )
     @action(
@@ -262,7 +265,10 @@ def set_background(self, request: Request, slug: str):
         request=FilePathSerializer,
         responses={
             200: OpenApiResponse(description="Success"),
-            400: OpenApiResponse(description="Bad request"),
+            400: OpenApiResponse(description="Bad request", response={"error": str}),
+            403: OpenApiResponse(description="Permission denied", response={"error": str}),
+            415: OpenApiResponse(description="Unsupported Media Type", response={"error": str}),
+            500: OpenApiResponse(description="Internal server error", response={"error": str}),
         },
     )
     @action(

authentik/lib/utils/file.py

@@ -1,5 +1,8 @@
 """file utils"""
 
+import os
+
+from django.core.exceptions import SuspiciousOperation
 from django.db.models import Model
 from django.http import HttpResponseBadRequest
 from rest_framework.fields import BooleanField, CharField, FileField
@@ -12,6 +15,15 @@
 LOGGER = get_logger()
 
 
+class FileValidationError(SuspiciousOperation):
+    """Custom exception for file validation errors."""
+
+    def __init__(self, message: str, status_code: int = 400):
+        super().__init__(message)
+        self.status_code = status_code
+        self.user_message = message
+
+
 class FileUploadSerializer(PassiveSerializer):
     """Serializer to upload file"""
 
@@ -30,19 +42,55 @@
     field = getattr(obj, field_name)
     file = request.FILES.get("file", None)
     clear = request.data.get("clear", "false").lower() == "true"
+
+    # If clearing or replacing, delete the old file first
+    if (clear or file) and field:
+        try:
+            LOGGER.debug(
+                "Deleting old file before setting new one",
+                field_name=field_name,
+                old_file=field.name if field else None,
+            )
+            # Delete old file but don't save model yet
+            field.delete(save=False)
+        except Exception as exc:
+            LOGGER.warning("Failed to delete old file", exc=exc)
+
     if clear:
-        # .delete() saves the model by default
-        field.delete()
+        # Save model after clearing
+        obj.save()
         return Response({})
+
     if file:
+        # Get the upload_to path from the model field
+        upload_to = field.field.upload_to
+        # If upload_to is set, ensure the file name includes the directory
+        if upload_to:
+            # Use basename to strip any path components from the filename
+            base_name = os.path.basename(file.name)
+            # Construct a clean path within the upload directory
+            file.name = f"{upload_to}/{base_name}"
         setattr(obj, field_name, file)
         try:
             obj.save()
+        except FileValidationError as exc:
+            LOGGER.warning(
+                "File validation failed",
+                error=exc.user_message,
+                status_code=exc.status_code,
+                field=field_name,
+            )
+            return Response({"error": exc.user_message}, status=exc.status_code)
         except PermissionError as exc:
             LOGGER.warning("Failed to save file", exc=exc)
-            return HttpResponseBadRequest()
+            return Response({"error": "Permission denied saving file"}, status=403)
+        except Exception as exc:
+            LOGGER.error("Unexpected error saving file", exc=exc)
+            return Response(
+                {"error": "An unexpected error occurred while saving the file"}, status=500
+            )
         return Response({})
-    return HttpResponseBadRequest()
+    return Response({"error": "No file provided"}, status=400)
 
 
 def set_file_url(request: Request, obj: Model, field: str):