litellm/litellm/proxy/common_utils/callback_utils.py

import copy
import json
import os
from collections.abc import Callable, Iterable, Mapping
from dataclasses import dataclass
from itertools import accumulate
from typing import TYPE_CHECKING, Any, Final, Literal, NoReturn, Optional, TypeAlias

from typing_extensions import ReadOnly, TypedDict, assert_never

import litellm
from litellm import get_secret
from litellm._logging import verbose_proxy_logger
from litellm.constants import (
    CLIENT_OUTPUT_CEILING_METADATA_KEY,
    CONSUMED_REQUEST_TAGS_METADATA_KEY,
    MAX_GUARDRAIL_SCAN_METADATA_HEADER_LENGTH,
    PRE_CALL_EXECUTED_GUARDRAILS_KEY,
    ROUTING_REQUEST_TAGS_METADATA_KEY,
    SESSION_DEPLOYMENT_AFFINITY_TTL_METADATA_KEY,
)
from litellm.integrations.custom_logger import CustomLogger
from litellm.litellm_core_utils.core_helpers import (
    get_metadata_variable_name_from_kwargs,
    get_or_create_metadata_bucket,
)
from litellm.litellm_core_utils.sensitive_data_masker import SensitiveDataMasker
from litellm.proxy._types import CommonProxyErrors, LiteLLMPromptInjectionParams
from litellm.proxy.common_utils.encrypt_decrypt_utils import (
    decrypt_value_helper,
    encrypt_value_helper,
)
from litellm.proxy.types_utils.utils import get_instance_fn
from litellm.types.guardrails import GuardrailEventHooks
from litellm.types.utils import (
    StandardLoggingGuardrailInformation,
    StandardLoggingPayload,
)

_CALLBACK_VAR_MASKER: Final = SensitiveDataMasker()
# Compound names that are credential-bearing but don't contain any of the
# default sensitive segments (so SensitiveDataMasker won't flag them).
_EXTRA_SENSITIVE_CALLBACK_KEYS: Final = {"gcs_path_service_account"}
# Sentinel prefix on encrypted callback_var values. Lets us detect
# already-encrypted input cheaply (no decrypt-attempt round trip) and
# avoid double-encrypting if `LITELLM_SALT_KEY` is rotated between writes.
_CALLBACK_VAR_ENCRYPTED_PREFIX: Final = "litellm_enc::"
# Metadata slots that hold operator-configured callback and secret-manager setup
# (and therefore integration credentials). Resolved from UserAPIKeyAuth during
# pre-call setup, never read back off the copies stamped into request metadata.
_CALLBACK_CONFIG_SLOTS: Final = frozenset({"logging", "callback_settings", "secret_manager_settings"})

blue_color_code: Final = "\033[94m"
reset_color_code: Final = "\033[0m"

TRUSTED_PILLAR_RESPONSE_HEADERS_METADATA_KEY: Final = "_pillar_response_headers_trusted"

GUARDRAIL_SCAN_IDS_METADATA_KEY: Final = "guardrail_scan_ids"
GUARDRAIL_SCAN_METADATA_METADATA_KEY: Final = "guardrail_scan_metadata"


class GuardrailScanMetadata(TypedDict):
    guardrail: ReadOnly[str | None]
    stage: ReadOnly[str]
    provider: ReadOnly[str]
    scan_id: ReadOnly[str]


if TYPE_CHECKING:
    from litellm.litellm_core_utils.litellm_logging import Logging as LiteLLMLogging


@dataclass(frozen=True, slots=True)
class _CallbackResolvedToClass:
    entry: str
    loaded: type
    tag: Literal["resolved_to_class"] = "resolved_to_class"


@dataclass(frozen=True, slots=True)
class _CallbackNotDispatchable:
    entry: str
    loaded: object
    tag: Literal["not_dispatchable"] = "not_dispatchable"


_CallbackLoadError: TypeAlias = _CallbackResolvedToClass | _CallbackNotDispatchable


def _classify_loaded_callback(entry: str, loaded: object) -> CustomLogger | Callable[..., object] | _CallbackLoadError:
    """
    Decide whether what a ``litellm_settings.callbacks`` dotted path resolved to can be dispatched.

    A dotted path only ever runs as a ``CustomLogger`` instance or as a callback function. Anything
    else (most commonly a class instead of an instance) used to load without complaint and then be
    skipped on every request, with no log line and no error.
    """
    if isinstance(loaded, CustomLogger) or (callable(loaded) and not isinstance(loaded, type)):
        return loaded
    if isinstance(loaded, type):
        return _CallbackResolvedToClass(entry=entry, loaded=loaded)
    return _CallbackNotDispatchable(entry=entry, loaded=loaded)


def _raise_callback_load_error(error: _CallbackLoadError) -> NoReturn:
    """The one edge that raises: map a load error onto config load's failure contract."""
    match error:
        case _CallbackResolvedToClass():
            module_path: Final = error.entry.rsplit(".", 1)[0] if "." in error.entry else error.entry
            raise ValueError(
                f"litellm_settings.callbacks entry '{error.entry}' resolved to the class "
                f"{error.loaded.__module__}.{error.loaded.__qualname__}, which is neither a "
                "CustomLogger instance nor a callable, so the proxy would never run it."
                f" Point it at an instance instead, e.g. add `proxy_handler_instance = {error.loaded.__name__}()` to "
                f'{module_path} and set `callbacks: ["{module_path}.proxy_handler_instance"]`.'
            )
        case _CallbackNotDispatchable():
            raise ValueError(
                f"litellm_settings.callbacks entry '{error.entry}' resolved to "
                f"{type(error.loaded).__name__} {error.loaded!r}, which is neither a "
                "CustomLogger instance nor a callable, so the proxy would never run it."
            )
    assert_never(error)


def _loaded_callback_or_raise(entry: str, loaded: object) -> CustomLogger | Callable[..., object]:
    resolved: Final = _classify_loaded_callback(entry=entry, loaded=loaded)
    if isinstance(resolved, _CallbackResolvedToClass | _CallbackNotDispatchable):
        _raise_callback_load_error(resolved)
    return resolved


def initialize_callbacks_on_proxy(
    value: Any,
    premium_user: bool,
    config_file_path: str,
    litellm_settings: dict,
    callback_specific_params: dict | None = None,
):
    if not isinstance(callback_specific_params, dict):
        callback_specific_params = {}
    from litellm.integrations.custom_logger import CustomLogger
    from litellm.litellm_core_utils.logging_callback_manager import (
        LoggingCallbackManager,
    )
    from litellm.proxy.proxy_server import prisma_client

    verbose_proxy_logger.debug("%sinitializing callbacks=%s on proxy%s", blue_color_code, value, reset_color_code)
    if isinstance(value, list):
        imported_list: Final[list[Any]] = []
        for callback in value:  # ["presidio", <my-custom-callback>]
            if isinstance(callback, str) and callback == "compression_interception":
                from litellm.integrations.compression_interception.handler import (
                    CompressionInterceptionLogger,
                )

                compression_interception_obj = CompressionInterceptionLogger.initialize_from_proxy_config(
                    litellm_settings=litellm_settings,
                    callback_specific_params=callback_specific_params,
                )
                imported_list.append(compression_interception_obj)
                continue

            if isinstance(callback, str) and callback == "code_interpreter_interception":
                from litellm.integrations.code_interpreter_interception.handler import (
                    CodeInterpreterInterceptionLogger,
                )

                code_interpreter_interception_obj = CodeInterpreterInterceptionLogger.initialize_from_proxy_config(
                    litellm_settings=litellm_settings,
                    callback_specific_params=callback_specific_params,
                )
                imported_list.append(code_interpreter_interception_obj)
                continue

            # check if callback is a custom logger compatible callback
            if isinstance(callback, str):
                callback = LoggingCallbackManager._add_custom_callback_generic_api_str(callback)
            if isinstance(callback, str) and callback in litellm._known_custom_logger_compatible_callbacks:
                imported_list.append(callback)
            elif isinstance(callback, str) and callback == "presidio":
                from litellm.proxy.guardrails.guardrail_hooks.presidio import (
                    _OPTIONAL_PresidioPIIMasking,
                )

                presidio_logging_only: bool | None = litellm_settings.get("presidio_logging_only", None)
                if presidio_logging_only is not None:
                    presidio_logging_only = bool(presidio_logging_only)  # validate boolean given

                _presidio_params = {}
                if "presidio" in callback_specific_params and isinstance(callback_specific_params["presidio"], dict):
                    _presidio_params = callback_specific_params["presidio"]

                params: dict[str, Any] = {
                    "logging_only": presidio_logging_only,
                    **_presidio_params,
                }
                pii_masking_object = _OPTIONAL_PresidioPIIMasking(**params)
                imported_list.append(pii_masking_object)
            elif isinstance(callback, str) and callback == "llamaguard_moderations":
                try:
                    from litellm_enterprise.enterprise_callbacks.llama_guard import (
                        _ENTERPRISE_LlamaGuard,
                    )
                except ImportError:
                    raise Exception(
                        "MissingTrying to use Llama Guard" + CommonProxyErrors.missing_enterprise_package.value
                    )

                if premium_user is not True:
                    raise Exception("Trying to use Llama Guard" + CommonProxyErrors.not_premium_user.value)

                llama_guard_object = _ENTERPRISE_LlamaGuard()
                imported_list.append(llama_guard_object)
            elif isinstance(callback, str) and callback == "hide_secrets":
                try:
                    from litellm_enterprise.enterprise_callbacks.secret_detection import (
                        _ENTERPRISE_SecretDetection,
                    )
                except ImportError:
                    raise Exception(
                        "Trying to use Secret Detection" + CommonProxyErrors.missing_enterprise_package.value
                    )

                if premium_user is not True:
                    raise Exception("Trying to use secret hiding" + CommonProxyErrors.not_premium_user.value)

                _secret_detection_object = _ENTERPRISE_SecretDetection()
                imported_list.append(_secret_detection_object)
            elif isinstance(callback, str) and callback == "openai_moderations":
                try:
                    from enterprise.enterprise_hooks.openai_moderation import (
                        _ENTERPRISE_OpenAI_Moderation,
                    )
                except ImportError:
                    raise Exception(
                        "Trying to use OpenAI Moderations Check,"
                        + CommonProxyErrors.missing_enterprise_package_docker.value
                    )

                if premium_user is not True:
                    raise Exception("Trying to use OpenAI Moderations Check" + CommonProxyErrors.not_premium_user.value)

                openai_moderations_object = _ENTERPRISE_OpenAI_Moderation()
                imported_list.append(openai_moderations_object)
            elif isinstance(callback, str) and callback == "lakera_prompt_injection":
                from litellm.proxy.guardrails.guardrail_hooks.lakera_ai import (
                    lakeraAI_Moderation,
                )

                init_params = {}
                if "lakera_prompt_injection" in callback_specific_params and isinstance(
                    callback_specific_params["lakera_prompt_injection"], dict
                ):
                    init_params = callback_specific_params["lakera_prompt_injection"]
                lakera_moderations_object = lakeraAI_Moderation(**init_params)
                imported_list.append(lakera_moderations_object)
            elif isinstance(callback, str) and callback == "aporia_prompt_injection":
                from litellm.proxy.guardrails.guardrail_hooks.aporia_ai.aporia_ai import (
                    AporiaGuardrail,
                )

                aporia_guardrail_object = AporiaGuardrail()
                imported_list.append(aporia_guardrail_object)
            elif isinstance(callback, str) and callback == "google_text_moderation":
                try:
                    from enterprise.enterprise_hooks.google_text_moderation import (
                        _ENTERPRISE_GoogleTextModeration,
                    )
                except ImportError:
                    raise Exception(
                        "Trying to use Google Text Moderation,"
                        + CommonProxyErrors.missing_enterprise_package_docker.value
                    )

                if premium_user is not True:
                    raise Exception("Trying to use Google Text Moderation" + CommonProxyErrors.not_premium_user.value)

                google_text_moderation_obj = _ENTERPRISE_GoogleTextModeration()
                imported_list.append(google_text_moderation_obj)
            elif isinstance(callback, str) and callback == "llmguard_moderations":
                try:
                    from litellm_enterprise.enterprise_callbacks.llm_guard import (
                        _ENTERPRISE_LLMGuard,
                    )
                except ImportError:
                    raise Exception("Trying to use Llm Guard" + CommonProxyErrors.missing_enterprise_package.value)

                if premium_user is not True:
                    raise Exception("Trying to use Llm Guard" + CommonProxyErrors.not_premium_user.value)

                llm_guard_moderation_obj = _ENTERPRISE_LLMGuard()
                imported_list.append(llm_guard_moderation_obj)
            elif isinstance(callback, str) and callback == "blocked_user_check":
                try:
                    from enterprise.enterprise_hooks.blocked_user_list import (
                        _ENTERPRISE_BlockedUserList,
                    )
                except ImportError:
                    raise Exception(
                        "Trying to use Blocked User List" + CommonProxyErrors.missing_enterprise_package_docker.value
                    )

                if premium_user is not True:
                    raise Exception("Trying to use ENTERPRISE BlockedUser" + CommonProxyErrors.not_premium_user.value)

                blocked_user_list = _ENTERPRISE_BlockedUserList(prisma_client=prisma_client)
                imported_list.append(blocked_user_list)
            elif isinstance(callback, str) and callback == "banned_keywords":
                try:
                    from enterprise.enterprise_hooks.banned_keywords import (
                        _ENTERPRISE_BannedKeywords,
                    )
                except ImportError:
                    raise Exception(
                        "Trying to use Banned Keywords" + CommonProxyErrors.missing_enterprise_package_docker.value
                    )

                if premium_user is not True:
                    raise Exception("Trying to use ENTERPRISE BannedKeyword" + CommonProxyErrors.not_premium_user.value)

                banned_keywords_obj = _ENTERPRISE_BannedKeywords()
                imported_list.append(banned_keywords_obj)
            elif isinstance(callback, str) and callback == "detect_prompt_injection":
                from litellm.proxy.hooks.prompt_injection_detection import (
                    _OPTIONAL_PromptInjectionDetection,
                )

                prompt_injection_params = None
                if "prompt_injection_params" in litellm_settings:
                    prompt_injection_params_in_config = litellm_settings["prompt_injection_params"]
                    prompt_injection_params = LiteLLMPromptInjectionParams(**prompt_injection_params_in_config)

                prompt_injection_detection_obj = _OPTIONAL_PromptInjectionDetection(
                    prompt_injection_params=prompt_injection_params,
                )
                imported_list.append(prompt_injection_detection_obj)
            elif isinstance(callback, str) and callback == "batch_redis_requests":
                from litellm.proxy.hooks.batch_redis_get import (
                    _PROXY_BatchRedisRequests,
                )

                batch_redis_obj = _PROXY_BatchRedisRequests()
                imported_list.append(batch_redis_obj)
            elif isinstance(callback, str) and callback == "azure_content_safety":
                from litellm.proxy.hooks.azure_content_safety import (
                    _PROXY_AzureContentSafety,
                )

                azure_content_safety_params = litellm_settings["azure_content_safety_params"]
                for k, v in azure_content_safety_params.items():
                    if v is not None and isinstance(v, str) and v.startswith("os.environ/"):
                        azure_content_safety_params[k] = get_secret(v)

                azure_content_safety_obj = _PROXY_AzureContentSafety(
                    **azure_content_safety_params,
                )
                imported_list.append(azure_content_safety_obj)
            elif isinstance(callback, str) and callback == "websearch_interception":
                from litellm.integrations.websearch_interception.handler import (
                    WebSearchInterceptionLogger,
                )

                websearch_interception_obj = WebSearchInterceptionLogger.initialize_from_proxy_config(
                    litellm_settings=litellm_settings,
                    callback_specific_params=callback_specific_params,
                )
                imported_list.append(websearch_interception_obj)
            elif isinstance(callback, str) and callback == "datadog_cost_management":
                from litellm.integrations.datadog.datadog_cost_management import (
                    DatadogCostManagementLogger,
                )

                init_params = {}
                if "datadog_cost_management" in callback_specific_params and isinstance(
                    callback_specific_params["datadog_cost_management"], dict
                ):
                    init_params = callback_specific_params["datadog_cost_management"]
                datadog_cost_management_obj = DatadogCostManagementLogger(**init_params)
                imported_list.append(datadog_cost_management_obj)
            elif isinstance(callback, CustomLogger):
                imported_list.append(callback)
            else:
                verbose_proxy_logger.debug(
                    "%s attempting to import custom calback=%s %s", blue_color_code, callback, reset_color_code
                )
                imported_list.append(
                    _loaded_callback_or_raise(
                        entry=callback,
                        loaded=get_instance_fn(
                            value=callback,
                            config_file_path=config_file_path,
                        ),
                    )
                )
        if isinstance(litellm.callbacks, list):
            litellm.callbacks.extend(imported_list)
        else:
            litellm.callbacks = imported_list

        if "prometheus" in value:
            from litellm.integrations.prometheus import PrometheusLogger

            PrometheusLogger._mount_metrics_endpoint()
    else:
        litellm.callbacks = [
            _loaded_callback_or_raise(
                entry=value,
                loaded=get_instance_fn(
                    value=value,
                    config_file_path=config_file_path,
                ),
            )
        ]
    verbose_proxy_logger.debug("%s Initialized Callbacks - %s %s", blue_color_code, litellm.callbacks, reset_color_code)


def get_model_group_from_litellm_kwargs(kwargs: dict) -> str | None:
    _litellm_params: Final = kwargs.get("litellm_params", None) or {}
    _metadata: Final = _litellm_params.get(get_metadata_variable_name_from_kwargs(kwargs)) or {}
    _model_group: Final = _metadata.get("model_group", None)
    if _model_group is not None:
        return _model_group

    return None


def get_model_group_from_request_data(data: dict) -> str | None:
    _metadata: Final = data.get("metadata", None) or {}
    _model_group: Final = _metadata.get("model_group", None)
    if _model_group is not None:
        return _model_group

    return None


def get_remaining_tokens_and_requests_from_request_data(data: dict) -> dict[str, str]:
    """
    Helper function to return x-litellm-key-remaining-tokens-{model_group} and x-litellm-key-remaining-requests-{model_group}

    Returns {} when api_key + model rpm/tpm limit is not set

    """
    headers: Final = {}
    _metadata: Final = data.get("metadata", None) or {}
    model_group: Final = get_model_group_from_request_data(data)

    # The h11 package considers "/" or ":" invalid and raise a LocalProtocolError
    h11_model_group_name: Final = model_group.replace("/", "-").replace(":", "-") if model_group else None

    # Remaining Requests
    remaining_requests_variable_name: Final = f"litellm-key-remaining-requests-{model_group}"
    remaining_requests: Final = _metadata.get(remaining_requests_variable_name, None)
    if remaining_requests:
        headers[f"x-litellm-key-remaining-requests-{h11_model_group_name}"] = remaining_requests

    # Remaining Tokens
    remaining_tokens_variable_name: Final = f"litellm-key-remaining-tokens-{model_group}"
    remaining_tokens: Final = _metadata.get(remaining_tokens_variable_name, None)
    if remaining_tokens:
        headers[f"x-litellm-key-remaining-tokens-{h11_model_group_name}"] = remaining_tokens

    return headers


def _serialize_scan_metadata_header(entries: Iterable[object], *, max_length: int) -> str | None:
    """Compact JSON list of scan metadata entries, dropping trailing entries so the header fits in max_length."""
    encoded: Final = tuple(json.dumps(entry, separators=(",", ":")) for entry in entries)
    lengths: Final = tuple(accumulate(len(item) + 1 for item in encoded))
    kept: Final = sum(1 for length in lengths if length + 1 <= max_length)
    if kept == 0:
        return None
    return f"[{','.join(encoded[:kept])}]"


def get_logging_caching_headers(request_data: dict) -> dict | None:
    _metadata: Final[dict] = {}
    metadata_bucket: Final = request_data.get("metadata")
    litellm_metadata_bucket: Final = request_data.get("litellm_metadata")
    if isinstance(metadata_bucket, dict):
        _metadata.update(metadata_bucket)
    if isinstance(litellm_metadata_bucket, dict):
        # Batch/file routes store proxy tracking in litellm_metadata while
        # user-facing metadata stays in metadata; merge both for headers.
        _metadata.update(litellm_metadata_bucket)
    headers: Final = {}
    if "applied_guardrails" in _metadata:
        headers["x-litellm-applied-guardrails"] = ",".join(_metadata["applied_guardrails"])

    scan_ids: Final = _metadata.get(GUARDRAIL_SCAN_IDS_METADATA_KEY)
    if scan_ids:
        headers["x-litellm-guardrail-scan-id"] = ",".join(scan_ids)

    scan_metadata: Final = _metadata.get(GUARDRAIL_SCAN_METADATA_METADATA_KEY)
    scan_metadata_header: Final = (
        _serialize_scan_metadata_header(scan_metadata, max_length=MAX_GUARDRAIL_SCAN_METADATA_HEADER_LENGTH)
        if isinstance(scan_metadata, (list, tuple))
        else None
    )
    if scan_metadata_header:
        headers["x-litellm-guardrail-scan-metadata"] = scan_metadata_header

    if "applied_policies" in _metadata:
        headers["x-litellm-applied-policies"] = ",".join(_metadata["applied_policies"])

    if "policy_sources" in _metadata:
        sources: Final = _metadata["policy_sources"]
        if isinstance(sources, dict) and sources:
            # Use ';' as delimiter — matched_via reasons may contain commas
            headers["x-litellm-policy-sources"] = "; ".join(f"{name}={reason}" for name, reason in sources.items())

    if "semantic-similarity" in _metadata:
        headers["x-litellm-semantic-similarity"] = str(_metadata["semantic-similarity"])

    is_trusted_pillar_metadata: Final = _metadata.get(TRUSTED_PILLAR_RESPONSE_HEADERS_METADATA_KEY) is True
    pillar_headers: Final = _metadata.get("pillar_response_headers")
    if is_trusted_pillar_metadata and isinstance(pillar_headers, dict):
        headers.update(
            {
                key: str(value)
                for key, value in pillar_headers.items()
                if isinstance(key, str) and key.lower().startswith("x-pillar-")
            }
        )
    elif is_trusted_pillar_metadata and "pillar_flagged" in _metadata:
        headers["x-pillar-flagged"] = str(_metadata["pillar_flagged"]).lower()

    return headers


LITELLM_PROXY_INTERNAL_METADATA_KEYS: Final = frozenset(
    {
        "applied_policies",
        "applied_guardrails",
        GUARDRAIL_SCAN_IDS_METADATA_KEY,
        GUARDRAIL_SCAN_METADATA_METADATA_KEY,
        "policy_sources",
        "guardrails",
        "guardrail_config",
        "_guardrail_pipelines",
        "_pipeline_managed_guardrails",
        PRE_CALL_EXECUTED_GUARDRAILS_KEY,
        SESSION_DEPLOYMENT_AFFINITY_TTL_METADATA_KEY,
        CONSUMED_REQUEST_TAGS_METADATA_KEY,
        CLIENT_OUTPUT_CEILING_METADATA_KEY,
        ROUTING_REQUEST_TAGS_METADATA_KEY,
        "disable_global_guardrails",
        "disable_global_guardrail",
        "opted_out_global_guardrails",
        "pillar_response_headers",
        "_pillar_response_headers_trusted",
        "pillar_flagged",
        "pillar_scanners",
        "pillar_evidence",
        "pillar_evidence_truncated",
        "pillar_session_id_response",
        "standard_logging_object",
        "proxy_server_request",
        "secret_fields",
    }
)


def sanitize_openai_provider_metadata(
    metadata: Mapping[str, object] | None,
) -> Mapping[str, object] | None:
    """
    Keep only provider-safe OpenAI metadata entries (string keys -> string values).

    Strips LiteLLM proxy-internal tracking fields that must not be forwarded to
    OpenAI batch/file APIs.
    """
    if metadata is None:
        return None
    sanitized: Final[dict[str, str]] = {}
    for key, value in metadata.items():
        if key in LITELLM_PROXY_INTERNAL_METADATA_KEYS:
            continue
        if isinstance(value, str):
            sanitized[key] = value
        else:
            verbose_proxy_logger.debug(
                "sanitize_openai_provider_metadata: dropping key %r with non-string value of type %s",
                key,
                type(value).__name__,
            )
    return None if metadata and not sanitized else sanitized


def add_guardrail_to_applied_guardrails_header(request_data: dict, guardrail_name: str | None):
    if guardrail_name is None:
        return
    _, _metadata = get_or_create_metadata_bucket(request_data)
    if "applied_guardrails" in _metadata:
        if guardrail_name not in _metadata["applied_guardrails"]:
            _metadata["applied_guardrails"].append(guardrail_name)
    else:
        _metadata["applied_guardrails"] = [guardrail_name]


def add_guardrail_scan_id(
    request_data: dict[str, object],
    scan_id: str | None,
    *,
    guardrail_name: str | None,
    provider: str,
    stage: GuardrailEventHooks,
) -> None:
    """
    Record a provider scan id, keyed to the guardrail execution that produced it, so it can be surfaced to the caller.

    Guardrails only return scan details to the client when they block, so allowed requests carry no
    audit trail. Ids recorded here become the x-litellm-guardrail-scan-id response header, and the
    (guardrail, stage, provider, scan_id) entries become the x-litellm-guardrail-scan-metadata header.
    """
    if not scan_id:
        return
    _, _metadata = get_or_create_metadata_bucket(request_data)
    existing: Final = _metadata.get(GUARDRAIL_SCAN_IDS_METADATA_KEY)
    scan_ids: Final[tuple[object, ...]] = tuple(existing) if isinstance(existing, (list, tuple)) else ()
    if scan_id not in scan_ids:
        _metadata[GUARDRAIL_SCAN_IDS_METADATA_KEY] = (*scan_ids, scan_id)

    entry: Final[GuardrailScanMetadata] = {
        "guardrail": guardrail_name,
        "stage": stage.value,
        "provider": provider,
        "scan_id": scan_id,
    }
    existing_entries: Final = _metadata.get(GUARDRAIL_SCAN_METADATA_METADATA_KEY)
    entries: Final[tuple[object, ...]] = tuple(existing_entries) if isinstance(existing_entries, (list, tuple)) else ()
    if entry not in entries:
        _metadata[GUARDRAIL_SCAN_METADATA_METADATA_KEY] = (*entries, entry)


def add_policy_to_applied_policies_header(request_data: dict, policy_name: str | None):
    """
    Add a policy name to the applied_policies list in request metadata.

    This is used to track which policies were applied to a request,
    similar to how applied_guardrails tracks guardrails.
    """
    if policy_name is None:
        return
    _, _metadata = get_or_create_metadata_bucket(request_data)
    if "applied_policies" in _metadata:
        if policy_name not in _metadata["applied_policies"]:
            _metadata["applied_policies"].append(policy_name)
    else:
        _metadata["applied_policies"] = [policy_name]


def add_policy_sources_to_metadata(request_data: dict, policy_sources: dict[str, str]):
    """
    Store policy match reasons in metadata for x-litellm-policy-sources header.

    Args:
        request_data: The request data dict
        policy_sources: Map of policy_name -> matched_via reason
    """
    if not policy_sources:
        return
    _, _metadata = get_or_create_metadata_bucket(request_data)
    existing = _metadata.get("policy_sources", {})
    if not isinstance(existing, dict):
        existing = {}
    existing.update(policy_sources)
    _metadata["policy_sources"] = existing


def add_guardrail_response_to_standard_logging_object(
    litellm_logging_obj: Optional["LiteLLMLogging"],
    guardrail_response: StandardLoggingGuardrailInformation,
):
    if litellm_logging_obj is None:
        return
    standard_logging_object: Final[StandardLoggingPayload | None] = litellm_logging_obj.model_call_details.get(
        "standard_logging_object"
    )
    if standard_logging_object is None:
        return
    guardrail_information = standard_logging_object.get("guardrail_information", [])
    if guardrail_information is None:
        guardrail_information = []
    guardrail_information.append(guardrail_response)
    standard_logging_object["guardrail_information"] = guardrail_information

    return standard_logging_object


def process_callback(_callback: str, callback_type: str, environment_variables: dict) -> dict:
    """Process a single callback and return its data with environment variables"""
    env_vars: Final = CustomLogger.get_callback_env_vars(_callback)

    env_vars_dict: Final[dict[str, str | None]] = {}
    for _var in env_vars:
        stored_value = environment_variables.get(_var, None)
        env_vars_dict[_var] = stored_value if stored_value is not None else os.getenv(_var)

    return {"name": _callback, "variables": env_vars_dict, "type": callback_type}


def normalize_callback_names(callbacks: Iterable[object] | None) -> list[object]:
    if callbacks is None:
        return []
    return [c.lower() if isinstance(c, str) else c for c in callbacks]


def strip_callback_config(metadata: dict[str, object] | None) -> dict[str, object] | None:
    """Return key/team metadata without the slots that carry callback credentials."""
    if not isinstance(metadata, dict):
        return metadata
    return {k: v for k, v in metadata.items() if k not in _CALLBACK_CONFIG_SLOTS}


def encrypt_callback_vars(metadata: Any) -> Any:
    """Return a deep copy of metadata with callback_vars values encrypted at rest.

    Idempotent: a value that already decrypts cleanly is left unchanged so
    round-trips through edit forms don't double-encrypt.
    """
    return _transform_callback_vars(metadata, _encrypt_if_plaintext)


def decrypt_callback_vars(metadata: Any) -> Any:
    """Return a deep copy of metadata with callback_vars values decrypted.

    Legacy plaintext rows pass through unchanged (decrypt failure → original).
    """
    return _transform_callback_vars(metadata, _decrypt_or_passthrough)


def _transform_callback_vars(metadata: object, transform: Callable[[str, Any], Any]) -> object:
    if not isinstance(metadata, dict):
        return metadata
    out: Final = copy.deepcopy(metadata)
    logging_entries: Final = out.get("logging")
    if isinstance(logging_entries, list):
        for entry in logging_entries:
            if isinstance(entry, dict) and isinstance(entry.get("callback_vars"), dict):
                entry["callback_vars"] = {k: transform(k, v) for k, v in entry["callback_vars"].items()}
    callback_settings: Final = out.get("callback_settings")
    if isinstance(callback_settings, dict) and isinstance(callback_settings.get("callback_vars"), dict):
        callback_settings["callback_vars"] = {k: transform(k, v) for k, v in callback_settings["callback_vars"].items()}
    return out


def is_sensitive_callback_key(
    key: str,
    extra: set[str] | None = None,
) -> bool:
    """Return ``True`` if ``key`` is present in ``extra`` (checked as-is), or
    if its lowercase form is in ``_EXTRA_SENSITIVE_CALLBACK_KEYS``, or if
    ``_CALLBACK_VAR_MASKER.is_sensitive_key`` matches it.
    """
    if extra and key in extra:
        return True
    if key.lower() in _EXTRA_SENSITIVE_CALLBACK_KEYS:
        return True
    return _CALLBACK_VAR_MASKER.is_sensitive_key(key)


def _encrypt_if_plaintext(key: str, value: object) -> object:
    if not isinstance(value, str) or not value:
        return value
    if not is_sensitive_callback_key(key):
        return value
    if value.startswith(_CALLBACK_VAR_ENCRYPTED_PREFIX):
        # Already encrypted — round-tripping ciphertext (e.g. UI Edit Settings
        # save without changing the field) must not double-encrypt. Cheap
        # prefix check is robust under salt-key rotation; a decrypt-based
        # idempotency check would mis-classify K1-encrypted blobs as
        # plaintext under K2 and wrap them a second time.
        return value
    try:
        return _CALLBACK_VAR_ENCRYPTED_PREFIX + encrypt_value_helper(value)
    except Exception:
        # No salt key / master key configured — leave the value as-is rather
        # than crash the write. Dev environments without LITELLM_SALT_KEY hit
        # this path; production always has a master key so encryption proceeds.
        return value


def _decrypt_or_passthrough(key: str, value: object) -> object:
    if not isinstance(value, str) or not value:
        return value
    if not value.startswith(_CALLBACK_VAR_ENCRYPTED_PREFIX):
        # Legacy plaintext rows or non-credential fields — return as-is.
        return value
    inner: Final = value[len(_CALLBACK_VAR_ENCRYPTED_PREFIX) :]
    decrypted: Final = decrypt_value_helper(value=inner, key=key, exception_type="debug", return_original_value=False)
    return decrypted if decrypted is not None else value