From 805f41c75776bcc3f90ecb09eb32f15c46c19362 Mon Sep 17 00:00:00 2001 From: AdrianoDev Date: Mon, 21 Sep 2026 22:48:52 +0200 Subject: [PATCH] fix(upload): tipo da magic bytes, non da Content-Type dichiarato (#36 punto 2) Il Content-Type del client non era verificato: un file "x.html" dichiarato image/png veniva salvato e poi servito per estensione come text/html, eseguito dal browser. Nuova detect_file_type() (PNG/JPEG/GIF/WebP/PDF) in files.py, usata anche dal logo in settings.py; l'estensione salvata segue sempre il tipo rilevato, mai il nome file dichiarato dal client. Il logo rifiuta SVG a prescindere (puo' portare script). Upload letti a blocchi da 1MB, interrotti appena superano il limite invece di bufferizzare tutto prima di controllare la dimensione. Co-Authored-By: Claude Sonnet 5 --- src/backend/api/routers/files.py | 85 ++++++++++++++++++++++------- src/backend/api/routers/settings.py | 49 ++++++++++------- src/backend/tests/test_files.py | 45 +++++++++++++++ src/backend/tests/test_settings.py | 81 +++++++++++++++++++++++++++ 4 files changed, 219 insertions(+), 41 deletions(-) diff --git a/src/backend/api/routers/files.py b/src/backend/api/routers/files.py index b2ee9b3..e6cc22e 100644 --- a/src/backend/api/routers/files.py +++ b/src/backend/api/routers/files.py @@ -26,6 +26,39 @@ ALLOWED_IMAGE_TYPES = {"image/jpeg", "image/png", "image/gif", "image/webp"} ALLOWED_PDF_TYPE = "application/pdf" ALLOWED_TYPES = ALLOWED_IMAGE_TYPES | {ALLOWED_PDF_TYPE} +# Magic-byte signatures: the client-declared Content-Type is only what the +# browser claimed and cannot be trusted (#36 point 2 — a file named "x.html" +# declared image/png was stored and then served by extension as text/html, +# which a browser executes). SVG is intentionally not recognised: it is not +# in ALLOWED_TYPES and can itself carry script content. +_MAGIC_SIGNATURES: list[tuple[bytes, str, str]] = [ + # (signature, mime type, canonical extension) + (b"\x89PNG\r\n\x1a\n", "image/png", ".png"), + (b"\xff\xd8\xff", "image/jpeg", ".jpg"), + (b"GIF87a", "image/gif", ".gif"), + (b"GIF89a", "image/gif", ".gif"), + (b"%PDF-", "application/pdf", ".pdf"), +] + +# Read uploads in chunks so a declared-huge body is rejected as soon as it +# crosses the limit, instead of being buffered into memory in full first. +_UPLOAD_CHUNK_SIZE = 1024 * 1024 + + +def detect_file_type(content: bytes) -> tuple[str, str] | None: + """Identify a file's real type from its magic bytes. + + Returns (mime_type, canonical_extension), or None when the content does + not match any type this system serves. + """ + for signature, mime, ext in _MAGIC_SIGNATURES: + if content.startswith(signature): + return mime, ext + # WebP: "RIFF" + 4-byte little-endian size + "WEBP"; the size varies. + if len(content) >= 12 and content[0:4] == b"RIFF" and content[8:12] == b"WEBP": + return "image/webp", ".webp" + return None + # Top-level folders nothing may DELETE through this router: statistics/reports are # audit records, the logo is only ever replaced via POST /api/settings/logo. _UNDELETABLE_PREFIXES = ("statistics/", "reports/", "logos/") @@ -178,25 +211,36 @@ async def upload_file( """Upload an image or PDF file. Files are stored in uploads/{recipe_id}/{version_id}/ - Thumbnails are generated for images. + Thumbnails are generated for images. The type is authoritatively + determined from the file's magic bytes, not the client-declared + Content-Type (#36 point 2): the stored extension always matches what was + actually detected, so it can never be served back as something else. """ - # Validate file type - if not validate_file_type(file.content_type): - raise HTTPException( - status_code=status.HTTP_400_BAD_REQUEST, - detail=f"File type {file.content_type} not allowed. Must be image or PDF.", - ) - - # Read file content - content = await file.read() + # Read in chunks, stopping as soon as the size limit is crossed instead + # of buffering an oversized body into memory first. + max_bytes = settings.max_upload_size_mb * 1024 * 1024 + buffer = bytearray() + while True: + chunk = await file.read(_UPLOAD_CHUNK_SIZE) + if not chunk: + break + buffer.extend(chunk) + if len(buffer) > max_bytes: + raise HTTPException( + status_code=status.HTTP_400_BAD_REQUEST, + detail=f"File size exceeds maximum {settings.max_upload_size_mb}MB", + ) + content = bytes(buffer) file_size = len(content) - # Validate file size - if not validate_file_size(file_size): + # Validate file type from the actual bytes, not file.content_type + detected = detect_file_type(content) + if detected is None: raise HTTPException( status_code=status.HTTP_400_BAD_REQUEST, - detail=f"File size {file_size} bytes exceeds maximum {settings.max_upload_size_mb}MB", + detail="File type not allowed. Must be a recognised image or PDF.", ) + detected_mime, detected_ext = detected # Determine storage path if recipe_id is not None and version_id is not None: @@ -206,18 +250,17 @@ async def upload_file( storage_dir.mkdir(parents=True, exist_ok=True) - # Sanitize the uploaded filename + # Sanitize the uploaded filename, then force its extension to the one + # actually detected: a declared/renamed extension is not trusted either. safe_filename = sanitize_filename(file.filename or "") + base_name = Path(safe_filename).stem or "file" + safe_filename = f"{base_name}{detected_ext}" # Generate unique filename if file already exists file_path = storage_dir / safe_filename counter = 1 while file_path.exists(): - name_parts = safe_filename.rsplit(".", 1) - if len(name_parts) == 2: - file_path = storage_dir / f"{name_parts[0]}_{counter}.{name_parts[1]}" - else: - file_path = storage_dir / f"{safe_filename}_{counter}" + file_path = storage_dir / f"{base_name}_{counter}{detected_ext}" counter += 1 # Write file @@ -225,7 +268,7 @@ async def upload_file( # Generate thumbnail for images thumbnail_path = None - if file.content_type in ALLOWED_IMAGE_TYPES: + if detected_mime in ALLOWED_IMAGE_TYPES: thumbnail_dir = storage_dir / "thumbnails" thumbnail_dir.mkdir(exist_ok=True) thumbnail_path = thumbnail_dir / file_path.name @@ -242,7 +285,7 @@ async def upload_file( return { "file_path": str(relative_path).replace("\\", "/"), "thumbnail_path": str(thumbnail_relative).replace("\\", "/") if thumbnail_relative else None, - "file_type": file.content_type, + "file_type": detected_mime, "file_size": file_size, } diff --git a/src/backend/api/routers/settings.py b/src/backend/api/routers/settings.py index c755033..c58fdab 100644 --- a/src/backend/api/routers/settings.py +++ b/src/backend/api/routers/settings.py @@ -1,6 +1,4 @@ """System settings router - read/update settings, upload company logo.""" -from pathlib import Path - from fastapi import APIRouter, Depends, File, HTTPException, UploadFile, status from sqlalchemy import select from sqlalchemy.ext.asyncio import AsyncSession @@ -8,6 +6,7 @@ from sqlalchemy.ext.asyncio import AsyncSession from src.backend.config import settings from src.backend.database import get_db from src.backend.api.middleware.api_key import get_current_user, require_admin_user +from src.backend.api.routers.files import ALLOWED_IMAGE_TYPES, detect_file_type from src.backend.models.orm.setting import SystemSetting from src.backend.models.orm.user import User @@ -126,35 +125,45 @@ async def upload_company_logo( """Upload company logo. Saves logo to uploads/logos/ and updates company_logo_path setting. - Only admins can upload the logo. + Only admins can upload the logo. The type is determined from the file's + magic bytes, not the client-declared Content-Type (#36 point 2). SVG is + refused outright: it can itself carry script content, and there is no + safe way to serve it inline for a logo shown on every page. """ - # Validate file type (must be image) - allowed_types = {"image/jpeg", "image/png", "image/gif", "image/webp", "image/svg+xml"} - if file.content_type not in allowed_types: - raise HTTPException( - status_code=status.HTTP_400_BAD_REQUEST, - detail=f"File type {file.content_type} not allowed. Must be an image.", - ) - - # Read file content - content = await file.read() + # Read in chunks, stopping as soon as the size limit is crossed instead + # of buffering an oversized body into memory first. + max_bytes = settings.max_upload_size_mb * 1024 * 1024 + buffer = bytearray() + chunk_size = 1024 * 1024 + while True: + chunk = await file.read(chunk_size) + if not chunk: + break + buffer.extend(chunk) + if len(buffer) > max_bytes: + raise HTTPException( + status_code=status.HTTP_400_BAD_REQUEST, + detail=f"File size exceeds maximum {settings.max_upload_size_mb}MB", + ) + content = bytes(buffer) file_size = len(content) - # Validate file size - max_bytes = settings.max_upload_size_mb * 1024 * 1024 - if file_size > max_bytes: + # Validate file type from the actual bytes (SVG never matches: refused) + detected = detect_file_type(content) + if detected is None or detected[0] not in ALLOWED_IMAGE_TYPES: raise HTTPException( status_code=status.HTTP_400_BAD_REQUEST, - detail=f"File size {file_size} bytes exceeds maximum {settings.max_upload_size_mb}MB", + detail="File type not allowed. Must be a recognised image (SVG is not accepted).", ) + _detected_mime, detected_ext = detected # Create logos directory logos_dir = settings.upload_path / "logos" logos_dir.mkdir(parents=True, exist_ok=True) - # Use a fixed filename for the company logo - file_extension = Path(file.filename).suffix - logo_filename = f"company_logo{file_extension}" + # Fixed filename for the company logo, extension forced to what was + # actually detected (not what the client claimed or the filename said). + logo_filename = f"company_logo{detected_ext}" logo_path = logos_dir / logo_filename # Write file diff --git a/src/backend/tests/test_files.py b/src/backend/tests/test_files.py index e9e0ab2..76e8bbf 100644 --- a/src/backend/tests/test_files.py +++ b/src/backend/tests/test_files.py @@ -65,6 +65,51 @@ class TestUploadFile: assert data["file_type"] == "image/jpeg" assert data["file_size"] > 0 + async def test_extension_forced_to_detected_type_ignores_filename( + self, client: AsyncClient, maker_user: User, tmp_path: Path + ): + """gitea#36 point 2: the stored extension follows the detected magic + bytes, not the client-supplied filename — even a misleading one.""" + jpeg_bytes = b"\xff\xd8\xff\xe0\x00\x10JFIF" + b"\x00" * 20 + b"\xff\xd9" + + with patch("src.backend.api.routers.files.settings") as mock_settings: + mock_settings.upload_path = tmp_path + mock_settings.max_upload_size_mb = 50 + + resp = await client.post( + "/api/files/upload", + headers=auth_headers(maker_user), + # Filename claims .txt; content is a real JPEG. + files={"file": ("photo.txt", io.BytesIO(jpeg_bytes), "text/plain")}, + ) + + assert resp.status_code == 200 + data = resp.json() + assert data["file_type"] == "image/jpeg" + assert data["file_path"].endswith(".jpg") + + async def test_upload_declared_type_lies_about_html_content( + self, client: AsyncClient, maker_user: User, tmp_path: Path + ): + """gitea#36 point 2: a file whose bytes are HTML must be rejected even + when the client declares Content-Type: image/png and names it .png — + this is what previously let a stored file be served back as + text/html by extension and executed by the browser.""" + html_bytes = b"" + + with patch("src.backend.api.routers.files.settings") as mock_settings: + mock_settings.upload_path = tmp_path + mock_settings.max_upload_size_mb = 50 + + resp = await client.post( + "/api/files/upload", + headers=auth_headers(maker_user), + files={"file": ("x.png", io.BytesIO(html_bytes), "image/png")}, + ) + + assert resp.status_code == 400 + assert not any(tmp_path.rglob("x.png")) + async def test_upload_invalid_type( self, client: AsyncClient, maker_user: User ): diff --git a/src/backend/tests/test_settings.py b/src/backend/tests/test_settings.py index 1ba48a1..ece91c3 100644 --- a/src/backend/tests/test_settings.py +++ b/src/backend/tests/test_settings.py @@ -1,4 +1,8 @@ """Tests for settings router (/api/settings).""" +import io +from pathlib import Path +from unittest.mock import patch + import pytest from httpx import AsyncClient from sqlalchemy.ext.asyncio import AsyncSession @@ -182,3 +186,80 @@ class TestUpdateSettingsValidation: json={"some_future_key": "anything"}, ) assert resp.status_code == 200 + + +class TestUploadLogo: + """POST /api/settings/logo tests (gitea#36 point 2).""" + + _PNG_BYTES = b"\x89PNG\r\n\x1a\n" + b"\x00" * 20 + + async def test_valid_png_accepted( + self, client: AsyncClient, admin_user: User, tmp_path: Path + ): + with patch("src.backend.api.routers.settings.settings") as mock_settings: + mock_settings.upload_path = tmp_path + mock_settings.max_upload_size_mb = 50 + + resp = await client.post( + "/api/settings/logo", + headers=auth_headers(admin_user), + files={"file": ("logo.png", io.BytesIO(self._PNG_BYTES), "image/png")}, + ) + + assert resp.status_code == 200 + assert resp.json()["logo_path"].endswith(".png") + + async def test_svg_refused_even_with_svg_content_type( + self, client: AsyncClient, admin_user: User, tmp_path: Path + ): + """SVG is refused outright, not just when it lies about its type.""" + svg_bytes = b'' + + with patch("src.backend.api.routers.settings.settings") as mock_settings: + mock_settings.upload_path = tmp_path + mock_settings.max_upload_size_mb = 50 + + resp = await client.post( + "/api/settings/logo", + headers=auth_headers(admin_user), + files={"file": ("logo.svg", io.BytesIO(svg_bytes), "image/svg+xml")}, + ) + + assert resp.status_code == 400 + assert not any(tmp_path.rglob("*.svg")) + + async def test_pdf_refused_for_logo( + self, client: AsyncClient, admin_user: User, tmp_path: Path + ): + """The logo slot only accepts images: files.py's PDF signature must + not slip through settings.py's narrower ALLOWED_IMAGE_TYPES check.""" + pdf_bytes = b"%PDF-1.4\n" + b"\x00" * 20 + + with patch("src.backend.api.routers.settings.settings") as mock_settings: + mock_settings.upload_path = tmp_path + mock_settings.max_upload_size_mb = 50 + + resp = await client.post( + "/api/settings/logo", + headers=auth_headers(admin_user), + files={"file": ("logo.pdf", io.BytesIO(pdf_bytes), "application/pdf")}, + ) + + assert resp.status_code == 400 + + async def test_html_disguised_as_png_refused( + self, client: AsyncClient, admin_user: User, tmp_path: Path + ): + html_bytes = b"" + + with patch("src.backend.api.routers.settings.settings") as mock_settings: + mock_settings.upload_path = tmp_path + mock_settings.max_upload_size_mb = 50 + + resp = await client.post( + "/api/settings/logo", + headers=auth_headers(admin_user), + files={"file": ("logo.png", io.BytesIO(html_bytes), "image/png")}, + ) + + assert resp.status_code == 400