"""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 subprocess import sys 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 # Repo root for the subprocess cwd so `python -m backend.app.utils.*` # resolves regardless of where Celery / pytest started. backend/app/utils # = parents[0]; backend/app = parents[1]; backend = parents[2]; repo root # = parents[3]. _REPO_ROOT = Path(__file__).resolve().parents[3] _PROBE_RUNNER_MODULE = "backend.app.utils._archive_probe_runner" @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. Runs via subprocess (not multiprocessing.Process) because Celery's prefork worker pool is daemon-mode and Python's multiprocessing forbids daemon processes from spawning children ("AssertionError: daemonic processes are not allowed to have children"). subprocess has no such restriction and still gives the crash isolation: a probe segfault/OOM exits non-zero rather than killing the worker. """ try: result = subprocess.run( [sys.executable, "-m", _PROBE_RUNNER_MODULE, str(path)], capture_output=True, text=True, timeout=timeout, cwd=str(_REPO_ROOT), ) except subprocess.TimeoutExpired: return ProbeResult(ok=False, crashed=True, reason="archive probe timed out") if result.returncode != 0: # Negative = killed by signal (segfault); positive = unhandled # exception or OOM-kill. Either way: poison-pill signature. return ProbeResult( ok=False, crashed=True, reason=f"archive probe crashed (exit {result.returncode})", ) last_line = result.stdout.strip().splitlines()[-1:] or [""] try: outcome = json.loads(last_line[0]) except json.JSONDecodeError as exc: return ProbeResult( ok=False, crashed=True, reason=f"archive probe produced no parseable result: {exc}", ) if outcome.get("status") == "ok": return ProbeResult(ok=True) return ProbeResult( ok=False, crashed=False, reason=outcome.get("detail") or "archive probe rejected", ) def _run_probe(path_str: str) -> tuple[str, str | None]: """Pure-Python body of the archive probe — bomb-guard + integrity test. Returns ('ok', None) or ('error', reason). Caught exceptions become clean 'error' rejections; uncaught crashes in the subprocess become non-zero exit codes (poison-pill signature) handled by probe_archive. Exposed at the module level so the subprocess runner and tests both call the same code path. """ path = Path(path_str) try: total, test_bad = _inspect_archive(path) except Exception as exc: # noqa: BLE001 — clean rejection return ("error", f"{type(exc).__name__}: {exc}") if total is not None and total > MAX_ARCHIVE_UNCOMPRESSED_BYTES: gib = total / (1024 ** 3) return ("error", f"uncompressed size {gib:.1f} GiB exceeds the bomb-guard cap") if test_bad is not None: return ("error", f"integrity test failed at member {test_bad!r}") return ("ok", None) def _inspect_archive(path: Path): """Return (total_uncompressed_bytes | None, first_bad_member | None) for the archive. Format detected by extension OR magic bytes (so a mis-named archive is still bomb-guarded + integrity-tested, matching the extractor's gate); raises on a structurally-broken container (caught by the child as a clean rejection).""" from ..services.archive_extractor import detect_archive_format fmt = detect_archive_format(path) if fmt == "zip": import zipfile with zipfile.ZipFile(path) as zf: total = sum(zi.file_size for zi in zf.infolist()) return total, zf.testzip() if fmt == "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 fmt == "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") # Not a recognised archive — nothing to test; treat as clean. return None, None