diff --git a/app/tests/test_url_guard.py b/app/tests/test_url_guard.py index fae5cdc..0218b74 100644 --- a/app/tests/test_url_guard.py +++ b/app/tests/test_url_guard.py @@ -11,6 +11,7 @@ from url_guard import ( validate_url, _address_allowed_at_connect, _guarded_getaddrinfo, + _proxy_endpoint, install_socket_guard, ) @@ -106,15 +107,25 @@ class AddressResolutionTests(unittest.TestCase): class ConnectAddressPolicyTests(unittest.TestCase): - """Connect-time policy: allow global + loopback, block everything else.""" + """Connect-time policy: allow global, plus loopback only when the caller has + established that this destination is the operator's configured proxy.""" 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_loopback_blocked_by_default(self): + # A blanket loopback allowance is what let manifest-derived media URLs + # reach services on the server's own loopback interface. + self.assertFalse(_address_allowed_at_connect("127.0.0.1")) + self.assertFalse(_address_allowed_at_connect("::1")) + + def test_loopback_allowed_only_when_opted_in(self): + self.assertTrue(_address_allowed_at_connect("127.0.0.1", allow_loopback=True)) + self.assertTrue(_address_allowed_at_connect("::1", allow_loopback=True)) + + def test_opt_in_does_not_widen_beyond_loopback(self): + self.assertFalse(_address_allowed_at_connect("169.254.169.254", allow_loopback=True)) + self.assertFalse(_address_allowed_at_connect("10.0.0.5", allow_loopback=True)) def test_link_local_metadata_blocked(self): self.assertFalse(_address_allowed_at_connect("169.254.169.254")) @@ -127,7 +138,37 @@ class ConnectAddressPolicyTests(unittest.TestCase): self.assertFalse(_address_allowed_at_connect("::ffff:169.254.169.254")) +class ProxyEndpointParsingTests(unittest.TestCase): + def test_explicit_port(self): + self.assertEqual(_proxy_endpoint("http://127.0.0.1:9050"), ("127.0.0.1", 9050)) + + def test_default_port_per_scheme(self): + self.assertEqual(_proxy_endpoint("socks5://127.0.0.1"), ("127.0.0.1", 1080)) + self.assertEqual(_proxy_endpoint("http://127.0.0.1"), ("127.0.0.1", 80)) + + def test_bare_host_port(self): + self.assertEqual(_proxy_endpoint("127.0.0.1:8080"), ("127.0.0.1", 8080)) + + def test_hostname_lowercased(self): + self.assertEqual(_proxy_endpoint("http://LocalHost.:9050"), ("localhost", 9050)) + + def test_ipv6_literal(self): + self.assertEqual(_proxy_endpoint("http://[::1]:9050"), ("::1", 9050)) + + def test_empty_and_invalid(self): + self.assertIsNone(_proxy_endpoint("")) + self.assertIsNone(_proxy_endpoint(" ")) + self.assertIsNone(_proxy_endpoint(None)) + self.assertIsNone(_proxy_endpoint("http://")) + + class GuardedGetaddrinfoTests(unittest.TestCase): + def setUp(self): + # Default state: no proxy configured, so no loopback destination allowed. + saved = set(url_guard._allowed_loopback_endpoints) + url_guard._allowed_loopback_endpoints = set() + self.addCleanup(lambda: setattr(url_guard, "_allowed_loopback_endpoints", saved)) + 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): @@ -139,9 +180,35 @@ class GuardedGetaddrinfoTests(unittest.TestCase): results = _guarded_getaddrinfo("mixed", 80) self.assertEqual([r[4][0] for r in results], ["142.250.1.1"]) - def test_loopback_passes(self): + def test_loopback_blocked_without_matching_proxy(self): + # The advisory case: an m3u8 segment URL pointing at a loopback service. with mock.patch("url_guard._real_getaddrinfo", return_value=_addrinfo("127.0.0.1")): - results = _guarded_getaddrinfo("localproxy", 9050) + with self.assertRaises(socket.gaierror): + _guarded_getaddrinfo("127.0.0.1", 9999) + + def test_loopback_allowed_at_configured_proxy_endpoint(self): + url_guard._allowed_loopback_endpoints = {("127.0.0.1", 9050)} + with mock.patch("url_guard._real_getaddrinfo", return_value=_addrinfo("127.0.0.1")): + results = _guarded_getaddrinfo("127.0.0.1", 9050) + self.assertEqual([r[4][0] for r in results], ["127.0.0.1"]) + + def test_loopback_blocked_at_other_port_on_proxy_host(self): + # Same host as the proxy, different port: still off limits. + url_guard._allowed_loopback_endpoints = {("127.0.0.1", 9050)} + with mock.patch("url_guard._real_getaddrinfo", return_value=_addrinfo("127.0.0.1")): + with self.assertRaises(socket.gaierror): + _guarded_getaddrinfo("127.0.0.1", 9999) + + def test_proxy_reachable_by_hostname(self): + url_guard._allowed_loopback_endpoints = {("localhost", 9050)} + with mock.patch("url_guard._real_getaddrinfo", return_value=_addrinfo("127.0.0.1")): + results = _guarded_getaddrinfo("localhost", 9050) + self.assertEqual([r[4][0] for r in results], ["127.0.0.1"]) + + def test_string_port_is_normalised(self): + url_guard._allowed_loopback_endpoints = {("127.0.0.1", 9050)} + with mock.patch("url_guard._real_getaddrinfo", return_value=_addrinfo("127.0.0.1")): + results = _guarded_getaddrinfo("127.0.0.1", "9050") self.assertEqual([r[4][0] for r in results], ["127.0.0.1"]) @@ -172,16 +239,44 @@ class AllowPrivateBypassTests(unittest.TestCase): class InstallSocketGuardTests(unittest.TestCase): + def setUp(self): + original, saved = socket.getaddrinfo, set(url_guard._allowed_loopback_endpoints) + self.addCleanup(lambda: setattr(socket, "getaddrinfo", original)) + self.addCleanup(lambda: setattr(url_guard, "_allowed_loopback_endpoints", saved)) + # Keep the host's own environment out of the assertions below. + patcher = mock.patch("url_guard.urllib.request.getproxies", return_value={}) + self.getproxies = patcher.start() + self.addCleanup(patcher.stop) + 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 + 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) + + def test_no_proxy_means_no_loopback_allowance(self): + install_socket_guard() + self.assertEqual(url_guard._allowed_loopback_endpoints, set()) + + def test_explicit_proxy_is_registered(self): + install_socket_guard(proxy_urls=("socks5://127.0.0.1:9050",)) + self.assertEqual(url_guard._allowed_loopback_endpoints, {("127.0.0.1", 9050)}) + + def test_unset_proxy_option_is_ignored(self): + # ytdl_opts.get('proxy') is None when the operator configured no proxy. + install_socket_guard(proxy_urls=(None,)) + self.assertEqual(url_guard._allowed_loopback_endpoints, set()) + + def test_environment_proxies_are_registered(self): + self.getproxies.return_value = {"http": "http://127.0.0.1:8080"} + install_socket_guard() + self.assertEqual(url_guard._allowed_loopback_endpoints, {("127.0.0.1", 8080)}) + + def test_endpoints_reset_between_installs(self): + install_socket_guard(proxy_urls=("http://127.0.0.1:8080",)) + install_socket_guard(proxy_urls=(None,)) + self.assertEqual(url_guard._allowed_loopback_endpoints, set()) if __name__ == "__main__": diff --git a/app/url_guard.py b/app/url_guard.py index 8b0bd92..5823ec1 100644 --- a/app/url_guard.py +++ b/app/url_guard.py @@ -30,12 +30,23 @@ all of these: import ipaddress import logging import socket +import urllib.request from urllib.parse import urlsplit log = logging.getLogger('url_guard') _ALLOWED_SCHEMES = ('http', 'https') +# Ports to assume when a configured proxy URL omits one, per proxy scheme. +_PROXY_DEFAULT_PORTS = { + 'http': 80, + 'https': 443, + 'socks4': 1080, + 'socks4a': 1080, + 'socks5': 1080, + 'socks5h': 1080, +} + # Hostnames that must be blocked without needing a lookup. ``localhost`` and any # subdomain of it are conventionally loopback, and the GCP metadata name is a # well-known SSRF target that may resolve via a resolver we don't control. @@ -68,40 +79,107 @@ def _address_is_global(addr: str) -> bool: return ip is not None and ip.is_global -def _address_allowed_at_connect(addr: str) -> bool: +def _address_allowed_at_connect(addr: str, allow_loopback: bool = False) -> 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. + Permits global addresses only. Loopback is permitted just for the specific + host:port of an operator-configured proxy (see ``_loopback_endpoint_allowed``), + never as a blanket rule: media URLs that yt-dlp derives from a remote manifest + are attacker-controlled and reach this policy without passing ``validate_url``, + so a general loopback allowance would let a hostile playlist read any service + on the server's loopback interface. Blocks link-local (cloud metadata at + 169.254.169.254), private (RFC1918), unique-local and every other non-global + range. """ ip = _normalise_ip(addr) - return ip is not None and (ip.is_global or ip.is_loopback) + if ip is None: + return False + return ip.is_global or (allow_loopback and ip.is_loopback) + + +def _proxy_endpoint(proxy_url: str): + """Parse a proxy URL into a ``(hostname, port)`` pair, or ``None`` if it has + no usable host. Used to scope the loopback allowance to that endpoint alone.""" + if not isinstance(proxy_url, str) or not proxy_url.strip(): + return None + candidate = proxy_url.strip() + if '://' not in candidate: + # Bare host:port, as accepted by the *_proxy environment variables. + candidate = '//' + candidate + try: + parts = urlsplit(candidate) + hostname, port = parts.hostname, parts.port + except ValueError: + return None + if not hostname: + return None + if port is None: + port = _PROXY_DEFAULT_PORTS.get(parts.scheme.lower()) + return (hostname.rstrip('.').lower(), port) + + +def _collect_proxy_endpoints(proxy_urls) -> set: + """Endpoints of every proxy this download may legitimately dial: the explicit + yt-dlp ``proxy`` option plus the ``*_proxy`` environment variables yt-dlp falls + back to. All are operator-configured, unlike the URLs inside fetched media.""" + candidates = list(proxy_urls) + list(urllib.request.getproxies().values()) + return {ep for ep in map(_proxy_endpoint, candidates) if ep is not None} # Captured at import so re-installing the guard never wraps the wrapper. _real_getaddrinfo = socket.getaddrinfo +# Populated by install_socket_guard; empty means no loopback destination is allowed. +_allowed_loopback_endpoints: set = set() + + +def _normalise_port(port): + if isinstance(port, str): + try: + return int(port) + except ValueError: + try: + return socket.getservbyname(port) + except OSError: + return None + return port + + +def _loopback_endpoint_allowed(host, port) -> bool: + if not _allowed_loopback_endpoints or host is None: + return False + return (str(host).rstrip('.').lower(), _normalise_port(port)) in _allowed_loopback_endpoints + 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])] + # Mirrors getaddrinfo(host, port, ...): port is the first optional argument. + port = args[0] if args else kwargs.get('port') + allow_loopback = _loopback_endpoint_allowed(host, port) + allowed = [r for r in results if _address_allowed_at_connect(r[4][0], allow_loopback)] if not allowed: raise socket.gaierror(f'Refusing to connect to non-global address for host {host!r}') return allowed -def install_socket_guard(allow_private: bool = False) -> None: +def install_socket_guard(allow_private: bool = False, proxy_urls=()) -> 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. + every resolved address at connect time, covering redirects, DNS rebinding and + manifest-derived media URLs 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. + + *proxy_urls* are the operator's configured proxies (yt-dlp's ``proxy`` option; + the ``*_proxy`` environment variables are picked up automatically). A proxy on + loopback is reachable at its own host:port, and nothing else on loopback is. + That costs proxied setups nothing: yt-dlp resolves the proxy itself at exactly + that host:port, and a media URL is either handed to the proxy unresolved or + resolved on its own merits — never inheriting the proxy's allowance. When *allow_private* is set (``ALLOW_PRIVATE_ADDRESSES``), the guard is not installed at all, so proxy/VPN setups that route through private or Fake-IP @@ -109,6 +187,12 @@ def install_socket_guard(allow_private: bool = False) -> None: """ if allow_private: return + _allowed_loopback_endpoints.clear() + _allowed_loopback_endpoints.update(_collect_proxy_endpoints(proxy_urls)) + for host, port in sorted(_allowed_loopback_endpoints, key=lambda ep: (ep[0], ep[1] or 0)): + ip = _normalise_ip(host) + if ip is not None and ip.is_loopback: + log.info(f'Allowing connections to configured loopback proxy {host}:{port}') socket.getaddrinfo = _guarded_getaddrinfo diff --git a/app/ytdl.py b/app/ytdl.py index 2f51230..77e758c 100644 --- a/app/ytdl.py +++ b/app/ytdl.py @@ -657,10 +657,12 @@ class Download: 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. Skipped when - # ALLOW_PRIVATE_ADDRESSES trusts the environment (e.g. Fake-IP proxies). - install_socket_guard(self.allow_private) + # saw the submitted URL string; this catches redirects, DNS rebinding and + # attacker-controlled media URLs pulled from a remote manifest, none of + # which it can see. The configured proxy is passed so that a proxy on + # loopback stays reachable at its own address without opening up the rest + # of loopback. Skipped when ALLOW_PRIVATE_ADDRESSES trusts the environment. + install_socket_guard(self.allow_private, proxy_urls=(self.ytdl_opts.get('proxy'),)) log.info(f"Starting download for: {self.info.title} ({self.info.url})") try: debug_logging = logging.getLogger().isEnabledFor(logging.DEBUG)