From 25e6e1b10fa0d141ce8837ef0646af9ae5519e5a Mon Sep 17 00:00:00 2001 From: Bryan Van Deusen Date: Wed, 20 May 2026 20:39:17 -0400 Subject: [PATCH] =?UTF-8?q?feat(fc3c):=20FileValidator=20=E2=80=94=20magic?= =?UTF-8?q?-byte=20head/tail=20check=20for=20fresh=20writes?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Co-Authored-By: Claude Opus 4.7 (1M context) --- backend/app/services/file_validator.py | 111 +++++++++++++++++++++++++ tests/test_file_validator.py | 85 +++++++++++++++++++ 2 files changed, 196 insertions(+) create mode 100644 backend/app/services/file_validator.py create mode 100644 tests/test_file_validator.py diff --git a/backend/app/services/file_validator.py b/backend/app/services/file_validator.py new file mode 100644 index 0000000..791b142 --- /dev/null +++ b/backend/app/services/file_validator.py @@ -0,0 +1,111 @@ +"""Magic-byte head/tail validator for fresh gallery-dl writes. + +Catches truncated downloads (servers lying about Content-Length, TCP +RST mid-transfer, etc.) before they enter the library. O(1) per file +— reads only head and tail bytes; never decodes pixels. + +Fail-open: unknown formats and unreadable files pass through. Format +detection is by extension; we only validate what the downloader is +expected to write. +""" + +from __future__ import annotations + +from dataclasses import dataclass +from pathlib import Path + + +JPEG_HEAD = b"\xff\xd8\xff" +JPEG_TAIL = b"\xff\xd9" + +PNG_HEAD = b"\x89PNG\r\n\x1a\n" +PNG_TAIL = b"\x00\x00\x00\x00IEND\xaeB`\x82" + +GIF_HEAD_87A = b"GIF87a" +GIF_HEAD_89A = b"GIF89a" +GIF_TAIL = b"\x3b" + +WEBP_RIFF = b"RIFF" +WEBP_FORMAT = b"WEBP" + + +@dataclass(frozen=True) +class ValidationResult: + ok: bool + format: str | None = None + reason: str | None = None + size: int = 0 + + +VALIDATED_EXTENSIONS = {".jpg", ".jpeg", ".png", ".gif", ".webp"} + + +def is_validatable(path: Path) -> bool: + return path.suffix.lower() in VALIDATED_EXTENSIONS + + +def validate_file(path: Path) -> ValidationResult: + """Returns ok=False if the file is structurally incomplete. + + Never raises. Unknown formats return ok=True with format=None. + Unreadable files surface as ok=False so callers can quarantine. + """ + ext = path.suffix.lower() + if ext not in VALIDATED_EXTENSIONS: + return ValidationResult(ok=True, format=None) + + try: + size = path.stat().st_size + except FileNotFoundError: + return ValidationResult(ok=False, reason="file not found") + except OSError as exc: + return ValidationResult(ok=False, reason=f"stat failed: {exc}") + + if size < 16: + return ValidationResult(ok=False, format=ext.lstrip("."), reason="too small", size=size) + + try: + with path.open("rb") as fh: + head = fh.read(32) + fh.seek(max(0, size - 32)) + tail = fh.read(32) + except OSError as exc: + return ValidationResult(ok=False, reason=f"read failed: {exc}", size=size) + + if ext in (".jpg", ".jpeg"): + if not head.startswith(JPEG_HEAD): + return ValidationResult(ok=False, format="jpeg", reason="missing JPEG SOI marker", size=size) + if not tail.endswith(JPEG_TAIL): + return ValidationResult(ok=False, format="jpeg", reason="missing JPEG EOI marker", size=size) + return ValidationResult(ok=True, format="jpeg", size=size) + + if ext == ".png": + if not head.startswith(PNG_HEAD): + return ValidationResult(ok=False, format="png", reason="missing PNG signature", size=size) + if not tail.endswith(PNG_TAIL): + return ValidationResult(ok=False, format="png", reason="missing PNG IEND chunk", size=size) + return ValidationResult(ok=True, format="png", size=size) + + if ext == ".gif": + if not (head.startswith(GIF_HEAD_87A) or head.startswith(GIF_HEAD_89A)): + return ValidationResult(ok=False, format="gif", reason="missing GIF signature", size=size) + if not tail.endswith(GIF_TAIL): + return ValidationResult(ok=False, format="gif", reason="missing GIF trailer", size=size) + return ValidationResult(ok=True, format="gif", size=size) + + if ext == ".webp": + if not (head.startswith(WEBP_RIFF) and WEBP_FORMAT in head[:16]): + return ValidationResult(ok=False, format="webp", reason="missing WebP RIFF/WEBP marker", size=size) + try: + chunk_size = int.from_bytes(head[4:8], "little") + except Exception: + return ValidationResult(ok=False, format="webp", reason="malformed WebP chunk size", size=size) + if chunk_size + 8 != size: + return ValidationResult( + ok=False, format="webp", + reason=f"WebP RIFF size mismatch: header says {chunk_size + 8}, file is {size}", + size=size, + ) + return ValidationResult(ok=True, format="webp", size=size) + + return ValidationResult(ok=True, format=None, size=size) diff --git a/tests/test_file_validator.py b/tests/test_file_validator.py new file mode 100644 index 0000000..5a36ed2 --- /dev/null +++ b/tests/test_file_validator.py @@ -0,0 +1,85 @@ +from backend.app.services.file_validator import ( + is_validatable, + validate_file, +) + + +def test_is_validatable_only_known_extensions(tmp_path): + assert is_validatable(tmp_path / "x.jpg") + assert is_validatable(tmp_path / "x.JPEG") + assert is_validatable(tmp_path / "x.png") + assert is_validatable(tmp_path / "x.gif") + assert is_validatable(tmp_path / "x.webp") + assert not is_validatable(tmp_path / "x.json") + assert not is_validatable(tmp_path / "x.mp4") + + +def test_valid_jpeg_passes(tmp_path): + from PIL import Image + p = tmp_path / "good.jpg" + Image.new("RGB", (32, 32), (200, 50, 50)).save(p, "JPEG") + r = validate_file(p) + assert r.ok is True + assert r.format == "jpeg" + assert r.size > 0 + + +def test_truncated_jpeg_fails(tmp_path): + from PIL import Image + p = tmp_path / "bad.jpg" + Image.new("RGB", (32, 32), (200, 50, 50)).save(p, "JPEG") + raw = p.read_bytes() + p.write_bytes(raw[:-16]) + r = validate_file(p) + assert r.ok is False + assert r.format == "jpeg" + assert "EOI" in r.reason + + +def test_missing_jpeg_soi_fails(tmp_path): + p = tmp_path / "junk.jpg" + p.write_bytes(b"\x00" * 64) + r = validate_file(p) + assert r.ok is False + assert "SOI" in r.reason + + +def test_truncated_png_fails(tmp_path): + from PIL import Image + p = tmp_path / "bad.png" + Image.new("RGB", (32, 32), (10, 10, 10)).save(p, "PNG") + raw = p.read_bytes() + p.write_bytes(raw[:-12]) + r = validate_file(p) + assert r.ok is False + assert r.format == "png" + + +def test_gif_signature_and_trailer(tmp_path): + p = tmp_path / "bad.gif" + p.write_bytes(b"GIF89a" + b"\x00" * 64) + r = validate_file(p) + assert r.ok is False + assert r.format == "gif" + + +def test_unknown_extension_passes(tmp_path): + p = tmp_path / "x.bin" + p.write_bytes(b"\x00" * 64) + r = validate_file(p) + assert r.ok is True + assert r.format is None + + +def test_missing_file_fails_cleanly(tmp_path): + r = validate_file(tmp_path / "nope.jpg") + assert r.ok is False + assert "not found" in r.reason + + +def test_too_small_fails(tmp_path): + p = tmp_path / "tiny.jpg" + p.write_bytes(b"\xff\xd8\xff") + r = validate_file(p) + assert r.ok is False + assert r.reason == "too small"