bvandeusen
5bb25245a5
Patreon attachment downloads land with sanitized URL-blob filenames
(01_https___www.patreon.com_media-u_v3_<id>) whose Path.suffix is junk, never
.zip — so the extension-only is_archive() filed them as opaque PostAttachments
and never extracted them ("No images attached to this post").
Add archive_extractor.detect_archive_format() — extension first, then magic-byte
sniff (zipfile.is_zipfile + RAR/7z signatures). is_archive(), extract_archive(),
and safe_probe._inspect_archive() (the bomb-guard) all route through it, so a
mis-named/extension-less archive is now detected, bomb-guarded, integrity-tested,
AND extracted regardless of filename. Stops new ones; part 2 re-extracts the
already-imported backlog.
Tests: mis-named zip detected + extracted; non-archive dotted name not
misdetected; _inspect_archive on a mis-named zip; signature updated.
Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
79 lines
2.8 KiB
Python
79 lines
2.8 KiB
Python
"""Layer-3 subprocess-isolated probe tests.
|
|
|
|
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 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)
|
|
assert total == 150
|
|
assert bad is None
|
|
|
|
|
|
def test_inspect_archive_detects_misnamed_zip(tmp_path):
|
|
"""A zip with a mangled / extension-less name (Patreon URL-blob attachment)
|
|
is still bomb-guarded + integrity-tested via magic-byte detection."""
|
|
z = tmp_path / "01_https___www.patreon.com_media-u_v3_131083093"
|
|
_zip(z, {"a.txt": b"x" * 100})
|
|
total, bad = safe_probe._inspect_archive(z)
|
|
assert total == 100
|
|
assert bad is None
|
|
|
|
|
|
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
|
|
status, detail = safe_probe._run_probe(str(z))
|
|
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
|