Shortcuts on this page
r m x toggle line displays
j k next/prev highlighted chunk
0 (zero) top of page
1 (one) first highlighted chunk
1# Copyright New York University and the TUF contributors
2# SPDX-License-Identifier: MIT OR Apache-2.0
4"""Proxy environment variable handling with Urllib3"""
6from __future__ import annotations
8from typing import Any
9from urllib.request import getproxies
11from urllib3 import BaseHTTPResponse, PoolManager, ProxyManager
12from urllib3.util.url import parse_url
15# TODO: ProxyEnvironment could implement the whole PoolManager.RequestMethods
16# Mixin: We only need request() so nothing else is currently implemented
17class ProxyEnvironment:
18 """A PoolManager manager for automatic proxy handling based on env variables
20 Keeps track of PoolManagers for different proxy urls based on proxy
21 environment variables. Use `get_pool_manager()` or `request()` to access
22 the right manager for a scheme/host.
24 Supports '*_proxy' variables, with special handling for 'no_proxy' and
25 'all_proxy'.
26 """
28 def __init__(
29 self,
30 **kw_args: Any, # noqa: ANN401
31 ) -> None:
32 self._pool_managers: dict[str | None, PoolManager] = {}
33 self._kw_args = kw_args
35 self._proxies = getproxies()
36 self._all_proxy = self._proxies.pop("all", None)
37 no_proxy = self._proxies.pop("no", None)
38 if no_proxy is None:
39 self._no_proxy_hosts = []
40 else:
41 # split by comma, remove leading periods
42 self._no_proxy_hosts = [
43 h.lstrip(".") for h in no_proxy.replace(" ", "").split(",") if h
44 ]
46 def _get_proxy(self, scheme: str | None, host: str | None) -> str | None:
47 """Get a proxy url for scheme and host based on proxy env variables"""
49 if host is None:
50 # urllib3 only handles http/https but we can do something reasonable
51 # even for schemes that don't require host (like file)
52 return None
54 # does host match any of the "no_proxy" hosts?
55 for no_proxy_host in self._no_proxy_hosts:
56 # wildcard match, exact hostname match, or parent domain match
57 if no_proxy_host in ("*", host) or host.endswith(
58 f".{no_proxy_host}"
59 ):
60 return None
62 if scheme in self._proxies:
63 return self._proxies[scheme]
64 if self._all_proxy is not None:
65 return self._all_proxy
67 return None
69 def get_pool_manager(
70 self, scheme: str | None, host: str | None
71 ) -> PoolManager:
72 """Get a poolmanager for scheme and host.
74 Returns a ProxyManager if that is correct based on current proxy env
75 variables, otherwise returns a PoolManager
76 """
78 proxy = self._get_proxy(scheme, host)
79 if proxy not in self._pool_managers:
80 if proxy is None:
81 self._pool_managers[proxy] = PoolManager(**self._kw_args)
82 else:
83 self._pool_managers[proxy] = ProxyManager(
84 proxy,
85 **self._kw_args,
86 )
88 return self._pool_managers[proxy]
90 def request(
91 self,
92 method: str,
93 url: str,
94 **request_kw: Any, # noqa: ANN401
95 ) -> BaseHTTPResponse:
96 """Make a request using a PoolManager chosen based on url and
97 proxy environment variables.
98 """
99 u = parse_url(url)
100 manager = self.get_pool_manager(u.scheme, u.host)
101 return manager.request(method, url, **request_kw)