fix(guardrails): align banned_keywords + azure_content_safety call_type gates with runtime route_type

The hooks gated on ``call_type == "completion"`` but the proxy ingress
passes ``route_type`` straight through as ``call_type`` —
``"acompletion"`` for /v1/chat/completions and ``"aresponses"`` for
/v1/responses. Tests passed because they used the literal sync
``"completion"`` value, masking the gap.

Switch both hooks to ``is_text_content_call_type`` (matches the
canonical runtime values: completion / acompletion / aresponses) and
update existing tests to assert against runtime values, plus parametrize
a regression test that pins the gate.
This commit is contained in:
user 2026-05-04 21:27:24 +00:00
parent f4e7dde2d8
commit abbefccad4
No known key found for this signature in database
4 changed files with 144 additions and 10 deletions

View file

@ -11,7 +11,10 @@ from typing import Literal
import litellm
from litellm.caching.caching import DualCache
from litellm.proxy._types import UserAPIKeyAuth
from litellm.proxy.guardrails._content_utils import iter_message_text
from litellm.proxy.guardrails._content_utils import (
is_text_content_call_type,
iter_message_text,
)
from litellm.integrations.custom_logger import CustomLogger
from litellm._logging import verbose_proxy_logger
from fastapi import HTTPException
@ -74,8 +77,7 @@ class _ENTERPRISE_BannedKeywords(CustomLogger):
- check if user id part of blocked list
"""
self.print_verbose("Inside Banned Keyword List Pre-Call Hook")
if call_type == "completion":
# Covers multimodal list content + Responses-API input.
if is_text_content_call_type(call_type):
for text in iter_message_text(data):
self.test_violation(test_str=text)

View file

@ -8,7 +8,31 @@ skip the other shapes — these helpers normalise that so every hook sees
every text fragment.
"""
from typing import Any, Callable, Dict, Iterator, List
from typing import Any, Callable, Dict, FrozenSet, Iterator, List
# Call types whose body carries free-form chat / prompt text that
# text-content guardrails (banned keywords, content moderation, secret
# detection, …) should inspect. The proxy ingress passes ``route_type``
# straight through as ``call_type``, so the literal values here are
# what the guardrail dispatcher actually receives:
#
# /v1/chat/completions -> "acompletion"
# /v1/responses -> "aresponses"
#
# ``"completion"`` is included for SDK / internal callers that invoke
# ``pre_call_hook`` directly with the sync name. Embedding, moderation,
# audio, and transcription endpoints are deliberately excluded — text
# guardrails on those paths are a separate scope.
TEXT_CONTENT_CALL_TYPES: FrozenSet[str] = frozenset(
{"completion", "acompletion", "aresponses"}
)
def is_text_content_call_type(call_type: str) -> bool:
"""Return True if ``call_type`` carries free-form text that text
guardrails should inspect (Chat Completions or Responses API)."""
return call_type in TEXT_CONTENT_CALL_TYPES
def _iter_text_parts_in_content(content: Any) -> Iterator[str]:

View file

@ -8,7 +8,10 @@ from litellm._logging import verbose_proxy_logger
from litellm.caching.caching import DualCache
from litellm.integrations.custom_logger import CustomLogger
from litellm.proxy._types import UserAPIKeyAuth
from litellm.proxy.guardrails._content_utils import iter_message_text
from litellm.proxy.guardrails._content_utils import (
is_text_content_call_type,
iter_message_text,
)
class _PROXY_AzureContentSafety(
@ -119,8 +122,7 @@ class _PROXY_AzureContentSafety(
):
verbose_proxy_logger.debug("Inside Azure Content-Safety Pre-Call Hook")
try:
if call_type == "completion":
# Covers multimodal list content + Responses-API input.
if is_text_content_call_type(call_type):
for text in iter_message_text(data):
await self.test_violation(content=text, source="input")

View file

@ -433,7 +433,13 @@ async def test_lasso_masking_writes_back_responses_api_input(user_api_key, monke
def test_banned_keywords_blocks_multimodal_content(monkeypatch):
"""VERIA-11: a banned word hidden in a multimodal text part is now caught."""
"""VERIA-11: a banned word hidden in a multimodal text part is now caught.
Uses ``acompletion`` the value the proxy ingress actually passes
for ``/v1/chat/completions``. Asserting against the literal sync
``"completion"`` would pass even if the hook's call-type gate were
misaligned with the runtime, so the test wouldn't catch regressions.
"""
monkeypatch.setattr("litellm.banned_keywords_list", ["forbidden"], raising=False)
from enterprise.enterprise_hooks.banned_keywords import _ENTERPRISE_BannedKeywords
from fastapi import HTTPException
@ -455,7 +461,7 @@ def test_banned_keywords_blocks_multimodal_content(monkeypatch):
}
]
},
call_type="completion",
call_type="acompletion",
)
import asyncio
@ -477,7 +483,7 @@ def test_banned_keywords_blocks_responses_api_input(monkeypatch):
user_api_key_dict=UserAPIKeyAuth(api_key="hashed", user_id="u"),
cache=DualCache(),
data={"input": "this contains forbidden content"},
call_type="completion",
call_type="aresponses",
)
import asyncio
@ -486,6 +492,59 @@ def test_banned_keywords_blocks_responses_api_input(monkeypatch):
asyncio.run(_run())
@pytest.mark.parametrize("call_type", ["completion", "acompletion", "aresponses"])
def test_banned_keywords_fires_on_text_content_call_types(monkeypatch, call_type):
"""Locks the call-type gate to the runtime ``route_type`` values the
proxy actually emits pinning a regression where the hook had
``call_type == "completion"`` and silently no-op'd both
``acompletion`` (chat completions) and ``aresponses`` (Responses API).
"""
monkeypatch.setattr("litellm.banned_keywords_list", ["forbidden"], raising=False)
from enterprise.enterprise_hooks.banned_keywords import _ENTERPRISE_BannedKeywords
from fastapi import HTTPException
guard = _ENTERPRISE_BannedKeywords()
import asyncio
with pytest.raises(HTTPException):
asyncio.run(
guard.async_pre_call_hook(
user_api_key_dict=UserAPIKeyAuth(api_key="hashed", user_id="u"),
cache=DualCache(),
data={
"messages": [{"role": "user", "content": "forbidden text"}],
"input": "forbidden text",
},
call_type=call_type,
)
)
def test_banned_keywords_skips_non_text_call_types(monkeypatch):
"""Embedding / moderation / audio paths don't carry chat text and
aren't in the text-guardrail scope. They must not trigger the hook
even when the request body otherwise looks like a chat payload.
"""
monkeypatch.setattr("litellm.banned_keywords_list", ["forbidden"], raising=False)
from enterprise.enterprise_hooks.banned_keywords import _ENTERPRISE_BannedKeywords
guard = _ENTERPRISE_BannedKeywords()
import asyncio
for call_type in ("aembedding", "amoderation", "aspeech", "atranscription"):
# Should return without raising, even though the data carries the banned word.
asyncio.run(
guard.async_pre_call_hook(
user_api_key_dict=UserAPIKeyAuth(api_key="hashed", user_id="u"),
cache=DualCache(),
data={"input": "forbidden text"},
call_type=call_type,
)
)
@pytest.mark.asyncio
async def test_banned_keywords_post_call_checks_all_choices(monkeypatch, user_api_key):
"""Krrish blocker: ``n>1`` responses must not bypass post-call checks by
@ -515,6 +574,53 @@ async def test_banned_keywords_post_call_checks_all_choices(monkeypatch, user_ap
# ── Azure Content Safety ──────────────────────────────────────────────────────
@pytest.mark.asyncio
@pytest.mark.parametrize(
"call_type, data",
[
(
"acompletion",
{
"messages": [
{
"role": "user",
"content": [
{"type": "text", "text": "scan me"},
{"type": "image_url", "image_url": {"url": "..."}},
],
}
]
},
),
("aresponses", {"input": "scan me"}),
],
)
async def test_azure_content_safety_pre_call_fires_on_runtime_call_types(
user_api_key, call_type, data
):
"""The proxy ingress passes ``route_type`` straight through as
``call_type`` ``acompletion`` for chat completions and
``aresponses`` for the Responses API. The hook must inspect text
fragments under both, not only the literal ``"completion"`` string
used by some SDK callers."""
from litellm.proxy.hooks.azure_content_safety import _PROXY_AzureContentSafety
guard = _PROXY_AzureContentSafety.__new__(_PROXY_AzureContentSafety)
seen = []
async def fake_test_violation(content, source=None):
seen.append((content, source))
guard.test_violation = fake_test_violation
await guard.async_pre_call_hook(
user_api_key_dict=user_api_key,
cache=DualCache(),
data=data,
call_type=call_type,
)
assert ("scan me", "input") in seen
@pytest.mark.asyncio
async def test_azure_content_safety_post_call_checks_all_choices(user_api_key):
"""Krrish blocker: ``n>1`` responses must not bypass Azure Content Safety