This commit is contained in:
Ninad Phalak 2026-09-05 20:48:22 -04:00 committed by GitHub
commit b696f17714
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
13 changed files with 1714 additions and 2 deletions

View file

@ -0,0 +1,33 @@
from typing import TYPE_CHECKING, Final
from litellm.types.guardrails import SupportedGuardrailIntegrations
from .llm_shield_proxy import LLMShieldProxyGuardrail
if TYPE_CHECKING:
from litellm.types.guardrails import Guardrail, LitellmParams
def initialize_guardrail(litellm_params: "LitellmParams", guardrail: "Guardrail") -> LLMShieldProxyGuardrail:
import litellm
_llm_shield_guardrail_callback: Final = LLMShieldProxyGuardrail(
api_key=litellm_params.api_key,
api_base=litellm_params.api_base,
guardrail_name=guardrail.get("guardrail_name", ""),
event_hook=litellm_params.mode,
default_on=litellm_params.default_on,
)
litellm.logging_callback_manager.add_litellm_callback(_llm_shield_guardrail_callback)
return _llm_shield_guardrail_callback
guardrail_initializer_registry: Final = { # mutable-ok: module-level registry, built once and never mutated
SupportedGuardrailIntegrations.LLM_SHIELD_PROXY.value: initialize_guardrail,
}
guardrail_class_registry: Final = { # mutable-ok: module-level registry, built once and never mutated
SupportedGuardrailIntegrations.LLM_SHIELD_PROXY.value: LLMShieldProxyGuardrail,
}

View file

@ -0,0 +1,57 @@
# Example LiteLLM Proxy configuration for LLM Shield Proxy
# LLM Shield Proxy is a self-hosted PII gateway: https://github.com/ninadphalak/LLM-Shield-Proxy
#
# Unlike a masking guardrail, LLM Shield Proxy's substitution is reversible. Personal data is
# replaced with placeholders before the request goes to the provider, and the original
# values are put back into the model's reply, so the end user still sees real data while
# the provider never received it.
model_list:
- model_name: gpt-4o
litellm_params:
model: openai/gpt-4o
api_key: os.environ/OPENAI_API_KEY
guardrails:
# Both modes belong on ONE entry. pre_call redacts the outbound request and post_call
# restores the reply; listing only pre_call would send placeholders back to the user.
- guardrail_name: "llm_shield_proxy"
litellm_params:
guardrail: llm_shield_proxy
mode: ["pre_call", "post_call"]
default_on: true
# Your own LLM Shield Proxy deployment. Defaults to http://localhost:8000, and also reads
# LLM_SHIELD_PROXY_API_BASE from the environment.
api_base: "http://localhost:8000"
# A virtual key configured on that deployment. Also reads LLM_SHIELD_PROXY_API_KEY.
api_key: os.environ/LLM_SHIELD_PROXY_API_KEY
# Usage:
#
# 1. Run LLM Shield Proxy somewhere the proxy can reach:
# pip install llm-shield-proxy
# llm-shield-proxy --port 8000
#
# 2. Point this config at it and start the proxy:
# export LLM_SHIELD_PROXY_API_KEY="your-virtual-key"
# litellm --config example_config.yaml
#
# 3. Send a request containing personal data:
# curl http://localhost:4000/v1/chat/completions \
# -H "Authorization: Bearer sk-1234" \
# -H "Content-Type: application/json" \
# -d '{"model":"gpt-4o","messages":[{"role":"user","content":"Email jane.doe@example.com the invoice"}]}'
#
# The provider receives a stand-in value in place of the address. The reply you get
# back carries the real address again.
#
# Notes:
#
# - Requests are refused if LLM Shield Proxy is unreachable or returns an error, rather than
# being forwarded. Sending them on would hand the provider exactly the data this
# guardrail exists to withhold.
# - Restoring a value requires the request and the reply to share a session. LiteLLM's
# session id is used when present; otherwise one is generated per request.
# - Streaming replies are restored as chunks arrive. A placeholder split across two
# chunks is held back until it is complete, so partial values are never emitted.
# - Only text is redacted; images and audio pass through untouched.

View file

@ -0,0 +1,660 @@
# +-------------------------------------------------------------+
#
# Use LLM Shield Proxy for reversible PII redaction
# https://github.com/ninadphalak/LLM-Shield-Proxy
#
# +-------------------------------------------------------------+
import os
import uuid
from collections.abc import AsyncGenerator, Callable, Mapping, Sequence
from typing import (
TYPE_CHECKING,
Any, # noqa: TID251 # **kwargs forwards verbatim to CustomGuardrail.__init__
ClassVar,
Final,
Literal,
Optional,
TypeAlias,
)
import httpx
from litellm._logging import verbose_proxy_logger
from litellm.exceptions import GuardrailRaisedException
from litellm.integrations.custom_guardrail import (
CustomGuardrail,
log_guardrail_information,
)
from litellm.llms.custom_httpx.http_handler import (
get_async_httpx_client,
httpxSpecialProvider,
)
from litellm.proxy._types import UserAPIKeyAuth
from litellm.types.guardrails import GuardrailEventHooks
from litellm.types.utils import GenericGuardrailAPIInputs
if TYPE_CHECKING:
from litellm.caching.caching import DualCache
from litellm.litellm_core_utils.litellm_logging import Logging as LiteLLMLoggingObj
GUARDRAIL_NAME: Final = "llm_shield_proxy"
_DEFAULT_API_BASE: Final = "http://localhost:8000"
_REDACT_PATH: Final = "/v1/guard/redact"
_REHYDRATE_PATH: Final = "/v1/guard/rehydrate"
_REHYDRATE_STREAM_PATH: Final = "/v1/guard/rehydrate/stream"
# The session id ties a redact call to the rehydrate calls that undo it. It is
# stored on the request dict rather than on the guardrail instance: the proxy
# registers one instance process-wide, so instance attributes would be shared
# across concurrent requests.
_SESSION_METADATA_KEY: Final = "llm_shield_session_id"
# Roles whose text the application author wrote and the caller never sees. Their
# PII is still redacted outbound, but it is not restorable from the reply.
_PRIVILEGED_ROLES: Final = frozenset({"system", "developer"})
# Vault ids are minted here and never derived from anything the caller sends. The
# vault holds the plaintext behind every placeholder, so an id a caller could
# supply or guess would let one user rehydrate another user's values by getting a
# placeholder echoed back. The per-process prefix means a caller cannot even name
# a vault this process uses.
_VAULT_PREFIX: Final = f"litellm-{uuid.uuid4().hex}"
_DEFAULT_TIMEOUT_SECONDS: Final = 10.0
# The proxy's own request dict. Mutable by design: a pre-call guardrail rewrites
# the caller's payload in place, which is the entire point of the hook.
# mutable-ok: the shape is fixed by CustomLogger's hook signatures.
MutableRequest: TypeAlias = dict
# A JSON body on its way to httpx, which requires a real dict rather than a view.
# mutable-ok: handed straight to the HTTP client.
JsonBody: TypeAlias = dict
# One redactable span: the text as it stands, and the write that puts the
# replacement back where it came from.
# How far a tool_result chain is followed. Real payloads nest one or two deep; the
# bound is what stops a crafted one from becoming an unbounded walk.
_MAX_CONTENT_DEPTH: Final = 8
_Slot: TypeAlias = tuple[str, Callable[[str], None]] # mutable-ok: Callable's param list.
# The accumulator the collectors below append into. It never escapes
# _locate_request_texts, which freezes it into a tuple before returning.
_SlotSink: TypeAlias = list[_Slot] # mutable-ok: accumulator passed between collectors.
# Sliding windows keyed by streaming choice index, threaded through one stream.
_CarryWindows: TypeAlias = dict # mutable-ok: per-choice windows advanced in place.
# A caller-owned list whose entries are rewritten in place, such as a Completions
# `prompt` sent as an array of strings.
MutableSeq: TypeAlias = list # mutable-ok: the request payload's own list.
def _collect(container: MutableRequest, key: str, slots: _SlotSink) -> None:
"""Records the string at `key`, along with the write that replaces it."""
value: Final = container.get(key)
if isinstance(value, str) and value:
slots.append((value, lambda new, c=container, k=key: c.__setitem__(k, new)))
def _collect_entry(entries: MutableSeq, index: int, slots: _SlotSink) -> None:
"""Records a string held directly in a list, rather than under a key."""
value: Final = entries[index]
if isinstance(value, str) and value:
slots.append((value, lambda new, e=entries, i=index: e.__setitem__(i, new)))
def _collect_prompt(data: MutableRequest, slots: _SlotSink) -> None:
"""The Completions API sends its text in `prompt`, and its tail in `suffix`."""
_collect(data, "suffix", slots)
prompt: Final = data.get("prompt")
if isinstance(prompt, str):
_collect(data, "prompt", slots)
return
if isinstance(prompt, dict):
# A Responses API PromptObject. `variables` are substituted into the stored
# prompt on the provider side, so they are caller text. `id` and `version`
# identify which prompt to use and must arrive unchanged.
variables: Final = prompt.get("variables")
if isinstance(variables, dict):
for name in tuple(variables):
_collect(variables, name, slots)
return
if not isinstance(prompt, list):
return
for index in range(len(prompt)):
_collect_entry(prompt, index, slots)
def _collect_content(container: MutableRequest, slots: _SlotSink) -> None:
"""Collects `content`, a string or a list of typed parts.
An Anthropic tool_result nests its own content, so this has to descend. It walks
with an explicit stack and a depth bound rather than by recursion: the nesting is
caller controlled, and an unbounded descent is a JSON bomb.
"""
# Walked in document order: the shield maps its replies back by position, so the
# order spans are collected in is part of the contract.
pending: Final[list] = [(container, 0)] # mutable-ok: local queue, never escapes.
cursor = 0 # rebind-ok: advances through the queue.
while cursor < len(pending):
node, depth = pending[cursor]
cursor += 1
content = node.get("content")
if isinstance(content, str):
_collect(node, "content", slots)
continue
if depth >= _MAX_CONTENT_DEPTH:
continue
for part in content if isinstance(content, list) else ():
if not isinstance(part, dict):
continue
# Image and audio parts have no text and fall through untouched.
_collect(part, "text", slots)
if "content" in part:
pending.append((part, depth + 1))
def _collect_participant_name(message: MutableRequest, slots: _SlotSink) -> None:
"""Redacts `name` where it identifies a person, never where it names a function.
On a user or assistant turn `name` is the participant, which is personal data.
On a tool or function turn the same field carries the function's name and has
to reach the provider unchanged, or the call no longer routes.
"""
if message.get("role") in ("tool", "function"):
return
_collect(message, "name", slots)
def _collect_tool_arguments(message: MutableRequest, slots: _SlotSink) -> None:
"""Tool arguments carry the values a user asked the model to act on."""
for tool_call in message.get("tool_calls") or ():
function = tool_call.get("function") if isinstance(tool_call, dict) else None # rebind-ok: loop variable.
if isinstance(function, dict):
_collect(function, "arguments", slots)
legacy: Final = message.get("function_call")
if isinstance(legacy, dict):
_collect(legacy, "arguments", slots)
def _collect_system(data: MutableRequest, slots: _SlotSink) -> None:
"""Anthropic's /v1/messages carries its system prompt at the top level."""
system: Final = data.get("system")
if isinstance(system, str):
_collect(data, "system", slots)
return
for part in system if isinstance(system, list) else ():
if isinstance(part, dict):
_collect(part, "text", slots)
def _collect_responses_fields(data: MutableRequest, slots: _SlotSink, privileged: _SlotSink) -> None:
"""The Responses API sends text outside `messages`, in `instructions` and `input`.
`instructions` is written by the application, not by the caller, so it is
collected into the privileged sink; `input` is the caller's own text.
"""
_collect(data, "instructions", privileged)
request_input: Final = data.get("input")
if isinstance(request_input, str):
_collect(data, "input", slots)
return
if not isinstance(request_input, list):
return
for index, item in enumerate(request_input):
if isinstance(item, str):
# The embeddings and moderations shape: `input` as an array of strings.
_collect_entry(request_input, index, slots)
continue
if not isinstance(item, dict):
continue
_collect_content(item, slots)
# A function_call item holds `arguments`; a function_call_output holds `output`.
_collect(item, "arguments", slots)
_collect(item, "output", slots)
def _choice_index(choice: object) -> int:
"""Streaming choices are matched across chunks by their index."""
index: Final = getattr(choice, "index", 0)
return index if isinstance(index, int) else 0
class LLMShieldProxyGuardrail(CustomGuardrail):
"""Redacts PII before it leaves the proxy and restores it in the response.
Unlike a masking guardrail, the substitution is reversible. Outbound text is
replaced with placeholders held in a session vault inside the user's own LLM
Shield deployment; the model's reply is then restored so the end user sees the
original values while the provider never received them.
Streaming is restored incrementally rather than by buffering the response. LLM
Shield holds back only the trailing characters that could still turn out to be
part of a placeholder, so tokens are forwarded as they arrive and a placeholder
split across two chunks is never emitted in fragments.
"""
# Our redaction and restoration run in the native lifecycle hooks below. Without
# this the proxy would route every event through the unified apply_guardrail path
# and the streaming hook would never fire.
use_native_lifecycle_hooks: ClassVar[bool] = True
def __init__(
self,
guardrail_name: str = GUARDRAIL_NAME,
api_base: str | None = None,
api_key: str | None = None,
**kwargs: Any, # noqa: LIT008 # kwargs-ok: forwarded verbatim to CustomGuardrail.__init__
) -> None:
self.async_handler = get_async_httpx_client(llm_provider=httpxSpecialProvider.GuardrailCallback)
env_base: Final = os.environ.get("LLM_SHIELD_PROXY_API_BASE")
self.api_base: Final = (api_base or env_base or _DEFAULT_API_BASE).rstrip("/")
self.api_key: Final = api_key or os.environ.get("LLM_SHIELD_PROXY_API_KEY")
super().__init__(guardrail_name=guardrail_name, **kwargs)
@classmethod
def get_supported_event_hooks(cls) -> list[GuardrailEventHooks]: # mutable-ok: parent's signature.
return [GuardrailEventHooks.pre_call, GuardrailEventHooks.post_call] # mutable-ok: parent's signature.
# --- transport ---------------------------------------------------------------
def _headers(self, session_id: str) -> JsonBody:
headers: Final[JsonBody] = { # mutable-ok: httpx requires a real dict.
"Content-Type": "application/json",
"X-Session-ID": session_id,
}
if self.api_key:
headers["Authorization"] = f"Bearer {self.api_key}"
return headers
async def _call_shield(self, path: str, session_id: str, payload: JsonBody) -> Mapping[str, object]:
"""Posts to LLM Shield Proxy, failing closed on any transport or status error.
A redaction guardrail that fails open sends the very data it exists to
protect to a third-party provider, so an unreachable or erroring shield
blocks the request instead of passing it through.
"""
try:
response: Final = await self.async_handler.post(
f"{self.api_base}{path}",
headers=self._headers(session_id),
json=payload,
timeout=_DEFAULT_TIMEOUT_SECONDS,
)
response.raise_for_status()
return response.json()
except httpx.HTTPStatusError as exc:
verbose_proxy_logger.exception("LLM Shield Proxy returned %s for %s", exc.response.status_code, path)
raise GuardrailRaisedException(
guardrail_name=self.guardrail_name,
message=f"LLM Shield Proxy returned {exc.response.status_code}; blocking the request.",
) from exc
except Exception as exc:
verbose_proxy_logger.exception("LLM Shield Proxy call to %s failed", path)
raise GuardrailRaisedException(
guardrail_name=self.guardrail_name,
message="LLM Shield Proxy is unreachable; blocking the request.",
) from exc
async def _redact(self, texts: Sequence[str], session_id: str) -> Sequence[str]:
payload: Final[JsonBody] = {"texts": list(texts)} # mutable-ok: JSON body for httpx.
body: Final = await self._call_shield(_REDACT_PATH, session_id, payload)
return self._same_length_or_raise(body.get("texts"), texts, "redact")
async def _rehydrate(self, texts: Sequence[str], session_id: str) -> Sequence[str]:
payload: Final[JsonBody] = {"texts": list(texts)} # mutable-ok: JSON body for httpx.
body: Final = await self._call_shield(_REHYDRATE_PATH, session_id, payload)
return self._same_length_or_raise(body.get("texts"), texts, "rehydrate")
def _same_length_or_raise(self, returned: object, sent: Sequence[str], operation: str) -> Sequence[str]:
"""Guards the positional mapping the callers rely on to write results back."""
if not isinstance(returned, list) or len(returned) != len(sent):
raise GuardrailRaisedException(
guardrail_name=self.guardrail_name,
message=f"LLM Shield Proxy {operation} returned an unexpected payload; blocking the request.",
)
return tuple(returned)
# --- session ------------------------------------------------------------------
@staticmethod
def _mint_session_id(data: MutableRequest) -> str:
"""Mints a vault id for this request, overwriting anything already there.
Redaction and restoration both happen inside one request/response pair, so
a fresh id per request is all that is needed, and it is what keeps one
caller from reaching another caller's vault.
"""
session_id: Final = f"{_VAULT_PREFIX}-{uuid.uuid4().hex}"
metadata: Final = data.setdefault("metadata", {}) # mutable-ok: per-request store.
if isinstance(metadata, dict):
metadata[_SESSION_METADATA_KEY] = session_id
return session_id
@staticmethod
def _session_id(data: MutableRequest) -> str:
"""Reads back the vault id minted while redacting this request.
Falls back to an unused id rather than to anything the caller supplied: a
reply that cannot be restored is a visible placeholder, while trusting a
caller-supplied id would hand them someone else's plaintext.
"""
metadata: Final = data.get("metadata")
existing: Final = metadata.get(_SESSION_METADATA_KEY) if isinstance(metadata, dict) else None
if isinstance(existing, str) and existing.startswith(_VAULT_PREFIX):
return existing
return f"{_VAULT_PREFIX}-{uuid.uuid4().hex}"
# --- request traversal --------------------------------------------------------
@staticmethod
def _locate_request_texts(
data: MutableRequest,
) -> tuple[Sequence[_Slot], Sequence[_Slot]]:
"""Finds every redactable span, split by whether the caller can see it.
Anything missed here reaches the provider in the clear while the guardrail
still reports as enabled, so the walk covers every request shape that
carries text.
The split exists because the response is restored against one vault only.
Server-authored spans -- system and developer turns, Anthropic's top-level
`system`, the Responses API `instructions` -- go into a vault nothing is
ever restored against, so a caller who gets the model to echo one of their
placeholders back receives the placeholder, not the value behind it.
"""
slots: Final[_SlotSink] = [] # mutable-ok: accumulator, frozen on return.
privileged: Final[_SlotSink] = [] # mutable-ok: accumulator, frozen on return.
for message in data.get("messages") or ():
if isinstance(message, dict):
sink = privileged if message.get("role") in _PRIVILEGED_ROLES else slots
_collect_content(message, sink)
_collect_participant_name(message, sink)
_collect_tool_arguments(message, sink)
_collect_responses_fields(data, slots, privileged)
_collect_prompt(data, slots)
_collect_system(data, privileged)
return tuple(slots), tuple(privileged)
# --- hooks --------------------------------------------------------------------
@log_guardrail_information
async def async_pre_call_hook(
self,
user_api_key_dict: UserAPIKeyAuth,
cache: "DualCache",
data: MutableRequest,
call_type: str,
) -> MutableRequest | None:
"""Replaces PII anywhere in the outbound request with vault placeholders."""
if self.should_run_guardrail(data=data, event_type=GuardrailEventHooks.pre_call) is not True:
return data
slots, privileged = self._locate_request_texts(data)
if not slots and not privileged:
return data
session_id: Final = self._mint_session_id(data)
if privileged:
# A vault of its own, whose id is deliberately never stored: the
# response is restored against `session_id` alone, so nothing the
# model emits can turn one of these placeholders back into plaintext.
await self._redact_into(privileged, f"{_VAULT_PREFIX}-{uuid.uuid4().hex}")
if slots:
await self._redact_into(slots, session_id)
return data
async def _redact_into(self, slots: Sequence[_Slot], session_id: str) -> None:
"""Redacts every span in `slots` under one vault and writes the result back."""
redacted: Final = await self._redact(tuple(text for text, _ in slots), session_id)
for (_, write), replacement in zip(slots, redacted):
write(replacement)
@log_guardrail_information
async def async_post_call_success_hook(
self,
data: MutableRequest,
user_api_key_dict: UserAPIKeyAuth,
response: Any,
) -> Any:
"""Restores the original values in a non-streaming response."""
if self.should_run_guardrail(data=data, event_type=GuardrailEventHooks.post_call) is not True:
return response
if self._is_anthropic_message_response(response):
return await self._restore_anthropic_response(response, data)
text_blocks: Final = self._responses_api_text_blocks(response)
if text_blocks:
return await self._restore_responses_api_response(response, text_blocks, data)
choices: Final = getattr(response, "choices", None)
if not choices:
return response
pending: Final = tuple(
(choice.message, choice.message.content)
for choice in choices
if getattr(choice, "message", None) is not None
and isinstance(getattr(choice.message, "content", None), str)
and choice.message.content
)
if not pending:
return response
restored: Final = await self._rehydrate(tuple(text for _, text in pending), self._session_id(data))
for (message, _), replacement in zip(pending, restored):
message.content = replacement
return response
@staticmethod
def _is_anthropic_message_response(response: object) -> bool:
"""Anthropic's native /v1/messages reply arrives as a plain dict."""
return (
isinstance(response, dict)
and response.get("type") == "message"
and isinstance(response.get("content"), list)
)
async def _restore_anthropic_response(self, response: MutableRequest, data: MutableRequest) -> MutableRequest:
"""Restores text blocks in an Anthropic native message reply.
This shape has no `choices`, so without its own branch the reply would go
back to the caller still carrying placeholders.
"""
blocks: Final = tuple(
block
for block in response["content"]
if isinstance(block, dict) and block.get("type") == "text" and isinstance(block.get("text"), str)
)
if not blocks:
return response
restored: Final = await self._rehydrate(tuple(block["text"] for block in blocks), self._session_id(data))
for block, replacement in zip(blocks, restored):
block["text"] = replacement
return response
@staticmethod
def _responses_api_text_blocks(response: object) -> Sequence[object]:
"""Text blocks in a Responses API reply.
That shape carries `output` items rather than `choices`, so it needs its own
walk; without one the reply goes back to the caller still holding
placeholders even though the request was redacted correctly. Blocks come
through as dicts or as objects depending on how far the reply has been
deserialised, so both are handled.
"""
blocks: Final[list[object]] = [] # mutable-ok: accumulator, frozen on return.
for item in getattr(response, "output", None) or ():
for block in getattr(item, "content", None) or ():
if isinstance(block, dict):
if isinstance(block.get("text"), str) and block["text"]:
blocks.append(block)
elif isinstance(getattr(block, "text", None), str) and block.text:
blocks.append(block)
return tuple(blocks)
@staticmethod
def _block_text(block: object) -> str:
return block["text"] if isinstance(block, dict) else block.text
async def _restore_responses_api_response(
self, response: Any, blocks: Sequence[object], data: MutableRequest
) -> Any:
"""Puts the original values back into a Responses API reply."""
restored: Final = await self._rehydrate(
tuple(self._block_text(block) for block in blocks), self._session_id(data)
)
for block, replacement in zip(blocks, restored):
if isinstance(block, dict):
block["text"] = replacement
else:
block.text = replacement
return response
async def async_post_call_streaming_iterator_hook(
self,
user_api_key_dict: UserAPIKeyAuth,
response: Any,
request_data: MutableRequest,
) -> AsyncGenerator[Any, None]:
"""Restores original values incrementally, without buffering the stream.
Each choice is its own token stream, so the sliding window is tracked per
choice index. One shared window would splice the characters held back for
one choice onto the next. The windows are locals of this generator, so they
are scoped to a single stream and cannot leak between concurrent requests.
"""
if self.should_run_guardrail(data=request_data, event_type=GuardrailEventHooks.post_call) is not True:
async for chunk in response:
yield chunk
return
session_id: Final = self._session_id(request_data)
carries: Final[dict] = {} # mutable-ok: per-choice windows, local to this stream.
last_chunk = None # rebind-ok: tracks the most recent chunk for the final flush.
async for chunk in response:
last_chunk = chunk
for choice in getattr(chunk, "choices", None) or ():
await self._restore_choice(choice, carries, session_id)
yield chunk
# A stream that ended without a finish_reason can still leave text held back.
if last_chunk is not None and any(carries.values()):
async for trailing in self._flush_trailing(last_chunk, carries, session_id):
yield trailing
async def _restore_choice(self, choice: Any, carries: _CarryWindows, session_id: str) -> None:
"""Restores one choice's delta, advancing that choice's own window."""
delta: Final = getattr(choice, "delta", None)
if delta is None:
return
index: Final = _choice_index(choice)
carry: Final = carries.get(index, "")
text: Final = getattr(delta, "content", None)
is_final: Final = bool(getattr(choice, "finish_reason", None))
if not isinstance(text, str) or not text:
# Nothing to restore here, but a final chunk still has to flush the window.
if is_final and carry:
flushed, flushed_carry = await self._stream_step("", carry, True, session_id)
carries[index] = flushed_carry # rebind-ok: this choice's window advances.
if flushed:
delta.content = flushed
return
emitted, remaining = await self._stream_step(text, carry, is_final, session_id)
carries[index] = remaining # rebind-ok: this choice's window advances.
delta.content = emitted
async def _flush_trailing(
self, last_chunk: Any, carries: _CarryWindows, session_id: str
) -> AsyncGenerator[Any, None]:
"""Empties every window still holding text, one chunk per choice.
Driven by the windows rather than by the last chunk's choices. A choice that
finished earlier is not present in the terminal chunk, and flushing only what
that chunk carries would drop its held text and truncate its answer.
"""
for index in sorted(carries):
carry = carries[index]
if not carry:
continue
text, remaining = await self._stream_step("", carry, True, session_id)
carries[index] = remaining # rebind-ok: this choice's window advances.
if not text:
continue
chunk = self._chunk_for_choice(last_chunk, index)
if chunk is None:
continue
chunk.choices[0].delta.content = text
yield chunk
@staticmethod
def _chunk_for_choice(last_chunk: Any, index: int) -> Any:
"""A single-choice copy of the last chunk, carrying only `index`.
Emitting one choice per chunk keeps a flush from reading as content on a
choice it does not belong to.
"""
chunk: Final = last_chunk.model_copy(deep=True)
raw_choices: Final = getattr(chunk, "choices", None)
if not raw_choices:
return None
choices: Final[tuple] = tuple(raw_choices)
matching: Final = tuple(choice for choice in choices if _choice_index(choice) == index)
kept: Final = matching[0] if matching else choices[0]
if getattr(kept, "delta", None) is None:
return None
kept.index = index
# The terminal signal, if there was one, already went out with the real chunk.
kept.finish_reason = None
chunk.choices = [kept] # mutable-ok: the chunk model requires a list.
return chunk
async def _stream_step(self, text: str, carry: str, final: bool, session_id: str) -> tuple[str, str]:
"""Returns ``(text safe to emit now, window still being held)``."""
body: Final = await self._call_shield(
_REHYDRATE_STREAM_PATH,
session_id,
# mutable-ok: JSON request body for httpx.
{"text": text, "carry": carry, "final": final}, # mutable-ok: JSON request body for httpx.
)
emitted: Final = body.get("text")
remaining: Final = body.get("carry")
if not isinstance(emitted, str) or not isinstance(remaining, str):
raise GuardrailRaisedException(
guardrail_name=self.guardrail_name,
message="LLM Shield Proxy stream rehydration returned an unexpected payload.",
)
return emitted, remaining
# --- unified API (powers the UI "Test guardrail" button) -----------------------
@log_guardrail_information
async def apply_guardrail(
self,
inputs: GenericGuardrailAPIInputs,
request_data: MutableRequest,
input_type: Literal["request", "response"],
logging_obj: Optional["LiteLLMLoggingObj"] = None,
) -> GenericGuardrailAPIInputs:
texts: Final = inputs.get("texts")
if not texts:
return inputs
replaced: Final = (
await self._redact(tuple(texts), self._mint_session_id(request_data))
if input_type == "request"
else await self._rehydrate(tuple(texts), self._session_id(request_data))
)
# Return a new mapping rather than rewriting the caller's, so this stays a
# pure transform of the inputs it was handed.
merged: Final[JsonBody] = {**inputs, "texts": list(replaced)} # mutable-ok: TypedDict.
return merged

View file

@ -137,6 +137,7 @@ class SupportedGuardrailIntegrations(Enum):
COMPRESR = "compresr"
STRAIKER = "straiker"
ALICE = "alice"
LLM_SHIELD_PROXY = "llm_shield_proxy"
class Role(Enum):

View file

@ -0,0 +1,24 @@
from pydantic import Field
from .base import GuardrailConfigModel
class LLMShieldProxyGuardrailConfigModel(GuardrailConfigModel):
api_key: str | None = Field(
default=None,
description=(
"The virtual key for the LLM Shield Proxy instance. If not provided, the "
"`LLM_SHIELD_PROXY_API_KEY` environment variable is checked."
),
)
api_base: str | None = Field(
default=None,
description=(
"The base URL of the LLM Shield Proxy instance. If not provided, the `LLM_SHIELD_PROXY_API_BASE` "
"environment variable is checked, then `http://localhost:8000`."
),
)
@staticmethod
def ui_friendly_name() -> str:
return "LLM Shield Proxy"

View file

@ -30,6 +30,10 @@ external = [
# grows over time; typing it concretely (`object`) broke that forwarding call outright —
# basedpyright turned every named param into a reportArgumentType error. Any is correct here.
"litellm/proxy/guardrails/guardrail_hooks/alice/alice.py" = ["ANN401"]
# Same reason: `**kwargs` forwards verbatim to CustomGuardrail.__init__, and the lifecycle
# hook signatures inherit `Any` for `response` from CustomLogger, so narrowing them here
# would break the override rather than describe it.
"litellm/proxy/guardrails/guardrail_hooks/llm_shield_proxy/llm_shield_proxy.py" = ["ANN401"]
[lint.mccabe]
max-complexity = 15

View file

@ -0,0 +1,901 @@
import json
from types import SimpleNamespace
from unittest.mock import AsyncMock
import pytest
from httpx import Request, Response
import litellm
from litellm.exceptions import GuardrailRaisedException
from litellm.proxy.guardrails.guardrail_hooks.llm_shield_proxy.llm_shield_proxy import (
GUARDRAIL_NAME,
LLMShieldProxyGuardrail,
)
from litellm.proxy.guardrails.init_guardrails import init_guardrails_v2
from litellm.types.guardrails import GuardrailEventHooks
from litellm.types.utils import Choices, Delta, Message, ModelResponse, ModelResponseStream, StreamingChoices
def _guardrail(**overrides: object) -> LLMShieldProxyGuardrail:
params: dict[str, object] = {
"api_key": "test-key",
"api_base": "http://shield.test",
"guardrail_name": GUARDRAIL_NAME,
"event_hook": "pre_call",
"default_on": True,
}
params.update(overrides)
return LLMShieldProxyGuardrail(**params)
def _response(payload: dict, status_code: int = 200) -> Response:
return Response(
status_code=status_code,
json=payload,
request=Request("POST", "http://shield.test/v1/guard/redact"),
)
def _mock_post(guardrail: LLMShieldProxyGuardrail, *payloads: dict) -> AsyncMock:
"""Queues one shield response per expected call."""
mock = AsyncMock(side_effect=[_response(p) for p in payloads])
guardrail.async_handler.post = mock # type: ignore[method-assign]
return mock
def _chunk(content: str | None, finish_reason: str | None = None) -> ModelResponseStream:
return ModelResponseStream(
choices=[StreamingChoices(index=0, delta=Delta(content=content), finish_reason=finish_reason)]
)
async def _drain(generator) -> list:
return [chunk async for chunk in generator]
def test_llm_shield_guardrail_config(monkeypatch: pytest.MonkeyPatch):
"""Should register through init_guardrails_v2 like any other provider."""
monkeypatch.setattr(litellm, "guardrail_name_config_map", {})
monkeypatch.setenv("LLM_SHIELD_PROXY_API_KEY", "test-key")
init_guardrails_v2(
all_guardrails=[
{
"guardrail_name": "llm_shield_proxy",
"litellm_params": {"guardrail": "llm_shield_proxy", "mode": "pre_call", "default_on": True},
}
],
config_file_path="",
)
registered = [cb for cb in litellm.callbacks if isinstance(cb, LLMShieldProxyGuardrail)]
assert len(registered) == 1
assert registered[0].guardrail_name == "llm_shield_proxy"
class TestLLMShieldProxyInitialization:
def test_api_base_defaults_to_localhost(self, monkeypatch: pytest.MonkeyPatch):
monkeypatch.delenv("LLM_SHIELD_PROXY_API_BASE", raising=False)
assert _guardrail(api_base=None).api_base == "http://localhost:8000"
def test_api_base_reads_environment(self, monkeypatch: pytest.MonkeyPatch):
monkeypatch.setenv("LLM_SHIELD_PROXY_API_BASE", "http://shield.internal:9000")
assert _guardrail(api_base=None).api_base == "http://shield.internal:9000"
def test_trailing_slash_is_stripped(self):
assert _guardrail(api_base="http://shield.test/").api_base == "http://shield.test"
def test_both_modes_can_be_enabled_on_one_entry(self):
"""Redaction and restoration are two halves of one config entry.
A deployment that lists only pre_call would redact the request and then hand
the placeholders straight back to the end user.
"""
guardrail = _guardrail(event_hook=["pre_call", "post_call"])
data: dict = {"messages": []}
assert guardrail.should_run_guardrail(data=data, event_type=GuardrailEventHooks.pre_call) is True
assert guardrail.should_run_guardrail(data=data, event_type=GuardrailEventHooks.post_call) is True
assert guardrail.should_run_guardrail(data=data, event_type=GuardrailEventHooks.during_call) is False
class TestRedaction:
@pytest.mark.asyncio
async def test_string_content_is_redacted(self):
guardrail = _guardrail()
_mock_post(guardrail, {"texts": ["Email [EMAIL_1] about it"]})
data = {"messages": [{"role": "user", "content": "Email a@b.com about it"}]}
result = await guardrail.async_pre_call_hook(
user_api_key_dict=None, cache=None, data=data, call_type="completion"
)
assert result["messages"][0]["content"] == "Email [EMAIL_1] about it"
@pytest.mark.asyncio
async def test_multimodal_text_parts_are_redacted(self):
"""The list content shape is a historical bypass; text parts must be covered."""
guardrail = _guardrail()
_mock_post(guardrail, {"texts": ["call [PHONE_1]"]})
data = {
"messages": [
{
"role": "user",
"content": [
{"type": "text", "text": "call 555-0100"},
{"type": "image_url", "image_url": {"url": "http://x/y.png"}},
],
}
]
}
await guardrail.async_pre_call_hook(user_api_key_dict=None, cache=None, data=data, call_type="completion")
assert data["messages"][0]["content"][0]["text"] == "call [PHONE_1]"
assert data["messages"][0]["content"][1]["image_url"]["url"] == "http://x/y.png"
@pytest.mark.asyncio
async def test_request_without_text_is_untouched(self):
"""No text to redact means no call to LLM Shield Proxy.
This deliberately uses a request with no caller text at all. An earlier
version used a Responses-API `input`, which asserted the very bypass that
let `input` reach the provider unredacted.
"""
guardrail = _guardrail()
mock = _mock_post(guardrail)
data = {"model": "gpt-4o", "temperature": 0.2}
await guardrail.async_pre_call_hook(user_api_key_dict=None, cache=None, data=data, call_type="completion")
mock.assert_not_called()
@pytest.mark.asyncio
async def test_session_id_is_reused_across_hooks(self):
"""Rehydration can only resolve tokens minted under the same session."""
guardrail = _guardrail()
mock = _mock_post(guardrail, {"texts": ["[EMAIL_1]"]}, {"texts": ["a@b.com"]})
data = {"messages": [{"role": "user", "content": "a@b.com"}]}
await guardrail.async_pre_call_hook(user_api_key_dict=None, cache=None, data=data, call_type="completion")
await guardrail._rehydrate(["[EMAIL_1]"], guardrail._session_id(data))
sessions = {call.kwargs["headers"]["X-Session-ID"] for call in mock.call_args_list}
assert len(sessions) == 1
class TestRequestCoverage:
"""Every request shape that carries caller text must be redacted.
A shape missed here is not a cosmetic gap: the guardrail reports as enabled
while the raw value goes to the provider.
"""
@pytest.mark.asyncio
async def test_responses_api_string_input_is_redacted(self):
"""Measured against a live provider: `input` reached the model unredacted."""
guardrail = _guardrail()
mock = _mock_post(guardrail, {"texts": ["Email [EMAIL_1] the invoice"]})
data = {"input": "Email jane.doe@example.com the invoice"}
await guardrail.async_pre_call_hook(user_api_key_dict=None, cache=None, data=data, call_type="aresponses")
assert mock.call_args_list[0].kwargs["json"]["texts"] == ["Email jane.doe@example.com the invoice"]
assert data["input"] == "Email [EMAIL_1] the invoice"
@pytest.mark.asyncio
async def test_responses_api_list_input_is_redacted(self):
guardrail = _guardrail()
_mock_post(guardrail, {"texts": ["[EMAIL_1]", "[PHONE_1]"]})
data = {
"input": [
{"role": "user", "content": "jane.doe@example.com"},
{"role": "user", "content": [{"type": "input_text", "text": "555-0100"}]},
]
}
await guardrail.async_pre_call_hook(user_api_key_dict=None, cache=None, data=data, call_type="aresponses")
assert data["input"][0]["content"] == "[EMAIL_1]"
assert data["input"][1]["content"][0]["text"] == "[PHONE_1]"
@pytest.mark.asyncio
async def test_tool_call_arguments_are_redacted(self):
"""Tool arguments carry the values the user asked the model to act on."""
guardrail = _guardrail()
_mock_post(guardrail, {"texts": ['{"email": "[EMAIL_1]"}']})
data = {
"messages": [
{
"role": "assistant",
"content": None,
"tool_calls": [
{
"id": "call_1",
"type": "function",
"function": {"name": "send", "arguments": '{"email": "jane.doe@example.com"}'},
}
],
}
]
}
await guardrail.async_pre_call_hook(user_api_key_dict=None, cache=None, data=data, call_type="completion")
assert data["messages"][0]["tool_calls"][0]["function"]["arguments"] == '{"email": "[EMAIL_1]"}'
@pytest.mark.asyncio
async def test_responses_api_instructions_are_redacted(self):
"""`instructions` is provider-bound text that sits outside `messages`."""
guardrail = _guardrail()
mock = _mock_post(guardrail, {"texts": ["contact [EMAIL_1]"]})
data = {"instructions": "contact jane.doe@example.com", "input": ""}
await guardrail.async_pre_call_hook(user_api_key_dict=None, cache=None, data=data, call_type="aresponses")
assert mock.call_args_list[0].kwargs["json"]["texts"] == ["contact jane.doe@example.com"]
assert data["instructions"] == "contact [EMAIL_1]"
@pytest.mark.asyncio
async def test_legacy_function_call_arguments_are_redacted(self):
guardrail = _guardrail()
_mock_post(guardrail, {"texts": ['{"email": "[EMAIL_1]"}']})
data = {
"messages": [
{
"role": "assistant",
"function_call": {"name": "send", "arguments": '{"email": "jane.doe@example.com"}'},
}
]
}
await guardrail.async_pre_call_hook(user_api_key_dict=None, cache=None, data=data, call_type="completion")
assert data["messages"][0]["function_call"]["arguments"] == '{"email": "[EMAIL_1]"}'
@pytest.mark.asyncio
async def test_completions_prompt_is_redacted(self):
"""/v1/completions puts its text in a top-level `prompt`, not in messages."""
guardrail = _guardrail()
mock = _mock_post(guardrail, {"texts": ["Email [EMAIL_1]"]})
data = {"prompt": "Email jane.doe@example.com"}
await guardrail.async_pre_call_hook(user_api_key_dict=None, cache=None, data=data, call_type="atext_completion")
assert mock.call_args_list[0].kwargs["json"]["texts"] == ["Email jane.doe@example.com"]
assert data["prompt"] == "Email [EMAIL_1]"
@pytest.mark.asyncio
async def test_completions_prompt_array_is_redacted(self):
"""`prompt` also accepts an array, and each entry is provider-bound."""
guardrail = _guardrail()
_mock_post(guardrail, {"texts": ["[EMAIL_1]", "[PHONE_1]"]})
data = {"prompt": ["jane.doe@example.com", "555-0100"]}
await guardrail.async_pre_call_hook(user_api_key_dict=None, cache=None, data=data, call_type="atext_completion")
assert data["prompt"] == ["[EMAIL_1]", "[PHONE_1]"]
@pytest.mark.asyncio
async def test_responses_function_call_items_are_redacted(self):
"""Responses input items hold tool data in `arguments` and `output`."""
guardrail = _guardrail()
_mock_post(guardrail, {"texts": ['{"email": "[EMAIL_1]"}', "sent to [EMAIL_1]"]})
data = {
"input": [
{"type": "function_call", "name": "send", "arguments": '{"email": "jane.doe@example.com"}'},
{"type": "function_call_output", "call_id": "c1", "output": "sent to jane.doe@example.com"},
]
}
await guardrail.async_pre_call_hook(user_api_key_dict=None, cache=None, data=data, call_type="aresponses")
assert data["input"][0]["arguments"] == '{"email": "[EMAIL_1]"}'
assert data["input"][1]["output"] == "sent to [EMAIL_1]"
@pytest.mark.asyncio
async def test_anthropic_system_prompt_is_redacted(self):
"""/v1/messages carries its system prompt at the top level, not in messages."""
guardrail = _guardrail()
mock = _mock_post(guardrail, {"texts": ["the user is [EMAIL_1]"]})
data = {"system": "the user is jane.doe@example.com", "messages": []}
await guardrail.async_pre_call_hook(
user_api_key_dict=None, cache=None, data=data, call_type="anthropic_messages"
)
assert mock.call_args_list[0].kwargs["json"]["texts"] == ["the user is jane.doe@example.com"]
assert data["system"] == "the user is [EMAIL_1]"
@pytest.mark.asyncio
async def test_anthropic_system_blocks_are_redacted(self):
"""`system` also accepts a list of text blocks."""
guardrail = _guardrail()
_mock_post(guardrail, {"texts": ["[EMAIL_1]"]})
data = {"system": [{"type": "text", "text": "jane.doe@example.com"}], "messages": []}
await guardrail.async_pre_call_hook(
user_api_key_dict=None, cache=None, data=data, call_type="anthropic_messages"
)
assert data["system"][0]["text"] == "[EMAIL_1]"
@pytest.mark.asyncio
async def test_string_array_input_is_redacted(self):
"""Embeddings and moderations send `input` as an array of bare strings."""
guardrail = _guardrail()
_mock_post(guardrail, {"texts": ["[EMAIL_1]", "[PHONE_1]"]})
data = {"input": ["jane.doe@example.com", "555-0100"]}
await guardrail.async_pre_call_hook(user_api_key_dict=None, cache=None, data=data, call_type="aembedding")
assert data["input"] == ["[EMAIL_1]", "[PHONE_1]"]
@pytest.mark.asyncio
async def test_participant_name_is_redacted(self):
"""`name` on a user turn identifies a person."""
guardrail = _guardrail()
mock = _mock_post(guardrail, {"texts": ["hi", "[PERSON_1]"]})
data = {"messages": [{"role": "user", "name": "Jane Doe", "content": "hi"}]}
await guardrail.async_pre_call_hook(user_api_key_dict=None, cache=None, data=data, call_type="completion")
assert mock.call_args_list[0].kwargs["json"]["texts"] == ["hi", "Jane Doe"]
assert data["messages"][0]["name"] == "[PERSON_1]"
@pytest.mark.asyncio
async def test_tool_function_name_is_left_alone(self):
"""On a tool turn the same field is the function name.
Redacting it would stop the call routing, so this asserts it is never sent
to the shield at all.
"""
guardrail = _guardrail()
mock = _mock_post(guardrail, {"texts": ["result"]})
data = {"messages": [{"role": "tool", "name": "get_weather", "content": "result"}]}
await guardrail.async_pre_call_hook(user_api_key_dict=None, cache=None, data=data, call_type="completion")
assert data["messages"][0]["name"] == "get_weather"
assert mock.call_args_list[0].kwargs["json"]["texts"] == ["result"]
@pytest.mark.asyncio
async def test_anthropic_tool_result_content_is_redacted(self):
"""A tool_result nests its own content, as a string or as more blocks."""
guardrail = _guardrail()
_mock_post(guardrail, {"texts": ["[EMAIL_1]", "[EMAIL_2]"]})
data = {
"messages": [
{
"role": "user",
"content": [
{"type": "tool_result", "tool_use_id": "t1", "content": "found jane.doe@example.com"},
{
"type": "tool_result",
"tool_use_id": "t2",
"content": [{"type": "text", "text": "also bob@example.com"}],
},
],
}
]
}
await guardrail.async_pre_call_hook(user_api_key_dict=None, cache=None, data=data, call_type="completion")
assert data["messages"][0]["content"][0]["content"] == "[EMAIL_1]"
assert data["messages"][0]["content"][1]["content"][0]["text"] == "[EMAIL_2]"
@pytest.mark.asyncio
async def test_deeply_nested_tool_results_are_bounded(self):
"""Nesting is caller controlled, so the descent has to stop somewhere.
The walk must terminate on a payload built to be pathological, rather than
following it as far as it goes.
"""
guardrail = _guardrail()
captured: list = []
async def echo(url, headers, json, timeout): # noqa: ARG001
captured.append(json["texts"])
return _response({"texts": list(json["texts"])})
guardrail.async_handler.post = AsyncMock(side_effect=echo) # type: ignore[method-assign]
deep: dict = {"type": "tool_result", "content": "past-the-bound@example.com"}
for _ in range(200):
deep = {"type": "tool_result", "content": [deep]}
data = {"messages": [{"role": "user", "content": [{"type": "text", "text": "shallow"}, deep]}]}
await guardrail.async_pre_call_hook(user_api_key_dict=None, cache=None, data=data, call_type="completion")
sent = captured[0]
assert "shallow" in sent
assert "past-the-bound@example.com" not in sent, "the walk followed the chain past its bound"
assert len(sent) < 200
@pytest.mark.asyncio
async def test_responses_prompt_object_variables_are_redacted(self):
"""A PromptObject's variables are substituted into the prompt provider side.
The id and version pick which stored prompt to run and have to arrive
unchanged; the variables are caller text.
"""
guardrail = _guardrail()
_mock_post(guardrail, {"texts": ["[EMAIL_1]"]})
data = {"prompt": {"id": "pmpt_123", "version": "2", "variables": {"customer": "jane.doe@example.com"}}}
await guardrail.async_pre_call_hook(user_api_key_dict=None, cache=None, data=data, call_type="aresponses")
assert data["prompt"]["variables"]["customer"] == "[EMAIL_1]"
assert data["prompt"]["id"] == "pmpt_123"
assert data["prompt"]["version"] == "2"
@pytest.mark.asyncio
async def test_completions_suffix_is_redacted(self):
"""LiteLLM forwards the legacy `suffix` to providers that support it."""
guardrail = _guardrail()
_mock_post(guardrail, {"texts": ["signed [EMAIL_1]", "write to [EMAIL_1]"]})
data = {"prompt": "write to jane.doe@example.com", "suffix": "signed jane.doe@example.com"}
await guardrail.async_pre_call_hook(user_api_key_dict=None, cache=None, data=data, call_type="atext_completion")
assert data["suffix"] == "signed [EMAIL_1]"
@pytest.mark.asyncio
async def test_every_shape_in_one_request_is_redacted(self):
guardrail = _guardrail()
mock = _mock_post(guardrail, {"texts": ["a", "b", "c", "d"]})
data = {
"messages": [
{"role": "user", "content": "one"},
{"role": "user", "content": [{"type": "text", "text": "two"}]},
{
"role": "assistant",
"tool_calls": [{"function": {"name": "f", "arguments": "three"}}],
},
],
"input": "four",
}
await guardrail.async_pre_call_hook(user_api_key_dict=None, cache=None, data=data, call_type="completion")
assert mock.call_args_list[0].kwargs["json"]["texts"] == ["one", "two", "three", "four"]
assert data["messages"][0]["content"] == "a"
assert data["messages"][1]["content"][0]["text"] == "b"
assert data["messages"][2]["tool_calls"][0]["function"]["arguments"] == "c"
assert data["input"] == "d"
class TestRestoration:
@pytest.mark.asyncio
async def test_openai_shape_is_restored(self):
guardrail = _guardrail(event_hook="post_call")
_mock_post(guardrail, {"texts": ["a@b.com"]})
response = ModelResponse(choices=[Choices(index=0, message=Message(role="assistant", content="[EMAIL_1]"))])
result = await guardrail.async_post_call_success_hook(
data={"messages": []}, user_api_key_dict=None, response=response
)
assert result.choices[0].message.content == "a@b.com"
@pytest.mark.asyncio
async def test_responses_api_shape_is_restored(self):
"""The Responses API reply carries output items, not choices.
Measured against a live provider: once the request side was fixed the reply
came back still holding the placeholder, because this shape has no choices
to walk.
"""
guardrail = _guardrail(event_hook="post_call")
_mock_post(guardrail, {"texts": ["a@b.com"]})
response = SimpleNamespace(output=[SimpleNamespace(content=[{"type": "output_text", "text": "[EMAIL_1]"}])])
result = await guardrail.async_post_call_success_hook(
data={"messages": []}, user_api_key_dict=None, response=response
)
assert result.output[0].content[0]["text"] == "a@b.com"
@pytest.mark.asyncio
async def test_responses_api_object_blocks_are_restored(self):
"""Blocks arrive as objects too, depending on how far the reply is parsed."""
guardrail = _guardrail(event_hook="post_call")
_mock_post(guardrail, {"texts": ["a@b.com"]})
block = SimpleNamespace(text="[EMAIL_1]")
response = SimpleNamespace(output=[SimpleNamespace(content=[block])])
await guardrail.async_post_call_success_hook(data={"messages": []}, user_api_key_dict=None, response=response)
assert block.text == "a@b.com"
@pytest.mark.asyncio
async def test_anthropic_message_shape_is_restored(self):
"""The /v1/messages reply is a plain dict with no choices.
Measured against a live provider: without its own branch the reply went
back to the caller still carrying the placeholder, even though the
request had been redacted correctly.
"""
guardrail = _guardrail(event_hook="post_call")
_mock_post(guardrail, {"texts": ["a@b.com"]})
response = {
"type": "message",
"role": "assistant",
"content": [{"type": "text", "text": "[EMAIL_1]"}],
}
result = await guardrail.async_post_call_success_hook(
data={"messages": []}, user_api_key_dict=None, response=response
)
assert result["content"][0]["text"] == "a@b.com"
@pytest.mark.asyncio
async def test_anthropic_non_text_blocks_are_left_alone(self):
guardrail = _guardrail(event_hook="post_call")
_mock_post(guardrail, {"texts": ["a@b.com"]})
response = {
"type": "message",
"role": "assistant",
"content": [
{"type": "text", "text": "[EMAIL_1]"},
{"type": "tool_use", "id": "t1", "name": "lookup", "input": {}},
],
}
result = await guardrail.async_post_call_success_hook(
data={"messages": []}, user_api_key_dict=None, response=response
)
assert result["content"][0]["text"] == "a@b.com"
assert result["content"][1] == {"type": "tool_use", "id": "t1", "name": "lookup", "input": {}}
class TestVaultIsolation:
"""The vault id must never be something a caller can choose.
The vault holds the plaintext behind every placeholder. If a caller could name
the vault, they could send a placeholder, have the model echo it back, and get
another caller's value restored into their own reply.
"""
@pytest.mark.asyncio
async def test_caller_supplied_session_id_is_not_used(self):
guardrail = _guardrail()
mock = _mock_post(guardrail, {"texts": ["[EMAIL_1]"]})
data = {
"messages": [{"role": "user", "content": "a@b.com"}],
"metadata": {"llm_shield_session_id": "victim-session"},
"litellm_session_id": "victim-session",
}
await guardrail.async_pre_call_hook(user_api_key_dict=None, cache=None, data=data, call_type="completion")
used = mock.call_args_list[0].kwargs["headers"]["X-Session-ID"]
assert used != "victim-session"
assert data["metadata"]["llm_shield_session_id"] == used
@pytest.mark.asyncio
async def test_restore_ignores_a_foreign_session_id(self):
"""A reply is left unrestored rather than resolved against another vault."""
guardrail = _guardrail(event_hook="post_call")
mock = _mock_post(guardrail, {"texts": ["[EMAIL_1]"]})
data = {"metadata": {"llm_shield_session_id": "victim-session"}}
response = ModelResponse(choices=[Choices(index=0, message=Message(role="assistant", content="[EMAIL_1]"))])
await guardrail.async_post_call_success_hook(data=data, user_api_key_dict=None, response=response)
assert mock.call_args_list[0].kwargs["headers"]["X-Session-ID"] != "victim-session"
@pytest.mark.asyncio
async def test_each_request_gets_its_own_vault(self):
guardrail = _guardrail()
mock = _mock_post(guardrail, {"texts": ["[EMAIL_1]"]}, {"texts": ["[EMAIL_1]"]})
for _ in range(2):
await guardrail.async_pre_call_hook(
user_api_key_dict=None,
cache=None,
data={"messages": [{"role": "user", "content": "a@b.com"}]},
call_type="completion",
)
seen = {call.kwargs["headers"]["X-Session-ID"] for call in mock.call_args_list}
assert len(seen) == 2
@pytest.mark.parametrize(
"data",
[
pytest.param(
{"messages": [{"role": "system", "content": "S"}, {"role": "user", "content": "U"}]},
id="system-turn",
),
pytest.param(
{"messages": [{"role": "developer", "content": "S"}, {"role": "user", "content": "U"}]},
id="developer-turn",
),
pytest.param(
{"system": "S", "messages": [{"role": "user", "content": "U"}]},
id="anthropic-top-level-system",
),
pytest.param({"instructions": "S", "input": "U"}, id="responses-instructions"),
],
)
def test_server_authored_text_is_split_from_the_callers(self, data: dict) -> None:
"""Every request shape must sort its server-authored spans out of the caller's."""
caller, privileged = LLMShieldProxyGuardrail._locate_request_texts(data)
assert [text for text, _ in caller] == ["U"]
assert [text for text, _ in privileged] == ["S"]
@pytest.mark.asyncio
async def test_a_system_prompt_gets_a_vault_of_its_own(self) -> None:
"""The reply is restored against the caller's vault, so the two cannot be one."""
guardrail = _guardrail()
mock = _mock_post(guardrail, {"texts": ["[EMAIL_1]"]}, {"texts": ["[EMAIL_2]"]})
data = {
"messages": [
{"role": "system", "content": "escalate to admin@corp.internal"},
{"role": "user", "content": "email a@b.com"},
]
}
await guardrail.async_pre_call_hook(
user_api_key_dict=None, cache=None, data=data, call_type="completion"
)
privileged_id, caller_id = (
call.kwargs["headers"]["X-Session-ID"] for call in mock.call_args_list
)
assert privileged_id != caller_id
assert guardrail._session_id(data) == caller_id
@pytest.mark.asyncio
async def test_the_system_prompt_vault_id_is_never_stored(self) -> None:
"""Nothing can restore against the system vault later, because its id is not kept.
This is what stops a caller from having the model echo a placeholder out of a
system prompt they cannot see and receiving the plaintext behind it.
"""
guardrail = _guardrail()
mock = _mock_post(guardrail, {"texts": ["[EMAIL_1]"]}, {"texts": ["[EMAIL_2]"]})
data = {
"messages": [
{"role": "system", "content": "escalate to admin@corp.internal"},
{"role": "user", "content": "email a@b.com"},
]
}
await guardrail.async_pre_call_hook(
user_api_key_dict=None, cache=None, data=data, call_type="completion"
)
privileged_id = mock.call_args_list[0].kwargs["headers"]["X-Session-ID"]
assert privileged_id not in json.dumps(data, default=str)
class TestFailClosed:
@pytest.mark.asyncio
async def test_unreachable_shield_blocks_the_request(self):
"""Failing open would send the PII upstream, defeating the guardrail."""
guardrail = _guardrail()
guardrail.async_handler.post = AsyncMock(side_effect=ConnectionError("refused"))
with pytest.raises(GuardrailRaisedException):
await guardrail.async_pre_call_hook(
user_api_key_dict=None,
cache=None,
data={"messages": [{"role": "user", "content": "a@b.com"}]},
call_type="completion",
)
@pytest.mark.asyncio
async def test_error_status_blocks_the_request(self):
guardrail = _guardrail()
guardrail.async_handler.post = AsyncMock(return_value=_response({"error": "nope"}, status_code=500))
with pytest.raises(GuardrailRaisedException):
await guardrail.async_pre_call_hook(
user_api_key_dict=None,
cache=None,
data={"messages": [{"role": "user", "content": "a@b.com"}]},
call_type="completion",
)
@pytest.mark.asyncio
async def test_short_payload_blocks_the_request(self):
"""A response that loses an entry would silently misalign the write-back."""
guardrail = _guardrail()
_mock_post(guardrail, {"texts": []})
with pytest.raises(GuardrailRaisedException):
await guardrail.async_pre_call_hook(
user_api_key_dict=None,
cache=None,
data={"messages": [{"role": "user", "content": "a@b.com"}]},
call_type="completion",
)
class TestStreamingRehydration:
@pytest.mark.asyncio
async def test_split_placeholder_is_not_emitted_in_fragments(self):
"""The window holds back a partial placeholder and releases it once complete."""
guardrail = _guardrail(event_hook="post_call")
# Shield holds "[EMAIL" back, then releases the restored value.
_mock_post(
guardrail,
{"text": "Email ", "carry": "[EMAIL"},
{"text": "a@b.com about it", "carry": ""},
)
async def stream():
yield _chunk("Email [EMAIL")
yield _chunk("_1] about it", finish_reason="stop")
chunks = await _drain(
guardrail.async_post_call_streaming_iterator_hook(
user_api_key_dict=None, response=stream(), request_data={"messages": []}
)
)
emitted = [c.choices[0].delta.content for c in chunks]
assert emitted == ["Email ", "a@b.com about it"]
# No fragment of the placeholder ever reached the client.
assert not any("[EMAIL" in (text or "") for text in emitted)
@pytest.mark.asyncio
async def test_carry_is_returned_to_the_next_call(self):
guardrail = _guardrail(event_hook="post_call")
mock = _mock_post(
guardrail,
{"text": "", "carry": "hold"},
{"text": "held-and-more", "carry": ""},
)
async def stream():
yield _chunk("hold")
yield _chunk("-and-more", finish_reason="stop")
await _drain(
guardrail.async_post_call_streaming_iterator_hook(
user_api_key_dict=None, response=stream(), request_data={"messages": []}
)
)
assert mock.call_args_list[0].kwargs["json"]["carry"] == ""
assert mock.call_args_list[1].kwargs["json"]["carry"] == "hold"
assert mock.call_args_list[1].kwargs["json"]["final"] is True
@pytest.mark.asyncio
async def test_every_choice_is_restored(self):
"""With n>1 a later choice must not be handed back still holding a placeholder."""
guardrail = _guardrail(event_hook="post_call")
_mock_post(
guardrail,
{"text": "first@example.com", "carry": ""},
{"text": "second@example.com", "carry": ""},
)
async def stream():
yield ModelResponseStream(
choices=[
StreamingChoices(index=0, delta=Delta(content="[EMAIL_1]"), finish_reason="stop"),
StreamingChoices(index=1, delta=Delta(content="[EMAIL_2]"), finish_reason="stop"),
]
)
chunks = await _drain(
guardrail.async_post_call_streaming_iterator_hook(
user_api_key_dict=None, response=stream(), request_data={"messages": []}
)
)
restored = [choice.delta.content for choice in chunks[0].choices]
assert restored == ["first@example.com", "second@example.com"]
@pytest.mark.asyncio
async def test_choice_windows_do_not_cross_contaminate(self):
"""Each choice is its own token stream, so each carries its own window.
One shared window would send the characters held back for choice 0 up
against choice 1's next delta and splice the two streams together.
"""
guardrail = _guardrail(event_hook="post_call")
mock = _mock_post(
guardrail,
{"text": "", "carry": "A-held"},
{"text": "", "carry": "B-held"},
{"text": "a-done", "carry": ""},
{"text": "b-done", "carry": ""},
)
async def stream():
yield ModelResponseStream(
choices=[
StreamingChoices(index=0, delta=Delta(content="a1")),
StreamingChoices(index=1, delta=Delta(content="b1")),
]
)
yield ModelResponseStream(
choices=[
StreamingChoices(index=0, delta=Delta(content="a2"), finish_reason="stop"),
StreamingChoices(index=1, delta=Delta(content="b2"), finish_reason="stop"),
]
)
await _drain(
guardrail.async_post_call_streaming_iterator_hook(
user_api_key_dict=None, response=stream(), request_data={"messages": []}
)
)
sent = [call.kwargs["json"] for call in mock.call_args_list]
assert sent[2]["carry"] == "A-held", "choice 0 must get its own window back"
assert sent[3]["carry"] == "B-held", "choice 1 must get its own window back"
@pytest.mark.asyncio
async def test_a_choice_missing_from_the_last_chunk_still_flushes(self):
"""Held text must not be dropped because its choice ended earlier.
Choice 1 finishes and stops appearing, then the stream ends without a
finish_reason for choice 0. Flushing only the terminal chunk's choices would
discard whatever choice 1 was still holding and truncate its answer.
"""
guardrail = _guardrail(event_hook="post_call")
_mock_post(
guardrail,
{"text": "", "carry": "held-0"},
{"text": "", "carry": "held-1"},
{"text": "zero-done", "carry": ""},
{"text": "one-done", "carry": ""},
)
async def stream():
yield ModelResponseStream(
choices=[
StreamingChoices(index=0, delta=Delta(content="a")),
StreamingChoices(index=1, delta=Delta(content="b")),
]
)
yield ModelResponseStream(choices=[StreamingChoices(index=0, delta=Delta(content=None))])
chunks = await _drain(
guardrail.async_post_call_streaming_iterator_hook(
user_api_key_dict=None, response=stream(), request_data={"messages": []}
)
)
flushed = {
choice.index: choice.delta.content for chunk in chunks for choice in chunk.choices if choice.delta.content
}
assert flushed.get(1) == "one-done", "choice 1's held text was dropped"
assert flushed.get(0) == "zero-done"
@pytest.mark.asyncio
async def test_chunks_are_forwarded_as_they_arrive(self):
"""Restoration must not buffer the stream into a single terminal chunk."""
guardrail = _guardrail(event_hook="post_call")
_mock_post(
guardrail,
{"text": "one ", "carry": ""},
{"text": "two ", "carry": ""},
{"text": "three", "carry": ""},
)
async def stream():
yield _chunk("one ")
yield _chunk("two ")
yield _chunk("three", finish_reason="stop")
chunks = await _drain(
guardrail.async_post_call_streaming_iterator_hook(
user_api_key_dict=None, response=stream(), request_data={"messages": []}
)
)
assert len(chunks) == 3
assert [c.choices[0].delta.content for c in chunks] == ["one ", "two ", "three"]

View file

@ -0,0 +1,6 @@
<svg width="24" height="24" viewBox="0 0 24 24" fill="none" xmlns="http://www.w3.org/2000/svg">
<rect width="24" height="24" rx="8" fill="#0B1F33"/>
<path d="M12 4L18 6.35V11.5C18 15.28 15.62 18.2 12 19.4C8.38 18.2 6 15.28 6 11.5V6.35L12 4Z" fill="#BDEFF7"/>
<path d="M12 4L18 6.35V11.5C18 15.28 15.62 18.2 12 19.4V4Z" fill="#3FC5DE"/>
<path d="M12 4L18 6.35V11.5C18 15.28 15.62 18.2 12 19.4C8.38 18.2 6 15.28 6 11.5V6.35L12 4Z" stroke="#6FE3F2" stroke-width="0.9" stroke-linejoin="round"/>
</svg>

After

Width:  |  Height:  |  Size: 498 B

View file

@ -73,7 +73,9 @@ interface GuardrailPreset {
provider: string;
categoryName?: string;
guardrailNameSuggestion: string;
mode: string;
// A guardrail that both rewrites the request and repairs the response needs two
// modes seeded, not one; the form already normalises either shape.
mode: string | string[];
defaultOn: boolean;
}

View file

@ -2,7 +2,9 @@ export interface GuardrailPreset {
provider: string;
categoryName?: string;
guardrailNameSuggestion: string;
mode: string;
// A guardrail that both rewrites the request and repairs the response needs two
// modes seeded, not one; the form already normalises either shape.
mode: string | string[];
defaultOn: boolean;
}
@ -318,4 +320,12 @@ export const GUARDRAIL_PRESETS: Record<string, GuardrailPreset> = {
mode: "pre_call",
defaultOn: false,
},
llm_shield_proxy: {
provider: "LLM Shield Proxy",
guardrailNameSuggestion: "LLM Shield Proxy",
// Both halves are required. With only pre_call the request is redacted and the
// placeholders are handed straight back to the caller.
mode: ["pre_call", "post_call"],
defaultOn: false,
},
};

View file

@ -28,6 +28,7 @@ const EXPECTED_PARTNER_LOGO_FILES: Record<string, string> = {
repelloai: "repelloai.png",
straiker: "straiker.svg",
alice: "alice.svg",
llm_shield_proxy: "llm_shield_proxy.svg",
};
describe("guardrail_garden_data logos", () => {

View file

@ -474,6 +474,16 @@ export const PARTNER_GUARDRAIL_CARDS: GuardrailCardInfo[] = [
tags: ["Content Moderation", "Prompt Injection", "PII", "Policy"],
providerKey: "Alice",
},
{
id: "llm_shield_proxy",
name: "LLM Shield Proxy",
description:
"Self-hosted PII redaction that puts the original values back into the model's response, so the provider never receives personal data while the end user still sees it.",
category: "partner",
logo: guardrailLogoMap["LLM Shield Proxy"],
tags: ["PII", "Data Privacy", "Compliance", "Streaming"],
providerKey: "LLM Shield Proxy",
},
];
export const ALL_CARDS = [...LITELLM_CONTENT_FILTER_CARDS, ...PARTNER_GUARDRAIL_CARDS];

View file

@ -1,6 +1,7 @@
import aimSecurityLogo from "../../../../../public/assets/logos/aim_security.jpeg";
import aktoLogo from "../../../../../public/assets/logos/akto.svg";
import aliceLogo from "../../../../../public/assets/logos/alice.svg";
import llmShieldProxyLogo from "../../../../../public/assets/logos/llm_shield_proxy.svg";
import aporiaLogo from "../../../../../public/assets/logos/aporia.png";
import bedrockLogo from "../../../../../public/assets/logos/bedrock.svg";
import catoNetworksLogo from "../../../../../public/assets/logos/cato_networks.svg";
@ -85,6 +86,7 @@ export const guardrail_provider_map: Record<string, string> = {
QostodianNexus: "qostodian_nexus",
Repelloai: "repelloai",
Alice: "alice",
"LLM Shield Proxy": "llm_shield_proxy",
};
// Function to populate provider map from API response - updates the original map
@ -208,6 +210,7 @@ export const guardrailLogoMap = {
"RepelloAI Argus": repelloAiLogo.src,
Straiker: straikerLogo.src,
Alice: aliceLogo.src,
"LLM Shield Proxy": llmShieldProxyLogo.src,
} satisfies Record<string, string>;
export const getGuardrailLogo = (displayName: string): string | undefined =>