Add proxy_assert_hostname and proxy_assert_fingerprint to ProxyManager
ADD
changelog/2409.feature.rst
+Added ``proxy_assert_hostname`` and ``proxy_assert_fingerprint`` kwargs to ``ProxyManager``.

MODIFY
dummyserver/certs/README.rst
dummyserver/certs/README.rst
@@ -6,7 +6,7 @@ Here's how you can regenerate the certificates::
import trustme
ca = trustme.CA()
- server_cert = ca.issue_cert(u"localhost")
+ server_cert = ca.issue_cert("localhost")
ca.cert_pem.write_to_path("cacert.pem")
ca.private_key_pem.write_to_path("cacert.key")

MODIFY
src/urllib3/connection.py
src/urllib3/connection.py
@@ -84,6 +84,8 @@ _TYPE_BODY = Union[bytes, IO[Any], Iterable[bytes], str]
class ProxyConfig(NamedTuple):
ssl_context: Optional["ssl.SSLContext"]
use_forwarding_for_https: bool
+ assert_hostname: Union[None, str, "Literal[False]"]
+ assert_fingerprint: Optional[str]
class HTTPConnection(_HTTPConnection):
@@ -541,9 +543,9 @@ class HTTPSConnection(HTTPConnection):
ca_cert_data=self.ca_cert_data,
server_hostname=hostname,
ssl_context=ssl_context,
+ assert_hostname=proxy_config.assert_hostname,
+ assert_fingerprint=proxy_config.assert_fingerprint,
# Features that aren't implemented for proxies yet:
- assert_fingerprint=None,
- assert_hostname=None,
cert_file=None,
key_file=None,
key_password=None,

MODIFY
src/urllib3/poolmanager.py
src/urllib3/poolmanager.py
@@ -511,6 +511,12 @@ class ProxyManager(PoolManager):
private. IP address, target hostname, SNI, and port are always visible
to an HTTPS proxy even when this flag is disabled.
+ :param proxy_assert_hostname:
+ The hostname of the certificate to verify against.
+
+ :param proxy_assert_fingerprint:
+ The fingerprint of the certificate to verify against.
+
Example:
.. code-block:: python
@@ -541,6 +547,8 @@ class ProxyManager(PoolManager):
proxy_headers: Optional[Mapping[str, str]] = None,
proxy_ssl_context: Optional["ssl.SSLContext"] = None,
use_forwarding_for_https: bool = False,
+ proxy_assert_hostname: Union[None, str, "Literal[False]"] = None,
+ proxy_assert_fingerprint: Optional[str] = None,
**connection_pool_kw: Any,
) -> None:
@@ -560,7 +568,12 @@ class ProxyManager(PoolManager):
self.proxy = proxy
self.proxy_headers = proxy_headers or {}
self.proxy_ssl_context = proxy_ssl_context
- self.proxy_config = ProxyConfig(proxy_ssl_context, use_forwarding_for_https)
+ self.proxy_config = ProxyConfig(
+ proxy_ssl_context,
+ use_forwarding_for_https,
+ proxy_assert_hostname,
+ proxy_assert_fingerprint,
+ )
connection_pool_kw["_proxy"] = self.proxy
connection_pool_kw["_proxy_headers"] = self.proxy_headers

MODIFY
test/conftest.py
test/conftest.py
@@ -3,7 +3,6 @@ import platform
import socket
import ssl
import sys
-import threading
from pathlib import Path
from typing import AbstractSet, Any, Dict, Generator, NamedTuple, Optional, Tuple
@@ -13,7 +12,7 @@ from tornado import ioloop, web
from dummyserver.handlers import TestingApp
from dummyserver.proxy import ProxyHandler
-from dummyserver.server import HAS_IPV6, run_tornado_app
+from dummyserver.server import HAS_IPV6, run_loop_in_thread, run_tornado_app
from dummyserver.testcase import HTTPSDummyServerTestCase
from urllib3.util import ssl_
@@ -59,18 +58,13 @@ def run_server_in_thread(
scheme: str, host: str, tmpdir: Path, ca: trustme.CA, server_cert: trustme.LeafCert
) -> Generator[ServerConfig, None, None]:
ca_cert_path = str(tmpdir / "ca.pem")
- server_cert_path = str(tmpdir / "server.pem")
- server_key_path = str(tmpdir / "server.key")
ca.cert_pem.write_to_path(ca_cert_path)
- server_cert.private_key_pem.write_to_path(server_key_path)
- server_cert.cert_chain_pems[0].write_to_path(server_cert_path)
- server_certs = {"keyfile": server_key_path, "certfile": server_cert_path}
+ server_certs = _write_cert_to_dir(server_cert, tmpdir)
io_loop = ioloop.IOLoop.current()
app = web.Application([(r".*", TestingApp)])
server, port = run_tornado_app(app, io_loop, server_certs, scheme, host)
- server_thread = threading.Thread(target=io_loop.start)
- server_thread.start()
+ server_thread = run_loop_in_thread(io_loop)
yield ServerConfig("https", host, port, ca_cert_path)
@@ -105,8 +99,7 @@ def run_server_and_proxy_in_thread(
)
proxy_config = ServerConfig(proxy_scheme, proxy_host, proxy_port, ca_cert_path)
- server_thread = threading.Thread(target=io_loop.start)
- server_thread.start()
+ loop_thread = run_loop_in_thread(io_loop)
yield (proxy_config, server_config)
@@ -114,7 +107,7 @@ def run_server_and_proxy_in_thread(
io_loop.add_callback(proxy_app.stop)
io_loop.add_callback(io_loop.stop)
- server_thread.join()
+ loop_thread.join()
@pytest.fixture(params=["localhost", "127.0.0.1", "::1"])
@@ -162,6 +155,21 @@ def no_san_server_with_different_commmon_name(
yield cfg
+@pytest.fixture
+def san_proxy_with_server(
+ loopback_host: str, tmp_path_factory: pytest.TempPathFactory
+) -> Generator[Tuple[ServerConfig, ServerConfig], None, None]:
+ tmpdir = tmp_path_factory.mktemp("certs")
+ ca = trustme.CA()
+ proxy_cert = ca.issue_cert(loopback_host)
+ server_cert = ca.issue_cert("localhost")
+
+ with run_server_and_proxy_in_thread(
+ "https", loopback_host, tmpdir, ca, proxy_cert, server_cert
+ ) as cfg:
+ yield cfg
+
+
@pytest.fixture
def no_san_proxy_with_server(
tmp_path_factory: pytest.TempPathFactory,

MODIFY
test/test_util.py
test/test_util.py
@@ -799,7 +799,12 @@ class TestUtil:
def test_connection_requires_http_tunnel_http_proxy(self) -> None:
proxy = parse_url("http://proxy:8080")
- proxy_config = ProxyConfig(ssl_context=None, use_forwarding_for_https=False)
+ proxy_config = ProxyConfig(
+ ssl_context=None,
+ use_forwarding_for_https=False,
+ assert_hostname=None,
+ assert_fingerprint=None,
+ )
destination_scheme = "http"
assert not connection_requires_http_tunnel(
proxy, proxy_config, destination_scheme
@@ -810,7 +815,12 @@ class TestUtil:
def test_connection_requires_http_tunnel_https_proxy(self) -> None:
proxy = parse_url("https://proxy:8443")
- proxy_config = ProxyConfig(ssl_context=None, use_forwarding_for_https=False)
+ proxy_config = ProxyConfig(
+ ssl_context=None,
+ use_forwarding_for_https=False,
+ assert_hostname=None,
+ assert_fingerprint=None,
+ )
destination_scheme = "http"
assert not connection_requires_http_tunnel(
proxy, proxy_config, destination_scheme

MODIFY
test/with_dummyserver/test_proxy_poolmanager.py
test/with_dummyserver/test_proxy_poolmanager.py
+import binascii
+import hashlib
+import ipaddress
import os.path
+import pathlib
import shutil
import socket
import ssl
@@ -651,6 +655,93 @@ class TestIPv6HTTPProxyManager(IPv6HTTPDummyProxyTestCase):
class TestHTTPSProxyVerification:
+ @staticmethod
+ def _get_proxy_fingerprint_md5(ca_path: str) -> str:
+ proxy_pem_path = pathlib.Path(ca_path).parent / "proxy.pem"
+ proxy_der = ssl.PEM_cert_to_DER_cert(proxy_pem_path.read_text())
+ proxy_hashed = hashlib.md5(proxy_der).digest()
+ fingerprint = binascii.hexlify(proxy_hashed).decode("ascii")
+ return fingerprint
+
+ @staticmethod
+ def _get_certificate_formatted_proxy_host(host: str) -> str:
+ try:
+ addr = ipaddress.ip_address(host)
+ except ValueError:
+ return host
+
+ if addr.version != 6:
+ return host
+
+ # Transform ipv6 like '::1' to 0:0:0:0:0:0:0:1 via '0000:0000:0000:0000:0000:0000:0000:0001'
+ return addr.exploded.replace("0000", "0").replace("000", "")
+
+ def test_https_proxy_assert_fingerprint_md5(
+ self, no_san_proxy_with_server: Tuple[ServerConfig, ServerConfig]
+ ) -> None:
+ proxy, server = no_san_proxy_with_server
+ proxy_url = f"https://{proxy.host}:{proxy.port}"
+ destination_url = f"https://{server.host}:{server.port}"
+
+ proxy_fingerprint = self._get_proxy_fingerprint_md5(proxy.ca_certs)
+ with proxy_from_url(
+ proxy_url,
+ ca_certs=proxy.ca_certs,
+ proxy_assert_fingerprint=proxy_fingerprint,
+ ) as https:
+ https.request("GET", destination_url)
+
+ def test_https_proxy_assert_fingerprint_md5_non_matching(
+ self, no_san_proxy_with_server: Tuple[ServerConfig, ServerConfig]
+ ) -> None:
+ proxy, server = no_san_proxy_with_server
+ proxy_url = f"https://{proxy.host}:{proxy.port}"
+ destination_url = f"https://{server.host}:{server.port}"
+
+ proxy_fingerprint = self._get_proxy_fingerprint_md5(proxy.ca_certs)
+ new_char = "b" if proxy_fingerprint[5] == "a" else "a"
+ proxy_fingerprint = proxy_fingerprint[:5] + new_char + proxy_fingerprint[6:]
+
+ with proxy_from_url(
+ proxy_url,
+ ca_certs=proxy.ca_certs,
+ proxy_assert_fingerprint=proxy_fingerprint,
+ ) as https:
+ with pytest.raises(MaxRetryError) as e:
+ https.request("GET", destination_url)
+
+ assert "Fingerprints did not match" in str(e)
+
+ def test_https_proxy_assert_hostname(
+ self, san_proxy_with_server: Tuple[ServerConfig, ServerConfig]
+ ) -> None:
+ proxy, server = san_proxy_with_server
+ destination_url = f"https://{server.host}:{server.port}"
+
+ with proxy_from_url(
+ proxy.base_url, ca_certs=proxy.ca_certs, proxy_assert_hostname=proxy.host
+ ) as https:
+ https.request("GET", destination_url)
+
+ def test_https_proxy_assert_hostname_non_matching(
+ self, san_proxy_with_server: Tuple[ServerConfig, ServerConfig]
+ ) -> None:
+ proxy, server = san_proxy_with_server
+ destination_url = f"https://{server.host}:{server.port}"
+
+ proxy_hostname = "example.com"
+ with proxy_from_url(
+ proxy.base_url,
+ ca_certs=proxy.ca_certs,
+ proxy_assert_hostname=proxy_hostname,
+ ) as https:
+ with pytest.raises(MaxRetryError) as e:
+ https.request("GET", destination_url)
+
+ proxy_host = self._get_certificate_formatted_proxy_host(proxy.host)
+ msg = f"hostname \\'{proxy_hostname}\\' doesn\\'t match \\'{proxy_host}\\'"
+ assert msg in str(e)
+
def test_https_proxy_hostname_verification(
self, no_localhost_san_server: ServerConfig
) -> None:

Add proxy_assert_hostname and proxy_assert_fingerprint to ProxyManager