fix: scope the connect-time loopback allowance to the configured proxy

The connect-time guard permitted every loopback address so that a locally
configured proxy (proxy: http://127.0.0.1:9050) stayed reachable. But the
connect guard is the only check that media URLs derived from remote metadata
ever face — validate_url sees just the submitted URL — so that blanket
allowance let a remote manifest steer the download subprocess at services
bound to the server's loopback interface, with the response written to the
download directory and served back by the UI.

Permit loopback only at the host:port of a proxy the operator configured,
taken from yt-dlp's proxy option and the *_proxy environment variables.
Nothing is lost: when a proxy is in use yt-dlp hands it the media URL instead
of resolving that URL locally, so the two cases never overlap. Every other
loopback destination now falls under the same is_global policy as the rest.

Reported by m3rl1nu5 (https://github.com/hai135) in GHSA-73g4-qhhq-c32c.

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

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):
+    def test_loopback_blocked_by_default(self):
-        # Loopback stays reachable so locally-configured proxies keep working.
+        # A blanket loopback allowance is what let manifest-derived media URLs
-        self.assertTrue(_address_allowed_at_connect("127.0.0.1"))
+        # reach services on the server's own loopback interface.
-        self.assertTrue(_address_allowed_at_connect("::1"))
+        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
+        install_socket_guard()
-        try:
+        self.assertIs(socket.getaddrinfo, url_guard._guarded_getaddrinfo)
-            install_socket_guard()
+        # Re-installing must not wrap the wrapper (real fn captured at import).
-            self.assertIs(socket.getaddrinfo, url_guard._guarded_getaddrinfo)
+        install_socket_guard()
-            # Re-installing must not wrap the wrapper (real fn captured at import).
+        self.assertIs(socket.getaddrinfo, url_guard._guarded_getaddrinfo)
-            install_socket_guard()
+
-            self.assertIs(socket.getaddrinfo, url_guard._guarded_getaddrinfo)
+    def test_no_proxy_means_no_loopback_allowance(self):
-        finally:
+        install_socket_guard()
-            socket.getaddrinfo = original
+        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__":

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
+    Permits global addresses only. Loopback is permitted just for the specific
-    proxies (e.g. ``proxy: http://127.0.0.1:9050``) keep working. Blocks the SSRF
+    host:port of an operator-configured proxy (see ``_loopback_endpoint_allowed``),
-    targets that matter: link-local (cloud metadata at 169.254.169.254), private
+    never as a blanket rule: media URLs that yt-dlp derives from a remote manifest
-    (RFC1918), unique-local and every other non-global, non-loopback range.
+    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
+    every resolved address at connect time, covering redirects, DNS rebinding and
-    for any networking backend that resolves through Python's socket module
+    manifest-derived media URLs for any networking backend that resolves through
-    (urllib, requests). Native resolvers — notably curl_cffi/libcurl used by
+    Python's socket module (urllib, requests). Native resolvers — notably
-    ``--impersonate`` — bypass this and rely on network isolation as the backstop.
+    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
 
 

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
+        # saw the submitted URL string; this catches redirects, DNS rebinding and
-        # to internal hosts (cloud metadata, RFC1918) that it cannot. Skipped when
+        # attacker-controlled media URLs pulled from a remote manifest, none of
-        # ALLOW_PRIVATE_ADDRESSES trusts the environment (e.g. Fake-IP proxies).
+        # which it can see. The configured proxy is passed so that a proxy on
-        install_socket_guard(self.allow_private)
+        # 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)