diff --git a/backend/app/downloads/egress.py b/backend/app/downloads/egress.py index a5649c06..472851ba 100644 --- a/backend/app/downloads/egress.py +++ b/backend/app/downloads/egress.py @@ -233,7 +233,7 @@ def open_ydl(opts: dict) -> yt_dlp.YoutubeDL: # assembles, the same every caller handed yt-dlp before. ydl = yt_dlp.YoutubeDL(opts) # type: ignore[arg-type] if opts.get(GUARDED): - ydl.add_post_processor(HttpOnlyFormats(ydl), when="pre_process") + ydl.add_post_processor(HttpOnlyFormats(), when="pre_process") return ydl @@ -250,9 +250,23 @@ def proxy_url() -> str: return _gate.url -# The protocols yt-dlp fetches over http(s) — the only ones the gate sees. Everything else (rtmp, -# rtsp, mms, f4m's own fetcher, ws) would be dialled by ffmpeg or another tool directly. -_HTTP_PROTOCOLS = frozenset({"http", "https", "m3u8_native", "http_dash_segments"}) +# The protocols fetched over http(s) through the proxy option — the only ones the gate sees: yt-dlp's +# own fetchers (http, f4m, ism, the DASH segments and their generator, native HLS — which a bare m3u8 +# also gets, as a guarded download sets hls_prefer_native) and ffmpeg's http input (a live stream or +# a clip), which takes the proxy too. Everything else (rtmp, rtsp, mms, ws) ffmpeg or another tool +# would dial directly. +_HTTP_PROTOCOLS = frozenset( + { + "http", + "https", + "f4m", + "ism", + "m3u8", + "m3u8_native", + "http_dash_segments", + "http_dash_segments_generator", + } +) class HttpOnlyFormats(PostProcessor): @@ -260,6 +274,11 @@ class HttpOnlyFormats(PostProcessor): yt-dlp re-reads its formats after ("The pre-processors may have modified the formats"), so the selector never sees them. With nothing left, yt-dlp reports no formats — a refusal, not a leak.""" + def set_downloader(self, downloader): + # Not the job's postprocessor hooks: they would report this filter as a "processing" step + # before a byte is fetched (yt-dlp attaches them here, once per call). + self._downloader = downloader + def run(self, information): # yt-dlp's stub types the info dict as a TypedDict with optional keys; it is a dict here. info = cast("dict[str, Any]", information) diff --git a/backend/tests/test_egress.py b/backend/tests/test_egress.py index 012dd3d1..19d980bf 100644 --- a/backend/tests/test_egress.py +++ b/backend/tests/test_egress.py @@ -273,6 +273,76 @@ def test_a_format_fetched_past_the_gate_is_never_chosen(tmp_path): ydl.process_ie_result(only_rtmp, download=False) +def test_a_format_yt_dlp_fetches_itself_is_kept(): + # f4m, ism and the DASH generator fetch through yt-dlp's own urlopen; a bare m3u8 goes to the + # native HLS fetcher under hls_prefer_native (set for every guarded download). All ask the gate. + kept = [ + {"format_id": p, "url": f"https://{PUBLIC}/{p}", "protocol": p} + for p in ("f4m", "ism", "m3u8", "m3u8_native", "http_dash_segments_generator", "https") + ] + raw = {"format_id": "rtmp", "url": f"rtmp://{PUBLIC}/live", "protocol": "rtmp"} + _, info = egress.HttpOnlyFormats().run({"formats": [*kept, raw]}) + assert [f["format_id"] for f in info["formats"]] == [f["format_id"] for f in kept] + + +def test_a_single_url_result_past_the_gate_loses_its_url(): + _, info = egress.HttpOnlyFormats().run({"url": f"rtmp://{INTERNAL}/live"}) + assert info["url"] is None + _, info = egress.HttpOnlyFormats().run({"url": f"https://{PUBLIC}/a.mp4"}) + assert info["url"] == f"https://{PUBLIC}/a.mp4" + + +def test_the_format_filter_says_no_phase_of_its_own(tmp_path): + # It is a post-processor only to run before the format choice; the job's postprocessor hook + # must not see it (it painted every guarded job "processing" before a byte was fetched). + calls: list[dict] = [] + opts = _guarded_opts(tmp_path, "http://127.0.0.1:9", postprocessor_hooks=[calls.append]) + with egress.open_ydl(opts) as ydl: + ydl.run_all_pps("pre_process", {"id": "x", "formats": []}) + assert calls == [] + + +def test_an_allowed_tunnel_carries_bytes_both_ways(gate): + # The CONNECT path a site served over https takes: a 200, then the bytes the client sent after + # the head (in the same packet here) reach the far end, and its answer comes back. + far = socket.socket() + far.bind((PUBLIC, 0)) + far.listen() + + def echo(): + conn, _ = far.accept() + with conn: + conn.sendall(b"PONG:" + conn.recv(64)) + + threading.Thread(target=echo, daemon=True).start() + try: + with socket.create_connection(("127.0.0.1", gate.server_address[1]), timeout=5) as s: + s.sendall(f"CONNECT {PUBLIC}:{far.getsockname()[1]} HTTP/1.1\r\n\r\nPING".encode()) + got = b"" + while b"PONG:PING" not in got: + chunk = s.recv(4096) + if not chunk: + break + got += chunk + assert got.startswith(b"HTTP/1.1 200") + assert got.endswith(b"PONG:PING") + finally: + far.close() + + +def test_the_workers_own_gate_refuses_our_network(): + # Every other gate here is built with a test rule; this is the one `proxy_url()` starts. + internal = _Listener() + try: + port = int(egress.proxy_url().rsplit(":", 1)[1]) + with socket.create_connection(("127.0.0.1", port), timeout=5) as s: + s.sendall(f"CONNECT {INTERNAL}:{internal.port} HTTP/1.1\r\n\r\n".encode()) + assert s.recv(4096).startswith(b"HTTP/1.1 403") + assert internal.hits == 0 + finally: + internal.close() + + def _clip_hits(tmp_path: Path, segment: str, proxy: str) -> int: """How many connections the internal listener takes while a clip (`download_ranges`, which yt-dlp hands to ffmpeg even for an HLS format) is cut from a playlist whose one segment is diff --git a/backend/tests/test_facebook_impersonation.py b/backend/tests/test_facebook_impersonation.py index 61cfb150..f1a0971b 100644 --- a/backend/tests/test_facebook_impersonation.py +++ b/backend/tests/test_facebook_impersonation.py @@ -18,7 +18,7 @@ import yt_dlp from yt_dlp.networking.exceptions import TransportError from yt_dlp.networking.impersonate import ImpersonateTarget -from app.downloads import adapters, pipeline +from app.downloads import adapters, egress, pipeline from app.downloads import errors as dlerrors _URL = "https://www.facebook.com/reel/806379462330494/?fs=e&fs=e" @@ -160,6 +160,16 @@ def test_a_cookie_group_post_downloads_the_posts_own_video(ydl, tmp_path): # Never the post link: that is where the feed video comes from. assert web["extracted"] == [watch, watch] assert [o.get("impersonate") for o in seen[1:]] == [None, _CHROME] + # The post page read and both attempts all go through the egress gate. + assert all(o.get(egress.GUARDED) and o["proxy"] == egress.proxy_url() for o in seen), seen + + +def test_a_youtube_download_skips_the_gate(ydl, tmp_path): + seen, outcomes, _ = ydl + outcomes += [{"id": "dQw4w9WgXcQ"}] + yt = adapters.resolve_adapter("youtube", "dQw4w9WgXcQ") + pipeline._extract_with_fallback(1, 1, "https://youtu.be/dQw4w9WgXcQ", {}, tmp_path, yt) + assert seen and not any(o.get(egress.GUARDED) or o.get("proxy") for o in seen) def test_a_group_post_without_a_single_video_fails_saying_so(ydl, tmp_path): diff --git a/backend/tests/test_facebook_titles.py b/backend/tests/test_facebook_titles.py index 145eb320..4bebe1c6 100644 --- a/backend/tests/test_facebook_titles.py +++ b/backend/tests/test_facebook_titles.py @@ -10,7 +10,7 @@ from sqlalchemy import select, update from app.config import settings from app.db import SessionLocal -from app.downloads import adapters, backfill, pipeline +from app.downloads import adapters, backfill, egress, pipeline from app.downloads.models import DownloadJob, MediaAsset _URL = "https://www.facebook.com/reel/1860682468422402/" @@ -61,6 +61,7 @@ def test_a_placeholder_with_nothing_behind_it_takes_the_signed_out_pages_text(mo assert len(seen) == 1 assert seen[0]["skip_download"] is True assert not seen[0].get("cookiefile") + assert seen[0][egress.GUARDED] is True and seen[0]["proxy"] == egress.proxy_url() def test_a_real_title_reads_nothing_more(monkeypatch):