fix: let the download reach the PO token provider (closes #1064)

The image ships yt-dlp's bgutil PO token provider and starts it on
loopback, where the plugin dials it at http://127.0.0.1:4416. Since
482381d scoped the connect-time allowance to the configured proxy, the
download subprocess could no longer resolve it:

    Refusing to connect to non-global address for host '127.0.0.1'

which surfaces as the plugin's "Error reaching GET .../ping". Metadata
extraction runs in the main process and installs no guard, so titles kept
resolving while the download itself ran without a token — and YouTube
increasingly answers those with 403.

The allowance already had the right shape for this; it was just named for
its only user. Endpoints the operator or the image configured are now
allowed as a class: install_socket_guard takes service_urls alongside
proxy_urls, and ytdl derives them from the bundled default plus any
base_url set through the youtubepot-bgutilhttp (or the deprecated youtube
getpot_bgutil_baseurl) extractor argument. The bundled server runs either
way, so it stays allowed when a base URL is configured.

Matching stays exact host:port on the configured string, so nothing else
on loopback opens up: a hostile media URL naming the endpoint reaches a
token server with two endpoints and nothing worth reading.

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

app/tests/test_url_guard.py

@@ -12,7 +12,7 @@ from url_guard import (
     _address_allowed_at_connect,
     _address_is_global,
     _guarded_getaddrinfo,
-    _proxy_endpoint,
+    _url_endpoint,
     install_socket_guard,
 )
 
@@ -121,19 +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", is_proxy_endpoint=True))
+        self.assertTrue(_address_allowed_at_connect("127.0.0.1", is_allowed_endpoint=True))
-        self.assertTrue(_address_allowed_at_connect("::1", is_proxy_endpoint=True))
+        self.assertTrue(_address_allowed_at_connect("::1", is_allowed_endpoint=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("10.1.20.30", is_allowed_endpoint=True))
-        self.assertTrue(_address_allowed_at_connect("192.168.1.10", is_proxy_endpoint=True))
+        self.assertTrue(_address_allowed_at_connect("192.168.1.10", is_allowed_endpoint=True))
-        self.assertTrue(_address_allowed_at_connect("fd00::1", is_proxy_endpoint=True))
+        self.assertTrue(_address_allowed_at_connect("fd00::1", is_allowed_endpoint=True))
 
     def test_opt_in_still_rejects_non_addresses(self):
-        self.assertFalse(_address_allowed_at_connect("not-an-ip", is_proxy_endpoint=True))
+        self.assertFalse(_address_allowed_at_connect("not-an-ip", is_allowed_endpoint=True))
 
     def test_link_local_metadata_blocked(self):
         self.assertFalse(_address_allowed_at_connect("169.254.169.254"))
@@ -187,36 +187,36 @@ class TunnelledIPv4Tests(unittest.TestCase):
             self.assertIsNotNone(validate_url("http://nat64.example/x"))
 
 
-class ProxyEndpointParsingTests(unittest.TestCase):
+class EndpointParsingTests(unittest.TestCase):
     def test_explicit_port(self):
-        self.assertEqual(_proxy_endpoint("http://127.0.0.1:9050"), ("127.0.0.1", 9050))
+        self.assertEqual(_url_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(_url_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))
+        self.assertEqual(_url_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))
+        self.assertEqual(_url_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))
+        self.assertEqual(_url_endpoint("http://LocalHost.:9050"), ("localhost", 9050))
 
     def test_ipv6_literal(self):
-        self.assertEqual(_proxy_endpoint("http://[::1]:9050"), ("::1", 9050))
+        self.assertEqual(_url_endpoint("http://[::1]:9050"), ("::1", 9050))
 
     def test_empty_and_invalid(self):
-        self.assertIsNone(_proxy_endpoint(""))
+        self.assertIsNone(_url_endpoint(""))
-        self.assertIsNone(_proxy_endpoint("   "))
+        self.assertIsNone(_url_endpoint("   "))
-        self.assertIsNone(_proxy_endpoint(None))
+        self.assertIsNone(_url_endpoint(None))
-        self.assertIsNone(_proxy_endpoint("http://"))
+        self.assertIsNone(_url_endpoint("http://"))
 
 
 class GuardedGetaddrinfoTests(unittest.TestCase):
     def setUp(self):
         # Default state: no proxy configured, so no loopback destination allowed.
-        saved = set(url_guard._allowed_proxy_endpoints)
+        saved = set(url_guard._allowed_endpoints)
-        url_guard._allowed_proxy_endpoints = set()
+        url_guard._allowed_endpoints = set()
-        self.addCleanup(lambda: setattr(url_guard, "_allowed_proxy_endpoints", saved))
+        self.addCleanup(lambda: setattr(url_guard, "_allowed_endpoints", saved))
 
     def test_internal_only_raises(self):
         with mock.patch("url_guard._real_getaddrinfo", return_value=_addrinfo("169.254.169.254")):
@@ -235,27 +235,27 @@ class GuardedGetaddrinfoTests(unittest.TestCase):
             with self.assertRaises(socket.gaierror):
                 _guarded_getaddrinfo("127.0.0.1", 9999)
 
-    def test_loopback_allowed_at_configured_proxy_endpoint(self):
+    def test_loopback_allowed_at_configured_url_endpoint(self):
-        url_guard._allowed_proxy_endpoints = {("127.0.0.1", 9050)}
+        url_guard._allowed_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_proxy_endpoints = {("127.0.0.1", 9050)}
+        url_guard._allowed_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_proxy_endpoints = {("localhost", 9050)}
+        url_guard._allowed_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_proxy_endpoints = {("127.0.0.1", 9050)}
+        url_guard._allowed_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"])
@@ -263,22 +263,38 @@ class GuardedGetaddrinfoTests(unittest.TestCase):
     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)}
+        url_guard._allowed_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)}
+        url_guard._allowed_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_pot_provider_reachable_on_loopback(self):
+        # #1064: the bundled PO token provider listens on loopback, and blocking
+        # it left every default install downloading YouTube without a token.
+        url_guard._allowed_endpoints = {("127.0.0.1", 4416)}
+        with mock.patch("url_guard._real_getaddrinfo", return_value=_addrinfo("127.0.0.1")):
+            results = _guarded_getaddrinfo("127.0.0.1", 4416)
+        self.assertEqual([r[4][0] for r in results], ["127.0.0.1"])
+
+    def test_other_loopback_service_still_blocked(self):
+        # MeTube's own port is one hop away on the same interface: allowing the
+        # token provider must not allow the rest of loopback.
+        url_guard._allowed_endpoints = {("127.0.0.1", 4416)}
+        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", 8081)
+
     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)}
+        url_guard._allowed_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)
@@ -312,9 +328,9 @@ class AllowPrivateBypassTests(unittest.TestCase):
 
 class InstallSocketGuardTests(unittest.TestCase):
     def setUp(self):
-        original, saved = socket.getaddrinfo, set(url_guard._allowed_proxy_endpoints)
+        original, saved = socket.getaddrinfo, set(url_guard._allowed_endpoints)
         self.addCleanup(lambda: setattr(socket, "getaddrinfo", original))
-        self.addCleanup(lambda: setattr(url_guard, "_allowed_proxy_endpoints", saved))
+        self.addCleanup(lambda: setattr(url_guard, "_allowed_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()
@@ -329,26 +345,45 @@ class InstallSocketGuardTests(unittest.TestCase):
 
     def test_no_proxy_means_no_loopback_allowance(self):
         install_socket_guard()
-        self.assertEqual(url_guard._allowed_proxy_endpoints, set())
+        self.assertEqual(url_guard._allowed_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_proxy_endpoints, {("127.0.0.1", 9050)})
+        self.assertEqual(url_guard._allowed_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_proxy_endpoints, set())
+        self.assertEqual(url_guard._allowed_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_proxy_endpoints, {("127.0.0.1", 8080)})
+        self.assertEqual(url_guard._allowed_endpoints, {("127.0.0.1", 8080)})
+
+    def test_service_url_is_registered(self):
+        install_socket_guard(service_urls=("http://127.0.0.1:4416",))
+        self.assertEqual(url_guard._allowed_endpoints, {("127.0.0.1", 4416)})
+
+    def test_service_and_proxy_endpoints_coexist(self):
+        install_socket_guard(
+            proxy_urls=("socks5://10.1.20.30:1080",),
+            service_urls=("http://127.0.0.1:4416",),
+        )
+        self.assertEqual(
+            url_guard._allowed_endpoints,
+            {("10.1.20.30", 1080), ("127.0.0.1", 4416)},
+        )
+
+    def test_service_urls_reset_between_installs(self):
+        install_socket_guard(service_urls=("http://127.0.0.1:4416",))
+        install_socket_guard()
+        self.assertEqual(url_guard._allowed_endpoints, set())
 
     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_proxy_endpoints, set())
+        self.assertEqual(url_guard._allowed_endpoints, set())
 
 
 if __name__ == "__main__":

app/tests/test_ytdl_utils.py

@@ -77,6 +77,7 @@ from ytdl import (
     MusicMetadataPreProcessor,
     _compact_persisted_entry,
     _convert_srt_to_txt_file,
+    _pot_provider_urls,
     _AlbumArtistPostProcessor,
     _resolve_outtmpl_fields,
     _sanitize_entry_for_pickle,
@@ -1071,5 +1072,44 @@ class ShortTitleForFailedUrlTests(unittest.TestCase):
         self.assertEqual(_short_title_for_failed_url(malformed), malformed)
 
 
+class PotProviderUrlsTests(unittest.TestCase):
+    """#1064: the connect-time guard must let the download reach the PO token
+    provider, so it has to know every endpoint yt-dlp might dial for one."""
+
+    def test_bundled_provider_by_default(self):
+        self.assertEqual(_pot_provider_urls({}), ("http://127.0.0.1:4416",))
+
+    def test_configured_base_url_is_added(self):
+        urls = _pot_provider_urls({
+            "extractor_args": {"youtubepot-bgutilhttp": {"base_url": ["http://pot:4416"]}},
+        })
+        # The bundled server runs regardless, so both stay reachable.
+        self.assertEqual(urls, ("http://127.0.0.1:4416", "http://pot:4416"))
+
+    def test_deprecated_base_url_arg_is_honoured(self):
+        urls = _pot_provider_urls({
+            "extractor_args": {"youtube": {"getpot_bgutil_baseurl": ["http://pot:4416"]}},
+        })
+        self.assertEqual(urls, ("http://127.0.0.1:4416", "http://pot:4416"))
+
+    def test_unrelated_extractor_args_are_ignored(self):
+        urls = _pot_provider_urls({
+            "extractor_args": {"youtube": {"player_client": ["web"]}},
+        })
+        self.assertEqual(urls, ("http://127.0.0.1:4416",))
+
+    def test_malformed_extractor_args_do_not_raise(self):
+        # YTDL_OPTIONS is operator-supplied JSON and reaches here unvalidated.
+        for opts in (
+            {"extractor_args": None},
+            {"extractor_args": "youtube:player_client=web"},
+            {"extractor_args": {"youtubepot-bgutilhttp": "http://pot:4416"}},
+            {"extractor_args": {"youtubepot-bgutilhttp": {"base_url": []}}},
+        ):
+            with self.subTest(opts=opts):
+                self.assertEqual(_pot_provider_urls(opts), ("http://127.0.0.1:4416",))
+
+
 if __name__ == "__main__":
+
     unittest.main()

app/url_guard.py

@@ -37,8 +37,8 @@ log = logging.getLogger('url_guard')
 
 _ALLOWED_SCHEMES = ('http', 'https')
 
-# Ports to assume when a configured proxy URL omits one, per proxy scheme.
+# Ports to assume when a configured endpoint URL omits one, per scheme.
-_PROXY_DEFAULT_PORTS = {
+_SCHEME_DEFAULT_PORTS = {
     'http': 80,
     'https': 443,
     'socks4': 1080,
@@ -129,31 +129,32 @@ 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, is_proxy_endpoint: bool = False) -> bool:
+def _address_allowed_at_connect(addr: str, is_allowed_endpoint: bool = False) -> bool:
     """True if *addr* may be connected to at download time.
 
     Permits global addresses, and anything at all when the destination is an
-    operator-configured proxy (see ``_is_proxy_endpoint``). Internal addresses
+    endpoint the operator or the image configured — a proxy, or the PO token
-    are otherwise refused with no blanket exception: media URLs that yt-dlp
+    provider (see ``_is_allowed_endpoint``). Internal addresses are otherwise
-    derives from a remote manifest are attacker-controlled and reach this policy
+    refused with no blanket exception: media URLs that yt-dlp derives from a
-    without passing ``validate_url``, so any range opened here is a range a
+    remote manifest are attacker-controlled and reach this policy without passing
-    hostile playlist can read from the server's own network. Blocks link-local
+    ``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
-    return is_proxy_endpoint or all(ip.is_global for ip in ips)
+    return is_allowed_endpoint or all(ip.is_global for ip in ips)
 
 
-def _proxy_endpoint(proxy_url: str):
+def _url_endpoint(url: str):
-    """Parse a proxy URL into a ``(hostname, port)`` pair, or ``None`` if it has
+    """Parse a configured URL into a ``(hostname, port)`` pair, or ``None`` if it
-    no usable host. Used to scope the internal-address allowance to that endpoint
+    has no usable host. Used to scope the internal-address allowance to that
-    alone."""
+    endpoint alone."""
-    if not isinstance(proxy_url, str) or not proxy_url.strip():
+    if not isinstance(url, str) or not url.strip():
         return None
-    candidate = proxy_url.strip()
+    candidate = url.strip()
     if '://' not in candidate:
         # Bare host:port, as accepted by the *_proxy environment variables.
         candidate = '//' + candidate
@@ -165,23 +166,28 @@ def _proxy_endpoint(proxy_url: str):
     if not hostname:
         return None
     if port is None:
-        port = _PROXY_DEFAULT_PORTS.get(parts.scheme.lower())
+        port = _SCHEME_DEFAULT_PORTS.get(parts.scheme.lower())
     return (hostname.rstrip('.').lower(), port)
 
 
+def _endpoints(urls) -> set:
+    """The parseable endpoints among *urls*, dropping any that name no host."""
+    return {ep for ep in map(_url_endpoint, urls) if ep is not None}
+
+
 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}
+    return _endpoints(candidates)
 
 
 # Captured at import so re-installing the guard never wraps the wrapper.
 _real_getaddrinfo = socket.getaddrinfo
 
 # Populated by install_socket_guard; empty means no internal destination is allowed.
-_allowed_proxy_endpoints: set = set()
+_allowed_endpoints: set = set()
 
 
 def _normalise_port(port):
@@ -196,28 +202,29 @@ def _normalise_port(port):
     return port
 
 
-def _is_proxy_endpoint(host, port) -> bool:
+def _is_allowed_endpoint(host, port) -> bool:
-    """True when host:port is exactly an endpoint the operator configured as a
+    """True when host:port is exactly one of the endpoints this download is
-    proxy. Matching is on the configured host *string*, not on the resolved
+    configured to dial — a proxy or the PO token provider. Matching is on the
-    address, so a hostile media URL cannot borrow the allowance by resolving to
+    configured host *string*, not on the resolved address, so a hostile media URL
-    the same address under a different name."""
+    cannot borrow the allowance by resolving to the same address under a
-    if not _allowed_proxy_endpoints or host is None:
+    different name."""
+    if not _allowed_endpoints or host is None:
         return False
-    return (str(host).rstrip('.').lower(), _normalise_port(port)) in _allowed_proxy_endpoints
+    return (str(host).rstrip('.').lower(), _normalise_port(port)) in _allowed_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')
-    is_proxy = _is_proxy_endpoint(host, port)
+    is_configured = _is_allowed_endpoint(host, port)
-    allowed = [r for r in results if _address_allowed_at_connect(r[4][0], is_proxy)]
+    allowed = [r for r in results if _address_allowed_at_connect(r[4][0], is_configured)]
     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, proxy_urls=()) -> None:
+def install_socket_guard(allow_private: bool = False, proxy_urls=(), service_urls=()) -> None:
     """Enforce the no-internal-hosts policy at actual connection time.
 
     ``validate_url`` only checks the *submitted* URL string; yt-dlp then follows
@@ -230,12 +237,16 @@ 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 is
+    the ``*_proxy`` environment variables are picked up automatically), and
-    reachable at its own host:port wherever it lives — loopback, the LAN, a VPN
+    *service_urls* the helper services the download itself has to reach — the PO
-    range — and nothing else internal is. That costs proxied setups nothing and
+    token provider this image ships and starts on loopback. Each is reachable at
-    gives away nothing: yt-dlp resolves the proxy itself at exactly that host:port,
+    its own host:port wherever it lives — loopback, the LAN, a VPN range — and
-    and a media URL is either handed to the proxy unresolved or resolved on its own
+    nothing else internal is. That costs those setups nothing and gives away
-    merits — never inheriting the proxy's allowance.
+    little: yt-dlp dials each 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 allowance. A hostile media URL naming an allowed endpoint
+    reaches only what is listening there: a proxy that would have fetched it
+    anyway, or a token server with two endpoints and nothing to read.
 
     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
@@ -243,10 +254,13 @@ def install_socket_guard(allow_private: bool = False, proxy_urls=()) -> None:
     """
     if allow_private:
         return
-    _allowed_proxy_endpoints.clear()
+    proxy_endpoints = _collect_proxy_endpoints(proxy_urls)
-    _allowed_proxy_endpoints.update(_collect_proxy_endpoints(proxy_urls))
+    service_endpoints = _endpoints(service_urls) - proxy_endpoints
-    for host, port in sorted(_allowed_proxy_endpoints, key=lambda ep: (ep[0], ep[1] or 0)):
+    _allowed_endpoints.clear()
-        log.info(f'Allowing connections to configured proxy {host}:{port}')
+    _allowed_endpoints.update(proxy_endpoints | service_endpoints)
+    for label, endpoints in (('proxy', proxy_endpoints), ('service', service_endpoints)):
+        for host, port in sorted(endpoints, key=lambda ep: (ep[0], ep[1] or 0)):
+            log.info(f'Allowing connections to configured {label} {host}:{port}')
     socket.getaddrinfo = _guarded_getaddrinfo
 
 

app/ytdl.py

@@ -92,6 +92,36 @@ class _DownloadYtdlLogger:
 # vanish in the child can deadlock it silently before it does any work. This
 # app creates background threads (executors, notifier callbacks) well before
 # any download starts, so forcing fork there reproduces exactly that hazard.
+# The image ships yt-dlp's bgutil PO token provider and starts it on loopback
+# (docker-entrypoint.sh); the plugin dials this URL unless pointed elsewhere.
+# Without a token YouTube serves 403s, so the connect-time guard has to let the
+# download subprocess reach it.
+_POT_PROVIDER_DEFAULT_URL = 'http://127.0.0.1:4416'
+
+# extractor-arg keys the bgutil HTTP provider reads its base URL from: the
+# current one first, then the deprecated form it still honours.
+_POT_PROVIDER_BASE_URL_ARGS = (
+    ('youtubepot-bgutilhttp', 'base_url'),
+    ('youtube', 'getpot_bgutil_baseurl'),
+)
+
+
+def _pot_provider_urls(ytdl_opts: dict) -> tuple:
+    """Every PO token provider endpoint this download may dial: the bundled one,
+    plus any the operator pointed yt-dlp at through ``extractor_args``. The
+    bundled server runs either way, so it stays allowed even when a base URL is
+    configured."""
+    urls = [_POT_PROVIDER_DEFAULT_URL]
+    extractor_args = ytdl_opts.get('extractor_args')
+    if isinstance(extractor_args, dict):
+        for ie_key, arg in _POT_PROVIDER_BASE_URL_ARGS:
+            section = extractor_args.get(ie_key)
+            values = section.get(arg) if isinstance(section, dict) else None
+            if values:
+                urls.append(values[0])
+    return tuple(urls)
+
+
 _MP_CTX = (
     multiprocessing.get_context("fork")
     if sys.platform.startswith("linux") and "fork" in multiprocessing.get_all_start_methods()
@@ -764,10 +794,15 @@ 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 an
+        # which it can see. The configured proxy and the PO token provider are
-        # internal address stays reachable at its own host:port without opening up
+        # passed so that each stays reachable at its own host:port without opening
-        # anything else. Skipped when ALLOW_PRIVATE_ADDRESSES trusts the environment.
+        # up anything else. Skipped when ALLOW_PRIVATE_ADDRESSES trusts the
-        install_socket_guard(self.allow_private, proxy_urls=(self.ytdl_opts.get('proxy'),))
+        # environment.
+        install_socket_guard(
+            self.allow_private,
+            proxy_urls=(self.ytdl_opts.get('proxy'),),
+            service_urls=_pot_provider_urls(self.ytdl_opts),
+        )
         log.info(f"Starting download for: {self.info.title} ({self.info.url})")
         # Bound outside the try so the except branch can read what was captured
         # before the error was raised.