test/with_dummyserver/test_proxy_poolmanager.py - main - niceSOFT/python3-urllib3

test/with_dummyserver/test_proxy_poolmanager.py
1 213 строк48 KB
Andrey Petrov
Isolate HTTPS proxy TLS settings from target TLS (#5093)
30 июл 2026, 01:20
30 июл 2026, 01:2007408ce
from __future__ import annotations

import binascii
import contextlib
import hashlib
import ipaddress
import os.path
import pathlib
import shutil
import socket
import ssl
import tempfile
import typing
from test import LONG_TIMEOUT, SHORT_TIMEOUT, resolvesLocalhostFQDN, withPyOpenSSL
from test.conftest import ServerConfig

import pytest
import trustme

import urllib3.exceptions
from dummyserver.socketserver import DEFAULT_CA, HAS_IPV6, get_unreachable_address
from dummyserver.testcase import (
    HypercornDummyProxyTestCase,
    IPv6HypercornDummyProxyTestCase,
)
from urllib3 import HTTPResponse
from urllib3._collections import HTTPHeaderDict
from urllib3.connection import HTTPSConnection, VerifiedHTTPSConnection
from urllib3.connectionpool import connection_from_url
from urllib3.exceptions import (
    ConnectTimeoutError,
    InsecureRequestWarning,
    MaxRetryError,
    ProxyError,
    ProxySchemeUnknown,
    ProxySchemeUnsupported,
    ReadTimeoutError,
    SSLError,
)
from urllib3.poolmanager import ProxyManager, proxy_from_url
from urllib3.util.retry import RequestHistory
from urllib3.util.ssl_ import create_urllib3_context
from urllib3.util.timeout import Timeout

from .. import TARPIT_HOST, requires_network


def assert_is_verified(pm: ProxyManager, *, proxy: bool, target: bool) -> None:
    pool = list(pm.pools._container.values())[-1]  # retrieve last pool entry
    connection = (
        pool.pool.queue[-1] if pool.pool is not None else None
    )  # retrieve last connection entry

    assert connection is not None
    assert connection.proxy_is_verified is proxy
    assert connection.is_verified is target


class TestHTTPProxyManager(HypercornDummyProxyTestCase):
    @classmethod
    def setup_class(cls) -> None:
        super().setup_class()
        cls.http_url = f"http://{cls.http_host}:{int(cls.http_port)}"
        cls.http_url_alt = f"http://{cls.http_host_alt}:{int(cls.http_port)}"
        cls.https_url = f"https://{cls.https_host}:{int(cls.https_port)}"
        cls.https_url_alt = f"https://{cls.https_host_alt}:{int(cls.https_port)}"
        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)}"

        # Generate another CA to test verification failure
        cls.certs_dir = tempfile.mkdtemp()
        bad_ca = trustme.CA()

        cls.bad_ca_path = os.path.join(cls.certs_dir, "ca_bad.pem")
        bad_ca.cert_pem.write_to_path(cls.bad_ca_path)

    @classmethod
    def teardown_class(cls) -> None:
        super().teardown_class()
        shutil.rmtree(cls.certs_dir)

    def test_basic_proxy(self) -> None:
        with proxy_from_url(self.proxy_url, ca_certs=DEFAULT_CA) as http:
            r = http.request("GET", f"{self.http_url}/")
            assert r.status == 200

            r = http.request("GET", f"{self.https_url}/")
            assert r.status == 200

    def test_https_proxy(self) -> None:
        with proxy_from_url(self.https_proxy_url, ca_certs=DEFAULT_CA) 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_is_verified_http_proxy_to_http_target(self) -> None:
        with proxy_from_url(self.proxy_url, ca_certs=DEFAULT_CA) as http:
            r = http.request("GET", f"{self.http_url}/")
            assert r.status == 200
            assert_is_verified(http, proxy=False, target=False)

    def test_is_verified_http_proxy_to_https_target(self) -> None:
        with proxy_from_url(self.proxy_url, ca_certs=DEFAULT_CA) as http:
            r = http.request("GET", f"{self.https_url}/")
            assert r.status == 200
            assert_is_verified(http, proxy=False, target=True)

    def test_is_verified_https_proxy_to_http_target(self) -> None:
        with proxy_from_url(self.https_proxy_url, ca_certs=DEFAULT_CA) as https:
            r = https.request("GET", f"{self.http_url}/")
            assert r.status == 200
            assert_is_verified(https, proxy=True, target=False)

    def test_is_verified_https_proxy_to_http_target_without_proxy_config(self) -> None:
        with proxy_from_url(self.https_proxy_url, ca_certs=DEFAULT_CA) as https:
            https.connection_pool_kw["_proxy_config"] = None
            r = https.request("GET", f"{self.http_url}/")
            assert r.status == 200
            assert_is_verified(https, proxy=True, target=False)

    def test_is_verified_https_proxy_to_https_target(self) -> None:
        with proxy_from_url(self.https_proxy_url, ca_certs=DEFAULT_CA) as https:
            r = https.request("GET", f"{self.https_url}/")
            assert r.status == 200
            assert_is_verified(https, proxy=True, target=True)

    def test_http_and_https_kwarg_ca_cert_data_proxy(self) -> None:
        with open(DEFAULT_CA) as pem_file:
            pem_file_data = pem_file.read()
        with proxy_from_url(self.https_proxy_url, ca_cert_data=pem_file_data) 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(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,
            ca_certs=DEFAULT_CA,
        ) 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

    @withPyOpenSSL
    def test_https_proxy_with_proxy_ssl_context_pyopenssl(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:
            response = https.request("GET", f"{self.http_url}/")
            assert response.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:
            r = https.request("GET", f"{self.http_url}/")
            assert r.status == 200

            with pytest.raises(
                ProxySchemeUnsupported, match="isn't available on non-native SSLContext"
            ):
                https.request("GET", f"{self.https_url}/")

    def test_https_proxy_forwarding_for_https(self) -> None:
        with proxy_from_url(
            self.https_proxy_url,
            ca_certs=DEFAULT_CA,
            use_forwarding_for_https=True,
        ) as https:
            r = https.request("GET", f"{self.http_url}/")
            assert r.status == 200

            r = https.request("GET", f"{self.https_url}/")
            assert r.status == 200

    @requires_network()
    @pytest.mark.parametrize(
        "proxy_ssl_context_kw",
        [
            ("ssl_context",),
            ("proxy_ssl_context",),
            ("ssl_context", "proxy_ssl_context"),
        ],
    )
    def test_https_proxy_forwarding_for_https_with_custom_context(
        self, proxy_ssl_context_kw: tuple[str, ...]
    ) -> None:
        """
        Test that an HTTP request succeeds when using a forwarding HTTPS
        proxy with an SSL context provided via either
        ``proxy_ssl_context``, ``ssl_context`` (fallback), or both.
        """

        proxy_ctx = ssl.create_default_context()
        proxy_ctx.load_verify_locations(DEFAULT_CA)

        warning_checker: contextlib.AbstractContextManager[typing.Any]
        if "ssl_context" in proxy_ssl_context_kw:
            warning_checker = pytest.warns(FutureWarning, match="ssl_context")
        else:
            warning_checker = contextlib.nullcontext()
        with warning_checker:
            proxy = proxy_from_url(
                self.https_proxy_url,
                **{kw: proxy_ctx for kw in proxy_ssl_context_kw},
                use_forwarding_for_https=True,
            )

        with proxy:
            resp = proxy.request("GET", self.https_url)
            assert resp.status == 200

    def test_nagle_proxy(self) -> None:
        """Test that proxy connections do not have TCP_NODELAY turned on"""
        with ProxyManager(self.proxy_url) as http:
            hc2 = http.connection_from_host(self.http_host, self.http_port)
            conn = hc2._get_conn()
            try:
                hc2._make_request(conn, "GET", f"{self.http_url}/")
                tcp_nodelay_setting = conn.sock.getsockopt(  # type: ignore[attr-defined]
                    socket.IPPROTO_TCP, socket.TCP_NODELAY
                )
                assert tcp_nodelay_setting == 0, (
                    "Expected TCP_NODELAY for proxies to be set "
                    "to zero, instead was %s" % tcp_nodelay_setting
                )
            finally:
                conn.close()

    @pytest.mark.parametrize("proxy_scheme", ["http", "https"])
    @pytest.mark.parametrize("target_scheme", ["http", "https"])
    def test_proxy_conn_fail_from_dns(
        self, proxy_scheme: str, target_scheme: str
    ) -> None:
        host, port = get_unreachable_address()
        with proxy_from_url(
            f"{proxy_scheme}://{host}:{port}/", retries=1, timeout=LONG_TIMEOUT
        ) as http:
            if target_scheme == "https":
                target_url = self.https_url
            else:
                target_url = self.http_url

            with pytest.raises(MaxRetryError) as e:
                http.request("GET", f"{target_url}/")
            assert isinstance(e.value.reason, ProxyError)
            assert isinstance(
                e.value.reason.original_error, urllib3.exceptions.NameResolutionError
            )

    def test_oldapi(self) -> None:
        with ProxyManager(
            connection_from_url(self.proxy_url), ca_certs=DEFAULT_CA  # type: ignore[arg-type]
        ) as http:
            r = http.request("GET", f"{self.http_url}/")
            assert r.status == 200

            r = http.request("GET", f"{self.https_url}/")
            assert r.status == 200

    @resolvesLocalhostFQDN()
    def test_proxy_https_fqdn(self) -> None:
        with proxy_from_url(self.proxy_url, ca_certs=DEFAULT_CA) as http:
            r = http.request("GET", f"{self.https_url_fqdn}/")
            assert r.status == 200

    def test_proxy_verified(self) -> None:
        with proxy_from_url(
            self.proxy_url, cert_reqs="REQUIRED", ca_certs=self.bad_ca_path
        ) as http:
            with http._new_pool(
                "https", self.https_host, self.https_port
            ) as https_pool:
                with pytest.raises(MaxRetryError) as e:
                    https_pool.request("GET", "/", retries=0)
            assert isinstance(e.value.reason, SSLError)
            assert (
                "certificate verify failed" in str(e.value.reason)
                # PyPy is more specific
                or "self signed certificate in certificate chain" in str(e.value.reason)
            ), f"Expected 'certificate verify failed', instead got: {e.value.reason!r}"

            http = proxy_from_url(
                self.proxy_url, cert_reqs="REQUIRED", ca_certs=DEFAULT_CA
            )
            with http._new_pool(
                "https", self.https_host, self.https_port
            ) as https_pool2:
                with contextlib.closing(https_pool._new_conn()) as conn:
                    assert conn.__class__ == VerifiedHTTPSConnection
                    https_pool2.request(
                        "GET", "/"
                    )  # Should succeed without exceptions.

            http = proxy_from_url(
                self.proxy_url, cert_reqs="REQUIRED", ca_certs=DEFAULT_CA
            )
            with http._new_pool(
                "https", "127.0.0.1", self.https_port
            ) as https_fail_pool:
                with pytest.raises(
                    MaxRetryError, match="doesn't match|IP address mismatch"
                ) as e:
                    https_fail_pool.request("GET", "/", retries=0)
                assert isinstance(e.value.reason, SSLError)

    def test_redirect(self) -> None:
        with proxy_from_url(self.proxy_url) as http:
            r = http.request(
                "GET",
                f"{self.http_url}/redirect",
                fields={"target": f"{self.http_url}/"},
                redirect=False,
            )

            assert r.status == 303

            r = http.request(
                "GET",
                f"{self.http_url}/redirect",
                fields={"target": f"{self.http_url}/"},
            )

            assert r.status == 200
            assert r.data == b"Dummy server!"

    def test_cross_host_redirect(self) -> None:
        with proxy_from_url(self.proxy_url) as http:
            cross_host_location = f"{self.http_url_alt}/echo?a=b"
            with pytest.raises(MaxRetryError):
                http.request(
                    "GET",
                    f"{self.http_url}/redirect",
                    fields={"target": cross_host_location},
                    retries=0,
                )

            r = http.request(
                "GET",
                f"{self.http_url}/redirect",
                fields={"target": f"{self.http_url_alt}/echo?a=b"},
                retries=1,
            )
            assert isinstance(r, HTTPResponse)
            assert r._pool is not None
            assert r._pool.host != self.http_host_alt

    _sensitive_headers = {
        "Authorization": "foo",
        "Proxy-Authorization": "bar",
        "Cookie": "foo=bar",
    }

    @pytest.mark.parametrize(
        "sensitive_headers",
        (_sensitive_headers, {k.lower(): v for k, v in _sensitive_headers.items()}),
        ids=("capitalized", "lowercase"),
    )
    def test_cross_host_redirect_remove_headers_via_proxy_manager(
        self, sensitive_headers: dict[str, str]
    ) -> None:
        headers_url = f"{self.http_url_alt}/headers"
        initial_url = f"{self.http_url}/redirect?target={headers_url}"
        with proxy_from_url(self.proxy_url) as proxy_mgr:
            r = proxy_mgr.request(
                "GET", initial_url, headers=sensitive_headers, retries=1
            )
            assert r.status == 200
            assert r.retries is not None
            assert r.retries.history == (
                RequestHistory(
                    method="GET",
                    url=initial_url,
                    error=None,
                    status=303,
                    redirect_location=headers_url,
                ),
            )
            data = r.json()
            for header in sensitive_headers:
                assert header not in data

    @pytest.mark.parametrize(
        "sensitive_headers",
        (_sensitive_headers, {k.lower(): v for k, v in _sensitive_headers.items()}),
        ids=("capitalized", "lowercase"),
    )
    def test_cross_host_redirect_remove_headers_via_pool(
        self, sensitive_headers: dict[str, str]
    ) -> None:
        headers_url = f"{self.http_url_alt}/headers"
        initial_url = f"{self.http_url}/redirect?target={headers_url}"
        with proxy_from_url(self.proxy_url) as proxy_mgr:
            pool = proxy_mgr.connection_from_url(self.http_url)
            r = pool.urlopen(
                "GET",
                initial_url,
                headers=sensitive_headers,
                retries=1,
                redirect=True,
                assert_same_host=False,
                preload_content=True,
            )
            assert r.status == 200
            assert r.retries is not None
            assert r.retries.history == (
                RequestHistory(
                    method="GET",
                    url=initial_url,
                    error=None,
                    status=303,
                    redirect_location=headers_url,
                ),
            )
            data = r.json()
            for header in sensitive_headers:
                assert header not in data

    def test_cross_protocol_redirect(self) -> None:
        with proxy_from_url(self.proxy_url, ca_certs=DEFAULT_CA) as http:
            cross_protocol_location = f"{self.https_url}/echo?a=b"
            with pytest.raises(MaxRetryError):
                http.request(
                    "GET",
                    f"{self.http_url}/redirect",
                    fields={"target": cross_protocol_location},
                    retries=0,
                )

            r = http.request(
                "GET",
                f"{self.http_url}/redirect",
                fields={"target": f"{self.https_url}/echo?a=b"},
                retries=1,
            )
            assert isinstance(r, HTTPResponse)
            assert r._pool is not None
            assert r._pool.host == self.https_host

    def test_headers(self) -> None:
        with proxy_from_url(
            self.proxy_url,
            headers={"Foo": "bar"},
            proxy_headers={"Hickory": "dickory"},
            ca_certs=DEFAULT_CA,
        ) as http:
            r = http.request_encode_url("GET", f"{self.http_url}/headers")
            returned_headers = r.json()
            assert returned_headers.get("Foo") == "bar"
            assert returned_headers.get("Hickory") == "dickory"
            assert returned_headers.get("Host") == f"{self.http_host}:{self.http_port}"

            r = http.request_encode_url("GET", f"{self.http_url_alt}/headers")
            returned_headers = r.json()
            assert returned_headers.get("Foo") == "bar"
            assert returned_headers.get("Hickory") == "dickory"
            assert (
                returned_headers.get("Host") == f"{self.http_host_alt}:{self.http_port}"
            )

            r = http.request_encode_url("GET", f"{self.https_url}/headers")
            returned_headers = r.json()
            assert returned_headers.get("Foo") == "bar"
            assert returned_headers.get("Hickory") is None
            assert (
                returned_headers.get("Host") == f"{self.https_host}:{self.https_port}"
            )

            r = http.request_encode_body("POST", f"{self.http_url}/headers")
            returned_headers = r.json()
            assert returned_headers.get("Foo") == "bar"
            assert returned_headers.get("Hickory") == "dickory"
            assert returned_headers.get("Host") == f"{self.http_host}:{self.http_port}"

            r = http.request_encode_url(
                "GET", f"{self.http_url}/headers", headers={"Baz": "quux"}
            )
            returned_headers = r.json()
            assert returned_headers.get("Foo") is None
            assert returned_headers.get("Baz") == "quux"
            assert returned_headers.get("Hickory") == "dickory"
            assert returned_headers.get("Host") == f"{self.http_host}:{self.http_port}"

            r = http.request_encode_url(
                "GET", f"{self.https_url}/headers", headers={"Baz": "quux"}
            )
            returned_headers = r.json()
            assert returned_headers.get("Foo") is None
            assert returned_headers.get("Baz") == "quux"
            assert returned_headers.get("Hickory") is None
            assert (
                returned_headers.get("Host") == f"{self.https_host}:{self.https_port}"
            )

            r = http.request_encode_body(
                "GET", f"{self.http_url}/headers", headers={"Baz": "quux"}
            )
            returned_headers = r.json()
            assert returned_headers.get("Foo") is None
            assert returned_headers.get("Baz") == "quux"
            assert returned_headers.get("Hickory") == "dickory"
            assert returned_headers.get("Host") == f"{self.http_host}:{self.http_port}"

            r = http.request_encode_body(
                "GET", f"{self.https_url}/headers", headers={"Baz": "quux"}
            )
            returned_headers = r.json()
            assert returned_headers.get("Foo") is None
            assert returned_headers.get("Baz") == "quux"
            assert returned_headers.get("Hickory") is None
            assert (
                returned_headers.get("Host") == f"{self.https_host}:{self.https_port}"
            )

    def test_https_headers(self) -> None:
        with proxy_from_url(
            self.https_proxy_url,
            headers={"Foo": "bar"},
            proxy_headers={"Hickory": "dickory"},
            ca_certs=DEFAULT_CA,
        ) as http:
            r = http.request_encode_url("GET", f"{self.http_url}/headers")
            returned_headers = r.json()
            assert returned_headers.get("Foo") == "bar"
            assert returned_headers.get("Hickory") == "dickory"
            assert returned_headers.get("Host") == f"{self.http_host}:{self.http_port}"

            r = http.request_encode_url("GET", f"{self.http_url_alt}/headers")
            returned_headers = r.json()
            assert returned_headers.get("Foo") == "bar"
            assert returned_headers.get("Hickory") == "dickory"
            assert (
                returned_headers.get("Host") == f"{self.http_host_alt}:{self.http_port}"
            )

            r = http.request_encode_body(
                "GET", f"{self.https_url}/headers", headers={"Baz": "quux"}
            )
            returned_headers = r.json()
            assert returned_headers.get("Foo") is None
            assert returned_headers.get("Baz") == "quux"
            assert returned_headers.get("Hickory") is None
            assert (
                returned_headers.get("Host") == f"{self.https_host}:{self.https_port}"
            )

    def test_https_headers_forwarding_for_https(self) -> None:
        with proxy_from_url(
            self.https_proxy_url,
            headers={"Foo": "bar"},
            proxy_headers={"Hickory": "dickory"},
            ca_certs=DEFAULT_CA,
            use_forwarding_for_https=True,
        ) as http:
            r = http.request_encode_url("GET", f"{self.https_url}/headers")
            returned_headers = r.json()
            assert returned_headers.get("Foo") == "bar"
            assert returned_headers.get("Hickory") == "dickory"
            assert (
                returned_headers.get("Host") == f"{self.https_host}:{self.https_port}"
            )

    def test_headerdict(self) -> None:
        default_headers = HTTPHeaderDict(a="b")
        proxy_headers = HTTPHeaderDict()
        proxy_headers.add("foo", "bar")

        with proxy_from_url(
            self.proxy_url, headers=default_headers, proxy_headers=proxy_headers
        ) as http:
            request_headers = HTTPHeaderDict(baz="quux")
            r = http.request("GET", f"{self.http_url}/headers", headers=request_headers)
            returned_headers = r.json()
            assert returned_headers.get("Foo") == "bar"
            assert returned_headers.get("Baz") == "quux"

    def test_proxy_pooling(self) -> None:
        with proxy_from_url(self.proxy_url, cert_reqs="NONE") as http:
            for x in range(2):
                http.urlopen("GET", self.http_url)
            assert len(http.pools) == 1

            for x in range(2):
                http.urlopen("GET", self.http_url_alt)
            assert len(http.pools) == 1

            for x in range(2):
                with pytest.warns(InsecureRequestWarning):
                    http.urlopen("GET", self.https_url)
            assert len(http.pools) == 2

            for x in range(2):
                with pytest.warns(InsecureRequestWarning):
                    http.urlopen("GET", self.https_url_alt)
            assert len(http.pools) == 3

    def test_proxy_pooling_ext(self) -> None:
        with proxy_from_url(self.proxy_url) as http:
            hc1 = http.connection_from_url(self.http_url)
            hc2 = http.connection_from_host(self.http_host, self.http_port)
            hc3 = http.connection_from_url(self.http_url_alt)
            hc4 = http.connection_from_host(self.http_host_alt, self.http_port)
            assert hc1 == hc2
            assert hc2 == hc3
            assert hc3 == hc4

            sc1 = http.connection_from_url(self.https_url)
            sc2 = http.connection_from_host(
                self.https_host, self.https_port, scheme="https"
            )
            sc3 = http.connection_from_url(self.https_url_alt)
            sc4 = http.connection_from_host(
                self.https_host_alt, self.https_port, scheme="https"
            )
            assert sc1 == sc2
            assert sc2 != sc3
            assert sc3 == sc4

    @requires_network()
    @pytest.mark.parametrize(
        ["proxy_scheme", "target_scheme", "use_forwarding_for_https"],
        [
            ("http", "http", False),
            ("https", "http", False),
            # 'use_forwarding_for_https' is only valid for HTTPS+HTTPS.
            ("https", "https", True),
        ],
    )
    def test_forwarding_proxy_request_timeout(
        self, proxy_scheme: str, target_scheme: str, use_forwarding_for_https: bool
    ) -> None:
        proxy_url = self.https_proxy_url if proxy_scheme == "https" else self.proxy_url
        target_url = f"{target_scheme}://{TARPIT_HOST}"

        with proxy_from_url(
            proxy_url,
            ca_certs=DEFAULT_CA,
            use_forwarding_for_https=use_forwarding_for_https,
        ) as proxy:
            with pytest.raises(MaxRetryError) as e:
                timeout = Timeout(connect=LONG_TIMEOUT, read=SHORT_TIMEOUT)
                proxy.request("GET", target_url, timeout=timeout)

            # We sent the request to the proxy but didn't get any response
            # so we're not sure if that's being caused by the proxy or the
            # target so we put the blame on the target.
            assert isinstance(e.value.reason, ReadTimeoutError)

    @requires_network()
    @pytest.mark.parametrize(
        ["proxy_scheme", "target_scheme"], [("http", "https"), ("https", "https")]
    )
    def test_tunneling_proxy_request_timeout(
        self, proxy_scheme: str, target_scheme: str
    ) -> None:
        proxy_url = self.https_proxy_url if proxy_scheme == "https" else self.proxy_url
        target_url = f"{target_scheme}://{TARPIT_HOST}"

        with proxy_from_url(
            proxy_url,
            ca_certs=DEFAULT_CA,
        ) as proxy:
            with pytest.raises(MaxRetryError) as e:
                timeout = Timeout(connect=LONG_TIMEOUT, read=SHORT_TIMEOUT)
                proxy.request("GET", target_url, timeout=timeout)

            assert isinstance(e.value.reason, ReadTimeoutError)

    @requires_network()
    @pytest.mark.parametrize(
        ["proxy_scheme", "target_scheme", "use_forwarding_for_https"],
        [
            ("http", "http", False),
            ("https", "http", False),
            # 'use_forwarding_for_https' is only valid for HTTPS+HTTPS.
            ("https", "https", True),
        ],
    )
    def test_forwarding_proxy_connect_timeout(
        self, proxy_scheme: str, target_scheme: str, use_forwarding_for_https: bool
    ) -> None:
        proxy_url = f"{proxy_scheme}://{TARPIT_HOST}"
        target_url = self.https_url if target_scheme == "https" else self.http_url

        with proxy_from_url(
            proxy_url,
            ca_certs=DEFAULT_CA,
            timeout=SHORT_TIMEOUT,
            use_forwarding_for_https=use_forwarding_for_https,
        ) as proxy:
            with pytest.raises(MaxRetryError) as e:
                proxy.request("GET", target_url)

            assert isinstance(e.value.reason, ProxyError)
            assert isinstance(e.value.reason.original_error, ConnectTimeoutError)

    @requires_network()
    @pytest.mark.parametrize(
        ["proxy_scheme", "target_scheme"], [("http", "https"), ("https", "https")]
    )
    def test_tunneling_proxy_connect_timeout(
        self, proxy_scheme: str, target_scheme: str
    ) -> None:
        proxy_url = f"{proxy_scheme}://{TARPIT_HOST}"
        target_url = self.https_url if target_scheme == "https" else self.http_url

        with proxy_from_url(
            proxy_url, ca_certs=DEFAULT_CA, timeout=SHORT_TIMEOUT
        ) as proxy:
            with pytest.raises(MaxRetryError) as e:
                proxy.request("GET", target_url)

            assert isinstance(e.value.reason, ProxyError)
            assert isinstance(e.value.reason.original_error, ConnectTimeoutError)

    @requires_network()
    @pytest.mark.parametrize(
        ["target_scheme", "use_forwarding_for_https"],
        [
            ("http", False),
            ("https", False),
            ("https", True),
        ],
    )
    def test_https_proxy_tls_error(
        self, target_scheme: str, use_forwarding_for_https: str
    ) -> None:
        target_url = self.https_url if target_scheme == "https" else self.http_url
        proxy_ctx = ssl.create_default_context()
        with proxy_from_url(
            self.https_proxy_url,
            proxy_ssl_context=proxy_ctx,
            use_forwarding_for_https=use_forwarding_for_https,
        ) as proxy:
            with pytest.raises(MaxRetryError) as e:
                proxy.request("GET", target_url)
            assert isinstance(e.value.reason, ProxyError)
            assert isinstance(e.value.reason.original_error, SSLError)

    @requires_network()
    @pytest.mark.parametrize("proxy_scheme", ["http", "https"])
    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)
        ctx = ssl.create_default_context()

        with proxy_from_url(
            proxy_url,
            proxy_ssl_context=proxy_ctx,
            ssl_context=ctx,
        ) as proxy:
            with pytest.raises(MaxRetryError) as e:
                proxy.request("GET", self.https_url)
            assert isinstance(e.value.reason, SSLError)

    @requires_network()
    def test_forwarding_proxy_tls_error_with_fallback(self) -> None:
        """
        Test that when only ``ssl_context`` is passed to a forwarding
        HTTPS proxy, a ``FutureWarning`` is emitted and if the TLS
        handshake with the proxy fails using that context, the error is
        a ``ProxyError`` wrapping the original TLS error.
        """
        ctx = ssl.create_default_context()

        with pytest.warns(FutureWarning, match="ssl_context"):
            proxy = proxy_from_url(
                self.https_proxy_url,
                ssl_context=ctx,
                use_forwarding_for_https=True,
            )

        with proxy:
            with pytest.raises(MaxRetryError) as e:
                proxy.request("GET", self.https_url)
            assert isinstance(e.value.reason, ProxyError)
            assert isinstance(e.value.reason.original_error, SSLError)

    @requires_network()
    def test_forwarding_proxy_ssl_context_precedence(self) -> None:
        """
        Test that when both ``ssl_context`` and ``proxy_ssl_context``
        are passed to a forwarding HTTPS proxy, ``proxy_ssl_context`` is
        used for the proxy connection and ``ssl_context`` is ignored
        (with a warning).
        """
        ssl_context = create_urllib3_context(cert_reqs=ssl.CERT_REQUIRED)
        proxy_ssl_context = create_urllib3_context(cert_reqs=ssl.CERT_NONE)

        with pytest.warns(FutureWarning, match="ssl_context"):
            proxy = proxy_from_url(
                self.https_proxy_url,
                proxy_ssl_context=proxy_ssl_context,
                ssl_context=ssl_context,
                use_forwarding_for_https=True,
            )

        with proxy:
            pool = proxy.connection_from_url(self.https_url)
            with contextlib.closing(pool._new_conn()) as conn:
                conn = typing.cast(HTTPSConnection, conn)
                conn.connect()
                assert isinstance(conn.sock, ssl.SSLSocket)
                assert conn.sock.context is proxy_ssl_context
                assert proxy_ssl_context.verify_mode == ssl.CERT_NONE

    @requires_network()
    def test_forwarding_proxy_ssl_context_fallback(self) -> None:
        """
        Test that when only ``ssl_context`` is passed to a forwarding
        HTTPS proxy, a ``FutureWarning`` is emitted and ``ssl_context``
        is used as the proxy TLS context (fallback).
        """
        ssl_context = create_urllib3_context(cert_reqs=ssl.CERT_NONE)

        with pytest.warns(FutureWarning, match="ssl_context"):
            proxy = proxy_from_url(
                self.https_proxy_url,
                ssl_context=ssl_context,
                cert_reqs=ssl.CERT_REQUIRED,
                ca_certs=DEFAULT_CA,
                use_forwarding_for_https=True,
            )

        assert proxy.proxy_ssl_context is ssl_context
        assert proxy.proxy_config is not None
        assert proxy.proxy_config.ssl_context is None

        with proxy:
            pool = proxy.connection_from_url(self.https_url)
            with contextlib.closing(pool._new_conn()) as conn:
                conn = typing.cast(HTTPSConnection, conn)
                conn.connect()
                assert isinstance(conn.sock, ssl.SSLSocket)
                assert conn.sock.context is ssl_context
                assert ssl_context.verify_mode == ssl.CERT_REQUIRED

    @requires_network()
    def test_https_proxy_to_http_target_ssl_context_fallback(self) -> None:
        ssl_context = create_urllib3_context()
        ssl_context.load_verify_locations(DEFAULT_CA)

        with proxy_from_url(
            self.https_proxy_url,
            ssl_context=ssl_context,
        ) as proxy:
            pool = proxy.connection_from_url(self.http_url)
            with contextlib.closing(pool._new_conn()) as conn:
                conn = typing.cast(HTTPSConnection, conn)
                conn.connect()
                assert isinstance(conn.sock, ssl.SSLSocket)
                assert conn.sock.context is ssl_context

    @requires_network()
    def test_forwarding_non_https_proxy_ssl_context_fallback(self) -> None:
        """
        Test that when only ``ssl_context`` is passed to a forwarding
        non-HTTPS proxy, no warning or error is emitted.
        """
        ssl_context = ssl.create_default_context(cafile=DEFAULT_CA)
        # No warning/error should be emitted even when ``ssl_context``
        # is provided to a non-HTTPS proxy for some reason.
        proxy = proxy_from_url(
            self.proxy_url,
            ssl_context=ssl_context,
            use_forwarding_for_https=True,
        )
        with proxy:
            resp = proxy.request("GET", self.https_url)
            assert resp.status == 200

    def test_scheme_host_case_insensitive(self) -> None:
        """Assert that upper-case schemes and hosts are normalized."""
        with proxy_from_url(self.proxy_url.upper(), ca_certs=DEFAULT_CA) as http:
            r = http.request("GET", f"{self.http_url.upper()}/")
            assert r.status == 200

            r = http.request("GET", f"{self.https_url.upper()}/")
            assert r.status == 200

    @pytest.mark.parametrize(
        "url, error_msg",
        [
            (
                "127.0.0.1",
                "Proxy URL had no scheme, should start with http:// or https://",
            ),
            (
                "localhost:8080",
                "Proxy URL had no scheme, should start with http:// or https://",
            ),
            (
                "ftp://google.com",
                "Proxy URL had unsupported scheme ftp, should use http:// or https://",
            ),
        ],
    )
    def test_invalid_schema(self, url: str, error_msg: str) -> None:
        with pytest.raises(ProxySchemeUnknown, match=error_msg):
            proxy_from_url(url)


@pytest.mark.skipif(not HAS_IPV6, reason="Only runs on IPv6 systems")
class TestIPv6HTTPProxyManager(IPv6HypercornDummyProxyTestCase):
    @classmethod
    def setup_class(cls) -> None:
        super().setup_class()
        cls.http_url = f"http://{cls.http_host}:{int(cls.http_port)}"
        cls.http_url_alt = f"http://{cls.http_host_alt}:{int(cls.http_port)}"
        cls.https_url = f"https://{cls.https_host}:{int(cls.https_port)}"
        cls.https_url_alt = f"https://{cls.https_host_alt}:{int(cls.https_port)}"
        cls.proxy_url = f"http://[{cls.proxy_host}]:{int(cls.proxy_port)}"

    def test_basic_ipv6_proxy(self) -> None:
        with proxy_from_url(self.proxy_url, ca_certs=DEFAULT_CA) as http:
            r = http.request("GET", f"{self.http_url}/")
            assert r.status == 200

            r = http.request("GET", f"{self.https_url}/")
            assert r.status == 200


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_ctx = urllib3.util.ssl_.create_urllib3_context(verify_flags=0)
        proxy_ctx.load_verify_locations(proxy.ca_certs)
        proxy_fingerprint = self._get_proxy_fingerprint_md5(proxy.ca_certs)
        with proxy_from_url(
            proxy_url,
            ca_certs=server.ca_certs,
            proxy_ssl_context=proxy_ctx,
            proxy_assert_fingerprint=proxy_fingerprint,
        ) as https:
            https.request("GET", destination_url)

    @pytest.mark.parametrize("use_forwarding_for_https", [False, True])
    def test_https_proxy_assert_fingerprint_md5_non_matching(
        self,
        no_san_proxy_with_server: tuple[ServerConfig, ServerConfig],
        use_forwarding_for_https: bool,
    ) -> 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_ctx = urllib3.util.ssl_.create_urllib3_context(verify_flags=0)
        proxy_ctx.load_verify_locations(proxy.ca_certs)
        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=server.ca_certs,
            proxy_ssl_context=proxy_ctx,
            proxy_assert_fingerprint=proxy_fingerprint,
            use_forwarding_for_https=use_forwarding_for_https,
        ) 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)

    @pytest.mark.parametrize("use_forwarding_for_https", [False, True])
    def test_https_proxy_assert_hostname_non_matching(
        self,
        san_proxy_with_server: tuple[ServerConfig, ServerConfig],
        use_forwarding_for_https: bool,
    ) -> 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,
            use_forwarding_for_https=use_forwarding_for_https,
        ) 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:
        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)

    def test_https_proxy_ipv4_san(
        self, ipv4_san_proxy_with_server: tuple[ServerConfig, ServerConfig]
    ) -> None:
        proxy, server = ipv4_san_proxy_with_server
        proxy_url = f"https://{proxy.host}:{proxy.port}"
        destination_url = f"https://{server.host}:{server.port}"
        with proxy_from_url(proxy_url, ca_certs=proxy.ca_certs) as https:
            r = https.request("GET", destination_url)
            assert r.status == 200

    def test_https_proxy_ipv6_san(
        self, ipv6_san_proxy_with_server: tuple[ServerConfig, ServerConfig]
    ) -> None:
        proxy, server = ipv6_san_proxy_with_server
        proxy_url = f"https://[{proxy.host}]:{proxy.port}"
        destination_url = f"https://{server.host}:{server.port}"
        with proxy_from_url(proxy_url, ca_certs=proxy.ca_certs) as https:
            r = https.request("GET", destination_url)
            assert r.status == 200

    @pytest.mark.parametrize("target_scheme", ["http", "https"])
    def test_https_proxy_no_san(
        self,
        no_san_proxy_with_server: tuple[ServerConfig, ServerConfig],
        target_scheme: str,
    ) -> None:
        proxy, server = no_san_proxy_with_server
        proxy_url = f"https://{proxy.host}:{proxy.port}"
        destination_url = f"{target_scheme}://{server.host}:{server.port}"

        with proxy_from_url(proxy_url, ca_certs=proxy.ca_certs) as https:
            with pytest.raises(MaxRetryError) as e:
                https.request("GET", destination_url)
            assert isinstance(e.value.reason, ProxyError)

            ssl_error = e.value.reason.original_error
            assert isinstance(ssl_error, SSLError)
            assert (
                "no appropriate subjectAltName fields were found" in str(ssl_error)
                or "Hostname mismatch, certificate is not valid for 'localhost'"
                in str(ssl_error)
                or "Empty Subject Alternative Name extension" in str(ssl_error)
            )

    def test_https_proxy_no_san_hostname_checks_common_name(
        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_ctx = urllib3.util.ssl_.create_urllib3_context(verify_flags=0)
        proxy_ctx.load_verify_locations(proxy.ca_certs)
        try:
            proxy_ctx.hostname_checks_common_name = True
        # PyPy doesn't like us setting 'hostname_checks_common_name'
        # but also has it enabled by default so we need to handle that.
        except AttributeError:
            pass
        if getattr(proxy_ctx, "hostname_checks_common_name", False) is not True:
            pytest.skip("Test requires 'SSLContext.hostname_checks_common_name=True'")

        with proxy_from_url(
            proxy_url, ca_certs=server.ca_certs, proxy_ssl_context=proxy_ctx
        ) as https:
            https.request("GET", destination_url)

    @pytest.mark.parametrize(
        "target_tls_kwargs",
        [
            pytest.param({"server_hostname": "example.com"}, id="server-hostname"),
            pytest.param({"assert_hostname": "example.com"}, id="assert-hostname"),
            pytest.param({"assert_fingerprint": "00" * 32}, id="assert-fingerprint"),
        ],
    )
    def test_https_proxy_forwarding_ignores_target_identity_settings(
        self,
        san_proxy_with_server: tuple[ServerConfig, ServerConfig],
        target_tls_kwargs: dict[str, typing.Any],
    ) -> None:
        proxy, server = san_proxy_with_server
        proxy_ctx = create_urllib3_context()
        proxy_ctx.load_verify_locations(proxy.ca_certs)

        with proxy_from_url(
            proxy.base_url,
            proxy_ssl_context=proxy_ctx,
            proxy_assert_hostname=proxy.host,
            use_forwarding_for_https=True,
            **target_tls_kwargs,
        ) as https:
            pool = https.connection_from_url(server.base_url)
            with contextlib.closing(pool._new_conn()) as conn:
                conn = typing.cast(HTTPSConnection, conn)
                conn.connect()
                assert isinstance(conn.sock, ssl.SSLSocket)
                assert conn.sock.context is proxy_ctx

    def test_https_proxy_forwarding_ignores_target_client_certificate(
        self,
        san_proxy_with_server: tuple[ServerConfig, ServerConfig],
    ) -> None:
        proxy, server = san_proxy_with_server
        proxy_ctx = create_urllib3_context()
        proxy_ctx.load_verify_locations(proxy.ca_certs)
        missing_cert = str(pathlib.Path(proxy.ca_certs).with_name("missing-cert.pem"))

        with proxy_from_url(
            proxy.base_url,
            proxy_ssl_context=proxy_ctx,
            cert_file=missing_cert,
            use_forwarding_for_https=True,
        ) as https:
            pool = https.connection_from_url(server.base_url)
            with contextlib.closing(pool._new_conn()) as conn:
                conn = typing.cast(HTTPSConnection, conn)
                conn.connect()
                assert isinstance(conn.sock, ssl.SSLSocket)
                assert conn.sock.context is proxy_ctx

    def test_https_proxy_forwarding_ignores_target_tls_policy(
        self,
        san_proxy_with_server: tuple[ServerConfig, ServerConfig],
    ) -> None:
        proxy, server = san_proxy_with_server
        proxy_ctx = create_urllib3_context(cert_reqs=ssl.CERT_REQUIRED)
        proxy_ctx.load_verify_locations(proxy.ca_certs)
        proxy_cert_store_stats = proxy_ctx.cert_store_stats()

        with proxy_from_url(
            proxy.base_url,
            proxy_ssl_context=proxy_ctx,
            cert_reqs=ssl.CERT_NONE,
            ca_certs=DEFAULT_CA,
            use_forwarding_for_https=True,
        ) as https:
            for target_url in (server.base_url, "https://example.com/"):
                pool = https.connection_from_url(target_url)
                with contextlib.closing(pool._new_conn()) as conn:
                    conn = typing.cast(HTTPSConnection, conn)
                    conn.connect()
                    assert isinstance(conn.sock, ssl.SSLSocket)
                    assert conn.sock.context is proxy_ctx

        assert proxy_ctx.verify_mode == ssl.CERT_REQUIRED
        assert proxy_ctx.cert_store_stats() == proxy_cert_store_stats