Use proxy_ssl_context as SSL ctx when forwarding

Use proxy_ssl_context when fowarding to both a HTTPS destination
(which occurs when use_forwarding_for_https=True)
and to a HTTP destination.

Raise an exception if both a proxy_ssl_context and a ssl_context
are provided when use_forwarding_for_https=True.

Add tests to verify these cases and modify
test_proxy_https_target_tls_error in test_proxy_poolmanager
to account for this new behavior.

Fixes urllib3#2577

changelog/2577.bugfix.rst

@@ -0,0 +1,5 @@
+The ``proxy_ssl_context`` argument for ``urllib3.ProxyManager`` will be used
+for the connection to the proxy when the connection is forwarded, either because
+``use_forwarding_for_https`` is set to ``True`` or because the target is a HTTP destination.
+Previously, ``proxy_ssl_context`` was not used in these situations. If ``ssl_context``
+was passed as a keyword argument, it was used instead.

src/urllib3/connection.py

@@ -421,7 +421,21 @@ def __init__(
         self.key_file = key_file
         self.cert_file = cert_file
         self.key_password = key_password
-        self.ssl_context = ssl_context
+
+        if proxy_config and proxy_config.use_forwarding_for_https and ssl_context:
+            if proxy_config.ssl_context:
+                raise ValueError(
+                    "Parameter ssl_context cannot be set when use_forwarding_for_https is True"
+                )
+            else:
+                raise ValueError(
+                    "Parameter ssl_context cannot be set when use_forwarding_for_https is True. Set parameter proxy_ssl_context instead"
+                )
+
+        if proxy_config and proxy_config.use_forwarding_for_https:
+            self.ssl_context = proxy_config.ssl_context
+        else:
+            self.ssl_context = ssl_context
         self.server_hostname = server_hostname
         self.ssl_version = None
         self.ssl_minimum_version = None
@@ -500,6 +514,11 @@ def connect(self) -> None:
                 SystemTimeWarning,
             )
 
+        if self._connecting_to_proxy and self.proxy_config:
+            ssl_context = self.proxy_config.ssl_context
+        else:
+            ssl_context = self.ssl_context
+
         sock_and_verified = _ssl_wrap_socket_and_match_hostname(
             sock=sock,
             cert_reqs=self.cert_reqs,
@@ -513,7 +532,7 @@ def connect(self) -> None:
             key_file=self.key_file,
             key_password=self.key_password,
             server_hostname=server_hostname,
-            ssl_context=self.ssl_context,
+            ssl_context=ssl_context,
             tls_in_tls=tls_in_tls,
             assert_hostname=self.assert_hostname,
             assert_fingerprint=self.assert_fingerprint,

test/with_dummyserver/test_proxy_poolmanager.py

@@ -78,17 +78,28 @@ def test_https_proxy(self) -> None:
     def test_https_proxy_with_proxy_ssl_context(self) -> None:
         proxy_ssl_context = create_urllib3_context()
         proxy_ssl_context.load_verify_locations(DEFAULT_CA)
+        ctx = ssl.create_default_context()
+        ctx.load_verify_locations(DEFAULT_CA)
         with proxy_from_url(
             self.https_proxy_url,
             proxy_ssl_context=proxy_ssl_context,
-            ca_certs=DEFAULT_CA,
+            ssl_context=ctx,
         ) as https:
             r = https.request("GET", f"{self.https_url}/")
             assert r.status == 200
 
             r = https.request("GET", f"{self.http_url}/")
             assert r.status == 200
 
+    def test_https_proxy_with_proxy_ssl_context_to_http_target(self) -> None:
+        proxy_ssl_context = create_urllib3_context()
+        proxy_ssl_context.load_verify_locations(DEFAULT_CA)
+        with proxy_from_url(
+            self.https_proxy_url, proxy_ssl_context=proxy_ssl_context
+        ) as https:
+            r = https.request("GET", f"{self.http_url}/")
+            assert r.status == 200
+
     @withPyOpenSSL
     def test_https_proxy_pyopenssl_not_supported(self) -> None:
         with proxy_from_url(self.https_proxy_url, ca_certs=DEFAULT_CA) as https:
@@ -394,6 +405,45 @@ def test_https_headers_forwarding_for_https(self) -> None:
                 returned_headers.get("Host") == f"{self.https_host}:{self.https_port}"
             )
 
+    def test_forwarding_for_https_with_proxy_ctx(self) -> None:
+        proxy_ctx = ssl.create_default_context()
+        proxy_ctx.load_verify_locations(DEFAULT_CA)
+        with proxy_from_url(
+            self.https_proxy_url,
+            proxy_ssl_context=proxy_ctx,
+            use_forwarding_for_https=True,
+        ) as proxy:
+            proxy.request("GET", f"{self.https_url}/")
+
+    def test_forwarding_for_https_error_with_ssl_ctx(self) -> None:
+        proxy_ctx = ssl.create_default_context()
+        proxy_ctx.load_verify_locations(DEFAULT_CA)
+        ctx = ssl.create_default_context()
+        ctx.load_verify_locations(DEFAULT_CA)
+        with proxy_from_url(
+            self.https_proxy_url,
+            proxy_ssl_context=proxy_ctx,
+            use_forwarding_for_https=True,
+            ssl_context=ctx,
+        ) as proxy:
+            with pytest.raises(
+                ValueError,
+                match="Parameter ssl_context cannot be set when use_forwarding_for_https is True",
+            ):
+                proxy.request("GET", self.https_url)
+
+    def test_forwarding_for_https_error_with_only_ssl_ctx(self) -> None:
+        ctx = ssl.create_default_context()
+        ctx.load_verify_locations(DEFAULT_CA)
+        with proxy_from_url(
+            self.https_proxy_url, use_forwarding_for_https=True, ssl_context=ctx
+        ) as proxy:
+            with pytest.raises(
+                ValueError,
+                match="Parameter ssl_context cannot be set when use_forwarding_for_https is True. Set parameter proxy_ssl_context instead",
+            ):
+                proxy.request("GET", self.https_url)
+
     def test_headerdict(self) -> None:
         default_headers = HTTPHeaderDict(a="b")
         proxy_headers = HTTPHeaderDict()
@@ -571,19 +621,13 @@ def test_https_proxy_tls_error(
 
     @requires_network()
     @pytest.mark.parametrize(
-        ["proxy_scheme", "use_forwarding_for_https"],
+        ["proxy_scheme"],
         [
-            ("http", False),
-            ("https", False),
-            ("https", True),
+            ("http",),
+            ("https",),
         ],
     )
-    def test_proxy_https_target_tls_error(
-        self, proxy_scheme: str, use_forwarding_for_https: str
-    ) -> None:
-        if proxy_scheme == "https" and use_forwarding_for_https:
-            pytest.skip("Test is expected to fail due to urllib3/urllib3#2577")
-
+    def test_proxy_https_target_tls_error(self, proxy_scheme: str) -> None:
         proxy_url = self.https_proxy_url if proxy_scheme == "https" else self.proxy_url
         proxy_ctx = ssl.create_default_context()
         proxy_ctx.load_verify_locations(DEFAULT_CA)
@@ -593,7 +637,7 @@ def test_proxy_https_target_tls_error(
             proxy_url,
             proxy_ssl_context=proxy_ctx,
             ssl_context=ctx,
-            use_forwarding_for_https=use_forwarding_for_https,
+            use_forwarding_for_https=False,
         ) as proxy:
             with pytest.raises(MaxRetryError) as e:
                 proxy.request("GET", self.https_url)