diff --git a/backend/.basedpyright/baseline.json b/backend/.basedpyright/baseline.json
index 8af825b4..af71727c 100644
--- a/backend/.basedpyright/baseline.json
+++ b/backend/.basedpyright/baseline.json
@@ -364,25 +364,7 @@
                 }
             }
         ],
-        "./app/downloads/adapters.py": [
-            {
-                "code": "reportArgumentType",
-                "range": {
-                    "startColumn": 30,
-                    "endColumn": 38,
-                    "lineCount": 1
-                }
-            }
-        ],
         "./app/downloads/backfill.py": [
-            {
-                "code": "reportArgumentType",
-                "range": {
-                    "startColumn": 34,
-                    "endColumn": 38,
-                    "lineCount": 1
-                }
-            },
             {
                 "code": "reportReturnType",
                 "range": {
@@ -413,14 +395,6 @@
             }
         ],
         "./app/downloads/pipeline.py": [
-            {
-                "code": "reportArgumentType",
-                "range": {
-                    "startColumn": 34,
-                    "endColumn": 38,
-                    "lineCount": 1
-                }
-            },
             {
                 "code": "reportAttributeAccessIssue",
                 "range": {
diff --git a/backend/app/downloads/adapters.py b/backend/app/downloads/adapters.py
index ee693f6c..1a932fb3 100644
--- a/backend/app/downloads/adapters.py
+++ b/backend/app/downloads/adapters.py
@@ -14,10 +14,10 @@ Resolution works both before extraction (from source_kind / URL host) and after
 import re
 from urllib.parse import urlparse
 
-import yt_dlp
 from yt_dlp.networking.impersonate import ImpersonateTarget
 
 from app.config import settings
+from app.downloads import egress
 from app.downloads.stated import StatedFailure
 
 # Facebook's "Cannot parse data": the page yt-dlp got was a wall, not the content. Both the Chrome retry
@@ -33,6 +33,9 @@ class SourceAdapter:
     """Generic source: yt-dlp handles everything and the ref is already a fetchable URL."""
 
     key = "generic"
+    # SEC-17: every fetch of this source's download goes through the egress gate (`egress`). A
+    # source's own host list is no guarantee — its pages, redirects and formats can point anywhere.
+    egress_guarded = True
 
     # --- extraction (worker._extract_with_fallback / backfill) ---------------------------------
     def resolve_url(self, url: str, ydl_opts: dict) -> str:
@@ -92,6 +95,9 @@ class SourceAdapter:
 
 class YouTubeAdapter(SourceAdapter):
     key = "youtube"
+    # Not through the gate: YouTube's fetches go to Google's fixed hosts, and the PO-token provider
+    # it needs is the sidecar on our internal network, which the gate exists to refuse.
+    egress_guarded = False
     # Errors worth retrying with a different player-client set (YouTube per-client flakiness).
     _RETRIABLE = (
         "not a bot",
@@ -179,7 +185,7 @@ class FacebookAdapter(SourceAdapter):
         post_id = facebook_group_post_id(url)
         if post_id is None or not ydl_opts.get("cookiefile"):
             return url
-        with yt_dlp.YoutubeDL(ydl_opts) as ydl:
+        with egress.open_ydl(ydl_opts) as ydl:
             raw = ydl.urlopen(url).read(self._POST_PAGE_MAX_BYTES)
         page = raw.decode("utf-8", "replace")
         # Measured: a refused or expired cookie gets the login page (~450 KB, a login
diff --git a/backend/app/downloads/backfill.py b/backend/app/downloads/backfill.py
index a79ed4e2..ca0cc804 100644
--- a/backend/app/downloads/backfill.py
+++ b/backend/app/downloads/backfill.py
@@ -16,7 +16,6 @@ Run once per environment:
 
 from urllib.parse import urlsplit
 
-import yt_dlp
 from sqlalchemy import select, update
 from sqlalchemy.orm import Session
 
@@ -25,7 +24,7 @@ from app.config import settings
 from app.db import SessionLocal
 from app.downloads import adapters
 from app.downloads import edit as editmod
-from app.downloads import formats, service
+from app.downloads import egress, formats, service
 from app.downloads.models import DownloadJob, MediaAsset
 from app.models import DISPLAY_TEXT_MAX
 from app.titles import normalize_title
@@ -43,11 +42,15 @@ def extract_meta(url: str, adapter: adapters.SourceAdapter, asset_id: int) -> di
 def _extract_meta_scoped(url: str, adapter: adapters.SourceAdapter, asset_id: int) -> dict | None:
     for clients in adapter.client_sets():
         opts = formats.build_ydl_opts(
-            {}, "%(id)s.%(ext)s", lambda d: None, extra=adapter.ydl_opts_extra(clients)
+            {},
+            "%(id)s.%(ext)s",
+            lambda d: None,
+            extra=adapter.ydl_opts_extra(clients),
+            egress=adapter.egress_guarded,
         )
         opts["skip_download"] = True
         try:
-            with yt_dlp.YoutubeDL(opts) as ydl:
+            with egress.open_ydl(opts) as ydl:
                 info = ydl.extract_info(url, download=False)
             if info:
                 return info
diff --git a/backend/app/downloads/egress.py b/backend/app/downloads/egress.py
new file mode 100644
index 00000000..991e8ce8
--- /dev/null
+++ b/backend/app/downloads/egress.py
@@ -0,0 +1,264 @@
+"""The egress gate: every fetch of a user-supplied download goes through here, and this is where the
+address is judged — at the moment of the connection, not when the URL was typed (SEC-17).
+
+`urlguard` judges the URL the user typed, at enqueue and again in the worker. What that cannot see is
+every request yt-dlp makes after it: a public host answering 3xx with an internal `Location`, a page
+whose formats point inside, a name that resolves public for the check and private for the fetch (DNS
+rebinding). So a guarded download runs with yt-dlp's `proxy` set to this gate, a small HTTP proxy on
+the worker's loopback. For each request it resolves the host ONCE, refuses the connection if any
+address the name has is one the server must not fetch from (`urlguard.is_forbidden`), and otherwise
+connects to the very address it vetted — there is no second lookup for a rebinding answer to slip into.
+
+Measured (yt-dlp 2026.8.19, 2026-11-06) on the worker image, with a logging proxy: the `proxy` option
+reaches every path a download takes — the Requests handler, the CurlCFFI one an impersonated request
+uses, the native HLS downloader and an ffmpeg HLS download (segments included). ffmpeg itself also
+speaks rtmp, rtsp, ftp, tcp and more, which no HTTP proxy covers, so a guarded download keeps only the
+formats it fetches over http(s) (`HttpOnlyFormats`, added by `open_ydl`) and fetches HLS natively.
+
+A refused connection is answered `403` before anything is dialled, so a live internal port and a dead
+one read the same; `errors.user_error_text` then stores the one redacted sentence for both.
+
+Not used for YouTube: those downloads talk to Google's fixed hosts and to the PO-token sidecar, which
+lives on the internal network on purpose (`adapters.YouTubeAdapter.egress_guarded`).
+"""
+
+import ipaddress
+import select
+import socket
+import socketserver
+import threading
+from collections.abc import Callable
+from typing import Any, cast
+from urllib.parse import urlsplit
+
+import yt_dlp
+from yt_dlp.postprocessor.common import PostProcessor
+
+from app import logs
+from app.downloads import urlguard
+
+log = logs.get_logger(__name__)
+
+Resolver = Callable[..., list]
+Forbidden = Callable[[ipaddress.IPv4Address | ipaddress.IPv6Address], bool]
+
+_BLOCKED = b"HTTP/1.1 403 Forbidden\r\nContent-Length: 0\r\nConnection: close\r\n\r\n"
+_BAD_GATEWAY = b"HTTP/1.1 502 Bad Gateway\r\nContent-Length: 0\r\nConnection: close\r\n\r\n"
+_BAD_REQUEST = b"HTTP/1.1 400 Bad Request\r\nContent-Length: 0\r\nConnection: close\r\n\r\n"
+_MAX_HEAD = 64 * 1024
+# A tunnel with no traffic either way for this long is closed. Above yt-dlp's own socket_timeout
+# (30 s, `formats.build_ydl_opts`), so the gate never ends a transfer yt-dlp would still wait for.
+_IDLE_S = 120
+# Hop-by-hop headers a proxy must not pass on; `Connection: close` replaces them, so one connection
+# carries one request and a reused client connection can never be steered to a second host unvetted.
+_HOP = (b"proxy-", b"connection:", b"keep-alive:")
+
+
+def vetted_address(
+    host: str, port: int, *, resolve: Resolver = socket.getaddrinfo, forbidden: Forbidden
+) -> tuple[str, int] | None:
+    """The address to connect to for `host:port`, or None when it must not be fetched: the name has
+    no answer, or ANY of its addresses is forbidden (a round-robin answer is not a coin flip). IPv4
+    only, as every download is (`source_address` 0.0.0.0 in `formats.build_ydl_opts`)."""
+    try:
+        infos = resolve(host, port, socket.AF_INET, socket.SOCK_STREAM)
+    except (socket.gaierror, UnicodeError, OSError):
+        # silent-ok: "no answer" is a refusal, and the caller's refusal is the line that is logged.
+        return None
+    addrs = [info[4] for info in infos]
+    if not addrs:
+        return None
+    for addr in addrs:
+        try:
+            ip = ipaddress.ip_address(addr[0])
+        except ValueError:
+            # silent-ok: an answer that is not an address is refused like a forbidden one, and the
+            # caller logs the refusal.
+            return None
+        if forbidden(ip):
+            return None
+    return addrs[0][0], addrs[0][1]
+
+
+def _read_head(sock: socket.socket) -> bytes | None:
+    buf = b""
+    while b"\r\n\r\n" not in buf:
+        if len(buf) > _MAX_HEAD:
+            return None
+        chunk = sock.recv(8192)
+        if not chunk:
+            return None
+        buf += chunk
+    return buf
+
+
+def _pipe(a: socket.socket, b: socket.socket) -> None:
+    socks = [a, b]
+    while True:
+        ready, _, _ = select.select(socks, [], [], _IDLE_S)
+        if not ready:
+            return
+        for s in ready:
+            data = s.recv(65536)
+            if not data:
+                return
+            (b if s is a else a).sendall(data)
+
+
+class _Handler(socketserver.BaseRequestHandler):
+    def handle(self) -> None:
+        client: socket.socket = self.request
+        head = _read_head(client)
+        if head is None:
+            return
+        head_part, _, body = head.partition(b"\r\n\r\n")
+        lines = head_part.split(b"\r\n")
+        try:
+            method, target, version = lines[0].decode("latin1").split(" ", 2)
+        except ValueError:
+            # silent-ok: the only client is yt-dlp on this loopback; a malformed request is answered
+            # 400 and reaches its own error, which the job records.
+            client.sendall(_BAD_REQUEST)
+            return
+        if method == "CONNECT":
+            host, _, port_s = target.rpartition(":")
+            host = host.strip("[]")
+            upstream = self._dial(host, port_s)
+            if upstream is None:
+                return
+            with upstream:
+                client.sendall(b"HTTP/1.1 200 Connection established\r\n\r\n")
+                if body:
+                    upstream.sendall(body)
+                _pipe(client, upstream)
+            return
+        u = urlsplit(target)
+        if u.scheme != "http" or not u.hostname:
+            client.sendall(_BLOCKED)
+            return
+        try:
+            port = u.port or 80
+        except ValueError:
+            # silent-ok: as for a malformed request line above — answered 400, recorded by the job.
+            client.sendall(_BAD_REQUEST)
+            return
+        upstream = self._dial(u.hostname, str(port))
+        if upstream is None:
+            return
+        path = (u.path or "/") + (f"?{u.query}" if u.query else "")
+        headers = [h for h in lines[1:] if not h.lower().startswith(_HOP)]
+        request = b"\r\n".join(
+            [f"{method} {path} {version}".encode("latin1"), *headers, b"Connection: close"]
+        )
+        with upstream:
+            upstream.sendall(request + b"\r\n\r\n" + body)
+            _pipe(client, upstream)
+
+    def _dial(self, host: str, port_s: str) -> socket.socket | None:
+        """Connect to the address `host` was vetted at, or answer the client and return None."""
+        client: socket.socket = self.request
+        try:
+            port = int(port_s)
+        except ValueError:
+            # silent-ok: as for a malformed request line above — answered 400, recorded by the job.
+            client.sendall(_BAD_REQUEST)
+            return None
+        gate = cast("_Gate", self.server)
+        addr = vetted_address(host, port, resolve=gate.resolve, forbidden=gate.forbidden)
+        if addr is None:
+            log.warning(
+                "egress refused a connection to %s:%s",
+                host,  # log-pii-ok: the host a download tried to reach, never its path or query
+                port,
+                extra=logs.event("download.egress.refused", port=port),
+            )
+            client.sendall(_BLOCKED)
+            return None
+        try:
+            return socket.create_connection(addr, timeout=_IDLE_S)
+        except OSError:
+            # silent-ok: a public host that does not answer; yt-dlp reports it as the job's own
+            # network failure, which the job records.
+            client.sendall(_BAD_GATEWAY)
+            return None
+
+
+class _Gate(socketserver.ThreadingTCPServer):
+    daemon_threads = True
+    allow_reuse_address = True
+
+    def __init__(self, resolve: Resolver, forbidden: Forbidden):
+        self.resolve = resolve
+        self.forbidden = forbidden
+        super().__init__(("127.0.0.1", 0), _Handler)
+
+    @property
+    def url(self) -> str:
+        return f"http://127.0.0.1:{self.server_address[1]}"
+
+
+def start_gate(
+    *, resolve: Resolver = socket.getaddrinfo, forbidden: Forbidden = urlguard.is_forbidden
+) -> _Gate:
+    """A gate serving on its own thread. The worker uses the one `proxy_url` starts; a test starts
+    its own, with a resolver and a judgement it controls."""
+    gate = _Gate(resolve, forbidden)
+    threading.Thread(target=gate.serve_forever, name="egress-gate", daemon=True).start()
+    return gate
+
+
+# The options key `formats.build_ydl_opts` marks a guarded download with; `open_ydl` reads it. yt-dlp
+# keeps unknown keys in its params and never acts on them.
+GUARDED = "siftlode_egress_guarded"
+
+
+def open_ydl(opts: dict) -> yt_dlp.YoutubeDL:
+    """The one way this package opens yt-dlp (a contract test holds it): a guarded download also
+    gets `HttpOnlyFormats`, which yt-dlp takes only as an instance, not through the options dict."""
+    # yt-dlp's stub types its params as a TypedDict; these are the plain dict `formats.build_ydl_opts`
+    # 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")
+    return ydl
+
+
+_gate: _Gate | None = None
+_gate_lock = threading.Lock()
+
+
+def proxy_url() -> str:
+    """This process's gate, started on first use (the worker's jobs share it)."""
+    global _gate
+    with _gate_lock:
+        if _gate is None:
+            _gate = start_gate()
+        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"})
+
+
+class HttpOnlyFormats(PostProcessor):
+    """Drops every format a guarded download would fetch past the gate. Runs as `pre_process`, which
+    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 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)
+        if info.get("formats"):
+            info["formats"] = [f for f in info["formats"] if _over_http(f)]
+        elif info.get("url") and not _over_http(info):
+            info["url"] = None
+        return [], information
+
+
+def _over_http(fmt: dict) -> bool:
+    url = fmt.get("manifest_url") or fmt.get("url") or ""
+    if urlsplit(url).scheme not in ("http", "https"):
+        return False
+    protocol = fmt.get("protocol") or urlsplit(fmt.get("url") or "").scheme
+    return all(p in _HTTP_PROTOCOLS for p in protocol.split("+"))
diff --git a/backend/app/downloads/formats.py b/backend/app/downloads/formats.py
index 959753c9..95066371 100644
--- a/backend/app/downloads/formats.py
+++ b/backend/app/downloads/formats.py
@@ -28,6 +28,7 @@ import urllib.parse
 from collections.abc import Iterator
 
 from app import logs
+from app.downloads import egress as egress_gate
 
 log = logs.get_logger(__name__)
 
@@ -294,6 +295,8 @@ def build_ydl_opts(
     extra: dict | None = None,
     cookiefile: str | None = None,
     max_bytes: int | None = None,
+    *,
+    egress: bool,
 ) -> dict:
     """Assemble yt-dlp options for a single download to `outtmpl` (a full path template).
 
@@ -309,7 +312,11 @@ def build_ydl_opts(
     a file announced larger than it rather than filling the disk and only then being caught by the
     quota accounting. Its ffmpeg downloader (a `download_ranges` fetch) does not read it, nor does
     a fragmented download (YouTube's formats: measured, 718 MB fetched with 1 MB left) — so the
-    pipeline also holds every produced file to what is left by its real size."""
+    pipeline also holds every produced file to what is left by its real size.
+
+    `egress` (SEC-17) routes every fetch through the worker's egress gate, which refuses a connection
+    into our own network, and keeps only the formats fetched over http(s). Required, with no default:
+    a new caller has to say which kind of download it builds (`SourceAdapter.egress_guarded`)."""
     s = normalize(spec)
     h = s["max_height"]
     opts: dict = {
@@ -364,6 +371,13 @@ def build_ydl_opts(
     if cookiefile:
         opts["cookiefile"] = cookiefile
 
+    # SEC-17. AFTER the adapter merge, like the cookie: an adapter must not be able to point a
+    # guarded download past the gate.
+    if egress:
+        opts["proxy"] = egress_gate.proxy_url()
+        opts["hls_prefer_native"] = True
+        opts[egress_gate.GUARDED] = True
+
     if s["mode"] == "a":
         opts["format"] = "bestaudio/best"
         opts["postprocessors"].append(
diff --git a/backend/app/downloads/pipeline.py b/backend/app/downloads/pipeline.py
index 236b162d..cecda260 100644
--- a/backend/app/downloads/pipeline.py
+++ b/backend/app/downloads/pipeline.py
@@ -17,7 +17,7 @@ from app.downloads import clip as clipmod
 from app.downloads import edit as editmod
 from app.downloads import errors as dlerrors
 from app.downloads import adapters, credentials, formats, mediakind, playable, probe, quota
-from app.downloads import backfill
+from app.downloads import backfill, egress
 from app.downloads import service, storage
 from app.downloads import urlguard
 from app.downloads.jobdb import (
@@ -394,6 +394,7 @@ def _extract_with_fallback(
             _make_progress_hook(job_id, asset_id),
             cookiefile=cookiefile,
             max_bytes=max_bytes,
+            egress=adapter.egress_guarded,
         ),
     )
     last_exc: Exception | None = None
@@ -405,6 +406,7 @@ def _extract_with_fallback(
             extra=adapter.ydl_opts_extra(clients),
             cookiefile=cookiefile,
             max_bytes=max_bytes,
+            egress=adapter.egress_guarded,
         )
         opts["postprocessor_hooks"] = [_make_pp_hook(job_id)]
         if section is not None:
@@ -415,7 +417,7 @@ def _extract_with_fallback(
         # (Per-ATTEMPT only; a single extract_info that itself runs past the stall timeout is still
         # killed — acceptable, as socket_timeout + extractor_retries bound one attempt well under it.)
         try:
-            with yt_dlp.YoutubeDL(opts) as ydl:
+            with egress.open_ydl(opts) as ydl:
                 return ydl.extract_info(target, download=True)
         except yt_dlp.utils.DownloadError as exc:
             last_exc = exc
diff --git a/backend/app/downloads/urlguard.py b/backend/app/downloads/urlguard.py
index 854aa732..69e86e8a 100644
--- a/backend/app/downloads/urlguard.py
+++ b/backend/app/downloads/urlguard.py
@@ -10,22 +10,12 @@ What this closes: a URL whose host resolves to a loopback / private / link-local
 is refused, at enqueue AND again in the worker before extraction. Both, because DNS can answer
 differently between the two moments, and the worker is where the socket is actually opened.
 
-What it does NOT close, stated plainly rather than implied, because both need the same fix — gating
-the connection itself with a custom socket factory inside yt-dlp, a much larger change than this:
-
-* DNS rebinding: a host that passes the worker's check and then answers yt-dlp's own resolution
-  with an internal address.
-* HTTP redirects: a public host that answers 3xx with an internal ``Location``. yt-dlp follows it,
-  and only the URL the user TYPED was ever judged here. An attacker-controlled host or any public
-  open redirect reaches it, so this is the cheaper of the two. It does NOT usually yield the
-  response body — a non-media body fails the mp4 remux and stores nothing — but it does hand back
-  the readout above: the redaction in ``errors.user_error_text`` fires only for the ``network``
-  class, so a dead internal port (refused → ``network`` → redacted) is distinguishable from a live
-  one (an HTTP answer that is not media → ``unsupported_host``/``format_unavailable``, raw text
-  kept). Tracked as SEC-17.
-
-The two checks here stop the straightforward case (typing ``http://127.0.0.1:6379`` or an internal
-hostname into the box), which is the reachable one.
+What this alone cannot see is every request yt-dlp makes AFTER the typed URL: a public host
+answering 3xx with an internal ``Location``, and DNS rebinding (a name that passes this check and
+then resolves inside for yt-dlp's own lookup). Those are closed one layer down, by `egress`: a guarded
+download's every connection goes through a gate that judges the address it actually dials, with
+this module's `is_forbidden`. The checks here stay as the early, structured refusal — a user typing
+``http://127.0.0.1:6379`` hears why at once instead of from a failed job.
 
 Not used for YouTube jobs: those resolve to a fixed public host from an 11-char id, so there is no
 user-controlled destination to guard.
@@ -62,7 +52,7 @@ class BlockedUrl(Exception):
         self.reason: UrlBlockedReason = reason
 
 
-def _is_forbidden(ip: ipaddress.IPv4Address | ipaddress.IPv6Address) -> bool:
+def is_forbidden(ip: ipaddress.IPv4Address | ipaddress.IPv6Address) -> bool:
     """Every address family we must not open a connection to. Checked explicitly rather than
     leaning on ``is_private`` alone, whose coverage differs between IPv4 and IPv6."""
     # ::ffff:127.0.0.1 is loopback wearing an IPv6 hat — unwrap before judging it.
@@ -113,5 +103,5 @@ def check_url(raw: str) -> None:
             ip = ipaddress.ip_address(addr)
         except ValueError:  # getaddrinfo returned something unparseable; treat as unusable
             raise BlockedUrl(REASON_PRIVATE) from None
-        if _is_forbidden(ip):
+        if is_forbidden(ip):
             raise BlockedUrl(REASON_PRIVATE)
diff --git a/backend/tests/test_clip.py b/backend/tests/test_clip.py
index 9f5a10bf..0c74189e 100644
--- a/backend/tests/test_clip.py
+++ b/backend/tests/test_clip.py
@@ -531,6 +531,9 @@ def test_a_youtube_clip_hands_the_range_to_yt_dlp(monkeypatch, tmp_path):
         def __enter__(self):
             return self
 
+        def add_post_processor(self, pp, when="post_process"):
+            pass  # the egress format filter; nothing here to filter
+
         def __exit__(self, *_a):
             return False
 
diff --git a/backend/tests/test_egress.py b/backend/tests/test_egress.py
new file mode 100644
index 00000000..4ce83b2d
--- /dev/null
+++ b/backend/tests/test_egress.py
@@ -0,0 +1,322 @@
+"""SEC-17: the egress gate judges the address a guarded download actually dials.
+
+Two loopback addresses stand in for the two sides: 127.0.0.2 plays a public host, 127.0.0.1 our own
+network. The gate under test refuses exactly 127.0.0.1, so every assertion is about the judgement and
+the wiring, with real sockets, a real yt-dlp and a real ffmpeg — no mocked transport.
+"""
+
+import contextlib
+import http.server
+import ipaddress
+import socket
+import socketserver
+import threading
+from pathlib import Path
+
+import pytest
+import yt_dlp
+from yt_dlp.utils import download_range_func
+
+from app.downloads import adapters, egress, errors, formats
+
+PUBLIC = "127.0.0.2"
+INTERNAL = "127.0.0.1"
+
+
+def _forbid_internal(ip: ipaddress.IPv4Address | ipaddress.IPv6Address) -> bool:
+    return str(ip) == INTERNAL
+
+
+@pytest.fixture
+def gate():
+    g = egress.start_gate(forbidden=_forbid_internal)
+    yield g
+    g.shutdown()
+    g.server_close()
+
+
+class _Listener:
+    """A TCP listener that only counts who connected — the internal service under attack."""
+
+    def __init__(self, host: str = INTERNAL, port: int = 0):
+        self.hits = 0
+        self.sock = socket.socket()
+        self.sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
+        self.sock.bind((host, port))
+        self.sock.listen()
+        self.port = self.sock.getsockname()[1]
+        threading.Thread(target=self._serve, daemon=True).start()
+
+    def _serve(self):
+        while True:
+            try:
+                conn, _ = self.sock.accept()
+            except OSError:
+                return
+            self.hits += 1
+            with contextlib.suppress(OSError):
+                conn.sendall(b"HTTP/1.1 200 OK\r\nContent-Length: 2\r\n\r\nok")
+            conn.close()
+
+    def close(self):
+        self.sock.close()
+
+
+def _public_server(routes: dict, port: int = 0, directory: Path | None = None):
+    """An HTTP server on the 'public' address: `routes` maps a path to (status, headers); anything
+    else is served from `directory`."""
+
+    class H(http.server.SimpleHTTPRequestHandler):
+        def __init__(self, *a, **k):
+            super().__init__(*a, directory=str(directory or "."), **k)
+
+        def log_message(self, *a):
+            pass
+
+        def do_GET(self):
+            if self.path in routes:
+                status, headers, body = routes[self.path]
+                self.send_response(status)
+                for k, v in headers.items():
+                    self.send_header(k, v)
+                self.send_header("Content-Length", str(len(body)))
+                self.end_headers()
+                self.wfile.write(body)
+                return
+            super().do_GET()
+
+    srv = socketserver.ThreadingTCPServer((PUBLIC, port), H)
+    srv.daemon_threads = True
+    threading.Thread(target=srv.serve_forever, daemon=True).start()
+    return srv
+
+
+def _guarded_opts(tmp_path: Path, gate_url: str, **extra) -> dict:
+    opts = formats.build_ydl_opts({}, str(tmp_path / "%(id)s.%(ext)s"), lambda d: None, egress=True)
+    opts.update(retries=0, fragment_retries=0, extractor_retries=0, socket_timeout=5)
+    opts["proxy"] = gate_url  # the test's gate (it lets 127.0.0.2 through), not the worker's
+    opts.update(extra)
+    return opts
+
+
+def _download_error(opts: dict, url: str) -> str:
+    with pytest.raises(yt_dlp.utils.DownloadError) as exc, egress.open_ydl(opts) as ydl:
+        ydl.extract_info(url, download=True)
+    return str(exc.value)
+
+
+# --- the judgement ---------------------------------------------------------------------------------
+
+
+def test_a_name_with_any_forbidden_answer_is_refused():
+    def resolve(answers):
+        return lambda host, port, *a: [(2, 1, 6, "", (ip, port)) for ip in answers]
+
+    assert egress.vetted_address(
+        "x", 80, resolve=resolve([PUBLIC]), forbidden=_forbid_internal
+    ) == (
+        PUBLIC,
+        80,
+    )
+    assert (
+        egress.vetted_address(
+            "x", 80, resolve=resolve([PUBLIC, INTERNAL]), forbidden=_forbid_internal
+        )
+        is None
+    )
+
+    def no_answer(*a):
+        raise socket.gaierror("no such name")
+
+    assert egress.vetted_address("x", 80, resolve=no_answer, forbidden=_forbid_internal) is None
+
+
+def test_the_real_judgement_refuses_our_own_networks():
+    # The worker's gate uses urlguard's judgement unchanged — the same one the enqueue check uses.
+    for addr in ("127.0.0.1", "10.1.2.3", "172.18.0.5", "192.168.1.1", "169.254.169.254"):
+        assert egress.vetted_address(addr, 80, forbidden=egress.urlguard.is_forbidden) is None, addr
+    assert egress.vetted_address("93.184.215.14", 80, forbidden=egress.urlguard.is_forbidden)
+
+
+def test_the_gate_dials_the_address_it_vetted_not_a_second_answer(gate):
+    # DNS rebinding: a name that answers public for the check and private for the fetch. The gate
+    # resolves once per connection and dials that answer.
+    public = _public_server({"/": (200, {}, b"public")})
+    internal = _Listener(INTERNAL, public.server_address[1])
+    calls = []
+
+    def rebinding(host, port, *a):
+        calls.append(host)
+        ip = PUBLIC if len(calls) == 1 else INTERNAL
+        return [(2, 1, 6, "", (ip, port))]
+
+    gate.resolve = rebinding
+    try:
+        with socket.create_connection(("127.0.0.1", gate.server_address[1])) as s:
+            s.sendall(
+                f"GET http://rebind.test:{public.server_address[1]}/ HTTP/1.1\r\n"
+                f"Host: rebind.test\r\n\r\n".encode()
+            )
+            answer = b"".join(iter(lambda: s.recv(4096), b""))
+        assert answer.endswith(b"public")
+        assert calls == ["rebind.test"]
+        assert internal.hits == 0
+    finally:
+        public.shutdown()
+        internal.close()
+
+
+def test_a_tunnel_into_the_network_is_refused_before_it_is_dialled(gate):
+    internal = _Listener()
+    try:
+        with socket.create_connection(("127.0.0.1", gate.server_address[1])) as s:
+            s.sendall(f"CONNECT {INTERNAL}:{internal.port} HTTP/1.1\r\n\r\n".encode())
+            answer = s.recv(4096)
+        assert answer.startswith(b"HTTP/1.1 403")
+        assert internal.hits == 0
+    finally:
+        internal.close()
+
+
+# --- a real yt-dlp through it ----------------------------------------------------------------------
+
+
+def test_a_redirect_into_the_network_reaches_it_unguarded_and_not_through_the_gate(tmp_path, gate):
+    internal = _Listener()
+    public = _public_server(
+        {"/v.mp4": (302, {"Location": f"http://{INTERNAL}:{internal.port}/v.mp4"}, b"")}
+    )
+    url = f"http://{PUBLIC}:{public.server_address[1]}/v.mp4"
+    try:
+        # The hole, as it was: yt-dlp follows the 3xx into our network.
+        bare = {"outtmpl": str(tmp_path / "bare" / "%(id)s.%(ext)s"), "quiet": True, "retries": 0}
+        with contextlib.suppress(yt_dlp.utils.DownloadError), yt_dlp.YoutubeDL(bare) as ydl:
+            ydl.extract_info(url, download=True)
+        assert internal.hits >= 1
+        before = internal.hits
+
+        _download_error(_guarded_opts(tmp_path, gate.url), url)
+        assert internal.hits == before
+    finally:
+        public.shutdown()
+        internal.close()
+
+
+def test_a_live_and_a_dead_internal_port_leave_the_same_text(tmp_path, gate):
+    live = _Listener()
+    dead = _Listener()
+    dead_port = dead.port
+    dead.close()
+    public = _public_server(
+        {
+            "/live.mp4": (302, {"Location": f"http://{INTERNAL}:{live.port}/x"}, b""),
+            "/dead.mp4": (302, {"Location": f"http://{INTERNAL}:{dead_port}/x"}, b""),
+        }
+    )
+    base = f"http://{PUBLIC}:{public.server_address[1]}"
+    try:
+        stored = []
+        for name in ("live", "dead"):
+            msg = _download_error(_guarded_opts(tmp_path / name, gate.url), f"{base}/{name}.mp4")
+            stored.append(errors.user_error_text("url", msg, errors.classify(msg)))
+        assert stored[0] == stored[1]
+        assert live.hits == 0
+    finally:
+        public.shutdown()
+        live.close()
+
+
+def test_a_format_fetched_past_the_gate_is_never_chosen(tmp_path):
+    # ffmpeg speaks rtmp, rtsp, tcp… directly, where no HTTP proxy is asked. A guarded download keeps
+    # only the formats fetched over http(s).
+    opts = _guarded_opts(tmp_path, "http://127.0.0.1:9")
+    info = {
+        "id": "x",
+        "title": "x",
+        "extractor": "generic",
+        "extractor_key": "Generic",
+        "webpage_url": f"http://{PUBLIC}/page",
+        "formats": [
+            {"format_id": "rtmp", "url": f"rtmp://{INTERNAL}/live", "ext": "flv", "height": 1080},
+            {"format_id": "web", "url": f"http://{PUBLIC}/a.mp4", "ext": "mp4", "height": 360},
+        ],
+    }
+    with egress.open_ydl(opts) as ydl:
+        chosen = ydl.process_ie_result(dict(info), download=False)
+    assert chosen["format_id"] == "web"
+
+    only_rtmp = dict(info, formats=[info["formats"][0]])
+    with (
+        pytest.raises(yt_dlp.utils.YoutubeDLError, match="Requested format is not available"),
+        egress.open_ydl(opts) as ydl,
+    ):
+        ydl.process_ie_result(only_rtmp, download=False)
+
+
+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
+    `segment` — formatted with the listener's address."""
+    internal = _Listener()
+    playlist = (
+        "#EXTM3U\n#EXT-X-VERSION:3\n#EXT-X-TARGETDURATION:2\n#EXTINF:2.0,\n"
+        + segment.format(host=INTERNAL, port=internal.port)
+        + "\n#EXT-X-ENDLIST\n"
+    ).encode()
+    public = _public_server(
+        {"/p.m3u8": (200, {"Content-Type": "application/vnd.apple.mpegurl"}, playlist)}
+    )
+    try:
+        opts = _guarded_opts(tmp_path, proxy, download_ranges=download_range_func([], [(0, 1)]))
+        with contextlib.suppress(yt_dlp.utils.DownloadError), egress.open_ydl(opts) as ydl:
+            ydl.extract_info(f"http://{PUBLIC}:{public.server_address[1]}/p.m3u8", download=True)
+        return internal.hits
+    finally:
+        public.shutdown()
+        internal.close()
+
+
+def test_a_clip_cut_by_ffmpeg_reaches_a_segment_only_through_the_gate(tmp_path, gate):
+    # Without the gate, ffmpeg fetches a hostile playlist's segment from inside our network — the
+    # control that shows this path is real. Through the gate it is never dialled.
+    segment = "http://{host}:{port}/s.ts"
+    assert _clip_hits(tmp_path / "bare", segment, proxy="") >= 1
+    assert _clip_hits(tmp_path / "gated", segment, proxy=gate.url) == 0
+
+
+def test_ffmpeg_itself_refuses_a_raw_tcp_segment(tmp_path):
+    # A boundary, not our code: the HTTP proxy cannot see a tcp:// segment, and measured, ffmpeg's
+    # HLS reader does not open one from an http playlist. This fails if an ffmpeg upgrade starts to.
+    assert _clip_hits(tmp_path, "tcp://{host}:{port}/", proxy="") == 0
+
+
+# --- the wiring ------------------------------------------------------------------------------------
+
+
+def test_only_youtube_downloads_skip_the_gate():
+    assert adapters.SourceAdapter.egress_guarded is True
+    assert adapters.resolve_adapter("url", "https://www.facebook.com/x").egress_guarded is True
+    assert adapters.resolve_adapter("url", "https://vimeo.com/1").egress_guarded is True
+    assert adapters.resolve_adapter("youtube", "dQw4w9WgXcQ").egress_guarded is False
+
+
+def test_a_guarded_download_goes_through_the_gate_whatever_its_adapter_adds():
+    opts = formats.build_ydl_opts(
+        {}, "/tmp/x", lambda d: None, extra={"proxy": "", "hls_prefer_native": False}, egress=True
+    )
+    assert opts["proxy"] == egress.proxy_url()
+    assert opts["hls_prefer_native"] is True
+    assert opts[egress.GUARDED] is True
+    assert "proxy" not in formats.build_ydl_opts({}, "/tmp/x", lambda d: None, egress=False)
+
+
+def test_every_yt_dlp_this_app_opens_goes_through_open_ydl():
+    app = Path(egress.__file__).resolve().parents[1]
+    offenders = [
+        f"{p.relative_to(app)}:{n}"
+        for p in sorted(app.rglob("*.py"))
+        if p.name != "egress.py"
+        for n, line in enumerate(p.read_text(encoding="utf-8").splitlines(), 1)
+        if "YoutubeDL(" in line
+    ]
+    assert offenders == []
diff --git a/backend/tests/test_facebook_impersonation.py b/backend/tests/test_facebook_impersonation.py
index fc5c7840..61cfb150 100644
--- a/backend/tests/test_facebook_impersonation.py
+++ b/backend/tests/test_facebook_impersonation.py
@@ -73,6 +73,9 @@ def ydl(monkeypatch):
         def __enter__(self):
             return self
 
+        def add_post_processor(self, pp, when="post_process"):
+            pass  # the egress format filter; nothing here to filter
+
         def __exit__(self, *exc):
             return False
 
diff --git a/backend/tests/test_facebook_titles.py b/backend/tests/test_facebook_titles.py
index 25372183..145eb320 100644
--- a/backend/tests/test_facebook_titles.py
+++ b/backend/tests/test_facebook_titles.py
@@ -36,6 +36,9 @@ def _fake_ydl(monkeypatch, outcome):
         def __enter__(self):
             return self
 
+        def add_post_processor(self, pp, when="post_process"):
+            pass  # the egress format filter; nothing here to filter
+
         def __exit__(self, *exc):
             return False
 
@@ -43,7 +46,7 @@ def _fake_ydl(monkeypatch, outcome):
             seen.append({**self.opts, "url": url})
             return outcome
 
-    monkeypatch.setattr(backfill.yt_dlp, "YoutubeDL", FakeYDL)
+    monkeypatch.setattr("yt_dlp.YoutubeDL", FakeYDL)
     return seen
 
 
diff --git a/backend/tests/test_failure_lines.py b/backend/tests/test_failure_lines.py
index e5122140..de021f3a 100644
--- a/backend/tests/test_failure_lines.py
+++ b/backend/tests/test_failure_lines.py
@@ -359,6 +359,9 @@ def test_the_metadata_backfill_names_the_asset_and_not_the_download(sink, monkey
         def __enter__(self):
             return self
 
+        def add_post_processor(self, pp, when="post_process"):
+            pass  # the egress format filter; nothing here to filter
+
         def __exit__(self, *exc):
             return False
 
@@ -367,7 +370,7 @@ def test_the_metadata_backfill_names_the_asset_and_not_the_download(sink, monkey
                 f"ERROR: [youtube] dQw4w9WgXcQ: Video unavailable ({target})"
             )
 
-    monkeypatch.setattr(backfill.yt_dlp, "YoutubeDL", _Refusing)
+    monkeypatch.setattr("yt_dlp.YoutubeDL", _Refusing)
     assert backfill.extract_meta(url, adapters.YouTubeAdapter(), 11) is None
 
     lines = [o for o in sink.objects() if "meta extract failed" in o["msg"]]
@@ -386,13 +389,16 @@ def test_the_metadata_backfill_takes_out_a_generic_name_whatever_it_holds(sink,
         def __enter__(self):
             return self
 
+        def add_post_processor(self, pp, when="post_process"):
+            pass  # the egress format filter; nothing here to filter
+
         def __exit__(self, *exc):
             return False
 
         def extract_info(self, target, download):
             raise yt_dlp.utils.DownloadError("ERROR: [generic] Our Trip: Day 1: Unable to download")
 
-    monkeypatch.setattr(backfill.yt_dlp, "YoutubeDL", _Refusing)
+    monkeypatch.setattr("yt_dlp.YoutubeDL", _Refusing)
     assert backfill.extract_meta(url, adapters.SourceAdapter(), 11) is None
 
     lines = [o for o in sink.objects() if "meta extract failed" in o["msg"]]
@@ -449,6 +455,9 @@ def test_a_retried_attempt_names_the_host_and_not_the_video(sink, monkeypatch, t
         def __enter__(self):
             return self
 
+        def add_post_processor(self, pp, when="post_process"):
+            pass  # the egress format filter; nothing here to filter
+
         def __exit__(self, *exc):
             return False
 
@@ -463,6 +472,7 @@ def test_a_retried_attempt_names_the_host_and_not_the_video(sink, monkeypatch, t
         resolve_url=lambda target, opts: target,
         ydl_opts_extra=lambda clients: {},
         retriable=lambda text: True,
+        egress_guarded=False,
     )
     monkeypatch.setattr(pipeline.yt_dlp, "YoutubeDL", _YDL)
     monkeypatch.setattr(pipeline, "_touch_job", lambda job_id: None)
diff --git a/backend/tests/test_formats.py b/backend/tests/test_formats.py
index a0369c5d..d6ef2c61 100644
--- a/backend/tests/test_formats.py
+++ b/backend/tests/test_formats.py
@@ -15,16 +15,20 @@ from app.downloads import formats
 
 
 def _opts(spec):
-    return formats.build_ydl_opts(spec, "/tmp/%(id)s.%(ext)s", lambda d: None)
+    return formats.build_ydl_opts(spec, "/tmp/%(id)s.%(ext)s", lambda d: None, egress=False)
 
 
 def test_adapter_extra_opts_are_merged_in():
     # The SourceAdapter contributes source-specific opts (e.g. YouTube's extractor_args) via `extra`.
     extra = {"extractor_args": {"youtube": {"player_client": ["web"]}}}
-    opts = formats.build_ydl_opts({}, "/tmp/%(id)s.%(ext)s", lambda d: None, extra=extra)
+    opts = formats.build_ydl_opts(
+        {}, "/tmp/%(id)s.%(ext)s", lambda d: None, extra=extra, egress=False
+    )
     assert opts["extractor_args"]["youtube"]["player_client"] == ["web"]
     # No extra → no extractor_args (a generic source contributes nothing).
-    assert "extractor_args" not in formats.build_ydl_opts({}, "/tmp/x", lambda d: None)
+    assert "extractor_args" not in formats.build_ydl_opts(
+        {}, "/tmp/x", lambda d: None, egress=False
+    )
 
 
 def test_mp4_default_selects_avc1_with_progressive_fallback():
@@ -113,7 +117,7 @@ def test_a_playlist_url_can_only_yield_one_item():
 
 
 def test_the_remaining_quota_becomes_a_hard_size_ceiling():
-    opts = formats.build_ydl_opts({}, "/tmp/x", lambda d: None, max_bytes=5_000_000)
+    opts = formats.build_ydl_opts({}, "/tmp/x", lambda d: None, max_bytes=5_000_000, egress=False)
     assert opts["max_filesize"] == 5_000_000
 
 
@@ -122,7 +126,7 @@ def test_no_budget_means_no_size_filter():
     # enqueue, not an impossible yt-dlp filter to express here.
     for budget in (None, 0, -1):
         assert "max_filesize" not in formats.build_ydl_opts(
-            {}, "/tmp/x", lambda d: None, max_bytes=budget
+            {}, "/tmp/x", lambda d: None, max_bytes=budget, egress=False
         )
 
 
@@ -130,7 +134,12 @@ def test_a_source_adapter_cannot_raise_the_size_ceiling():
     # The adapter merge happens after max_filesize is set, so this pins that no adapter option can
     # widen a storage cap — the mirror of the cookiefile rule, in the opposite direction.
     opts = formats.build_ydl_opts(
-        {}, "/tmp/x", lambda d: None, extra={"max_filesize": 999_999_999}, max_bytes=5_000_000
+        {},
+        "/tmp/x",
+        lambda d: None,
+        extra={"max_filesize": 999_999_999},
+        max_bytes=5_000_000,
+        egress=False,
     )
     assert opts["max_filesize"] == 5_000_000
 
@@ -197,7 +206,7 @@ def test_yt_dlp_refuses_a_file_over_the_cap_without_raising_and_the_watch_sees_i
         for cap, refused in ((1_000, True), (10_000_000, False)):
             out = tmp_path / f"out-{cap}"
             opts = formats.build_ydl_opts(
-                {}, str(out / "%(id)s.%(ext)s"), lambda d: None, max_bytes=cap
+                {}, str(out / "%(id)s.%(ext)s"), lambda d: None, max_bytes=cap, egress=False
             )
             with formats.watching_size_cap() as over, yt_dlp.YoutubeDL(opts) as ydl:
                 ydl.extract_info(url, download=True)
diff --git a/backend/tests/test_ytdlp_logging.py b/backend/tests/test_ytdlp_logging.py
index bc98cacc..57ff10f2 100644
--- a/backend/tests/test_ytdlp_logging.py
+++ b/backend/tests/test_ytdlp_logging.py
@@ -216,7 +216,7 @@ def test_a_line_with_no_url_is_left_alone(sink):
 
 
 def test_the_download_options_carry_the_logger():
-    opts = formats.build_ydl_opts({}, "%(id)s.%(ext)s", lambda d: None)
+    opts = formats.build_ydl_opts({}, "%(id)s.%(ext)s", lambda d: None, egress=False)
 
     assert opts["logger"] is formats._YDL_LOG
 
@@ -227,7 +227,7 @@ def test_a_source_adapter_cannot_silence_it():
     adapter must not be able to take away the one account of what went wrong that a failed job
     leaves behind."""
     opts = formats.build_ydl_opts(
-        {}, "%(id)s.%(ext)s", lambda d: None, extra={"logger": None, "quiet": False}
+        {}, "%(id)s.%(ext)s", lambda d: None, extra={"logger": None, "quiet": False}, egress=False
     )
 
     assert opts["logger"] is formats._YDL_LOG