scriptoriumadmin/litellm
1
0
Fork 0
mirror of https://github.com/BerriAI/litellm.git synced 2026-09-14 23:21:35 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 156

feat(guardrails): enforce unreachable_fallback generically in the orchestration layer

Browse source 
Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com>
This commit is contained in:
yucheng 2026-08-06 03:35:51 +00:00
parent b617e672e3
commit 2e7a8d4984
6 changed files with 251 additions and 3 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

9
litellm/integrations/custom_guardrail.py
View file

@ -121,6 +121,7 @@ class CustomGuardrail(CustomLogger):
sticky_session_routing: bool = True,
run_in_parallel: bool = False,
only_scan_new_messages: bool = False,
unreachable_fallback: str | None = None,
**kwargs,
):
"""
@ -142,6 +143,10 @@ class CustomGuardrail(CustomLogger):
run_in_parallel: When True, this pre_call or post_call guardrail runs concurrently with
other opted-in guardrails of the same hook. Only safe for block-only guardrails that
do not mutate the request or response.
unreachable_fallback: 'fail_closed' (default) propagates guardrail failures that are not
deliberate blocks, 'fail_open' lets the request continue. Enforced by the proxy
guardrail orchestration for every guardrail, so individual guardrails do not need to
implement it themselves.
"""
self.guardrail_name = guardrail_name
self.supported_event_hooks = supported_event_hooks
@ -158,6 +163,10 @@ class CustomGuardrail(CustomLogger):
self.sticky_session_routing: bool = sticky_session_routing
self.run_in_parallel: bool = run_in_parallel
self.only_scan_new_messages: bool = only_scan_new_messages
if not hasattr(self, "unreachable_fallback"):
self.unreachable_fallback: Literal["fail_closed", "fail_open"] = (
"fail_open" if unreachable_fallback == "fail_open" else "fail_closed"
)
if supported_event_hooks:
## validate event_hook is in supported_event_hooks

3
litellm/proxy/guardrails/guardrail_registry.py
View file

@ -497,6 +497,9 @@ class InMemoryGuardrailHandler:
"skip_tool_message_in_guardrail",
getattr(litellm_params, "skip_tool_message_in_guardrail", None),
)
custom_guardrail_callback.unreachable_fallback = (
"fail_open" if getattr(litellm_params, "unreachable_fallback", None) == "fail_open" else "fail_closed"
)
configured_run_in_parallel: Final = getattr(litellm_params, "run_in_parallel", None)
if configured_run_in_parallel is not None:
custom_guardrail_callback.run_in_parallel = bool(configured_run_in_parallel)

35
litellm/proxy/utils.py
View file

@ -359,6 +359,35 @@ def _enrich_http_exception_with_guardrail_context(exc: BaseException, callback:
detail.setdefault("guardrail_mode", event_hook)
def _guardrail_fails_open(exc: BaseException, callback: "CustomGuardrail") -> bool:
"""
True when `exc` is a guardrail failure the orchestration layer should swallow
because the guardrail is configured with `unreachable_fallback: fail_open`.
Deliberate interventions (blocks, reroutes, passthrough responses) are always
honored, so only technical failures such as a timeout, a connection error or a
bug inside the guardrail let the request through. This is enforced here rather
than inside each guardrail so every guardrail, including custom ones, supports
fail_open / fail_closed without implementing it.
"""
if not isinstance(exc, Exception):
return False
if CustomGuardrail._is_guardrail_intervention(exc):
return False
return getattr(callback, "unreachable_fallback", "fail_closed") == "fail_open"
def _log_guardrail_fail_open(exc: BaseException, callback: "CustomGuardrail", hook_type: str) -> None:
verbose_proxy_logger.critical(
"Guardrail %s failed during %s and is configured with unreachable_fallback='fail_open', "
"allowing the request to proceed: %s: %s",
getattr(callback, "guardrail_name", None) or type(callback).__name__,
hook_type,
type(exc).__name__,
exc,
)
def _exception_changes_request_flow(exc: BaseException) -> bool:
"""
True for guardrail exceptions the proxy turns into an alternate request flow
@ -1108,6 +1137,9 @@ class ProxyLogging:
status = "error"
error_type = type(e).__name__
_enrich_http_exception_with_guardrail_context(e, callback)
if _guardrail_fails_open(e, callback):
_log_guardrail_fail_open(e, callback, "pre_call")
return None
# Re-raise the exception to maintain existing behavior
raise
finally:
@ -1626,6 +1658,9 @@ class ProxyLogging:
status = "error"
error_type = type(e).__name__
_enrich_http_exception_with_guardrail_context(e, callback)
if _guardrail_fails_open(e, callback):
_log_guardrail_fail_open(e, callback, hook_type)
return None
raise
finally:
ProxyLogging._emit_guardrail_metrics(

7
litellm/types/guardrails.py
View file

@ -848,9 +848,10 @@ class BaseLitellmParams(ContentFilterConfigModel): # works for new and patch up
unreachable_fallback: Literal["fail_closed", "fail_open"] = Field(
default="fail_closed",
description=(
"Behavior when a guardrail endpoint is unreachable due to network errors. "
"Implemented by guardrail='generic_guardrail_api', 'akto', 'vigil_guard', 'repelloai', 'headroom', and 'compresr'. "
"'fail_closed' raises an error (default). 'fail_open' logs a critical error and allows the request to proceed."
"Behavior when a guardrail fails for a reason other than a deliberate block, such as a "
"network error, a timeout, or a bug in the guardrail. Enforced by the proxy guardrail "
"orchestration for every guardrail, including custom ones. 'fail_closed' propagates the "
"error (default). 'fail_open' logs a critical error and allows the request to proceed."
),
)

40
tests/test_litellm/proxy/guardrails/test_guardrail_registry.py
View file

@ -558,3 +558,43 @@ def test_reinitialized_judge_guardrail_uses_lazy_router_provider():
finally:
for cb_list, snapshot in zip(lists, snapshots):
cb_list[:] = snapshot
@pytest.mark.parametrize(
"configured, expected",
[(None, "fail_closed"), ("fail_open", "fail_open"), ("fail_closed", "fail_closed")],
)
def test_initialize_guardrail_wires_unreachable_fallback_generically(configured, expected):
"""
Any guardrail, including ones whose constructor knows nothing about
unreachable_fallback, must end up with the configured value so the
orchestration layer can enforce fail_open / fail_closed for it.
"""
from litellm.proxy.guardrails import guardrail_registry as registry_module
def _initializer(litellm_params, guardrail):
return CustomGuardrail(
guardrail_name=guardrail["guardrail_name"],
event_hook=GuardrailEventHooks.pre_call,
default_on=True,
)
registry_module.guardrail_initializer_registry["unreachable_fallback_test"] = _initializer
try:
params = {"guardrail": "unreachable_fallback_test", "mode": "pre_call"}
if configured is not None:
params["unreachable_fallback"] = configured
handler = InMemoryGuardrailHandler()
result = handler.initialize_guardrail(
guardrail={"guardrail_name": "cf-unreachable-fallback", "litellm_params": params},
)
stored = handler.guardrail_id_to_custom_guardrail[result["guardrail_id"]]
assert stored.unreachable_fallback == expected
finally:
registry_module.guardrail_initializer_registry.pop("unreachable_fallback_test", None)
def test_custom_guardrail_defaults_to_fail_closed():
assert CustomGuardrail(guardrail_name="g").unreachable_fallback == "fail_closed"

160
tests/test_litellm/proxy/utils/proxy_logging/test_guardrail_pipeline.py
View file

@ -775,3 +775,163 @@ async def test_process_prompt_template_async_get_prompt_error_raises(proxy_loggi
prompt_version=None,
call_type="completion",
)
# ---------------------------------------------------------------------------
# generic unreachable_fallback enforcement (fail_open / fail_closed)
# ---------------------------------------------------------------------------
def _make_failing_guardrail(exc: Exception, unreachable_fallback: str):
cb = _make_guardrail()
cb.should_run_guardrail = MagicMock(return_value=True)
cb.unreachable_fallback = unreachable_fallback
cb.async_pre_call_hook = AsyncMock(side_effect=exc)
return cb
@pytest.mark.asyncio
@pytest.mark.parametrize(
"exc",
[
ConnectionError("guardrail endpoint unreachable"),
asyncio.TimeoutError(),
RuntimeError("bug inside the guardrail"),
HTTPException(status_code=500, detail={"error": "guardrail provider exploded"}),
],
)
async def test_pre_call_guardrail_failure_is_swallowed_when_fail_open(
proxy_logging, make_user_api_key_auth, exc
):
cb = _make_failing_guardrail(exc, "fail_open")
proxy_logging._should_use_guardrail_load_balancing = MagicMock(return_value=False)
out = await proxy_logging._process_guardrail_callback(
callback=cb,
data={"model": "m"},
user_api_key_dict=make_user_api_key_auth(),
call_type="completion",
event_type=GuardrailEventHooks.pre_call,
)
assert out is None
@pytest.mark.asyncio
async def test_pre_call_guardrail_failure_propagates_when_fail_closed(
proxy_logging, make_user_api_key_auth
):
cb = _make_failing_guardrail(ConnectionError("guardrail endpoint unreachable"), "fail_closed")
proxy_logging._should_use_guardrail_load_balancing = MagicMock(return_value=False)
with pytest.raises(ConnectionError):
await proxy_logging._process_guardrail_callback(
callback=cb,
data={"model": "m"},
user_api_key_dict=make_user_api_key_auth(),
call_type="completion",
event_type=GuardrailEventHooks.pre_call,
)
@pytest.mark.asyncio
@pytest.mark.parametrize(
"exc",
[
HTTPException(status_code=400, detail={"error": "violated content policy"}),
HTTPException(status_code=403, detail={"error": "blocked"}),
litellm.exceptions.GuardrailRaisedException(guardrail_name="g", message="blocked"),
SensitiveDataRouteException(
session_id="s", route_to_model="on-prem", guardrail_name="g", sticky_session_routing=True
),
ModifyResponseException(message="passthrough violation", model="m", request_data={}),
],
)
async def test_pre_call_intentional_block_still_raises_with_fail_open(
proxy_logging, make_user_api_key_auth, exc
):
cb = _make_failing_guardrail(exc, "fail_open")
proxy_logging._should_use_guardrail_load_balancing = MagicMock(return_value=False)
with pytest.raises(type(exc)):
await proxy_logging._process_guardrail_callback(
callback=cb,
data={"model": "m"},
user_api_key_dict=make_user_api_key_auth(),
call_type="completion",
event_type=GuardrailEventHooks.pre_call,
)
@pytest.mark.asyncio
@pytest.mark.parametrize("hook_type", ["during_call", "post_call"])
async def test_run_guardrail_with_metrics_fail_open_swallows_non_block(monkeypatch, hook_type):
async def task():
raise ConnectionError("guardrail endpoint unreachable")
cb = _make_guardrail()
cb.unreachable_fallback = "fail_open"
monkeypatch.setattr(litellm, "callbacks", [])
assert await ProxyLogging._run_guardrail_with_metrics(cb, task(), hook_type) is None
@pytest.mark.asyncio
@pytest.mark.parametrize("hook_type", ["during_call", "post_call"])
async def test_run_guardrail_with_metrics_fail_closed_propagates(monkeypatch, hook_type):
async def task():
raise ConnectionError("guardrail endpoint unreachable")
cb = _make_guardrail()
cb.unreachable_fallback = "fail_closed"
monkeypatch.setattr(litellm, "callbacks", [])
with pytest.raises(ConnectionError):
await ProxyLogging._run_guardrail_with_metrics(cb, task(), hook_type)
@pytest.mark.asyncio
async def test_run_guardrail_with_metrics_fail_open_keeps_block(monkeypatch):
async def task():
raise HTTPException(status_code=400, detail={"error": "blocked"})
cb = _make_guardrail()
cb.unreachable_fallback = "fail_open"
monkeypatch.setattr(litellm, "callbacks", [])
with pytest.raises(HTTPException):
await ProxyLogging._run_guardrail_with_metrics(cb, task(), "post_call")
@pytest.mark.asyncio
async def test_pre_call_hook_continues_to_next_guardrail_after_fail_open(
proxy_logging, make_user_api_key_auth, monkeypatch
):
class _FailOpenGuardrail(CustomGuardrail):
async def async_pre_call_hook(self, user_api_key_dict, cache, data, call_type):
raise ConnectionError("guardrail endpoint unreachable")
class _BlockingGuardrail(CustomGuardrail):
async def async_pre_call_hook(self, user_api_key_dict, cache, data, call_type):
raise HTTPException(status_code=400, detail={"error": "blocked by second guardrail"})
unreachable = _FailOpenGuardrail(
guardrail_name="unreachable",
event_hook=GuardrailEventHooks.pre_call,
default_on=True,
unreachable_fallback="fail_open",
)
blocking = _BlockingGuardrail(
guardrail_name="blocking", event_hook=GuardrailEventHooks.pre_call, default_on=True
)
monkeypatch.setattr(litellm, "callbacks", [unreachable, blocking])
proxy_logging._should_use_guardrail_load_balancing = MagicMock(return_value=False)
with pytest.raises(HTTPException) as exc_info:
await proxy_logging.pre_call_hook(
user_api_key_dict=make_user_api_key_auth(),
data={"model": "m", "messages": [{"role": "user", "content": "hi"}]},
call_type="completion",
)
assert exc_info.value.detail["error"] == "blocked by second guardrail"