✨(backend) add file upload feature & tests

For the coming features we will need to store files on the meet side.
(for instance user backgrounds).

This commits adds a new Model to manage files, and the associated
serializers & viewsets. All are tested.

This work was heavily inspired by the work done by our friends at
https://github.com/suitenumerique/drive
It build on the same architecture design (upload directly to S3 but
download goes through our proxy), but model is much much simplier
(no folders, no file sharing, etc.).

src/backend/core/api/filters.py

@@ -0,0 +1,67 @@
+"""API filters for meet' core application."""
+
+from django.utils.translation import gettext_lazy as _
+
+import django_filters
+from django_filters import BooleanFilter
+
+from core import models
+
+
+class FileFilter(django_filters.FilterSet):
+    """
+    Custom filter for filtering files.
+    """
+
+    class Meta:
+        model = models.File
+        fields = ["type"]
+
+
+class ListFileFilter(FileFilter):
+    """Filter class dedicated to the file viewset list method."""
+
+    is_creator_me = django_filters.BooleanFilter(
+        method="filter_is_creator_me", label=_("Creator is me")
+    )
+
+    is_deleted = BooleanFilter(field_name="deleted_at", method="filter_is_deleted")
+
+    class Meta:
+        model = models.File
+        fields = ["is_creator_me", "type", "upload_state", "is_deleted"]
+
+    def filter_is_deleted(self, queryset, name, value):
+        """
+        Filter files based on whether they are deleted or not.
+
+        Example:
+            - /api/v1.0/files/?is_deleted=false
+                → Filters files that were not deleted
+        """
+        if value is None:
+            return queryset
+
+        lookup = "__".join([name, "isnull"])
+        return queryset.filter(**{lookup: not value})
+
+    # pylint: disable=unused-argument
+    def filter_is_creator_me(self, queryset, name, value):
+        """
+        Filter files based on the `creator` being the current user.
+
+        Example:
+            - /api/v1.0/files/?is_creator_me=true
+                → Filters files created by the logged-in user
+            - /api/v1.0/files/?is_creator_me=false
+                → Filters files created by other users
+        """
+        user = self.request.user
+
+        if not user.is_authenticated:
+            return queryset
+
+        if value:
+            return queryset.filter(creator=user)
+
+        return queryset.exclude(creator=user)

src/backend/core/api/permissions.py

@@ -1,5 +1,7 @@
 """Permission handlers for the Meet core app."""
 
+from django.http import Http404
+
 from rest_framework import permissions
 
 from ..models import RoleChoices
@@ -106,3 +108,23 @@ class HasLiveKitRoomAccess(permissions.BasePermission):
         if not request.auth or not hasattr(request.auth, "video"):
             return False
         return request.auth.video.room == str(obj.id)
+
+
+class FilePermission(IsAuthenticated):
+    """
+    Permissions applying to the file API endpoint.
+    Handling soft deletions specificities
+    """
+
+    def has_object_permission(self, request, view, obj):
+        """
+        Return a 404 on deleted files or if the user is not the owner
+        """
+
+        if obj.deleted_at is not None or obj.hard_deleted_at is not None:
+            raise Http404
+
+        if obj.creator != request.user:
+            raise Http404
+
+        return obj.get_abilities(request.user).get(view.action, False)

src/backend/core/api/serializers.py

@@ -1,10 +1,15 @@
 """Client serializers for the Meet core app."""
 
 # pylint: disable=abstract-method,no-name-in-module
+import logging
+from os.path import splitext
 from typing import Literal
+from urllib.parse import quote
 
 from django.conf import settings
 from django.core.exceptions import SuspiciousOperation
+
+# pylint: disable=abstract-method,no-name-in-module
 from django.utils.translation import gettext_lazy as _
 
 from django_pydantic_field.rest_framework import SchemaField
@@ -15,6 +20,8 @@ from timezone_field.rest_framework import TimeZoneSerializerField
 
 from core import models, utils
 
+logger = logging.getLogger(__name__)
+
 
 class UserSerializer(serializers.ModelSerializer):
     """Serialize users."""
@@ -27,6 +34,15 @@ class UserSerializer(serializers.ModelSerializer):
         read_only_fields = ["id", "email", "full_name", "short_name"]
 
 
+class UserLightSerializer(serializers.ModelSerializer):
+    """Serialize users with limited fields."""
+
+    class Meta:
+        model = models.User
+        fields = ["id", "full_name", "short_name"]
+        read_only_fields = ["id", "full_name", "short_name"]
+
+
 class ResourceAccessSerializerMixin:
     """
     A serializer mixin to share controlling that the logged-in user submitting a room access object
@@ -377,3 +393,143 @@ class UpdateParticipantSerializer(BaseParticipantsManagementSerializer):
             )
 
         return attrs
+
+
+class ListFileSerializer(serializers.ModelSerializer):
+    """Serialize File model for the API."""
+
+    url = serializers.SerializerMethodField(read_only=True)
+    creator = UserLightSerializer(read_only=True)
+    abilities = serializers.SerializerMethodField(read_only=True)
+
+    class Meta:
+        model = models.File
+        fields = [
+            "id",
+            "created_at",
+            "updated_at",
+            "title",
+            "type",
+            "creator",
+            "deleted_at",
+            "hard_deleted_at",
+            "filename",
+            "upload_state",
+            "mimetype",
+            "size",
+            "description",
+            "url",
+            "abilities",
+        ]
+        read_only_fields = [
+            "id",
+            "created_at",
+            "updated_at",
+            "creator",
+            "deleted_at",
+            "hard_deleted_at",
+            "filename",
+            "upload_state",
+            "mimetype",
+            "size",
+            "url",
+            "abilities",
+        ]
+
+    def get_url(self, obj):
+        """Return the URL of the file."""
+        if obj.is_pending_upload:
+            return None
+
+        return f"{settings.MEDIA_BASE_URL}{settings.MEDIA_URL}{quote(obj.file_key)}"
+
+    def get_abilities(self, file) -> dict:
+        """Return abilities of the logged-in user on the instance."""
+        request = self.context.get("request")
+        if not request:
+            return {}
+
+        return file.get_abilities(request.user)
+
+
+class FileSerializer(ListFileSerializer):
+    """Default serializer File model for the API."""
+
+    def create(self, validated_data):
+        raise NotImplementedError("Create method can not be used.")
+
+
+class CreateFileSerializer(ListFileSerializer):
+    """Serializer used to create a new file"""
+
+    title = serializers.CharField(max_length=255, required=False)
+    policy = serializers.SerializerMethodField()
+
+    class Meta:
+        model = models.File
+        fields = [*ListFileSerializer.Meta.fields, "policy"]
+        read_only_fields = [
+            *(
+                field
+                for field in ListFileSerializer.Meta.read_only_fields
+                if field != "filename"
+            ),
+            "policy",
+        ]
+
+    def get_fields(self):
+        """Force the id field to be writable."""
+        fields = super().get_fields()
+        fields["id"].read_only = False
+
+        return fields
+
+    def validate_id(self, value):
+        """Ensure the provided ID does not already exist when creating a new file."""
+        request = self.context.get("request")
+
+        # Only check this on POST (creation)
+        if request and models.File.objects.filter(id=value).exists():
+            raise serializers.ValidationError(
+                "A file with this ID already exists. You cannot override it.",
+                code="file_create_existing_id",
+            )
+
+        return value
+
+    def validate(self, attrs):
+        """Validate extension and fill title."""
+        # we run the default validation first to make sure the base data in attrs is ok
+        attrs = super().validate(attrs)
+
+        filename_root, ext = splitext(attrs["filename"])
+
+        if settings.FILE_UPLOAD_APPLY_RESTRICTIONS:
+            config_for_file_type = settings.FILE_UPLOAD_RESTRICTIONS[attrs["type"]]
+            if ext.lower() not in config_for_file_type["allowed_extensions"]:
+                logger.info(
+                    "create_item: file extension not allowed %s for filename %s",
+                    ext,
+                    attrs["filename"],
+                )
+                raise serializers.ValidationError(
+                    {"filename": _("This file extension is not allowed.")},
+                    code="item_create_file_extension_not_allowed",
+                )
+
+        # The title will be the filename if not provided
+        if not attrs.get("title", None):
+            attrs["title"] = filename_root
+
+        return attrs
+
+    def get_policy(self, file):
+        """Return the policy to use if the item is a file."""
+
+        if file.upload_state == models.FileUploadStateChoices.READY:
+            return None
+
+        return utils.generate_upload_policy(file)
+
+    def update(self, instance, validated_data):
+        raise NotImplementedError("Update method can not be used.")

src/backend/core/api/viewsets.py

@@ -1,16 +1,26 @@
 """API endpoints"""
+# pylint: disable=too-many-lines
 
+import re
 import uuid
 from logging import getLogger
-from urllib.parse import urlparse
+from urllib.parse import unquote, urlparse
 
 from django.conf import settings
+from django.core.files.storage import default_storage
 from django.db.models import Q
 from django.http import Http404
 from django.shortcuts import get_object_or_404
 from django.utils.text import slugify
 
-from rest_framework import decorators, mixins, pagination, viewsets
+from django_filters import rest_framework as django_filters
+from rest_framework import (
+    decorators,
+    filters,
+    mixins,
+    pagination,
+    viewsets,
+)
 from rest_framework import (
     exceptions as drf_exceptions,
 )
@@ -22,6 +32,7 @@ from rest_framework import (
 )
 
 from core import enums, models, utils
+from core.api.filters import ListFileFilter
 from core.recording.enums import FileExtension
 from core.recording.event.authentication import StorageEventAuthentication
 from core.recording.event.exceptions import (
@@ -56,6 +67,7 @@ from core.services.participants_management import (
 )
 from core.services.room_creation import RoomCreation
 from core.services.subtitle import SubtitleException, SubtitleService
+from core.tasks.file import process_file_deletion
 
 from ..authentication.livekit import LiveKitTokenAuthentication
 from . import permissions, serializers, throttling
@@ -66,6 +78,17 @@ from .feature_flag import FeatureFlag
 logger = getLogger(__name__)
 
 
+FILE_FOLDER = settings.FILE_UPLOAD_PATH
+UUID_REGEX = (
+    r"[a-fA-F0-9]{8}-[a-fA-F0-9]{4}-[a-fA-F0-9]{4}-[a-fA-F0-9]{4}-[a-fA-F0-9]{12}"
+)
+FILE_EXT_REGEX = r"[\d\w]+"
+MEDIA_STORAGE_URL_PATTERN = re.compile(
+    f"{settings.MEDIA_URL:s}"
+    rf"(?P<key>{FILE_FOLDER:s}/(?P<pk>{UUID_REGEX:s})/\.{FILE_EXT_REGEX:s})$"
+)
+
+
 class NestedGenericViewSet(viewsets.GenericViewSet):
     """
     A generic Viewset aims to be used in a nested route context.
@@ -77,20 +100,20 @@ class NestedGenericViewSet(viewsets.GenericViewSet):
     lookup_fields: list[str] = ["pk"]
     lookup_url_kwargs: list[str] = []
 
-    def __getattribute__(self, item):
+    def __getattribute__(self, file):
         """
         This method is overridden to allow to get the last lookup field or lookup url kwarg
         when accessing the `lookup_field` or `lookup_url_kwarg` attribute. This is useful
         to keep compatibility with all methods used by the parent class `GenericViewSet`.
         """
-        if item in ["lookup_field", "lookup_url_kwarg"]:
-            return getattr(self, item + "s", [None])[-1]
+        if file in ["lookup_field", "lookup_url_kwarg"]:
+            return getattr(self, file + "s", [None])[-1]
 
-        return super().__getattribute__(item)
+        return super().__getattribute__(file)
 
     def get_queryset(self):
         """
-        Get the list of items for this view.
+        Get the list of files for this view.
 
         `lookup_fields` attribute is enumerated here to perform the nested lookup.
         """
@@ -793,7 +816,7 @@ class RecordingViewSet(
         # Extract the original URL from the request header
         original_url = request.META.get("HTTP_X_ORIGINAL_URL")
         if not original_url:
-            logger.debug("Missing HTTP_X_ORIGINAL_URL header in subrequest")
+            logger.warning("Missing HTTP_X_ORIGINAL_URL header in subrequest")
             raise drf_exceptions.PermissionDenied()
 
         logger.debug("Original url: '%s'", original_url)
@@ -810,7 +833,7 @@ class RecordingViewSet(
         try:
             return match.groupdict()
         except (ValueError, AttributeError) as exc:
-            logger.debug("Failed to extract parameters from subrequest URL: %s", exc)
+            logger.warning("Failed to extract parameters from subrequest URL: %s", exc)
             raise drf_exceptions.PermissionDenied() from exc
 
     @decorators.action(detail=False, methods=["get"], url_path="media-auth")
@@ -834,7 +857,7 @@ class RecordingViewSet(
         recording_id = url_params["recording_id"]
 
         extension = url_params["extension"]
-        if extension not in [item.value for item in FileExtension]:
+        if extension not in [file.value for file in FileExtension]:
             raise drf_exceptions.ValidationError({"detail": "Unsupported extension."})
 
         try:
@@ -858,3 +881,302 @@ class RecordingViewSet(
         request = utils.generate_s3_authorization_headers(recording.key)
 
         return drf_response.Response("authorized", headers=request.headers, status=200)
+
+
+# pylint: disable=too-many-public-methods
+class FileViewSet(
+    SerializerPerActionMixin,
+    mixins.CreateModelMixin,
+    mixins.DestroyModelMixin,
+    mixins.UpdateModelMixin,
+    mixins.ListModelMixin,
+    viewsets.GenericViewSet,
+):
+    """
+    FileViewSet API.
+
+    This viewset provides CRUD operations and additional actions for managing files.
+
+    ### API Endpoints:
+    1. **List**: Retrieve a paginated list of files.
+       Example: GET /files/?page=2
+    2. **Retrieve**: Get a specific file by its ID.
+       Example: GET /files/{id}/
+    3. **Create**: Create a new file.
+       Example: POST /files/
+    4. **Update**: Update a file by its ID.
+       Example: PUT /files/{id}/
+    5. **Delete**: Soft delete a file by its ID.
+       Example: DELETE /files/{id}/
+
+
+    ### Ordering: created_at, updated_at, title
+
+        Example:
+        - Ascending: GET /api/v1.0/files/?ordering=created_at
+
+    ### Filtering:
+        - `is_creator_me=true`: Returns files created by the current user.
+        - `is_creator_me=false`: Returns files created by other users.
+        - `is_deleted=false`: Returns files that are not (soft) deleted
+
+        Example:
+        - GET /api/v1.0/files/?is_creator_me=true
+        - GET /api/v1.0/files/?is_creator_me=false&is_deleted=false
+
+    ### Notes:
+    - Implements soft delete logic to retain file
+    """
+
+    ordering = ["-updated_at"]
+    ordering_fields = ["created_at", "updated_at", "title"]
+    pagination_class = Pagination
+    permission_classes = [
+        permissions.FilePermission,
+    ]
+    queryset = models.File.objects.filter(hard_deleted_at__isnull=True)
+    default_serializer_class = serializers.FileSerializer
+    serializer_classes = {
+        "list": serializers.ListFileSerializer,
+        "create": serializers.CreateFileSerializer,
+    }
+    filter_backends = (django_filters.DjangoFilterBackend, filters.OrderingFilter)
+    filterset_class = ListFileFilter
+
+    def get_queryset(self):
+        """Get queryset that defaults to the the current request user."""
+        user = self.request.user
+        queryset = super().get_queryset().select_related("creator")
+
+        if not user.is_authenticated:
+            return queryset.none()
+
+        # For now, we force the filtering on the current user in all cases, might evolve later
+        queryset = queryset.filter(creator=user)
+        return queryset
+
+    def get_response_for_queryset(self, queryset, context=None):
+        """Return paginated response for the queryset if requested."""
+        context = context or self.get_serializer_context()
+        page = self.paginate_queryset(queryset)
+        if page is not None:
+            serializer = self.get_serializer(page, many=True, context=context)
+            result = self.get_paginated_response(serializer.data)
+            return result
+
+        serializer = self.get_serializer(queryset, many=True, context=context)
+        return drf_response.Response(serializer.data)
+
+    def perform_create(self, serializer):
+        """Set the current user as creator of the newly created file."""
+        serializer.save(creator=self.request.user)
+
+    def perform_destroy(self, instance):
+        """Override to implement a soft delete instead of dumping the record in database."""
+        instance.soft_delete()
+
+    @decorators.action(detail=True, methods=["post"], url_path="upload-ended")
+    def upload_ended(self, request, *args, **kwargs):
+        """
+        Check the actual uploaded file and mark it as ready.
+        """
+
+        file = self.get_object()
+
+        if not file.is_pending_upload:
+            raise drf_exceptions.ValidationError(
+                {"file": "This action is only available for files in PENDING state."},
+                code="file_upload_state_not_pending",
+            )
+
+        s3_client = default_storage.connection.meta.client
+
+        head_response = s3_client.head_object(
+            Bucket=default_storage.bucket_name, Key=file.file_key
+        )
+        file_size = head_response["ContentLength"]
+
+        if settings.FILE_UPLOAD_APPLY_RESTRICTIONS:
+            config_for_file_type = settings.FILE_UPLOAD_RESTRICTIONS[file.type]
+            if file_size > config_for_file_type["max_size"]:
+                self._complete_file_deletion(file)
+                logger.info(
+                    "upload_ended: file size (%s) for file %s higher than the allowed max size",
+                    file_size,
+                    file.file_key,
+                )
+                raise drf_exceptions.ValidationError(
+                    detail="The file size is higher than the allowed max size.",
+                    code="file_size_exceeded",
+                )
+
+        # python-magic recommends using at least the first 2048 bytes
+        # to reduce incorrect identification.
+        # This is a tradeoff between pulling in the whole file and the most likely relevant bytes
+        # of the file for mime type identification.
+        if file_size > 2048:
+            range_response = s3_client.get_object(
+                Bucket=default_storage.bucket_name,
+                Key=file.file_key,
+                Range="bytes=0-2047",
+            )
+            file_head = range_response["Body"].read()
+        else:
+            file_head = s3_client.get_object(
+                Bucket=default_storage.bucket_name, Key=file.file_key
+            )["Body"].read()
+
+        # Use improved MIME type detection combining magic bytes and file extension
+        logger.info("upload_ended: detecting mimetype for file: %s", file.file_key)
+        mimetype = utils.detect_mimetype(file_head, filename=file.filename)
+
+        if settings.FILE_UPLOAD_APPLY_RESTRICTIONS:
+            config_for_file_type = settings.FILE_UPLOAD_RESTRICTIONS[file.type]
+            allowed_file_mimetypes = config_for_file_type["allowed_mimetypes"]
+            if mimetype not in allowed_file_mimetypes:
+                self._complete_file_deletion(file)
+                logger.warning(
+                    "upload_ended: mimetype not allowed %s for file %s",
+                    mimetype,
+                    file.file_key,
+                )
+                raise drf_exceptions.ValidationError(
+                    detail="The file type is not allowed.",
+                    code="file_type_not_allowed",
+                )
+
+        file.upload_state = models.FileUploadStateChoices.READY
+        file.mimetype = mimetype
+        file.size = file_size
+
+        file.save(update_fields=["upload_state", "mimetype", "size"])
+
+        if head_response["ContentType"] != mimetype:
+            logger.info(
+                "upload_ended: content type mismatch between object storage and file,"
+                " updating from %s to %s",
+                head_response["ContentType"],
+                mimetype,
+            )
+            s3_client.copy_object(
+                Bucket=default_storage.bucket_name,
+                Key=file.file_key,
+                CopySource={
+                    "Bucket": default_storage.bucket_name,
+                    "Key": file.file_key,
+                },
+                ContentType=mimetype,
+                Metadata=head_response["Metadata"],
+                MetadataDirective="REPLACE",
+            )
+
+        # Not yet implemented
+        # Change the file.upload_state when this will be done
+        # malware_detection.analyse_file(file.file_key, file_id=file.id)
+
+        serializer = self.get_serializer(file)
+
+        return drf_response.Response(serializer.data, status=drf_status.HTTP_200_OK)
+
+    def _complete_file_deletion(self, file):
+        """Delete a file completely."""
+        file.soft_delete()
+        file.hard_delete()
+        process_file_deletion.delay(file.id)
+
+    def _authorize_subrequest(self, request, pattern):
+        """
+        Authorize access based on the original URL of an Nginx subrequest
+        and user permissions. Returns a dictionary of URL parameters if authorized.
+
+        The original url is passed by nginx in the "HTTP_X_ORIGINAL_URL" header.
+        See corresponding ingress configuration in Helm chart and read about the
+        nginx.ingress.kubernetes.io/auth-url annotation to understand how the Nginx ingress
+        is configured to do this.
+
+        Based on the original url and the logged in user, we must decide if we authorize Nginx
+        to let this request go through (by returning a 200 code) or if we block it (by returning
+        a 403 error). Note that we return 403 errors without any further details for security
+        reasons.
+
+        Parameters:
+        - pattern: The regex pattern to extract identifiers from the URL.
+
+        Returns:
+        - A dictionary of URL parameters if the request is authorized.
+        Raises:
+        - PermissionDenied if authorization fails.
+        """
+        # Extract the original URL from the request header
+        original_url = request.META.get("HTTP_X_ORIGINAL_URL")
+        if not original_url:
+            logger.warning("Missing HTTP_X_ORIGINAL_URL header in subrequest")
+            raise drf_exceptions.PermissionDenied()
+
+        parsed_url = urlparse(original_url)
+        match = pattern.search(unquote(parsed_url.path))
+
+        if not match:
+            logger.warning(
+                "Subrequest URL '%s' did not match pattern '%s'",
+                parsed_url.path,
+                pattern,
+            )
+            raise drf_exceptions.PermissionDenied()
+
+        try:
+            url_params = match.groupdict()
+        except (ValueError, AttributeError) as exc:
+            logger.warning("Failed to extract parameters from subrequest URL: %s", exc)
+            raise drf_exceptions.PermissionDenied() from exc
+
+        pk = url_params.get("pk")
+        if not pk:
+            logger.warning("File ID (pk) not found in URL parameters: %s", url_params)
+            raise drf_exceptions.PermissionDenied()
+
+        # Fetch the file and check if the user has access
+        queryset = models.File.objects.all()
+        # No suspicious analysis implemented yet
+        # queryset = self._filter_suspicious_files(queryset, request.user)
+        try:
+            file = queryset.get(pk=pk)
+        except models.File.DoesNotExist as exc:
+            logger.warning("File with ID '%s' does not exist", pk)
+            raise drf_exceptions.PermissionDenied() from exc
+
+        user_abilities = file.get_abilities(request.user)
+        if not user_abilities.get(self.action, False):
+            logger.warning(
+                "User '%s' lacks permission for file '%s'", request.user.id, pk
+            )
+            raise drf_exceptions.PermissionDenied()
+
+        logger.debug(
+            "Subrequest authorization successful. Extracted parameters: %s", url_params
+        )
+        return url_params, request.user.id, file
+
+    @decorators.action(detail=False, methods=["get"], url_path="media-auth")
+    def media_auth(self, request, *args, **kwargs):
+        """
+        This view is used by an Nginx subrequest to control access to an file's
+        attachment file.
+
+        When we let the request go through, we compute authorization headers that will be added to
+        the request going through thanks to the nginx.ingress.kubernetes.io/auth-response-headers
+        annotation. The request will then be proxied to the object storage backend who will
+        respond with the file after checking the signature included in headers.
+        """
+        url_params, _, file = self._authorize_subrequest(
+            request, MEDIA_STORAGE_URL_PATTERN
+        )
+
+        if file.is_pending_upload:
+            logger.warning("File '%s' is not ready", file.id)
+            raise drf_exceptions.PermissionDenied()
+
+        # Generate S3 authorization headers using the extracted URL parameters
+        request = utils.generate_s3_authorization_headers(f"{url_params.get('key'):s}")
+
+        return drf_response.Response("authorized", headers=request.headers, status=200)

src/backend/core/factories.py

@@ -2,8 +2,11 @@
 Core application factories
 """
 
+from io import BytesIO
+
 from django.conf import settings
 from django.contrib.auth.hashers import make_password
+from django.core.files.storage import default_storage
 from django.utils.text import slugify
 
 import factory.fuzzy
@@ -153,3 +156,42 @@ class ApplicationDomainFactory(factory.django.DjangoModelFactory):
 
     domain = factory.Faker("domain_name")
     application = factory.SubFactory(ApplicationFactory)
+
+
+class FileFactory(factory.django.DjangoModelFactory):
+    """A factory to create files"""
+
+    class Meta:
+        model = models.File
+        skip_postgeneration_save = True
+
+    title = factory.Sequence(lambda n: f"file{n}")
+    creator = factory.SubFactory(UserFactory)
+    deleted_at = None
+    type = factory.fuzzy.FuzzyChoice([t[0] for t in models.FileTypeChoices.choices])
+    filename = factory.lazy_attribute(lambda o: fake.file_name())
+    upload_state = None
+    size = None
+
+    @factory.post_generation
+    def update_upload_state(self, create, extracted, **kwargs):
+        """Change the upload state of a file."""
+        if create and extracted:
+            self.upload_state = extracted
+            self.save()
+
+    @factory.post_generation
+    def upload_bytes(self, create, extracted, **kwargs):
+        """Save content of the file into the storage"""
+        if create and extracted is not None:
+            content = (
+                extracted
+                if isinstance(extracted, bytes)
+                else str(extracted).encode("utf-8")
+            )
+
+            self.filename = kwargs.get("filename", self.filename or "content.txt")
+            self.size = len(content)
+            self.save()
+
+            default_storage.save(self.file_key, BytesIO(content))

src/backend/core/migrations/0017_file.py

@@ -0,0 +1,42 @@
+# Generated by Django 5.2.11 on 2026-03-03 15:22
+
+import django.db.models.deletion
+import uuid
+from django.conf import settings
+from django.db import migrations, models
+
+
+class Migration(migrations.Migration):
+
+    dependencies = [
+        ('core', '0016_recording_options'),
+    ]
+
+    operations = [
+        migrations.CreateModel(
+            name='File',
+            fields=[
+                ('id', models.UUIDField(default=uuid.uuid4, editable=False, help_text='primary key for the record as UUID', primary_key=True, serialize=False, verbose_name='id')),
+                ('created_at', models.DateTimeField(auto_now_add=True, help_text='date and time at which a record was created', verbose_name='created on')),
+                ('updated_at', models.DateTimeField(auto_now=True, help_text='date and time at which a record was last updated', verbose_name='updated on')),
+                ('type', models.CharField(choices=[('background_image', 'Background image')], max_length=25)),
+                ('title', models.CharField(max_length=255, verbose_name='title')),
+                ('deleted_at', models.DateTimeField(blank=True, null=True)),
+                ('hard_deleted_at', models.DateTimeField(blank=True, null=True)),
+                ('filename', models.CharField(max_length=255)),
+                ('upload_state', models.CharField(choices=[('pending', 'Pending'), ('ready', 'Ready')], max_length=25)),
+                ('mimetype', models.CharField(blank=True, max_length=255, null=True)),
+                ('size', models.BigIntegerField(blank=True, null=True)),
+                ('description', models.TextField(blank=True, null=True)),
+                ('malware_detection_info', models.JSONField(blank=True, default=dict, help_text='Malware detection info when the analysis status is unsafe.', null=True)),
+                ('creator', models.ForeignKey(blank=True, null=True, on_delete=django.db.models.deletion.RESTRICT, related_name='files_created', to=settings.AUTH_USER_MODEL)),
+            ],
+            options={
+                'verbose_name': 'File',
+                'verbose_name_plural': 'Files',
+                'db_table': 'file',
+                'ordering': ('created_at',),
+                'indexes': [models.Index(fields=['creator', 'type', '-created_at'], name='file_creator_730cce_idx')],
+            },
+        ),
+    ]

src/backend/core/models.py

@@ -1,11 +1,14 @@
 """
 Declare and configure the models for the Meet core application
+# pylint: disable=too-many-lines
 """
+# pylint: disable=too-many-lines
 
 import secrets
 import uuid
 from datetime import datetime, timedelta
 from logging import getLogger
+from os.path import splitext
 from typing import List, Optional
 
 from django.conf import settings
@@ -14,7 +17,7 @@ from django.contrib.auth.base_user import AbstractBaseUser
 from django.contrib.postgres.fields import ArrayField
 from django.core import mail, validators
 from django.core.exceptions import PermissionDenied, ValidationError
-from django.db import models
+from django.db import models, transaction
 from django.utils import timezone
 from django.utils.text import capfirst, slugify
 from django.utils.translation import gettext_lazy as _
@@ -829,3 +832,186 @@ class ApplicationDomain(BaseModel):
 
         self.domain = self.domain.lower().strip()
         super().save(*args, **kwargs)
+
+
+class FileUploadStateChoices(models.TextChoices):
+    """Possible states of a file."""
+
+    PENDING = "pending", _("Pending")
+    # Commented out for now, as we may need this when we implement the malware detection logic.
+    # ANALYZING = "analyzing", _("Analyzing")
+    # SUSPICIOUS = "suspicious", _("Suspicious")
+    # FILE_TOO_LARGE_TO_ANALYZE = (
+    #     "file_too_large_to_analyze",
+    #     _("File too large to analyze"),
+    # )
+    READY = "ready", _("Ready")
+
+
+class FileTypeChoices(models.TextChoices):
+    """Defines the possible types of a file."""
+
+    BACKGROUND_IMAGE = "background_image", _("Background image")
+
+
+class File(BaseModel):
+    """File uploaded by a user."""
+
+    type = models.CharField(
+        max_length=25,
+        choices=FileTypeChoices.choices,
+        null=False,
+        blank=False,
+    )
+    title = models.CharField(_("title"), max_length=255)
+    creator = models.ForeignKey(
+        User,
+        on_delete=models.RESTRICT,
+        related_name="files_created",
+        blank=True,
+        null=True,
+    )
+    deleted_at = models.DateTimeField(null=True, blank=True)
+    hard_deleted_at = models.DateTimeField(null=True, blank=True)
+
+    filename = models.CharField(max_length=255, null=False, blank=False)
+
+    upload_state = models.CharField(
+        max_length=25,
+        choices=FileUploadStateChoices.choices,
+    )
+    mimetype = models.CharField(max_length=255, null=True, blank=True)
+    size = models.BigIntegerField(null=True, blank=True)
+    description = models.TextField(null=True, blank=True)
+    malware_detection_info = models.JSONField(
+        null=True,
+        blank=True,
+        default=dict,
+        help_text=_("Malware detection info when the analysis status is unsafe."),
+    )
+
+    class Meta:
+        db_table = "file"
+        verbose_name = _("File")
+        verbose_name_plural = _("Files")
+        ordering = ("created_at",)
+        indexes = [
+            models.Index(fields=["creator", "type", "-created_at"]),
+        ]
+
+    def __str__(self):
+        return str(self.title)
+
+    def save(self, *args, **kwargs):
+        """Set the upload state to pending if it's the first save and it's a file."""
+
+        if self.created_at is None:
+            self.upload_state = FileUploadStateChoices.PENDING
+
+        return super().save(*args, **kwargs)
+
+    def delete(self, using=None, keep_parents=False):
+        if self.deleted_at is None:
+            raise RuntimeError("The file must be soft deleted before being deleted.")
+
+        return super().delete(using, keep_parents)
+
+    @property
+    def is_pending_upload(self):
+        """Return whether the file is in a pending upload state"""
+        return self.upload_state == FileUploadStateChoices.PENDING
+
+    @property
+    def extension(self):
+        """Return the extension related to the filename."""
+        if self.filename is None:
+            raise RuntimeError(
+                "The file must have a filename to compute its extension."
+            )
+
+        _, extension = splitext(self.filename)
+
+        if extension:
+            return extension.lstrip(".")
+
+        return None
+
+    @property
+    def key_base(self):
+        """Key base of the location where the file is stored in object storage."""
+        if not self.pk:
+            raise RuntimeError(
+                "The file instance must be saved before requesting a storage key."
+            )
+
+        return f"{settings.FILE_UPLOAD_PATH}/{self.pk!s}"
+
+    @property
+    def file_key(self):
+        """Key used to store the file in object storage."""
+        _, extension = splitext(self.filename)
+        # We store only the extension in the storage system to avoid
+        # leaking Personal Information in logs, etc.
+        return f"{self.key_base}/{extension!s}"
+
+    def get_abilities(self, user):
+        """
+        Compute and return abilities for a given user on the file.
+        """
+        # Characteristics that are based only on specific access
+        is_creator = user == self.creator
+        retrieve = is_creator
+        is_deleted = self.deleted_at is not None
+        can_update = is_creator and not is_deleted and user.is_authenticated
+        can_hard_delete = is_creator and user.is_authenticated
+        can_destroy = can_hard_delete and not is_deleted
+
+        return {
+            "destroy": can_destroy,
+            "hard_delete": can_hard_delete,
+            "retrieve": retrieve,
+            "media_auth": retrieve and not is_deleted,
+            "partial_update": can_update,
+            "update": can_update,
+            "upload_ended": can_update and user.is_authenticated,
+        }
+
+    @transaction.atomic
+    def soft_delete(self):
+        """
+        Soft delete the file.
+        We still keep the .delete() method untouched for programmatic purposes.
+        """
+        if self.deleted_at:
+            raise RuntimeError("This file is already deleted.")
+
+        self.deleted_at = timezone.now()
+        self.save(update_fields=["deleted_at"])
+
+    def hard_delete(self):
+        """
+        Hard delete the file.
+        We still keep the .delete() method untouched for programmatic purposes.
+        """
+        if self.hard_deleted_at:
+            raise ValidationError(
+                {
+                    "hard_deleted_at": ValidationError(
+                        _("This file is already hard deleted."),
+                        code="file_hard_delete_already_effective",
+                    )
+                }
+            )
+
+        if self.deleted_at is None:
+            raise ValidationError(
+                {
+                    "hard_deleted_at": ValidationError(
+                        _("To hard delete a file, it must first be soft deleted."),
+                        code="file_hard_delete_should_soft_delete_first",
+                    )
+                }
+            )
+
+        self.hard_deleted_at = timezone.now()
+        self.save(update_fields=["hard_deleted_at"])

src/backend/core/tasks/_task.py

@@ -0,0 +1,49 @@
+# ruff: noqa: PLC0415
+
+from django.conf import settings
+
+
+def task(*d_args, **d_kwargs):
+    """
+    Decorator compatible with Celery's @app.task, but works without Celery.
+
+    If Celery is available, returns a real Celery task.
+    If not, returns the original function and provides `.delay()`/`.apply_async()`
+    as synchronous fallbacks (so existing call sites don't break).
+
+    Notes:
+        Mostly LLM-generated.
+    """
+
+    def _fallback_wrap(func):
+        def delay(*args, **kwargs):
+            return func(*args, **kwargs)
+
+        def apply_async(args=None, kwargs=None, **_options):
+            return func(*(args or ()), **(kwargs or {}))
+
+        func.delay = delay
+        func.apply_async = apply_async
+        return func
+
+    # Handle bare decorator usage: @task
+    if len(d_args) == 1 and callable(d_args[0]) and not d_kwargs:
+        func = d_args[0]
+        if settings.CELERY_ENABLED:
+            from meet.celery_app import app as _celery_app
+
+            return _celery_app.task(func)
+        return _fallback_wrap(func)
+
+    # Handle parameterized usage: @task(...), e.g. @task(bind=True)
+    def _decorate(func):
+        if settings.CELERY_ENABLED:
+            from meet.celery_app import app as _celery_app
+
+            return _celery_app.task(*d_args, **d_kwargs)(func)
+        return _fallback_wrap(func)
+
+    return _decorate
+
+
+__all__ = ("task",)

src/backend/core/tasks/file.py

@@ -0,0 +1,36 @@
+"""
+Tasks related to files.
+"""
+
+import logging
+
+from django.core.files.storage import default_storage
+
+from core.models import File
+from core.tasks._task import task
+
+logger = logging.getLogger(__name__)
+
+
+@task
+def process_file_deletion(file_id):
+    """
+    Process the deletion of a file.
+    Definitely delete it in the database.
+    Delete the files from the storage.
+    """
+    logger.info("Processing item deletion for %s", file_id)
+    try:
+        file = File.objects.get(id=file_id)
+    except File.DoesNotExist:
+        logger.error("Item %s does not exist", file_id)
+        return
+
+    if file.hard_deleted_at is None:
+        logger.error("To process an item deletion, it must be hard deleted first.")
+        return
+
+    logger.info("Deleting file %s", file.file_key)
+    default_storage.delete(file.file_key)
+
+    file.delete()

src/backend/core/tests/files/test_api_files_create.py

@@ -0,0 +1,323 @@
+"""
+Tests for files API endpoint in meet's core app: create
+"""
+
+from concurrent.futures import ThreadPoolExecutor
+from urllib.parse import parse_qs, urlparse
+from uuid import uuid4
+
+from django.utils import timezone
+
+import pytest
+from freezegun import freeze_time
+from rest_framework import status
+from rest_framework.test import APIClient
+
+from core import factories
+from core.models import File, FileTypeChoices, FileUploadStateChoices
+
+pytestmark = pytest.mark.django_db
+
+
+def test_api_files_create_anonymous():
+    """Anonymous users should not be allowed to create items."""
+    response = APIClient().post(
+        "/api/v1.0/files/",
+        {
+            "title": "My file",
+            "type": FileTypeChoices.BACKGROUND_IMAGE,
+        },
+    )
+
+    assert response.status_code == status.HTTP_401_UNAUTHORIZED
+    assert not File.objects.exists()
+
+
+def test_api_files_create_authenticated_success():
+    """
+    Authenticated users should be able to create files and should automatically be declared
+    as the owner of the newly created file.
+    """
+    user = factories.UserFactory()
+
+    client = APIClient()
+    client.force_login(user)
+
+    response = client.post(
+        "/api/v1.0/files/",
+        {
+            "title": "my file",
+            "filename": "my_file.png",
+            "type": FileTypeChoices.BACKGROUND_IMAGE,
+        },
+        format="json",
+    )
+
+    assert response.status_code == 201, response.json()
+    file = File.objects.get()
+    assert file.title == "my file"
+    assert file.creator == user
+    assert file.type == FileTypeChoices.BACKGROUND_IMAGE
+    assert file.upload_state == FileUploadStateChoices.PENDING
+
+
+def test_api_files_create_file_authenticated_no_filename():
+    """
+    Creating a file item without providing a filename should fail.
+    """
+    user = factories.UserFactory()
+
+    client = APIClient()
+    client.force_login(user)
+
+    response = client.post(
+        "/api/v1.0/files/",
+        {
+            "title": "my item",
+            "type": FileTypeChoices.BACKGROUND_IMAGE,
+        },
+        format="json",
+    )
+    assert response.status_code == 400
+    assert response.json() == {"filename": ["This field is required."]}
+
+
+def test_api_files_create_file_authenticated_success():
+    """
+    Authenticated users should be able to create a file file and must provide a filename.
+    """
+    user = factories.UserFactory()
+
+    client = APIClient()
+    client.force_login(user)
+
+    now = timezone.now()
+    with freeze_time(now):
+        response = client.post(
+            "/api/v1.0/files/",
+            {
+                "type": FileTypeChoices.BACKGROUND_IMAGE,
+                "title": "Eiffle tower",
+                "filename": "file.png",
+            },
+            format="json",
+        )
+    assert response.status_code == 201
+    file = File.objects.get()
+    assert file.title == "Eiffle tower"
+    assert file.type == FileTypeChoices.BACKGROUND_IMAGE
+    assert file.filename == "file.png"
+
+    response_data = response.json()
+    assert response_data["creator"] is not None, response_data
+
+    assert response.json().get("policy") is not None
+
+    policy = response.json()["policy"]
+    policy_parsed = urlparse(policy)
+
+    assert policy_parsed.scheme == "http"
+    assert policy_parsed.netloc == "localhost:9000"
+    assert policy_parsed.path == f"/meet-media-storage/files/{file.id!s}/.png"
+
+    query_params = parse_qs(policy_parsed.query)
+
+    assert query_params.pop("X-Amz-Algorithm") == ["AWS4-HMAC-SHA256"]
+    assert query_params.pop("X-Amz-Credential") == [
+        f"meet/{now.strftime('%Y%m%d')}/us-east-1/s3/aws4_request"
+    ]
+    assert query_params.pop("X-Amz-Date") == [now.strftime("%Y%m%dT%H%M%SZ")]
+    assert query_params.pop("X-Amz-Expires") == ["60"]
+    assert query_params.pop("X-Amz-SignedHeaders") == ["host;x-amz-acl"]
+    assert query_params.pop("X-Amz-Signature") is not None
+
+    assert len(query_params) == 0
+
+
+def test_api_files_create_file_authenticated_extension_not_allowed():
+    """
+    Creating a file item with an extension not allowed should fail.
+    """
+    user = factories.UserFactory()
+    client = APIClient()
+    client.force_login(user)
+    response = client.post(
+        "/api/v1.0/files/",
+        {
+            "type": FileTypeChoices.BACKGROUND_IMAGE,
+            "title": "Paris tower",
+            "filename": "file.notallowed",
+        },
+        format="json",
+    )
+    assert response.status_code == 400
+    assert response.json() == {"filename": ["This file extension is not allowed."]}
+
+
+def test_api_files_create_file_authenticated_extension_case_insensitive():
+    """
+    Creating a file item with an extension, no matter the case used, should be allowed.
+    """
+    user = factories.UserFactory()
+    client = APIClient()
+    client.force_login(user)
+    response = client.post(
+        "/api/v1.0/files/",
+        {
+            "type": FileTypeChoices.BACKGROUND_IMAGE,
+            "filename": "file.JPG",
+        },
+        format="json",
+    )
+    assert response.status_code == 201, response.json()
+    file = File.objects.get()
+    assert file.title == "file"
+
+
+def test_api_files_create_file_authenticated_not_checking_extension(settings):
+    """
+    Creating a file with an extension not allowed should not fail when restrictions are disabled.
+    """
+    settings.FILE_UPLOAD_APPLY_RESTRICTIONS = False
+    user = factories.UserFactory()
+    client = APIClient()
+    client.force_login(user)
+    response = client.post(
+        "/api/v1.0/files/",
+        {
+            "type": FileTypeChoices.BACKGROUND_IMAGE,
+            "filename": "file.notallowed",
+        },
+        format="json",
+    )
+    assert response.status_code == 201, response.json()
+    file = File.objects.get()
+    assert file.title == "file"
+
+
+def test_api_files_create_file_authenticated_no_extension_but_checking_it_should_fail(
+    settings,
+):
+    """
+    Creating a file without an extension but checking the extension should fail.
+    """
+    settings.FILE_UPLOAD_APPLY_RESTRICTIONS = True
+    user = factories.UserFactory()
+    client = APIClient()
+    client.force_login(user)
+    response = client.post(
+        "/api/v1.0/files/",
+        {
+            "type": FileTypeChoices.BACKGROUND_IMAGE,
+            "filename": "file",
+        },
+        format="json",
+    )
+    assert response.status_code == 400
+    assert response.json() == {"filename": ["This file extension is not allowed."]}
+
+
+def test_api_files_create_file_authenticated_hidden_file_but_checking_extension_should_fail(
+    settings,
+):
+    """
+    Creating a hidden file (starting with a dot) but checking the extension should fail.
+    """
+    settings.FILE_UPLOAD_APPLY_RESTRICTIONS = True
+    user = factories.UserFactory()
+    client = APIClient()
+    client.force_login(user)
+    response = client.post(
+        "/api/v1.0/files/",
+        {
+            "type": FileTypeChoices.BACKGROUND_IMAGE,
+            "filename": ".file",
+        },
+    )
+
+    assert response.status_code == 400
+    assert response.json() == {"filename": ["This file extension is not allowed."]}
+
+
+def test_api_files_create_force_id_success():
+    """It should be possible to force the item ID when creating a   item."""
+    user = factories.UserFactory()
+    client = APIClient()
+    client.force_login(user)
+
+    forced_id = uuid4()
+
+    response = client.post(
+        "/api/v1.0/files/",
+        {
+            "id": str(forced_id),
+            "title": "my item",
+            "type": FileTypeChoices.BACKGROUND_IMAGE,
+            "filename": "my_file.png",
+        },
+        format="json",
+    )
+
+    assert response.status_code == 201, response.json()
+    files = File.objects.all()
+    assert len(files) == 1
+    assert files[0].id == forced_id
+
+
+def test_api_files_create_force_id_existing():
+    """
+    It should not be possible to use the ID of an existing file when forcing ID on creation.
+    """
+    user = factories.UserFactory()
+    client = APIClient()
+    client.force_login(user)
+
+    file = factories.FileFactory()
+
+    response = client.post(
+        "/api/v1.0/files/",
+        {
+            "id": str(file.id),
+            "title": "my file",
+            "type": FileTypeChoices.BACKGROUND_IMAGE,
+            "filename": "my_file.png",
+        },
+        format="json",
+    )
+
+    assert response.status_code == 400
+    assert response.json() == {
+        "id": ["A file with this ID already exists. You cannot override it."]
+    }
+
+
+@pytest.mark.django_db(transaction=True)
+def test_api_files_create_file_race_condition():
+    """
+    It should be possible to create several files at the same time
+    without causing any race conditions or data integrity issues.
+    """
+
+    def create_item(title):
+        user = factories.UserFactory()
+        client = APIClient()
+        client.force_login(user)
+        return client.post(
+            "/api/v1.0/files/",
+            {
+                "title": title,
+                "type": FileTypeChoices.BACKGROUND_IMAGE,
+                "filename": "my_file.png",
+            },
+            format="json",
+        )
+
+    with ThreadPoolExecutor(max_workers=2) as executor:
+        future1 = executor.submit(create_item, "my item 1")
+        future2 = executor.submit(create_item, "my item 2")
+
+        response1 = future1.result()
+        response2 = future2.result()
+
+        assert response1.status_code == 201
+        assert response2.status_code == 201

src/backend/core/tests/files/test_api_files_delete.py

@@ -0,0 +1,45 @@
+"""
+Tests for files API endpoint in meet's core app: delete
+"""
+
+import pytest
+from rest_framework.test import APIClient
+
+from core import factories, models
+
+pytestmark = pytest.mark.django_db
+
+
+def test_api_files_delete_anonymous():
+    """Anonymous users should not be allowed to destroy a file."""
+    file = factories.FileFactory()
+    existing_items = models.File.objects.all().count()
+
+    response = APIClient().delete(
+        f"/api/v1.0/files/{file.id!s}/",
+    )
+
+    assert response.status_code == 401
+    assert models.File.objects.count() == existing_items
+
+
+def test_api_files_delete_authenticated_owner():
+    """
+    Authenticated users should be able to delete a item they own.
+    """
+    user = factories.UserFactory()
+
+    client = APIClient()
+    client.force_login(user)
+
+    file = factories.FileFactory(creator=user)
+
+    response = client.delete(
+        f"/api/v1.0/files/{file.id}/",
+    )
+
+    assert response.status_code == 204
+
+    # Make sure it is only a soft delete
+    file.refresh_from_db()
+    assert file.deleted_at is not None

src/backend/core/tests/files/test_api_files_list.py

@@ -0,0 +1,160 @@
+"""
+Tests for files API endpoint in meet's core app: list
+"""
+
+from unittest import mock
+
+from django.utils import timezone
+
+import pytest
+from faker import Faker
+from rest_framework.pagination import PageNumberPagination
+from rest_framework.test import APIClient
+
+from core import factories, models
+
+fake = Faker()
+pytestmark = pytest.mark.django_db
+
+
+def test_api_files_list_anonymous_not_allowed():
+    """
+    Anonymous users should not be allowed to list files whatever the
+    """
+    response = APIClient().get("/api/v1.0/files/")
+    assert response.status_code == 401
+
+
+def test_api_files_list_authentificated_user_allowed():
+    """
+    Authentificated users should be allowed to list files
+    """
+    user = factories.UserFactory()
+    client = APIClient()
+    client.force_login(user)
+
+    response = client.get("/api/v1.0/files/")
+    assert response.status_code == 200
+    assert response.data == {"count": 0, "next": None, "previous": None, "results": []}
+
+
+def test_api_files_list_format():
+    """Validate the format of files as returned by the list view."""
+    user = factories.UserFactory()
+    client = APIClient()
+    client.force_login(user)
+
+    file = factories.FileFactory(
+        type=models.FileTypeChoices.BACKGROUND_IMAGE,
+        title="item 1",
+        creator=user,
+    )
+
+    # A file from another user should not appear
+    factories.FileFactory(
+        type=models.FileTypeChoices.BACKGROUND_IMAGE,
+        title="item 2",
+    )
+
+    # hard deleted item should not appear
+    factories.FileFactory(
+        type=models.FileTypeChoices.BACKGROUND_IMAGE,
+        hard_deleted_at=timezone.now(),
+        title="hard deleted item",
+        creator=user,
+    )
+
+    response = client.get("/api/v1.0/files/")
+
+    assert response.status_code == 200
+    content = response.json()
+    results = content.pop("results")
+    assert content == {
+        "count": 1,
+        "next": None,
+        "previous": None,
+    }
+    assert len(results) == 1
+    assert results == [
+        {
+            "id": str(file.id),
+            "created_at": file.created_at.isoformat().replace("+00:00", "Z"),
+            "creator": {
+                "id": str(file.creator.id),
+                "full_name": file.creator.full_name,
+                "short_name": file.creator.short_name,
+            },
+            "title": file.title,
+            "updated_at": file.updated_at.isoformat().replace("+00:00", "Z"),
+            "type": models.FileTypeChoices.BACKGROUND_IMAGE,
+            "upload_state": file.upload_state,
+            "url": None,
+            "mimetype": file.mimetype,
+            "filename": file.filename,
+            "size": None,
+            "description": None,
+            "deleted_at": None,
+            "hard_deleted_at": None,
+            "abilities": {
+                "destroy": True,
+                "hard_delete": True,
+                "media_auth": True,
+                "partial_update": True,
+                "retrieve": True,
+                "update": True,
+                "upload_ended": True,
+            },
+        }
+    ]
+
+
+@mock.patch.object(PageNumberPagination, "get_page_size", return_value=2)
+def test_api_files_list_pagination(
+    _mock_page_size,
+):
+    """Pagination should work as expected."""
+    user = factories.UserFactory()
+
+    client = APIClient()
+    client.force_login(user)
+
+    file_ids = [
+        str(file.id)
+        for file in factories.FileFactory.create_batch(
+            3,
+            creator=user,
+            type=models.FileTypeChoices.BACKGROUND_IMAGE,
+        )
+    ]
+    # Get page 1
+    response = client.get(
+        "/api/v1.0/files/",
+    )
+
+    assert response.status_code == 200
+    content = response.json()
+
+    assert content["count"] == 3
+    assert content["next"] == "http://testserver/api/v1.0/files/?page=2"
+    assert content["previous"] is None
+
+    assert len(content["results"]) == 2
+    for item in content["results"]:
+        file_ids.remove(item["id"])
+
+    # Get page 2
+    response = client.get(
+        "/api/v1.0/files/?page=2",
+    )
+
+    assert response.status_code == 200
+    content = response.json()
+
+    assert content["count"] == 3
+    assert content["next"] is None
+    assert content["previous"] == "http://testserver/api/v1.0/files/"
+
+    assert len(content["results"]) == 1
+    for item in content["results"]:
+        file_ids.remove(item["id"])
+    assert file_ids == []

src/backend/core/tests/files/test_api_files_list_filters.py

@@ -0,0 +1,216 @@
+"""
+Tests for files API endpoint in meet's core app: list
+"""
+
+import pytest
+from faker import Faker
+from rest_framework.test import APIClient
+
+from core import factories, models
+
+fake = Faker()
+pytestmark = pytest.mark.django_db
+
+
+# Filters: unknown field
+
+
+def test_api_files_list_filter_unknown_field():
+    """
+    Trying to filter by an unknown field should do nothing.
+    """
+    user = factories.UserFactory()
+    client = APIClient()
+    client.force_login(user)
+
+    factories.FileFactory(type=models.FileTypeChoices.BACKGROUND_IMAGE)
+    expected_ids = {
+        str(file.id)
+        for file in factories.FileFactory.create_batch(
+            2, creator=user, type=models.FileTypeChoices.BACKGROUND_IMAGE
+        )
+    }
+
+    response = client.get("/api/v1.0/files/?unknown=true")
+
+    assert response.status_code == 200
+    results = response.json()["results"]
+    assert len(results) == 2
+    assert {result["id"] for result in results} == expected_ids
+
+
+# Filters: is_creator_me
+
+
+def test_api_files_list_filter_is_creator_me_true():
+    """
+    Authenticated users should be able to filter files they created.
+    """
+    user = factories.UserFactory()
+    client = APIClient()
+    client.force_login(user)
+
+    factories.FileFactory.create_batch(
+        2, creator=user, type=models.FileTypeChoices.BACKGROUND_IMAGE
+    )
+    factories.FileFactory.create_batch(2, type=models.FileTypeChoices.BACKGROUND_IMAGE)
+
+    response = client.get("/api/v1.0/files/?is_creator_me=true")
+
+    assert response.status_code == 200
+    results = response.json()["results"]
+    assert len(results) == 2
+
+    # Ensure all results are created by the current user
+    for result in results:
+        assert result["creator"] == {
+            "id": str(user.id),
+            "full_name": user.full_name,
+            "short_name": user.short_name,
+        }
+
+
+def test_api_files_list_filter_is_creator_me_invalid():
+    """Filtering with an invalid `is_creator_me` value should do nothing."""
+    user = factories.UserFactory()
+    client = APIClient()
+    client.force_login(user)
+
+    factories.FileFactory.create_batch(
+        3, creator=user, type=models.FileTypeChoices.BACKGROUND_IMAGE
+    )
+
+    response = client.get("/api/v1.0/files/?is_creator_me=invalid")
+
+    assert response.status_code == 200
+    results = response.json()["results"]
+    assert len(results) == 3
+
+
+# Filters: type
+
+
+def test_api_files_list_filter_type_and_upload_status():
+    """
+    Authenticated users should be able to filter files by their type and upload status.
+
+    This test will make more sense when other types are added to the API
+    """
+
+    user = factories.UserFactory()
+
+    client = APIClient()
+    client.force_login(user)
+
+    file = factories.FileFactory(
+        creator=user,
+        type=models.FileTypeChoices.BACKGROUND_IMAGE,
+        update_upload_state=models.FileUploadStateChoices.PENDING,
+    )
+    assert file.upload_state == models.FileUploadStateChoices.PENDING
+
+    expected_files = factories.FileFactory.create_batch(
+        2,
+        creator=user,
+        type=models.FileTypeChoices.BACKGROUND_IMAGE,
+        update_upload_state=models.FileUploadStateChoices.READY,
+    )
+    expected_files_ids = {str(file.id) for file in expected_files}
+
+    # Filter by type: background_image & upload state
+    response = client.get("/api/v1.0/files/?type=background_image&upload_state=ready")
+
+    assert response.status_code == 200
+    assert response.json()["count"] == 2
+
+    results = response.json()["results"]
+
+    # Ensure all results are background images
+    results_ids = {result["id"] for result in results}
+    assert results_ids == expected_files_ids
+    for result in results:
+        assert result["type"] == models.FileTypeChoices.BACKGROUND_IMAGE
+        assert result["upload_state"] == models.FileUploadStateChoices.READY
+
+    # Second request without the upload_state filter, to check that all 3 show up
+    response = client.get("/api/v1.0/files/?type=background_image")
+
+    assert response.status_code == 200
+    assert response.json()["count"] == 3
+
+    results = response.json()["results"]
+
+    # Ensure all results are background images
+    results_ids = {result["id"] for result in results}
+    assert results_ids == {str(file.id) for file in expected_files + [file]}
+    for result in results:
+        assert result["type"] == models.FileTypeChoices.BACKGROUND_IMAGE
+
+
+def test_api_files_list_filter_is_deleted():
+    """
+    Authenticated users should be able to filter files by their deletion status.
+    """
+    user = factories.UserFactory()
+    client = APIClient()
+    client.force_login(user)
+
+    not_deleted_file = factories.FileFactory(creator=user)
+
+    deleted_files = factories.FileFactory.create_batch(2, creator=user)
+    for file in deleted_files:
+        file.soft_delete()
+
+    # No filters
+    response_no_filters = client.get("/api/v1.0/files/")
+
+    assert response_no_filters.status_code == 200
+    assert response_no_filters.json()["count"] == 3
+
+    results = response_no_filters.json()["results"]
+
+    results_ids = {result["id"] for result in results}
+    assert results_ids == {str(file.id) for file in [*deleted_files, not_deleted_file]}
+
+    # Filters deleted
+    response_filter_deleted = client.get("/api/v1.0/files/?is_deleted=true")
+
+    assert response_filter_deleted.status_code == 200
+    assert response_filter_deleted.json()["count"] == 2
+
+    results = response_filter_deleted.json()["results"]
+
+    results_ids = {result["id"] for result in results}
+    assert results_ids == {str(file.id) for file in deleted_files}
+
+    # Filters not deleted
+    response_filter_not_deleted = client.get("/api/v1.0/files/?is_deleted=false")
+
+    assert response_filter_not_deleted.status_code == 200
+    assert response_filter_not_deleted.json()["count"] == 1
+
+    results = response_filter_not_deleted.json()["results"]
+
+    # Ensure all results are deleted
+    results_ids = {result["id"] for result in results}
+    assert results_ids == {str(file.id) for file in [not_deleted_file]}
+
+
+def test_api_files_list_filter_unknown_type():
+    """
+    Filtering by an unknown type should return an empty list
+    """
+
+    user = factories.UserFactory()
+
+    client = APIClient()
+    client.force_login(user)
+
+    factories.FileFactory.create_batch(3, creator=user)
+
+    response = client.get("/api/v1.0/files/?type=unknown")
+
+    assert response.status_code == 400
+    assert response.json() == {
+        "type": ["Select a valid choice. unknown is not one of the available choices."]
+    }

src/backend/core/tests/files/test_api_files_list_ordering.py

@@ -0,0 +1,71 @@
+"""Test the ordering of items."""
+
+import operator
+
+import pytest
+from rest_framework.test import APIClient
+
+from core import factories, models
+
+pytestmark = pytest.mark.django_db
+
+
+def test_api_files_list_ordering_default():
+    """items should be ordered by descending "updated_at" by default"""
+    user = factories.UserFactory()
+    client = APIClient()
+    client.force_login(user)
+
+    factories.FileFactory.create_batch(
+        4, creator=user, type=models.FileTypeChoices.BACKGROUND_IMAGE
+    )
+    response = client.get("/api/v1.0/files/")
+
+    assert response.status_code == 200
+    results = response.json()["results"]
+    assert len(results) == 4
+
+    # Check that results are sorted by descending "updated_at" as expected
+    for i in range(3):
+        assert operator.ge(results[i]["updated_at"], results[i + 1]["updated_at"])
+
+
+def test_api_files_list_ordering_by_fields():
+    """It should be possible to order by several fields"""
+    user = factories.UserFactory()
+    client = APIClient()
+    client.force_login(user)
+
+    factories.FileFactory.create_batch(
+        4, creator=user, type=models.FileTypeChoices.BACKGROUND_IMAGE
+    )
+
+    for parameter in [
+        "created_at",
+        "-created_at",
+        "updated_at",
+        "-updated_at",
+    ]:
+        is_descending = parameter.startswith("-")
+        field = parameter.lstrip("-")
+        querystring = f"?ordering={parameter}"
+
+        response = client.get(f"/api/v1.0/files/{querystring:s}")
+        assert response.status_code == 200
+        results = response.json()["results"]
+        assert len(results) == 4
+
+        # Check that results are sorted by the field in querystring as expected
+        compare = operator.ge if is_descending else operator.le
+        for i in range(3):
+            operator1 = (
+                results[i][field].lower()
+                if isinstance(results[i][field], str)
+                else results[i][field]
+            )
+            operator2 = (
+                results[i + 1][field].lower()
+                if isinstance(results[i + 1][field], str)
+                else results[i + 1][field]
+            )
+            assert compare(operator1, operator2)

src/backend/core/tests/files/test_api_files_media_auth.py

@@ -0,0 +1,141 @@
+"""
+Test file uploads API endpoint for users in meet's core app.
+"""
+
+from io import BytesIO
+from urllib.parse import quote, urlparse
+
+from django.conf import settings
+from django.core.files.storage import default_storage
+from django.utils import timezone
+
+import pytest
+import requests
+from freezegun import freeze_time
+from rest_framework.test import APIClient
+
+from core import factories, models
+
+pytestmark = pytest.mark.django_db
+
+
+def test_api_files_media_auth_anonymous_not_authorized():
+    """Anonymous users should not be allowed to retrieve a file"""
+    file = factories.FileFactory(
+        type=models.FileTypeChoices.BACKGROUND_IMAGE,
+        update_upload_state=models.FileUploadStateChoices.READY,
+    )
+
+    original_url = f"http://localhost/media/{file.file_key:s}"
+    response = APIClient().get(
+        "/api/v1.0/files/media-auth/", HTTP_X_ORIGINAL_URL=original_url
+    )
+
+    assert response.status_code == 401
+
+
+def test_api_files_media_get_own():
+    """
+    Authenticated user should be allowed to retrieve their own file.
+    """
+    user = factories.UserFactory()
+
+    file = factories.FileFactory(
+        type=models.FileTypeChoices.BACKGROUND_IMAGE,
+        update_upload_state=models.FileUploadStateChoices.READY,
+        creator=user,
+    )
+
+    client = APIClient()
+    client.force_login(user)
+
+    default_storage.save(
+        file.file_key,
+        BytesIO(b"my prose"),
+    )
+
+    original_url = f"http://localhost/media/{file.file_key:s}"
+    now = timezone.now()
+    with freeze_time(now):
+        response = client.get(
+            "/api/v1.0/files/media-auth/", HTTP_X_ORIGINAL_URL=original_url
+        )
+
+    assert response.status_code == 200
+
+    authorization = response["Authorization"]
+    assert "AWS4-HMAC-SHA256 Credential=" in authorization
+    assert (
+        "SignedHeaders=host;x-amz-content-sha256;x-amz-date, Signature="
+        in authorization
+    )
+    assert response["X-Amz-Date"] == now.strftime("%Y%m%dT%H%M%SZ")
+
+    s3_url = urlparse(settings.AWS_S3_ENDPOINT_URL)
+    file_url = f"{settings.AWS_S3_ENDPOINT_URL:s}/meet-media-storage/{file.file_key:s}"
+    response = requests.get(
+        file_url,
+        headers={
+            "authorization": authorization,
+            "x-amz-date": response["x-amz-date"],
+            "x-amz-content-sha256": response["x-amz-content-sha256"],
+            "Host": f"{s3_url.hostname:s}:{s3_url.port:d}",
+        },
+        timeout=1,
+    )
+    assert response.content.decode("utf-8") == "my prose"
+
+
+def test_api_files_media_auth_file_pending():
+    """
+    Users who have a specific access to an file, whatever the role, should not be able to
+    retrieve related attachments if the file is not ready.
+    """
+    user = factories.UserFactory()
+    client = APIClient()
+    client.force_login(user)
+
+    file = factories.FileFactory(
+        type=models.FileTypeChoices.BACKGROUND_IMAGE,
+        upload_state=models.FileUploadStateChoices.PENDING,
+        creator=user,
+    )
+
+    key = file.file_key
+
+    original_url = quote(f"http://localhost/media/{key:s}")
+    response = client.get(
+        "/api/v1.0/files/media-auth/", HTTP_X_ORIGINAL_URL=original_url
+    )
+
+    assert response.status_code == 403
+
+
+def test_api_files_media_auth_own_file_deleted():
+    """
+    This function tests the access restrictions on deleted files through the media
+    authorization API endpoint. It ensures that a user cannot retrieve a file that is deleted.
+    """
+    user = factories.UserFactory()
+
+    file = factories.FileFactory(
+        type=models.FileTypeChoices.BACKGROUND_IMAGE,
+        update_upload_state=models.FileUploadStateChoices.READY,
+        creator=user,
+    )
+
+    client = APIClient()
+    client.force_login(user)
+
+    default_storage.save(
+        file.file_key,
+        BytesIO(b"my prose"),
+    )
+    file.soft_delete()
+
+    original_url = f"http://localhost/media/{file.file_key:s}"
+    response = client.get(
+        "/api/v1.0/files/media-auth/", HTTP_X_ORIGINAL_URL=original_url
+    )
+
+    assert response.status_code == 403

src/backend/core/tests/files/test_api_files_update.py

@@ -0,0 +1,67 @@
+"""
+Tests for files API endpoint in meet's core app: update
+"""
+
+import pytest
+from rest_framework.test import APIClient
+
+from core import factories
+from core.api import serializers
+
+pytestmark = pytest.mark.django_db
+
+
+def test_api_files_update_anonymous_forbidden():
+    """
+    Anonymous users should not be allowed to update an file when link
+    configuration does not allow it.
+    """
+
+    file = factories.FileFactory()
+
+    old_file_values = serializers.FileSerializer(instance=file).data
+    new_file_values = serializers.FileSerializer(instance=factories.FileFactory()).data
+
+    response = APIClient().put(
+        f"/api/v1.0/files/{file.id!s}/",
+        new_file_values,
+        format="json",
+    )
+    assert response.status_code == 401
+    assert response.json() == {
+        "detail": "Authentication credentials were not provided."
+    }
+
+    file.refresh_from_db()
+    item_values = serializers.FileSerializer(instance=file).data
+    assert item_values == old_file_values
+
+
+def test_api_files_update_description_and_title():
+    """
+    Test the description and title of an file can be updated.
+    """
+    user = factories.UserFactory()
+
+    client = APIClient()
+    client.force_login(user)
+
+    file = factories.FileFactory(
+        description="Old description",
+        title="Old title",
+        creator=user,
+    )
+
+    response = client.patch(
+        f"/api/v1.0/files/{file.id!s}/",
+        {"description": "New description", "title": "New title"},
+        format="json",
+    )
+    assert response.status_code == 200
+    result = response.json()
+    assert result["description"] == "New description"
+    assert result["title"] == "New title"
+
+    file.refresh_from_db()
+    assert file.description == "New description"
+    assert file.title == "New title"

src/backend/core/tests/files/test_api_files_upload_ended.py

@@ -0,0 +1,272 @@
+"""Test related to item upload ended API."""
+
+import logging
+from io import BytesIO
+
+from django.core.files.storage import default_storage
+
+import pytest
+from rest_framework.test import APIClient
+
+from core import factories, models
+from core.models import FileTypeChoices, FileUploadStateChoices
+
+pytestmark = pytest.mark.django_db
+
+
+def test_api_file_upload_ended_anonymous():
+    """Anonymous users should not be allowed to end an upload."""
+    file = factories.FileFactory()
+    response = APIClient().post(f"/api/v1.0/files/{file.id!s}/upload-ended/")
+
+    assert response.status_code == 401
+
+
+def test_api_file_upload_ended_non_creator_not_found():
+    """Users without write permissions should not be allowed to end an upload."""
+    user = factories.UserFactory()
+    client = APIClient()
+    client.force_login(user)
+
+    file = factories.FileFactory()
+
+    response = client.post(f"/api/v1.0/files/{file.id!s}/upload-ended/")
+
+    assert response.status_code == 404
+
+
+def test_api_file_upload_ended_on_wrong_upload_state():
+    """
+    Users should not be allowed to end an upload on files that are not in the PENDING upload state.
+    """
+    user = factories.UserFactory()
+    client = APIClient()
+    client.force_login(user)
+
+    file = factories.FileFactory(
+        type=FileTypeChoices.BACKGROUND_IMAGE,
+        creator=user,
+        update_upload_state=FileUploadStateChoices.READY,
+    )
+
+    response = client.post(f"/api/v1.0/files/{file.id!s}/upload-ended/")
+
+    assert response.status_code == 400
+    assert response.json() == {
+        "file": "This action is only available for files in PENDING state."
+    }
+
+
+def test_api_file_upload_ended_success(settings):
+    """
+    Users should be able to end an upload on files that are files and in the UPLOADING upload state.
+    """
+    user = factories.UserFactory()
+    client = APIClient()
+    client.force_login(user)
+
+    settings.FILE_UPLOAD_APPLY_RESTRICTIONS = True
+    settings.FILE_UPLOAD_RESTRICTIONS = {
+        "background_image": {
+            **settings.FILE_UPLOAD_RESTRICTIONS["background_image"],
+            "allowed_mimetypes": ["text/html", "text/plain"],
+        },
+    }
+
+    file = factories.FileFactory(
+        type=FileTypeChoices.BACKGROUND_IMAGE,
+        filename="my_file.txt",
+        mimetype="text/html",
+        creator=user,
+    )
+
+    default_storage.save(
+        file.file_key,
+        BytesIO(b"my prose"),
+    )
+
+    response = client.post(f"/api/v1.0/files/{file.id!s}/upload-ended/")
+
+    assert response.status_code == 200
+
+    file.refresh_from_db()
+    assert file.upload_state == FileUploadStateChoices.READY
+    assert file.mimetype == "text/plain"
+    assert file.size == 8
+
+    assert response.json()["mimetype"] == "text/plain"
+
+
+def test_api_file_upload_ended_mimetype_not_allowed(settings, caplog):
+    """
+    Test that the API returns a 400 when the mimetype is not allowed.
+    File should be deleted and the file should be deleted from the storage.
+    """
+    settings.RESTRICT_UPLOAD_FILE_TYPE = True
+    settings.FILE_UPLOAD_RESTRICTIONS = {
+        "background_image": {
+            **settings.FILE_UPLOAD_RESTRICTIONS["background_image"],
+            "allowed_mimetypes": ["application/pdf"],
+        }
+    }
+
+    user = factories.UserFactory()
+    client = APIClient()
+    client.force_login(user)
+
+    file = factories.FileFactory(
+        type=FileTypeChoices.BACKGROUND_IMAGE, filename="my_file.txt", creator=user
+    )
+
+    default_storage.save(
+        file.file_key,
+        BytesIO(b"my prose"),
+    )
+
+    with caplog.at_level(logging.WARNING):
+        response = client.post(f"/api/v1.0/files/{file.id!s}/upload-ended/")
+
+    assert response.status_code == 400
+    assert (
+        f"upload_ended: mimetype not allowed text/plain for file {file.file_key}"
+        in caplog.text
+    )
+
+    assert not models.File.objects.filter(id=file.id).exists()
+    assert not default_storage.exists(file.file_key)
+
+
+def test_api_file_upload_ended_mimetype_not_allowed_not_checking_mimetype(settings):
+    """
+    Test that the API returns a 200 when the mimetype is not allowed but not checking the mimetype.
+    """
+    settings.FILE_UPLOAD_APPLY_RESTRICTIONS = False
+    settings.FILE_UPLOAD_RESTRICTIONS = {
+        "background_image": {
+            **settings.FILE_UPLOAD_RESTRICTIONS["background_image"],
+            "allowed_mimetypes": ["application/pdf"],
+        }
+    }
+    user = factories.UserFactory()
+    client = APIClient()
+    client.force_login(user)
+
+    file = factories.FileFactory(
+        type=FileTypeChoices.BACKGROUND_IMAGE, filename="my_file.txt", creator=user
+    )
+
+    default_storage.save(
+        file.file_key,
+        BytesIO(b"my prose"),
+    )
+
+    response = client.post(f"/api/v1.0/files/{file.id!s}/upload-ended/")
+
+    assert response.status_code == 200
+
+    file.refresh_from_db()
+    assert file.upload_state == FileUploadStateChoices.READY
+    assert file.mimetype == "text/plain"
+    assert file.size == 8
+
+    assert response.json()["mimetype"] == "text/plain"
+
+
+def test_api_upload_ended_mismatch_mimetype_with_object_storage(settings, caplog):
+    """
+    Object on storage should have the same mimetype than the one saved in the
+    File object.
+    """
+    settings.FILE_UPLOAD_APPLY_RESTRICTIONS = True
+    settings.FILE_UPLOAD_RESTRICTIONS = {
+        "background_image": {
+            **settings.FILE_UPLOAD_RESTRICTIONS["background_image"],
+            "allowed_mimetypes": ["text/html", "application/pdf"],
+        }
+    }
+
+    user = factories.UserFactory()
+    client = APIClient()
+    client.force_login(user)
+
+    file = factories.FileFactory(
+        type=FileTypeChoices.BACKGROUND_IMAGE,
+        filename="my_file.pdf",
+        title="my_file.pdf",
+        creator=user,
+    )
+
+    s3_client = default_storage.connection.meta.client
+
+    s3_client.put_object(
+        Bucket=default_storage.bucket_name,
+        Key=file.file_key,
+        ContentType="text/html",
+        Body=BytesIO(
+            b'<meta http-equiv="refresh" content="0; url=https://fichiers.numerique.gouv.fr">'
+        ),
+        Metadata={
+            "foo": "bar",
+        },
+    )
+
+    head_object = s3_client.head_object(
+        Bucket=default_storage.bucket_name, Key=file.file_key
+    )
+
+    assert head_object["ContentType"] == "text/html"
+    with caplog.at_level(logging.INFO, logger="core.api.viewsets"):
+        response = client.post(f"/api/v1.0/files/{file.id!s}/upload-ended/")
+    assert (
+        "upload_ended: content type mismatch between object storage and file,"
+        " updating from text/html to application/pdf" in caplog.text
+    )
+    assert response.status_code == 200
+
+    file.refresh_from_db()
+
+    assert file.mimetype == "application/pdf"
+
+    head_object = s3_client.head_object(
+        Bucket=default_storage.bucket_name, Key=file.file_key
+    )
+    assert head_object["ContentType"] == "application/pdf"
+    assert head_object["Metadata"] == {"foo": "bar"}
+
+
+def test_api_upload_ended_file_size_exceeded(settings, caplog):
+    """
+    Test when the file size exceed the allowed max upload file size
+    should return a 400 and delete the file.
+    """
+
+    settings.FILE_UPLOAD_RESTRICTIONS = {
+        "background_image": {
+            **settings.FILE_UPLOAD_RESTRICTIONS["background_image"],
+            "max_size": 0,
+        }
+    }
+
+    user = factories.UserFactory()
+    client = APIClient()
+    client.force_login(user)
+
+    file = factories.FileFactory(
+        type=FileTypeChoices.BACKGROUND_IMAGE, filename="my_file.txt", creator=user
+    )
+
+    default_storage.save(
+        file.file_key,
+        BytesIO(b"my prose"),
+    )
+
+    with caplog.at_level(logging.INFO, logger="core.api.viewsets"):
+        response = client.post(f"/api/v1.0/files/{file.id!s}/upload-ended/")
+    assert (
+        f"upload_ended: file size (8) for file {file.file_key} higher than the allowed max size"
+        in caplog.text
+    )
+    assert response.status_code == 400
+
+    assert not models.File.objects.filter(id=file.id).exists()
+    assert not default_storage.exists(file.file_key)

src/backend/core/urls.py

@@ -14,6 +14,7 @@ router = DefaultRouter()
 router.register("users", viewsets.UserViewSet, basename="users")
 router.register("rooms", viewsets.RoomViewSet, basename="rooms")
 router.register("recordings", viewsets.RecordingViewSet, basename="recordings")
+router.register("files", viewsets.FileViewSet, basename="files")
 router.register(
     "resource-accesses", viewsets.ResourceAccessViewSet, basename="resource_accesses"
 )

src/backend/core/utils.py

@@ -19,6 +19,7 @@ from django.conf import settings
 from django.core.files.storage import default_storage
 
 import aiohttp
+import boto3
 import botocore
 import magic
 from asgiref.sync import async_to_sync
@@ -382,8 +383,10 @@ def detect_mimetype(file_buffer: bytes, filename: str | None = None) -> str:
         # Use guess_file_type (Python 3.13+) instead of deprecated guess_type
         mimetype_from_extension, _ = mimetypes.guess_file_type(filename, strict=False)
 
-    logger.info("detect_mimetype: mimetype_from_content: %s", mimetype_from_content)
-    logger.info("detect_mimetype: mimetype_from_extension: %s", mimetype_from_extension)
+    logger.debug("detect_mimetype: mimetype_from_content: %s", mimetype_from_content)
+    logger.debug(
+        "detect_mimetype: mimetype_from_extension: %s", mimetype_from_extension
+    )
 
     # Strategy: Prefer content-based detection, but use extension if:
     # 1. Content detection returns generic types (application/octet-stream, text/plain)
@@ -411,3 +414,44 @@ def detect_mimetype(file_buffer: bytes, filename: str | None = None) -> str:
 
     # Default to content-based detection (most reliable)
     return mimetype_from_content or "application/octet-stream"
+
+
+def generate_upload_policy(file):
+    """
+    Generate a S3 upload policy for a given file.
+
+    Notes:
+        Originally taken from https://github.com/suitenumerique/drive/blob/564822d31f071c6dfacd112ef4b7146c73077cd9/src/backend/core/api/utils.py#L102  # pylint: disable=line-too-long
+    """
+
+    key = file.file_key
+
+    # This settings should be used if the backend application and the frontend application
+    # can't connect to the object storage with the same domain. This is the case in the
+    # docker compose stack used in development. The frontend application will use localhost
+    # to connect to the object storage while the backend application will use the object storage
+    # service name declared in the docker compose stack.
+    # This is needed because the domain name is used to compute the signature. So it can't be
+    # changed dynamically by the frontend application.
+    if settings.AWS_S3_DOMAIN_REPLACE:
+        s3_client = boto3.client(
+            "s3",
+            aws_access_key_id=settings.AWS_S3_ACCESS_KEY_ID,
+            aws_secret_access_key=settings.AWS_S3_SECRET_ACCESS_KEY,
+            endpoint_url=settings.AWS_S3_DOMAIN_REPLACE,
+            config=botocore.client.Config(
+                region_name=settings.AWS_S3_REGION_NAME,
+                signature_version=settings.AWS_S3_SIGNATURE_VERSION,
+            ),
+        )
+    else:
+        s3_client = default_storage.connection.meta.client
+
+    # Generate the policy
+    policy = s3_client.generate_presigned_url(
+        ClientMethod="put_object",
+        Params={"Bucket": default_storage.bucket_name, "Key": key, "ACL": "private"},
+        ExpiresIn=settings.AWS_S3_UPLOAD_POLICY_EXPIRATION,
+    )
+
+    return policy