🔒️(backend) prevent mismatch mimetype between object storage and app

If there is a mismatch mimetype between the object in the object storage
and items in the application, this can lead to previewing files in the
browser that are not allowed. We want to prevent this, the mimetype in
the object storage is overriden in that case.

CHANGELOG.md

@@ -16,6 +16,10 @@ and this project adheres to
 
 - 🔥(backend) remove usage of atomic transaction for item creation
 
+### Security
+
+- 🔒️(backend) prevent mismatch mimetype between object storage and application
+
 ## [v0.11.1] - 2026-01-13
 
 ### Fixed

src/backend/core/api/utils.py

@@ -1,5 +1,6 @@
 """Util to generate S3 authorization headers for object storage access control"""
 
+import logging
 import mimetypes
 from datetime import datetime
 
@@ -10,6 +11,8 @@ import boto3
 import botocore
 import magic
 
+logger = logging.getLogger(__name__)
+
 
 def flat_to_nested(items):
     """
@@ -188,6 +191,9 @@ 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)
+
     # Strategy: Prefer content-based detection, but use extension if:
     # 1. Content detection returns generic types (application/octet-stream, text/plain)
     # 2. Content detection fails or returns None

src/backend/core/api/viewsets.py

@@ -23,6 +23,7 @@ from django.utils.text import slugify
 
 import posthog
 import rest_framework as drf
+from botocore.exceptions import ClientError
 from corsheaders.middleware import (
     ACCESS_CONTROL_ALLOW_METHODS,
     ACCESS_CONTROL_ALLOW_ORIGIN,
@@ -708,6 +709,7 @@ class ItemViewSet(
             )["Body"].read()
 
         # Use improved MIME type detection combining magic bytes and file extension
+        logger.info("upload_ended: detecting mimetype for file: %s", item.file_key)
         mimetype = utils.detect_mimetype(file_head, filename=item.filename)
 
         if (
@@ -731,6 +733,35 @@ class ItemViewSet(
 
         item.save(update_fields=["upload_state", "mimetype", "size"])
 
+        if head_response["ContentType"] != mimetype:
+            logger.info(
+                "upload_ended: content type mismatch between object storage and item,"
+                " updating from %s to %s",
+                head_response["ContentType"],
+                mimetype,
+            )
+            try:
+                s3_client.copy_object(
+                    Bucket=default_storage.bucket_name,
+                    Key=item.file_key,
+                    CopySource={
+                        "Bucket": default_storage.bucket_name,
+                        "Key": item.file_key,
+                    },
+                    ContentType=mimetype,
+                    Metadata=head_response["Metadata"],
+                    MetadataDirective="REPLACE",
+                )
+            except ClientError as error:
+                # Log an exception but don't stop the action.
+                logger.exception(
+                    "Changing content type of item %s on object storage failed with error code %s"
+                    " and error message %s",
+                    item.id,
+                    error.response["Error"]["Code"],
+                    error.response["Error"]["Message"],
+                )
+
         malware_detection.analyse_file(item.file_key, item_id=item.id)
 
         serializer = self.get_serializer(item)

src/backend/core/tests/items/test_api_item_upload_ended.py

@@ -302,3 +302,57 @@ def test_api_item_upload_ended_mimetype_not_allowed_not_checking_mimetype(settin
     assert item.size == 8
 
     assert response.json()["mimetype"] == "text/plain"
+
+
+def test_api_upload_ended_mismatch_mimetype_with_object_storage(caplog):
+    """
+    Object on storage should have the same mimetype than the one saved in the
+    Item object.
+    """
+    user = factories.UserFactory()
+    client = APIClient()
+    client.force_login(user)
+
+    item = factories.ItemFactory(
+        type=ItemTypeChoices.FILE,
+        filename="my_file.pdf",
+        title="my_file.pdf",
+        users=[(user, "owner")],
+    )
+
+    s3_client = default_storage.connection.meta.client
+
+    s3_client.put_object(
+        Bucket=default_storage.bucket_name,
+        Key=item.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=item.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/items/{item.id!s}/upload-ended/")
+    assert (
+        "upload_ended: content type mismatch between object storage and item, "
+        "updating from text/html to application/pdf" in caplog.text
+    )
+    assert response.status_code == 200
+
+    item.refresh_from_db()
+
+    assert item.mimetype == "application/pdf"
+
+    head_object = s3_client.head_object(
+        Bucket=default_storage.bucket_name, Key=item.file_key
+    )
+    assert head_object["ContentType"] == "application/pdf"
+    assert head_object["Metadata"] == {"foo": "bar"}