"""Every ffmpeg/ffprobe the downloads package runs over a library file names the demuxers it may use (`edit.INPUT_GUARD`, a downloaded thumbnail's `edit.THUMB_GUARD`). Without the list, ffmpeg picks a demuxer from the file's content and name, and both are the user's for an upload (the remote site's for a thumbnail): a playlist-type demuxer opens whatever local paths its entries name. These tests read the argv each builder hands to ffmpeg — the guard is a flag, so the flag is what is held — and scan the package for an ffmpeg/ffprobe call written without one. Not in scope: `app/plex`, whose ffmpeg reads the admin's own Plex library, not user-supplied bytes. """ import ast import shutil import subprocess from pathlib import Path import pytest from app.downloads import edit, playable DOWNLOADS = Path(__file__).resolve().parents[1] / "app" / "downloads" # The demuxers that read paths named inside a file. `concat` is allowed only where the list is ours. PLAYLIST_DEMUXERS = {"hls", "dash", "concat", "image2", "ffconcat", "tee"} MEDIA = set(edit.MEDIA_DEMUXERS) def _whitelists(cmd: list[str]) -> list[set[str] | None]: """For each `-i` in `cmd`, the demuxers the `-format_whitelist` given since the previous input allows — None when that input has none.""" out, current = [], None for i, tok in enumerate(cmd): if tok == "-format_whitelist": current = set(cmd[i + 1].split(",")) elif tok == "-i": out.append(current) current = None return out def _assert_media_guarded(cmd: list[str], extra: set[str] = frozenset()) -> None: lists = _whitelists(cmd) assert lists, cmd # a builder that opens no input is not what this test is about for allowed in lists: assert allowed is not None, f"an input with no -format_whitelist: {cmd}" assert allowed <= MEDIA | extra, allowed - MEDIA - extra assert not (allowed - extra) & PLAYLIST_DEMUXERS def test_the_media_list_names_no_demuxer_that_reads_other_paths(): assert not MEDIA & PLAYLIST_DEMUXERS assert edit.INPUT_GUARD == ["-format_whitelist", ",".join(edit.MEDIA_DEMUXERS)] def test_the_phone_safe_conversion_is_guarded(): p = playable.Plan("mp4", True, True, True, True) _assert_media_guarded(playable.build_cmd(Path("in.mov"), Path("out.mp4"), p)) @pytest.mark.parametrize( "spec", [ {"trim": {"start_s": 1, "end_s": 3}, "accurate": False}, {"trim": {"start_s": 1, "end_s": 3}, "accurate": True}, {"crop": {"x": 0, "y": 0, "w": 101, "h": 51}}, ], ids=["fast trim", "accurate trim", "crop"], ) def test_a_clip_is_guarded(spec): spec = edit.normalize_edit_spec(spec) _assert_media_guarded(edit.build_edit_cmd(Path("in.mp4"), Path("out.mp4"), spec, "mp4")) @pytest.mark.parametrize("accurate", [True, False], ids=["accurate join", "fast join"]) def test_a_join_is_guarded(accurate, tmp_path): segs = [{"start_s": 0, "end_s": 1}, {"start_s": 2, "end_s": 3}] spec = edit.normalize_edit_spec({"segments": segs, "accurate": accurate}) cmd, _ = edit.build_concat_plan(Path("in.mp4"), Path("out.mp4"), spec, "mp4", tmp_path) # The fast join reads a concat list WE write; the whitelist rides down to the file it names. _assert_media_guarded(cmd, extra=set() if accurate else {"concat"}) def test_the_poster_and_the_filmstrip_are_guarded(): _assert_media_guarded(edit.build_poster_cmd(Path("in.mp4"), Path("p.jpg"), 2.0)) geom = edit.storyboard_geometry(60) _assert_media_guarded(edit.build_storyboard_cmd(Path("in.mp4"), Path("s.jpg"), 60, geom)) def test_a_downloaded_thumbnail_opens_as_a_single_image_only(): (allowed,) = _whitelists(edit.build_thumbnail_jpg_cmd(Path("t.webp"), Path("t.jpg"))) assert allowed == {"png_pipe", "jpeg_pipe", "webp_pipe"} def test_every_ffprobe_is_guarded(monkeypatch): seen = [] class _Done: stdout = "{}" def _run(args, **_k): seen.append(args) return _Done() monkeypatch.setattr(edit.subprocess, "run", _run) edit.run_ffprobe_json(Path("x.mp4"), show_streams=True) edit.probe_duration(Path("x.mp4")) assert len(seen) == 2 for args in seen: i = args.index("-format_whitelist") assert set(args[i + 1].split(",")) == MEDIA def _unguarded_calls(source: str) -> list[int]: """Lines of a list literal that starts an ffmpeg/ffprobe command or hands one an input (`-i`) and carries no `*…GUARD` / `*guard` splat and no "-format_whitelist" of its own. A builder that adds its input in a later `+=` must guard that list — that is where the input is.""" bad = [] for node in ast.walk(ast.parse(source)): if not isinstance(node, ast.List): continue consts = {e.value for e in node.elts if isinstance(e, ast.Constant)} if not consts & {"-i", "ffprobe"}: continue guarded = "-format_whitelist" in consts or any( isinstance(e, ast.Starred) and isinstance(v := e.value, ast.Name | ast.Attribute) and (v.id if isinstance(v, ast.Name) else v.attr).upper().endswith("GUARD") for e in node.elts ) if not guarded: bad.append(node.lineno) return bad def test_no_ffmpeg_call_in_the_package_opens_an_input_unguarded(): found = { f"{p.name}:{line}" for p in sorted(DOWNLOADS.rglob("*.py")) for line in _unguarded_calls(p.read_text(encoding="utf-8")) } assert found == set() def test_the_scan_sees_an_unguarded_call(): # The scan above is green over nothing if it cannot see one: both shapes it must catch. assert _unguarded_calls('cmd = ["ffmpeg", "-y", "-i", src]') == [1] assert _unguarded_calls('a = ["ffprobe", "-v", "error", str(p)]') == [1] assert _unguarded_calls('cmd = [*INPUT_GUARD, "-i", src]') == [] @pytest.mark.skipif(shutil.which("ffmpeg") is None, reason="ffmpeg not available") @pytest.mark.parametrize("ext", ["png", "webp", "jpg"]) def test_the_thumbnail_guard_still_opens_every_thumbnail_a_download_keeps(ext, tmp_path): # `pipeline._THUMB_SUFFIXES`: what the guard must still let through (a .jpg that is really a # PNG goes the ffmpeg route too — `thumbnail_to_jpg` copies only true JPEG bytes). src = tmp_path / f"t.{ext}" fmt = ["-f", "image2", "-c:v", "png"] if ext == "jpg" else [] subprocess.run( ["ffmpeg", "-y", "-loglevel", "error", "-f", "lavfi", "-i", "testsrc2=size=64x48", "-frames:v", "1", *fmt, str(src)], check=True, ) # fmt: skip out = edit.thumbnail_to_jpg(src, tmp_path / "poster.jpg") assert out is not None and edit._is_jpeg(out)