mirror of
https://github.com/BerriAI/litellm.git
synced 2026-10-09 03:18:44 +00:00
fix(guardrails): classify content parts by declared type before reading text
A part tagged `type: "image_url"` that also carries a `text` key was scanned as text and returned before the image branch ran, while the provider transformations branch on `type` and send it to the model as an image. That let a caller defeat an IMAGE-modality guardrail, and defeat on_unscannable_image, by pairing the image with a benign decoy string. Classify by the declared type first, so an image_url part always takes the image path regardless of what other fields it carries. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
This commit is contained in:
parent
93f2ae537d
commit
4c98cf95b5
2 changed files with 52 additions and 4 deletions
|
|
@ -381,13 +381,20 @@ class BedrockGuardrail(CustomGuardrail, BaseAWSLLM):
|
|||
if not isinstance(item, dict):
|
||||
return None
|
||||
part: Final = cast(Mapping[str, object], item) # cast-ok: narrowed to dict on the line above
|
||||
# Classify by the declared type before reading any field. Provider
|
||||
# transformations branch on `type`, so a part tagged image_url reaches the
|
||||
# model as an image even when it also carries a `text` key. Reading `text`
|
||||
# first would scan that decoy and forward the image unscanned, which is the
|
||||
# bypass this whole extractor exists to close.
|
||||
if part.get("type") == "image_url":
|
||||
image_url: Final = self._get_image_url(item=part)
|
||||
if image_url is None:
|
||||
return None
|
||||
return await self._build_image_content_item(image_url=image_url)
|
||||
text: Final = part.get("text")
|
||||
if isinstance(text, str):
|
||||
return BedrockContentItem(text=BedrockTextContent(text=text))
|
||||
image_url: Final = self._get_image_url(item=part)
|
||||
if image_url is None:
|
||||
return None
|
||||
return await self._build_image_content_item(image_url=image_url)
|
||||
return None
|
||||
|
||||
@staticmethod
|
||||
def _get_image_url(item: Mapping[str, object]) -> str | None:
|
||||
|
|
|
|||
|
|
@ -5323,6 +5323,47 @@ class TestBedrockGuardrailImageInput:
|
|||
{"image": {"format": "png", "source": {"bytes": self._PNG_DATA_URI.split(",")[1]}}},
|
||||
]
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_image_part_carrying_a_text_field_is_still_scanned_as_an_image(self):
|
||||
"""A part tagged image_url reaches the model as an image, text field or not.
|
||||
|
||||
Provider transformations branch on `type`, so reading `text` first would scan
|
||||
the decoy and forward the image unscanned - the exact bypass this change closes.
|
||||
"""
|
||||
messages = [
|
||||
{
|
||||
"role": "user",
|
||||
"content": [
|
||||
{
|
||||
"type": "image_url",
|
||||
"image_url": {"url": self._PNG_DATA_URI},
|
||||
"text": "just a friendly note",
|
||||
}
|
||||
],
|
||||
}
|
||||
]
|
||||
|
||||
request = await self._guardrail().convert_to_bedrock_format(source="INPUT", messages=messages)
|
||||
|
||||
assert request["content"] == [
|
||||
{"image": {"format": "png", "source": {"bytes": self._PNG_DATA_URI.split(",")[1]}}}
|
||||
]
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_unscannable_image_part_carrying_a_text_field_still_blocks(self):
|
||||
"""The decoy text must not turn an unscannable image into a scanned request."""
|
||||
messages = [
|
||||
{
|
||||
"role": "user",
|
||||
"content": [{"type": "image_url", "image_url": {"url": self._GIF_DATA_URI}, "text": "hello"}],
|
||||
}
|
||||
]
|
||||
|
||||
with pytest.raises(HTTPException) as exc_info:
|
||||
await self._guardrail().convert_to_bedrock_format(source="INPUT", messages=messages)
|
||||
|
||||
assert exc_info.value.status_code == 400
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_unscannable_image_blocks_the_request_by_default(self):
|
||||
"""ApplyGuardrail takes png/jpeg only, and the image reaches the model either way.
|
||||
|
|
|
|||
Loading…
Add table
Reference in a new issue