From dae47930aa2c8153cb2df1a4f9084b7eaa52e04b Mon Sep 17 00:00:00 2001 From: hpiclaranet Date: Sat, 22 Aug 2026 00:40:02 +0100 Subject: [PATCH 1/3] fix(anthropic): stream tag duplication, 529 mapping, image fetch hardening Three reliability fixes in the Anthropic provider: - Stream: a thinking block followed directly by a tool_use block left current_block_type as 'thinking' (input_json_delta matched no type branch), so the closing think tag was emitted a second time at MessageStopEvent. Tool-use deltas now mark the block type as 'tool_use'. - Errors: HTTP 529 OverloadedError fell through the APIError catch-all into InvokeBadRequestError; it is now mapped to InvokeServerUnavailableError (server-side capacity, retryable). - Image fetch: requests.get had no timeout (hung URLs could block the worker indefinitely), no size cap, accepted any URL scheme and followed redirects. It now uses a (5s connect, 15s read) timeout, a 10 MB cap, http(s)-only URLs, and no redirect following. Tests use real SDK stream event objects; the five bug-target tests fail against the pre-fix code and the seven behavior guards pass on both. --- models/anthropic/manifest.yaml | 2 +- models/anthropic/models/llm/llm.py | 63 ++- .../tests/test_stream_and_hardening.py | 454 ++++++++++++++++++ 3 files changed, 510 insertions(+), 9 deletions(-) create mode 100644 models/anthropic/tests/test_stream_and_hardening.py diff --git a/models/anthropic/manifest.yaml b/models/anthropic/manifest.yaml index 59f848457..874aae278 100644 --- a/models/anthropic/manifest.yaml +++ b/models/anthropic/manifest.yaml @@ -25,4 +25,4 @@ resource: model: enabled: false type: plugin -version: 0.3.27 +version: 0.3.29 diff --git a/models/anthropic/models/llm/llm.py b/models/anthropic/models/llm/llm.py index ea7c9d101..7279f8bd7 100644 --- a/models/anthropic/models/llm/llm.py +++ b/models/anthropic/models/llm/llm.py @@ -1136,6 +1136,11 @@ def _handle_chat_generate_stream_response( ) elif hasattr(chunk.delta, "text"): current_block_type = "text" + elif hasattr(chunk.delta, "type") and chunk.delta.type == "input_json_delta": + # Tool-use block: mark as non-thinking so the closing + # think tag (already emitted at the block transition + # above) is not emitted a second time at stream end. + current_block_type = "tool_use" elif hasattr(chunk.delta, "type") and chunk.delta.type == "redacted_thinking": current_block_type = "redacted_thinking" assistant_prompt_message = AssistantPromptMessage(content="\n") @@ -1428,6 +1433,10 @@ def _create_image_content(self, content: ImagePromptMessageContent) -> dict: return result + # Cap for URL-fetched images: fail early with a clear error instead of + # bloating the request payload (base64 inflates size ~33%). + MAX_IMAGE_FETCH_BYTES = 10 * 1024 * 1024 + def _process_image_data(self, data: str) -> tuple[str, str]: """Process image data from URL or data URI.""" if data.startswith("data:"): @@ -1435,19 +1444,53 @@ def _process_image_data(self, data: str) -> tuple[str, str]: header, encoded = data.split(";base64,", 1) mime_type = header.replace("data:", "") return mime_type, encoded - + # Fetch from URL + url = data.strip() + if not url.lower().startswith(("http://", "https://")): + raise ValueError( + f"Unsupported image URL scheme: only http(s) URLs are allowed ({data})" + ) try: - response = requests.get(data) + # Bounded timeout (a hung URL must not block the worker indefinitely) + # and no redirect following (the fetched host is exactly the one given). + # stream=True + iter_content keeps memory bounded: a hostile server + # cannot push more than ~MAX_IMAGE_FETCH_BYTES into the process. + response = requests.get( + url, + timeout=(5.0, 15.0), + allow_redirects=False, + stream=True, + ) response.raise_for_status() - - with Image.open(io.BytesIO(response.content)) as img: + # raise_for_status only raises for >= 400: reject 3xx explicitly. + if 300 <= response.status_code < 400: + raise ValueError("Redirect responses are not allowed") + + length = int(response.headers.get("Content-Length") or 0) + if length > self.MAX_IMAGE_FETCH_BYTES: + raise ValueError( + f"Image exceeds the {self.MAX_IMAGE_FETCH_BYTES // (1024 * 1024)} MB limit" + ) + + chunks: list[bytes] = [] + received = 0 + for chunk in response.iter_content(chunk_size=64 * 1024): + received += len(chunk) + if received > self.MAX_IMAGE_FETCH_BYTES: + raise ValueError( + f"Image exceeds the {self.MAX_IMAGE_FETCH_BYTES // (1024 * 1024)} MB limit" + ) + chunks.append(chunk) + content = b"".join(chunks) + + with Image.open(io.BytesIO(content)) as img: img_format = img.format or "jpeg" # Default to jpeg if format is None mime_type = f"image/{img_format.lower()}" - - base64_data = base64.b64encode(response.content).decode("utf-8") + + base64_data = base64.b64encode(content).decode("utf-8") return mime_type, base64_data - + except Exception as ex: raise ValueError(f"Failed to fetch image from {data}: {ex}") from ex @@ -1668,7 +1711,11 @@ def _invoke_error_mapping(self) -> dict[type[InvokeError], list[type[Exception]] anthropic.APIConnectionError, anthropic.APITimeoutError, ], - InvokeServerUnavailableError: [anthropic.InternalServerError], + InvokeServerUnavailableError: [ + anthropic.InternalServerError, + # 529 Overloaded is a server-side capacity error, not a client error. + anthropic.OverloadedError, + ], InvokeRateLimitError: [anthropic.RateLimitError], InvokeAuthorizationError: [ anthropic.AuthenticationError, diff --git a/models/anthropic/tests/test_stream_and_hardening.py b/models/anthropic/tests/test_stream_and_hardening.py new file mode 100644 index 000000000..c71de6652 --- /dev/null +++ b/models/anthropic/tests/test_stream_and_hardening.py @@ -0,0 +1,454 @@ +"""Regression tests for stream state machine, error mapping, and image fetching. + +- The streaming handler used to emit the closing think tag a second time when a + thinking block was followed directly by a tool_use block (current_block_type + was left as "thinking" on input_json_delta and re-closed at MessageStopEvent). +- HTTP 529 OverloadedError fell through the APIError catch-all and was mapped + to InvokeBadRequestError instead of InvokeServerUnavailableError. +- Image URL fetching had no timeout, no size cap, no scheme restriction and + followed redirects. +""" + +import base64 +import io + +import anthropic +import httpx +import pytest +from anthropic.types import ( + ContentBlockDeltaEvent, + ContentBlockStartEvent, + InputJSONDelta, + Message, + MessageDeltaEvent, + MessageStartEvent, + MessageStopEvent, + TextBlock, + TextDelta, + ThinkingBlock, + ThinkingDelta, + ToolUseBlock, + Usage, +) +from anthropic.types.raw_message_delta_event import Delta as MessageDelta +from dify_plugin.entities.model.message import UserPromptMessage +from dify_plugin.errors.model import ( + InvokeRateLimitError, + InvokeServerUnavailableError, +) +from models.llm import llm as llm_module +from models.llm.llm import AnthropicLargeLanguageModel +from PIL import Image + +OPEN_THINK = chr(60) + "think" + chr(62) +CLOSE_THINK = chr(60) + "/think" + chr(62) + + +def _message_start(input_tokens: int = 10) -> MessageStartEvent: + message = Message( + id="msg_1", + content=[], + model="claude-sonnet-4-5", + role="assistant", + type="message", + usage=Usage(input_tokens=input_tokens, output_tokens=1), + ) + return MessageStartEvent(type="message_start", message=message) + + +def _message_delta(stop_reason: str, output_tokens: int = 5) -> MessageDeltaEvent: + return MessageDeltaEvent( + type="message_delta", + delta=MessageDelta(stop_reason=stop_reason), + usage={"output_tokens": output_tokens}, + ) + + +def _run_stream(events: list) -> list: + return list( + AnthropicLargeLanguageModel()._handle_chat_generate_stream_response( + "claude-sonnet-4-5", + {"anthropic_api_key": "test-key"}, + events, + [UserPromptMessage(content="hi")], + ) + ) + + +def _emitted_text(chunks: list) -> str: + return "".join( + str(chunk.delta.message.content) for chunk in chunks if chunk.delta.message.content + ) + + +def test_thinking_then_tool_use_closes_think_tag_once() -> None: + events = [ + _message_start(), + ContentBlockStartEvent( + type="content_block_start", + index=0, + content_block=ThinkingBlock(thinking="", signature="", type="thinking"), + ), + ContentBlockDeltaEvent( + type="content_block_delta", + index=0, + delta=ThinkingDelta(type="thinking_delta", thinking="hmm"), + ), + ContentBlockStartEvent( + type="content_block_start", + index=1, + content_block=ToolUseBlock( + id="toolu_1", input={}, name="get_weather", type="tool_use" + ), + ), + ContentBlockDeltaEvent( + type="content_block_delta", + index=1, + delta=InputJSONDelta(type="input_json_delta", partial_json='{"city": "Lisbon"}'), + ), + _message_delta("tool_use"), + MessageStopEvent(type="message_stop"), + ] + + chunks = _run_stream(events) + text = _emitted_text(chunks) + + assert text.count(OPEN_THINK) == 1 + assert text.count(CLOSE_THINK) == 1 + assert "hmm" in text + + last = chunks[-1] + assert last.delta.finish_reason == "tool_use" + tool_calls = last.delta.message.tool_calls or [] + assert len(tool_calls) == 1 + assert tool_calls[0].function.name == "get_weather" + assert tool_calls[0].function.arguments == '{"city": "Lisbon"}' + + +def test_thinking_then_text_closes_think_tag_once() -> None: + events = [ + _message_start(), + ContentBlockStartEvent( + type="content_block_start", + index=0, + content_block=ThinkingBlock(thinking="", signature="", type="thinking"), + ), + ContentBlockDeltaEvent( + type="content_block_delta", + index=0, + delta=ThinkingDelta(type="thinking_delta", thinking="hmm"), + ), + ContentBlockStartEvent( + type="content_block_start", + index=1, + content_block=TextBlock(text="", type="text"), + ), + ContentBlockDeltaEvent( + type="content_block_delta", + index=1, + delta=TextDelta(type="text_delta", text="hi there"), + ), + _message_delta("end_turn"), + MessageStopEvent(type="message_stop"), + ] + + chunks = _run_stream(events) + text = _emitted_text(chunks) + + assert text.count(OPEN_THINK) == 1 + assert text.count(CLOSE_THINK) == 1 + assert "hi there" in text + + +def test_thinking_then_stop_closes_think_tag_once() -> None: + events = [ + _message_start(), + ContentBlockStartEvent( + type="content_block_start", + index=0, + content_block=ThinkingBlock(thinking="", signature="", type="thinking"), + ), + ContentBlockDeltaEvent( + type="content_block_delta", + index=0, + delta=ThinkingDelta(type="thinking_delta", thinking="hmm"), + ), + _message_delta("max_tokens"), + MessageStopEvent(type="message_stop"), + ] + + chunks = _run_stream(events) + text = _emitted_text(chunks) + + assert text.count(OPEN_THINK) == 1 + assert text.count(CLOSE_THINK) == 1 + + +def test_text_only_never_emits_think_tags() -> None: + events = [ + _message_start(), + ContentBlockStartEvent( + type="content_block_start", + index=0, + content_block=TextBlock(text="", type="text"), + ), + ContentBlockDeltaEvent( + type="content_block_delta", + index=0, + delta=TextDelta(type="text_delta", text="plain"), + ), + _message_delta("end_turn"), + MessageStopEvent(type="message_stop"), + ] + + chunks = _run_stream(events) + text = _emitted_text(chunks) + + assert OPEN_THINK not in text + assert CLOSE_THINK not in text + assert "plain" in text + + +def test_thinking_then_parallel_tool_calls_closes_think_tag_once() -> None: + events = [ + _message_start(), + ContentBlockStartEvent( + type="content_block_start", + index=0, + content_block=ThinkingBlock(thinking="", signature="", type="thinking"), + ), + ContentBlockDeltaEvent( + type="content_block_delta", + index=0, + delta=ThinkingDelta(type="thinking_delta", thinking="hmm"), + ), + ContentBlockStartEvent( + type="content_block_start", + index=1, + content_block=ToolUseBlock( + id="toolu_1", input={}, name="get_weather", type="tool_use" + ), + ), + ContentBlockDeltaEvent( + type="content_block_delta", + index=1, + delta=InputJSONDelta(type="input_json_delta", partial_json='{"city": "Lisbon"}'), + ), + ContentBlockStartEvent( + type="content_block_start", + index=2, + content_block=ToolUseBlock( + id="toolu_2", input={}, name="get_time", type="tool_use" + ), + ), + ContentBlockDeltaEvent( + type="content_block_delta", + index=2, + delta=InputJSONDelta(type="input_json_delta", partial_json='{"zone": "UTC"}'), + ), + _message_delta("tool_use"), + MessageStopEvent(type="message_stop"), + ] + + chunks = _run_stream(events) + text = _emitted_text(chunks) + + assert text.count(OPEN_THINK) == 1 + assert text.count(CLOSE_THINK) == 1 + + tool_calls = chunks[-1].delta.message.tool_calls or [] + assert [(tc.function.name, tc.function.arguments) for tc in tool_calls] == [ + ("get_weather", '{"city": "Lisbon"}'), + ("get_time", '{"zone": "UTC"}'), + ] + + +def test_thinking_then_empty_tool_input_closes_think_tag_once() -> None: + # A tool_use block that never emits an input_json_delta (empty input): + # no block transition occurs, so the close tag must come at stream end — once. + events = [ + _message_start(), + ContentBlockStartEvent( + type="content_block_start", + index=0, + content_block=ThinkingBlock(thinking="", signature="", type="thinking"), + ), + ContentBlockDeltaEvent( + type="content_block_delta", + index=0, + delta=ThinkingDelta(type="thinking_delta", thinking="hmm"), + ), + ContentBlockStartEvent( + type="content_block_start", + index=1, + content_block=ToolUseBlock(id="toolu_1", input={}, name="noop", type="tool_use"), + ), + _message_delta("tool_use"), + MessageStopEvent(type="message_stop"), + ] + + chunks = _run_stream(events) + text = _emitted_text(chunks) + + assert text.count(OPEN_THINK) == 1 + assert text.count(CLOSE_THINK) == 1 + + tool_calls = chunks[-1].delta.message.tool_calls or [] + assert len(tool_calls) == 1 + assert tool_calls[0].function.name == "noop" + assert tool_calls[0].function.arguments == "{}" + + +def _status_error(cls: type, status_code: int): + response = httpx.Response( + status_code, + request=httpx.Request("POST", "https://api.anthropic.com/v1/messages"), + ) + return cls(message="err", response=response, body=None) + + +def test_529_overloaded_maps_to_server_unavailable() -> None: + llm = AnthropicLargeLanguageModel() + transformed = llm._transform_invoke_error(_status_error(anthropic.OverloadedError, 529)) + assert isinstance(transformed, InvokeServerUnavailableError) + + +def test_500_and_429_mapping_unchanged() -> None: + llm = AnthropicLargeLanguageModel() + assert isinstance( + llm._transform_invoke_error(_status_error(anthropic.InternalServerError, 500)), + InvokeServerUnavailableError, + ) + assert isinstance( + llm._transform_invoke_error(_status_error(anthropic.RateLimitError, 429)), + InvokeRateLimitError, + ) + + +def _png_bytes() -> bytes: + buf = io.BytesIO() + Image.new("RGB", (1, 1), color="white").save(buf, format="PNG") + return buf.getvalue() + + +class _FakeResponse: + def __init__(self, content: bytes, status_code: int = 200) -> None: + self.content = content + self.status_code = status_code + self.headers = {"Content-Length": str(len(content))} + + def raise_for_status(self) -> None: + if self.status_code >= 400: + raise httpx.HTTPStatusError( + f"HTTP {self.status_code}", + request=httpx.Request("GET", "http://x/img"), + response=httpx.Response(self.status_code), + ) + + def iter_content(self, chunk_size: int = 65536): + for i in range(0, len(self.content), chunk_size): + yield self.content[i : i + chunk_size] + + def close(self) -> None: + pass + + +def test_image_fetch_uses_timeout_and_no_redirects(monkeypatch) -> None: + captured: dict = {} + + def fake_get(url, **kwargs): + captured["url"] = url + captured["kwargs"] = kwargs + return _FakeResponse(_png_bytes()) + + monkeypatch.setattr(llm_module.requests, "get", fake_get) + + mime, b64 = AnthropicLargeLanguageModel()._process_image_data("http://x/img.png") + + assert mime == "image/png" + assert base64.b64decode(b64) == _png_bytes() + assert captured["kwargs"]["timeout"] == (5.0, 15.0) + assert captured["kwargs"]["allow_redirects"] is False + assert captured["kwargs"]["stream"] is True + + +def test_image_fetch_strips_surrounding_whitespace(monkeypatch) -> None: + captured: dict = {} + + def fake_get(url, **kwargs): + captured["url"] = url + return _FakeResponse(_png_bytes()) + + monkeypatch.setattr(llm_module.requests, "get", fake_get) + + AnthropicLargeLanguageModel()._process_image_data(" http://x/img.png ") + + assert captured["url"] == "http://x/img.png" + + +def test_image_fetch_rejects_non_http_schemes(monkeypatch) -> None: + def fail_get(*args, **kwargs): + raise AssertionError("requests.get must not be called for non-http(s) URLs") + + monkeypatch.setattr(llm_module.requests, "get", fail_get) + + with pytest.raises(ValueError, match="only http\\(s\\) URLs are allowed"): + AnthropicLargeLanguageModel()._process_image_data("file:///etc/passwd") + + +def test_image_fetch_rejects_redirects(monkeypatch) -> None: + monkeypatch.setattr( + llm_module.requests, "get", lambda url, **kw: _FakeResponse(b"", status_code=302) + ) + + with pytest.raises(ValueError, match="Redirect responses are not allowed"): + AnthropicLargeLanguageModel()._process_image_data("http://x/img.png") + + +def test_image_fetch_rejects_redirect_even_with_valid_body(monkeypatch) -> None: + # A 3xx with an image-like body must still be rejected, not consumed. + monkeypatch.setattr( + llm_module.requests, + "get", + lambda url, **kw: _FakeResponse(_png_bytes(), status_code=302), + ) + + with pytest.raises(ValueError, match="Redirect responses are not allowed"): + AnthropicLargeLanguageModel()._process_image_data("http://x/img.png") + + +def test_image_fetch_rejects_oversized_content(monkeypatch) -> None: + too_big = b"x" * (AnthropicLargeLanguageModel.MAX_IMAGE_FETCH_BYTES + 1) + monkeypatch.setattr(llm_module.requests, "get", lambda url, **kw: _FakeResponse(too_big)) + + with pytest.raises(ValueError, match="exceeds the 10 MB limit"): + AnthropicLargeLanguageModel()._process_image_data("http://x/img.png") + + +def test_image_fetch_rejects_oversized_content_length_header(monkeypatch) -> None: + # The declared Content-Length alone must trip the cap before downloading. + response = _FakeResponse(b"") + response.headers = {"Content-Length": str(AnthropicLargeLanguageModel.MAX_IMAGE_FETCH_BYTES + 1)} + monkeypatch.setattr(llm_module.requests, "get", lambda url, **kw: response) + + with pytest.raises(ValueError, match="exceeds the 10 MB limit"): + AnthropicLargeLanguageModel()._process_image_data("http://x/img.png") + + +def test_image_fetch_rejects_non_image_content(monkeypatch) -> None: + monkeypatch.setattr( + llm_module.requests, "get", lambda url, **kw: _FakeResponse(b"not an image") + ) + + with pytest.raises(ValueError, match="Failed to fetch image"): + AnthropicLargeLanguageModel()._process_image_data("http://x/img.png") + + +def test_data_uri_still_works() -> None: + png = _png_bytes() + data_uri = "data:image/png;base64," + base64.b64encode(png).decode("utf-8") + + mime, b64 = AnthropicLargeLanguageModel()._process_image_data(data_uri) + + assert mime == "image/png" + assert base64.b64decode(b64) == png From a7eee8db36b74760b754c4766c065b93fb6a3244 Mon Sep 17 00:00:00 2001 From: hpiclaranet Date: Sat, 22 Aug 2026 00:56:59 +0100 Subject: [PATCH 2/3] fix(anthropic): close streamed image responses on all exit paths Wrap the streamed fetch in the requests context manager so the connection is released deterministically on early rejection (3xx / Content-Length), mid-stream cap, error responses and success. Also cover the chunked (no Content-Length) over-cap path with a deterministic test that asserts iteration stops right after the cap, and make the redirect test assert the body is never read. Review findings from the second-opinion validation round of this PR. --- models/anthropic/models/llm/llm.py | 48 ++++++++++--------- .../tests/test_stream_and_hardening.py | 45 ++++++++++++++--- 2 files changed, 64 insertions(+), 29 deletions(-) diff --git a/models/anthropic/models/llm/llm.py b/models/anthropic/models/llm/llm.py index 7279f8bd7..06d2c2fef 100644 --- a/models/anthropic/models/llm/llm.py +++ b/models/anthropic/models/llm/llm.py @@ -1456,37 +1456,39 @@ def _process_image_data(self, data: str) -> tuple[str, str]: # and no redirect following (the fetched host is exactly the one given). # stream=True + iter_content keeps memory bounded: a hostile server # cannot push more than ~MAX_IMAGE_FETCH_BYTES into the process. - response = requests.get( + # The context manager guarantees the connection is released on every + # exit path (early rejection, mid-stream cap, success). + with requests.get( url, timeout=(5.0, 15.0), allow_redirects=False, stream=True, - ) - response.raise_for_status() - # raise_for_status only raises for >= 400: reject 3xx explicitly. - if 300 <= response.status_code < 400: - raise ValueError("Redirect responses are not allowed") - - length = int(response.headers.get("Content-Length") or 0) - if length > self.MAX_IMAGE_FETCH_BYTES: - raise ValueError( - f"Image exceeds the {self.MAX_IMAGE_FETCH_BYTES // (1024 * 1024)} MB limit" - ) - - chunks: list[bytes] = [] - received = 0 - for chunk in response.iter_content(chunk_size=64 * 1024): - received += len(chunk) - if received > self.MAX_IMAGE_FETCH_BYTES: + ) as response: + response.raise_for_status() + # raise_for_status only raises for >= 400: reject 3xx explicitly. + if 300 <= response.status_code < 400: + raise ValueError("Redirect responses are not allowed") + + length = int(response.headers.get("Content-Length") or 0) + if length > self.MAX_IMAGE_FETCH_BYTES: raise ValueError( f"Image exceeds the {self.MAX_IMAGE_FETCH_BYTES // (1024 * 1024)} MB limit" ) - chunks.append(chunk) - content = b"".join(chunks) - with Image.open(io.BytesIO(content)) as img: - img_format = img.format or "jpeg" # Default to jpeg if format is None - mime_type = f"image/{img_format.lower()}" + chunks: list[bytes] = [] + received = 0 + for chunk in response.iter_content(chunk_size=64 * 1024): + received += len(chunk) + if received > self.MAX_IMAGE_FETCH_BYTES: + raise ValueError( + f"Image exceeds the {self.MAX_IMAGE_FETCH_BYTES // (1024 * 1024)} MB limit" + ) + chunks.append(chunk) + content = b"".join(chunks) + + with Image.open(io.BytesIO(content)) as img: + img_format = img.format or "jpeg" # Default to jpeg if format is None + mime_type = f"image/{img_format.lower()}" base64_data = base64.b64encode(content).decode("utf-8") return mime_type, base64_data diff --git a/models/anthropic/tests/test_stream_and_hardening.py b/models/anthropic/tests/test_stream_and_hardening.py index c71de6652..598d4fd76 100644 --- a/models/anthropic/tests/test_stream_and_hardening.py +++ b/models/anthropic/tests/test_stream_and_hardening.py @@ -352,6 +352,12 @@ def iter_content(self, chunk_size: int = 65536): def close(self) -> None: pass + def __enter__(self) -> "_FakeResponse": + return self + + def __exit__(self, *exc: object) -> None: + self.close() + def test_image_fetch_uses_timeout_and_no_redirects(monkeypatch) -> None: captured: dict = {} @@ -406,12 +412,14 @@ def test_image_fetch_rejects_redirects(monkeypatch) -> None: def test_image_fetch_rejects_redirect_even_with_valid_body(monkeypatch) -> None: - # A 3xx with an image-like body must still be rejected, not consumed. - monkeypatch.setattr( - llm_module.requests, - "get", - lambda url, **kw: _FakeResponse(_png_bytes(), status_code=302), - ) + # A 3xx with an image-like body must still be rejected before any body read. + response = _FakeResponse(_png_bytes(), status_code=302) + + def no_iter(*args, **kwargs): + raise AssertionError("iter_content must not be called for a 3xx response") + + response.iter_content = no_iter + monkeypatch.setattr(llm_module.requests, "get", lambda url, **kw: response) with pytest.raises(ValueError, match="Redirect responses are not allowed"): AnthropicLargeLanguageModel()._process_image_data("http://x/img.png") @@ -435,6 +443,31 @@ def test_image_fetch_rejects_oversized_content_length_header(monkeypatch) -> Non AnthropicLargeLanguageModel()._process_image_data("http://x/img.png") +def test_image_fetch_rejects_chunked_body_over_cap(monkeypatch) -> None: + # No Content-Length (chunked transfer): the cap must trip on the bounded + # iter_content accumulation, and iteration must stop right after the cap. + too_big = b"x" * (AnthropicLargeLanguageModel.MAX_IMAGE_FETCH_BYTES + 1) + response = _FakeResponse(too_big) + response.headers = {} + read_count = {"n": 0} + original_iter = response.iter_content + + def counting_iter(chunk_size: int = 65536): + for chunk in original_iter(chunk_size): + read_count["n"] += 1 + yield chunk + + response.iter_content = counting_iter + monkeypatch.setattr(llm_module.requests, "get", lambda url, **kw: response) + + with pytest.raises(ValueError, match="exceeds the 10 MB limit"): + AnthropicLargeLanguageModel()._process_image_data("http://x/img.png") + + # 10 MB / 64 KB = 160 full chunks; the 161st trips the cap. It must not + # drain the rest of the body. + assert read_count["n"] <= 162 + + def test_image_fetch_rejects_non_image_content(monkeypatch) -> None: monkeypatch.setattr( llm_module.requests, "get", lambda url, **kw: _FakeResponse(b"not an image") From 8cf658d77a7e1320ff49273973926e2ebb8f4df0 Mon Sep 17 00:00:00 2001 From: hpiclaranet Date: Sat, 22 Aug 2026 01:01:58 +0100 Subject: [PATCH 3/3] test(anthropic): make the chunked over-cap test discriminating The chunked-body test used a body of cap+1 bytes (161 chunks), so a regression that drained the whole stream would have read the same chunk count as a correct early abort. Use a 2x-cap body (320 chunks): a correct abort reads 161, a full drain reads 320 and fails the bound. Test-rigor finding from the second-opinion validation round of this PR. --- models/anthropic/tests/test_stream_and_hardening.py | 10 +++++++--- 1 file changed, 7 insertions(+), 3 deletions(-) diff --git a/models/anthropic/tests/test_stream_and_hardening.py b/models/anthropic/tests/test_stream_and_hardening.py index 598d4fd76..01c439fd8 100644 --- a/models/anthropic/tests/test_stream_and_hardening.py +++ b/models/anthropic/tests/test_stream_and_hardening.py @@ -446,7 +446,10 @@ def test_image_fetch_rejects_oversized_content_length_header(monkeypatch) -> Non def test_image_fetch_rejects_chunked_body_over_cap(monkeypatch) -> None: # No Content-Length (chunked transfer): the cap must trip on the bounded # iter_content accumulation, and iteration must stop right after the cap. - too_big = b"x" * (AnthropicLargeLanguageModel.MAX_IMAGE_FETCH_BYTES + 1) + # The body is 2x the cap on purpose: with a cap+1 body a "drain the whole + # stream" regression would read the same chunk count as a correct early + # abort, so it would not be caught by the read-count assertion. + too_big = b"x" * (2 * AnthropicLargeLanguageModel.MAX_IMAGE_FETCH_BYTES) response = _FakeResponse(too_big) response.headers = {} read_count = {"n": 0} @@ -463,8 +466,9 @@ def counting_iter(chunk_size: int = 65536): with pytest.raises(ValueError, match="exceeds the 10 MB limit"): AnthropicLargeLanguageModel()._process_image_data("http://x/img.png") - # 10 MB / 64 KB = 160 full chunks; the 161st trips the cap. It must not - # drain the rest of the body. + # 10 MB / 64 KB = 160 full chunks; the 161st trips the cap, so a correct + # early abort reads exactly 161 chunks. A regression that drains the + # stream would read all 320 and fail this bound. assert read_count["n"] <= 162