FabledCurator/backend/app/utils/safe_probe.py

"""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
    # Container duration in seconds (videos only) — the Tier-1 video near-dup
    # key (#871). None when not a video / ffprobe didn't report it.
    duration: float | 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",
                # format=duration is the reliable container duration (stream
                # duration is often N/A for remuxed/streamed mp4) — the #871
                # video near-dup key.
                "-show_entries", "stream=width,height:format=duration",
                "-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:
        parsed = json.loads(out.stdout) or {}
    except json.JSONDecodeError as exc:
        return ProbeResult(ok=False, crashed=False, reason=f"ffprobe output parse failed: {exc}")
    streams = parsed.get("streams") or []
    if not streams:
        return ProbeResult(ok=False, crashed=False, reason="no decodable video stream")
    duration = None
    raw_dur = (parsed.get("format") or {}).get("duration")
    try:
        duration = float(raw_dur) if raw_dur is not None else None
    except (TypeError, ValueError):
        duration = None
    return ProbeResult(
        ok=True, width=streams[0].get("width"), height=streams[0].get("height"),
        duration=duration,
    )


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 _over_cap(total) -> bool:
    return total is not None and total > MAX_ARCHIVE_UNCOMPRESSED_BYTES


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).

    The size total comes from the archive's directory, so an archive over the
    cap is rejected before the integrity test decompresses anything: on a
    multi-gigabyte archive that test alone outlasts the probe's timeout."""
    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())
            if _over_cap(total):
                return total, None
            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())
            if _over_cap(total):
                return total, None
            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)
            if _over_cap(total):
                return total, 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