diff --git a/backend/app/utils/_archive_probe_runner.py b/backend/app/utils/_archive_probe_runner.py new file mode 100644 index 0000000..f4b0c88 --- /dev/null +++ b/backend/app/utils/_archive_probe_runner.py @@ -0,0 +1,33 @@ +"""Subprocess entrypoint for safe_probe.probe_archive — see safe_probe.py. + +probe_archive spawns this via subprocess (not multiprocessing.Process) +because Celery's prefork worker pool runs tasks in DAEMON processes and +Python's multiprocessing forbids daemon processes from spawning children +("AssertionError: daemonic processes are not allowed to have children", +operator-flagged 2026-05-30 — every archive import failed at task +startup). subprocess has no such restriction; we still get crash- +isolation because a probe segfault/OOM exits non-zero rather than +killing the worker. + +Prints a single JSON line on stdout: {"status": "ok"|"error", +"detail": "..."?}. Exit code 0 for clean outcomes; non-zero exit +(signal / OOM-kill / unhandled exception) is the poison-pill signature +the parent maps to ProbeResult(crashed=True). +""" +import json +import sys + +from .safe_probe import _run_probe + + +def main() -> int: + if len(sys.argv) != 2: + print(json.dumps({"status": "error", "detail": "usage: "})) + return 2 + status, detail = _run_probe(sys.argv[1]) + print(json.dumps({"status": status, "detail": detail})) + return 0 + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/backend/app/utils/safe_probe.py b/backend/app/utils/safe_probe.py index 02c387c..d8e5360 100644 --- a/backend/app/utils/safe_probe.py +++ b/backend/app/utils/safe_probe.py @@ -28,8 +28,8 @@ Operator-requested 2026-05-28 (Layer 3). """ import json -import multiprocessing as mp import subprocess +import sys from dataclasses import dataclass from pathlib import Path @@ -40,6 +40,13 @@ ARCHIVE_PROBE_TIMEOUT_SECONDS = 120 # 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: @@ -91,54 +98,68 @@ def probe_video(path: Path, *, timeout: float = VIDEO_PROBE_TIMEOUT_SECONDS) -> 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) + """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 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. + 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 {proc.exitcode})", + reason=f"archive probe crashed (exit {result.returncode})", ) + last_line = result.stdout.strip().splitlines()[-1:] or [""] 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": + 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=detail) + return ProbeResult( + ok=False, crashed=False, + reason=outcome.get("detail") or "archive probe rejected", + ) -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.""" +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) 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 + return ("error", f"{type(exc).__name__}: {exc}") 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 + return ("error", f"uncompressed size {gib:.1f} GiB exceeds the bomb-guard cap") if test_bad is not None: - q.put(("error", f"integrity test failed at member {test_bad!r}")) - return - q.put(("ok", None)) + return ("error", f"integrity test failed at member {test_bad!r}") + return ("ok", None) def _inspect_archive(path: Path, ext: str): diff --git a/tests/test_safe_probe.py b/tests/test_safe_probe.py index 6dd7ea2..98a67b3 100644 --- a/tests/test_safe_probe.py +++ b/tests/test_safe_probe.py @@ -1,12 +1,11 @@ """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. +The bomb-guard cap is exercised against `_run_probe` directly (in-process, +where a monkeypatch on the module constant takes effect) — the real probe +runs in a subprocess that re-imports the module, so a parent-process +monkeypatch wouldn't reach the spawned interpreter. """ -import multiprocessing as mp import zipfile from pathlib import Path @@ -46,16 +45,14 @@ def test_inspect_archive_reports_size_and_clean_integrity(tmp_path): 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.""" +def test_run_probe_bomb_guard(tmp_path, monkeypatch): + """In-process call to _run_probe so the monkeypatched cap takes effect. + The public probe_archive runs in a subprocess which re-imports the + module and wouldn't see the patched constant.""" 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) + status, detail = safe_probe._run_probe(str(z)) assert status == "error" assert "bomb-guard cap" in detail