Adriano/TieMeasureFlow
1
0
Fork 0
Code Issues 4 Pull Requests Actions Packages Projects Releases 3 Wiki Activity

fix(upload): tipo da magic bytes, non da Content-Type dichiarato (#36 punto 2)

Browse Source 
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 <noreply@anthropic.com>
This commit is contained in:
AdrianoDev
2026-09-21 22:48:52 +02:00
parent bfe66a5710
commit 805f41c757
4 changed files with 219 additions and 41 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/backend/api/routers/files.py
+64 -21
View File
@@ -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,
}
src/backend/api/routers/settings.py
+29 -20
View File
@@ -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
src/backend/tests/test_files.py
+45
View File
@@ -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"<html><body><script>alert(1)</script></body></html>"
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
):
src/backend/tests/test_settings.py
+81
View File
@@ -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'<svg xmlns="http://www.w3.org/2000/svg"><script>alert(1)</script></svg>'
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"<html><body><script>alert(1)</script></body></html>"
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