From 758a16aee0769d2981ab15a0fb0d38fb069596b6 Mon Sep 17 00:00:00 2001 From: Vinit Kumar Date: Tue, 4 Aug 2026 19:25:40 +0530 Subject: [PATCH 1/7] fix: require boolean private network opt-in --- json2xml/utils.py | 2 ++ lat.md/behavior.md | 2 +- lat.md/tests.md | 4 ++++ tests/test_utils.py | 12 ++++++++++++ 4 files changed, 19 insertions(+), 1 deletion(-) diff --git a/json2xml/utils.py b/json2xml/utils.py index 1284bf64..fb4c2e9b 100644 --- a/json2xml/utils.py +++ b/json2xml/utils.py @@ -109,6 +109,8 @@ 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) diff --git a/lat.md/behavior.md b/lat.md/behavior.md index 9c3afba7..5ba4cdb8 100644 --- a/lat.md/behavior.md +++ b/lat.md/behavior.md @@ -14,7 +14,7 @@ The input helpers convert files, strings, URLs, and stdin into Python data struc 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 reads at most 10 MiB after content decoding. Trusted library callers can opt into private-network access only with an actual boolean while retaining the response limit. ## User examples diff --git a/lat.md/tests.md b/lat.md/tests.md index 5022bc69..de8bfdf8 100644 --- a/lat.md/tests.md +++ b/lat.md/tests.md @@ -34,6 +34,10 @@ 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 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. + ## 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..31c2dbf0 100644 --- a/tests/test_utils.py +++ b/tests/test_utils.py @@ -244,6 +244,18 @@ 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 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=allow_private_networks, # type: ignore[arg-type] + ) + 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"): From a2cd2f76d00b49b416e01e7696ccd572a3238fc6 Mon Sep 17 00:00:00 2001 From: Vinit Kumar Date: Tue, 4 Aug 2026 19:26:13 +0530 Subject: [PATCH 2/7] fix: wrap invalid Unicode hostnames --- json2xml/utils.py | 2 +- lat.md/behavior.md | 2 +- lat.md/tests.md | 4 ++++ tests/test_utils.py | 11 +++++++++++ 4 files changed, 17 insertions(+), 2 deletions(-) diff --git a/json2xml/utils.py b/json2xml/utils.py index fb4c2e9b..1520861b 100644 --- a/json2xml/utils.py +++ b/json2xml/utils.py @@ -86,7 +86,7 @@ def _validate_url(url: str, allow_private_networks: bool) -> None: port or (443 if parsed.scheme == "https" else 80), type=socket.SOCK_STREAM, ) - except OSError as error: + except (OSError, UnicodeError) as error: raise URLReadError("URL hostname could not be resolved") from error addresses = { ip_address(str(info[4][0]).split("%", 1)[0]) diff --git a/lat.md/behavior.md b/lat.md/behavior.md index 5ba4cdb8..9cf52ae4 100644 --- a/lat.md/behavior.md +++ b/lat.md/behavior.md @@ -8,7 +8,7 @@ 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 diff --git a/lat.md/tests.md b/lat.md/tests.md index de8bfdf8..6f4a6505 100644 --- a/lat.md/tests.md +++ b/lat.md/tests.md @@ -38,6 +38,10 @@ URL input should stop reading once the decoded response exceeds its configured l 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. + ## 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 31c2dbf0..2d71723c 100644 --- a/tests/test_utils.py +++ b/tests/test_utils.py @@ -244,6 +244,17 @@ 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") + # @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( From 997844a34bd4b870dcc45449469b0065045ee623 Mon Sep 17 00:00:00 2001 From: Vinit Kumar Date: Tue, 4 Aug 2026 19:28:35 +0530 Subject: [PATCH 3/7] fix: pin URL requests to validated DNS addresses --- json2xml/utils.py | 82 ++++++++++++++++++++++++++++++++++++--------- lat.md/behavior.md | 2 +- lat.md/tests.md | 4 +++ tests/test_utils.py | 62 ++++++++++++++++++++++++++++++++-- 4 files changed, 131 insertions(+), 19 deletions(-) diff --git a/json2xml/utils.py b/json2xml/utils.py index 1520861b..4128f925 100644 --- a/json2xml/utils.py +++ b/json2xml/utils.py @@ -5,7 +5,7 @@ import socket from ipaddress import ip_address from typing import Any -from urllib.parse import urlsplit +from urllib.parse import SplitResult, urlsplit, urlunsplit __lazy_modules__ = ["urllib3"] @@ -59,8 +59,10 @@ 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, allow_private_networks: bool +) -> tuple[SplitResult, str | None]: + """Validate a URL and return the public address the request must use.""" try: parsed = urlsplit(url) port = parsed.port @@ -74,11 +76,11 @@ def _validate_url(url: str, allow_private_networks: bool) -> None: if parsed.hostname is None: raise URLReadError("URL must include a hostname") if allow_private_networks: - return + return parsed, None hostname = parsed.hostname try: - addresses = {ip_address(hostname)} + addresses = [ip_address(hostname)] except ValueError: try: address_info = socket.getaddrinfo( @@ -88,13 +90,63 @@ def _validate_url(url: str, allow_private_networks: bool) -> None: ) 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 parsed, str(addresses[0]) + + +def _request_url( + http: Any, + parsed: SplitResult, + validated_address: str | None, + params: dict[str, str] | None, + timeout: Any, +) -> Any: + """Issue a GET directly to the validated address when one is required.""" + request_options = { + "fields": params, + "timeout": timeout, + "retries": False, + "redirect": False, + "preload_content": False, + } + if validated_address is None: + return http.request("GET", parsed.geturl(), **request_options) + + 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, + headers={"Host": authority}, + **request_options, + ) def readfromurl( @@ -117,19 +169,17 @@ def readfromurl( or max_response_bytes <= 0 ): raise URLReadError("Maximum response size must be a positive integer") - _validate_url(url, allow_private_networks) + parsed, validated_address = _validate_url(url, 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, + response = _request_url( + http, + parsed, + validated_address, + params, + timeout, ) if response.status != 200: raise URLReadError("URL is not returning correct response") diff --git a/lat.md/behavior.md b/lat.md/behavior.md index 9cf52ae4..8c72a8e3 100644 --- a/lat.md/behavior.md +++ b/lat.md/behavior.md @@ -14,7 +14,7 @@ The input helpers convert files, strings, URLs, and stdin into Python data struc 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 opt into private-network access only with an actual boolean 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 reads at most 10 MiB after content decoding. Trusted library callers can opt into private-network access only with an actual boolean while retaining the response limit. ## User examples diff --git a/lat.md/tests.md b/lat.md/tests.md index 6f4a6505..7857a4b7 100644 --- a/lat.md/tests.md +++ b/lat.md/tests.md @@ -42,6 +42,10 @@ Private-network access should require an actual boolean so truthy strings or num 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 2d71723c..7e89037c 100644 --- a/tests/test_utils.py +++ b/tests/test_utils.py @@ -220,6 +220,56 @@ 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]] + @patch("json2xml.utils._get_http_client") + @patch("json2xml.utils.socket.getaddrinfo") + def test_readfromurl_pins_validated_dns_address( + self, mock_getaddrinfo: Mock, mock_get_http_client: Mock + ) -> None: + """Test the HTTP connection cannot resolve a validated hostname again.""" + mock_getaddrinfo.return_value = [ + (socket.AF_INET, socket.SOCK_STREAM, 6, "", ("93.184.216.34", 8443)) + ] + 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( + "https://rebind.example:8443/data.json?existing=yes", + params={"added": "yes"}, + ) + + assert result == {"ok": True} + http.request.assert_not_called() + http.connection_from_host.assert_called_once_with( + "93.184.216.34", + port=8443, + scheme="https", + pool_kwargs={ + "assert_hostname": "rebind.example", + "server_hostname": "rebind.example", + }, + ) + pool.request.assert_called_once_with( + "GET", + "/data.json?existing=yes", + fields={"added": "yes"}, + headers={"Host": "rebind.example:8443"}, + 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"): @@ -290,7 +340,11 @@ 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() @@ -308,7 +362,11 @@ def test_readfromurl_limits_decoded_response_size( 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=16, + allow_private_networks=True, + ) http.request.assert_called_once_with( "GET", From a7e13168d62072715f56c80c3c4770aeb2dd5698 Mon Sep 17 00:00:00 2001 From: Vinit Kumar Date: Tue, 4 Aug 2026 19:31:32 +0530 Subject: [PATCH 4/7] fix: bound incremental response decompression --- json2xml/utils.py | 71 ++++++++++++++++++++++++++++++++++++++++-- lat.md/behavior.md | 2 +- tests/test_utils.py | 76 +++++++++++++++++++++++++++++++++++++++++++-- 3 files changed, 142 insertions(+), 7 deletions(-) diff --git a/json2xml/utils.py b/json2xml/utils.py index 4128f925..18f00eca 100644 --- a/json2xml/utils.py +++ b/json2xml/utils.py @@ -3,6 +3,7 @@ import json import socket +import zlib from ipaddress import ip_address from typing import Any from urllib.parse import SplitResult, urlsplit, urlunsplit @@ -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 @@ -149,6 +151,71 @@ def _request_url( ) +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": + has_zlib_header = ( + len(first_chunk) >= 2 + and first_chunk[0] & 0x0F == 8 + and (first_chunk[0] << 8 | first_chunk[1]) % 31 == 0 + ) + return zlib.decompressobj( + zlib.MAX_WBITS if has_zlib_header else -zlib.MAX_WBITS + ) + raise URLReadError(f"Unsupported Content-Encoding: {encoding}") + + +def _read_response_data(response: Any, max_response_bytes: int) -> bytes: + """Read a response without allowing 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 + + first_chunk = response.read( + COMPRESSED_READ_CHUNK_BYTES, + decode_content=False, + ) + decoder = _compression_decoder(encoding, first_chunk) + response_data = bytearray() + compressed_chunk = first_chunk + try: + while compressed_chunk: + 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") + unconsumed = decoder.unconsumed_tail + if unconsumed == pending: + raise URLReadError("URL returned invalid compressed data") + pending = unconsumed + compressed_chunk = response.read( + COMPRESSED_READ_CHUNK_BYTES, + decode_content=False, + ) + + remaining_bytes = max_response_bytes + 1 - len(response_data) + response_data.extend(decoder.flush(remaining_bytes)) + except zlib.error as error: + raise URLReadError("URL returned invalid compressed data") from error + + if len(response_data) > max_response_bytes: + raise URLReadError("URL response exceeds maximum size") + if not decoder.eof or decoder.unused_data: + raise URLReadError("URL returned invalid compressed data") + return bytes(response_data) + + def readfromurl( url: str, params: dict[str, str] | None = None, @@ -192,9 +259,7 @@ def readfromurl( 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") + response_data = _read_response_data(response, max_response_bytes) 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 8c72a8e3..ff981286 100644 --- a/lat.md/behavior.md +++ b/lat.md/behavior.md @@ -14,7 +14,7 @@ The input helpers convert files, strings, URLs, and stdin into Python data struc 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 pins each public request to a validated address while retaining the original Host header and TLS hostname. It reads at most 10 MiB after content decoding. Trusted library callers can opt into private-network access only with an actual boolean 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 into a 10 MiB bounded output and rejects unsupported encodings. Trusted library callers can opt into private-network access only with an actual boolean while retaining the response limit. ## User examples diff --git a/tests/test_utils.py b/tests/test_utils.py index 7e89037c..5e8bbf68 100644 --- a/tests/test_utils.py +++ b/tests/test_utils.py @@ -1,8 +1,10 @@ """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 unittest.mock import Mock, patch @@ -349,14 +351,81 @@ def test_readfromurl_rejects_invalid_content_lengths( response.read.assert_not_called() response.close.assert_called_once_with() + @pytest.mark.parametrize( + ("encoding", "compressed"), + [ + ("gzip", gzip.compress(b'{"ok":true}')), + ("deflate", zlib.compress(b'{"ok":true}')), + ( + "deflate", + (lambda compressor: compressor.compress(b'{"ok":true}') + compressor.flush())( + zlib.compressobj(wbits=-zlib.MAX_WBITS) + ), + ), + ], + ) + @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} + + @pytest.mark.parametrize( + ("encoding", "compressed"), + [ + ("br", b"unsupported"), + ("gzip", b"not a gzip stream"), + ("gzip", gzip.compress(b'{"ok":true}')[:-8]), + ("gzip", gzip.compress(b'{"ok":true}') + b"trailing data"), + ], + ) + @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" * 1_000_000) + b'"}') + 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()) @@ -377,6 +446,7 @@ def test_readfromurl_limits_decoded_response_size( redirect=False, preload_content=False, ) + response.read.assert_called_once_with(64 * 1024, decode_content=False) response.close.assert_called_once_with() From 828cf9a2abcb5bdccba7f18dd47c25d3b68f5c56 Mon Sep 17 00:00:00 2001 From: Vinit Kumar Date: Tue, 4 Aug 2026 19:32:41 +0530 Subject: [PATCH 5/7] docs: add security migration guidance --- README.rst | 6 ++++-- RELEASE_NOTES.md | 21 +++++++++++++++++++++ 2 files changed, 25 insertions(+), 2 deletions(-) diff --git a/README.rst b/README.rst index f32125ce..fe2ca034 100644 --- a/README.rst +++ b/README.rst @@ -240,8 +240,10 @@ 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 decoded content. Gzip and deflate responses 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..921f4a44 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 are decoded incrementally with bounded zlib output, so compressed bodies cannot allocate beyond `max_response_bytes` 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. +- Decoded URL responses remain 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. From 20a090f3a6e735a69979a05b63d4f4dd32613b46 Mon Sep 17 00:00:00 2001 From: Vinit Kumar Date: Tue, 4 Aug 2026 19:34:28 +0530 Subject: [PATCH 6/7] test: complete URL boundary coverage --- json2xml/utils.py | 15 ++------------- tests/test_utils.py | 24 ++++++++++++++++++++++-- 2 files changed, 24 insertions(+), 15 deletions(-) diff --git a/json2xml/utils.py b/json2xml/utils.py index 18f00eca..19a2a1c3 100644 --- a/json2xml/utils.py +++ b/json2xml/utils.py @@ -121,10 +121,7 @@ def _request_url( return http.request("GET", parsed.geturl(), **request_options) 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 + hostname = parsed.hostname.encode("idna").decode("ascii") port = parsed.port or (443 if parsed.scheme == "https" else 80) authority = f"[{hostname}]" if ":" in hostname else hostname @@ -195,22 +192,14 @@ def _read_response_data(response: Any, max_response_bytes: int) -> bytes: response_data.extend(decoded_chunk) if len(response_data) > max_response_bytes: raise URLReadError("URL response exceeds maximum size") - unconsumed = decoder.unconsumed_tail - if unconsumed == pending: - raise URLReadError("URL returned invalid compressed data") - pending = unconsumed + pending = decoder.unconsumed_tail compressed_chunk = response.read( COMPRESSED_READ_CHUNK_BYTES, decode_content=False, ) - - remaining_bytes = max_response_bytes + 1 - len(response_data) - response_data.extend(decoder.flush(remaining_bytes)) except zlib.error as error: raise URLReadError("URL returned invalid compressed data") from error - if len(response_data) > max_response_bytes: - raise URLReadError("URL response exceeds maximum size") if not decoder.eof or decoder.unused_data: raise URLReadError("URL returned invalid compressed data") return bytes(response_data) diff --git a/tests/test_utils.py b/tests/test_utils.py index 5e8bbf68..15f1a2a9 100644 --- a/tests/test_utils.py +++ b/tests/test_utils.py @@ -6,7 +6,7 @@ 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 @@ -316,7 +316,7 @@ def test_readfromurl_rejects_non_boolean_private_network_opt_in( with pytest.raises(URLReadError, match="must be a boolean"): readfromurl( "http://127.0.0.1/private.json", - allow_private_networks=allow_private_networks, # type: ignore[arg-type] + allow_private_networks=cast(Any, allow_private_networks), ) def test_readfromurl_rejects_invalid_response_limit(self) -> None: @@ -351,6 +351,26 @@ def test_readfromurl_rejects_invalid_content_lengths( 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"), [ From f5784ad86a6f8287f68da1ab7b936c3fa263bd3e Mon Sep 17 00:00:00 2001 From: Vinit Kumar Date: Tue, 4 Aug 2026 19:54:06 +0530 Subject: [PATCH 7/7] fix: tests --- README.rst | 7 +- RELEASE_NOTES.md | 4 +- json2xml/utils.py | 216 +++++++++++++++++++++++---------- lat.md/behavior.md | 2 +- lat.md/tests.md | 4 + tests/test_utils.py | 284 +++++++++++++++++++++++++++++++++++++++----- 6 files changed, 414 insertions(+), 103 deletions(-) diff --git a/README.rst b/README.rst index fe2ca034..f21730a3 100644 --- a/README.rst +++ b/README.rst @@ -241,9 +241,10 @@ You can use the json2xml library in the following ways: URL reads accept only credential-free HTTP(S), reject redirects and non-public destinations by default, pin public connections to their validated DNS address, -and stop after 10 MiB of decoded content. Gzip and deflate responses 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: +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 921f4a44..2df1c0ae 100644 --- a/RELEASE_NOTES.md +++ b/RELEASE_NOTES.md @@ -3,14 +3,14 @@ ## 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 are decoded incrementally with bounded zlib output, so compressed bodies cannot allocate beyond `max_response_bytes` before rejection. +- 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. -- Decoded URL responses remain limited to 10 MiB by default. Set `max_response_bytes` explicitly when a trusted endpoint needs a different positive limit. +- 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`. diff --git a/json2xml/utils.py b/json2xml/utils.py index 19a2a1c3..39b6fb2a 100644 --- a/json2xml/utils.py +++ b/json2xml/utils.py @@ -61,13 +61,11 @@ def readfromjson(filename: str) -> JSONValue: # @lat: [[behavior#URL security boundaries]] -def _validate_url( - url: str, allow_private_networks: bool -) -> tuple[SplitResult, str | None]: - """Validate a URL and return the public address the request must use.""" +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 @@ -77,17 +75,26 @@ def _validate_url( 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 parsed, None + 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)] except ValueError: try: address_info = socket.getaddrinfo( hostname, - port or (443 if parsed.scheme == "https" else 80), + port, type=socket.SOCK_STREAM, ) except (OSError, UnicodeError) as error: @@ -99,29 +106,22 @@ def _validate_url( 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 parsed, str(addresses[0]) + return str(addresses[0]) -def _request_url( +def _request_via_validated_address( http: Any, parsed: SplitResult, - validated_address: str | None, + validated_address: str, params: dict[str, str] | None, timeout: Any, ) -> Any: - """Issue a GET directly to the validated address when one is required.""" - request_options = { - "fields": params, - "timeout": timeout, - "retries": False, - "redirect": False, - "preload_content": False, - } - if validated_address is None: - return http.request("GET", parsed.geturl(), **request_options) - + """Issue a GET directly to an address already validated as public.""" assert parsed.hostname is not None - hostname = parsed.hostname.encode("idna").decode("ascii") + 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 @@ -143,8 +143,23 @@ def _request_url( return pool.request( "GET", request_target, + fields=params, headers={"Host": authority}, - **request_options, + 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 ) @@ -153,38 +168,31 @@ def _compression_decoder(encoding: str, first_chunk: bytes) -> Any: if encoding in {"gzip", "x-gzip"}: return zlib.decompressobj(16 + zlib.MAX_WBITS) if encoding == "deflate": - has_zlib_header = ( - len(first_chunk) >= 2 - and first_chunk[0] & 0x0F == 8 - and (first_chunk[0] << 8 | first_chunk[1]) % 31 == 0 - ) - return zlib.decompressobj( - zlib.MAX_WBITS if has_zlib_header else -zlib.MAX_WBITS + 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 _read_response_data(response: Any, max_response_bytes: int) -> bytes: - """Read a response without allowing 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 - - first_chunk = response.read( - COMPRESSED_READ_CHUNK_BYTES, - decode_content=False, - ) - decoder = _compression_decoder(encoding, first_chunk) +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) @@ -193,18 +201,79 @@ def _read_response_data(response: Any, max_response_bytes: int) -> bytes: if len(response_data) > max_response_bytes: raise URLReadError("URL response exceeds maximum size") pending = decoder.unconsumed_tail - compressed_chunk = response.read( + + 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, - decode_content=False, + 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( url: str, params: dict[str, str] | None = None, @@ -225,30 +294,45 @@ def readfromurl( or max_response_bytes <= 0 ): raise URLReadError("Maximum response size must be a positive integer") - parsed, validated_address = _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 = _request_url( - http, - parsed, - validated_address, - params, - timeout, - ) + 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 = _read_response_data(response, max_response_bytes) + 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 ff981286..91e87479 100644 --- a/lat.md/behavior.md +++ b/lat.md/behavior.md @@ -14,7 +14,7 @@ The input helpers convert files, strings, URLs, and stdin into Python data struc 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 pins each public request to a validated address while retaining the original Host header and TLS hostname. It incrementally decodes gzip and deflate bodies into a 10 MiB bounded output and rejects unsupported encodings. Trusted library callers can opt into private-network access only with an actual boolean 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 7857a4b7..3117d041 100644 --- a/lat.md/tests.md +++ b/lat.md/tests.md @@ -34,6 +34,10 @@ 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. diff --git a/tests/test_utils.py b/tests/test_utils.py index 15f1a2a9..b16fc565 100644 --- a/tests/test_utils.py +++ b/tests/test_utils.py @@ -223,19 +223,73 @@ def test_readfromurl_rejects_hostnames_resolving_to_private_networks( 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_dns_address( - self, mock_getaddrinfo: Mock, mock_get_http_client: Mock + 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 the HTTP connection cannot resolve a validated hostname again.""" + """Test pinned HTTP(S), default ports, and IPv6 authority handling.""" mock_getaddrinfo.return_value = [ - (socket.AF_INET, socket.SOCK_STREAM, 6, "", ("93.184.216.34", 8443)) + ( + socket.AF_INET, + socket.SOCK_STREAM, + 6, + "", + (validated_address, expected_port), + ) ] - response = Mock( - status=200, - headers={"Content-Length": "11"}, - ) + response = Mock(status=200, headers={"Content-Length": "11"}) response.read.return_value = b'{"ok":true}' pool = Mock() pool.request.return_value = response @@ -244,27 +298,57 @@ def test_readfromurl_pins_validated_dns_address( timeout = Mock() mock_get_http_client.return_value = (urllib3, http, timeout) - result = readfromurl( - "https://rebind.example:8443/data.json?existing=yes", - params={"added": "yes"}, - ) + result = readfromurl(url, params={"added": "yes"}) assert result == {"ok": True} http.request.assert_not_called() http.connection_from_host.assert_called_once_with( - "93.184.216.34", - port=8443, - scheme="https", - pool_kwargs={ - "assert_hostname": "rebind.example", - "server_hostname": "rebind.example", - }, + 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": "rebind.example:8443"}, + 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, @@ -307,6 +391,23 @@ def test_readfromurl_wraps_invalid_unicode_hostnames( 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( @@ -326,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( @@ -374,13 +479,22 @@ def test_readfromurl_limits_uncompressed_response_without_length( @pytest.mark.parametrize( ("encoding", "compressed"), [ - ("gzip", gzip.compress(b'{"ok":true}')), - ("deflate", zlib.compress(b'{"ok":true}')), - ( + 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", ), ], ) @@ -405,13 +519,118 @@ def test_readfromurl_decodes_supported_compression_incrementally( 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"), [ - ("br", b"unsupported"), - ("gzip", b"not a gzip stream"), - ("gzip", gzip.compress(b'{"ok":true}')[:-8]), - ("gzip", gzip.compress(b'{"ok":true}') + b"trailing data"), + 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") @@ -440,7 +659,10 @@ def test_readfromurl_limits_decoded_response_size( self, mock_get_http_client: Mock ) -> None: """Test compressed URL reads stop at the configured decoded-byte limit.""" - compressed = gzip.compress(b'{"value":"' + (b"x" * 1_000_000) + b'"}') + compressed = gzip.compress( + b'{"value":"' + (b"x" * 10_000) + b'"}', + mtime=0, + ) response = Mock( status=200, headers={"Content-Encoding": "gzip"}, @@ -453,7 +675,7 @@ def test_readfromurl_limits_decoded_response_size( with pytest.raises(URLReadError, match="maximum size"): readfromurl( "https://8.8.8.8/data.json", - max_response_bytes=16, + max_response_bytes=128, allow_private_networks=True, ) @@ -466,7 +688,7 @@ def test_readfromurl_limits_decoded_response_size( redirect=False, preload_content=False, ) - response.read.assert_called_once_with(64 * 1024, decode_content=False) + response.read.assert_called_once_with(129, decode_content=False) response.close.assert_called_once_with()