diff --git a/README.rst b/README.rst index f32125ce..f21730a3 100644 --- a/README.rst +++ b/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 diff --git a/RELEASE_NOTES.md b/RELEASE_NOTES.md index 8fee9d76..2df1c0ae 100644 --- a/RELEASE_NOTES.md +++ b/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. diff --git a/json2xml/utils.py b/json2xml/utils.py index 1284bf64..39b6fb2a 100644 --- a/json2xml/utils.py +++ b/json2xml/utils.py @@ -3,9 +3,10 @@ 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"] @@ -13,6 +14,7 @@ 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: diff --git a/lat.md/behavior.md b/lat.md/behavior.md index 9c3afba7..91e87479 100644 --- a/lat.md/behavior.md +++ b/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 diff --git a/lat.md/tests.md b/lat.md/tests.md index 5022bc69..3117d041 100644 --- a/lat.md/tests.md +++ b/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. diff --git a/tests/test_utils.py b/tests/test_utils.py index abbe36ad..b16fc565 100644 --- a/tests/test_utils.py +++ b/tests/test_utils.py @@ -1,10 +1,12 @@ """Test module for json2xml.utils functionality.""" +import gzip import json import socket import tempfile import threading +import zlib from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer -from typing import TYPE_CHECKING, ClassVar +from typing import TYPE_CHECKING, Any, ClassVar, cast from unittest.mock import Mock, patch import pytest @@ -220,6 +222,140 @@ def test_readfromurl_rejects_hostnames_resolving_to_private_networks( with pytest.raises(URLReadError, match="public network address"): readfromurl("https://internal.example/data.json") + # @lat: [[tests#Input readers#URL reader pins validated DNS addresses]] + @pytest.mark.parametrize( + ( + "url", + "validated_address", + "expected_port", + "expected_host", + "expected_pool_kwargs", + "uses_dns", + ), + [ + ( + "https://rebind.example/data.json?existing=yes", + "93.184.216.34", + 443, + "rebind.example", + { + "assert_hostname": "rebind.example", + "server_hostname": "rebind.example", + }, + True, + ), + ( + "http://rebind.example/data.json?existing=yes", + "93.184.216.34", + 80, + "rebind.example", + None, + True, + ), + ( + "https://[2606:4700:4700::1111]:8443/data.json?existing=yes", + "2606:4700:4700::1111", + 8443, + "[2606:4700:4700::1111]:8443", + { + "assert_hostname": "2606:4700:4700::1111", + "server_hostname": "2606:4700:4700::1111", + }, + False, + ), + ], + ids=["https-default-port", "http-default-port", "https-ipv6"], + ) + @patch("json2xml.utils._get_http_client") + @patch("json2xml.utils.socket.getaddrinfo") + def test_readfromurl_pins_validated_address_with_correct_authority( + self, + mock_getaddrinfo: Mock, + mock_get_http_client: Mock, + url: str, + validated_address: str, + expected_port: int, + expected_host: str, + expected_pool_kwargs: dict[str, str] | None, + uses_dns: bool, + ) -> None: + """Test pinned HTTP(S), default ports, and IPv6 authority handling.""" + mock_getaddrinfo.return_value = [ + ( + socket.AF_INET, + socket.SOCK_STREAM, + 6, + "", + (validated_address, expected_port), + ) + ] + response = Mock(status=200, headers={"Content-Length": "11"}) + response.read.return_value = b'{"ok":true}' + pool = Mock() + pool.request.return_value = response + http = Mock() + http.connection_from_host.return_value = pool + timeout = Mock() + mock_get_http_client.return_value = (urllib3, http, timeout) + + result = readfromurl(url, params={"added": "yes"}) + + assert result == {"ok": True} + http.request.assert_not_called() + http.connection_from_host.assert_called_once_with( + validated_address, + port=expected_port, + scheme="https" if url.startswith("https:") else "http", + pool_kwargs=expected_pool_kwargs, + ) + pool.request.assert_called_once_with( + "GET", + "/data.json?existing=yes", + fields={"added": "yes"}, + headers={"Host": expected_host}, + timeout=timeout, + retries=False, + redirect=False, + preload_content=False, + ) + if uses_dns: + mock_getaddrinfo.assert_called_once() + else: + mock_getaddrinfo.assert_not_called() + response.close.assert_called_once_with() + + @patch("json2xml.utils._get_http_client") + def test_readfromurl_uses_direct_request_for_private_network_opt_in( + self, mock_get_http_client: Mock + ) -> None: + """Test trusted private-network reads retain the complete request URL.""" + url = "https://private.example/data.json?existing=yes" + response = Mock(status=200, headers={"Content-Length": "11"}) + response.read.return_value = b'{"ok":true}' + http = Mock() + http.request.return_value = response + timeout = Mock() + mock_get_http_client.return_value = (urllib3, http, timeout) + + result = readfromurl( + url, + params={"added": "yes"}, + allow_private_networks=True, + ) + + assert result == {"ok": True} + http.connection_from_host.assert_not_called() + http.request.assert_called_once_with( + "GET", + url, + fields={"added": "yes"}, + timeout=timeout, + retries=False, + redirect=False, + preload_content=False, + ) + response.close.assert_called_once_with() + def test_readfromurl_rejects_unsupported_schemes_and_credentials(self) -> None: """Test URL reads accept only credential-free HTTP and HTTPS URLs.""" with pytest.raises(URLReadError, match="HTTP or HTTPS"): @@ -244,6 +380,46 @@ def test_readfromurl_rejects_unresolvable_hostnames( with pytest.raises(URLReadError, match="could not be resolved"): readfromurl("https://unresolvable.example/data.json") + # @lat: [[tests#Input readers#URL reader wraps invalid Unicode hostnames]] + @patch("json2xml.utils.socket.getaddrinfo") + def test_readfromurl_wraps_invalid_unicode_hostnames( + self, mock_getaddrinfo: Mock + ) -> None: + """Test malformed IDNA hostnames preserve the URL reader error contract.""" + mock_getaddrinfo.side_effect = UnicodeError("invalid IDNA label") + + with pytest.raises(URLReadError, match="could not be resolved"): + readfromurl("https://invalid-unicode.example/data.json") + + @patch("json2xml.utils._get_http_client") + @patch("json2xml.utils.socket.getaddrinfo") + def test_readfromurl_wraps_idna_failure_when_building_pinned_request( + self, mock_getaddrinfo: Mock, mock_get_http_client: Mock + ) -> None: + """Test IDNA failure after resolution still raises URLReadError.""" + mock_getaddrinfo.return_value = [ + (socket.AF_INET, socket.SOCK_STREAM, 6, "", ("93.184.216.34", 443)) + ] + http = Mock() + mock_get_http_client.return_value = (urllib3, http, Mock()) + + with pytest.raises(URLReadError, match="could not be resolved"): + readfromurl("https://\ud800.example/data.json") + + http.connection_from_host.assert_not_called() + + # @lat: [[tests#Input readers#URL reader requires a boolean private-network opt-in]] + @pytest.mark.parametrize("allow_private_networks", ["false", 1, None]) + def test_readfromurl_rejects_non_boolean_private_network_opt_in( + self, allow_private_networks: object + ) -> None: + """Test only an actual boolean can opt into private-network access.""" + with pytest.raises(URLReadError, match="must be a boolean"): + readfromurl( + "http://127.0.0.1/private.json", + allow_private_networks=cast(Any, allow_private_networks), + ) + def test_readfromurl_rejects_invalid_response_limit(self) -> None: """Test callers cannot disable the response cap with a non-positive value.""" with pytest.raises(URLReadError, match="positive integer"): @@ -251,7 +427,11 @@ def test_readfromurl_rejects_invalid_response_limit(self) -> None: @pytest.mark.parametrize( ("content_length", "message"), - [("17", "maximum size"), ("not-a-number", "invalid Content-Length")], + [ + ("17", "maximum size"), + ("not-a-number", "invalid Content-Length"), + ("-1", "invalid Content-Length"), + ], ) @patch("json2xml.utils._get_http_client") def test_readfromurl_rejects_invalid_content_lengths( @@ -267,25 +447,237 @@ def test_readfromurl_rejects_invalid_content_lengths( mock_get_http_client.return_value = (urllib3, http, Mock()) with pytest.raises(URLReadError, match=message): - readfromurl("https://8.8.8.8/data.json", max_response_bytes=16) + readfromurl( + "https://8.8.8.8/data.json", + max_response_bytes=16, + allow_private_networks=True, + ) response.read.assert_not_called() response.close.assert_called_once_with() + @patch("json2xml.utils._get_http_client") + def test_readfromurl_limits_uncompressed_response_without_length( + self, mock_get_http_client: Mock + ) -> None: + """Test an undeclared uncompressed body cannot exceed the byte limit.""" + response = Mock(status=200, headers={}) + response.read.return_value = b"x" * 17 + http = Mock() + http.request.return_value = response + mock_get_http_client.return_value = (urllib3, http, Mock()) + + with pytest.raises(URLReadError, match="maximum size"): + readfromurl( + "https://8.8.8.8/data.json", + max_response_bytes=16, + allow_private_networks=True, + ) + + response.read.assert_called_once_with(17, decode_content=False) + + @pytest.mark.parametrize( + ("encoding", "compressed"), + [ + pytest.param( + "gzip", + gzip.compress(b'{"ok":true}', mtime=0), + id="gzip", + ), + pytest.param( + "deflate", + zlib.compress(b'{"ok":true}'), + id="zlib-deflate", + ), + pytest.param( + "deflate", + (lambda compressor: compressor.compress(b'{"ok":true}') + compressor.flush())( + zlib.compressobj(wbits=-zlib.MAX_WBITS) + ), + id="raw-deflate", + ), + ], + ) + @patch("json2xml.utils._get_http_client") + def test_readfromurl_decodes_supported_compression_incrementally( + self, + mock_get_http_client: Mock, + encoding: str, + compressed: bytes, + ) -> None: + """Test bounded decoding preserves supported compressed responses.""" + response = Mock(status=200, headers={"Content-Encoding": encoding}) + response.read.side_effect = [compressed, b""] + http = Mock() + http.request.return_value = response + mock_get_http_client.return_value = (urllib3, http, Mock()) + + result = readfromurl( + "https://8.8.8.8/data.json", + allow_private_networks=True, + ) + + assert result == {"ok": True} + + @patch("json2xml.utils._get_http_client") + def test_readfromurl_respects_compressed_content_length( + self, mock_get_http_client: Mock + ) -> None: + """Test a declared compressed size bounds raw response reads.""" + compressed = gzip.compress(b'{"ok":true}', mtime=0) + response = Mock( + status=200, + headers={ + "Content-Encoding": "gzip", + "Content-Length": str(len(compressed)), + }, + ) + response.read.side_effect = [compressed] + http = Mock() + http.request.return_value = response + mock_get_http_client.return_value = (urllib3, http, Mock()) + + result = readfromurl( + "https://8.8.8.8/data.json", + allow_private_networks=True, + ) + + assert result == {"ok": True} + response.read.assert_called_once_with( + len(compressed), + decode_content=False, + ) + + @patch("json2xml.utils._get_http_client") + def test_readfromurl_rejects_compressed_body_over_declared_length( + self, mock_get_http_client: Mock + ) -> None: + """Test compressed reads reject bytes beyond the declared length.""" + compressed = gzip.compress(b'{"ok":true}', mtime=0) + response = Mock( + status=200, + headers={"Content-Encoding": "gzip", "Content-Length": "1"}, + ) + response.read.return_value = compressed + http = Mock() + http.request.return_value = response + mock_get_http_client.return_value = (urllib3, http, Mock()) + + with pytest.raises(URLReadError, match="declared Content-Length"): + readfromurl( + "https://8.8.8.8/data.json", + allow_private_networks=True, + ) + + @patch("json2xml.utils._get_http_client") + def test_readfromurl_rejects_incomplete_compressed_body( + self, mock_get_http_client: Mock + ) -> None: + """Test compressed reads reject EOF before the declared length.""" + compressed = gzip.compress(b'{"ok":true}', mtime=0) + response = Mock( + status=200, + headers={ + "Content-Encoding": "gzip", + "Content-Length": str(len(compressed)), + }, + ) + response.read.side_effect = [compressed[:10], b""] + http = Mock() + http.request.return_value = response + mock_get_http_client.return_value = (urllib3, http, Mock()) + + with pytest.raises(URLReadError, match="did not match Content-Length"): + readfromurl( + "https://8.8.8.8/data.json", + allow_private_networks=True, + ) + + # @lat: [[tests#Input readers#URL reader bounds encoded response size]] + @patch("json2xml.utils._get_http_client") + def test_readfromurl_caps_compressed_bytes_without_content_length( + self, mock_get_http_client: Mock + ) -> None: + """Test compressed input is bounded even when decoded output is small.""" + compressed = gzip.compress(b'{"ok":true}', mtime=0) + response = Mock(status=200, headers={"Content-Encoding": "gzip"}) + response.read.side_effect = [compressed] + http = Mock() + http.request.return_value = response + mock_get_http_client.return_value = (urllib3, http, Mock()) + + with pytest.raises(URLReadError, match="maximum size"): + readfromurl( + "https://8.8.8.8/data.json", + max_response_bytes=16, + allow_private_networks=True, + ) + + response.read.assert_called_once_with(17, decode_content=False) + + @pytest.mark.parametrize( + ("encoding", "compressed"), + [ + pytest.param("br", b"unsupported", id="unsupported"), + pytest.param("gzip", b"not a gzip stream", id="invalid-gzip"), + pytest.param("deflate", b"x", id="invalid-short-deflate"), + pytest.param( + "gzip", + gzip.compress(b'{"ok":true}', mtime=0)[:-8], + id="truncated-gzip", + ), + pytest.param( + "gzip", + gzip.compress(b'{"ok":true}', mtime=0) + b"trailing data", + id="trailing-gzip", + ), + ], + ) + @patch("json2xml.utils._get_http_client") + def test_readfromurl_rejects_unsafe_or_invalid_compression( + self, + mock_get_http_client: Mock, + encoding: str, + compressed: bytes, + ) -> None: + """Test unsafe or malformed compressed responses fail closed.""" + response = Mock(status=200, headers={"Content-Encoding": encoding}) + response.read.side_effect = [compressed, b""] + http = Mock() + http.request.return_value = response + mock_get_http_client.return_value = (urllib3, http, Mock()) + + with pytest.raises(URLReadError, match="compressed|Content-Encoding"): + readfromurl( + "https://8.8.8.8/data.json", + allow_private_networks=True, + ) + @patch("json2xml.utils._get_http_client") # @lat: [[tests#Input readers#URL reader limits decoded response size]] def test_readfromurl_limits_decoded_response_size( self, mock_get_http_client: Mock ) -> None: - """Test URL reads stop after the configured decoded-byte limit.""" - response = Mock(status=200, headers={}) - response.read.return_value = b'{"value":"payload larger than limit"}' + """Test compressed URL reads stop at the configured decoded-byte limit.""" + compressed = gzip.compress( + b'{"value":"' + (b"x" * 10_000) + b'"}', + mtime=0, + ) + response = Mock( + status=200, + headers={"Content-Encoding": "gzip"}, + ) + response.read.side_effect = [compressed, b""] http = Mock() http.request.return_value = response mock_get_http_client.return_value = (urllib3, http, Mock()) with pytest.raises(URLReadError, match="maximum size"): - readfromurl("https://8.8.8.8/data.json", max_response_bytes=16) + readfromurl( + "https://8.8.8.8/data.json", + max_response_bytes=128, + allow_private_networks=True, + ) http.request.assert_called_once_with( "GET", @@ -296,6 +688,7 @@ def test_readfromurl_limits_decoded_response_size( redirect=False, preload_content=False, ) + response.read.assert_called_once_with(129, decode_content=False) response.close.assert_called_once_with()