diff --git a/litellm/proxy/guardrails/guardrail_hooks/ovalix/ovalix.py b/litellm/proxy/guardrails/guardrail_hooks/ovalix/ovalix.py index 4a20adf0e82..6644a3d3902 100644 --- a/litellm/proxy/guardrails/guardrail_hooks/ovalix/ovalix.py +++ b/litellm/proxy/guardrails/guardrail_hooks/ovalix/ovalix.py @@ -54,6 +54,7 @@ class OvalixGuardrailBlockedException(GuardrailRaisedException): guardrail_name=guardrail_name, message=message, should_wrap_with_default_message=should_wrap_with_default_message, + blocked_content=True, ) diff --git a/litellm/proxy/guardrails/guardrail_hooks/straiker/straiker.py b/litellm/proxy/guardrails/guardrail_hooks/straiker/straiker.py index bb866db4009..7cca1ae2d63 100644 --- a/litellm/proxy/guardrails/guardrail_hooks/straiker/straiker.py +++ b/litellm/proxy/guardrails/guardrail_hooks/straiker/straiker.py @@ -536,13 +536,14 @@ class StraikerGuardrail(CustomGuardrail): request_data: dict, input_type: Literal["request", "response"], message: str, + blocked_content: bool = False, ) -> NoReturn: if input_type == "request": raise GuardrailRaisedException( guardrail_name=self.guardrail_name or GUARDRAIL_NAME, message=message, should_wrap_with_default_message=False, - blocked_content=True, + blocked_content=blocked_content, ) raise ModifyResponseException( message=message, @@ -624,6 +625,7 @@ class StraikerGuardrail(CustomGuardrail): request_data=request_data, input_type=input_type, message=parsed.blocked_reason or DEFAULT_BLOCK_MESSAGE, + blocked_content=True, ) if parsed.action == "GUARDRAIL_INTERVENED": is_streamed_response: Final = input_type == "response" and _is_streamed_request(request_data) @@ -632,6 +634,7 @@ class StraikerGuardrail(CustomGuardrail): request_data=request_data, input_type=input_type, message=parsed.blocked_reason or DEFAULT_BLOCK_MESSAGE, + blocked_content=True, ) return self._intervened_inputs(inputs, parsed) return inputs diff --git a/litellm/proxy/openai_files_endpoints/batch_guardrails.py b/litellm/proxy/openai_files_endpoints/batch_guardrails.py index 237d3f6577a..f8325ff5d36 100644 --- a/litellm/proxy/openai_files_endpoints/batch_guardrails.py +++ b/litellm/proxy/openai_files_endpoints/batch_guardrails.py @@ -47,10 +47,15 @@ _SUMMARY_LIMIT: Final = 50 _SCAN_METADATA_KEY: Final = "litellm_metadata" -# `metadata` is dropped rather than diffed: guardrail dispatch writes its bookkeeping into it -# whenever the payload has one, and a record's own metadata is not scanned content on the -# online path either. -_INJECTED_KEYS: Final = frozenset({_SCAN_METADATA_KEY, "metadata"}) +# Set by pre_call_hook when a guardrail rerouted the request to a different model. +_ROUTE_APPLIED_KEY: Final = "sensitive_data_routing_applied" + +# Dropped before dispatch and restored afterwards rather than diffed. Guardrail dispatch writes +# its bookkeeping into `metadata`, and a record's own metadata is not scanned content on the +# online path either. `guardrails` is dropped because guardrail selection reads it ahead of the +# proxy-injected list, so leaving it would let a record's own body opt out of the chain its key +# and team selected; online that key can only add to the list, never replace it. +_INJECTED_KEYS: Final = frozenset({_SCAN_METADATA_KEY, "metadata", "guardrails"}) # Only what guardrail dispatch reads. The parent OTel span is deliberately left out: parenting one # guardrail span per record would put tens of thousands of spans on a single upload's trace. @@ -97,7 +102,14 @@ class UnscannableRecord: url: str | None -BatchScanFailure: TypeAlias = UnparseableRecord | UnscannableRecord +@dataclass(frozen=True, slots=True) +class UnroutableRecord: + line_number: int + custom_id: str | None + guardrail: str | None + + +BatchScanFailure: TypeAlias = UnparseableRecord | UnscannableRecord | UnroutableRecord @dataclass(frozen=True, slots=True) @@ -190,6 +202,12 @@ def raise_public(failure: BatchScanFailure) -> NoReturn: "and its body has no messages, prompt or input, so guardrails cannot read it. " "Give the record a chat, completion, embedding, responses or messages body" ) + case UnroutableRecord(line_number=line_number, custom_id=custom_id, guardrail=guardrail): + raise _rejected( + f"Batch input line {line_number}{_describe(custom_id)} was routed to a different model by " + f"{guardrail or 'a guardrail'}, and every record of a batch file goes to one provider, so " + "the file cannot be submitted. Send that record outside the batch" + ) case _: assert_never(failure) @@ -211,9 +229,16 @@ def _is_content_block(exc: BaseException) -> bool: the guardrail to fail closed, so treating it as a block would turn "refuse this request" into "drop this record and submit the rest", which is the silent loss of enforcement this whole path exists to prevent. A guardrail that does not say it blocked content aborts the upload. + + Guardrails that report a technical failure as an ``HTTPException`` carrying a block status + are caught by ``__cause__``: raising ``from`` the underlying error is a deliberate statement + that something else caused this, which a verdict on content never is. Implicit context is + left alone, since a block raised inside an unrelated ``except`` would read as a failure. """ if isinstance(exc, GuardrailRaisedException): return exc.blocked_content + if exc.__cause__ is not None: + return False return is_guardrail_intervention(exc) @@ -334,7 +359,8 @@ async def _scan_record( scan_input: Final[dict[str, object]] = copy.deepcopy(body) # mutable-ok: pre_call_hook mutates the dict it is given own_injected: Final = MappingProxyType({key: body[key] for key in _INJECTED_KEYS if key in body}) - scan_input.pop("metadata", None) + for injected in _INJECTED_KEYS: + scan_input.pop(injected, None) # Deep, and per record: `headers` and `tags` are nested containers shared with the upload # request and with every other record in the window, and a guardrail that writes into one in # place would otherwise leak across records and back into the request. The narrowing above @@ -355,6 +381,14 @@ async def _scan_record( return RecordDropped(line_number=record.line_number, custom_id=custom_id, guardrail=_naming_guardrail(exc)) raise + rerouted: Final = scanned.get("metadata") + if isinstance(rerouted, dict) and rerouted.get(_ROUTE_APPLIED_KEY): + return UnroutableRecord( + line_number=record.line_number, + custom_id=custom_id, + guardrail=rerouted.get("sensitive_data_routing_guardrail"), + ) + compared: Final = (frozenset(body) | frozenset(scanned)) - _INJECTED_KEYS if _fingerprint(scanned, compared) == _fingerprint(body, compared): return None @@ -449,14 +483,20 @@ async def scan_batch_input_file( break if not problems: await drain() + except BaseException: + redactions.close() + raise finally: file_source.seek(0) if problems: + redactions.close() worst: Final = _worst(tuple(problems)) if isinstance(worst, BaseException): raise worst return worst + if not changes: + redactions.close() return BatchScanResult( changes=tuple(sorted(changes, key=lambda change: change.line_number)), scanned_records=sum(scanned), diff --git a/tests/test_litellm/proxy/guardrails/guardrail_hooks/test_straiker.py b/tests/test_litellm/proxy/guardrails/guardrail_hooks/test_straiker.py index 36a2e205ea7..a3d86034f70 100644 --- a/tests/test_litellm/proxy/guardrails/guardrail_hooks/test_straiker.py +++ b/tests/test_litellm/proxy/guardrails/guardrail_hooks/test_straiker.py @@ -1067,3 +1067,29 @@ async def test_anthropic_non_streaming_response_reports_usage(): payload = _posted_payload(g) assert payload["usage"] == {"input_tokens": 10, "output_tokens": 5} assert payload["response"]["finish_reason"] == "end_turn" + + +def test_fail_closed_backend_failure_is_not_reported_as_a_content_verdict(): + """A drop-one-record consumer must be able to tell a verdict from an outage; _fail is not a verdict.""" + from litellm.exceptions import GuardrailRaisedException + + guardrail = _make_guardrail() + + with pytest.raises(GuardrailRaisedException) as unreachable: + guardrail._fail( + inputs={}, + request_data={"model": "m"}, + input_type="request", + error="connection refused", + is_unreachable=True, + ) + assert unreachable.value.blocked_content is False + + with pytest.raises(GuardrailRaisedException) as verdict: + guardrail._block( + request_data={"model": "m"}, + input_type="request", + message="blocked", + blocked_content=True, + ) + assert verdict.value.blocked_content is True diff --git a/tests/test_litellm/proxy/openai_files_endpoint/test_batch_guardrails.py b/tests/test_litellm/proxy/openai_files_endpoint/test_batch_guardrails.py index 5715f2558b3..758f213b081 100644 --- a/tests/test_litellm/proxy/openai_files_endpoint/test_batch_guardrails.py +++ b/tests/test_litellm/proxy/openai_files_endpoint/test_batch_guardrails.py @@ -672,6 +672,143 @@ async def test_an_unreachable_guardrail_aborts_instead_of_quietly_dropping_the_r await _scan_full(_jsonl(_record("a"), _record("b", content="tripwire")), FakeProxyLogging(_hook)) +@pytest.mark.asyncio +async def test_a_guardrail_subclass_that_blocks_content_drops_only_that_record(): + """A subclass has to opt in too, or a real block takes the whole upload down with it.""" + from litellm.proxy.guardrails.guardrail_hooks.ovalix.ovalix import OvalixGuardrailBlockedException + + def _hook(data): + if "tripwire" in data["messages"][0]["content"]: + raise OvalixGuardrailBlockedException(guardrail_name="ovalix", message="blocked") + + result = await _scan_full(_jsonl(_record("a"), _record("b", content="tripwire")), FakeProxyLogging(_hook)) + + assert result.changes == (RecordDropped(line_number=2, custom_id="b", guardrail="ovalix"),) + assert result.submitted_records == 1 + + +@pytest.mark.asyncio +async def test_a_record_a_guardrail_rerouted_aborts_rather_than_shipping_to_the_original_provider(): + """pre_call_hook honours a reroute by rewriting `model`; a batch file cannot follow it.""" + from litellm.proxy.openai_files_endpoints.batch_guardrails import UnroutableRecord + + def _hook(data): + if "tripwire" in data["messages"][0]["content"]: + data["model"] = "on-prem-model" + data["metadata"] = { + "sensitive_data_routing_applied": True, + "sensitive_data_routing_guardrail": "router-guard", + } + + failure = await _scan(_jsonl(_record("a"), _record("b", content="tripwire")), FakeProxyLogging(_hook)) + + assert failure == UnroutableRecord(line_number=2, custom_id="b", guardrail="router-guard") + with pytest.raises(HTTPException) as caught: + raise_public(failure) + assert "routed to a different model" in str(caught.value.detail) + + +@pytest.mark.asyncio +async def test_the_scan_spool_is_closed_when_nothing_will_read_it(): + """The spool is opened for every scan, so a clean file must not leave a temp handle behind.""" + result = await _scan_full(_jsonl(_record("a")), FakeProxyLogging()) + + assert result.changes == () + assert result.redactions.closed + + +@pytest.mark.asyncio +async def test_the_scan_spool_is_closed_when_the_upload_is_refused(): + def _hook(data): + if "tripwire" in data["messages"][0]["content"]: + raise RuntimeError("infrastructure is down") + + source = _jsonl(_record("a"), _record("b", content="tripwire")) + spools = [] + import litellm.proxy.openai_files_endpoints.batch_guardrails as bg + + real = bg.tempfile.SpooledTemporaryFile + + def _tracking(*args, **kwargs): + handle = real(*args, **kwargs) + spools.append(handle) + return handle + + bg.tempfile.SpooledTemporaryFile = _tracking + try: + with pytest.raises(RuntimeError): + await _scan_full(source, FakeProxyLogging(_hook)) + finally: + bg.tempfile.SpooledTemporaryFile = real + + assert spools and all(handle.closed for handle in spools) + + +@pytest.mark.asyncio +async def test_the_scan_spool_is_closed_when_a_record_escapes_the_iterator(): + """A raise from inside the read loop bypasses the per-record outcome path entirely.""" + import litellm.proxy.openai_files_endpoints.batch_guardrails as bg + + spools = [] + real = bg.tempfile.SpooledTemporaryFile + + def _tracking(*args, **kwargs): + handle = real(*args, **kwargs) + spools.append(handle) + return handle + + bg.tempfile.SpooledTemporaryFile = _tracking + try: + with pytest.raises(json.JSONDecodeError): + await _scan_full(io.BytesIO(b"{not json at all}\n"), FakeProxyLogging()) + finally: + bg.tempfile.SpooledTemporaryFile = real + + assert spools and all(handle.closed for handle in spools) + + +@pytest.mark.asyncio +async def test_a_technical_failure_dressed_as_a_block_status_still_aborts(): + """xecguard and purview report an unreachable backend as HTTPException(400) under fail-closed.""" + + def _hook(data): + if "tripwire" in data["messages"][0]["content"]: + try: + raise ConnectionError("backend unreachable") + except ConnectionError as exc: + raise HTTPException( + status_code=400, detail={"error": "XecGuard API unreachable (block_on_error=True)"} + ) from exc + + with pytest.raises(HTTPException): + await _scan_full(_jsonl(_record("a"), _record("b", content="tripwire")), FakeProxyLogging(_hook)) + + +@pytest.mark.asyncio +async def test_a_record_body_cannot_opt_itself_out_of_the_guardrail_chain(): + """Guardrail selection reads a body-level `guardrails` key first; online it can only add.""" + seen = [] + + await _scan_full( + _jsonl({**_record("a"), "body": {**_record("a")["body"], "guardrails": []}}), + FakeProxyLogging(lambda d: seen.append(sorted(d))), + metadata={"guardrails": ["team-guard"]}, + ) + + assert seen and "guardrails" not in seen[0] + + +@pytest.mark.asyncio +async def test_a_redacted_record_keeps_its_own_guardrails_key(): + """Stripping it for the scan must not rewrite what the caller asked the provider to run.""" + record = _record("m", content="my secret is here") + record["body"]["guardrails"] = ["extra-guard"] + + body = await _rewritten_body(record, _redact_containing("secret")) + + assert body["guardrails"] == ["extra-guard"] + + @pytest.mark.asyncio async def test_a_400_that_is_not_a_guardrail_decision_still_aborts(): """A guardrail's own HTTP client can raise a 400 because OUR payload was rejected, not the content."""