IPv6 authorities in proxy CONNECT requests are not bracketed

[streambinder]

Bug

In httpcore 1.0.9 (and current master, 10a6582), both tunnel implementations build the CONNECT target without brackets:

target = b"%b:%d" % (self._remote_origin.host, self._remote_origin.port)

• httpcore/_async/http_proxy.py (AsyncTunnelHTTPConnection.handle_async_request)
• httpcore/_sync/http_proxy.py (generated counterpart)

For an unbracketed IPv6 remote origin this produces an ambiguous authority on the wire:

CONNECT 2001:db8::1:443 HTTP/1.1
Host: 2001:db8::1:443

Per RFC 9110, Section 9.3.6, the CONNECT target uses the authority form, and per RFC 3986, Section 3.2.2, an IP literal in an authority must be enclosed in brackets. The correct form is:

CONNECT [2001:db8::1]:443 HTTP/1.1
Host: [2001:db8::1]:443

Proxies that strictly parse the CONNECT authority can reject or misparse HTTPS requests tunnelled to IPv6 destinations.

Prepared fix

I cannot open a pull request or an issue here because both are restricted to collaborators, so the complete fix is included inline as a patch at the end of this post (4 files, based on master 10a6582; it applies cleanly with git apply).

The change is narrow: bracket the host only when formatting the CONNECT authority (when the host contains a colon, i.e. it is an IPv6 literal), and use that authority for both the request target and the Host header. The stored host stays unbracketed for the network connection, TLS server_hostname, and connection-pool identity. DNS names, IPv4 addresses, proxy headers, and explicit ports are unchanged, and there is no public API change. The sync module and sync tests are regenerated with scripts/unasync.py.

Tests

New parametrized regression test test_proxy_tunneling_connect_authority in tests/_async/test_http_proxy.py (sync version generated), using a recording subclass of the existing mock backend/stream and exercising the real tunnel path. It asserts the exact CONNECT request bytes and Host header, covers a compressed IPv6 literal on port 443 and on port 8443, an IPv4-mapped IPv6 literal (::ffff:192.0.2.1), unchanged DNS and IPv4 authorities, and that a successful CONNECT proceeds through the mocked TLS upgrade (hostname unbracketed) and the tunneled request.

• Before the fix: 9 failed, 6 passed (all IPv6 cases failed; DNS/IPv4 passed).
• After the fix: proxy tests 32 passed; full suite 229 passed, 6 xpassed; coverage 100%.
• scripts/check passes: ruff format --diff, mypy, ruff check, scripts/unasync.py --check.

Could a maintainer please apply the patch below (or open a PR from it), or let me know the preferred way to contribute this while PR and issue creation are collaborators-only?

Patch

Fix IPv6 authorities in proxy CONNECT requests (4 files)

diff --git a/httpcore/_async/http_proxy.py b/httpcore/_async/http_proxy.py
index cc9d920..42d9f3f 100644
--- a/httpcore/_async/http_proxy.py
+++ b/httpcore/_async/http_proxy.py
@@ -268,7 +268,22 @@ class AsyncTunnelHTTPConnection(AsyncConnectionInterface):
 
         async with self._connect_lock:
             if not self._connected:
-                target = b"%b:%d" % (self._remote_origin.host, self._remote_origin.port)
+                # The CONNECT target must use the authority form.
+                # IPv6 literals must be enclosed in brackets, for example
+                # "[2001:db8::1]:443". See RFC 9110, Section 9.3.6 and
+                # RFC 3986, Section 3.2.2. The remote origin host itself
+                # remains unbracketed, as it is used for the network
+                # connection and for TLS hostname verification.
+                if b":" in self._remote_origin.host:
+                    target = b"[%b]:%d" % (
+                        self._remote_origin.host,
+                        self._remote_origin.port,
+                    )
+                else:
+                    target = b"%b:%d" % (
+                        self._remote_origin.host,
+                        self._remote_origin.port,
+                    )
 
                 connect_url = URL(
                     scheme=self._proxy_origin.scheme,
diff --git a/httpcore/_sync/http_proxy.py b/httpcore/_sync/http_proxy.py
index ecca88f..ee52905 100644
--- a/httpcore/_sync/http_proxy.py
+++ b/httpcore/_sync/http_proxy.py
@@ -268,7 +268,22 @@ class TunnelHTTPConnection(ConnectionInterface):
 
         with self._connect_lock:
             if not self._connected:
-                target = b"%b:%d" % (self._remote_origin.host, self._remote_origin.port)
+                # The CONNECT target must use the authority form.
+                # IPv6 literals must be enclosed in brackets, for example
+                # "[2001:db8::1]:443". See RFC 9110, Section 9.3.6 and
+                # RFC 3986, Section 3.2.2. The remote origin host itself
+                # remains unbracketed, as it is used for the network
+                # connection and for TLS hostname verification.
+                if b":" in self._remote_origin.host:
+                    target = b"[%b]:%d" % (
+                        self._remote_origin.host,
+                        self._remote_origin.port,
+                    )
+                else:
+                    target = b"%b:%d" % (
+                        self._remote_origin.host,
+                        self._remote_origin.port,
+                    )
 
                 connect_url = URL(
                     scheme=self._proxy_origin.scheme,
diff --git a/tests/_async/test_http_proxy.py b/tests/_async/test_http_proxy.py
index 84a984b..b931176 100644
--- a/tests/_async/test_http_proxy.py
+++ b/tests/_async/test_http_proxy.py
@@ -152,6 +152,47 @@ class HTTP1ThenHTTP2Backend(AsyncMockBackend):
         return HTTP1ThenHTTP2Stream(list(self._buffer))
 
 
+# A mock stream and backend that record the bytes written to the proxy
+# and the hostname used for the TLS upgrade, so that tests can assert
+# the exact CONNECT request that is sent over the wire.
+class RecordingStream(AsyncMockStream):
+    def __init__(self, buffer: typing.List[bytes]) -> None:
+        super().__init__(buffer)
+        self.writes: typing.List[bytes] = []
+        self.tls_server_hostname: typing.Optional[str] = None
+
+    async def write(
+        self, buffer: bytes, timeout: typing.Optional[float] = None
+    ) -> None:
+        self.writes.append(buffer)
+
+    async def start_tls(
+        self,
+        ssl_context: ssl.SSLContext,
+        server_hostname: typing.Optional[str] = None,
+        timeout: typing.Optional[float] = None,
+    ) -> AsyncNetworkStream:
+        self.tls_server_hostname = server_hostname
+        return self
+
+
+class RecordingBackend(AsyncMockBackend):
+    def __init__(self, buffer: typing.List[bytes]) -> None:
+        super().__init__(buffer)
+        self.stream: typing.Optional[RecordingStream] = None
+
+    async def connect_tcp(
+        self,
+        host: str,
+        port: int,
+        timeout: typing.Optional[float] = None,
+        local_address: typing.Optional[str] = None,
+        socket_options: typing.Optional[typing.Iterable[SOCKET_OPTION]] = None,
+    ) -> AsyncNetworkStream:
+        self.stream = RecordingStream(list(self._buffer))
+        return self.stream
+
+
 @pytest.mark.anyio
 async def test_proxy_tunneling_http2():
     """
@@ -268,6 +309,96 @@ async def test_proxy_tunneling_with_auth():
         assert response.content == b"Hello, world!"
 
 
+@pytest.mark.anyio
+@pytest.mark.parametrize(
+    "url, connect_bytes, tls_server_hostname",
+    [
+        # IPv6 literal, compressed form, default HTTPS port.
+        # RFC 9110, Section 9.3.6 requires the CONNECT target to use
+        # the authority form, and RFC 3986, Section 3.2.2 requires
+        # an IP literal in an authority to be enclosed in brackets.
+        (
+            "https://[2001:db8::1]/",
+            b"CONNECT [2001:db8::1]:443 HTTP/1.1\r\n"
+            b"Host: [2001:db8::1]:443\r\n"
+            b"Accept: */*\r\n\r\n",
+            "2001:db8::1",
+        ),
+        # IPv6 literal, compressed form, non-default port.
+        (
+            "https://[2001:db8::1]:8443/",
+            b"CONNECT [2001:db8::1]:8443 HTTP/1.1\r\n"
+            b"Host: [2001:db8::1]:8443\r\n"
+            b"Accept: */*\r\n\r\n",
+            "2001:db8::1",
+        ),
+        # IPv4-mapped IPv6 literal, default HTTPS port.
+        (
+            "https://[::ffff:192.0.2.1]/",
+            b"CONNECT [::ffff:192.0.2.1]:443 HTTP/1.1\r\n"
+            b"Host: [::ffff:192.0.2.1]:443\r\n"
+            b"Accept: */*\r\n\r\n",
+            "::ffff:192.0.2.1",
+        ),
+        # DNS name, default HTTPS port. The authority is unchanged.
+        (
+            "https://example.com/",
+            b"CONNECT example.com:443 HTTP/1.1\r\n"
+            b"Host: example.com:443\r\n"
+            b"Accept: */*\r\n\r\n",
+            "example.com",
+        ),
+        # IPv4 address, non-default port. The authority is unchanged.
+        (
+            "https://127.0.0.1:8443/",
+            b"CONNECT 127.0.0.1:8443 HTTP/1.1\r\n"
+            b"Host: 127.0.0.1:8443\r\n"
+            b"Accept: */*\r\n\r\n",
+            "127.0.0.1",
+        ),
+    ],
+)
+async def test_proxy_tunneling_connect_authority(
+    url: str, connect_bytes: bytes, tls_server_hostname: str
+) -> None:
+    """
+    The CONNECT request target and Host header must be a valid authority.
+
+    IPv6 literals must be bracketed, while DNS names and IPv4 addresses
+    must be sent unchanged. The hostname used for the TLS upgrade after
+    a successful CONNECT must remain unbracketed.
+    """
+    network_backend = RecordingBackend(
+        [
+            # The initial response to the proxy CONNECT
+            b"HTTP/1.1 200 OK\r\n\r\n",
+            # The actual response from the remote server
+            b"HTTP/1.1 200 OK\r\n",
+            b"Content-Type: plain/text\r\n",
+            b"Content-Length: 13\r\n",
+            b"\r\n",
+            b"Hello, world!",
+        ]
+    )
+
+    async with AsyncConnectionPool(
+        proxy=Proxy("http://localhost:8080/"),
+        network_backend=network_backend,
+    ) as proxy:
+        response = await proxy.request("GET", url)
+        assert response.status == 200
+        assert response.content == b"Hello, world!"
+
+    assert network_backend.stream is not None
+    # The first bytes written are the CONNECT request itself,
+    # including both the request target and the Host header.
+    assert network_backend.stream.writes[0] == connect_bytes
+    # The tunneled request is sent after the CONNECT succeeds.
+    assert network_backend.stream.writes[3].startswith(b"GET / HTTP/1.1\r\n")
+    # The TLS hostname must not include brackets.
+    assert network_backend.stream.tls_server_hostname == tls_server_hostname
+
+
 def test_proxy_headers():
     proxy = Proxy(
         url="http://localhost:8080/",
diff --git a/tests/_sync/test_http_proxy.py b/tests/_sync/test_http_proxy.py
index 966672d..f01e9c6 100644
--- a/tests/_sync/test_http_proxy.py
+++ b/tests/_sync/test_http_proxy.py
@@ -152,6 +152,47 @@ class HTTP1ThenHTTP2Backend(MockBackend):
         return HTTP1ThenHTTP2Stream(list(self._buffer))
 
 
+# A mock stream and backend that record the bytes written to the proxy
+# and the hostname used for the TLS upgrade, so that tests can assert
+# the exact CONNECT request that is sent over the wire.
+class RecordingStream(MockStream):
+    def __init__(self, buffer: typing.List[bytes]) -> None:
+        super().__init__(buffer)
+        self.writes: typing.List[bytes] = []
+        self.tls_server_hostname: typing.Optional[str] = None
+
+    def write(
+        self, buffer: bytes, timeout: typing.Optional[float] = None
+    ) -> None:
+        self.writes.append(buffer)
+
+    def start_tls(
+        self,
+        ssl_context: ssl.SSLContext,
+        server_hostname: typing.Optional[str] = None,
+        timeout: typing.Optional[float] = None,
+    ) -> NetworkStream:
+        self.tls_server_hostname = server_hostname
+        return self
+
+
+class RecordingBackend(MockBackend):
+    def __init__(self, buffer: typing.List[bytes]) -> None:
+        super().__init__(buffer)
+        self.stream: typing.Optional[RecordingStream] = None
+
+    def connect_tcp(
+        self,
+        host: str,
+        port: int,
+        timeout: typing.Optional[float] = None,
+        local_address: typing.Optional[str] = None,
+        socket_options: typing.Optional[typing.Iterable[SOCKET_OPTION]] = None,
+    ) -> NetworkStream:
+        self.stream = RecordingStream(list(self._buffer))
+        return self.stream
+
+
 
 def test_proxy_tunneling_http2():
     """
@@ -268,6 +309,96 @@ def test_proxy_tunneling_with_auth():
         assert response.content == b"Hello, world!"
 
 
+
+@pytest.mark.parametrize(
+    "url, connect_bytes, tls_server_hostname",
+    [
+        # IPv6 literal, compressed form, default HTTPS port.
+        # RFC 9110, Section 9.3.6 requires the CONNECT target to use
+        # the authority form, and RFC 3986, Section 3.2.2 requires
+        # an IP literal in an authority to be enclosed in brackets.
+        (
+            "https://[2001:db8::1]/",
+            b"CONNECT [2001:db8::1]:443 HTTP/1.1\r\n"
+            b"Host: [2001:db8::1]:443\r\n"
+            b"Accept: */*\r\n\r\n",
+            "2001:db8::1",
+        ),
+        # IPv6 literal, compressed form, non-default port.
+        (
+            "https://[2001:db8::1]:8443/",
+            b"CONNECT [2001:db8::1]:8443 HTTP/1.1\r\n"
+            b"Host: [2001:db8::1]:8443\r\n"
+            b"Accept: */*\r\n\r\n",
+            "2001:db8::1",
+        ),
+        # IPv4-mapped IPv6 literal, default HTTPS port.
+        (
+            "https://[::ffff:192.0.2.1]/",
+            b"CONNECT [::ffff:192.0.2.1]:443 HTTP/1.1\r\n"
+            b"Host: [::ffff:192.0.2.1]:443\r\n"
+            b"Accept: */*\r\n\r\n",
+            "::ffff:192.0.2.1",
+        ),
+        # DNS name, default HTTPS port. The authority is unchanged.
+        (
+            "https://example.com/",
+            b"CONNECT example.com:443 HTTP/1.1\r\n"
+            b"Host: example.com:443\r\n"
+            b"Accept: */*\r\n\r\n",
+            "example.com",
+        ),
+        # IPv4 address, non-default port. The authority is unchanged.
+        (
+            "https://127.0.0.1:8443/",
+            b"CONNECT 127.0.0.1:8443 HTTP/1.1\r\n"
+            b"Host: 127.0.0.1:8443\r\n"
+            b"Accept: */*\r\n\r\n",
+            "127.0.0.1",
+        ),
+    ],
+)
+def test_proxy_tunneling_connect_authority(
+    url: str, connect_bytes: bytes, tls_server_hostname: str
+) -> None:
+    """
+    The CONNECT request target and Host header must be a valid authority.
+
+    IPv6 literals must be bracketed, while DNS names and IPv4 addresses
+    must be sent unchanged. The hostname used for the TLS upgrade after
+    a successful CONNECT must remain unbracketed.
+    """
+    network_backend = RecordingBackend(
+        [
+            # The initial response to the proxy CONNECT
+            b"HTTP/1.1 200 OK\r\n\r\n",
+            # The actual response from the remote server
+            b"HTTP/1.1 200 OK\r\n",
+            b"Content-Type: plain/text\r\n",
+            b"Content-Length: 13\r\n",
+            b"\r\n",
+            b"Hello, world!",
+        ]
+    )
+
+    with ConnectionPool(
+        proxy=Proxy("http://localhost:8080/"),
+        network_backend=network_backend,
+    ) as proxy:
+        response = proxy.request("GET", url)
+        assert response.status == 200
+        assert response.content == b"Hello, world!"
+
+    assert network_backend.stream is not None
+    # The first bytes written are the CONNECT request itself,
+    # including both the request target and the Host header.
+    assert network_backend.stream.writes[0] == connect_bytes
+    # The tunneled request is sent after the CONNECT succeeds.
+    assert network_backend.stream.writes[3].startswith(b"GET / HTTP/1.1\r\n")
+    # The TLS hostname must not include brackets.
+    assert network_backend.stream.tls_server_hostname == tls_server_hostname
+
+
 def test_proxy_headers():
     proxy = Proxy(
         url="http://localhost:8080/",