fix(downloads): a user's download is judged at every connection it makes, not only at the URL typed

The URL guard judged the address the user typed; a public host's 3xx into our
network, a page's formats pointing inside, and a name rebound between the check
and the fetch all got past it, and the stored error told a live internal port
from a dead one. Every non-YouTube download now runs through an egress gate on
the worker's loopback: it resolves each host once, refuses any forbidden answer
before dialling, and connects to the address it vetted. Measured on the worker
image: yt-dlp's proxy reaches the Requests and CurlCFFI handlers, native HLS and
ffmpeg alike. Formats fetched over anything but http(s), which ffmpeg would dial
directly, are dropped, and one factory opens every yt-dlp so none skips it.
This commit is contained in:
2026-11-06 04:54:56 +01:00
parent 995a493bd3
commit 4e4fff87d4
14 changed files with 668 additions and 65 deletions
+8 -2
View File
@@ -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
+7 -4
View File
@@ -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
+264
View File
@@ -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("+"))
+15 -1
View File
@@ -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(
+4 -2
View File
@@ -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
+8 -18
View File
@@ -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)