Harden URL reads against rebinding and compression bombs

README.rst

@@ -240,8 +240,11 @@ You can use the json2xml library in the following ways:
     print(json2xml.Json2xml(data).to_xml())
 
 URL reads accept only credential-free HTTP(S), reject redirects and non-public
-destinations by default, and stop after 10 MiB of decoded content. Trusted
-library callers can opt into a private endpoint or choose a smaller limit:
+destinations by default, pin public connections to their validated DNS address,
+and stop after 10 MiB of encoded or decoded content. Gzip and deflate responses
+honor ``Content-Length`` and are decoded incrementally; other content encodings
+are rejected. Trusted library callers can opt into a private endpoint with the
+boolean ``True`` or choose a smaller limit:
 
 .. code-block:: python
 

RELEASE_NOTES.md

@@ -1,3 +1,24 @@
+# Unreleased
+
+## Security
+
+- Public URL reads now pin connections to a validated DNS address while preserving the requested Host header and HTTPS certificate hostname, preventing DNS-rebinding bypasses of private-network blocking.
+- Gzip and deflate responses honor valid `Content-Length` values and are decoded incrementally with bounded encoded input and zlib output, so compressed bodies cannot consume unbounded network I/O or memory before rejection.
+- `allow_private_networks` accepts only `True` or `False`; strings, numbers, and other values now raise `URLReadError` instead of relying on truthiness.
+- Malformed Unicode hostnames now consistently raise `URLReadError` when IDNA encoding or DNS resolution fails.
+
+## Migration guidance
+
+- URL reads continue to reject redirects and private, loopback, link-local, and other non-global destinations by default. Trusted callers that intentionally read a private endpoint must pass `allow_private_networks=True` as an actual boolean.
+- Encoded and decoded URL responses are limited to 10 MiB by default. Set `max_response_bytes` explicitly when a trusted endpoint needs a different positive limit.
+- URL response compression is limited to `gzip`, `x-gzip`, `deflate`, and `identity`. Servers returning Brotli, Zstandard, stacked encodings, or malformed compressed streams must be reconfigured or read outside `readfromurl()`.
+- XML 1.0-forbidden characters are now rejected instead of being serialized. Low-level serializer functions raise `ValueError`; `Json2xml.to_xml()` raises `InvalidDataError`.
+
+## Release prerequisite
+
+Publish a compatible accelerator version newer than `json2xml-rs==0.4.2`, then raise the `json2xml[fast]` and `uv.lock` minimum to that published version. Until then, the compatibility probe safely disables 0.4.2 and uses the Python backend.
+
+
 # json2xml 6.5.0
 
 Released 2026-07-15.

json2xml/utils.py

@@ -3,16 +3,18 @@
 
 import json
 import socket
+import zlib
 from ipaddress import ip_address
 from typing import Any
-from urllib.parse import urlsplit
+from urllib.parse import SplitResult, urlsplit, urlunsplit
 
 __lazy_modules__ = ["urllib3"]
 
 from .types import JSONValue
 
 DEFAULT_URL_TIMEOUT: Any | None = None
 DEFAULT_MAX_RESPONSE_BYTES = 10 * 1024 * 1024
+COMPRESSED_READ_CHUNK_BYTES = 64 * 1024
 _HTTP: Any | None = None
 
 
@@ -59,11 +61,11 @@ def readfromjson(filename: str) -> JSONValue:
 
 
 # @lat: [[behavior#URL security boundaries]]
-def _validate_url(url: str, allow_private_networks: bool) -> None:
-    """Reject URL forms that can escape the intended public HTTP boundary."""
+def _validate_url(url: str) -> SplitResult:
+    """Validate the URL form without performing network access."""
     try:
         parsed = urlsplit(url)
-        port = parsed.port
+        _ = parsed.port
     except (TypeError, ValueError) as error:
         raise URLReadError("URL is not valid") from error
 
@@ -73,28 +75,203 @@ def _validate_url(url: str, allow_private_networks: bool) -> None:
         raise URLReadError("URL must not contain credentials")
     if parsed.hostname is None:
         raise URLReadError("URL must include a hostname")
+    return parsed
+
+
+def _resolve_validated_address(
+    parsed: SplitResult, allow_private_networks: bool
+) -> str | None:
+    """Resolve and validate the public address used for the connection."""
     if allow_private_networks:
-        return
+        return None
 
+    assert parsed.hostname is not None
     hostname = parsed.hostname
+    port = parsed.port or (443 if parsed.scheme == "https" else 80)
     try:
-        addresses = {ip_address(hostname)}
+        addresses = [ip_address(hostname)]
     except ValueError:
         try:
             address_info = socket.getaddrinfo(
                 hostname,
-                port or (443 if parsed.scheme == "https" else 80),
+                port,
                 type=socket.SOCK_STREAM,
             )
-        except OSError as error:
+        except (OSError, UnicodeError) as error:
             raise URLReadError("URL hostname could not be resolved") from error
-        addresses = {
+        addresses = [
             ip_address(str(info[4][0]).split("%", 1)[0])
             for info in address_info
-        }
+        ]
 
     if not addresses or any(not address.is_global for address in addresses):
         raise URLReadError("URL must resolve only to a public network address")
+    return str(addresses[0])
+
+
+def _request_via_validated_address(
+    http: Any,
+    parsed: SplitResult,
+    validated_address: str,
+    params: dict[str, str] | None,
+    timeout: Any,
+) -> Any:
+    """Issue a GET directly to an address already validated as public."""
+    assert parsed.hostname is not None
+    try:
+        hostname = parsed.hostname.encode("idna").decode("ascii")
+    except UnicodeError as error:
+        raise URLReadError("URL hostname could not be resolved") from error
+
+    port = parsed.port or (443 if parsed.scheme == "https" else 80)
+    authority = f"[{hostname}]" if ":" in hostname else hostname
+    if parsed.port is not None:
+        authority = f"{authority}:{parsed.port}"
+    pool_kwargs = None
+    if parsed.scheme == "https":
+        pool_kwargs = {
+            "assert_hostname": hostname,
+            "server_hostname": hostname,
+        }
+    pool = http.connection_from_host(
+        validated_address,
+        port=port,
+        scheme=parsed.scheme,
+        pool_kwargs=pool_kwargs,
+    )
+    request_target = urlunsplit(("", "", parsed.path or "/", parsed.query, ""))
+    return pool.request(
+        "GET",
+        request_target,
+        fields=params,
+        headers={"Host": authority},
+        timeout=timeout,
+        retries=False,
+        redirect=False,
+        preload_content=False,
+    )
+
+
+def _has_zlib_header(data: bytes) -> bool:
+    """Return whether bytes begin with an RFC 1950 zlib header."""
+    if len(data) < 2:
+        return False
+    compression_method, flags = data[0], data[1]
+    return (
+        compression_method & 0x0F == 8
+        and (compression_method << 8 | flags) % 31 == 0
+    )
+
+
+def _compression_decoder(encoding: str, first_chunk: bytes) -> Any:
+    """Create a bounded-output decoder for supported content encodings."""
+    if encoding in {"gzip", "x-gzip"}:
+        return zlib.decompressobj(16 + zlib.MAX_WBITS)
+    if encoding == "deflate":
+        window_bits = (
+            zlib.MAX_WBITS if _has_zlib_header(first_chunk) else -zlib.MAX_WBITS
+        )
+        return zlib.decompressobj(window_bits)
+    raise URLReadError(f"Unsupported Content-Encoding: {encoding}")
+
+
+def _decompress_with_limit(
+    response: Any,
+    decoder: Any,
+    first_chunk: bytes,
+    max_response_bytes: int,
+    content_length: int | None,
+) -> bytes:
+    """Decode a compressed body with encoded and decoded byte limits."""
+    response_data = bytearray()
+    compressed_bytes = len(first_chunk)
+    compressed_chunk = first_chunk
+    try:
+        while compressed_chunk:
+            if compressed_bytes > max_response_bytes:
+                raise URLReadError("URL response exceeds maximum size")
+            if content_length is not None and compressed_bytes > content_length:
+                raise URLReadError("URL response exceeds declared Content-Length")
+
+            pending = compressed_chunk
+            while pending:
+                remaining_bytes = max_response_bytes + 1 - len(response_data)
+                decoded_chunk = decoder.decompress(pending, remaining_bytes)
+                response_data.extend(decoded_chunk)
+                if len(response_data) > max_response_bytes:
+                    raise URLReadError("URL response exceeds maximum size")
+                pending = decoder.unconsumed_tail
+
+            if content_length is not None and compressed_bytes == content_length:
+                break
+            compressed_bytes_max = (
+                content_length
+                if content_length is not None
+                else max_response_bytes + 1
+            )
+            read_size = min(
+                COMPRESSED_READ_CHUNK_BYTES,
+                compressed_bytes_max - compressed_bytes,
+            )
+            compressed_chunk = response.read(read_size, decode_content=False)
+            compressed_bytes += len(compressed_chunk)
+    except zlib.error as error:
+        raise URLReadError("URL returned invalid compressed data") from error
+
+    if content_length is not None and compressed_bytes != content_length:
+        raise URLReadError("URL response did not match Content-Length")
+    if not decoder.eof or decoder.unused_data:
+        raise URLReadError("URL returned invalid compressed data")
+    return bytes(response_data)
+
+
+def _read_response_data(
+    response: Any,
+    max_response_bytes: int,
+    content_length: int | None,
+) -> bytes:
+    """Read a response without allowing encoded or decoded output above the limit."""
+    encoding = response.headers.get("Content-Encoding", "").strip().lower()
+    if encoding in {"", "identity"}:
+        response_data = response.read(
+            max_response_bytes + 1,
+            decode_content=False,
+        )
+        if len(response_data) > max_response_bytes:
+            raise URLReadError("URL response exceeds maximum size")
+        return response_data
+
+    compressed_bytes_max = (
+        content_length if content_length is not None else max_response_bytes + 1
+    )
+    first_chunk = response.read(
+        min(COMPRESSED_READ_CHUNK_BYTES, compressed_bytes_max),
+        decode_content=False,
+    )
+    decoder = _compression_decoder(encoding, first_chunk)
+    return _decompress_with_limit(
+        response,
+        decoder,
+        first_chunk,
+        max_response_bytes,
+        content_length,
+    )
+
+
+def _validated_content_length(response: Any, max_response_bytes: int) -> int | None:
+    """Parse and bound a declared encoded response length."""
+    content_length = response.headers.get("Content-Length")
+    if content_length is None:
+        return None
+    try:
+        parsed_length = int(content_length)
+    except ValueError as error:
+        raise URLReadError("URL returned an invalid Content-Length") from error
+    if parsed_length < 0:
+        raise URLReadError("URL returned an invalid Content-Length")
+    if parsed_length > max_response_bytes:
+        raise URLReadError("URL response exceeds maximum size")
+    return parsed_length
 
 
 def readfromurl(
@@ -109,40 +286,53 @@ def readfromurl(
     Private-network access is available only through the explicit trusted-caller
     opt-in. Redirects and embedded credentials are always rejected.
     """
+    if not isinstance(allow_private_networks, bool):
+        raise URLReadError("allow_private_networks must be a boolean")
     if (
         isinstance(max_response_bytes, bool)
         or not isinstance(max_response_bytes, int)
         or max_response_bytes <= 0
     ):
         raise URLReadError("Maximum response size must be a positive integer")
-    _validate_url(url, allow_private_networks)
+    parsed = _validate_url(url)
+    validated_address = _resolve_validated_address(
+        parsed,
+        allow_private_networks,
+    )
 
     urllib3, http, timeout = _get_http_client()
     response = None
     try:
-        response = http.request(
-            "GET",
-            url,
-            fields=params,
-            timeout=timeout,
-            retries=False,
-            redirect=False,
-            preload_content=False,
-        )
+        if validated_address is None:
+            response = http.request(
+                "GET",
+                parsed.geturl(),
+                fields=params,
+                timeout=timeout,
+                retries=False,
+                redirect=False,
+                preload_content=False,
+            )
+        else:
+            response = _request_via_validated_address(
+                http,
+                parsed,
+                validated_address,
+                params,
+                timeout,
+            )
         if response.status != 200:
             raise URLReadError("URL is not returning correct response")
 
-        content_length = response.headers.get("Content-Length")
-        if content_length is not None:
-            try:
-                if int(content_length) > max_response_bytes:
-                    raise URLReadError("URL response exceeds maximum size")
-            except ValueError as error:
-                raise URLReadError("URL returned an invalid Content-Length") from error
-
-        response_data = response.read(max_response_bytes + 1, decode_content=True)
-        if len(response_data) > max_response_bytes:
-            raise URLReadError("URL response exceeds maximum size")
+        content_length = _validated_content_length(
+            response,
+            max_response_bytes,
+        )
+        response_data = _read_response_data(
+            response,
+            max_response_bytes,
+            content_length,
+        )
     except urllib3.exceptions.HTTPError as error:
         raise URLReadError("URL could not be read") from error
     finally:

lat.md/behavior.md

@@ -8,13 +8,13 @@ The input helpers convert files, strings, URLs, and stdin into Python data struc
 
 [[json2xml/utils.py#readfromjson]] wraps file and JSON decoding failures in `JSONReadError`. [[json2xml/utils.py#readfromstring]] rejects non-string inputs and malformed JSON with `StringReadError`.
 
-[[json2xml/utils.py#readfromurl]] lazily initializes the HTTP client, performs a bounded GET request, and raises `URLReadError` for network, status, size, decoding, and JSON failures.
+[[json2xml/utils.py#readfromurl]] lazily initializes the HTTP client, performs a bounded GET request, and raises `URLReadError` for hostname encoding, network, status, size, decoding, and JSON failures.
 
 ## URL security boundaries
 
 Remote JSON reads default to public, credential-free HTTP(S) targets and bounded decoded responses so callers do not accidentally expose internal services or unlimited memory.
 
-[[json2xml/utils.py#readfromurl]] disables redirects, rejects non-global resolved addresses, and reads at most 10 MiB after content decoding. Trusted library callers can explicitly opt into private-network access while retaining the response limit.
+[[json2xml/utils.py#readfromurl]] disables redirects, rejects non-global resolved addresses, and pins each public request to a validated address while retaining the original Host header and TLS hostname. It incrementally decodes gzip and deflate bodies with 10 MiB encoded and decoded limits, honors valid `Content-Length` values, and rejects unsupported encodings. Trusted library callers can opt into private-network access only with an actual boolean while retaining the response limits.
 
 ## User examples
 

lat.md/tests.md

@@ -34,6 +34,22 @@ URL input should reject unsupported schemes, embedded credentials, and private o
 
 URL input should stop reading once the decoded response exceeds its configured limit so compressed or oversized remote content cannot exhaust process memory.
 
+### URL reader bounds encoded response size
+
+Compressed URL input should honor valid `Content-Length` values and cap undeclared encoded bytes so network I/O remains bounded independently of decoded output size.
+
+### URL reader requires a boolean private-network opt-in
+
+Private-network access should require an actual boolean so truthy strings or numbers cannot silently disable destination validation.
+
+### URL reader wraps invalid Unicode hostnames
+
+Malformed IDNA hostnames should raise `URLReadError` so hostname encoding failures preserve the public URL-reader error contract.
+
+### URL reader pins validated DNS addresses
+
+Public URL reads should connect to a validated resolved address while preserving the original Host header and TLS hostname so DNS rebinding cannot redirect the connection.
+
 ## CLI failure messages
 
 These tests verify common command-line failures return short messages that name the broken input source and point users at the next valid action.