From e925d9f43b34a584bedf3d2a3e63f495dac9e50b Mon Sep 17 00:00:00 2001
From: xiaoxia <dev@xiaoxiajianji.com>
Date: Fri, 26 Jun 2026 18:21:46 +0800
Subject: [PATCH] fix(P2-5): Add server-side MIME type validation for file
 uploads

---
 apps/api/app/api/routes/upload.py | 64 ++++++++++++++++++++++++++++++-
 1 file changed, 62 insertions(+), 2 deletions(-)

diff --git a/apps/api/app/api/routes/upload.py b/apps/api/app/api/routes/upload.py
index a82fba7c1..f8b669a0c 100644
--- a/apps/api/app/api/routes/upload.py
+++ b/apps/api/app/api/routes/upload.py
@@ -1,3 +1,4 @@
+import os
 from typing import Any
 from uuid import uuid4
 
@@ -26,6 +27,59 @@ from packages.ports.workspace_member_repository import WorkspaceMemberRepository
 
 router = APIRouter()
 
+# Allowed MIME types for uploads
+ALLOWED_VIDEO_TYPES = {"video/mp4", "video/quicktime", "video/x-msvideo", "video/webm", "video/x-matroska"}
+ALLOWED_AUDIO_TYPES = {"audio/mpeg", "audio/wav", "audio/ogg", "audio/flac", "audio/aac"}
+ALLOWED_IMAGE_TYPES = {"image/jpeg", "image/png", "image/gif", "image/webp", "image/svg+xml"}
+ALLOWED_CONTENT_TYPES = ALLOWED_VIDEO_TYPES | ALLOWED_AUDIO_TYPES | ALLOWED_IMAGE_TYPES
+
+# Extension to MIME type mapping
+EXTENSION_TO_MIME = {
+    ".mp4": "video/mp4",
+    ".mov": "video/quicktime",
+    ".avi": "video/x-msvideo",
+    ".webm": "video/webm",
+    ".mkv": "video/x-matroska",
+    ".mp3": "audio/mpeg",
+    ".wav": "audio/wav",
+    ".ogg": "audio/ogg",
+    ".flac": "audio/flac",
+    ".aac": "audio/aac",
+    ".jpg": "image/jpeg",
+    ".jpeg": "image/jpeg",
+    ".png": "image/png",
+    ".gif": "image/gif",
+    ".webp": "image/webp",
+    ".svg": "image/svg+xml",
+}
+
+
+def _get_mime_type_from_filename(filename: str) -> str:
+    """Get MIME type from file extension."""
+    _, ext = os.path.splitext(filename.lower())
+    return EXTENSION_TO_MIME.get(ext, "application/octet-stream")
+
+
+def _validate_content_type(content_type: str, filename: str) -> str:
+    """Validate and normalize content type.
+    
+    If content_type is not provided or invalid, derive from filename.
+    """
+    if content_type and content_type in ALLOWED_CONTENT_TYPES:
+        return content_type
+    
+    # Try to get from filename
+    mime_from_file = _get_mime_type_from_filename(filename)
+    if mime_from_file in ALLOWED_CONTENT_TYPES:
+        return mime_from_file
+    
+    # Content type not allowed
+    raise HTTPException(
+        status_code=status.HTTP_400_BAD_REQUEST,
+        detail=f"File type '{content_type or mime_from_file}' is not allowed. "
+               f"Allowed types: video, audio, and image files.",
+    )
+
 
 def _require_project_and_library(
     workspace_id: str,
@@ -81,6 +135,9 @@ async def prepare_direct_upload(
             detail=f"File exceeds upload limit ({settings.OSS_DIRECT_UPLOAD_MAX_MB}MB)",
         )
 
+    # P2-5: Validate content type on server side
+    validated_content_type = _validate_content_type(request.content_type, request.filename)
+
     require_workspace_member(request.workspace_id, authenticated_user, workspace_member_repository)
     _require_project_and_library(
         request.workspace_id,
@@ -96,7 +153,7 @@ async def prepare_direct_upload(
     try:
         payload = storage_service.create_direct_upload_post(
             storage_key=storage_key,
-            content_type=request.content_type or "application/octet-stream",
+            content_type=validated_content_type,
             max_size_bytes=max_size_bytes,
             expires_seconds=settings.OSS_DIRECT_UPLOAD_EXPIRE_SECONDS,
         )
@@ -165,13 +222,16 @@ async def upload_asset(
     require_workspace_member(workspace_id, authenticated_user, workspace_member_repository)
     _require_project_and_library(workspace_id, project_id, library_id, project_repository, asset_library_repository)
 
+    # P2-5: Validate and normalize content type on server side
+    validated_content_type = _validate_content_type(file.content_type, file.filename)
+
     file_id = uuid4().hex[:8]
     storage_key = f"uploads/{file_id}/{file.filename}"
 
     file_url = storage_service.upload_file(
         file.file,
         storage_key,
-        content_type=file.content_type or "application/octet-stream",
+        content_type=validated_content_type,
     )
 
     job = _submit_ingest_job(