Coverage for .venv/lib/python3.13/site-packages/litellm/proxy/common_utils/callback_utils.py: 34%
405 statements
« prev ^ index » next coverage.py v7.15.2, created at 2026-10-10 12:01 +0000
1import copy
2import json
3import os
4from collections.abc import Callable, Iterable, Mapping
5from dataclasses import dataclass
6from itertools import accumulate
7from typing import TYPE_CHECKING, Any, Final, Literal, NoReturn, Optional, TypeAlias
9from typing_extensions import ReadOnly, TypedDict, assert_never
11import litellm
12from litellm import get_secret
13from litellm._logging import verbose_proxy_logger
14from litellm.constants import (
15 CLIENT_OUTPUT_CEILING_METADATA_KEY,
16 CONSUMED_REQUEST_TAGS_METADATA_KEY,
17 MAX_GUARDRAIL_SCAN_METADATA_HEADER_LENGTH,
18 PRE_CALL_EXECUTED_GUARDRAILS_KEY,
19 ROUTING_REQUEST_TAGS_METADATA_KEY,
20 SESSION_DEPLOYMENT_AFFINITY_TTL_METADATA_KEY,
21)
22from litellm.integrations.custom_logger import CustomLogger
23from litellm.litellm_core_utils.core_helpers import (
24 get_metadata_variable_name_from_kwargs,
25 get_or_create_metadata_bucket,
26)
27from litellm.litellm_core_utils.sensitive_data_masker import SensitiveDataMasker
28from litellm.proxy._types import CommonProxyErrors, LiteLLMPromptInjectionParams
29from litellm.proxy.common_utils.encrypt_decrypt_utils import (
30 decrypt_value_helper,
31 encrypt_value_helper,
32)
33from litellm.proxy.types_utils.utils import get_instance_fn
34from litellm.types.guardrails import GuardrailEventHooks
35from litellm.types.utils import (
36 StandardLoggingGuardrailInformation,
37 StandardLoggingPayload,
38)
40_CALLBACK_VAR_MASKER: Final = SensitiveDataMasker()
41# Compound names that are credential-bearing but don't contain any of the
42# default sensitive segments (so SensitiveDataMasker won't flag them).
43_EXTRA_SENSITIVE_CALLBACK_KEYS: Final = {"gcs_path_service_account"}
44# Sentinel prefix on encrypted callback_var values. Lets us detect
45# already-encrypted input cheaply (no decrypt-attempt round trip) and
46# avoid double-encrypting if `LITELLM_SALT_KEY` is rotated between writes.
47CALLBACK_VAR_ENCRYPTED_PREFIX: Final = "litellm_enc::"
48_CALLBACK_VAR_ENCRYPTED_PREFIX: Final = CALLBACK_VAR_ENCRYPTED_PREFIX
49# Metadata slots that hold operator-configured callback and secret-manager setup
50# (and therefore integration credentials). Resolved from UserAPIKeyAuth during
51# pre-call setup, never read back off the copies stamped into request metadata.
52_CALLBACK_CONFIG_SLOTS: Final = frozenset({"logging", "callback_settings", "secret_manager_settings"})
54blue_color_code: Final = "\033[94m"
55reset_color_code: Final = "\033[0m"
57TRUSTED_PILLAR_RESPONSE_HEADERS_METADATA_KEY: Final = "_pillar_response_headers_trusted"
59GUARDRAIL_SCAN_IDS_METADATA_KEY: Final = "guardrail_scan_ids"
60GUARDRAIL_SCAN_METADATA_METADATA_KEY: Final = "guardrail_scan_metadata"
63class GuardrailScanMetadata(TypedDict):
64 guardrail: ReadOnly[str | None]
65 stage: ReadOnly[str]
66 provider: ReadOnly[str]
67 scan_id: ReadOnly[str]
70if TYPE_CHECKING: 70 ↛ 71line 70 didn't jump to line 71 because the condition on line 70 was never true
71 from litellm.litellm_core_utils.litellm_logging import Logging as LiteLLMLogging
74@dataclass(frozen=True, slots=True)
75class _CallbackResolvedToClass:
76 entry: str
77 loaded: type
78 tag: Literal["resolved_to_class"] = "resolved_to_class"
81@dataclass(frozen=True, slots=True)
82class _CallbackNotDispatchable:
83 entry: str
84 loaded: object
85 tag: Literal["not_dispatchable"] = "not_dispatchable"
88_CallbackLoadError: TypeAlias = _CallbackResolvedToClass | _CallbackNotDispatchable
91def _classify_loaded_callback(entry: str, loaded: object) -> CustomLogger | Callable[..., object] | _CallbackLoadError:
92 """
93 Decide whether what a ``litellm_settings.callbacks`` dotted path resolved to can be dispatched.
95 A dotted path only ever runs as a ``CustomLogger`` instance or as a callback function. Anything
96 else (most commonly a class instead of an instance) used to load without complaint and then be
97 skipped on every request, with no log line and no error.
98 """
99 if isinstance(loaded, CustomLogger) or (callable(loaded) and not isinstance(loaded, type)):
100 return loaded
101 if isinstance(loaded, type):
102 return _CallbackResolvedToClass(entry=entry, loaded=loaded)
103 return _CallbackNotDispatchable(entry=entry, loaded=loaded)
106def _raise_callback_load_error(error: _CallbackLoadError) -> NoReturn:
107 """The one edge that raises: map a load error onto config load's failure contract."""
108 match error:
109 case _CallbackResolvedToClass():
110 module_path: Final = error.entry.rsplit(".", 1)[0] if "." in error.entry else error.entry
111 raise ValueError(
112 f"litellm_settings.callbacks entry '{error.entry}' resolved to the class "
113 f"{error.loaded.__module__}.{error.loaded.__qualname__}, which is neither a "
114 "CustomLogger instance nor a callable, so the proxy would never run it."
115 f" Point it at an instance instead, e.g. add `proxy_handler_instance = {error.loaded.__name__}()` to "
116 f'{module_path} and set `callbacks: ["{module_path}.proxy_handler_instance"]`.'
117 )
118 case _CallbackNotDispatchable():
119 raise ValueError(
120 f"litellm_settings.callbacks entry '{error.entry}' resolved to "
121 f"{type(error.loaded).__name__} {error.loaded!r}, which is neither a "
122 "CustomLogger instance nor a callable, so the proxy would never run it."
123 )
124 assert_never(error)
127def _loaded_callback_or_raise(entry: str, loaded: object) -> CustomLogger | Callable[..., object]:
128 resolved: Final = _classify_loaded_callback(entry=entry, loaded=loaded)
129 if isinstance(resolved, _CallbackResolvedToClass | _CallbackNotDispatchable):
130 _raise_callback_load_error(resolved)
131 return resolved
134def initialize_callbacks_on_proxy(
135 value: Any,
136 premium_user: bool,
137 config_file_path: str,
138 litellm_settings: dict,
139 callback_specific_params: dict | None = None,
140):
141 if not isinstance(callback_specific_params, dict):
142 callback_specific_params = {}
143 from litellm.integrations.custom_logger import CustomLogger
144 from litellm.litellm_core_utils.logging_callback_manager import (
145 LoggingCallbackManager,
146 )
147 from litellm.proxy.proxy_server import prisma_client
149 verbose_proxy_logger.debug("%sinitializing callbacks=%s on proxy%s", blue_color_code, value, reset_color_code)
150 if isinstance(value, list):
151 imported_list: Final[list[Any]] = []
152 for callback in value: # ["presidio", <my-custom-callback>]
153 if isinstance(callback, str) and callback == "compression_interception":
154 from litellm.integrations.compression_interception.handler import (
155 CompressionInterceptionLogger,
156 )
158 compression_interception_obj = CompressionInterceptionLogger.initialize_from_proxy_config(
159 litellm_settings=litellm_settings,
160 callback_specific_params=callback_specific_params,
161 )
162 imported_list.append(compression_interception_obj)
163 continue
165 if isinstance(callback, str) and callback == "code_interpreter_interception":
166 from litellm.integrations.code_interpreter_interception.handler import (
167 CodeInterpreterInterceptionLogger,
168 )
170 code_interpreter_interception_obj = CodeInterpreterInterceptionLogger.initialize_from_proxy_config(
171 litellm_settings=litellm_settings,
172 callback_specific_params=callback_specific_params,
173 )
174 imported_list.append(code_interpreter_interception_obj)
175 continue
177 # check if callback is a custom logger compatible callback
178 if isinstance(callback, str):
179 callback = LoggingCallbackManager._add_custom_callback_generic_api_str(callback)
180 if isinstance(callback, str) and callback in litellm._known_custom_logger_compatible_callbacks:
181 imported_list.append(callback)
182 elif isinstance(callback, str) and callback == "presidio":
183 from litellm.proxy.guardrails.guardrail_hooks.presidio import (
184 _OPTIONAL_PresidioPIIMasking,
185 )
187 presidio_logging_only: bool | None = litellm_settings.get("presidio_logging_only", None)
188 if presidio_logging_only is not None:
189 presidio_logging_only = bool(presidio_logging_only) # validate boolean given
191 _presidio_params = {}
192 if "presidio" in callback_specific_params and isinstance(callback_specific_params["presidio"], dict):
193 _presidio_params = callback_specific_params["presidio"]
195 params: dict[str, Any] = {
196 "logging_only": presidio_logging_only,
197 **_presidio_params,
198 }
199 pii_masking_object = _OPTIONAL_PresidioPIIMasking(**params)
200 imported_list.append(pii_masking_object)
201 elif isinstance(callback, str) and callback == "llamaguard_moderations":
202 try:
203 from litellm_enterprise.enterprise_callbacks.llama_guard import (
204 _ENTERPRISE_LlamaGuard,
205 )
206 except ImportError:
207 raise Exception(
208 "MissingTrying to use Llama Guard" + CommonProxyErrors.missing_enterprise_package.value
209 )
211 if premium_user is not True:
212 raise Exception("Trying to use Llama Guard" + CommonProxyErrors.not_premium_user.value)
214 llama_guard_object = _ENTERPRISE_LlamaGuard()
215 imported_list.append(llama_guard_object)
216 elif isinstance(callback, str) and callback == "hide_secrets":
217 try:
218 from litellm_enterprise.enterprise_callbacks.secret_detection import (
219 _ENTERPRISE_SecretDetection,
220 )
221 except ImportError:
222 raise Exception(
223 "Trying to use Secret Detection" + CommonProxyErrors.missing_enterprise_package.value
224 )
226 if premium_user is not True:
227 raise Exception("Trying to use secret hiding" + CommonProxyErrors.not_premium_user.value)
229 _secret_detection_object = _ENTERPRISE_SecretDetection()
230 imported_list.append(_secret_detection_object)
231 elif isinstance(callback, str) and callback == "openai_moderations":
232 try:
233 from enterprise.enterprise_hooks.openai_moderation import (
234 _ENTERPRISE_OpenAI_Moderation,
235 )
236 except ImportError:
237 raise Exception(
238 "Trying to use OpenAI Moderations Check,"
239 + CommonProxyErrors.missing_enterprise_package_docker.value
240 )
242 if premium_user is not True:
243 raise Exception("Trying to use OpenAI Moderations Check" + CommonProxyErrors.not_premium_user.value)
245 openai_moderations_object = _ENTERPRISE_OpenAI_Moderation()
246 imported_list.append(openai_moderations_object)
247 elif isinstance(callback, str) and callback == "lakera_prompt_injection":
248 from litellm.proxy.guardrails.guardrail_hooks.lakera_ai import (
249 lakeraAI_Moderation,
250 )
252 init_params = {}
253 if "lakera_prompt_injection" in callback_specific_params and isinstance(
254 callback_specific_params["lakera_prompt_injection"], dict
255 ):
256 init_params = callback_specific_params["lakera_prompt_injection"]
257 lakera_moderations_object = lakeraAI_Moderation(**init_params)
258 imported_list.append(lakera_moderations_object)
259 elif isinstance(callback, str) and callback == "aporia_prompt_injection":
260 from litellm.proxy.guardrails.guardrail_hooks.aporia_ai.aporia_ai import (
261 AporiaGuardrail,
262 )
264 aporia_guardrail_object = AporiaGuardrail()
265 imported_list.append(aporia_guardrail_object)
266 elif isinstance(callback, str) and callback == "google_text_moderation":
267 try:
268 from enterprise.enterprise_hooks.google_text_moderation import (
269 _ENTERPRISE_GoogleTextModeration,
270 )
271 except ImportError:
272 raise Exception(
273 "Trying to use Google Text Moderation,"
274 + CommonProxyErrors.missing_enterprise_package_docker.value
275 )
277 if premium_user is not True:
278 raise Exception("Trying to use Google Text Moderation" + CommonProxyErrors.not_premium_user.value)
280 google_text_moderation_obj = _ENTERPRISE_GoogleTextModeration()
281 imported_list.append(google_text_moderation_obj)
282 elif isinstance(callback, str) and callback == "llmguard_moderations":
283 try:
284 from litellm_enterprise.enterprise_callbacks.llm_guard import (
285 _ENTERPRISE_LLMGuard,
286 )
287 except ImportError:
288 raise Exception("Trying to use Llm Guard" + CommonProxyErrors.missing_enterprise_package.value)
290 if premium_user is not True:
291 raise Exception("Trying to use Llm Guard" + CommonProxyErrors.not_premium_user.value)
293 llm_guard_moderation_obj = _ENTERPRISE_LLMGuard()
294 imported_list.append(llm_guard_moderation_obj)
295 elif isinstance(callback, str) and callback == "blocked_user_check":
296 try:
297 from enterprise.enterprise_hooks.blocked_user_list import (
298 _ENTERPRISE_BlockedUserList,
299 )
300 except ImportError:
301 raise Exception(
302 "Trying to use Blocked User List" + CommonProxyErrors.missing_enterprise_package_docker.value
303 )
305 if premium_user is not True:
306 raise Exception("Trying to use ENTERPRISE BlockedUser" + CommonProxyErrors.not_premium_user.value)
308 blocked_user_list = _ENTERPRISE_BlockedUserList(prisma_client=prisma_client)
309 imported_list.append(blocked_user_list)
310 elif isinstance(callback, str) and callback == "banned_keywords":
311 try:
312 from enterprise.enterprise_hooks.banned_keywords import (
313 _ENTERPRISE_BannedKeywords,
314 )
315 except ImportError:
316 raise Exception(
317 "Trying to use Banned Keywords" + CommonProxyErrors.missing_enterprise_package_docker.value
318 )
320 if premium_user is not True:
321 raise Exception("Trying to use ENTERPRISE BannedKeyword" + CommonProxyErrors.not_premium_user.value)
323 banned_keywords_obj = _ENTERPRISE_BannedKeywords()
324 imported_list.append(banned_keywords_obj)
325 elif isinstance(callback, str) and callback == "detect_prompt_injection":
326 from litellm.proxy.hooks.prompt_injection_detection import (
327 _OPTIONAL_PromptInjectionDetection,
328 )
330 prompt_injection_params = None
331 if "prompt_injection_params" in litellm_settings:
332 prompt_injection_params_in_config = litellm_settings["prompt_injection_params"]
333 prompt_injection_params = LiteLLMPromptInjectionParams(**prompt_injection_params_in_config)
335 prompt_injection_detection_obj = _OPTIONAL_PromptInjectionDetection(
336 prompt_injection_params=prompt_injection_params,
337 )
338 imported_list.append(prompt_injection_detection_obj)
339 elif isinstance(callback, str) and callback == "batch_redis_requests":
340 from litellm.proxy.hooks.batch_redis_get import (
341 _PROXY_BatchRedisRequests,
342 )
344 batch_redis_obj = _PROXY_BatchRedisRequests()
345 imported_list.append(batch_redis_obj)
346 elif isinstance(callback, str) and callback == "azure_content_safety":
347 from litellm.proxy.hooks.azure_content_safety import (
348 _PROXY_AzureContentSafety,
349 )
351 azure_content_safety_params = litellm_settings["azure_content_safety_params"]
352 for k, v in azure_content_safety_params.items():
353 if v is not None and isinstance(v, str) and v.startswith("os.environ/"):
354 azure_content_safety_params[k] = get_secret(v)
356 azure_content_safety_obj = _PROXY_AzureContentSafety(
357 **azure_content_safety_params,
358 )
359 imported_list.append(azure_content_safety_obj)
360 elif isinstance(callback, str) and callback == "websearch_interception":
361 from litellm.integrations.websearch_interception.handler import (
362 WebSearchInterceptionLogger,
363 )
365 websearch_interception_obj = WebSearchInterceptionLogger.initialize_from_proxy_config(
366 litellm_settings=litellm_settings,
367 callback_specific_params=callback_specific_params,
368 )
369 imported_list.append(websearch_interception_obj)
370 elif isinstance(callback, str) and callback == "datadog_cost_management":
371 from litellm.integrations.datadog.datadog_cost_management import (
372 DatadogCostManagementLogger,
373 )
375 init_params = {}
376 if "datadog_cost_management" in callback_specific_params and isinstance(
377 callback_specific_params["datadog_cost_management"], dict
378 ):
379 init_params = callback_specific_params["datadog_cost_management"]
380 datadog_cost_management_obj = DatadogCostManagementLogger(**init_params)
381 imported_list.append(datadog_cost_management_obj)
382 elif isinstance(callback, CustomLogger):
383 imported_list.append(callback)
384 else:
385 verbose_proxy_logger.debug(
386 "%s attempting to import custom calback=%s %s", blue_color_code, callback, reset_color_code
387 )
388 imported_list.append(
389 _loaded_callback_or_raise(
390 entry=callback,
391 loaded=get_instance_fn(
392 value=callback,
393 config_file_path=config_file_path,
394 ),
395 )
396 )
397 if isinstance(litellm.callbacks, list):
398 litellm.callbacks.extend(imported_list)
399 else:
400 litellm.callbacks = imported_list
402 if "prometheus" in value:
403 from litellm.integrations.prometheus import PrometheusLogger
405 PrometheusLogger._mount_metrics_endpoint()
406 else:
407 litellm.callbacks = [
408 _loaded_callback_or_raise(
409 entry=value,
410 loaded=get_instance_fn(
411 value=value,
412 config_file_path=config_file_path,
413 ),
414 )
415 ]
416 verbose_proxy_logger.debug("%s Initialized Callbacks - %s %s", blue_color_code, litellm.callbacks, reset_color_code)
419def get_model_group_from_litellm_kwargs(kwargs: dict) -> str | None:
420 _litellm_params: Final = kwargs.get("litellm_params", None) or {}
421 _metadata: Final = _litellm_params.get(get_metadata_variable_name_from_kwargs(kwargs)) or {}
422 _model_group: Final = _metadata.get("model_group", None)
423 if _model_group is not None: 423 ↛ 424line 423 didn't jump to line 424 because the condition on line 423 was never true
424 return _model_group
426 return None
429def get_model_group_from_request_data(data: dict) -> str | None:
430 _metadata: Final = data.get("metadata", None) or {}
431 _model_group: Final = _metadata.get("model_group", None)
432 if _model_group is not None:
433 return _model_group
435 return None
438def get_remaining_tokens_and_requests_from_request_data(data: dict) -> dict[str, str]:
439 """
440 Helper function to return x-litellm-key-remaining-tokens-{model_group} and x-litellm-key-remaining-requests-{model_group}
442 Returns {} when api_key + model rpm/tpm limit is not set
444 """
445 headers: Final = {}
446 _metadata: Final = data.get("metadata", None) or {}
447 model_group: Final = get_model_group_from_request_data(data)
449 # The h11 package considers "/" or ":" invalid and raise a LocalProtocolError
450 h11_model_group_name: Final = model_group.replace("/", "-").replace(":", "-") if model_group else None
452 # Remaining Requests
453 remaining_requests_variable_name: Final = f"litellm-key-remaining-requests-{model_group}"
454 remaining_requests: Final = _metadata.get(remaining_requests_variable_name, None)
455 if remaining_requests: 455 ↛ 456line 455 didn't jump to line 456 because the condition on line 455 was never true
456 headers[f"x-litellm-key-remaining-requests-{h11_model_group_name}"] = remaining_requests
458 # Remaining Tokens
459 remaining_tokens_variable_name: Final = f"litellm-key-remaining-tokens-{model_group}"
460 remaining_tokens: Final = _metadata.get(remaining_tokens_variable_name, None)
461 if remaining_tokens: 461 ↛ 462line 461 didn't jump to line 462 because the condition on line 461 was never true
462 headers[f"x-litellm-key-remaining-tokens-{h11_model_group_name}"] = remaining_tokens
464 return headers
467def _serialize_scan_metadata_header(entries: Iterable[object], *, max_length: int) -> str | None:
468 """Compact JSON list of scan metadata entries, dropping trailing entries so the header fits in max_length."""
469 encoded: Final = tuple(json.dumps(entry, separators=(",", ":")) for entry in entries)
470 lengths: Final = tuple(accumulate(len(item) + 1 for item in encoded))
471 kept: Final = sum(1 for length in lengths if length + 1 <= max_length)
472 if kept == 0:
473 return None
474 return f"[{','.join(encoded[:kept])}]"
477def get_logging_caching_headers(request_data: dict) -> dict | None:
478 _metadata: Final[dict] = {}
479 metadata_bucket: Final = request_data.get("metadata")
480 litellm_metadata_bucket: Final = request_data.get("litellm_metadata")
481 if isinstance(metadata_bucket, dict):
482 _metadata.update(metadata_bucket)
483 if isinstance(litellm_metadata_bucket, dict):
484 # Batch/file routes store proxy tracking in litellm_metadata while
485 # user-facing metadata stays in metadata; merge both for headers.
486 _metadata.update(litellm_metadata_bucket)
487 headers: Final = {}
488 if "applied_guardrails" in _metadata:
489 headers["x-litellm-applied-guardrails"] = ",".join(_metadata["applied_guardrails"])
491 scan_ids: Final = _metadata.get(GUARDRAIL_SCAN_IDS_METADATA_KEY)
492 if scan_ids: 492 ↛ 493line 492 didn't jump to line 493 because the condition on line 492 was never true
493 headers["x-litellm-guardrail-scan-id"] = ",".join(scan_ids)
495 scan_metadata: Final = _metadata.get(GUARDRAIL_SCAN_METADATA_METADATA_KEY)
496 scan_metadata_header: Final = (
497 _serialize_scan_metadata_header(scan_metadata, max_length=MAX_GUARDRAIL_SCAN_METADATA_HEADER_LENGTH)
498 if isinstance(scan_metadata, (list, tuple))
499 else None
500 )
501 if scan_metadata_header: 501 ↛ 502line 501 didn't jump to line 502 because the condition on line 501 was never true
502 headers["x-litellm-guardrail-scan-metadata"] = scan_metadata_header
504 if "applied_policies" in _metadata:
505 headers["x-litellm-applied-policies"] = ",".join(_metadata["applied_policies"])
507 if "policy_sources" in _metadata:
508 sources: Final = _metadata["policy_sources"]
509 if isinstance(sources, dict) and sources: 509 ↛ 513line 509 didn't jump to line 513 because the condition on line 509 was always true
510 # Use ';' as delimiter — matched_via reasons may contain commas
511 headers["x-litellm-policy-sources"] = "; ".join(f"{name}={reason}" for name, reason in sources.items())
513 if "semantic-similarity" in _metadata: 513 ↛ 514line 513 didn't jump to line 514 because the condition on line 513 was never true
514 headers["x-litellm-semantic-similarity"] = str(_metadata["semantic-similarity"])
516 is_trusted_pillar_metadata: Final = _metadata.get(TRUSTED_PILLAR_RESPONSE_HEADERS_METADATA_KEY) is True
517 pillar_headers: Final = _metadata.get("pillar_response_headers")
518 if is_trusted_pillar_metadata and isinstance(pillar_headers, dict): 518 ↛ 519line 518 didn't jump to line 519 because the condition on line 518 was never true
519 headers.update(
520 {
521 key: str(value)
522 for key, value in pillar_headers.items()
523 if isinstance(key, str) and key.lower().startswith("x-pillar-")
524 }
525 )
526 elif is_trusted_pillar_metadata and "pillar_flagged" in _metadata: 526 ↛ 527line 526 didn't jump to line 527 because the condition on line 526 was never true
527 headers["x-pillar-flagged"] = str(_metadata["pillar_flagged"]).lower()
529 return headers
532LITELLM_PROXY_INTERNAL_METADATA_KEYS: Final = frozenset(
533 {
534 "applied_policies",
535 "applied_guardrails",
536 GUARDRAIL_SCAN_IDS_METADATA_KEY,
537 GUARDRAIL_SCAN_METADATA_METADATA_KEY,
538 "policy_sources",
539 "guardrails",
540 "guardrail_config",
541 "_guardrail_pipelines",
542 "_pipeline_managed_guardrails",
543 PRE_CALL_EXECUTED_GUARDRAILS_KEY,
544 SESSION_DEPLOYMENT_AFFINITY_TTL_METADATA_KEY,
545 CONSUMED_REQUEST_TAGS_METADATA_KEY,
546 CLIENT_OUTPUT_CEILING_METADATA_KEY,
547 ROUTING_REQUEST_TAGS_METADATA_KEY,
548 "disable_global_guardrails",
549 "disable_global_guardrail",
550 "opted_out_global_guardrails",
551 "pillar_response_headers",
552 "_pillar_response_headers_trusted",
553 "pillar_flagged",
554 "pillar_scanners",
555 "pillar_evidence",
556 "pillar_evidence_truncated",
557 "pillar_session_id_response",
558 "standard_logging_object",
559 "proxy_server_request",
560 "secret_fields",
561 }
562)
565def sanitize_openai_provider_metadata(
566 metadata: Mapping[str, object] | None,
567) -> Mapping[str, object] | None:
568 """
569 Keep only provider-safe OpenAI metadata entries (string keys -> string values).
571 Strips LiteLLM proxy-internal tracking fields that must not be forwarded to
572 OpenAI batch/file APIs.
573 """
574 if metadata is None: 574 ↛ 576line 574 didn't jump to line 576 because the condition on line 574 was always true
575 return None
576 sanitized: Final[dict[str, str]] = {}
577 for key, value in metadata.items():
578 if key in LITELLM_PROXY_INTERNAL_METADATA_KEYS:
579 continue
580 if isinstance(value, str):
581 sanitized[key] = value
582 else:
583 verbose_proxy_logger.debug(
584 "sanitize_openai_provider_metadata: dropping key %r with non-string value of type %s",
585 key,
586 type(value).__name__,
587 )
588 return None if metadata and not sanitized else sanitized
591def add_guardrail_to_applied_guardrails_header(request_data: dict, guardrail_name: str | None):
592 if guardrail_name is None: 592 ↛ 593line 592 didn't jump to line 593 because the condition on line 592 was never true
593 return
594 _, _metadata = get_or_create_metadata_bucket(request_data)
595 if "applied_guardrails" in _metadata:
596 if guardrail_name not in _metadata["applied_guardrails"]:
597 _metadata["applied_guardrails"].append(guardrail_name)
598 else:
599 _metadata["applied_guardrails"] = [guardrail_name]
602def add_guardrail_scan_id(
603 request_data: dict[str, object],
604 scan_id: str | None,
605 *,
606 guardrail_name: str | None,
607 provider: str,
608 stage: GuardrailEventHooks,
609) -> None:
610 """
611 Record a provider scan id, keyed to the guardrail execution that produced it, so it can be surfaced to the caller.
613 Guardrails only return scan details to the client when they block, so allowed requests carry no
614 audit trail. Ids recorded here become the x-litellm-guardrail-scan-id response header, and the
615 (guardrail, stage, provider, scan_id) entries become the x-litellm-guardrail-scan-metadata header.
616 """
617 if not scan_id:
618 return
619 _, _metadata = get_or_create_metadata_bucket(request_data)
620 existing: Final = _metadata.get(GUARDRAIL_SCAN_IDS_METADATA_KEY)
621 scan_ids: Final[tuple[object, ...]] = tuple(existing) if isinstance(existing, (list, tuple)) else ()
622 if scan_id not in scan_ids:
623 _metadata[GUARDRAIL_SCAN_IDS_METADATA_KEY] = (*scan_ids, scan_id)
625 entry: Final[GuardrailScanMetadata] = {
626 "guardrail": guardrail_name,
627 "stage": stage.value,
628 "provider": provider,
629 "scan_id": scan_id,
630 }
631 existing_entries: Final = _metadata.get(GUARDRAIL_SCAN_METADATA_METADATA_KEY)
632 entries: Final[tuple[object, ...]] = tuple(existing_entries) if isinstance(existing_entries, (list, tuple)) else ()
633 if entry not in entries:
634 _metadata[GUARDRAIL_SCAN_METADATA_METADATA_KEY] = (*entries, entry)
637def add_policy_to_applied_policies_header(request_data: dict, policy_name: str | None):
638 """
639 Add a policy name to the applied_policies list in request metadata.
641 This is used to track which policies were applied to a request,
642 similar to how applied_guardrails tracks guardrails.
643 """
644 if policy_name is None: 644 ↛ 645line 644 didn't jump to line 645 because the condition on line 644 was never true
645 return
646 _, _metadata = get_or_create_metadata_bucket(request_data)
647 if "applied_policies" in _metadata:
648 if policy_name not in _metadata["applied_policies"]: 648 ↛ exitline 648 didn't return from function 'add_policy_to_applied_policies_header' because the condition on line 648 was always true
649 _metadata["applied_policies"].append(policy_name)
650 else:
651 _metadata["applied_policies"] = [policy_name]
654def add_policy_sources_to_metadata(request_data: dict, policy_sources: dict[str, str]):
655 """
656 Store policy match reasons in metadata for x-litellm-policy-sources header.
658 Args:
659 request_data: The request data dict
660 policy_sources: Map of policy_name -> matched_via reason
661 """
662 if not policy_sources: 662 ↛ 663line 662 didn't jump to line 663 because the condition on line 662 was never true
663 return
664 _, _metadata = get_or_create_metadata_bucket(request_data)
665 existing = _metadata.get("policy_sources", {})
666 if not isinstance(existing, dict): 666 ↛ 667line 666 didn't jump to line 667 because the condition on line 666 was never true
667 existing = {}
668 existing.update(policy_sources)
669 _metadata["policy_sources"] = existing
672def add_guardrail_response_to_standard_logging_object(
673 litellm_logging_obj: Optional["LiteLLMLogging"],
674 guardrail_response: StandardLoggingGuardrailInformation,
675):
676 if litellm_logging_obj is None:
677 return
678 standard_logging_object: Final[StandardLoggingPayload | None] = litellm_logging_obj.model_call_details.get(
679 "standard_logging_object"
680 )
681 if standard_logging_object is None:
682 return
683 guardrail_information = standard_logging_object.get("guardrail_information", [])
684 if guardrail_information is None:
685 guardrail_information = []
686 guardrail_information.append(guardrail_response)
687 standard_logging_object["guardrail_information"] = guardrail_information
689 return standard_logging_object
692def process_callback(_callback: str, callback_type: str, environment_variables: dict) -> dict:
693 """Process a single callback and return its data with environment variables"""
694 env_vars: Final = CustomLogger.get_callback_env_vars(_callback)
696 env_vars_dict: Final[dict[str, str | None]] = {}
697 for _var in env_vars:
698 stored_value = environment_variables.get(_var, None)
699 env_vars_dict[_var] = stored_value if stored_value is not None else os.getenv(_var)
701 return {"name": _callback, "variables": env_vars_dict, "type": callback_type}
704def normalize_callback_names(callbacks: Iterable[object] | None) -> list[object]:
705 if callbacks is None:
706 return []
707 return [c.lower() if isinstance(c, str) else c for c in callbacks]
710def strip_callback_config(metadata: dict[str, object] | None) -> dict[str, object] | None:
711 """Return key/team metadata without the slots that carry callback credentials."""
712 if not isinstance(metadata, dict):
713 return metadata
714 return {k: v for k, v in metadata.items() if k not in _CALLBACK_CONFIG_SLOTS}
717def encrypt_callback_vars(metadata: object) -> Any:
718 """Return a deep copy of metadata with callback_vars values encrypted at rest.
720 Idempotent: a value that already decrypts cleanly is left unchanged so
721 round-trips through edit forms don't double-encrypt.
722 """
723 return _transform_callback_vars(metadata, _encrypt_if_plaintext)
726def decrypt_callback_vars(metadata: object) -> Any:
727 """Return a deep copy of metadata with callback_vars values decrypted.
729 Legacy plaintext rows pass through unchanged (decrypt failure → original).
730 """
731 return _transform_callback_vars(metadata, _decrypt_or_passthrough)
734def _transform_callback_vars(metadata: object, transform: Callable[[str, object], object]) -> object:
735 if not isinstance(metadata, dict): 735 ↛ 736line 735 didn't jump to line 736 because the condition on line 735 was never true
736 return metadata
737 out: Final = copy.deepcopy(metadata)
738 logging_entries: Final = out.get("logging")
739 if isinstance(logging_entries, list):
740 for entry in logging_entries:
741 if isinstance(entry, dict) and isinstance(entry.get("callback_vars"), dict): 741 ↛ 740line 741 didn't jump to line 740 because the condition on line 741 was always true
742 entry["callback_vars"] = {k: transform(k, v) for k, v in entry["callback_vars"].items()}
743 callback_settings: Final = out.get("callback_settings")
744 if isinstance(callback_settings, dict) and isinstance(callback_settings.get("callback_vars"), dict):
745 callback_settings["callback_vars"] = {k: transform(k, v) for k, v in callback_settings["callback_vars"].items()}
746 return out
749def is_sensitive_callback_key(
750 key: str,
751 extra: set[str] | None = None,
752) -> bool:
753 """Return ``True`` if ``key`` is present in ``extra`` (checked as-is), or
754 if its lowercase form is in ``_EXTRA_SENSITIVE_CALLBACK_KEYS``, or if
755 ``_CALLBACK_VAR_MASKER.is_sensitive_key`` matches it.
756 """
757 if extra and key in extra:
758 return True
759 if key.lower() in _EXTRA_SENSITIVE_CALLBACK_KEYS:
760 return True
761 return _CALLBACK_VAR_MASKER.is_sensitive_key(key)
764def _encrypt_if_plaintext(key: str, value: object) -> object:
765 if not isinstance(value, str) or not value:
766 return value
767 if not is_sensitive_callback_key(key):
768 return value
769 if value.startswith(_CALLBACK_VAR_ENCRYPTED_PREFIX):
770 # Already encrypted — round-tripping ciphertext (e.g. UI Edit Settings
771 # save without changing the field) must not double-encrypt. Cheap
772 # prefix check is robust under salt-key rotation; a decrypt-based
773 # idempotency check would mis-classify K1-encrypted blobs as
774 # plaintext under K2 and wrap them a second time.
775 return value
776 try:
777 return _CALLBACK_VAR_ENCRYPTED_PREFIX + encrypt_value_helper(value)
778 except Exception:
779 # No salt key / master key configured — leave the value as-is rather
780 # than crash the write. Dev environments without LITELLM_SALT_KEY hit
781 # this path; production always has a master key so encryption proceeds.
782 return value
785def _decrypt_or_passthrough(key: str, value: object) -> object:
786 if not isinstance(value, str) or not value:
787 return value
788 if not value.startswith(_CALLBACK_VAR_ENCRYPTED_PREFIX):
789 # Legacy plaintext rows or non-credential fields — return as-is.
790 return value
791 inner: Final = value[len(_CALLBACK_VAR_ENCRYPTED_PREFIX) :]
792 decrypted: Final = decrypt_value_helper(value=inner, key=key, exception_type="debug", return_original_value=False)
793 return decrypted if decrypted is not None else value