Harden URL reads against rebinding and compression bombs - #372
Merged
Conversation
Contributor
Reviewer's GuideHardens json2xml URL reading by pinning public HTTP(S) connections to DNS-validated addresses, adding bounded incremental decompression for gzip/deflate, tightening configuration validation, and updating tests and docs to cover the new behavior and migration guidance. Sequence diagram for hardened readfromurl URL readingsequenceDiagram
actor Caller
participant readfromurl
participant _validate_url
participant _request_url
participant _read_response_data
Caller->>readfromurl: readfromurl(url, allow_private_networks, max_response_bytes, params)
readfromurl->>_validate_url: _validate_url(url, allow_private_networks)
_validate_url-->>readfromurl: parsed, validated_address
readfromurl->>_request_url: _request_url(http, parsed, validated_address, params, timeout)
_request_url->>http: connection_from_host(validated_address, port, scheme, pool_kwargs)
_request_url->>pool: request("GET", request_target, headers, request_options)
_request_url-->>readfromurl: response
readfromurl->>_read_response_data: _read_response_data(response, max_response_bytes)
alt identity_or_no_encoding
_read_response_data->>response: read(max_response_bytes + 1, decode_content=False)
_read_response_data-->>readfromurl: response_data
else gzip_or_deflate
_read_response_data->>response: read(COMPRESSED_READ_CHUNK_BYTES, decode_content=False)
_read_response_data->>_compression_decoder: _compression_decoder(encoding, first_chunk)
loop while compressed_chunk
_read_response_data->>response: read(COMPRESSED_READ_CHUNK_BYTES, decode_content=False)
end
_read_response_data-->>readfromurl: response_data
end
readfromurl-->>Caller: parsed JSON data
File-Level Changes
Tips and commandsInteracting with Sourcery
Customizing Your ExperienceAccess your dashboard to:
Getting Help
|
Codecov Report✅ All modified and coverable lines are covered by tests. Additional details and impacted files@@ Coverage Diff @@
## master #372 +/- ##
=========================================
Coverage 100.00% 100.00%
=========================================
Files 7 7
Lines 833 920 +87
=========================================
+ Hits 833 920 +87
Flags with carried forward coverage won't be shown. Click here to find out more. ☔ View full report in Codecov by Harness. |
![sourcery-ai[bot]](https://avatars.githubusercontent.com/in/48477?s=60&v=4){kind=small}
Contributor
There was a problem hiding this comment.
Hey - I've found 2 issues, and left some high level feedback:
- In
_request_url, the IDNA encoding ofparsed.hostnamecan raiseUnicodeErrorthat is not wrapped inURLReadError, which breaks the documented error contract forreadfromurl; consider catchingUnicodeErrorthere and re-raisingURLReadErrorsimilar to_validate_url. - In
_read_response_data, the compressed path ignores a validContent-Lengthheader and will continue reading until EOF; if you want to bound I/O as well as decoded size, you might want to respectContent-Lengthor cap the total compressed bytes read.
Prompt for AI Agents
Please address the comments from this code review:
## Overall Comments
- In `_request_url`, the IDNA encoding of `parsed.hostname` can raise `UnicodeError` that is not wrapped in `URLReadError`, which breaks the documented error contract for `readfromurl`; consider catching `UnicodeError` there and re-raising `URLReadError` similar to `_validate_url`.
- In `_read_response_data`, the compressed path ignores a valid `Content-Length` header and will continue reading until EOF; if you want to bound I/O as well as decoded size, you might want to respect `Content-Length` or cap the total compressed bytes read.
## Individual Comments
### Comment 1
<location path="tests/test_utils.py" line_range="254-263" />
<code_context>
+
+ 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:
</code_context>
<issue_to_address>
**suggestion (testing):** Consider adding coverage for `_request_url` behavior with default ports, IPv6 literals, and non-pinned (public) URLs.
This test covers the pinned-address path for a hostname with an explicit non-standard HTTPS port. To more fully exercise `_request_url`, consider a parametrized test including: (1) HTTPS without an explicit port (expect `port=443` and no port in the `Host` header), (2) HTTP without an explicit port (expect `port=80`), (3) an IPv6 literal hostname (ensure `Host` is wrapped in `[]` and the pool target uses the validated address), and (4) `validated_address is None` (allow_private_networks=True) where `http.request` is called directly with the full URL and `connection_from_host` is not used. This would cover all branches and help prevent regressions in authority/Host handling.
Suggested implementation:
```python
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()
@pytest.mark.parametrize(
"url, validated_address, allow_private_networks, expected_scheme, expected_port, expected_host_header, expects_pinned",
[
# (1) HTTPS without explicit port -> port=443, Host header without port
(
"https://rebind.example/data.json?existing=yes",
"93.184.216.34",
False,
"https",
443,
"rebind.example",
True,
),
# (2) HTTP without explicit port -> port=80, Host header without port
(
"http://rebind.example/data.json?existing=yes",
"93.184.216.34",
False,
"http",
80,
"rebind.example",
True,
),
# (3) IPv6 literal with explicit port -> Host wrapped in [], pool uses validated address
(
"https://[2001:db8::1]:8443/data.json?existing=yes",
"2001:db8::1",
False,
"https",
8443,
"[2001:db8::1]:8443",
True,
),
# (4) Non-pinned (public) URL, validated_address is None and allow_private_networks=True
# -> http.request called directly, connection_from_host not used
(
"https://public.example/data.json?existing=yes",
None,
True,
"https",
None,
"public.example",
False,
),
],
)
@patch("json2xml.utils.get_http_client")
@patch("json2xml.utils.socket.getaddrinfo")
def test_request_url_handles_authority_and_host_header(
self,
mock_getaddrinfo: Mock,
mock_get_http_client: Mock,
url: str,
validated_address: str | None,
allow_private_networks: bool,
expected_scheme: str,
expected_port: int | None,
expected_host_header: str,
expects_pinned: bool,
) -> None:
"""Exercise `_request_url` authority handling for default ports, IPv6 literals, and public URLs."""
urllib3 = Mock()
http = Mock()
timeout = Mock()
pool = Mock()
response = Mock()
response.data = b'{"ok": true}'
pool.request.return_value = response
# When validated_address is provided, emulate successful resolution to that address.
# When it's None, emulate no pinned address resolution so the public URL path is taken.
if validated_address is not None:
mock_getaddrinfo.return_value = [
(None, None, None, None, (validated_address, 0)),
]
else:
mock_getaddrinfo.return_value = []
http.connection_from_host.return_value = pool
mock_get_http_client.return_value = (urllib3, http, timeout)
result = readfromurl(
url,
params={"added": "yes"},
allow_private_networks=allow_private_networks,
)
assert result == {"ok": True}
if expects_pinned:
# Pinned address path: connection_from_host is used, http.request is not.
http.request.assert_not_called()
http.connection_from_host.assert_called_once_with(
validated_address,
port=expected_port,
scheme=expected_scheme,
pool_kwargs={
"assert_hostname": urlsplit(url).hostname,
"server_hostname": urlsplit(url).hostname,
},
)
# Path and query should be stripped from the URL and passed to the pool.
parsed = urlsplit(url)
path_and_query = parsed.path
if parsed.query:
path_and_query += f"?{parsed.query}"
pool.request.assert_called_once_with(
"GET",
path_and_query,
fields={"added": "yes"},
headers={"Host": expected_host_header},
timeout=timeout,
retries=False,
redirect=False,
preload_content=False,
)
response.close.assert_called_once_with()
else:
# Non-pinned path: http.request is called directly with the full URL,
# and connection_from_host is not used at all.
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:
```
1. The new test uses `urlsplit` from `urllib.parse`. If this is not already imported at the top of `tests/test_utils.py`, add:
`from urllib.parse import urlsplit`.
2. Ensure `pytest`, `patch`, and `Mock` are imported consistently with the rest of the file (e.g., `from unittest.mock import Mock, patch`).
3. The behavior of `readfromurl` regarding `validated_address` and `allow_private_networks` may depend on how `socket.getaddrinfo` results are interpreted. Adjust the `mock_getaddrinfo.return_value` setup if your implementation uses a different pattern to derive `validated_address` (for example, multiple entries or additional flags).
4. If `_request_url` is a separate helper and you prefer testing it directly, you can refactor this parametrized test to call `_request_url` instead of `readfromurl`, while keeping the same expectations for `connection_from_host`, `http.request`, and the `Host` header.
</issue_to_address>
### Comment 2
<location path="json2xml/utils.py" line_range="64" />
<code_context>
# @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]:
</code_context>
<issue_to_address>
**issue (complexity):** Consider refactoring the new URL and decompression helpers into smaller, single-responsibility functions to clarify control flow and make the logic easier to follow and maintain.
1. Split `_validate_url` into pure validation and address resolution
Right now `_validate_url` both validates and returns connection metadata `(parsed, validated_address)`. You can keep behaviour while separating responsibilities and simplifying call sites:
```python
def _validate_url(url: str) -> SplitResult:
try:
parsed = urlsplit(url)
port = parsed.port
except (TypeError, ValueError) as error:
raise URLReadError("URL is not valid") from error
if parsed.scheme not in {"http", "https"}:
raise URLReadError("URL must use HTTP or HTTPS")
if parsed.username is not None or parsed.password is not None:
raise URLReadError("URL must not contain credentials")
if parsed.hostname is None:
raise URLReadError("URL must include a hostname")
return parsed
```
```python
def _resolve_validated_address(parsed: SplitResult, allow_private_networks: bool) -> str | None:
if allow_private_networks:
return 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,
type=socket.SOCK_STREAM,
)
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])
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])
```
Call site in `readfromurl` becomes simpler and matches the names:
```python
parsed = _validate_url(url)
validated_address = _resolve_validated_address(parsed, allow_private_networks)
```
2. Narrow `_request_url` to only the “validated address” path
You can keep the trivial path (`http.request("GET", parsed.geturl(), ...)`) directly in `readfromurl`, and have `_request_url` only handle the more complex routing. That avoids branching inside `_request_url` and makes control flow clearer:
```python
def _request_via_validated_address(
http: Any,
parsed: SplitResult,
validated_address: str,
params: dict[str, str] | None,
timeout: Any,
) -> Any:
request_options = {
"fields": params,
"timeout": timeout,
"retries": False,
"redirect": False,
"preload_content": False,
}
assert parsed.hostname is not None
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
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,
)
```
Then `readfromurl` becomes:
```python
urllib3, http, timeout = _get_http_client()
request_options = {
"fields": params,
"timeout": timeout,
"retries": False,
"redirect": False,
"preload_content": False,
}
if validated_address is None:
response = http.request("GET", parsed.geturl(), **request_options)
else:
response = _request_via_validated_address(
http, parsed, validated_address, params, timeout
)
```
3. Isolate deflate header detection out of `_compression_decoder`
You can keep `_compression_decoder` readable by pushing the bit-twiddling into a tiny helper:
```python
def _has_zlib_header(data: bytes) -> bool:
if len(data) < 2:
return False
cmf, flg = data[0], data[1]
# Compression method 8 (DEFLATE) and header checksum multiple of 31
return (cmf & 0x0F) == 8 and ((cmf << 8) | flg) % 31 == 0
```
```python
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":
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}")
```
4. Extract the decompression state machine from `_read_response_data`
The nested loops are correct but hard to follow. You can move the inner state machine into a focused helper that returns bytes, keeping `_read_response_data` thin:
```python
def _decompress_with_limit(
response: Any,
decoder: Any,
first_chunk: bytes,
max_response_bytes: int,
) -> bytes:
response_data = bytearray()
compressed_chunk = first_chunk
try:
while compressed_chunk:
pending = compressed_chunk
while pending:
remaining = max_response_bytes + 1 - len(response_data)
decoded = decoder.decompress(pending, remaining)
response_data.extend(decoded)
if len(response_data) > max_response_bytes:
raise URLReadError("URL response exceeds maximum size")
pending = decoder.unconsumed_tail
compressed_chunk = response.read(
COMPRESSED_READ_CHUNK_BYTES,
decode_content=False,
)
except zlib.error as error:
raise URLReadError("URL returned invalid compressed data") from error
if not decoder.eof or decoder.unused_data:
raise URLReadError("URL returned invalid compressed data")
return bytes(response_data)
```
Then `_read_response_data` focuses on “choose encoding path”:
```python
def _read_response_data(response: Any, max_response_bytes: int) -> bytes:
encoding = response.headers.get("Content-Encoding", "").strip().lower()
if encoding in {"", "identity"}:
data = response.read(max_response_bytes + 1, decode_content=False)
if len(data) > max_response_bytes:
raise URLReadError("URL response exceeds maximum size")
return data
first_chunk = response.read(
COMPRESSED_READ_CHUNK_BYTES,
decode_content=False,
)
decoder = _compression_decoder(encoding, first_chunk)
return _decompress_with_limit(response, decoder, first_chunk, max_response_bytes)
```
These changes preserve the new security and size-limit behaviour while flattening control flow and separating concerns into smaller, purpose-specific helpers.
</issue_to_address>Help me be more useful! Please click 👍 or 👎 on each comment and I'll use the feedback to improve your reviews.
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
1 participant
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Summary
allow_private_networksURLReadErrorVerification
uv run ruff check json2xml testsuvx ty check json2xml testsuv run pytest --cov=json2xml --cov-report=term --cov-fail-under=100 -q tests(486 passed, 100% coverage)cargo test --manifest-path rust/Cargo.toml(48 passed)lat checkhttps://pypi.org/pypi/json2xml/jsonRelease prerequisite
Before the next Python release, publish a compatible accelerator newer than
json2xml-rs==0.4.2, then raise thejson2xml[fast]anduv.lockminimum to that published version. The current compatibility probe safely disables 0.4.2 in the meantime.Summary by Sourcery
Harden URL-based JSON reads against DNS rebinding and compressed-response abuse while tightening URL validation and documenting the new security constraints.
Bug Fixes:
Enhancements:
Documentation:
Tests: