Coverage for .venv/lib/python3.13/site-packages/litellm/proxy/guardrails/guardrail_hooks/alice/alice.py: 32%
130 statements
« prev ^ index » next coverage.py v7.15.2, created at 2026-10-10 12:01 +0000
1# +-------------------------------------------------------------+
2#
3# Use Alice for your LLM calls
4# https://alice.io/
5#
6# +-------------------------------------------------------------+
8import json
9import os
10from collections.abc import Mapping
11from itertools import islice
12from typing import (
13 TYPE_CHECKING,
14 Final,
15 Literal,
16 Optional,
17)
19import httpx
20from typing_extensions import NotRequired, ReadOnly, TypedDict, Unpack
22from litellm._logging import verbose_proxy_logger
23from litellm.exceptions import GuardrailRaisedException, Timeout
24from litellm.integrations.custom_guardrail import (
25 CustomGuardrail,
26 log_guardrail_information,
27)
28from litellm.llms.custom_httpx.http_handler import (
29 get_async_httpx_client,
30 httpxSpecialProvider,
31)
32from litellm.types.guardrails import GuardrailEventHooks
33from litellm.types.utils import GenericGuardrailAPIInputs
35if TYPE_CHECKING: 35 ↛ 36line 35 didn't jump to line 36 because the condition on line 35 was never true
36 from litellm.litellm_core_utils.litellm_logging import Logging as LiteLLMLoggingObj
38GUARDRAIL_NAME: Final = "alice"
40_DEFAULT_API_BASE: Final = "https://api.alice.io"
41_EVALUATE_PATH: Final = "/v2/evaluate/litellm"
43_VERDICT_ALLOW: Final = "ALLOW"
44_VERDICT_BLOCK: Final = "BLOCK"
45_VERDICT_MASK: Final = "MASK"
46_VERDICT_DETECT: Final = "DETECT"
47_KNOWN_VERDICTS: Final = frozenset({_VERDICT_ALLOW, _VERDICT_BLOCK, _VERDICT_MASK, _VERDICT_DETECT})
49_DEFAULT_BLOCK_MESSAGE: Final = "Blocked by your organization's content policy."
51# apply_guardrail selects nothing: it forwards whichever of these came populated and lets Alice
52# decide what is worth evaluating. Only skip the call when every one of them is empty — there is
53# then genuinely nothing to send.
54_SELECTABLE_INPUT_FIELDS: Final = ("texts", "images", "tools", "tool_calls", "structured_messages")
56# Caps on the outbound copy of request_data. A payload deeper or wider than this is malformed
57# rather than large, and serializing it would cost more than the evaluation it feeds.
58_MAX_DEPTH: Final = 12
59_MAX_ITEMS: Final = 5000
61# request_data carries the caller's raw credentials under these keys, at any nesting depth —
62# a real captured payload puts inbound headers at request_data["proxy_server_request"]["headers"],
63# again under ["metadata"]["headers"] / ["litellm_metadata"]["headers"], and again under
64# ["metadata"]["requester_metadata"]["headers"], any of which can carry an Authorization or
65# x-api-key value. LiteLLM's own spend-log sanitizer excludes `secret_fields` for the same reason
66# (spend_tracking_utils._SENSITIVE_REQUEST_BODY_KEYS): `secret_fields.raw_headers` holds the
67# caller's Authorization / x-api-key in the clear, and `api_key` can carry a forwarded provider
68# credential. Stripping by key name rather than by path means a new nesting path can never
69# reintroduce the leak. Posting any of these to a third-party guardrail endpoint would be worse
70# than what the proxy already refuses to persist in its own audit trail — so none of them leave
71# the process.
72_CREDENTIAL_KEYS_TO_STRIP: Final = frozenset(
73 {"secret_fields", "api_key", "raw_headers", "headers", "provider_specific_header"}
74)
77class AliceReplacement(TypedDict):
78 """A masked substitution, positional against the texts that were submitted."""
80 index: ReadOnly[NotRequired[int]]
81 text: ReadOnly[NotRequired[str]]
84class AliceVerdict(TypedDict):
85 """Body returned by Alice's LiteLLM evaluate endpoint."""
87 verdict: ReadOnly[NotRequired[str]]
88 categories: ReadOnly[NotRequired["tuple[str, ...]"]]
89 correlation_id: ReadOnly[NotRequired[str]]
90 message: ReadOnly[NotRequired[str]]
91 replacements: ReadOnly[NotRequired["tuple[AliceReplacement, ...]"]]
94class _CustomGuardrailOptions(TypedDict, total=False, extra_items=object):
95 pass
98class AliceGuardrailMissingSecrets(Exception):
99 """Raised when the Alice API key is not configured."""
102class AliceGuardrail(CustomGuardrail):
103 """
104 Alice — policy-based guardrails for prompts and model responses.
106 This forwards the hook's arguments as it received them and enforces the verdict that comes
107 back, with one deliberate exception: any key named `secret_fields`, `api_key`, `raw_headers`,
108 `headers`, or `provider_specific_header` is dropped from `request_data` at any nesting depth
109 before it is serialized, and never reaches Alice. Short of that, it selects nothing and
110 renames nothing: which parts of a conversation are worth evaluating, and how a verdict is
111 reached, are decided by Alice — so changing either is a change on their side rather than a
112 LiteLLM upgrade. A batch with nothing selectable at all (no `texts`, `images`, `tools`,
113 `tool_calls`, or `structured_messages`) still skips the call, since there would be nothing to
114 send.
116 Known limitation: the unified guardrail's `streaming_transform_mode` defaults to
117 `block_only`, whose streaming path discards any returned text rewrite. A MASK verdict is
118 therefore a no-op on a streamed response — the original, unmasked text still reaches the
119 caller — while BLOCK continues to function on both streamed and non-streamed responses.
120 This is `during_call`'s documented behavior generally, not specific to Alice; configure a
121 masking-aware `streaming_transform_mode` if that gap matters for your traffic.
123 Alice evaluates against policies configured per *application*, and one proxy typically fronts
124 several, so the application is named on the virtual key rather than in this config:
126 curl $PROXY/key/generate -H "Authorization: Bearer $LITELLM_MASTER_KEY" \\
127 -d '{"key_alias": "payments-bot",
128 "metadata": {"alice_app_id": "payments-bot"}}'
130 Alice reads that off the authenticated key. Because the proxy strips caller-supplied
131 `user_api_key_*` from the request before a guardrail sees it, a caller cannot point its own
132 traffic at an application with laxer policies than the one its key was issued for.
134 Configuration example (litellm config YAML):
135 guardrails:
136 - guardrail_name: alice
137 litellm_params:
138 guardrail: alice
139 mode: [pre_call, post_call]
140 api_key: os.environ/ALICE_API_KEY
141 api_base: https://api.alice.io # optional
142 unreachable_fallback: fail_closed # optional
143 """
145 def __init__(
146 self,
147 api_key: str | None = None,
148 api_base: str | None = None,
149 unreachable_fallback: Literal["fail_closed", "fail_open"] = "fail_closed",
150 **kwargs: Unpack[ # kwargs-ok: forwarded verbatim to CustomGuardrail.__init__, whose param list is wide and evolving
151 _CustomGuardrailOptions
152 ],
153 ) -> None:
154 self.async_handler = get_async_httpx_client(llm_provider=httpxSpecialProvider.GuardrailCallback)
156 alice_api_key: Final = api_key or os.environ.get("ALICE_API_KEY")
157 if not alice_api_key:
158 raise AliceGuardrailMissingSecrets(
159 "Alice API key is required. Set the `ALICE_API_KEY` environment variable or "
160 "pass `api_key` in the guardrail config."
161 )
162 self.alice_api_key: str = alice_api_key
164 base: Final = (api_base or os.environ.get("ALICE_API_BASE") or _DEFAULT_API_BASE).rstrip("/")
165 self.api_base: str = f"{base}{_EVALUATE_PATH}"
166 self.unreachable_fallback: Literal["fail_closed", "fail_open"] = unreachable_fallback
168 if "supported_event_hooks" not in kwargs:
169 kwargs["supported_event_hooks"] = [ # mutable-ok: CustomGuardrail.__init__ requires a list here
170 GuardrailEventHooks.pre_call,
171 GuardrailEventHooks.during_call,
172 GuardrailEventHooks.post_call,
173 ]
175 super().__init__(**kwargs)
177 @log_guardrail_information
178 async def apply_guardrail(
179 self,
180 inputs: GenericGuardrailAPIInputs,
181 request_data: dict[str, object], # mutable-ok: overrides CustomGuardrail.apply_guardrail's plain-dict contract
182 input_type: Literal["request", "response"],
183 logging_obj: Optional["LiteLLMLoggingObj"] = None,
184 ) -> GenericGuardrailAPIInputs:
185 if not any(inputs.get(field) for field in _SELECTABLE_INPUT_FIELDS):
186 return inputs
188 try:
189 verdict: AliceVerdict = await self._evaluate(
190 inputs=inputs, request_data=request_data, input_type=input_type
191 )
192 except Timeout as e:
193 return self._on_unreachable(e, inputs)
194 except httpx.HTTPStatusError as e:
195 status_code: Final = getattr(getattr(e, "response", None), "status_code", None)
196 # Any 5xx is an outage on Alice's side, not our misconfiguration — route the whole
197 # class through the configured policy. A 4xx (rejected credential, bad request) is
198 # ours to fix and must never fail open, so it is deliberately left to propagate.
199 if isinstance(status_code, int) and 500 <= status_code < 600:
200 return self._on_unreachable(e, inputs)
201 raise
202 except httpx.RequestError as e:
203 return self._on_unreachable(e, inputs)
204 except (json.JSONDecodeError, UnicodeDecodeError, TypeError) as e:
205 # A body that cannot be decoded, cannot be parsed as JSON, or parses to something
206 # other than an object, is as unreachable as a dropped connection: this deployment's
207 # policy decides, not a raw exception. UnicodeDecodeError is named explicitly because
208 # it is a sibling of JSONDecodeError under ValueError, not a subclass of it.
209 return self._on_unreachable(e, inputs)
211 return self._enforce(verdict, inputs)
213 async def _evaluate(
214 self,
215 inputs: GenericGuardrailAPIInputs,
216 request_data: Mapping[str, object],
217 input_type: str,
218 ) -> AliceVerdict:
219 response: Final = await self.async_handler.post(
220 url=self.api_base,
221 json={ # mutable-ok: one-shot HTTP request body, never mutated after construction
222 "input_type": input_type,
223 "inputs": _json_safe(inputs),
224 "request_data": _json_safe(request_data, strip_keys=_CREDENTIAL_KEYS_TO_STRIP),
225 },
226 headers={ # mutable-ok: one-shot HTTP headers, never mutated after construction
227 "Content-Type": "application/json",
228 "af-api-key": self.alice_api_key,
229 },
230 )
231 response.raise_for_status()
232 body = response.json()
233 if not isinstance(body, dict):
234 raise TypeError("Alice returned a non-object body")
235 return body
237 def _enforce(self, verdict: AliceVerdict, inputs: GenericGuardrailAPIInputs) -> GenericGuardrailAPIInputs:
238 """Act on the verdict. An answer we cannot read is treated as unavailable, never as a pass."""
239 name: Final = verdict.get("verdict")
240 if name not in _KNOWN_VERDICTS:
241 return self._on_unreachable(ValueError(f"unrecognized verdict: {name!r}"), inputs)
243 if name == _VERDICT_BLOCK:
244 raise GuardrailRaisedException(
245 guardrail_name=GUARDRAIL_NAME,
246 message=verdict.get("message") or _DEFAULT_BLOCK_MESSAGE,
247 should_wrap_with_default_message=False,
248 blocked_content=True,
249 )
251 if name == _VERDICT_DETECT:
252 # Recorded by Alice and allowed through. The correlation id is what ties this request
253 # to that record; the evaluated text itself is never logged.
254 verbose_proxy_logger.warning(
255 "Alice guardrail: detection recorded, request allowed (correlation_id=%s, categories=%s)",
256 verdict.get("correlation_id"),
257 verdict.get("categories"),
258 )
259 return inputs
261 if name == _VERDICT_MASK:
262 self._apply_replacements(verdict, inputs)
264 return inputs
266 def _apply_replacements(self, verdict: AliceVerdict, inputs: GenericGuardrailAPIInputs) -> None:
267 """
268 Write each replacement onto the text it names.
270 Only `texts` is touched. The chat translation layer maps a returned `texts` list back onto
271 the request positionally, but takes a different branch entirely when `structured_messages`
272 comes back as a new object — which would drop these edits.
274 All-or-nothing: a single out-of-range or malformed replacement blocks the whole verdict
275 rather than being silently skipped, so content Alice meant to replace can never reach the
276 model unmasked alongside content that was replaced.
277 """
278 texts: Final = inputs.get("texts") or [] # mutable-ok: empty-list fallback, replaced wholesale below
279 replacements: Final = verdict.get("replacements") or [] # mutable-ok: empty-list fallback for iteration only
281 if not replacements:
282 raise self._mask_rejected(verdict)
284 for replacement in replacements:
285 index = replacement.get("index")
286 text = replacement.get("text")
287 if not (isinstance(index, int) and isinstance(text, str) and 0 <= index < len(texts)):
288 raise self._mask_rejected(verdict)
289 texts[index] = text
291 inputs["texts"] = texts
293 def _mask_rejected(self, verdict: AliceVerdict) -> GuardrailRaisedException:
294 """A MASK verdict that cannot be applied in full is refused outright, never partially —
295 see `_apply_replacements`."""
296 return GuardrailRaisedException(
297 guardrail_name=GUARDRAIL_NAME,
298 message=verdict.get("message") or _DEFAULT_BLOCK_MESSAGE,
299 should_wrap_with_default_message=False,
300 blocked_content=True,
301 )
303 def _on_unreachable(self, error: Exception, inputs: GenericGuardrailAPIInputs) -> GenericGuardrailAPIInputs:
304 """Apply the configured policy when Alice cannot be reached or cannot be understood."""
305 if self.unreachable_fallback == "fail_open":
306 verbose_proxy_logger.critical(
307 "Alice guardrail unreachable, allowing request per unreachable_fallback: %s",
308 error,
309 )
310 return inputs
311 raise GuardrailRaisedException(
312 guardrail_name=GUARDRAIL_NAME,
313 message="Alice guardrail is unavailable and this request cannot be checked",
314 should_wrap_with_default_message=False,
315 ) from error
317 @staticmethod
318 def get_config_model() -> type | None:
319 from litellm.types.proxy.guardrails.guardrail_hooks.alice import (
320 AliceGuardrailConfigModel,
321 )
323 return AliceGuardrailConfigModel
326def _json_safe(
327 value: object,
328 depth: int = 0,
329 seen: frozenset[int] = frozenset(),
330 strip_keys: frozenset[str] = frozenset(),
331) -> object:
332 """
333 Copy `value` into something `json.dumps` accepts, dropping only what cannot cross.
335 `request_data` carries live Python objects — an OpenTelemetry span among them — so it cannot
336 be serialized as it stands. What is dropped is decided by a mechanical rule rather than a
337 field list: a list drifts from what the far side needs, a rule cannot. Serializing naively
338 raises, and that error would be read as "guardrail unavailable" on every single request.
340 `strip_keys` drops a dict key by name at every depth it appears, not just the root — a caller
341 passes `_CREDENTIAL_KEYS_TO_STRIP` here so a credential nested under any path is caught the
342 same way a top-level one is, without maintaining a list of paths. The source object is never
343 mutated: every branch below builds a new container.
344 """
345 if isinstance(value, (str, int, float, bool)) or value is None:
346 return value
347 if depth >= _MAX_DEPTH or id(value) in seen:
348 return None
350 nested: Final = seen | frozenset((id(value),))
352 if isinstance(value, dict):
353 return {
354 key: _json_safe(item, depth + 1, nested, strip_keys)
355 for key, item in islice(value.items(), _MAX_ITEMS)
356 if isinstance(key, str) and key not in strip_keys
357 }
359 if isinstance(value, (list, tuple, set, frozenset)):
360 return [ # mutable-ok: return value is a one-shot list, discarded by the caller after use
361 _json_safe(item, depth + 1, nested, strip_keys) for item in islice(value, _MAX_ITEMS)
362 ]
364 dump: Final = getattr(value, "model_dump", None)
365 if callable(dump):
366 try:
367 return _json_safe(dump(mode="json"), depth + 1, nested, strip_keys)
368 except Exception: # noqa: BLE001 # a model that will not dump is one we drop
369 return None
371 # Everything json.dumps handles natively — str, int, float, bool, None, dict, list — is
372 # caught above, and a dict/list subclass is caught by isinstance. So whatever reaches here
373 # (bytes, datetime, an OpenTelemetry span) cannot cross the wire.
374 return None