Fix HTTP tunneling with IPv6 in older Python versions

Author: xNinjaKittyx

.github/workflows/ci.yml

@@ -58,6 +58,11 @@ jobs:
             os: ubuntu-24.04
             experimental: false
             nox-session: test_integration
+          # Test with 3.12.2 for https://github.com/urllib3/urllib3/pull/3620 patch
+          - python-version: "3.12.2"
+            os: ubuntu-24.04
+            experimental: false
+            nox-session: test-3.12
           # pypy
           - python-version: "pypy-3.10"
             os: ubuntu-24.04

changelog/3615.bugfix.rst

@@ -0,0 +1 @@
+Fixed incorrect `CONNECT` statement when using an IPv6 proxy with `connection_from_host`. Previously would not be wrapped in `[]`.

src/urllib3/connection.py

@@ -232,45 +232,94 @@ def set_tunnel(
         super().set_tunnel(host, port=port, headers=headers)
         self._tunnel_scheme = scheme
 
-    if sys.version_info < (3, 11, 4):
-
-        def _tunnel(self) -> None:
-            _MAXLINE = http.client._MAXLINE  # type: ignore[attr-defined]
-            connect = b"CONNECT %s:%d HTTP/1.0\r\n" % (  # type: ignore[str-format]
-                self._tunnel_host.encode("ascii"),  # type: ignore[union-attr]
-                self._tunnel_port,
-            )
-            headers = [connect]
-            for header, value in self._tunnel_headers.items():  # type: ignore[attr-defined]
-                headers.append(f"{header}: {value}\r\n".encode("latin-1"))
-            headers.append(b"\r\n")
-            # Making a single send() call instead of one per line encourages
-            # the host OS to use a more optimal packet size instead of
-            # potentially emitting a series of small packets.
-            self.send(b"".join(headers))
-            del headers
-
-            response = self.response_class(self.sock, method=self._method)  # type: ignore[attr-defined]
-            try:
-                (version, code, message) = response._read_status()  # type: ignore[attr-defined]
-
-                if code != http.HTTPStatus.OK:
-                    self.close()
-                    raise OSError(f"Tunnel connection failed: {code} {message.strip()}")
-                while True:
-                    line = response.fp.readline(_MAXLINE + 1)
-                    if len(line) > _MAXLINE:
-                        raise http.client.LineTooLong("header line")
-                    if not line:
-                        # for sites which EOF without sending a trailer
-                        break
-                    if line in (b"\r\n", b"\n", b""):
-                        break
+    if sys.version_info < (3, 11, 9) or ((3, 12) <= sys.version_info < (3, 12, 3)):
+        # Taken from python/cpython#100986 which was backported in 3.11.9 and 3.12.3.
+        # When using connection_from_host, host will come without brackets.
+        def _wrap_ipv6(self, ip: bytes) -> bytes:
+            if b":" in ip and ip[0] != b"["[0]:
+                return b"[" + ip + b"]"
+            return ip
+
+        if sys.version_info < (3, 11, 9):
+            # `_tunnel` copied from 3.11.13 backporting
+            # https://github.com/python/cpython/commit/0d4026432591d43185568dd31cef6a034c4b9261
+            # and https://github.com/python/cpython/commit/6fbc61070fda2ffb8889e77e3b24bca4249ab4d1
+            def _tunnel(self) -> None:
+                _MAXLINE = http.client._MAXLINE  # type: ignore[attr-defined]
+                connect = b"CONNECT %s:%d HTTP/1.0\r\n" % (  # type: ignore[str-format]
+                    self._wrap_ipv6(self._tunnel_host.encode("ascii")),  # type: ignore[union-attr]
+                    self._tunnel_port,
+                )
+                headers = [connect]
+                for header, value in self._tunnel_headers.items():  # type: ignore[attr-defined]
+                    headers.append(f"{header}: {value}\r\n".encode("latin-1"))
+                headers.append(b"\r\n")
+                # Making a single send() call instead of one per line encourages
+                # the host OS to use a more optimal packet size instead of
+                # potentially emitting a series of small packets.
+                self.send(b"".join(headers))
+                del headers
+
+                response = self.response_class(self.sock, method=self._method)  # type: ignore[attr-defined]
+                try:
+                    (version, code, message) = response._read_status()  # type: ignore[attr-defined]
+
+                    if code != http.HTTPStatus.OK:
+                        self.close()
+                        raise OSError(
+                            f"Tunnel connection failed: {code} {message.strip()}"
+                        )
+                    while True:
+                        line = response.fp.readline(_MAXLINE + 1)
+                        if len(line) > _MAXLINE:
+                            raise http.client.LineTooLong("header line")
+                        if not line:
+                            # for sites which EOF without sending a trailer
+                            break
+                        if line in (b"\r\n", b"\n", b""):
+                            break
+
+                        if self.debuglevel > 0:
+                            print("header:", line.decode())
+                finally:
+                    response.close()
+
+        elif (3, 12) <= sys.version_info < (3, 12, 3):
+            # `_tunnel` copied from 3.12.11 backporting
+            # https://github.com/python/cpython/commit/23aef575c7629abcd4aaf028ebd226fb41a4b3c8
+            def _tunnel(self) -> None:  # noqa: F811
+                connect = b"CONNECT %s:%d HTTP/1.1\r\n" % (  # type: ignore[str-format]
+                    self._wrap_ipv6(self._tunnel_host.encode("idna")),  # type: ignore[union-attr]
+                    self._tunnel_port,
+                )
+                headers = [connect]
+                for header, value in self._tunnel_headers.items():  # type: ignore[attr-defined]
+                    headers.append(f"{header}: {value}\r\n".encode("latin-1"))
+                headers.append(b"\r\n")
+                # Making a single send() call instead of one per line encourages
+                # the host OS to use a more optimal packet size instead of
+                # potentially emitting a series of small packets.
+                self.send(b"".join(headers))
+                del headers
+
+                response = self.response_class(self.sock, method=self._method)  # type: ignore[attr-defined]
+                try:
+                    (version, code, message) = response._read_status()  # type: ignore[attr-defined]
+
+                    self._raw_proxy_headers = http.client._read_headers(response.fp)  # type: ignore[attr-defined]
 
                     if self.debuglevel > 0:
-                        print("header:", line.decode())
-            finally:
-                response.close()
+                        for header in self._raw_proxy_headers:
+                            print("header:", header.decode())
+
+                    if code != http.HTTPStatus.OK:
+                        self.close()
+                        raise OSError(
+                            f"Tunnel connection failed: {code} {message.strip()}"
+                        )
+
+                finally:
+                    response.close()
 
     def connect(self) -> None:
         self.sock = self._new_conn()

test/with_dummyserver/test_socketlevel.py

@@ -22,6 +22,7 @@
 from test import LONG_TIMEOUT, SHORT_TIMEOUT, notWindows, resolvesLocalhostFQDN
 from threading import Event
 from unittest import mock
+from urllib.parse import urlparse
 
 import pytest
 import trustme
@@ -1289,7 +1290,7 @@ def echo_socket_handler(listener: socket.socket) -> None:
             r = conn.urlopen("GET", url, retries=0)
             assert r.status == 200
 
-    def test_connect_ipv6_addr(self) -> None:
+    def test_connect_ipv6_addr_from_host(self) -> None:
         ipv6_addr = "2001:4998:c:a06::2:4008"
 
         def echo_socket_handler(listener: socket.socket) -> None:
@@ -1329,13 +1330,75 @@ def echo_socket_handler(listener: socket.socket) -> None:
 
         with proxy_from_url(base_url, cert_reqs="NONE") as proxy:
             url = f"https://[{ipv6_addr}]"
+
+            # Try with connection_from_host
+            parsed_request_url = urlparse(url)
+
+            conn = proxy.connection_from_host(
+                scheme=parsed_request_url.scheme.lower(),
+                host=parsed_request_url.hostname,
+                port=parsed_request_url.port,
+            )
+            try:
+                with pytest.warns(InsecureRequestWarning):
+                    r = conn.urlopen("GET", url, retries=0)
+                assert r.status == 200
+            except MaxRetryError:
+                pytest.fail(
+                    "Invalid IPv6 format in HTTP CONNECT request when using connection_from_host"
+                )
+
+    def test_connect_ipv6_addr_from_url(self) -> None:
+        ipv6_addr = "2001:4998:c:a06::2:4008"
+
+        def echo_socket_handler(listener: socket.socket) -> None:
+            sock = listener.accept()[0]
+
+            buf = b""
+            while not buf.endswith(b"\r\n\r\n"):
+                buf += sock.recv(65536)
+            s = buf.decode("utf-8")
+
+            if s.startswith(f"CONNECT [{ipv6_addr}]:443"):
+                sock.send(b"HTTP/1.1 200 Connection Established\r\n\r\n")
+                ssl_sock = original_ssl_wrap_socket(
+                    sock,
+                    server_side=True,
+                    keyfile=DEFAULT_CERTS["keyfile"],
+                    certfile=DEFAULT_CERTS["certfile"],
+                )
+                buf = b""
+                while not buf.endswith(b"\r\n\r\n"):
+                    buf += ssl_sock.recv(65536)
+
+                ssl_sock.send(
+                    b"HTTP/1.1 200 OK\r\n"
+                    b"Content-Type: text/plain\r\n"
+                    b"Content-Length: 2\r\n"
+                    b"Connection: close\r\n"
+                    b"\r\n"
+                    b"Hi"
+                )
+                ssl_sock.close()
+            else:
+                sock.close()
+
+        self._start_server(echo_socket_handler)
+        base_url = f"http://{self.host}:{self.port}"
+
+        with proxy_from_url(base_url, cert_reqs="NONE") as proxy:
+            url = f"https://[{ipv6_addr}]"
+
+            # Try with connection_from_url
             conn = proxy.connection_from_url(url)
             try:
                 with pytest.warns(InsecureRequestWarning):
                     r = conn.urlopen("GET", url, retries=0)
                 assert r.status == 200
             except MaxRetryError:
-                pytest.fail("Invalid IPv6 format in HTTP CONNECT request")
+                pytest.fail(
+                    "Invalid IPv6 format in HTTP CONNECT request when using connection_from_url"
+                )
 
     @pytest.mark.parametrize("target_scheme", ["http", "https"])
     def test_https_proxymanager_connected_to_http_proxy(