From 82cf211b88532c00279a0911ae4d931ef92df7bf Mon Sep 17 00:00:00 2001 From: Manuel Raynaud Date: Wed, 28 Jan 2026 07:59:11 +0100 Subject: [PATCH] =?UTF-8?q?=F0=9F=94=92=EF=B8=8F(backend)=20prevent=20mism?= =?UTF-8?q?atch=20mimetype=20between=20object=20storage=20and=20app?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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 | 6 ++- src/backend/core/api/utils.py | 6 +++ src/backend/core/api/viewsets.py | 31 +++++++++++ .../tests/items/test_api_item_upload_ended.py | 54 +++++++++++++++++++ 4 files changed, 96 insertions(+), 1 deletion(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 44bcf7dd..6afa9005 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -10,12 +10,16 @@ and this project adheres to ### Added -- 🏗️ (ds_proxy) introduce how to use ds_proxy with Drive +- 🏗️(ds_proxy) introduce how to use ds_proxy with Drive ### Changed - 🔥(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 diff --git a/src/backend/core/api/utils.py b/src/backend/core/api/utils.py index facc01d3..fa2b31cd 100644 --- a/src/backend/core/api/utils.py +++ b/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 diff --git a/src/backend/core/api/viewsets.py b/src/backend/core/api/viewsets.py index e9431035..0343d4bc 100644 --- a/src/backend/core/api/viewsets.py +++ b/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) diff --git a/src/backend/core/tests/items/test_api_item_upload_ended.py b/src/backend/core/tests/items/test_api_item_upload_ended.py index df43aec3..1067d12b 100644 --- a/src/backend/core/tests/items/test_api_item_upload_ended.py +++ b/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'' + ), + 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"}