mirror of
https://github.com/BerriAI/litellm.git
synced 2026-10-17 04:34:25 +00:00
fix(guardrails): record every content filter match in logging_only and offer the mode in the UI (#46673)
* test(guardrails): pin content filter logging_only matches and UI modes in an integration test * fix(guardrails): record every content filter match in logging_only and offer the mode in the UI * style(tests): format the content filter logging_only integration test * refactor(guardrails): keep the logging_only predicate and scan override within the type gates * fix(guardrails): flag tool arguments a content filter would reject in logging_only and keep scanning * test(integration): expect guardrail_flagged for an observe-only content filter block in the scope config test * test(integration): pin logging_only flags for tool arguments an enforcing content filter rejects * refactor(guardrails): drop the content filter flagged branch no logging_only scan can reach * test(integration): pin OTel spans, Guardrails Monitor and a provider outage burst for a logging_only content filter * test(integration): pin MCP tool call verdicts and fail the Monitor cell when usage never rolls up * test(integration): pin category and competitor intent verdicts and the provider failure row in logging_only --------- Co-authored-by: yucheng-berri <293926326+yucheng-berri@users.noreply.github.com>
This commit is contained in:
yucheng-berri
GitHub
parent
a3251a3bfd
commit
f281f77bc0
8 changed files with 1071 additions and 35 deletions
|
|
@ -695,9 +695,7 @@ class CustomGuardrail(CustomLogger):
|
|||
supported_event_hooks: list[GuardrailEventHooks],
|
||||
) -> None:
|
||||
allowed_hooks: Final = frozenset(supported_event_hooks) | (
|
||||
frozenset((GuardrailEventHooks.logging_only,))
|
||||
if self.uses_apply_guardrail_interface() and not self.use_native_lifecycle_hooks
|
||||
else frozenset()
|
||||
frozenset((GuardrailEventHooks.logging_only,)) if self.accepts_logging_only_mode() else frozenset()
|
||||
)
|
||||
|
||||
def _validate_event_hook_list_is_in_supported_event_hooks(
|
||||
|
|
@ -869,8 +867,13 @@ class CustomGuardrail(CustomLogger):
|
|||
return True
|
||||
return False
|
||||
|
||||
def uses_apply_guardrail_interface(self) -> bool:
|
||||
return type(self).apply_guardrail is not CustomGuardrail.apply_guardrail
|
||||
@classmethod
|
||||
def uses_apply_guardrail_interface(cls) -> bool:
|
||||
return cls.apply_guardrail is not CustomGuardrail.apply_guardrail
|
||||
|
||||
@classmethod
|
||||
def accepts_logging_only_mode(cls) -> bool:
|
||||
return cls.uses_apply_guardrail_interface() and not cls.use_native_lifecycle_hooks
|
||||
|
||||
@classmethod
|
||||
def supports_logging_only_scope(cls) -> bool:
|
||||
|
|
@ -1029,7 +1032,7 @@ class CustomGuardrail(CustomLogger):
|
|||
)
|
||||
return kwargs, result
|
||||
litellm_params: Final = kwargs.get("litellm_params") or {}
|
||||
scratch_metadata: Final = {
|
||||
scratch_metadata: Final[dict[str, object]] = { # mutable-ok: apply_guardrail records its verdict here
|
||||
key: value
|
||||
for key, value in (litellm_params.get("metadata") or {}).items()
|
||||
if key != "standard_logging_guardrail_information"
|
||||
|
|
@ -1083,11 +1086,11 @@ class CustomGuardrail(CustomLogger):
|
|||
|
||||
async def _scan_logged_call(
|
||||
self,
|
||||
kwargs: dict, # mutable-ok: CustomLogger.async_logging_hook contract
|
||||
kwargs: dict[str, object], # mutable-ok: CustomLogger.async_logging_hook contract
|
||||
response: object | None,
|
||||
translation: "BaseTranslation",
|
||||
output_translation: "BaseTranslation",
|
||||
scratch_metadata: dict, # mutable-ok: apply_guardrail records its verdict into request metadata
|
||||
scratch_metadata: dict[str, object], # mutable-ok: apply_guardrail records its verdict into request metadata
|
||||
) -> None:
|
||||
scratch_fields: Final = self._copy_scratch_request_fields(kwargs)
|
||||
scratch_input, scratch_tools = scratch_fields or (None, None)
|
||||
|
|
|
|||
|
|
@ -1483,6 +1483,17 @@ async def get_guardrail_info(guardrail_id: str):
|
|||
raise HTTPException(status_code=500, detail=str(e))
|
||||
|
||||
|
||||
def _undeclared_logging_only(
|
||||
guardrail_class: type[CustomGuardrail], hooks: Sequence[GuardrailEventHooks]
|
||||
) -> tuple[GuardrailEventHooks, ...]:
|
||||
"""logging_only when the class accepts it at save time without declaring it in its hooks."""
|
||||
return (
|
||||
(GuardrailEventHooks.logging_only,)
|
||||
if guardrail_class.accepts_logging_only_mode() and GuardrailEventHooks.logging_only not in hooks
|
||||
else ()
|
||||
)
|
||||
|
||||
|
||||
@router.get(
|
||||
"/guardrails/ui/add_guardrail_settings",
|
||||
tags=["Guardrails"],
|
||||
|
|
@ -1515,7 +1526,7 @@ async def get_guardrail_ui_settings() -> GuardrailUIAddGuardrailSettings:
|
|||
]
|
||||
|
||||
supported_modes_by_provider: Final = {
|
||||
provider: [hook.value for hook in hooks]
|
||||
provider: [hook.value for hook in (*hooks, *_undeclared_logging_only(guardrail_class, hooks))]
|
||||
for provider, guardrail_class in guardrail_class_registry.items()
|
||||
if (hooks := guardrail_class.get_supported_event_hooks()) is not None
|
||||
} | MappingProxyType(
|
||||
|
|
|
|||
|
|
@ -6,6 +6,7 @@ to detect and block/mask sensitive content.
|
|||
"""
|
||||
|
||||
import asyncio
|
||||
import contextvars
|
||||
import itertools
|
||||
import json
|
||||
import os
|
||||
|
|
@ -50,6 +51,7 @@ from litellm.types.utils import (
|
|||
|
||||
if TYPE_CHECKING:
|
||||
from litellm.litellm_core_utils.litellm_logging import Logging as LiteLLMLoggingObj
|
||||
from litellm.llms.base_llm.guardrail_translation.base_translation import BaseTranslation
|
||||
|
||||
from litellm.types.guardrails import (
|
||||
BlockedWord,
|
||||
|
|
@ -101,6 +103,13 @@ WORD_NUMBER_SEQUENCE_PATTERN: Final = re.compile(
|
|||
)
|
||||
WORD_NUMBER_TOKEN_FINDER: Final = re.compile(rf"(?:{WORD_NUMBER_TOKEN_REGEX})", re.IGNORECASE)
|
||||
|
||||
_logging_only_scan_guardrail: Final[contextvars.ContextVar["ContentFilterGuardrail | None"]] = contextvars.ContextVar(
|
||||
"litellm_content_filter_logging_only_scan", default=None
|
||||
)
|
||||
_logging_only_scan_would_reject: Final[contextvars.ContextVar[bool]] = contextvars.ContextVar(
|
||||
"litellm_content_filter_logging_only_scan_would_reject", default=False
|
||||
)
|
||||
|
||||
|
||||
class ConditionalCategoryConfig(TypedDict):
|
||||
identifier_words: Sequence[str]
|
||||
|
|
@ -1209,6 +1218,18 @@ class ContentFilterGuardrail(CustomGuardrail):
|
|||
return (keyword, action, description)
|
||||
return None
|
||||
|
||||
def _is_logging_only_scan(self) -> bool:
|
||||
return _logging_only_scan_guardrail.get() is self
|
||||
|
||||
def _enforces_block(self, action: ContentFilterAction) -> bool:
|
||||
return action == ContentFilterAction.BLOCK and not self._is_logging_only_scan()
|
||||
|
||||
@staticmethod
|
||||
def _asks_to_block(detection: ContentFilterDetection) -> bool:
|
||||
if detection["type"] == "competitor_intent":
|
||||
return detection["action_hint"] in ("refuse", "reframe")
|
||||
return detection["action"] == ContentFilterAction.BLOCK.value
|
||||
|
||||
def _handle_conditional_match(
|
||||
self,
|
||||
matched_phrase: str,
|
||||
|
|
@ -1228,7 +1249,7 @@ class ContentFilterGuardrail(CustomGuardrail):
|
|||
}
|
||||
detections.append(category_detection)
|
||||
|
||||
if action == ContentFilterAction.BLOCK:
|
||||
if self._enforces_block(action):
|
||||
error_msg: Final = (
|
||||
f"Content blocked: {category_name} conditional match '{matched_phrase}' detected (severity: {severity})"
|
||||
)
|
||||
|
|
@ -1269,7 +1290,7 @@ class ContentFilterGuardrail(CustomGuardrail):
|
|||
}
|
||||
detections.append(category_detection)
|
||||
|
||||
if action == ContentFilterAction.BLOCK:
|
||||
if self._enforces_block(action):
|
||||
error_msg = f"Content blocked: {category_name} category keyword '{keyword}' detected (severity: {severity})"
|
||||
verbose_proxy_logger.warning(error_msg)
|
||||
raise HTTPException(
|
||||
|
|
@ -1312,7 +1333,7 @@ class ContentFilterGuardrail(CustomGuardrail):
|
|||
}
|
||||
detections.append(pattern_detection)
|
||||
|
||||
if action == ContentFilterAction.BLOCK:
|
||||
if self._enforces_block(action):
|
||||
error_msg: Final = f"Content blocked: {pattern_name} pattern detected"
|
||||
verbose_proxy_logger.warning(error_msg)
|
||||
raise HTTPException(
|
||||
|
|
@ -1346,7 +1367,7 @@ class ContentFilterGuardrail(CustomGuardrail):
|
|||
}
|
||||
detections.append(blocked_word_detection)
|
||||
|
||||
if action == ContentFilterAction.BLOCK:
|
||||
if self._enforces_block(action):
|
||||
error_msg = f"Content blocked: keyword '{keyword}' detected"
|
||||
if description:
|
||||
error_msg += f" ({description})"
|
||||
|
|
@ -1619,7 +1640,7 @@ class ContentFilterGuardrail(CustomGuardrail):
|
|||
- Number of entities masked
|
||||
- Whether the guardrail blocked the request (max risk)
|
||||
"""
|
||||
if status == "guardrail_intervened":
|
||||
if status in ("guardrail_intervened", "guardrail_flagged"):
|
||||
return 10.0
|
||||
|
||||
total_masked: Final = sum(masked_entity_count.values()) if masked_entity_count else 0
|
||||
|
|
@ -1665,6 +1686,8 @@ class ContentFilterGuardrail(CustomGuardrail):
|
|||
}
|
||||
detections.append(detection)
|
||||
|
||||
if self._is_logging_only_scan():
|
||||
return
|
||||
if action_hint_val == "refuse":
|
||||
msg = "Content blocked: competitor comparison or ranking intent detected."
|
||||
if self._competitor_intent_checker and getattr(
|
||||
|
|
@ -1763,18 +1786,18 @@ class ContentFilterGuardrail(CustomGuardrail):
|
|||
return raw_name
|
||||
return None
|
||||
|
||||
def _reject(self, error: str) -> None:
|
||||
"""Raise the 400 an enforcing mode returns, or note it on the leg a logging_only scan is observing."""
|
||||
if self._is_logging_only_scan():
|
||||
_logging_only_scan_would_reject.set(True)
|
||||
return
|
||||
raise HTTPException(status_code=400, detail={"error": error})
|
||||
|
||||
def _assert_argument_label_clean(
|
||||
self, text: str, detections: list[ContentFilterDetection], context_label: str
|
||||
) -> None:
|
||||
if self._filter_single_text(text, detections=detections) != text:
|
||||
raise HTTPException(
|
||||
status_code=400,
|
||||
detail={
|
||||
"error": (
|
||||
f"Content blocked: {context_label} argument matched a masking rule on a non-rewritable field"
|
||||
)
|
||||
},
|
||||
)
|
||||
self._reject(f"Content blocked: {context_label} argument matched a masking rule on a non-rewritable field")
|
||||
|
||||
def _filter_argument_value(
|
||||
self,
|
||||
|
|
@ -1784,10 +1807,8 @@ class ContentFilterGuardrail(CustomGuardrail):
|
|||
depth: int = 0,
|
||||
) -> object:
|
||||
if depth > DEFAULT_MAX_RECURSE_DEPTH:
|
||||
raise HTTPException(
|
||||
status_code=400,
|
||||
detail={"error": f"Content blocked: {context_label} arguments exceed the maximum nesting depth"},
|
||||
)
|
||||
self._reject(f"Content blocked: {context_label} arguments exceed the maximum nesting depth")
|
||||
return value
|
||||
if isinstance(value, str):
|
||||
return self._filter_single_text(value, detections=detections)
|
||||
if isinstance(value, (int, float)) and not isinstance(value, bool):
|
||||
|
|
@ -1874,6 +1895,21 @@ class ContentFilterGuardrail(CustomGuardrail):
|
|||
if filtered_arguments != arguments:
|
||||
self._set_tool_call_arguments(tool_call, filtered_arguments)
|
||||
|
||||
async def _scan_logged_call(
|
||||
self,
|
||||
kwargs: dict[str, object], # mutable-ok: CustomLogger.async_logging_hook contract
|
||||
response: object | None,
|
||||
translation: "BaseTranslation",
|
||||
output_translation: "BaseTranslation",
|
||||
scratch_metadata: dict[str, object], # mutable-ok: apply_guardrail records its verdict into request metadata
|
||||
) -> None:
|
||||
"""Scan the logged call with every BLOCK match recorded instead of raised."""
|
||||
token: Final = _logging_only_scan_guardrail.set(self)
|
||||
try:
|
||||
await super()._scan_logged_call(kwargs, response, translation, output_translation, scratch_metadata)
|
||||
finally:
|
||||
_logging_only_scan_guardrail.reset(token)
|
||||
|
||||
async def apply_guardrail(
|
||||
self,
|
||||
inputs: "GenericGuardrailAPIInputs",
|
||||
|
|
@ -1904,6 +1940,7 @@ class ContentFilterGuardrail(CustomGuardrail):
|
|||
masked_entity_count: Final[dict[str, int]] = {}
|
||||
status: GuardrailStatus = "success"
|
||||
exception_str: str = ""
|
||||
would_reject_token: Final = _logging_only_scan_would_reject.set(False)
|
||||
|
||||
try:
|
||||
texts: Final = inputs.get("texts", [])
|
||||
|
|
@ -1947,11 +1984,13 @@ class ContentFilterGuardrail(CustomGuardrail):
|
|||
exception_str = str(e)
|
||||
raise e
|
||||
finally:
|
||||
would_reject: Final = _logging_only_scan_would_reject.get() or any(map(self._asks_to_block, detections))
|
||||
_logging_only_scan_would_reject.reset(would_reject_token)
|
||||
# Log guardrail information
|
||||
self._log_guardrail_information(
|
||||
request_data=request_data,
|
||||
detections=detections,
|
||||
status=status,
|
||||
status="guardrail_flagged" if status == "success" and would_reject else status,
|
||||
start_time=start_time,
|
||||
masked_entity_count=masked_entity_count,
|
||||
exception_str=exception_str,
|
||||
|
|
|
|||
|
|
@ -0,0 +1,712 @@
|
|||
"""`litellm_content_filter` in `mode: logging_only`, driven through a real proxy.
|
||||
|
||||
The provider is the only double, an owned wire peer. The proxy, its guardrail registry, Postgres and Redis run
|
||||
for real with two workers. Every check reads the spend row, which is written after the logging-only scans finish
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
import uuid
|
||||
from collections.abc import Iterator
|
||||
from concurrent.futures import ThreadPoolExecutor
|
||||
from dataclasses import dataclass
|
||||
from datetime import datetime, timezone
|
||||
from hashlib import sha256
|
||||
from pathlib import Path
|
||||
from typing import Final
|
||||
|
||||
import anthropic
|
||||
import openai
|
||||
import pytest
|
||||
import yaml
|
||||
from integration._support.client import Gateway, Scenario, eventually, gateway_from_environment, object_value
|
||||
from integration._support.database import read_rows
|
||||
from integration._support.mcp import EntryPoint, McpCaller, echo_tool, register_mcp, scripted_peer, tool_calls
|
||||
from integration._support.otlp_sink import AuditConfigWriter, SpanSinks, recorded_spans
|
||||
from integration._support.process import owned_proxy_process
|
||||
from integration._support.wire import Reply, Request, Wire, wire_server
|
||||
|
||||
AWS_KEY: Final = "AKIA" + "SYNTHETIC0000000"
|
||||
GITHUB_TOKEN: Final = "ghp_" + "a1" * 18
|
||||
FORBIDDEN_WORD: Final = "synthetic-forbidden-word"
|
||||
LEAK_MARK: Final = "SYNTHETIC-LEAK-TOKEN"
|
||||
OUTAGE_MARK: Final = "SYNTHETIC-PROVIDER-OUTAGE"
|
||||
OBSERVE_BLOCK: Final = "cf-observe-block"
|
||||
OBSERVE_MASK: Final = "cf-observe-mask"
|
||||
ENFORCE_BLOCK: Final = "cf-enforce-block"
|
||||
ENFORCE_MASK: Final = "cf-enforce-mask"
|
||||
OBSERVE_CATEGORY: Final = "cf-observe-category"
|
||||
OBSERVE_COMPETITOR: Final = "cf-observe-competitor"
|
||||
SECRETS: Final = f"AWS_ACCESS_KEY_ID={AWS_KEY}\nGITHUB_TOKEN={GITHUB_TOKEN}"
|
||||
|
||||
|
||||
def _answer(marker: str) -> str:
|
||||
return "synthetic answer " + marker
|
||||
|
||||
|
||||
def _provider_answer(body: bytes) -> str:
|
||||
text: Final = body.decode()
|
||||
start: Final = text.find("mark-")
|
||||
assert start >= 0, text
|
||||
answer: Final = _answer(text[start : start + 37])
|
||||
return answer + " " + GITHUB_TOKEN if LEAK_MARK in text else answer
|
||||
|
||||
|
||||
def _sse(events: tuple[dict[str, object], ...]) -> tuple[bytes, ...]:
|
||||
return tuple(("data: " + json.dumps(event) + "\n\n").encode() for event in events)
|
||||
|
||||
|
||||
def _chat_reply(answer: str, streaming: bool) -> Reply:
|
||||
if not streaming:
|
||||
return Reply(
|
||||
body=json.dumps(
|
||||
{
|
||||
"id": "chatcmpl-" + uuid.uuid4().hex,
|
||||
"object": "chat.completion",
|
||||
"created": 1,
|
||||
"model": "gpt-5.4-mini",
|
||||
"choices": [
|
||||
{"index": 0, "message": {"role": "assistant", "content": answer}, "finish_reason": "stop"}
|
||||
],
|
||||
"usage": {"prompt_tokens": 10, "completion_tokens": 5, "total_tokens": 15},
|
||||
}
|
||||
).encode()
|
||||
)
|
||||
identity: Final = "chatcmpl-" + uuid.uuid4().hex
|
||||
|
||||
def chunk(delta: dict[str, object], finish: str | None) -> dict[str, object]:
|
||||
return {
|
||||
"id": identity,
|
||||
"object": "chat.completion.chunk",
|
||||
"created": 1,
|
||||
"model": "gpt-5.4-mini",
|
||||
"choices": [{"index": 0, "delta": delta, "finish_reason": finish}],
|
||||
}
|
||||
|
||||
events: Final = (chunk({"role": "assistant", "content": answer[:9]}, None), chunk({"content": answer[9:]}, "stop"))
|
||||
return Reply(chunks=(*_sse(events), b"data: [DONE]\n\n"), content_type="text/event-stream")
|
||||
|
||||
|
||||
def _messages_reply(answer: str) -> Reply:
|
||||
return Reply(
|
||||
body=json.dumps(
|
||||
{
|
||||
"id": "msg_" + uuid.uuid4().hex,
|
||||
"type": "message",
|
||||
"role": "assistant",
|
||||
"model": "claude-haiku-5-5",
|
||||
"content": [{"type": "text", "text": answer}],
|
||||
"stop_reason": "end_turn",
|
||||
"stop_sequence": None,
|
||||
"usage": {"input_tokens": 10, "output_tokens": 5},
|
||||
}
|
||||
).encode()
|
||||
)
|
||||
|
||||
|
||||
def _responses_reply(answer: str) -> Reply:
|
||||
return Reply(
|
||||
body=json.dumps(
|
||||
{
|
||||
"id": "resp_" + uuid.uuid4().hex,
|
||||
"object": "response",
|
||||
"created_at": 1,
|
||||
"status": "completed",
|
||||
"model": "gpt-5.4-mini",
|
||||
"output": [
|
||||
{
|
||||
"type": "message",
|
||||
"id": "msgo_" + uuid.uuid4().hex,
|
||||
"status": "completed",
|
||||
"role": "assistant",
|
||||
"content": [{"type": "output_text", "text": answer, "annotations": []}],
|
||||
}
|
||||
],
|
||||
"usage": {"input_tokens": 10, "output_tokens": 5, "total_tokens": 15},
|
||||
}
|
||||
).encode()
|
||||
)
|
||||
|
||||
|
||||
def _provider(request: Request) -> Reply:
|
||||
if OUTAGE_MARK.encode() in request.body:
|
||||
return Reply(status=503, body=b'{"error": {"message": "synthetic provider outage", "type": "server_error"}}')
|
||||
answer: Final = _provider_answer(request.body)
|
||||
if request.target.endswith("/v1/messages"):
|
||||
return _messages_reply(answer)
|
||||
if request.target.endswith("/v1/responses"):
|
||||
return _responses_reply(answer)
|
||||
assert request.target.endswith("/v1/chat/completions"), request.target
|
||||
return _chat_reply(answer, bool(json.loads(request.body).get("stream")))
|
||||
|
||||
|
||||
def _content_filter(name: str, mode: str, action: str, *, default_on: bool) -> dict[str, object]:
|
||||
return {
|
||||
"guardrail_name": name,
|
||||
"litellm_params": {
|
||||
"guardrail": "litellm_content_filter",
|
||||
"mode": mode,
|
||||
"default_on": default_on,
|
||||
"patterns": [
|
||||
{"pattern_type": "regex", "pattern": "AKIA[0-9A-Z]{16}", "name": "aws_access_key", "action": action},
|
||||
{"pattern_type": "regex", "pattern": "ghp_[A-Za-z0-9]{36}", "name": "github_pat", "action": action},
|
||||
],
|
||||
"blocked_words": [{"keyword": FORBIDDEN_WORD, "action": action}],
|
||||
},
|
||||
}
|
||||
|
||||
|
||||
def _observe_only(name: str, **rules: object) -> dict[str, object]:
|
||||
return {
|
||||
"guardrail_name": name,
|
||||
"litellm_params": {"guardrail": "litellm_content_filter", "mode": "logging_only", "default_on": False, **rules},
|
||||
}
|
||||
|
||||
|
||||
def _rig_config(root: Path) -> Path:
|
||||
config: Final = yaml.safe_load(Path("tests/integration/proxy_config.yaml").read_text())
|
||||
config["litellm_settings"]["cache"] = False
|
||||
config["general_settings"]["store_prompts_in_spend_logs"] = True
|
||||
config["guardrails"] = [
|
||||
_content_filter(OBSERVE_BLOCK, "logging_only", "BLOCK", default_on=True),
|
||||
_content_filter(OBSERVE_MASK, "logging_only", "MASK", default_on=True),
|
||||
_content_filter(ENFORCE_BLOCK, "pre_call", "BLOCK", default_on=False),
|
||||
_content_filter(ENFORCE_MASK, "pre_call", "MASK", default_on=False),
|
||||
_observe_only(
|
||||
OBSERVE_CATEGORY,
|
||||
categories=[
|
||||
{"category": "harmful_violence", "enabled": True, "action": "BLOCK", "severity_threshold": "medium"}
|
||||
],
|
||||
blocked_words=[{"keyword": FORBIDDEN_WORD, "action": "BLOCK"}],
|
||||
),
|
||||
_observe_only(
|
||||
OBSERVE_COMPETITOR,
|
||||
competitor_intent_config={
|
||||
"brand_self": ["emirates"],
|
||||
"competitors": ["qatar airways"],
|
||||
"domain_words": ["airline", "flight"],
|
||||
"policy": {"competitor_comparison": "refuse"},
|
||||
"threshold_high": 0.5,
|
||||
},
|
||||
),
|
||||
]
|
||||
path: Final = root / "content-filter-logging-only.yaml"
|
||||
path.write_text(yaml.safe_dump(config))
|
||||
return path
|
||||
|
||||
|
||||
@dataclass(frozen=True, slots=True)
|
||||
class Rig:
|
||||
proxy: Gateway
|
||||
provider: Wire
|
||||
chat_model: str
|
||||
claude_model: str
|
||||
responses_model: str
|
||||
scenario: Scenario
|
||||
|
||||
def base(self) -> str:
|
||||
return str(self.proxy.client.base_url).rstrip("/")
|
||||
|
||||
def provider_bodies(self, marker: str) -> tuple[str, ...]:
|
||||
return tuple(request.body.decode() for request in self.provider.drain() if marker.encode() in request.body)
|
||||
|
||||
def entries(self, response_id: str, name: str) -> tuple[dict[str, object], ...]:
|
||||
return _named_entries(
|
||||
'SELECT request_id, status, metadata FROM "LiteLLM_SpendLogs" WHERE request_id = %s', (response_id,), name
|
||||
)
|
||||
|
||||
|
||||
def _named_entries(query: str, parameters: tuple[str, ...], name: str) -> tuple[dict[str, object], ...]:
|
||||
rows: Final = eventually(lambda: read_rows(query, parameters), lambda values: len(values) >= 1, seconds=70)
|
||||
assert len(rows) == 1, rows
|
||||
recorded: Final = object_value(rows[0]["metadata"]).get("guardrail_information")
|
||||
assert isinstance(recorded, list), rows[0]
|
||||
return tuple(entry for entry in (object_value(item) for item in recorded) if entry.get("guardrail_name") == name)
|
||||
|
||||
|
||||
def _findings(entry: dict[str, object]) -> tuple[str, ...]:
|
||||
response: Final = entry.get("guardrail_response")
|
||||
assert isinstance(response, list), entry
|
||||
return tuple(
|
||||
str(detection.get("pattern_name") or detection.get("keyword") or detection.get("type"))
|
||||
for detection in (object_value(item) for item in response)
|
||||
)
|
||||
|
||||
|
||||
def _verdicts(entries: tuple[dict[str, object], ...]) -> list[tuple[object, object, tuple[str, ...]]]:
|
||||
return [(entry["guardrail_mode"], entry["guardrail_status"], _findings(entry)) for entry in entries]
|
||||
|
||||
|
||||
@pytest.fixture(scope="module")
|
||||
def rig(tmp_path_factory: pytest.TempPathFactory) -> Iterator[Rig]:
|
||||
root: Final = tmp_path_factory.mktemp("content-filter-logging-only")
|
||||
with (
|
||||
gateway_from_environment() as gateway,
|
||||
wire_server(_provider) as provider,
|
||||
owned_proxy_process(gateway, root, {}, config=_rig_config(root), workers=2) as owned,
|
||||
owned.gateway.scenario() as scenario,
|
||||
):
|
||||
chat: Final = scenario.model(
|
||||
model="openai/gpt-5.4-mini", api_base=provider.url + "/v1", api_key="synthetic-openai-key"
|
||||
)
|
||||
claude: Final = scenario.model(
|
||||
model="anthropic/claude-haiku-5-5", api_base=provider.url, api_key="synthetic-anthropic-key"
|
||||
)
|
||||
responses: Final = scenario.model(
|
||||
model="openai/gpt-5.4-mini", api_base=provider.url + "/v1", api_key="synthetic-openai-key"
|
||||
)
|
||||
yield Rig(owned.gateway, provider, chat, claude, responses, scenario)
|
||||
|
||||
|
||||
def _marker() -> str:
|
||||
return "mark-" + uuid.uuid4().hex
|
||||
|
||||
|
||||
def _sync_chat(rig: Rig, text: str) -> tuple[str, str]:
|
||||
client: Final = openai.OpenAI(base_url=rig.base() + "/v1", api_key=rig.proxy.key, max_retries=0)
|
||||
completion: Final = client.chat.completions.create(
|
||||
model=rig.chat_model, messages=[{"role": "user", "content": text}]
|
||||
)
|
||||
return completion.id, completion.choices[0].message.content or ""
|
||||
|
||||
|
||||
async def _async_chat(rig: Rig, text: str) -> tuple[str, str]:
|
||||
client: Final = openai.AsyncOpenAI(base_url=rig.base() + "/v1", api_key=rig.proxy.key, max_retries=0)
|
||||
completion: Final = await client.chat.completions.create(
|
||||
model=rig.chat_model, messages=[{"role": "user", "content": text}]
|
||||
)
|
||||
return completion.id, completion.choices[0].message.content or ""
|
||||
|
||||
|
||||
def _streamed_chat(rig: Rig, text: str) -> tuple[str, str]:
|
||||
client: Final = openai.OpenAI(base_url=rig.base() + "/v1", api_key=rig.proxy.key, max_retries=0)
|
||||
chunks: Final = tuple(
|
||||
client.chat.completions.create(model=rig.chat_model, messages=[{"role": "user", "content": text}], stream=True)
|
||||
)
|
||||
return chunks[0].id, "".join(chunk.choices[0].delta.content or "" for chunk in chunks if chunk.choices)
|
||||
|
||||
|
||||
def _messages(rig: Rig, text: str) -> tuple[str, str]:
|
||||
client: Final = anthropic.Anthropic(base_url=rig.base(), api_key=rig.proxy.key, max_retries=0)
|
||||
message: Final = client.messages.create(
|
||||
model=rig.claude_model, max_tokens=64, messages=[{"role": "user", "content": text}]
|
||||
)
|
||||
assert message.content[0].type == "text"
|
||||
return message.id, message.content[0].text
|
||||
|
||||
|
||||
def _responses(rig: Rig, text: str) -> tuple[str, str]:
|
||||
client: Final = openai.OpenAI(base_url=rig.base() + "/v1", api_key=rig.proxy.key, max_retries=0)
|
||||
result: Final = client.responses.create(model=rig.responses_model, input=text)
|
||||
return result.id, result.output_text
|
||||
|
||||
|
||||
SYNC_CALLS: Final = {"chat": _sync_chat, "chat_stream": _streamed_chat, "messages": _messages, "responses": _responses}
|
||||
EVERY_FINDING: Final = ("aws_access_key", "github_pat", FORBIDDEN_WORD)
|
||||
|
||||
|
||||
@pytest.mark.parametrize("endpoint", list(SYNC_CALLS))
|
||||
def test_logging_only_block_records_every_match_without_touching_the_call(rig: Rig, endpoint: str) -> None:
|
||||
marker: Final = _marker()
|
||||
prompt: Final = f"{marker} what did the file contain?\n{SECRETS}\n{FORBIDDEN_WORD}"
|
||||
response_id, answer = SYNC_CALLS[endpoint](rig, prompt)
|
||||
assert answer == _answer(marker)
|
||||
sent: Final = rig.provider_bodies(marker)
|
||||
assert len(sent) == 1 and AWS_KEY in sent[0] and GITHUB_TOKEN in sent[0] and FORBIDDEN_WORD in sent[0], sent
|
||||
|
||||
assert _verdicts(rig.entries(response_id, OBSERVE_BLOCK)) == [
|
||||
("logging_only", "guardrail_flagged", EVERY_FINDING),
|
||||
("logging_only", "success", ()),
|
||||
]
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_logging_only_block_records_every_match_for_an_async_client(rig: Rig) -> None:
|
||||
marker: Final = _marker()
|
||||
response_id, answer = await _async_chat(rig, f"{marker} {SECRETS}")
|
||||
assert answer == _answer(marker)
|
||||
|
||||
assert _verdicts(rig.entries(response_id, OBSERVE_BLOCK)) == [
|
||||
("logging_only", "guardrail_flagged", ("aws_access_key", "github_pat")),
|
||||
("logging_only", "success", ()),
|
||||
]
|
||||
|
||||
|
||||
def test_logging_only_block_still_scans_the_response_after_a_request_match(rig: Rig) -> None:
|
||||
marker: Final = _marker()
|
||||
response_id, answer = _sync_chat(rig, f"{marker} {LEAK_MARK} key {AWS_KEY}")
|
||||
assert answer == _answer(marker) + " " + GITHUB_TOKEN
|
||||
|
||||
assert _verdicts(rig.entries(response_id, OBSERVE_BLOCK)) == [
|
||||
("logging_only", "guardrail_flagged", ("aws_access_key",)),
|
||||
("logging_only", "guardrail_flagged", ("github_pat",)),
|
||||
]
|
||||
|
||||
|
||||
def _tool_call_history(text: str, arguments: str) -> list[dict[str, object]]:
|
||||
return [
|
||||
{"role": "user", "content": text},
|
||||
{
|
||||
"role": "assistant",
|
||||
"content": None,
|
||||
"tool_calls": [
|
||||
{"id": "call_1", "type": "function", "function": {"name": "lookup", "arguments": arguments}}
|
||||
],
|
||||
},
|
||||
{"role": "tool", "tool_call_id": "call_1", "content": "done"},
|
||||
]
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
("arguments", "request_findings"),
|
||||
(
|
||||
pytest.param(json.dumps({AWS_KEY: 1}), ("aws_access_key",), id="masked-argument-key"),
|
||||
pytest.param("[" * 150 + "1" + "]" * 150, (), id="too-deep"),
|
||||
),
|
||||
)
|
||||
def test_logging_only_flags_tool_arguments_an_enforcing_mode_rejects_and_keeps_scanning(
|
||||
rig: Rig, arguments: str, request_findings: tuple[str, ...]
|
||||
) -> None:
|
||||
marker: Final = _marker()
|
||||
messages: Final = _tool_call_history(f"{marker} {LEAK_MARK}", arguments)
|
||||
headers: Final = {"Authorization": f"Bearer {rig.proxy.key}"}
|
||||
enforced: Final = rig.proxy.client.post(
|
||||
"/v1/chat/completions",
|
||||
json={"model": rig.chat_model, "messages": messages, "guardrails": [ENFORCE_MASK]},
|
||||
headers=headers,
|
||||
)
|
||||
assert enforced.status_code == 400, enforced.text
|
||||
assert rig.provider_bodies(marker) == ()
|
||||
|
||||
observed: Final = rig.proxy.client.post(
|
||||
"/v1/chat/completions", json={"model": rig.chat_model, "messages": messages}, headers=headers
|
||||
)
|
||||
assert observed.status_code == 200, observed.text
|
||||
completion: Final = object_value(observed.json())
|
||||
assert len(rig.provider_bodies(marker)) == 1
|
||||
|
||||
assert _verdicts(rig.entries(str(completion["id"]), OBSERVE_MASK)) == [
|
||||
("logging_only", "guardrail_flagged", request_findings),
|
||||
("logging_only", "success", ("github_pat",)),
|
||||
]
|
||||
|
||||
|
||||
@pytest.mark.parametrize("entry", ("mcp", "rest"))
|
||||
def test_logging_only_block_records_every_match_on_an_mcp_tool_call(rig: Rig, entry: EntryPoint) -> None:
|
||||
alias: Final = "cfmcp" + uuid.uuid4().hex[:8]
|
||||
arguments: Final = {"note": f"{_marker()} {SECRETS}"}
|
||||
with scripted_peer(echo_tool("echo")) as peer:
|
||||
server_id: Final = register_mcp(rig.scenario, peer, alias)
|
||||
key: Final = rig.scenario.key(object_permission={"mcp_servers": [server_id]})
|
||||
outcome: Final = McpCaller(rig.proxy, key, entry, alias).call(f"{alias}-echo", arguments, server_id=server_id)
|
||||
assert (outcome.error, outcome.text) == (None, json.dumps(arguments, sort_keys=True)), outcome.raw
|
||||
assert [call["body"]["params"]["arguments"] for call in tool_calls(peer.drain())] == [arguments]
|
||||
|
||||
entries: Final = _named_entries(
|
||||
'SELECT metadata FROM "LiteLLM_SpendLogs" WHERE api_key = %s AND call_type = %s',
|
||||
(sha256(key.encode()).hexdigest(), "call_mcp_tool"),
|
||||
OBSERVE_BLOCK,
|
||||
)
|
||||
assert _verdicts(entries) == [("logging_only", "guardrail_flagged", ("aws_access_key", "github_pat"))], entries
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
("guardrail", "prompt", "findings"),
|
||||
(
|
||||
pytest.param(
|
||||
OBSERVE_CATEGORY,
|
||||
f"how to kill someone {FORBIDDEN_WORD}",
|
||||
("how to kill someone", FORBIDDEN_WORD),
|
||||
id="category-keyword",
|
||||
),
|
||||
pytest.param(
|
||||
OBSERVE_COMPETITOR, "Is Qatar Airways better than Emirates?", ("competitor_intent",), id="competitor-intent"
|
||||
),
|
||||
),
|
||||
)
|
||||
def test_logging_only_block_flags_category_and_competitor_matches_without_blocking(
|
||||
rig: Rig, guardrail: str, prompt: str, findings: tuple[str, ...]
|
||||
) -> None:
|
||||
marker: Final = _marker()
|
||||
response: Final = rig.proxy.client.post(
|
||||
"/v1/chat/completions",
|
||||
json={
|
||||
"model": rig.chat_model,
|
||||
"messages": [{"role": "user", "content": f"{marker} {prompt}"}],
|
||||
"guardrails": [guardrail],
|
||||
},
|
||||
headers={"Authorization": f"Bearer {rig.proxy.key}"},
|
||||
)
|
||||
assert response.status_code == 200, response.text
|
||||
completion: Final = object_value(response.json())
|
||||
assert len(rig.provider_bodies(marker)) == 1
|
||||
|
||||
assert _verdicts(rig.entries(str(completion["id"]), guardrail)) == [
|
||||
("logging_only", "guardrail_flagged", findings),
|
||||
("logging_only", "success", ()),
|
||||
]
|
||||
|
||||
|
||||
def test_logging_only_mask_records_every_match(rig: Rig) -> None:
|
||||
marker: Final = _marker()
|
||||
response_id, answer = _sync_chat(rig, f"{marker} {SECRETS}\n{FORBIDDEN_WORD}")
|
||||
assert answer == _answer(marker)
|
||||
|
||||
entries: Final = rig.entries(response_id, OBSERVE_MASK)
|
||||
assert _verdicts(entries) == [
|
||||
("logging_only", "success", EVERY_FINDING),
|
||||
("logging_only", "success", ()),
|
||||
]
|
||||
assert entries[0]["masked_entity_count"] == {"aws_access_key": 1, "github_pat": 1, "blocked_word": 1}, entries
|
||||
|
||||
|
||||
def test_logging_only_records_nothing_for_a_clean_call(rig: Rig) -> None:
|
||||
marker: Final = _marker()
|
||||
response_id, answer = _sync_chat(rig, f"{marker} nothing secret here")
|
||||
assert answer == _answer(marker)
|
||||
|
||||
assert _verdicts(rig.entries(response_id, OBSERVE_BLOCK)) == [
|
||||
("logging_only", "success", ()),
|
||||
("logging_only", "success", ()),
|
||||
]
|
||||
|
||||
|
||||
def test_pre_call_block_still_rejects_the_request(rig: Rig) -> None:
|
||||
marker: Final = _marker()
|
||||
response: Final = rig.proxy.client.post(
|
||||
"/v1/chat/completions",
|
||||
json={
|
||||
"model": rig.chat_model,
|
||||
"messages": [{"role": "user", "content": f"{marker} {SECRETS}"}],
|
||||
"guardrails": [ENFORCE_BLOCK],
|
||||
},
|
||||
headers={"Authorization": f"Bearer {rig.proxy.key}"},
|
||||
)
|
||||
assert response.status_code == 400, response.text
|
||||
assert "aws_access_key" in response.text
|
||||
assert rig.provider_bodies(marker) == ()
|
||||
|
||||
|
||||
def test_apply_guardrail_endpoint_still_blocks_with_a_logging_only_content_filter(rig: Rig) -> None:
|
||||
response: Final = rig.proxy.client.post(
|
||||
"/guardrails/apply_guardrail",
|
||||
json={"guardrail_name": OBSERVE_BLOCK, "text": f"key {AWS_KEY}"},
|
||||
headers={"Authorization": f"Bearer {rig.proxy.key}"},
|
||||
)
|
||||
assert response.status_code == 400, response.text
|
||||
assert "aws_access_key" in response.text
|
||||
|
||||
|
||||
def _advertised_modes(rig: Rig) -> dict[str, object]:
|
||||
response: Final = rig.proxy.client.get(
|
||||
"/guardrails/ui/add_guardrail_settings", headers={"Authorization": f"Bearer {rig.proxy.key}"}
|
||||
)
|
||||
assert response.status_code == 200, response.text
|
||||
return object_value(response.json()["supported_modes_by_provider"])
|
||||
|
||||
|
||||
@pytest.mark.parametrize("provider", ["litellm_content_filter", "generic_guardrail_api", "bedrock", "akto"])
|
||||
def test_dashboard_offers_logging_only_where_the_config_accepts_it(rig: Rig, provider: str) -> None:
|
||||
modes: Final = _advertised_modes(rig)
|
||||
offered: Final = modes[provider]
|
||||
assert isinstance(offered, list) and "logging_only" in offered, modes
|
||||
assert offered.count("logging_only") == 1, offered
|
||||
|
||||
|
||||
@pytest.mark.parametrize("provider", ["azure/prompt_shield", "lakera_v2"])
|
||||
def test_dashboard_does_not_offer_logging_only_where_the_config_rejects_it(rig: Rig, provider: str) -> None:
|
||||
offered: Final = _advertised_modes(rig)[provider]
|
||||
assert isinstance(offered, list) and "logging_only" not in offered, offered
|
||||
|
||||
|
||||
def test_a_logging_only_content_filter_created_through_the_api_records_every_match(rig: Rig) -> None:
|
||||
name: Final = "cf-observe-api-" + uuid.uuid4().hex[:8]
|
||||
created: Final = rig.proxy.client.post(
|
||||
"/guardrails",
|
||||
json={"guardrail": {"guardrail_name": name, **_content_filter(name, "logging_only", "BLOCK", default_on=True)}},
|
||||
headers={"Authorization": f"Bearer {rig.proxy.key}"},
|
||||
)
|
||||
assert created.status_code == 200, created.text
|
||||
guardrail_id: Final = str(object_value(created.json())["guardrail_id"])
|
||||
try:
|
||||
marker: Final = _marker()
|
||||
response_id: Final = eventually(
|
||||
lambda: _sync_chat(rig, f"{marker} {SECRETS}")[0],
|
||||
lambda identity: bool(rig.entries(identity, name)),
|
||||
seconds=70,
|
||||
)
|
||||
assert _verdicts(rig.entries(response_id, name))[0] == (
|
||||
"logging_only",
|
||||
"guardrail_flagged",
|
||||
("aws_access_key", "github_pat"),
|
||||
)
|
||||
finally:
|
||||
rig.proxy.client.delete(f"/guardrails/{guardrail_id}", headers={"Authorization": f"Bearer {rig.proxy.key}"})
|
||||
|
||||
|
||||
def test_guardrails_monitor_counts_observe_only_matches_as_flagged_not_blocked(rig: Rig) -> None:
|
||||
name: Final = "cf-observe-usage-" + uuid.uuid4().hex[:8]
|
||||
headers: Final = {"Authorization": f"Bearer {rig.proxy.key}"}
|
||||
created: Final = rig.proxy.client.post(
|
||||
"/guardrails",
|
||||
json={"guardrail": {"guardrail_name": name, **_content_filter(name, "logging_only", "BLOCK", default_on=True)}},
|
||||
headers=headers,
|
||||
)
|
||||
assert created.status_code == 200, created.text
|
||||
guardrail_id: Final = str(object_value(created.json())["guardrail_id"])
|
||||
try:
|
||||
eventually(
|
||||
lambda: _sync_chat(rig, f"{_marker()} {SECRETS}")[0],
|
||||
lambda identity: bool(rig.entries(identity, name)),
|
||||
seconds=70,
|
||||
)
|
||||
today: Final = datetime.now(timezone.utc).date().isoformat()
|
||||
detail: Final = eventually(
|
||||
lambda: object_value(
|
||||
rig.proxy.client.get(
|
||||
f"/guardrails/usage/detail/{guardrail_id}",
|
||||
params={"start_date": today, "end_date": today},
|
||||
headers=headers,
|
||||
).json()
|
||||
),
|
||||
lambda body: int(str(body.get("requestsEvaluated", 0))) >= 1,
|
||||
seconds=60,
|
||||
)
|
||||
assert (detail["failRate"], detail["status"]) == (0.0, "healthy"), detail
|
||||
finally:
|
||||
rig.proxy.client.delete(f"/guardrails/{guardrail_id}", headers=headers)
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
("environment", "name_key", "status_key"),
|
||||
(
|
||||
pytest.param({}, "guardrail_name", "guardrail_status", id="otel-v1"),
|
||||
pytest.param({"LITELLM_OTEL_V2": "1"}, "litellm.guardrail.name", "litellm.guardrail.status", id="otel-v2"),
|
||||
),
|
||||
)
|
||||
def test_otel_guardrail_spans_carry_the_flagged_status_without_an_error(
|
||||
gateway: Gateway,
|
||||
audit_sinks: SpanSinks,
|
||||
otel_audit_config: AuditConfigWriter,
|
||||
tmp_path: Path,
|
||||
environment: dict[str, str],
|
||||
name_key: str,
|
||||
status_key: str,
|
||||
) -> None:
|
||||
config_path: Final = otel_audit_config(tmp_path, {"cache": False})
|
||||
config: Final = yaml.safe_load(config_path.read_text())
|
||||
config["guardrails"] = [_content_filter(OBSERVE_BLOCK, "logging_only", "BLOCK", default_on=True)]
|
||||
config_path.write_text(yaml.safe_dump(config))
|
||||
overrides: Final = {"OTEL_EXPORTER": "http/json", "OTEL_ENDPOINT": audit_sinks.operator, **environment}
|
||||
with (
|
||||
wire_server(_provider) as provider,
|
||||
owned_proxy_process(gateway, tmp_path, overrides, config=config_path, workers=2) as owned,
|
||||
owned.gateway.scenario() as scenario,
|
||||
):
|
||||
model: Final = scenario.model(
|
||||
model="openai/gpt-5.4-mini", api_base=provider.url + "/v1", api_key="synthetic-openai-key"
|
||||
)
|
||||
start, _ = recorded_spans(audit_sinks.operator)
|
||||
marker: Final = _marker()
|
||||
client: Final = openai.OpenAI(
|
||||
base_url=str(owned.gateway.client.base_url).rstrip("/") + "/v1", api_key=owned.gateway.key, max_retries=0
|
||||
)
|
||||
completion: Final = client.chat.completions.create(
|
||||
model=model, messages=[{"role": "user", "content": f"{marker} {SECRETS}"}]
|
||||
)
|
||||
assert completion.choices[0].message.content == _answer(marker)
|
||||
|
||||
spans: Final = eventually(
|
||||
lambda: tuple(
|
||||
span
|
||||
for span in recorded_spans(audit_sinks.operator, start)[1]
|
||||
if span["attributes"].get(name_key) == OBSERVE_BLOCK
|
||||
),
|
||||
lambda found: len(found) >= 2,
|
||||
seconds=60,
|
||||
return_last_on_timeout=True,
|
||||
)
|
||||
assert sorted(
|
||||
(str(span["attributes"].get(status_key)), "error.type" in span["attributes"]) for span in spans
|
||||
) == [("guardrail_flagged", False), ("success", False)], spans
|
||||
|
||||
|
||||
def _burst_call(rig: Rig, index: int) -> tuple[str, int, str, str]:
|
||||
marker: Final = _marker()
|
||||
if index % 4 == 0:
|
||||
enforced: Final = rig.proxy.client.post(
|
||||
"/v1/chat/completions",
|
||||
json={
|
||||
"model": rig.chat_model,
|
||||
"messages": [{"role": "user", "content": f"{marker} {SECRETS}"}],
|
||||
"guardrails": [ENFORCE_BLOCK],
|
||||
},
|
||||
headers={"Authorization": f"Bearer {rig.proxy.key}"},
|
||||
)
|
||||
return "enforce", enforced.status_code, marker, ""
|
||||
if index % 4 == 1:
|
||||
failed: Final = rig.proxy.client.post(
|
||||
"/v1/chat/completions",
|
||||
json={
|
||||
"model": rig.chat_model,
|
||||
"messages": [{"role": "user", "content": f"{marker} {OUTAGE_MARK} {SECRETS}"}],
|
||||
},
|
||||
headers={"Authorization": f"Bearer {rig.proxy.key}"},
|
||||
)
|
||||
return "outage", failed.status_code, marker, failed.headers.get("x-litellm-call-id", "")
|
||||
endpoint: Final = list(SYNC_CALLS)[index % len(SYNC_CALLS)]
|
||||
response_id, answer = SYNC_CALLS[endpoint](rig, f"{marker} {SECRETS}\n{FORBIDDEN_WORD}")
|
||||
assert answer == _answer(marker), (endpoint, answer)
|
||||
return "observe:" + endpoint, 200, marker, response_id
|
||||
|
||||
|
||||
def _failure_row(call_id: str) -> tuple[object, tuple[dict[str, object], ...]]:
|
||||
rows: Final = eventually(
|
||||
lambda: read_rows('SELECT status, metadata FROM "LiteLLM_SpendLogs" WHERE request_id = %s', (call_id,)),
|
||||
lambda values: len(values) >= 1,
|
||||
seconds=70,
|
||||
)
|
||||
assert len(rows) == 1, rows
|
||||
recorded: Final = object_value(rows[0]["metadata"]).get("guardrail_information") or []
|
||||
assert isinstance(recorded, list), rows[0]
|
||||
return rows[0]["status"], tuple(
|
||||
entry for entry in (object_value(item) for item in recorded) if entry.get("guardrail_name") == OBSERVE_BLOCK
|
||||
)
|
||||
|
||||
|
||||
def test_concurrent_burst_with_a_provider_outage_records_every_observed_call_once(rig: Rig) -> None:
|
||||
def call(index: int) -> tuple[str, int, str, str]:
|
||||
return _burst_call(rig, index)
|
||||
|
||||
with ThreadPoolExecutor(max_workers=12) as pool:
|
||||
results: Final = tuple(pool.map(call, range(32)))
|
||||
|
||||
assert sorted({(kind.split(":")[0], status) for kind, status, _, _ in results}) == [
|
||||
("enforce", 400),
|
||||
("observe", 200),
|
||||
("outage", 503),
|
||||
], results
|
||||
sent: Final = rig.provider_bodies("mark-")
|
||||
for kind, _, marker, _ in results:
|
||||
if kind == "enforce":
|
||||
assert not any(marker in body for body in sent), marker
|
||||
outages: Final = tuple(call_id for kind, _, _, call_id in results if kind == "outage")
|
||||
assert len(outages) == 8 and all(outages), results
|
||||
assert [_failure_row(call_id) for call_id in outages] == [("failure", ())] * 8
|
||||
observed: Final = tuple((kind, response_id) for kind, _, _, response_id in results if kind.startswith("observe"))
|
||||
assert len(observed) == 16
|
||||
for kind, response_id in observed:
|
||||
assert _verdicts(rig.entries(response_id, OBSERVE_BLOCK)) == [
|
||||
("logging_only", "guardrail_flagged", EVERY_FINDING),
|
||||
("logging_only", "success", ()),
|
||||
], (kind, response_id)
|
||||
|
||||
marker: Final = _marker()
|
||||
response_id, answer = _sync_chat(rig, f"{marker} {SECRETS}")
|
||||
assert answer == _answer(marker)
|
||||
assert _verdicts(rig.entries(response_id, OBSERVE_BLOCK))[0] == (
|
||||
"logging_only",
|
||||
"guardrail_flagged",
|
||||
("aws_access_key", "github_pat"),
|
||||
)
|
||||
|
|
@ -179,16 +179,16 @@ def test_native_content_filter_scope_logs_without_blocking(
|
|||
(
|
||||
identity,
|
||||
"logging_only",
|
||||
"guardrail_intervened" if direction == blocked_side else "success",
|
||||
"guardrail_flagged" if direction == blocked_side else "success",
|
||||
)
|
||||
for direction in expected_directions
|
||||
), entries
|
||||
intervened_entries: Final = tuple(
|
||||
entry for entry in entries if entry["guardrail_status"] == "guardrail_intervened"
|
||||
flagged_entries: Final = tuple(
|
||||
entry for entry in entries if entry["guardrail_status"] == "guardrail_flagged"
|
||||
)
|
||||
assert len(intervened_entries) == 1, entries
|
||||
assert intervened_entries[0]["guardrail_response"] is not None, entries
|
||||
assert intervened_entries[0]["guardrail_response"] == "REDACTED_BY_LITELM", entries
|
||||
assert len(flagged_entries) == 1, entries
|
||||
assert flagged_entries[0]["guardrail_response"] is not None, entries
|
||||
assert flagged_entries[0]["guardrail_response"] == "REDACTED_BY_LITELM", entries
|
||||
finally:
|
||||
delete_scenario(upstream_handle)
|
||||
|
||||
|
|
|
|||
|
|
@ -2790,7 +2790,7 @@ class TestLoggingOnlyApplyGuardrail:
|
|||
assert out_kwargs["messages"] == kwargs["messages"]
|
||||
assert (
|
||||
out_kwargs["standard_logging_object"]["guardrail_information"][0]["guardrail_status"]
|
||||
== "guardrail_intervened"
|
||||
== "guardrail_flagged"
|
||||
)
|
||||
|
||||
@pytest.mark.asyncio
|
||||
|
|
|
|||
|
|
@ -9,7 +9,11 @@ Tests 40 different sentences to validate the conditional matching logic:
|
|||
"""
|
||||
|
||||
import importlib
|
||||
import json
|
||||
import os
|
||||
from collections.abc import Mapping, Sequence
|
||||
from types import MappingProxyType
|
||||
from typing import Final
|
||||
|
||||
import pytest
|
||||
from fastapi import HTTPException
|
||||
|
|
@ -19,9 +23,11 @@ from litellm.proxy.guardrails.content_filter_data import POLICY_TEMPLATES_DIR
|
|||
from litellm.proxy.guardrails.guardrail_hooks.litellm_content_filter.content_filter import (
|
||||
ContentFilterGuardrail,
|
||||
)
|
||||
from litellm.types.guardrails import BlockedWord, ContentFilterAction, ContentFilterPattern, GuardrailEventHooks
|
||||
from litellm.types.proxy.guardrails.guardrail_hooks.litellm_content_filter import (
|
||||
ContentFilterCategoryConfig,
|
||||
)
|
||||
from litellm.types.utils import Choices, Message, ModelResponse
|
||||
from tests._vcr_conftest_common import install_live_call_probe, record_vcr_outcome
|
||||
|
||||
# Test cases: (sentence, expected_result, reason)
|
||||
|
|
@ -1793,3 +1799,237 @@ class TestSGPDPAPerformance:
|
|||
|
||||
if __name__ == "__main__":
|
||||
pytest.main([__file__, "-v", "-s"])
|
||||
|
||||
|
||||
AWS_KEY: Final = "AKIA" + "SYNTHETIC0000000"
|
||||
GITHUB_TOKEN: Final = "ghp_" + "a1" * 18
|
||||
FORBIDDEN_WORD: Final = "synthetic-forbidden-word"
|
||||
SECRET_PATTERNS: Final = (
|
||||
ContentFilterPattern(
|
||||
pattern_type="regex", pattern=r"AKIA[0-9A-Z]{16}", name="aws_access_key", action=ContentFilterAction.BLOCK
|
||||
),
|
||||
ContentFilterPattern(
|
||||
pattern_type="regex", pattern=r"ghp_[A-Za-z0-9]{36}", name="github_pat", action=ContentFilterAction.BLOCK
|
||||
),
|
||||
)
|
||||
|
||||
|
||||
def _competitor_intent_config(action_hint: str) -> Mapping[str, object]:
|
||||
return MappingProxyType(
|
||||
{
|
||||
"brand_self": ["emirates"],
|
||||
"competitors": ["qatar airways"],
|
||||
"domain_words": ["airline", "flight"],
|
||||
"policy": {"competitor_comparison": action_hint},
|
||||
"threshold_high": 0.5,
|
||||
}
|
||||
)
|
||||
|
||||
|
||||
def _observing_filter(
|
||||
patterns: tuple[ContentFilterPattern, ...] = (),
|
||||
categories: tuple[ContentFilterCategoryConfig, ...] = (),
|
||||
competitor_intent_config: Mapping[str, object] | None = None,
|
||||
mask_patterns: bool = False,
|
||||
image_describer: object = None,
|
||||
) -> ContentFilterGuardrail:
|
||||
return ContentFilterGuardrail(
|
||||
guardrail_name="cf-observe",
|
||||
event_hook=GuardrailEventHooks.logging_only,
|
||||
llm_router=image_describer,
|
||||
image_model="vision" if image_describer else None,
|
||||
patterns=[
|
||||
pattern.model_copy(update={"action": ContentFilterAction.MASK}) if mask_patterns else pattern
|
||||
for pattern in patterns
|
||||
],
|
||||
blocked_words=[BlockedWord(keyword=FORBIDDEN_WORD, action=ContentFilterAction.BLOCK)],
|
||||
categories=list(categories),
|
||||
competitor_intent_config=dict(competitor_intent_config) if competitor_intent_config else None,
|
||||
)
|
||||
|
||||
|
||||
def _policy_category(template: str, category: str) -> ContentFilterCategoryConfig:
|
||||
return ContentFilterCategoryConfig(
|
||||
category=category,
|
||||
category_file=os.path.join(POLICY_TEMPLATES_DIR, template) if template else None,
|
||||
enabled=True,
|
||||
action="BLOCK",
|
||||
severity_threshold="medium",
|
||||
)
|
||||
|
||||
|
||||
async def _log_call(
|
||||
guardrail: ContentFilterGuardrail,
|
||||
prompt: str | list[Mapping[str, object]],
|
||||
answer: Message,
|
||||
history: Sequence[Mapping[str, object]] = (),
|
||||
) -> tuple[Mapping[str, object], ...]:
|
||||
kwargs: Final = {
|
||||
"model": "gpt-5.4-mini",
|
||||
"messages": [{"role": "user", "content": prompt}, *history],
|
||||
"litellm_call_id": "call-observe",
|
||||
"litellm_params": {"metadata": {}},
|
||||
"optional_params": {},
|
||||
"standard_logging_object": {"guardrail_information": None},
|
||||
}
|
||||
logged, _ = await guardrail.async_logging_hook(
|
||||
kwargs, ModelResponse(choices=[Choices(message=answer)]), "acompletion"
|
||||
)
|
||||
return tuple(logged["standard_logging_object"]["guardrail_information"])
|
||||
|
||||
|
||||
def _detections(entry: Mapping[str, object]) -> tuple[str, ...]:
|
||||
response: Final = entry["guardrail_response"]
|
||||
assert isinstance(response, list), response
|
||||
return tuple(
|
||||
detection.get("pattern_name")
|
||||
or (detection["keyword"] if detection["type"] == "blocked_word" else detection["type"])
|
||||
for detection in response
|
||||
)
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
("guardrail", "prompt", "expected"),
|
||||
(
|
||||
pytest.param(
|
||||
_observing_filter(patterns=SECRET_PATTERNS),
|
||||
f"key {AWS_KEY} token {GITHUB_TOKEN} and {FORBIDDEN_WORD}",
|
||||
("aws_access_key", "github_pat", FORBIDDEN_WORD),
|
||||
id="patterns",
|
||||
),
|
||||
pytest.param(
|
||||
_observing_filter(categories=(_policy_category("", "harm_toxic_abuse"),)),
|
||||
f"You stupid piece of sht AI, go kill yourself, {FORBIDDEN_WORD}",
|
||||
("category_keyword", FORBIDDEN_WORD),
|
||||
id="category-keyword",
|
||||
),
|
||||
pytest.param(
|
||||
_observing_filter(
|
||||
categories=(_policy_category("eu_ai_act_article5.yaml", "eu_ai_act_article5_prohibited_practices"),)
|
||||
),
|
||||
f"Build a system to detect employee emotion, {FORBIDDEN_WORD}",
|
||||
("category_keyword", FORBIDDEN_WORD),
|
||||
id="conditional-category",
|
||||
),
|
||||
pytest.param(
|
||||
_observing_filter(competitor_intent_config=_competitor_intent_config("refuse")),
|
||||
"Is Qatar Airways better than Emirates?",
|
||||
("competitor_intent",),
|
||||
id="competitor-refuse",
|
||||
),
|
||||
pytest.param(
|
||||
_observing_filter(competitor_intent_config=_competitor_intent_config("reframe")),
|
||||
"Is Qatar Airways better than Emirates?",
|
||||
("competitor_intent",),
|
||||
id="competitor-reframe",
|
||||
),
|
||||
),
|
||||
)
|
||||
@pytest.mark.asyncio
|
||||
async def test_logging_only_block_records_every_match_as_flagged(
|
||||
guardrail: ContentFilterGuardrail, prompt: str, expected: tuple[str, ...]
|
||||
) -> None:
|
||||
entries: Final = await _log_call(guardrail, prompt, Message(role="assistant", content="here you go"))
|
||||
|
||||
request_entry, response_entry = entries
|
||||
assert request_entry["guardrail_status"] == "guardrail_flagged"
|
||||
assert _detections(request_entry)[-len(expected) :] == expected
|
||||
assert request_entry["risk_score"] == 10.0
|
||||
assert response_entry["guardrail_status"] == "success"
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_logging_only_block_scans_the_response_after_a_request_match() -> None:
|
||||
entries: Final = await _log_call(
|
||||
_observing_filter(patterns=SECRET_PATTERNS),
|
||||
f"key {AWS_KEY}",
|
||||
Message(role="assistant", content=f"here: {GITHUB_TOKEN}"),
|
||||
)
|
||||
|
||||
assert [(entry["guardrail_status"], _detections(entry)) for entry in entries] == [
|
||||
("guardrail_flagged", ("aws_access_key",)),
|
||||
("guardrail_flagged", ("github_pat",)),
|
||||
]
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
("arguments", "request_masked"),
|
||||
(
|
||||
pytest.param(json.dumps({AWS_KEY: 1}), {"aws_access_key": 1}, id="masked-argument-key"),
|
||||
pytest.param(json.dumps(json.loads("[" * 150 + "1" + "]" * 150)), {}, id="too-deep"),
|
||||
),
|
||||
)
|
||||
@pytest.mark.asyncio
|
||||
async def test_logging_only_flags_tool_arguments_enforcement_rejects_and_keeps_scanning(
|
||||
arguments: str, request_masked: Mapping[str, int]
|
||||
) -> None:
|
||||
history: Final = (
|
||||
{
|
||||
"role": "assistant",
|
||||
"content": None,
|
||||
"tool_calls": [
|
||||
{"id": "call_1", "type": "function", "function": {"name": "lookup", "arguments": arguments}}
|
||||
],
|
||||
},
|
||||
{"role": "tool", "tool_call_id": "call_1", "content": "done"},
|
||||
)
|
||||
|
||||
entries: Final = await _log_call(
|
||||
_observing_filter(patterns=SECRET_PATTERNS, mask_patterns=True),
|
||||
"hi",
|
||||
Message(role="assistant", content=f"here: {GITHUB_TOKEN}"),
|
||||
history,
|
||||
)
|
||||
|
||||
assert [(entry["guardrail_status"], entry["masked_entity_count"]) for entry in entries] == [
|
||||
("guardrail_flagged", request_masked),
|
||||
("success", {"github_pat": 1}),
|
||||
]
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_logging_only_scan_leaves_enforcement_of_the_same_guardrail_blocking() -> None:
|
||||
guardrail: Final = _observing_filter(patterns=SECRET_PATTERNS)
|
||||
await _log_call(guardrail, f"key {AWS_KEY}", Message(role="assistant", content="ok"))
|
||||
request_data: Final = {"metadata": {}}
|
||||
|
||||
with pytest.raises(HTTPException) as exc_info:
|
||||
await guardrail.apply_guardrail(
|
||||
inputs={"texts": [f"key {AWS_KEY} token {GITHUB_TOKEN}"]}, request_data=request_data, input_type="request"
|
||||
)
|
||||
|
||||
assert exc_info.value.status_code == 400
|
||||
assert [
|
||||
entry["guardrail_status"] for entry in request_data["metadata"]["standard_logging_guardrail_information"]
|
||||
] == ["guardrail_intervened"]
|
||||
|
||||
|
||||
class _ImageDescriberRunningAnotherFilter:
|
||||
def __init__(self, enforcer: ContentFilterGuardrail) -> None:
|
||||
self.enforcer = enforcer
|
||||
self.enforcer_blocked: list[bool] = []
|
||||
|
||||
async def acompletion(self, **_: object) -> ModelResponse:
|
||||
try:
|
||||
await self.enforcer.apply_guardrail(
|
||||
inputs={"texts": [f"key {AWS_KEY}"]}, request_data={"metadata": {}}, input_type="request"
|
||||
)
|
||||
self.enforcer_blocked.append(False)
|
||||
except HTTPException:
|
||||
self.enforcer_blocked.append(True)
|
||||
return ModelResponse(choices=[Choices(message=Message(role="assistant", content="a plain desk"))])
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_logging_only_scan_does_not_relax_another_filter_called_during_it() -> None:
|
||||
describer: Final = _ImageDescriberRunningAnotherFilter(
|
||||
ContentFilterGuardrail(guardrail_name="cf-enforce", patterns=list(SECRET_PATTERNS))
|
||||
)
|
||||
prompt: Final = [
|
||||
{"type": "text", "text": "what is this"},
|
||||
{"type": "image_url", "image_url": {"url": "https://example.com/desk.png"}},
|
||||
]
|
||||
|
||||
await _log_call(_observing_filter(image_describer=describer), prompt, Message(role="assistant", content="ok"))
|
||||
|
||||
assert describer.enforcer_blocked == [True]
|
||||
|
|
|
|||
|
|
@ -41,6 +41,8 @@ from litellm.proxy.guardrails.content_filter_data import DATA_ROOTS
|
|||
from litellm.proxy.guardrails.guardrail_endpoints import (
|
||||
test_custom_code_guardrail as run_custom_code_test_endpoint,
|
||||
)
|
||||
from litellm.proxy.guardrails.guardrail_hooks.azure.prompt_shield import AzureContentSafetyPromptShieldGuardrail
|
||||
from litellm.proxy.guardrails.guardrail_hooks.litellm_content_filter.content_filter import ContentFilterGuardrail
|
||||
|
||||
MOCK_ADMIN_USER = UserAPIKeyAuth(user_role=LitellmUserRoles.PROXY_ADMIN)
|
||||
from litellm.proxy.guardrails.guardrail_registry import (
|
||||
|
|
@ -3143,7 +3145,36 @@ async def test_ui_settings_map_matches_runtime_supported_event_hooks():
|
|||
continue
|
||||
|
||||
assert provider in result.supported_modes_by_provider, provider
|
||||
assert result.supported_modes_by_provider[provider] == [hook.value for hook in declared], provider
|
||||
offered = result.supported_modes_by_provider[provider]
|
||||
assert offered[: len(declared)] == [hook.value for hook in declared], provider
|
||||
assert set(offered) - {hook.value for hook in declared} <= {"logging_only"}, provider
|
||||
assert offered.count("logging_only") == (
|
||||
1 if guardrail_class.accepts_logging_only_mode() or GuardrailEventHooks.logging_only in declared else 0
|
||||
), provider
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_ui_settings_offer_logging_only_where_the_generic_scan_runs():
|
||||
modes_by_provider = (await get_guardrail_ui_settings()).supported_modes_by_provider
|
||||
|
||||
assert modes_by_provider["litellm_content_filter"] == [
|
||||
*(hook.value for hook in ContentFilterGuardrail.get_supported_event_hooks()),
|
||||
"logging_only",
|
||||
]
|
||||
assert modes_by_provider["akto"].count("logging_only") == 1
|
||||
assert modes_by_provider["azure/prompt_shield"] == [
|
||||
hook.value for hook in AzureContentSafetyPromptShieldGuardrail.get_supported_event_hooks()
|
||||
]
|
||||
with pytest.raises(ValueError, match="not in the supported event hooks"):
|
||||
AzureContentSafetyPromptShieldGuardrail(
|
||||
guardrail_name="native-logging-only",
|
||||
api_key="k",
|
||||
api_base="https://example.invalid",
|
||||
event_hook="logging_only",
|
||||
)
|
||||
assert (
|
||||
ContentFilterGuardrail(guardrail_name="cf-logging-only", event_hook="logging_only").event_hook == "logging_only"
|
||||
)
|
||||
|
||||
|
||||
def test_content_filter_runtime_rejects_unsupported_mcp_hook():
|
||||
|
|
|
|||
Loading…
Add table
Reference in a new issue