feat(fc3c): FileValidator — magic-byte head/tail check for fresh writes
Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
This commit is contained in:
@@ -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)
|
||||
@@ -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"
|
||||
Reference in New Issue
Block a user