diff --git a/app/tests/test_url_guard.py b/app/tests/test_url_guard.py index 3cd1460..120a005 100644 --- a/app/tests/test_url_guard.py +++ b/app/tests/test_url_guard.py @@ -6,7 +6,13 @@ import socket import unittest from unittest import mock -from url_guard import validate_url +import url_guard +from url_guard import ( + validate_url, + _address_allowed_at_connect, + _guarded_getaddrinfo, + install_socket_guard, +) def _addrinfo(*addrs, family=socket.AF_INET): @@ -99,5 +105,58 @@ class AddressResolutionTests(unittest.TestCase): self.assertIsNotNone(validate_url("http://does-not-resolve.example/x")) +class ConnectAddressPolicyTests(unittest.TestCase): + """Connect-time policy: allow global + loopback, block everything else.""" + + def test_global_allowed(self): + self.assertTrue(_address_allowed_at_connect("142.250.1.1")) + + def test_loopback_allowed(self): + # Loopback stays reachable so locally-configured proxies keep working. + self.assertTrue(_address_allowed_at_connect("127.0.0.1")) + self.assertTrue(_address_allowed_at_connect("::1")) + + def test_link_local_metadata_blocked(self): + self.assertFalse(_address_allowed_at_connect("169.254.169.254")) + + def test_private_blocked(self): + self.assertFalse(_address_allowed_at_connect("10.0.0.5")) + self.assertFalse(_address_allowed_at_connect("192.168.1.10")) + + def test_ipv4_mapped_metadata_blocked(self): + self.assertFalse(_address_allowed_at_connect("::ffff:169.254.169.254")) + + +class GuardedGetaddrinfoTests(unittest.TestCase): + def test_internal_only_raises(self): + with mock.patch("url_guard._real_getaddrinfo", return_value=_addrinfo("169.254.169.254")): + with self.assertRaises(socket.gaierror): + _guarded_getaddrinfo("metadata", 80) + + def test_filters_internal_keeps_global(self): + # Split-horizon rebinding: keep the public address, drop the internal one. + with mock.patch("url_guard._real_getaddrinfo", return_value=_addrinfo("142.250.1.1", "10.0.0.1")): + results = _guarded_getaddrinfo("mixed", 80) + self.assertEqual([r[4][0] for r in results], ["142.250.1.1"]) + + def test_loopback_passes(self): + with mock.patch("url_guard._real_getaddrinfo", return_value=_addrinfo("127.0.0.1")): + results = _guarded_getaddrinfo("localproxy", 9050) + self.assertEqual([r[4][0] for r in results], ["127.0.0.1"]) + + +class InstallSocketGuardTests(unittest.TestCase): + def test_install_replaces_and_is_idempotent(self): + original = socket.getaddrinfo + try: + install_socket_guard() + self.assertIs(socket.getaddrinfo, url_guard._guarded_getaddrinfo) + # Re-installing must not wrap the wrapper (real fn captured at import). + install_socket_guard() + self.assertIs(socket.getaddrinfo, url_guard._guarded_getaddrinfo) + finally: + socket.getaddrinfo = original + + if __name__ == "__main__": unittest.main() diff --git a/app/url_guard.py b/app/url_guard.py index e75ec0b..d4b91ee 100644 --- a/app/url_guard.py +++ b/app/url_guard.py @@ -5,10 +5,16 @@ any ``http(s)`` URL. Without a guard, an attacker can make the server fetch internal endpoints (cloud metadata services, loopback, RFC1918 hosts, etc.) and have the response saved to the download directory and served back. -This module provides a single cheap validator applied at every URL ingress. It -intentionally does NOT attempt DNS-rebinding pinning, redirect-chain -re-validation, or validation of every media URL yt-dlp derives from remote -metadata — network isolation (e.g. Docker) remains the backstop for those. +This module provides two layers: + +* ``validate_url`` — a cheap validator applied at every URL ingress. +* ``install_socket_guard`` — a connect-time ``getaddrinfo`` guard installed in + the download subprocess, which re-validates every resolved address and so + covers redirects, DNS rebinding, and media URLs yt-dlp derives from remote + metadata — for any backend that resolves through Python's socket module. + +Native resolvers (curl_cffi/libcurl via ``--impersonate``) bypass the socket +guard; network isolation (e.g. Docker) remains the backstop for those. """ import ipaddress @@ -34,16 +40,60 @@ def _hostname_is_blocked(hostname: str) -> bool: return False -def _address_is_global(addr: str) -> bool: +def _normalise_ip(addr: str): + """Parse *addr*, unwrapping IPv4-mapped IPv6 (e.g. ``::ffff:169.254.169.254``) + so the embedded IPv4 address is judged on its own merits. Returns ``None`` + when *addr* is not a valid IP literal.""" try: ip = ipaddress.ip_address(addr) except ValueError: - return False - # Unwrap IPv4-mapped/compatible IPv6 (e.g. ::ffff:169.254.169.254) so the - # embedded IPv4 address is judged on its own merits. + return None if isinstance(ip, ipaddress.IPv6Address) and ip.ipv4_mapped is not None: ip = ip.ipv4_mapped - return ip.is_global + return ip + + +def _address_is_global(addr: str) -> bool: + ip = _normalise_ip(addr) + return ip is not None and ip.is_global + + +def _address_allowed_at_connect(addr: str) -> bool: + """True if *addr* may be connected to at download time. + + Permits global addresses and loopback — loopback so that locally-configured + proxies (e.g. ``proxy: http://127.0.0.1:9050``) keep working. Blocks the SSRF + targets that matter: link-local (cloud metadata at 169.254.169.254), private + (RFC1918), unique-local and every other non-global, non-loopback range. + """ + ip = _normalise_ip(addr) + return ip is not None and (ip.is_global or ip.is_loopback) + + +# Captured at import so re-installing the guard never wraps the wrapper. +_real_getaddrinfo = socket.getaddrinfo + + +def _guarded_getaddrinfo(host, *args, **kwargs): + results = _real_getaddrinfo(host, *args, **kwargs) + allowed = [r for r in results if _address_allowed_at_connect(r[4][0])] + if not allowed: + raise socket.gaierror(f'Refusing to connect to non-global address for host {host!r}') + return allowed + + +def install_socket_guard() -> None: + """Enforce the no-internal-hosts policy at actual connection time. + + ``validate_url`` only checks the *submitted* URL string; yt-dlp then follows + HTTP redirects and resolves media URLs from remote metadata without + re-validating them. Installing this in the download subprocess re-checks + every resolved address at connect time, covering redirects and DNS rebinding + for any networking backend that resolves through Python's socket module + (urllib, requests). Native resolvers — notably curl_cffi/libcurl used by + ``--impersonate`` — bypass this and rely on network isolation as the backstop. + """ + socket.getaddrinfo = _guarded_getaddrinfo def validate_url(url: str) -> str | None: diff --git a/app/ytdl.py b/app/ytdl.py index f4f4553..dc485fc 100644 --- a/app/ytdl.py +++ b/app/ytdl.py @@ -26,7 +26,7 @@ from dl_formats import get_format, get_opts, AUDIO_FORMATS, merge_ytdl_option_la from datetime import datetime from state_store import AtomicJsonStore, from_json_compatible, read_legacy_shelf, to_json_compatible from subscriptions import _entry_id -from url_guard import validate_url +from url_guard import validate_url, install_socket_guard log = logging.getLogger('ytdl') @@ -635,6 +635,10 @@ class Download: os.setpgrp() except OSError: pass + # Re-validate every outbound connection at fetch time. validate_url only + # saw the submitted URL string; this catches redirects and DNS rebinding + # to internal hosts (cloud metadata, RFC1918) that it cannot. + install_socket_guard() log.info(f"Starting download for: {self.info.title} ({self.info.url})") try: debug_logging = logging.getLogger().isEnabledFor(logging.DEBUG)