Change proxy TLS flow to match destination TLS flow
MODIFY
src/urllib3/connection.py
src/urllib3/connection.py
@@ -26,8 +26,8 @@ if TYPE_CHECKING:
from typing_extensions import Literal
from .util.ssl_ import _TYPE_PEER_CERT_RET_DICT
+ from .util.ssltransport import SSLTransport
-from .util.proxy import create_proxy_ssl_context
from .util.timeout import _DEFAULT_TIMEOUT, _TYPE_TIMEOUT, Timeout
from .util.util import to_bytes, to_str
@@ -51,8 +51,8 @@ from .exceptions import (
SystemTimeWarning,
)
from .util import SKIP_HEADER, SKIPPABLE_HEADERS, connection, ssl_
+from .util.ssl_ import assert_fingerprint as _assert_fingerprint
from .util.ssl_ import (
- assert_fingerprint,
create_urllib3_context,
resolve_cert_reqs,
resolve_ssl_version,
@@ -120,6 +120,10 @@ class HTTPConnection(_HTTPConnection):
#: Whether this connection verifies the host's certificate.
is_verified: bool = False
+ #: Whether this proxy connection (if used) verifies the proxy host's
+ #: certificate. If no HTTPS proxy is being used will be ``None``.
+ proxy_is_verified: Optional[bool] = None
+
source_address: Optional[Tuple[str, int]]
socket_options: Optional[connection._TYPE_SOCKET_OPTIONS]
_tunnel_host: Optional[str]
@@ -199,7 +203,7 @@ class HTTPConnection(_HTTPConnection):
"""
try:
- conn = connection.create_connection(
+ sock = connection.create_connection(
(self._dns_host, self.port),
self.timeout,
source_address=self.source_address,
@@ -218,7 +222,7 @@ class HTTPConnection(_HTTPConnection):
self, f"Failed to establish a new connection: {e}"
) from e
- return conn
+ return sock
def _is_using_tunnel(self) -> Optional[str]:
return self._tunnel_host
@@ -440,18 +444,19 @@ class HTTPSConnection(HTTPConnection):
def connect(self) -> None:
self._connecting_to_proxy = bool(self.proxy)
- # Add certificate verification
- conn = self._new_conn()
+
+ sock: Union[socket.socket, "ssl.SSLSocket"]
+ sock = self._new_conn()
hostname: str = self.host
tls_in_tls = False
if self._is_using_tunnel():
if self.tls_in_tls_required:
- conn = self._connect_tls_proxy(hostname, conn)
+ sock = self._connect_tls_proxy(hostname, sock)
tls_in_tls = True
self._connecting_to_proxy = False
- self.sock = conn
+ self.sock = sock
# Calls self._set_hostport(), so self.host is
# self._tunnel_host below.
@@ -478,67 +483,150 @@ class HTTPSConnection(HTTPConnection):
SystemTimeWarning,
)
- # Wrap socket using verification with the root certs in
- # trusted_root_certs
- default_ssl_context = False
- if self.ssl_context is None:
- default_ssl_context = True
- self.ssl_context = create_urllib3_context(
- ssl_version=resolve_ssl_version(self.ssl_version),
+ sock_and_verified = _ssl_wrap_socket_and_match_hostname(
+ sock=sock,
+ cert_reqs=self.cert_reqs,
+ ssl_version=self.ssl_version,
ssl_minimum_version=self.ssl_minimum_version,
ssl_maximum_version=self.ssl_maximum_version,
- cert_reqs=resolve_cert_reqs(self.cert_reqs),
+ ca_certs=self.ca_certs,
+ ca_cert_dir=self.ca_cert_dir,
+ ca_cert_data=self.ca_cert_data,
+ cert_file=self.cert_file,
+ key_file=self.key_file,
+ key_password=self.key_password,
+ server_hostname=server_hostname,
+ ssl_context=self.ssl_context,
+ tls_in_tls=tls_in_tls,
+ assert_hostname=self.assert_hostname,
+ assert_fingerprint=self.assert_fingerprint,
+ )
+ self.sock = sock_and_verified.socket
+ self.is_verified = sock_and_verified.is_verified
+ self._connecting_to_proxy = False
+
+ def _connect_tls_proxy(self, hostname: str, sock: socket.socket) -> "ssl.SSLSocket":
+ """
+ Establish a TLS connection to the proxy using the provided SSL context.
+ """
+ proxy_config = cast(
+ ProxyConfig, self.proxy_config
+ ) # `_connect_tls_proxy` is called when self._is_using_tunnel() is truthy.
+ ssl_context = proxy_config.ssl_context
+ sock_and_verified = _ssl_wrap_socket_and_match_hostname(
+ sock,
+ cert_reqs=self.cert_reqs,
+ ssl_version=self.ssl_version,
+ ssl_minimum_version=self.ssl_minimum_version,
+ ssl_maximum_version=self.ssl_maximum_version,
+ ca_certs=self.ca_certs,
+ ca_cert_dir=self.ca_cert_dir,
+ ca_cert_data=self.ca_cert_data,
+ server_hostname=hostname,
+ ssl_context=ssl_context,
+ # Features that aren't implemented for proxies yet:
+ assert_fingerprint=None,
+ assert_hostname=None,
+ cert_file=None,
+ key_file=None,
+ key_password=None,
+ tls_in_tls=False,
+ )
+ self.proxy_is_verified = sock_and_verified.is_verified
+ return sock_and_verified.socket # type: ignore[return-value]
+
+
+class _WrappedAndVerifiedSocket(NamedTuple):
+ """
+ Wrapped socket and whether the connection is
+ verified after the TLS handshake
+ """
+
+ socket: Union["ssl.SSLSocket", "SSLTransport"]
+ is_verified: bool
+
+
+def _ssl_wrap_socket_and_match_hostname(
+ sock: socket.socket,
+ *,
+ cert_reqs: Union[None, str, int],
+ ssl_version: Union[None, str, int],
+ ssl_minimum_version: Optional[int],
+ ssl_maximum_version: Optional[int],
+ cert_file: Optional[str],
+ key_file: Optional[str],
+ key_password: Optional[str],
+ ca_certs: Optional[str],
+ ca_cert_dir: Optional[str],
+ ca_cert_data: Union[None, str, bytes],
+ assert_hostname: Union[None, str, "Literal[False]"],
+ assert_fingerprint: Optional[str],
+ server_hostname: Optional[str],
+ ssl_context: Optional["ssl.SSLContext"],
+ tls_in_tls: bool = False,
+) -> _WrappedAndVerifiedSocket:
+ """Logic for constructing an SSLContext from all TLS parameters, passing
+ that down into ssl_wrap_socket, and then doing certificate verification
+ either via hostname or fingerprint. This function exists to guarantee
+ that both proxies and targets have the same behavior when connecting via TLS.
+ """
+ default_ssl_context = False
+ if ssl_context is None:
+ default_ssl_context = True
+ context = create_urllib3_context(
+ ssl_version=resolve_ssl_version(ssl_version),
+ ssl_minimum_version=ssl_minimum_version,
+ ssl_maximum_version=ssl_maximum_version,
+ cert_reqs=resolve_cert_reqs(cert_reqs),
)
+ else:
+ context = ssl_context
+
+ context.verify_mode = resolve_cert_reqs(cert_reqs)
+
# In some cases, we want to verify hostnames ourselves
if (
# `ssl` can't verify fingerprints or alternate hostnames
- self.assert_fingerprint
- or self.assert_hostname
+ assert_fingerprint
+ or assert_hostname
# We still support OpenSSL 1.0.2, which prevents us from verifying
# hostnames easily: https://github.com/pyca/pyopenssl/pull/933
or ssl_.IS_PYOPENSSL
or not ssl_.HAS_NEVER_CHECK_COMMON_NAME
):
- self.ssl_context.check_hostname = False
-
- context = self.ssl_context
- context.verify_mode = resolve_cert_reqs(self.cert_reqs)
+ context.check_hostname = False
# Try to load OS default certs if none are given.
- # Works well on Windows.
if (
- not self.ca_certs
- and not self.ca_cert_dir
- and not self.ca_cert_data
+ not ca_certs
+ and not ca_cert_dir
+ and not ca_cert_data
and default_ssl_context
and hasattr(context, "load_default_certs")
):
context.load_default_certs()
- self.sock = ssl_wrap_socket(
- sock=conn,
- keyfile=self.key_file,
- certfile=self.cert_file,
- key_password=self.key_password,
- ca_certs=self.ca_certs,
- ca_cert_dir=self.ca_cert_dir,
- ca_cert_data=self.ca_cert_data,
+ ssl_sock = ssl_wrap_socket(
+ sock=sock,
+ keyfile=key_file,
+ certfile=cert_file,
+ key_password=key_password,
+ ca_certs=ca_certs,
+ ca_cert_dir=ca_cert_dir,
+ ca_cert_data=ca_cert_data,
server_hostname=server_hostname,
ssl_context=context,
tls_in_tls=tls_in_tls,
)
- self._connecting_to_proxy = False
- if self.assert_fingerprint:
- assert_fingerprint(
- self.sock.getpeercert(binary_form=True), self.assert_fingerprint
- )
+ if assert_fingerprint:
+ _assert_fingerprint(ssl_sock.getpeercert(binary_form=True), assert_fingerprint)
elif (
context.verify_mode != ssl.CERT_NONE
and not context.check_hostname
- and self.assert_hostname is not False
+ and assert_hostname is not False
):
- cert: "_TYPE_PEER_CERT_RET_DICT" = self.sock.getpeercert() # type: ignore[assignment]
+ cert: "_TYPE_PEER_CERT_RET_DICT" = ssl_sock.getpeercert() # type: ignore[assignment]
# Need to signal to our match_hostname whether to use 'commonName' or not.
# If we're using our own constructed SSLContext we explicitly set 'False'
@@ -552,50 +640,14 @@ class HTTPSConnection(HTTPConnection):
_match_hostname(
cert,
- self.assert_hostname or server_hostname,
+ assert_hostname or server_hostname, # type: ignore[arg-type]
hostname_checks_common_name,
)
- self.is_verified = context.verify_mode == ssl.CERT_REQUIRED or bool(
- self.assert_fingerprint
- )
-
- def _connect_tls_proxy(self, hostname: str, conn: socket.socket) -> "ssl.SSLSocket":
- """
- Establish a TLS connection to the proxy using the provided SSL context.
- """
-
- proxy_config = cast(
- ProxyConfig, self.proxy_config
- ) # `_connect_tls_proxy` is called when self._is_using_tunnel() is truthy.
- ssl_context = proxy_config.ssl_context
-
- if ssl_context:
- # If the user provided a proxy context, we assume CA and client
- # certificates have already been set
- return ssl_wrap_socket(
- sock=conn,
- server_hostname=hostname,
- ssl_context=ssl_context,
- )
-
- ssl_context = create_proxy_ssl_context(
- self.ssl_version,
- self.cert_reqs,
- self.ca_certs,
- self.ca_cert_dir,
- self.ca_cert_data,
- )
-
- # If no cert was provided, use only the default options for server
- # certificate validation
- return ssl_wrap_socket(
- sock=conn,
- ca_certs=self.ca_certs,
- ca_cert_dir=self.ca_cert_dir,
- ca_cert_data=self.ca_cert_data,
- server_hostname=hostname,
- ssl_context=ssl_context,
+ return _WrappedAndVerifiedSocket(
+ socket=ssl_sock,
+ is_verified=context.verify_mode == ssl.CERT_REQUIRED
+ or bool(assert_fingerprint),
)

MODIFY
src/urllib3/util/proxy.py
src/urllib3/util/proxy.py
-from typing import TYPE_CHECKING, Optional, Union
+from typing import TYPE_CHECKING, Optional
-from .ssl_ import create_urllib3_context, resolve_cert_reqs, resolve_ssl_version
from .url import Url
if TYPE_CHECKING:
- import ssl
-
from ..connection import ProxyConfig
@@ -42,29 +39,3 @@ def connection_requires_http_tunnel(
# Otherwise always use a tunnel.
return True
-
-
-def create_proxy_ssl_context(
- ssl_version: Optional[Union[int, str]] = None,
- cert_reqs: Optional[Union[int, str]] = None,
- ca_certs: Optional[str] = None,
- ca_cert_dir: Optional[str] = None,
- ca_cert_data: Union[None, str, bytes] = None,
-) -> "ssl.SSLContext":
- """
- Generates a default proxy ssl context if one hasn't been provided by the
- user.
- """
- ssl_context = create_urllib3_context(
- ssl_version=resolve_ssl_version(ssl_version),
- cert_reqs=resolve_cert_reqs(cert_reqs),
- )
- if (
- not ca_certs
- and not ca_cert_dir
- and not ca_cert_data
- and hasattr(ssl_context, "load_default_certs")
- ):
- ssl_context.load_default_certs()
-
- return ssl_context

MODIFY
test/conftest.py
test/conftest.py
@@ -112,6 +112,19 @@ def no_san_server_with_different_commmon_name(
yield cfg
+@pytest.fixture
+def no_localhost_san_server(
+ tmp_path_factory: pytest.TempPathFactory,
+) -> Generator[ServerConfig, None, None]:
+ tmpdir = tmp_path_factory.mktemp("certs")
+ ca = trustme.CA()
+ # non localhost common name
+ server_cert = ca.issue_cert("example.com")
+
+ with run_server_in_thread("https", "localhost", tmpdir, ca, server_cert) as cfg:
+ yield cfg
+
+
@pytest.fixture
def ip_san_server(
tmp_path_factory: pytest.TempPathFactory,

MODIFY
test/test_util.py
test/test_util.py
@@ -24,7 +24,7 @@ from urllib3.exceptions import (
)
from urllib3.util import is_fp_closed
from urllib3.util.connection import _has_ipv6, allowed_gai_family, create_connection
-from urllib3.util.proxy import connection_requires_http_tunnel, create_proxy_ssl_context
+from urllib3.util.proxy import connection_requires_http_tunnel
from urllib3.util.request import _FAILEDTELL, make_headers, rewind_body
from urllib3.util.response import assert_header_parsing
from urllib3.util.ssl_ import (
@@ -789,10 +789,6 @@ class TestUtil:
proxy, proxy_config, destination_scheme
)
- def test_create_proxy_ssl_context(self) -> None:
- ssl_context = create_proxy_ssl_context(ssl_version=None, cert_reqs=None)
- ssl_context.verify_mode = ssl.CERT_REQUIRED
-
def test_assert_header_parsing_no_error_on_multipart(self) -> None:
from http import client

MODIFY
test/with_dummyserver/test_proxy_poolmanager.py
test/with_dummyserver/test_proxy_poolmanager.py
@@ -29,6 +29,7 @@ from urllib3.util.ssl_ import create_urllib3_context
from urllib3.util.timeout import Timeout
from .. import TARPIT_HOST, requires_network
+from ..conftest import ServerConfig
# Retry failed tests
pytestmark = pytest.mark.flaky
@@ -633,3 +634,34 @@ class TestIPv6HTTPProxyManager(IPv6HTTPDummyProxyTestCase):
r = http.request("GET", f"{self.https_url}/")
assert r.status == 200
+
+
+class TestHTTPSProxyVerification:
+ def test_https_proxy_hostname_verification(
+ self, no_localhost_san_server: ServerConfig
+ ) -> None:
+ bad_server = no_localhost_san_server
+ bad_proxy_url = f"https://{bad_server.host}:{bad_server.port}"
+
+ # An exception will be raised before we contact the destination domain.
+ test_url = "testing.com"
+ with proxy_from_url(bad_proxy_url, ca_certs=bad_server.ca_certs) as https:
+ with pytest.raises(MaxRetryError) as e:
+ https.request("GET", "http://%s/" % test_url)
+ assert isinstance(e.value.reason, ProxyError)
+
+ ssl_error = e.value.reason.original_error
+ assert isinstance(ssl_error, SSLError)
+ assert "hostname 'localhost' doesn't match" in str(
+ ssl_error
+ ) or "Hostname mismatch" in str(ssl_error)
+
+ with pytest.raises(MaxRetryError) as e:
+ https.request("GET", "https://%s/" % test_url)
+ assert isinstance(e.value.reason, ProxyError)
+
+ ssl_error = e.value.reason.original_error
+ assert isinstance(ssl_error, SSLError)
+ assert "hostname 'localhost' doesn't match" in str(
+ ssl_error
+ ) or "Hostname mismatch" in str(ssl_error)

Change proxy TLS flow to match destination TLS flow