diff --git a/backend/app/services/importer.py b/backend/app/services/importer.py index e2b94eb..0e366d9 100644 --- a/backend/app/services/importer.py +++ b/backend/app/services/importer.py @@ -30,6 +30,7 @@ from ..models import ( PostAttachment, Source, ) +from ..utils import safe_probe from ..utils.paths import derive_subdir, derive_top_level_artist, hash_suffixed_name from ..utils.phash import compute_phash, find_similar from ..utils.sidecar import find_sidecar, parse_sidecar @@ -407,6 +408,29 @@ class Importer: return ImportResult(status="attached") def _import_archive(self, source: Path) -> ImportResult: + # Layer-3 isolation: bomb-size guard + integrity test in a + # spawned child BEFORE extracting in this process. A + # decompression bomb or a native-lib crash on a malformed + # archive is contained to the child; we reject the file cleanly + # instead of OOMing/segfaulting the import worker. extract_archive + # is already fail-soft for plain exceptions, so this only adds + # the hard-crash protection. + probe = safe_probe.probe_archive(source) + if not probe.ok: + if probe.crashed: + return ImportResult( + status="failed", + error=f"archive probe crashed/timed out: {probe.reason}", + ) + # Clean rejection (bomb cap exceeded, integrity mismatch): + # still preserve the archive file itself as an attachment so + # nothing silently vanishes, matching extract_archive's + # fail-soft contract. + artist = self._resolve_artist(source) + post = self._post_for_sidecar(source, artist) + self._capture_attachment(source, post=post, artist=artist, resolved=True) + return ImportResult(status="attached") + artist = self._resolve_artist(source) post = self._post_for_sidecar(source, artist) member_ids: list[int] = [] @@ -446,7 +470,25 @@ class Importer: # Compute file dimensions (images only) and apply filters. width = height = None has_alpha = False - if not is_video(source): + if is_video(source): + # Layer-3 isolation: validate the container via ffprobe (a + # separate process) before the rest of the pipeline touches + # it. A corrupt video that would crash a decoder is rejected + # cleanly here, and we capture width/height for free (the + # importer didn't previously record video dimensions). + probe = safe_probe.probe_video(source) + if not probe.ok: + if probe.crashed: + return ImportResult( + status="failed", + error=f"video probe crashed/timed out: {probe.reason}", + ) + return ImportResult( + status="skipped", skip_reason=SkipReason.invalid_image, + error=probe.reason, + ) + width, height = probe.width, probe.height + else: try: with Image.open(source) as im: im.verify() diff --git a/backend/app/tasks/ml.py b/backend/app/tasks/ml.py index c4b8ddc..b114eec 100644 --- a/backend/app/tasks/ml.py +++ b/backend/app/tasks/ml.py @@ -71,6 +71,18 @@ def tag_and_embed(self, image_id: int) -> dict: embedder = get_embedder() if _is_video(src): + # Layer-3 isolation: ffprobe (a separate process) validates + # the container before we burn ~20 GPU ops sampling frames + # from it. A corrupt video that would crash the frame + # decoder is rejected cleanly here instead of taking down + # the ml-worker. Operator-flagged 2026-05-28. + from ..utils import safe_probe + vprobe = safe_probe.probe_video(src) + if not vprobe.ok: + return { + "status": "bad_video", "image_id": image_id, + "reason": vprobe.reason, + } frames = _sample_video_frames( src, int(os.environ.get("VIDEO_ML_FRAMES", "10")) ) diff --git a/backend/app/utils/safe_probe.py b/backend/app/utils/safe_probe.py new file mode 100644 index 0000000..02c387c --- /dev/null +++ b/backend/app/utils/safe_probe.py @@ -0,0 +1,170 @@ +"""Subprocess-isolated media probes (Layer 3 of import resilience). + +A malformed video or archive can hard-crash the worker process — a +decoder OOM, a native-lib segfault, or a decompression bomb. A hard +crash leaves no terminal flip, so the recovery sweep re-queues the row +and it crashes again: a poison-pill loop (the Layer-1 cap is the +backstop, but isolating the crash is better — the file gets a clean +terminal failure and the worker never dies). + +These probes run the risky read in a way that contains the blast: + +- Video: `ffprobe` is a separate binary, so a crash decoding the + container kills only ffprobe (non-zero exit), never the worker. Also + returns width/height, which the importer didn't previously capture + for videos. +- Archive: an uncompressed-size guard (catches decompression bombs + before they OOM anything) plus an integrity test in a spawned child + (catches native-lib crashes on a malformed archive). A child segfault + / OOM shows up as a non-zero exit code, not a dead worker. + +Images are intentionally NOT probed here: Pillow raises (it doesn't +segfault) on the realistic corrupt-image cases, the importer already +catches that as an invalid_image skip, and a subprocess per image would +wreck deep-scan throughput on a large library. Add an image branch only +if a real image-induced worker crash is ever observed. + +Operator-requested 2026-05-28 (Layer 3). +""" + +import json +import multiprocessing as mp +import subprocess +from dataclasses import dataclass +from pathlib import Path + +VIDEO_PROBE_TIMEOUT_SECONDS = 60 +ARCHIVE_PROBE_TIMEOUT_SECONDS = 120 +# Refuse archives whose total UNCOMPRESSED size exceeds this — the +# classic decompression-bomb guard (a 4 GB cap comfortably clears real +# art-pack archives while stopping a few-KB zip that expands to TB). +MAX_ARCHIVE_UNCOMPRESSED_BYTES = 4 * 1024 * 1024 * 1024 + + +@dataclass(frozen=True) +class ProbeResult: + ok: bool + # crashed=True means the probe HARD-FAILED (subprocess killed by a + # signal, OOM, or timeout) — the poison-pill signature. crashed=False + # with ok=False means a clean rejection (corrupt-but-handled, + # bomb-size-exceeded, integrity mismatch). Callers map crashed → a + # terminal 'failed', clean → a 'skipped'/'failed' of their choosing. + crashed: bool = False + reason: str | None = None + width: int | None = None + height: int | None = None + + +def probe_video(path: Path, *, timeout: float = VIDEO_PROBE_TIMEOUT_SECONDS) -> ProbeResult: + """Validate a video container + first video stream via ffprobe.""" + try: + out = subprocess.run( + [ + "ffprobe", "-v", "error", + "-select_streams", "v:0", + "-show_entries", "stream=width,height", + "-of", "json", str(path), + ], + capture_output=True, text=True, timeout=timeout, + ) + except subprocess.TimeoutExpired: + return ProbeResult(ok=False, crashed=True, reason="ffprobe timed out") + except OSError as exc: + # ffprobe missing / not executable — environmental, not the + # file's fault. Treat as a clean non-crash failure so the import + # path can decide (it currently proceeds without dims). + return ProbeResult(ok=False, crashed=False, reason=f"ffprobe unavailable: {exc}") + if out.returncode != 0: + return ProbeResult( + ok=False, crashed=False, + reason=f"ffprobe rejected the file: {out.stderr.strip()[:200]}", + ) + try: + streams = (json.loads(out.stdout) or {}).get("streams") or [] + except json.JSONDecodeError as exc: + return ProbeResult(ok=False, crashed=False, reason=f"ffprobe output parse failed: {exc}") + if not streams: + return ProbeResult(ok=False, crashed=False, reason="no decodable video stream") + return ProbeResult( + ok=True, width=streams[0].get("width"), height=streams[0].get("height"), + ) + + +def probe_archive(path: Path, *, timeout: float = ARCHIVE_PROBE_TIMEOUT_SECONDS) -> ProbeResult: + """Bomb-size guard + isolated integrity test for an archive.""" + ctx = mp.get_context("spawn") + q = ctx.Queue() + proc = ctx.Process(target=_archive_probe_target, args=(str(path), q)) + proc.start() + proc.join(timeout) + if proc.is_alive(): + proc.terminate() + proc.join(5) + return ProbeResult(ok=False, crashed=True, reason="archive probe timed out") + if proc.exitcode != 0: + # Negative exitcode = killed by signal (segfault); positive = + # the child os._exit'd or was OOM-killed. Either way the file + # hard-crashed the probe — the poison-pill signature. + return ProbeResult( + ok=False, crashed=True, + reason=f"archive probe crashed (exit {proc.exitcode})", + ) + try: + outcome = q.get(timeout=5) + except Exception: # noqa: BLE001 — empty queue / broken pipe + return ProbeResult(ok=False, crashed=True, reason="archive probe produced no result") + status, detail = outcome + if status == "ok": + return ProbeResult(ok=True) + return ProbeResult(ok=False, crashed=False, reason=detail) + + +def _archive_probe_target(path_str: str, q) -> None: + """Runs in the spawned child. Reads member sizes (bomb guard) then + runs the format's integrity test. Puts ('ok', None) or + ('error', reason). A crash/OOM here never reaches the queue — the + parent reads the non-zero exit code instead.""" + path = Path(path_str) + ext = path.suffix.lower() + try: + total, test_bad = _inspect_archive(path, ext) + except Exception as exc: # noqa: BLE001 — clean rejection + q.put(("error", f"{type(exc).__name__}: {exc}")) + return + if total is not None and total > MAX_ARCHIVE_UNCOMPRESSED_BYTES: + gib = total / (1024 ** 3) + q.put(("error", f"uncompressed size {gib:.1f} GiB exceeds the bomb-guard cap")) + return + if test_bad is not None: + q.put(("error", f"integrity test failed at member {test_bad!r}")) + return + q.put(("ok", None)) + + +def _inspect_archive(path: Path, ext: str): + """Return (total_uncompressed_bytes | None, first_bad_member | None) + for the archive. Format-specific; raises on a structurally-broken + container (caught by the child as a clean rejection).""" + if ext in (".zip", ".cbz"): + import zipfile + + with zipfile.ZipFile(path) as zf: + total = sum(zi.file_size for zi in zf.infolist()) + return total, zf.testzip() + if ext == ".rar": + import rarfile + + with rarfile.RarFile(path) as rf: + total = sum(getattr(ri, "file_size", 0) for ri in rf.infolist()) + rf.testrar() + return total, None + if ext == ".7z": + import py7zr + + with py7zr.SevenZipFile(path, "r") as zf: + info = zf.archiveinfo() + total = getattr(info, "uncompressed", None) + ok = zf.test() # True / None when all members pass + return total, (None if ok in (True, None) else "7z test reported corruption") + # Unknown extension — nothing to test; treat as clean. + return None, None diff --git a/tests/test_safe_probe.py b/tests/test_safe_probe.py new file mode 100644 index 0000000..6dd7ea2 --- /dev/null +++ b/tests/test_safe_probe.py @@ -0,0 +1,71 @@ +"""Layer-3 subprocess-isolated probe tests. + +The bomb-guard cap is exercised against `_archive_probe_target` directly +(in-process, where a monkeypatch on the module constant takes effect) — +spawn re-imports the module in the child, so a parent-process +monkeypatch wouldn't reach the spawned worker. +""" + +import multiprocessing as mp +import zipfile +from pathlib import Path + +from backend.app.utils import safe_probe + + +def _zip(path, entries): + with zipfile.ZipFile(path, "w") as zf: + for name, data in entries.items(): + zf.writestr(name, data) + + +def test_probe_archive_valid_zip(tmp_path): + z = tmp_path / "ok.zip" + _zip(z, {"a.jpg": b"hello", "b.png": b"world"}) + res = safe_probe.probe_archive(z) + assert res.ok is True + assert res.crashed is False + + +def test_probe_archive_corrupt_zip_clean_rejection(tmp_path): + z = tmp_path / "broken.zip" + z.write_bytes(b"PK\x03\x04 not really a zip past here") + res = safe_probe.probe_archive(z) + assert res.ok is False + # Corrupt-but-handled (zipfile raises BadZipFile in the child) — a + # clean rejection, not a hard crash. + assert res.crashed is False + assert res.reason + + +def test_inspect_archive_reports_size_and_clean_integrity(tmp_path): + z = tmp_path / "sized.zip" + _zip(z, {"a.txt": b"x" * 100, "b.txt": b"y" * 50}) + total, bad = safe_probe._inspect_archive(z, ".zip") + assert total == 150 + assert bad is None + + +def test_archive_probe_target_bomb_guard(tmp_path, monkeypatch): + """In-process call to the child target so the monkeypatched cap + takes effect. A normal zip whose uncompressed size exceeds the + (lowered) cap is rejected with the bomb-guard reason.""" + monkeypatch.setattr(safe_probe, "MAX_ARCHIVE_UNCOMPRESSED_BYTES", 10) + z = tmp_path / "bomb.zip" + _zip(z, {"big.txt": b"x" * 5000}) # 5000 uncompressed > 10-byte cap + q = mp.get_context("spawn").Queue() + safe_probe._archive_probe_target(str(z), q) + status, detail = q.get(timeout=5) + assert status == "error" + assert "bomb-guard cap" in detail + + +def test_probe_video_non_video_is_not_ok(tmp_path): + """A text file is not a decodable video. Whether ffprobe is present + (returncode != 0) or absent (OSError → 'unavailable'), the result is + ok=False. We don't assert on crashed/reason so the test is robust to + ffprobe presence in CI.""" + f = tmp_path / "nope.txt" + f.write_text("definitely not a video container") + res = safe_probe.probe_video(f) + assert res.ok is False