Construct Proxy-Authorization header from https://username:password

changelog/1999.bugfix.rst

@@ -0,0 +1 @@
+Added ability to specify ``Proxy-Authorization`` headers from the url

dummyserver/testcase.py

@@ -201,6 +201,7 @@ class HypercornDummyProxyTestCase:
     proxy_url: typing.ClassVar[str]
     https_proxy_port: typing.ClassVar[int]
     https_proxy_url: typing.ClassVar[str]
+    proxy_url_with_auth: typing.ClassVar[str]
 
     certs_dir: typing.ClassVar[str] = ""
     bad_ca_path: typing.ClassVar[str] = ""

src/urllib3/poolmanager.py

@@ -7,11 +7,14 @@
 from types import TracebackType
 from urllib.parse import urljoin
 
+from urllib3.util.request import make_headers
+
 from ._collections import HTTPHeaderDict, RecentlyUsedContainer
 from ._request_methods import RequestMethods
 from .connection import ProxyConfig
 from .connectionpool import HTTPConnectionPool, HTTPSConnectionPool, port_by_scheme
 from .exceptions import (
+    InvalidHeader,
     LocationValueError,
     MaxRetryError,
     ProxySchemeUnknown,
@@ -550,7 +553,7 @@ def __init__(
         proxy_url: str,
         num_pools: int = 10,
         headers: typing.Mapping[str, str] | None = None,
-        proxy_headers: typing.Mapping[str, str] | None = None,
+        proxy_headers: typing.MutableMapping[str, str] | None = None,
         proxy_ssl_context: ssl.SSLContext | None = None,
         use_forwarding_for_https: bool = False,
         proxy_assert_hostname: None | str | Literal[False] = None,
@@ -580,6 +583,23 @@ def __init__(
             proxy_assert_fingerprint,
         )
 
+        if proxy.auth is not None:
+            split = proxy.auth.split(":")
+            if len(split) == 2:
+                auth = make_headers(proxy_basic_auth=proxy.auth)["proxy-authorization"]
+                if (
+                    "proxy-authorization" in self.proxy_headers
+                    and self.proxy_headers["proxy-authorization"] != auth
+                ) or (
+                    headers
+                    and "proxy-authorization" in headers
+                    and headers["proxy-authorization"] != auth
+                ):
+                    raise InvalidHeader(
+                        "Proxy-Authorization given in headers or proxy_headers and URL don't match"
+                    )
+                self.proxy_headers["proxy-authorization"] = auth
+
         connection_pool_kw["_proxy"] = self.proxy
         connection_pool_kw["_proxy_headers"] = self.proxy_headers
         connection_pool_kw["_proxy_config"] = self.proxy_config

test/with_dummyserver/test_proxy_poolmanager.py

@@ -29,6 +29,7 @@
 from urllib3.exceptions import (
     ConnectTimeoutError,
     InsecureRequestWarning,
+    InvalidHeader,
     MaxRetryError,
     ProxyError,
     ProxySchemeUnknown,
@@ -37,6 +38,7 @@
     SSLError,
 )
 from urllib3.poolmanager import ProxyManager, proxy_from_url
+from urllib3.util.request import make_headers
 from urllib3.util.ssl_ import create_urllib3_context
 from urllib3.util.timeout import Timeout
 
@@ -64,6 +66,9 @@ def setup_class(cls) -> None:
         cls.https_url_fqdn = f"https://{cls.https_host}.:{int(cls.https_port)}"
         cls.proxy_url = f"http://{cls.proxy_host}:{int(cls.proxy_port)}"
         cls.https_proxy_url = f"https://{cls.proxy_host}:{int(cls.https_proxy_port)}"
+        cls.proxy_url_with_auth = (
+            f"http://user:password@{cls.proxy_host}:{int(cls.proxy_port)}"
+        )
 
         # Generate another CA to test verification failure
         cls.certs_dir = tempfile.mkdtemp()
@@ -320,6 +325,29 @@ def test_cross_protocol_redirect(self) -> None:
             assert r._pool is not None
             assert r._pool.host == self.https_host
 
+    def test_proxy_url_auth(self) -> None:
+        with proxy_from_url(self.proxy_url_with_auth) as http:
+            r = http.request_encode_url("GET", f"{self.http_url}/headers")
+            returned_headers = r.json()
+            assert (
+                returned_headers.get("Proxy-Authorization")
+                == "Basic dXNlcjpwYXNzd29yZA=="  # base64 encoded "user:password"
+            )
+        with pytest.raises(InvalidHeader):
+            proxy_from_url(
+                self.proxy_url_with_auth,
+                proxy_headers=make_headers(
+                    proxy_basic_auth="differentuser:differentpassword"
+                ),
+            )
+        with pytest.raises(InvalidHeader):
+            proxy_from_url(
+                self.proxy_url_with_auth,
+                headers=make_headers(
+                    proxy_basic_auth="differentuser:differentpassword"
+                ),
+            )
+
     def test_headers(self) -> None:
         with proxy_from_url(
             self.proxy_url,