fix: let a configured proxy live on any internal address (closes #1055)

Scoping the connect-time allowance to the configured proxy kept the exception
tied to loopback, so a proxy anywhere else internal — the common case of a
socks5 or HTTP proxy on the LAN — was refused with "Refusing to connect to
non-global address". The only workaround was ALLOW_PRIVATE_ADDRESSES, which
switches the whole guard off, a far larger concession than the setup needs.

The allowance was never really about loopback: it is about the operator having
named this host:port as a proxy. Widen it to any address at a configured proxy
endpoint and nothing is given away, because the match is on the configured host
string rather than the resolved address — a hostile media URL that resolves to
the proxy's address under another name gets no allowance, and one that names the
proxy endpoint itself only reaches the proxy. Every other internal destination
stays refused.

Also log each configured proxy endpoint, so the next report of this shape can be
diagnosed from the log rather than from the guard's source.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>

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
+    """Connect-time policy: allow global, plus anything at a destination the
-    established that this destination is the operator's configured proxy."""
+    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("127.0.0.1", is_proxy_endpoint=True))
-        self.assertTrue(_address_allowed_at_connect("::1", allow_loopback=True))
+        self.assertTrue(_address_allowed_at_connect("::1", is_proxy_endpoint=True))
 
-    def test_opt_in_does_not_widen_beyond_loopback(self):
+    def test_proxy_opt_in_covers_any_internal_range(self):
-        self.assertFalse(_address_allowed_at_connect("169.254.169.254", allow_loopback=True))
+        # A proxy is just as legitimately on the LAN or a VPN range as on
-        self.assertFalse(_address_allowed_at_connect("10.0.0.5", allow_loopback=True))
+        # 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)
+        saved = set(url_guard._allowed_proxy_endpoints)
-        url_guard._allowed_loopback_endpoints = set()
+        url_guard._allowed_proxy_endpoints = set()
-        self.addCleanup(lambda: setattr(url_guard, "_allowed_loopback_endpoints", saved))
+        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__":

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
+    Permits global addresses, and anything at all when the destination is an
-    host:port of an operator-configured proxy (see ``_loopback_endpoint_allowed``),
+    operator-configured proxy (see ``_is_proxy_endpoint``). Internal addresses
-    never as a blanket rule: media URLs that yt-dlp derives from a remote manifest
+    are otherwise refused with no blanket exception: media URLs that yt-dlp
-    are attacker-controlled and reach this policy without passing ``validate_url``,
+    derives from a remote manifest are attacker-controlled and reach this policy
-    so a general loopback allowance would let a hostile playlist read any service
+    without passing ``validate_url``, so any range opened here is a range a
-    on the server's loopback interface. Blocks link-local (cloud metadata at
+    hostile playlist can read from the server's own network. Blocks link-local
-    169.254.169.254), private (RFC1918), unique-local and every other non-global
+    (cloud metadata at 169.254.169.254), private (RFC1918), loopback,
-    range.
+    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 is_proxy_endpoint or all(ip.is_global for ip in ips)
-        return True
-    return allow_loopback and all(ip.is_loopback 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.
+# Populated by install_socket_guard; empty means no internal destination is allowed.
-_allowed_loopback_endpoints: set = set()
+_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:
+def _is_proxy_endpoint(host, port) -> bool:
-    if not _allowed_loopback_endpoints or host is None:
+    """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)
+    is_proxy = _is_proxy_endpoint(host, port)
-    allowed = [r for r in results if _address_allowed_at_connect(r[4][0], allow_loopback)]
+    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
+    the ``*_proxy`` environment variables are picked up automatically). A proxy is
-    loopback is reachable at its own host:port, and nothing else on loopback is.
+    reachable at its own host:port wherever it lives — loopback, the LAN, a VPN
-    That costs proxied setups nothing: yt-dlp resolves the proxy itself at exactly that host:port,
+    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_proxy_endpoints.clear()
-    _allowed_loopback_endpoints.update(_collect_proxy_endpoints(proxy_urls))
+    _allowed_proxy_endpoints.update(_collect_proxy_endpoints(proxy_urls))
-    for host, port in sorted(_allowed_loopback_endpoints, key=lambda ep: (ep[0], ep[1] or 0)):
+    for host, port in sorted(_allowed_proxy_endpoints, key=lambda ep: (ep[0], ep[1] or 0)):
-        ip = _normalise_ip(host)
+        log.info(f'Allowing connections to configured proxy {host}:{port}')
-        if ip is not None and ip.is_loopback:
-            log.info(f'Allowing connections to configured loopback proxy {host}:{port}')
     socket.getaddrinfo = _guarded_getaddrinfo
 
 

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
+        # which it can see. The configured proxy is passed so that a proxy on an
-        # loopback stays reachable at its own address without opening up the rest
+        # internal address stays reachable at its own host:port without opening up
-        # of loopback. Skipped when ALLOW_PRIVATE_ADDRESSES trusts the environment.
+        # 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: