ngclient: Add proxy environment variable handling

urllib3 does not handle this but we do want to support proxy users.

The environment variable handling is slightly simplified from the
requests implementation.

Signed-off-by: Jussi Kukkonen <[email protected]>

Author: jku

tests/test_updater_ng.py

@@ -330,8 +330,10 @@ def test_non_existing_target_file(self) -> None:
     def test_user_agent(self) -> None:
         # test default
         self.updater.refresh()
-        session = self.updater._fetcher._poolManager
-        ua = session.headers["User-Agent"]
+        poolmgr = self.updater._fetcher._proxy_env.get_pool_manager(
+            "http", "localhost"
+        )
+        ua = poolmgr.headers["User-Agent"]
         self.assertEqual(ua[:11], "python-tuf/")
 
         # test custom UA
@@ -343,8 +345,10 @@ def test_user_agent(self) -> None:
             config=UpdaterConfig(app_user_agent="MyApp/1.2.3"),
         )
         updater.refresh()
-        session = updater._fetcher._poolManager
-        ua = session.headers["User-Agent"]
+        poolmgr = updater._fetcher._proxy_env.get_pool_manager(
+            "http", "localhost"
+        )
+        ua = poolmgr.headers["User-Agent"]
 
         self.assertEqual(ua[:23], "MyApp/1.2.3 python-tuf/")
 

tuf/ngclient/_internal/proxy.py

@@ -0,0 +1,96 @@
+# Copyright New York University and the TUF contributors
+# SPDX-License-Identifier: MIT OR Apache-2.0
+
+"""Proxy environment variable handling with Urllib3"""
+
+from typing import Any
+from urllib.request import getproxies
+
+from urllib3 import BaseHTTPResponse, PoolManager, ProxyManager
+from urllib3.util.url import parse_url
+
+
+# TODO: ProxyEnvironment could implement the whole PoolManager.RequestMethods
+# Mixin: We only need request() so nothing else is currently implemented
+class ProxyEnvironment:
+    """A PoolManager manager for automatic proxy handling based on env variables
+
+    Keeps track of PoolManagers for different proxy urls based on proxy
+    environment variables. Use `get_pool_manager()` or `request()` to access
+    the right manager for a scheme/host.
+
+    Supports '*_proxy' variables, with special handling for 'no_proxy' and
+    'all_proxy'.
+    """
+
+    def __init__(
+        self,
+        **kw_args: Any,  # noqa: ANN401
+    ) -> None:
+        self._pool_managers: dict[str | None, PoolManager] = {}
+        self._kw_args = kw_args
+
+        self._proxies = getproxies()
+        self._all_proxy = self._proxies.pop("all", None)
+        no_proxy = self._proxies.pop("no", None)
+        if no_proxy is None:
+            self._no_proxy_hosts = []
+        else:
+            self._no_proxy_hosts = [
+                h for h in no_proxy.replace(" ", "").split(",") if h
+            ]
+
+    def _get_proxy(self, scheme: str | None, host: str | None) -> str | None:
+        """Get a proxy url for scheme and host based on proxy env variables"""
+
+        if host is None:
+            # urllib3 only handles http/https but we can do something reasonable
+            # even for schemes that don't require host (like file)
+            return None
+
+        # does host match "no_proxy" hosts?
+        for no_proxy_host in self._no_proxy_hosts:
+            # exact hostname match or parent domain match
+            if host == no_proxy_host or host.endswith(f".{no_proxy_host}"):
+                return None
+
+        if scheme in self._proxies:
+            return self._proxies[scheme]
+        if self._all_proxy is not None:
+            return self._all_proxy
+
+        return None
+
+    def get_pool_manager(
+        self, scheme: str | None, host: str | None
+    ) -> PoolManager:
+        """Get a poolmanager for scheme and host.
+
+        Returns a ProxyManager if that is correct based on current proxy env
+        variables, otherwise returns a PoolManager
+        """
+
+        proxy = self._get_proxy(scheme, host)
+        if proxy not in self._pool_managers:
+            if proxy is None:
+                self._pool_managers[proxy] = PoolManager(**self._kw_args)
+            else:
+                self._pool_managers[proxy] = ProxyManager(
+                    proxy,
+                    **self._kw_args,
+                )
+
+        return self._pool_managers[proxy]
+
+    def request(
+        self,
+        method: str,
+        url: str,
+        **request_kw: Any,  # noqa: ANN401
+    ) -> BaseHTTPResponse:
+        """Make a request using a PoolManager chosen based on url and
+        proxy environment variables.
+        """
+        u = parse_url(url)
+        manager = self.get_pool_manager(u.scheme, u.host)
+        return manager.request(method, url, **request_kw)

tuf/ngclient/urllib3_fetcher.py

@@ -15,6 +15,7 @@
 
 import tuf
 from tuf.api import exceptions
+from tuf.ngclient._internal.proxy import ProxyEnvironment
 from tuf.ngclient.fetcher import FetcherInterface
 
 if TYPE_CHECKING:
@@ -49,7 +50,7 @@ def __init__(
         if app_user_agent is not None:
             ua = f"{app_user_agent} {ua}"
 
-        self._poolManager = urllib3.PoolManager(headers={"User-Agent": ua})
+        self._proxy_env = ProxyEnvironment(headers={"User-Agent": ua})
 
     def _fetch(self, url: str) -> Iterator[bytes]:
         """Fetch the contents of HTTP/HTTPS url from a remote server.
@@ -72,7 +73,7 @@ def _fetch(self, url: str) -> Iterator[bytes]:
         #  - connect timeout (max delay before first byte is received)
         #  - read (gap) timeout (max delay between bytes received)
         try:
-            response = self._poolManager.request(
+            response = self._proxy_env.request(
                 "GET",
                 url,
                 preload_content=False,