diff --git a/app/tests/test_url_guard.py b/app/tests/test_url_guard.py index c2b4415..7279616 100644 --- a/app/tests/test_url_guard.py +++ b/app/tests/test_url_guard.py @@ -108,8 +108,8 @@ class AddressResolutionTests(unittest.TestCase): class ConnectAddressPolicyTests(unittest.TestCase): - """Connect-time policy: allow global, plus loopback only when the caller has - established that this destination is the operator's configured proxy.""" + """Connect-time policy: allow global, plus anything at a destination the + caller has established is the operator's configured proxy.""" def test_global_allowed(self): self.assertTrue(_address_allowed_at_connect("142.250.1.1")) @@ -121,12 +121,19 @@ class ConnectAddressPolicyTests(unittest.TestCase): 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)) + self.assertTrue(_address_allowed_at_connect("127.0.0.1", is_proxy_endpoint=True)) + self.assertTrue(_address_allowed_at_connect("::1", is_proxy_endpoint=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_proxy_opt_in_covers_any_internal_range(self): + # A proxy is just as legitimately on the LAN or a VPN range as on + # loopback (#1055): the allowance follows the operator's configured + # endpoint, not a particular address family. + self.assertTrue(_address_allowed_at_connect("10.1.20.30", is_proxy_endpoint=True)) + self.assertTrue(_address_allowed_at_connect("192.168.1.10", is_proxy_endpoint=True)) + self.assertTrue(_address_allowed_at_connect("fd00::1", is_proxy_endpoint=True)) + + def test_opt_in_still_rejects_non_addresses(self): + self.assertFalse(_address_allowed_at_connect("not-an-ip", is_proxy_endpoint=True)) def test_link_local_metadata_blocked(self): self.assertFalse(_address_allowed_at_connect("169.254.169.254")) @@ -207,9 +214,9 @@ class ProxyEndpointParsingTests(unittest.TestCase): 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)) + saved = set(url_guard._allowed_proxy_endpoints) + url_guard._allowed_proxy_endpoints = set() + self.addCleanup(lambda: setattr(url_guard, "_allowed_proxy_endpoints", saved)) def test_internal_only_raises(self): with mock.patch("url_guard._real_getaddrinfo", return_value=_addrinfo("169.254.169.254")): @@ -229,30 +236,53 @@ class GuardedGetaddrinfoTests(unittest.TestCase): _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)} + url_guard._allowed_proxy_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)} + url_guard._allowed_proxy_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)} + url_guard._allowed_proxy_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)} + url_guard._allowed_proxy_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_lan_proxy_reachable(self): + # #1055: a socks5 proxy on the LAN, refused while the allowance was + # loopback-only, which pushed operators to ALLOW_PRIVATE_ADDRESSES. + url_guard._allowed_proxy_endpoints = {("10.1.20.30", 1080)} + with mock.patch("url_guard._real_getaddrinfo", return_value=_addrinfo("10.1.20.30")): + results = _guarded_getaddrinfo("10.1.20.30", 1080) + self.assertEqual([r[4][0] for r in results], ["10.1.20.30"]) + + def test_other_lan_host_still_blocked(self): + # The allowance is the proxy's endpoint, not its subnet. + url_guard._allowed_proxy_endpoints = {("10.1.20.30", 1080)} + with mock.patch("url_guard._real_getaddrinfo", return_value=_addrinfo("10.1.20.31")): + with self.assertRaises(socket.gaierror): + _guarded_getaddrinfo("10.1.20.31", 1080) + + def test_proxy_address_not_borrowable_by_another_host(self): + # Matching is on the configured host string: a manifest URL that resolves + # to the proxy's address under its own name gets no allowance. + url_guard._allowed_proxy_endpoints = {("10.1.20.30", 1080)} + with mock.patch("url_guard._real_getaddrinfo", return_value=_addrinfo("10.1.20.30")): + with self.assertRaises(socket.gaierror): + _guarded_getaddrinfo("evil.example", 1080) + class AllowPrivateBypassTests(unittest.TestCase): """ALLOW_PRIVATE_ADDRESSES: trusted proxy/VPN environments opt out of the @@ -282,9 +312,9 @@ class AllowPrivateBypassTests(unittest.TestCase): class InstallSocketGuardTests(unittest.TestCase): def setUp(self): - original, saved = socket.getaddrinfo, set(url_guard._allowed_loopback_endpoints) + original, saved = socket.getaddrinfo, set(url_guard._allowed_proxy_endpoints) self.addCleanup(lambda: setattr(socket, "getaddrinfo", original)) - self.addCleanup(lambda: setattr(url_guard, "_allowed_loopback_endpoints", saved)) + self.addCleanup(lambda: setattr(url_guard, "_allowed_proxy_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() @@ -299,26 +329,26 @@ class InstallSocketGuardTests(unittest.TestCase): def test_no_proxy_means_no_loopback_allowance(self): install_socket_guard() - self.assertEqual(url_guard._allowed_loopback_endpoints, set()) + self.assertEqual(url_guard._allowed_proxy_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)}) + self.assertEqual(url_guard._allowed_proxy_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()) + self.assertEqual(url_guard._allowed_proxy_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)}) + self.assertEqual(url_guard._allowed_proxy_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()) + self.assertEqual(url_guard._allowed_proxy_endpoints, set()) if __name__ == "__main__": diff --git a/app/url_guard.py b/app/url_guard.py index ed30908..9dfea77 100644 --- a/app/url_guard.py +++ b/app/url_guard.py @@ -129,29 +129,28 @@ def _address_is_global(addr: str) -> bool: return bool(ips) and all(ip.is_global for ip in ips) -def _address_allowed_at_connect(addr: str, allow_loopback: bool = False) -> bool: +def _address_allowed_at_connect(addr: str, is_proxy_endpoint: bool = False) -> bool: """True if *addr* may be connected to at download time. - 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. + Permits global addresses, and anything at all when the destination is an + operator-configured proxy (see ``_is_proxy_endpoint``). Internal addresses + are otherwise refused with no blanket exception: media URLs that yt-dlp + derives from a remote manifest are attacker-controlled and reach this policy + without passing ``validate_url``, so any range opened here is a range a + hostile playlist can read from the server's own network. Blocks link-local + (cloud metadata at 169.254.169.254), private (RFC1918), loopback, + unique-local and every other non-global range. """ ips = _ips_to_judge(addr) if not ips: return False - if all(ip.is_global for ip in ips): - return True - return allow_loopback and all(ip.is_loopback for ip in ips) + return is_proxy_endpoint or all(ip.is_global for ip in ips) 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.""" + no usable host. Used to scope the internal-address allowance to that endpoint + alone.""" if not isinstance(proxy_url, str) or not proxy_url.strip(): return None candidate = proxy_url.strip() @@ -181,8 +180,8 @@ def _collect_proxy_endpoints(proxy_urls) -> set: # 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() +# Populated by install_socket_guard; empty means no internal destination is allowed. +_allowed_proxy_endpoints: set = set() def _normalise_port(port): @@ -197,18 +196,22 @@ def _normalise_port(port): return port -def _loopback_endpoint_allowed(host, port) -> bool: - if not _allowed_loopback_endpoints or host is None: +def _is_proxy_endpoint(host, port) -> bool: + """True when host:port is exactly an endpoint the operator configured as a + proxy. Matching is on the configured host *string*, not on the resolved + address, so a hostile media URL cannot borrow the allowance by resolving to + the same address under a different name.""" + if not _allowed_proxy_endpoints or host is None: return False - return (str(host).rstrip('.').lower(), _normalise_port(port)) in _allowed_loopback_endpoints + return (str(host).rstrip('.').lower(), _normalise_port(port)) in _allowed_proxy_endpoints def _guarded_getaddrinfo(host, *args, **kwargs): results = _real_getaddrinfo(host, *args, **kwargs) # 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)] + is_proxy = _is_proxy_endpoint(host, port) + allowed = [r for r in results if _address_allowed_at_connect(r[4][0], is_proxy)] if not allowed: raise socket.gaierror(f'Refusing to connect to non-global address for host {host!r}') return allowed @@ -227,9 +230,10 @@ def install_socket_guard(allow_private: bool = False, proxy_urls=()) -> None: 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, + the ``*_proxy`` environment variables are picked up automatically). A proxy is + reachable at its own host:port wherever it lives — loopback, the LAN, a VPN + range — and nothing else internal is. That costs proxied setups nothing and + gives away 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. @@ -239,12 +243,10 @@ def install_socket_guard(allow_private: bool = False, proxy_urls=()) -> 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}') + _allowed_proxy_endpoints.clear() + _allowed_proxy_endpoints.update(_collect_proxy_endpoints(proxy_urls)) + for host, port in sorted(_allowed_proxy_endpoints, key=lambda ep: (ep[0], ep[1] or 0)): + log.info(f'Allowing connections to configured proxy {host}:{port}') socket.getaddrinfo = _guarded_getaddrinfo diff --git a/app/ytdl.py b/app/ytdl.py index 77e758c..df26599 100644 --- a/app/ytdl.py +++ b/app/ytdl.py @@ -659,9 +659,9 @@ class Download: # Re-validate every outbound connection at fetch time. validate_url only # 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. + # which it can see. The configured proxy is passed so that a proxy on an + # internal address stays reachable at its own host:port without opening up + # anything else. 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: