Bracket IPv6 address literals when rendering URLs and authorities

src/httpcore2/httpcore2/_async/http_proxy.py

@@ -15,6 +15,7 @@
     enforce_bytes,
     enforce_headers,
     enforce_url,
+    format_host,
 )
 from .._ssl import default_ssl_context
 from .._synchronization import AsyncLock
@@ -261,7 +262,7 @@ async def handle_async_request(self, request: Request) -> Response:
 
         async with self._connect_lock:
             if not self._connected:
-                target = b"%b:%d" % (self._remote_origin.host, self._remote_origin.port)
+                target = b"%b:%d" % (format_host(self._remote_origin.host), self._remote_origin.port)
 
                 connect_url = URL(
                     scheme=self._proxy_origin.scheme,

src/httpcore2/httpcore2/_models.py

@@ -107,6 +107,22 @@ def enforce_stream(
 }
 
 
+def format_host(host: bytes) -> bytes:
+    """
+    Wrap an IPv6 address literal in square brackets.
+
+    `URL.host` and `Origin.host` hold the host in the form required to establish
+    a connection, which for an IPv6 address is the bare address. Wherever the
+    host is instead rendered as part of a URL or of an authority, an IPv6
+    address literal must be bracketed, as an `IP-literal`.
+
+    * https://tools.ietf.org/html/rfc3986#section-3.2.2
+    """
+    # A colon cannot occur in a registered name or in an IPv4 address, so its
+    # presence is enough to tell an IPv6 address apart.
+    return b"[%b]" % host if b":" in host else host
+
+
 def include_request_headers(
     headers: list[tuple[bytes, bytes]],
     *,
@@ -118,9 +134,9 @@ def include_request_headers(
     if b"host" not in headers_set:
         default_port = DEFAULT_PORTS.get(url.scheme)
         if url.port is None or url.port == default_port:
-            header_value = url.host
+            header_value = format_host(url.host)
         else:
-            header_value = b"%b:%d" % (url.host, url.port)
+            header_value = b"%b:%d" % (format_host(url.host), url.port)
         headers = [(b"Host", header_value)] + headers
 
     if content is not None and b"content-length" not in headers_set and b"transfer-encoding" not in headers_set:
@@ -171,7 +187,7 @@ def __eq__(self, other: typing.Any) -> bool:
 
     def __str__(self) -> str:
         scheme = self.scheme.decode("ascii")
-        host = self.host.decode("ascii")
+        host = format_host(self.host).decode("ascii")
         port = str(self.port)
         return f"{scheme}://{host}:{port}"
 
@@ -296,8 +312,8 @@ def __eq__(self, other: typing.Any) -> bool:
 
     def __bytes__(self) -> bytes:
         if self.port is None:
-            return b"%b://%b%b" % (self.scheme, self.host, self.target)
-        return b"%b://%b:%d%b" % (self.scheme, self.host, self.port, self.target)
+            return b"%b://%b%b" % (self.scheme, format_host(self.host), self.target)
+        return b"%b://%b:%d%b" % (self.scheme, format_host(self.host), self.port, self.target)
 
     def __repr__(self) -> str:
         return (

src/httpcore2/httpcore2/_sync/http_proxy.py

@@ -15,6 +15,7 @@
     enforce_bytes,
     enforce_headers,
     enforce_url,
+    format_host,
 )
 from .._ssl import default_ssl_context
 from .._synchronization import Lock
@@ -261,7 +262,7 @@ def handle_request(self, request: Request) -> Response:
 
         with self._connect_lock:
             if not self._connected:
-                target = b"%b:%d" % (self._remote_origin.host, self._remote_origin.port)
+                target = b"%b:%d" % (format_host(self._remote_origin.host), self._remote_origin.port)
 
                 connect_url = URL(
                     scheme=self._proxy_origin.scheme,

tests/httpcore2/_async/test_http_proxy.py

@@ -105,6 +105,90 @@ async def test_proxy_tunneling() -> None:
         assert not proxy.connections[0].can_handle_request(Origin(b"https", b"other.com", 443))
 
 
+class RecordingStream(AsyncMockStream):
+    """A mock stream that keeps everything written to it."""
+
+    def __init__(self, buffer: list[bytes], http2: bool = False) -> None:
+        super().__init__(buffer, http2)
+        self.written = b""
+
+    async def write(self, buffer: bytes, timeout: float | None = None) -> None:
+        self.written += buffer
+
+
+class RecordingBackend(AsyncMockBackend):
+    def __init__(self, buffer: list[bytes], http2: bool = False) -> None:
+        super().__init__(buffer, http2)
+        self.stream = RecordingStream(list(buffer), http2=http2)
+
+    async def connect_tcp(
+        self,
+        host: str,
+        port: int,
+        timeout: float | None = None,
+        local_address: str | None = None,
+        socket_options: typing.Iterable[SOCKET_OPTION] | None = None,
+    ) -> AsyncMockStream:
+        return self.stream
+
+
+@pytest.mark.anyio
+async def test_proxy_forwarding_to_ipv6_host() -> None:
+    """
+    Send an HTTP request for an IPv6 address literal via a proxy.
+
+    The absolute-form request target is a URI, so the address is bracketed.
+    """
+    network_backend = RecordingBackend(
+        [
+            b"HTTP/1.1 200 OK\r\n",
+            b"Content-Length: 0\r\n",
+            b"\r\n",
+        ]
+    )
+
+    async with AsyncConnectionPool(
+        proxy=Proxy("http://localhost:8080/"),
+        network_backend=network_backend,
+    ) as proxy:
+        response = await proxy.request("GET", "http://[::1]:1234/")
+        assert response.status == 200
+
+    request_line = network_backend.stream.written.split(b"\r\n")[0]
+    assert request_line == b"GET http://[::1]:1234/ HTTP/1.1"
+    assert b"\r\nHost: [::1]:1234\r\n" in network_backend.stream.written
+
+
+@pytest.mark.anyio
+async def test_proxy_tunneling_to_ipv6_host() -> None:
+    """
+    Send an HTTPS request for an IPv6 address literal via a proxy.
+
+    The CONNECT target is in authority-form, so the address is bracketed.
+    """
+    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-Length: 0\r\n",
+            b"\r\n",
+        ]
+    )
+
+    async with AsyncConnectionPool(
+        proxy=Proxy("http://localhost:8080/"),
+        network_backend=network_backend,
+    ) as proxy:
+        response = await proxy.request("GET", "https://[::1]:8443/")
+        assert response.status == 200
+
+    request_line = network_backend.stream.written.split(b"\r\n")[0]
+    assert request_line == b"CONNECT [::1]:8443 HTTP/1.1"
+    assert b"\r\nHost: [::1]:8443\r\n" in network_backend.stream.written
+
+
 # We need to adapt the mock backend here slightly in order to deal
 # with the proxy case. We do not want the initial connection to the proxy
 # to indicate an HTTP/2 connection, but we do want it to indicate HTTP/2

tests/httpcore2/_sync/test_http_proxy.py

@@ -105,6 +105,90 @@ def test_proxy_tunneling() -> None:
         assert not proxy.connections[0].can_handle_request(Origin(b"https", b"other.com", 443))
 
 
+class RecordingStream(MockStream):
+    """A mock stream that keeps everything written to it."""
+
+    def __init__(self, buffer: list[bytes], http2: bool = False) -> None:
+        super().__init__(buffer, http2)
+        self.written = b""
+
+    def write(self, buffer: bytes, timeout: float | None = None) -> None:
+        self.written += buffer
+
+
+class RecordingBackend(MockBackend):
+    def __init__(self, buffer: list[bytes], http2: bool = False) -> None:
+        super().__init__(buffer, http2)
+        self.stream = RecordingStream(list(buffer), http2=http2)
+
+    def connect_tcp(
+        self,
+        host: str,
+        port: int,
+        timeout: float | None = None,
+        local_address: str | None = None,
+        socket_options: typing.Iterable[SOCKET_OPTION] | None = None,
+    ) -> MockStream:
+        return self.stream
+
+
+
+def test_proxy_forwarding_to_ipv6_host() -> None:
+    """
+    Send an HTTP request for an IPv6 address literal via a proxy.
+
+    The absolute-form request target is a URI, so the address is bracketed.
+    """
+    network_backend = RecordingBackend(
+        [
+            b"HTTP/1.1 200 OK\r\n",
+            b"Content-Length: 0\r\n",
+            b"\r\n",
+        ]
+    )
+
+    with ConnectionPool(
+        proxy=Proxy("http://localhost:8080/"),
+        network_backend=network_backend,
+    ) as proxy:
+        response = proxy.request("GET", "http://[::1]:1234/")
+        assert response.status == 200
+
+    request_line = network_backend.stream.written.split(b"\r\n")[0]
+    assert request_line == b"GET http://[::1]:1234/ HTTP/1.1"
+    assert b"\r\nHost: [::1]:1234\r\n" in network_backend.stream.written
+
+
+
+def test_proxy_tunneling_to_ipv6_host() -> None:
+    """
+    Send an HTTPS request for an IPv6 address literal via a proxy.
+
+    The CONNECT target is in authority-form, so the address is bracketed.
+    """
+    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-Length: 0\r\n",
+            b"\r\n",
+        ]
+    )
+
+    with ConnectionPool(
+        proxy=Proxy("http://localhost:8080/"),
+        network_backend=network_backend,
+    ) as proxy:
+        response = proxy.request("GET", "https://[::1]:8443/")
+        assert response.status == 200
+
+    request_line = network_backend.stream.written.split(b"\r\n")[0]
+    assert request_line == b"CONNECT [::1]:8443 HTTP/1.1"
+    assert b"\r\nHost: [::1]:8443\r\n" in network_backend.stream.written
+
+
 # We need to adapt the mock backend here slightly in order to deal
 # with the proxy case. We do not want the initial connection to the proxy
 # to indicate an HTTP/2 connection, but we do want it to indicate HTTP/2

tests/httpcore2/test_models.py

@@ -19,6 +19,29 @@ def test_url_with_port() -> None:
     assert bytes(url) == b"https://www.example.com:443/"
 
 
+def test_url_with_ipv6_host() -> None:
+    """
+    `URL.host` holds the bare address, as needed to establish a connection, but
+    rendering the URL brackets it.
+    """
+    url = httpcore2.URL("https://[::1]/")
+    assert url == httpcore2.URL(scheme="https", host="::1", port=None, target="/")
+    assert url.host == b"::1"
+    assert bytes(url) == b"https://[::1]/"
+
+    url = httpcore2.URL("https://[::1]:8443/")
+    assert url.host == b"::1"
+    assert bytes(url) == b"https://[::1]:8443/"
+
+
+def test_url_origin_ipv6() -> None:
+    url = httpcore2.URL("https://[::1]:8443/")
+    origin = url.origin
+    assert origin == httpcore2.Origin(scheme=b"https", host=b"::1", port=8443)
+    assert origin.host == b"::1"
+    assert str(origin) == "https://[::1]:8443"
+
+
 def test_url_with_invalid_argument() -> None:
     with pytest.raises(TypeError) as exc_info:
         httpcore2.URL(123)  # type: ignore