diff --git a/.github/workflows/python-code-quality.yml b/.github/workflows/python-code-quality.yml index c198514d3a8..a900214ef55 100644 --- a/.github/workflows/python-code-quality.yml +++ b/.github/workflows/python-code-quality.yml @@ -141,3 +141,32 @@ jobs: UV_CACHE_DIR: /tmp/.uv-cache - name: Run tests/samples type checkers (mypy, pyrefly, ty) run: uv run --locked --group tool-runtime python -m scripts.workspace_poe_tasks ci-test-typing + + declarative-httpx-compatibility: + name: Declarative HTTPX Compatibility (${{ matrix.httpx-version }}) + strategy: + fail-fast: false + matrix: + httpx-version: ["0.27.0", "0.28.1"] + runs-on: ubuntu-latest + defaults: + run: + working-directory: ./python + steps: + - uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 # v7.0.1 + - uses: astral-sh/setup-uv@c18668ad3cf93ea998bef934396af7bb5c839dc7 # v10.2.0 + with: + version-file: "python/pyproject.toml" + - name: Install isolated declarative test environment + run: | + uv venv .venv-httpx --python 3.11 + uv pip install --python .venv-httpx/bin/python \ + -e packages/core -e packages/declarative \ + "httpx==${{ matrix.httpx-version }}" \ + pytest==9.1.1 pytest-asyncio==1.4.0 pytest-timeout==2.4.0 + uv pip check --python .venv-httpx/bin/python + - name: Test bounded HTTP handling + run: | + .venv-httpx/bin/python -m pytest \ + packages/declarative/tests/test_default_http_request_handler.py \ + -m "not integration" -q -o addopts='' diff --git a/python/packages/declarative/AGENTS.md b/python/packages/declarative/AGENTS.md index 3e3fe5879a3..059c1c605d7 100644 --- a/python/packages/declarative/AGENTS.md +++ b/python/packages/declarative/AGENTS.md @@ -14,6 +14,13 @@ YAML/JSON-based declarative agent and workflow definitions. Both state classes validate object attribute names but leave dictionary keys unchanged. +`DefaultHttpRequestHandler(max_response_body_bytes=...)` defaults to a 10 MiB +wire and decoded-stage body limit, including redirect and authentication responses. +It supports identity, gzip, and deflate, rejects unsupported or incomplete encodings, +and closes response streams without closing caller-owned clients. +HTTPX is constrained to the 0.27 and 0.28 series; focused CI checks both +series because response caching and default-header integration use its internals. + ## MCP Handler Lifetimes `DefaultMCPToolHandler` caches/coalesces sessions only without a `client_provider`. diff --git a/python/packages/declarative/README.md b/python/packages/declarative/README.md index 42a2a2bc305..b58e5b3a2a1 100644 --- a/python/packages/declarative/README.md +++ b/python/packages/declarative/README.md @@ -40,6 +40,31 @@ Explicit `Cookie` request headers remain supported. Response `Set-Cookie` header remain available in `HttpRequestResult.headers`; disabling persistence does not redact the response. +## HTTP response body limits + +`DefaultHttpRequestHandler` limits response bodies to **10 MiB** by default. +Configure a positive integer `max_response_body_bytes` on the handler when a host +requires a different bound. The bound applies to received bytes and each decoded +stage, including intermediate redirect and authentication responses. Responses +above the bound raise `httpx.StreamError` before text conversion or workflow +assignment, and their streams are closed without closing caller-owned clients. + +The handler restricts HTTPX's generated `Accept-Encoding` default to `gzip, deflate` +and supports identity, gzip, and deflate +responses, including up to four encoding stages. Unsupported encodings, +incomplete encoded bodies, and trailing encoded data raise `httpx.DecodingError`. +Empty encoded responses are validated unless the response is bodyless (HEAD, +1xx, 204, 205, 304, or a successful CONNECT response). Deflate input is retained +within the same bound until its wrapped or raw format can be validated. +Explicit request `Accept-Encoding` headers and client values that differ from +HTTPX's generated default are preserved but do not enable additional decoders. +Caller-supplied response hooks and transports must not +buffer response bodies before the handler applies its limit. + +HTTPX support is limited to the 0.27 and 0.28 series because the bounded +intermediate-response cache and generated default-header integration depend +on HTTPX internals. Compatibility tests cover HTTPX 0.27.0 and 0.28.1. + ## MCP handler session lifetime `DefaultMCPToolHandler()` reuses MCP sessions through a bounded LRU cache. diff --git a/python/packages/declarative/agent_framework_declarative/_workflows/_http_handler.py b/python/packages/declarative/agent_framework_declarative/_workflows/_http_handler.py index e73cb860223..d21cfa0e2d7 100644 --- a/python/packages/declarative/agent_framework_declarative/_workflows/_http_handler.py +++ b/python/packages/declarative/agent_framework_declarative/_workflows/_http_handler.py @@ -21,13 +21,16 @@ from __future__ import annotations import asyncio -from collections.abc import Awaitable, Callable, Mapping +import sys +import zlib +from collections.abc import AsyncGenerator, Awaitable, Callable, Mapping from dataclasses import dataclass, field, replace from http.cookiejar import CookieJar, DefaultCookiePolicy from typing import Any, Protocol, runtime_checkable from urllib.parse import parse_qsl, quote, urlencode, urlsplit import httpx +from httpx._client import ACCEPT_ENCODING __all__ = [ "DefaultHttpRequestHandler", @@ -142,6 +145,16 @@ class DefaultHttpRequestHandler: Explicit outbound ``Cookie`` headers and response ``Set-Cookie`` headers are preserved; the owned-client policy only prevents automatic cookie persistence. + Response bodies are limited to ``max_response_body_bytes`` (10 MiB by default), + both on the wire and after each gzip/deflate decoding stage. Other content encodings + are rejected. The limit also applies to redirect and authentication responses. + Caller-supplied response hooks must not eagerly read response bodies. + + Keyword Args: + client: Optional caller-owned HTTP client. + client_provider: Optional per-request client provider. + max_response_body_bytes: Positive wire and decoded response-body limit in bytes. + .. warning:: This handler performs **no** URL filtering or SSRF protection. Wrap or @@ -153,7 +166,13 @@ def __init__( *, client: httpx.AsyncClient | None = None, client_provider: ClientProvider | None = None, + max_response_body_bytes: int = 10 * 1024 * 1024, ) -> None: + if isinstance(max_response_body_bytes, bool) or not isinstance(max_response_body_bytes, int): + raise ValueError("max_response_body_bytes must be a positive integer.") + if max_response_body_bytes <= 0: + raise ValueError("max_response_body_bytes must be a positive integer.") + self._max_response_body_bytes = max_response_body_bytes self._owned_client: httpx.AsyncClient | None = None self._caller_client = client self._client_provider = client_provider @@ -193,6 +212,8 @@ async def send(self, info: HttpRequestInfo) -> HttpRequestResult: timeout = httpx.USE_CLIENT_DEFAULT headers = dict(info.headers) + if not _has_header(headers, "accept-encoding") and client.headers.get("accept-encoding") == ACCEPT_ENCODING: + headers["Accept-Encoding"] = "gzip, deflate" content: bytes | str | None = None if info.body is not None: content = info.body @@ -224,7 +245,19 @@ async def send(self, info: HttpRequestInfo) -> HttpRequestResult: ) request.url = url - response = await client.send(request) + auth = client.auth + if auth is None: + auth = httpx.BasicAuth(url.username, url.password) if url.username or url.password else httpx.Auth() + response = await client.send( + request, + stream=True, + follow_redirects=False, + auth=_ResponseLimitAuth( + auth, + limit=self._max_response_body_bytes, + follow_redirects=client.follow_redirects, + ), + ) # Preserve multi-value headers (e.g. multiple Set-Cookie) as list[str]. # Normalize names to lowercase so lookups are consistent and case @@ -296,3 +329,130 @@ def _has_header(headers: Mapping[str, str], name: str) -> bool: """Case-insensitive header presence check.""" needle = name.lower() return any(key.lower() == needle for key in headers) + + +class _ResponseLimitAuth(httpx.Auth): + """Read every response within bounds before HTTPX can buffer intermediate bodies.""" + + def __init__(self, auth: httpx.Auth, *, limit: int, follow_redirects: bool) -> None: + self._auth = auth + self._limit = limit + self._follow_redirects = follow_redirects + + async def async_auth_flow(self, request: httpx.Request) -> AsyncGenerator[httpx.Request, httpx.Response]: + flow = self._auth.async_auth_flow(request) + try: + request = await flow.__anext__() + while True: + response = yield request + await _read_bounded_response(response, self._limit) + if self._follow_redirects and response.next_request is not None: + request = response.next_request + continue + try: + request = await flow.asend(response) + except StopAsyncIteration: + return + finally: + await flow.aclose() + + +async def _read_bounded_response(response: httpx.Response, limit: int) -> None: + """Cache only bounded decoded content, preserving HTTPX text and auth behavior.""" + try: + if response.is_stream_consumed: + if len(response.content) > limit: + raise httpx.StreamError("HTTP response body exceeds the configured limit.") + return + + encodings = [ + encoding.strip().lower() + for encoding in response.headers.get("content-encoding", "").split(",") + if encoding.strip().lower() != "identity" and encoding.strip() + ] + if ( + response.request.method == "HEAD" + or 100 <= response.status_code < 200 + or response.status_code in {204, 205, 304} + or (response.request.method == "CONNECT" and 200 <= response.status_code < 300) + ): + encodings = [] + if len(encodings) > 4: + raise httpx.DecodingError("HTTP response has too many content encoding stages.", request=response.request) + decoders = [_BoundedContentDecoder(encoding, limit) for encoding in reversed(encodings)] + body = bytearray() + received = 0 + async for chunk in response.aiter_raw(): + received += len(chunk) + if received > limit: + raise httpx.StreamError("HTTP response body exceeds the configured limit.") + for decoder in decoders: + chunk = decoder.decode(chunk) + if len(body) + len(chunk) > limit: + raise httpx.StreamError("HTTP response body exceeds the configured limit.") + body.extend(chunk) + for index, decoder in enumerate(decoders): + chunk = decoder.finish() + for downstream in decoders[index + 1 :]: + chunk = downstream.decode(chunk) + if len(body) + len(chunk) > limit: + raise httpx.StreamError("HTTP response body exceeds the configured limit.") + body.extend(chunk) + # HTTPX rereads intermediate responses during auth/redirect handling. Populate + # its content cache so those reads cannot trigger automatic decoding again. + response._content = bytes(body) # pyright: ignore[reportPrivateUsage] + finally: + await response.aclose() + + +class _BoundedContentDecoder: + def __init__(self, encoding: str, limit: int) -> None: + if encoding not in {"gzip", "deflate"}: + raise httpx.DecodingError(f"Unsupported HTTP response content encoding: {encoding}.") + self._limit = limit + self._decoded = 0 + self._input = bytearray() + self._decoder = zlib.decompressobj(zlib.MAX_WBITS | 16) if encoding == "gzip" else None + + def decode(self, data: bytes) -> bytes: + if self._decoder is None: + # Raw deflate can share a wrapped stream's prefix. Retain bounded input + # until finalization so a failed candidate never reaches downstream stages. + if len(self._input) + len(data) > self._limit: + raise httpx.StreamError("HTTP response body exceeds the configured limit.") + self._input.extend(data) + return b"" + try: + decoded = self._decoder.decompress(data, min(self._limit - self._decoded + 1, sys.maxsize)) + except zlib.error as exc: + raise httpx.DecodingError(str(exc)) from exc + self._decoded += len(decoded) + if self._decoded > self._limit: + raise httpx.StreamError("HTTP response body exceeds the configured limit.") + if self._decoder.unused_data: + raise httpx.DecodingError("Unexpected data after HTTP response content encoding.") + return decoded + + def finish(self) -> bytes: + if self._decoder is not None: + if not self._decoder.eof: + raise httpx.DecodingError("Incomplete HTTP response content encoding.") + return b"" + + data = bytes(self._input) + self._input.clear() + failures: list[httpx.DecodingError | httpx.StreamError] = [] + for mode in (zlib.MAX_WBITS, -zlib.MAX_WBITS): + self._decoder = zlib.decompressobj(mode) + self._decoded = 0 + try: + decoded = self.decode(data) + if not self._decoder.eof: + raise httpx.DecodingError("Incomplete HTTP response content encoding.") + return decoded + except (httpx.DecodingError, httpx.StreamError) as exc: + failures.append(exc) + for error in failures: + if isinstance(error, httpx.StreamError): + raise error + raise failures[-1] diff --git a/python/packages/declarative/pyproject.toml b/python/packages/declarative/pyproject.toml index 5914619d8c6..72e46bf055a 100644 --- a/python/packages/declarative/pyproject.toml +++ b/python/packages/declarative/pyproject.toml @@ -23,7 +23,7 @@ classifiers = [ ] dependencies = [ "agent-framework-core>=1.20.0,<2", - "httpx>=0.27,<1", + "httpx>=0.27,<0.29", "powerfx>=0.0.32,<0.0.35; python_version < '3.14'", "pyyaml>=6.0,<7.0", ] diff --git a/python/packages/declarative/tests/test_default_http_request_handler.py b/python/packages/declarative/tests/test_default_http_request_handler.py index 7d4391fa43f..637e2d064ca 100644 --- a/python/packages/declarative/tests/test_default_http_request_handler.py +++ b/python/packages/declarative/tests/test_default_http_request_handler.py @@ -12,7 +12,9 @@ from __future__ import annotations import asyncio +import gzip import sys +import zlib from collections.abc import AsyncIterator from typing import Any from unittest.mock import AsyncMock, patch @@ -47,6 +49,21 @@ def _make_handler(transport: httpx.MockTransport) -> DefaultHttpRequestHandler: return DefaultHttpRequestHandler(client=client) +class _ResponseStream(httpx.AsyncByteStream): + def __init__(self, chunks: list[bytes]) -> None: + self.chunks = chunks + self.closed = False + self.reads = 0 + + async def __aiter__(self) -> AsyncIterator[bytes]: + for chunk in self.chunks: + self.reads += 1 + yield chunk + + async def aclose(self) -> None: + self.closed = True + + class TestRequestComposition: @pytest.mark.asyncio async def test_query_parameters_merged_into_url(self) -> None: @@ -750,6 +767,336 @@ async def test_empty_method_raises(self) -> None: await handler.send(HttpRequestInfo(method="", url="https://x.test/")) +class TestResponseBodyLimits: + @pytest.mark.parametrize("limit", [sys.maxsize, sys.maxsize + 1, 10**100]) + @pytest.mark.parametrize("encoding", ["gzip", "deflate"]) + async def test_large_response_body_limit_accepts_small_encoded_body(self, limit: int, encoding: str) -> None: + body = gzip.compress(b"ok") if encoding == "gzip" else zlib.compress(b"ok") + stream = _ResponseStream([body]) + response = httpx.Response(200, headers={"Content-Encoding": encoding}, stream=stream) + async with ( + httpx.AsyncClient(transport=httpx.MockTransport(lambda request: response)) as client, + DefaultHttpRequestHandler(client=client, max_response_body_bytes=limit) as handler, + ): + result = await handler.send(HttpRequestInfo(method="GET", url="https://api.example.test/items")) + assert result.body == "ok" + assert stream.closed + + @pytest.mark.parametrize("client_source", ["caller", "provider"]) + @pytest.mark.parametrize( + ("client_encoding", "request_encoding", "expected"), + [ + ("identity", None, "identity"), + ("deflate;q=0.5", None, "deflate;q=0.5"), + ("identity", "gzip", "gzip"), + ], + ) + async def test_selected_client_encoding_policy_is_preserved( + self, client_source: str, client_encoding: str, request_encoding: str | None, expected: str + ) -> None: + requests: list[httpx.Request] = [] + + def respond(request: httpx.Request) -> httpx.Response: + requests.append(request) + return httpx.Response(200, text="ok") + + async with httpx.AsyncClient( + headers={"Accept-Encoding": client_encoding}, transport=httpx.MockTransport(respond) + ) as client: + + async def provider(info: HttpRequestInfo) -> httpx.AsyncClient: + return client + + async with DefaultHttpRequestHandler( + client=client if client_source == "caller" else None, + client_provider=provider if client_source == "provider" else None, + ) as handler: + result = await handler.send( + HttpRequestInfo( + method="GET", + url="https://api.example.test/items", + headers={"Accept-Encoding": request_encoding} if request_encoding is not None else {}, + ) + ) + assert result.body == "ok" + assert requests[0].headers["Accept-Encoding"] == expected + assert client.headers["Accept-Encoding"] == client_encoding + assert not client.is_closed + + @pytest.mark.parametrize("encoding", ["identity", "gzip", "gzip, gzip"]) + async def test_response_body_above_default_limit_is_rejected(self, encoding: str) -> None: + body = b"a" * (10 * 1024 * 1024 + 1) + for _ in range(encoding.count("gzip")): + body = gzip.compress(body) + + stream = _ResponseStream([body]) + response = httpx.Response(200, headers={"Content-Encoding": encoding}, stream=stream) + async with httpx.AsyncClient(transport=httpx.MockTransport(lambda request: response)) as client: + async with DefaultHttpRequestHandler(client=client) as handler: + with pytest.raises(httpx.StreamError, match="response body.*limit"): + await handler.send(HttpRequestInfo(method="GET", url="https://api.example.test/items")) + assert stream.closed + assert not client.is_closed + + @pytest.mark.parametrize("encoding", ["identity", "gzip", "gzip, gzip", "deflate", "raw-deflate"]) + @pytest.mark.parametrize("size", [127, 128, 129]) + async def test_configured_limit_counts_decoded_bytes(self, encoding: str, size: int) -> None: + body = b"a" * size + for _ in range(encoding.count("gzip")): + body = gzip.compress(body) + if encoding == "deflate": + body = zlib.compress(body) + elif encoding == "raw-deflate": + compressor = zlib.compressobj(wbits=-zlib.MAX_WBITS) + body = compressor.compress(body) + compressor.flush() + stream = _ResponseStream([body[index : index + 1] for index in range(len(body))]) + response = httpx.Response(200, headers={"Content-Encoding": encoding.replace("raw-", "")}, stream=stream) + async with httpx.AsyncClient(transport=httpx.MockTransport(lambda request: response)) as client: + async with DefaultHttpRequestHandler(client=client, max_response_body_bytes=128) as handler: + if size > 128: + with pytest.raises(httpx.StreamError, match="response body.*limit"): + await handler.send(HttpRequestInfo(method="GET", url="https://api.example.test/items")) + else: + result = await handler.send(HttpRequestInfo(method="GET", url="https://api.example.test/items")) + assert result.body == "a" * size + assert stream.closed + + async def test_wire_limit_stops_before_next_chunk(self) -> None: + stream = _ResponseStream([b"a" * 129, b"unused"]) + response = httpx.Response(200, stream=stream) + async with ( + httpx.AsyncClient(transport=httpx.MockTransport(lambda request: response)) as client, + DefaultHttpRequestHandler(client=client, max_response_body_bytes=128) as handler, + ): + with pytest.raises(httpx.StreamError, match="response body.*limit"): + await handler.send(HttpRequestInfo(method="GET", url="https://api.example.test/items")) + assert stream.reads == 1 + assert stream.closed + + async def test_intermediate_decoding_stage_is_limited(self) -> None: + body = gzip.compress(gzip.compress(b"a" * 64, compresslevel=0)) + stream = _ResponseStream([body]) + response = httpx.Response(200, headers={"Content-Encoding": "gzip, gzip"}, stream=stream) + async with ( + httpx.AsyncClient(transport=httpx.MockTransport(lambda request: response)) as client, + DefaultHttpRequestHandler(client=client, max_response_body_bytes=64) as handler, + ): + with pytest.raises(httpx.StreamError, match="response body.*limit"): + await handler.send(HttpRequestInfo(method="GET", url="https://api.example.test/items")) + assert stream.closed + + @pytest.mark.parametrize("intermediate_status", [307, 401]) + async def test_intermediate_response_is_limited_before_next_request(self, intermediate_status: int) -> None: + requests: list[httpx.Request] = [] + stream = _ResponseStream([gzip.compress(b"a" * 129)]) + + def respond(request: httpx.Request) -> httpx.Response: + requests.append(request) + if len(requests) > 1: + return httpx.Response(200, text="ok") + return httpx.Response( + intermediate_status, + headers={ + "Content-Encoding": "gzip", + "Location": "/next", + "WWW-Authenticate": 'Digest realm="test", nonce="test", qop="auth"', + }, + stream=stream, + ) + + async with ( + httpx.AsyncClient( + transport=httpx.MockTransport(respond), + follow_redirects=True, + auth=httpx.DigestAuth("user", "pass"), + ) as client, + DefaultHttpRequestHandler(client=client, max_response_body_bytes=128) as handler, + ): + with pytest.raises(httpx.StreamError, match="response body.*limit"): + await handler.send(HttpRequestInfo(method="GET", url="https://api.example.test/items")) + assert len(requests) == 1 + assert stream.closed + + async def test_authentication_retry_retains_decoded_response(self) -> None: + requests: list[httpx.Request] = [] + + def respond(request: httpx.Request) -> httpx.Response: + requests.append(request) + return httpx.Response( + 401 if len(requests) == 1 else 200, + headers={ + "Content-Encoding": "gzip", + "WWW-Authenticate": 'Digest realm="test", nonce="test", qop="auth"', + "Content-Type": "text/plain; charset=iso-8859-1", + }, + stream=_ResponseStream([gzip.compress(b"\xe9")]), + ) + + async with ( + httpx.AsyncClient(transport=httpx.MockTransport(respond), auth=httpx.DigestAuth("user", "pass")) as client, + DefaultHttpRequestHandler(client=client, max_response_body_bytes=128) as handler, + ): + result = await handler.send(HttpRequestInfo(method="GET", url="https://api.example.test/items")) + assert len(requests) == 2 + assert requests[1].headers["Authorization"].startswith("Digest ") + assert result.status_code == 200 + assert result.body == "\u00e9" + assert result.headers["content-encoding"] == ["gzip"] + + async def test_redirect_limit_and_cross_host_headers_are_preserved(self) -> None: + requests: list[httpx.Request] = [] + streams: list[_ResponseStream] = [] + + def respond(request: httpx.Request) -> httpx.Response: + requests.append(request) + stream = _ResponseStream([gzip.compress(b"ok")]) + streams.append(stream) + return httpx.Response( + 307, + headers={"Content-Encoding": "gzip", "Location": "https://other.example.test/next"}, + stream=stream, + ) + + async with ( + httpx.AsyncClient( + transport=httpx.MockTransport(respond), + follow_redirects=True, + max_redirects=1, + auth=httpx.BasicAuth("user", "pass"), + ) as client, + DefaultHttpRequestHandler(client=client) as handler, + ): + with pytest.raises(httpx.TooManyRedirects): + await handler.send(HttpRequestInfo(method="GET", url="https://api.example.test/items")) + assert len(requests) == 2 + assert "Authorization" in requests[0].headers + assert "Authorization" not in requests[1].headers + assert all(stream.closed for stream in streams) + + async def test_cancellation_closes_pending_response(self) -> None: + started = asyncio.Event() + + class PendingStream(_ResponseStream): + async def __aiter__(self) -> AsyncIterator[bytes]: + started.set() + await asyncio.Event().wait() + yield b"unused" + + stream = PendingStream([]) + response = httpx.Response(200, stream=stream) + async with ( + httpx.AsyncClient(transport=httpx.MockTransport(lambda request: response)) as client, + DefaultHttpRequestHandler(client=client) as handler, + ): + task = asyncio.create_task( + handler.send(HttpRequestInfo(method="GET", url="https://api.example.test/items")) + ) + await started.wait() + assert task.cancel() + with pytest.raises(asyncio.CancelledError): + _ = await task + assert stream.closed + assert not client.is_closed + + @pytest.mark.parametrize( + ("method", "status"), + [ + ("HEAD", 200), + ("GET", 100), + ("GET", 101), + ("GET", 199), + ("GET", 204), + ("GET", 205), + ("GET", 304), + ("CONNECT", 200), + ("CONNECT", 299), + ], + ) + async def test_empty_bodyless_response_does_not_require_encoded_body(self, method: str, status: int) -> None: + stream = _ResponseStream([]) + response = httpx.Response(status, headers={"Content-Encoding": "gzip"}, stream=stream) + async with ( + httpx.AsyncClient(transport=httpx.MockTransport(lambda request: response)) as client, + DefaultHttpRequestHandler(client=client) as handler, + ): + result = await handler.send(HttpRequestInfo(method=method, url="https://api.example.test/items")) + assert result.body == "" + assert stream.closed + + @pytest.mark.parametrize("encoding", ["gzip", "deflate"]) + @pytest.mark.parametrize(("method", "status"), [("GET", 200), ("CONNECT", 300)]) + async def test_empty_encoded_response_with_body_is_rejected(self, encoding: str, method: str, status: int) -> None: + stream = _ResponseStream([]) + response = httpx.Response(status, headers={"Content-Encoding": encoding}, stream=stream) + async with ( + httpx.AsyncClient(transport=httpx.MockTransport(lambda request: response)) as client, + DefaultHttpRequestHandler(client=client) as handler, + ): + with pytest.raises(httpx.DecodingError, match="Incomplete"): + await handler.send(HttpRequestInfo(method=method, url="https://api.example.test/items")) + assert stream.closed + + @pytest.mark.parametrize("chunk_size", [1, 166]) + async def test_raw_deflate_with_wrapped_header_prefix_is_decoded(self, chunk_size: int) -> None: + body = bytes.fromhex("789c0063ff") + b"a" * 156 + bytes.fromhex("010000ffff") + stream = _ResponseStream([body[index : index + chunk_size] for index in range(0, len(body), chunk_size)]) + response = httpx.Response(200, headers={"Content-Encoding": "deflate"}, stream=stream) + async with ( + httpx.AsyncClient(transport=httpx.MockTransport(lambda request: response)) as client, + DefaultHttpRequestHandler(client=client, max_response_body_bytes=256) as handler, + ): + result = await handler.send(HttpRequestInfo(method="GET", url="https://api.example.test/items")) + assert result.body == "a" * 156 + assert stream.closed + + @pytest.mark.parametrize("encoding", ["gzip, deflate", "deflate, gzip", "deflate, deflate"]) + @pytest.mark.parametrize("size", [128, 129]) + async def test_mixed_encoding_finalization_preserves_limit(self, encoding: str, size: int) -> None: + body = b"a" * size + for stage in encoding.split(", "): + body = gzip.compress(body) if stage == "gzip" else zlib.compress(body) + stream = _ResponseStream([body[index : index + 1] for index in range(len(body))]) + response = httpx.Response(200, headers={"Content-Encoding": encoding}, stream=stream) + async with ( + httpx.AsyncClient(transport=httpx.MockTransport(lambda request: response)) as client, + DefaultHttpRequestHandler(client=client, max_response_body_bytes=128) as handler, + ): + if size > 128: + with pytest.raises(httpx.StreamError, match="response body.*limit"): + await handler.send(HttpRequestInfo(method="GET", url="https://api.example.test/items")) + else: + result = await handler.send(HttpRequestInfo(method="GET", url="https://api.example.test/items")) + assert result.body == "a" * size + assert stream.closed + + @pytest.mark.parametrize( + ("encoding", "body"), + [ + pytest.param("br", b"body", id="unsupported"), + pytest.param("gzip", b"invalid", id="invalid-gzip"), + pytest.param("gzip", gzip.compress(b"body", mtime=0)[:-1], id="incomplete-gzip"), + pytest.param("deflate", zlib.compress(b"body")[:-1], id="incomplete-deflate"), + pytest.param("gzip", gzip.compress(b"body", mtime=0) + b"extra", id="trailing-gzip-data"), + pytest.param("gzip, gzip, gzip, gzip, gzip", b"body", id="excessive-encoding-stages"), + ], + ) + async def test_invalid_content_encoding_closes_response(self, encoding: str, body: bytes) -> None: + stream = _ResponseStream([body]) + response = httpx.Response(200, headers={"Content-Encoding": encoding}, stream=stream) + async with ( + httpx.AsyncClient(transport=httpx.MockTransport(lambda request: response)) as client, + DefaultHttpRequestHandler(client=client) as handler, + ): + with pytest.raises(httpx.DecodingError): + await handler.send(HttpRequestInfo(method="GET", url="https://api.example.test/items")) + assert stream.closed + + @pytest.mark.parametrize("limit", [0, -1, True, 1.5, None]) + def test_invalid_response_body_limit_is_rejected(self, limit: Any) -> None: + with pytest.raises(ValueError, match="positive integer"): + DefaultHttpRequestHandler(max_response_body_bytes=limit) + + class TestAsyncContextManager: @pytest.mark.asyncio async def test_context_manager_closes_owned_client(self) -> None: diff --git a/python/uv.lock b/python/uv.lock index 13726f1d35c..55deaba74b1 100644 --- a/python/uv.lock +++ b/python/uv.lock @@ -552,7 +552,7 @@ dev = [ [package.metadata] requires-dist = [ { name = "agent-framework-core", editable = "packages/core" }, - { name = "httpx", specifier = ">=0.27,<1" }, + { name = "httpx", specifier = ">=0.27,<0.29" }, { name = "powerfx", marker = "python_full_version < '3.14'", specifier = ">=0.0.32,<0.0.35" }, { name = "pyyaml", specifier = ">=6.0,<7.0" }, ]