scriptoriumadmin/litellm
1
0
Fork 0
mirror of https://github.com/BerriAI/litellm.git synced 2026-10-07 02:59:05 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 103

fix(logging): redact failure error text and honor excluded fields on failure callbacks

Browse source 
Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com>
This commit is contained in:
yucheng 2026-09-29 08:21:55 +00:00
parent 85dc7cb62e
commit d733288425
16 changed files with 933 additions and 18 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

22
litellm/integrations/custom_logger.py
View file

@ -4,7 +4,7 @@ import re
import traceback
from collections.abc import AsyncGenerator, Mapping, Sequence
from types import MappingProxyType
from typing import TYPE_CHECKING, Any, ClassVar, Final, Optional
from typing import TYPE_CHECKING, Any, ClassVar, Final, Optional, cast
from pydantic import BaseModel
@ -24,6 +24,7 @@ from litellm.types.utils import (
StandardAuditLogPayload,
StandardCallbackDynamicParams,
StandardLoggingPayload,
StandardLoggingPayloadErrorInformation,
)
if TYPE_CHECKING:
@ -960,6 +961,19 @@ class CustomLogger: # https://docs.litellm.ai/docs/observability/custom_callbac
model_response_dict: Final = model_response.model_dump()
standard_logging_object_copy["response"] = model_response_dict
if turn_off_message_logging:
# Error text can quote the prompt; callbacks never scrub it themselves,
# so redact it regardless of `redacts_messages_itself`.
if standard_logging_object_copy.get("error_str"):
standard_logging_object_copy["error_str"] = "redacted-by-litellm"
error_information: Final = standard_logging_object_copy.get("error_information")
if isinstance(error_information, Mapping):
from litellm.litellm_core_utils.redact_messages import redact_error_information
standard_logging_object_copy["error_information"] = redact_error_information(
cast(StandardLoggingPayloadErrorInformation, dict(error_information))
)
params: Final = model_call_details.get("litellm_params")
request: Final = params.get("proxy_server_request") if isinstance(params, dict) else None
redacted_params: Final = (
@ -967,9 +981,15 @@ class CustomLogger: # https://docs.litellm.ai/docs/observability/custom_callbac
if turn_off_message_logging and isinstance(params, dict) and isinstance(request, dict)
else EMPTY_MAPPING
)
redacted_failure_fields: Final = (
MappingProxyType({"traceback_exception": "redacted-by-litellm"})
if turn_off_message_logging and model_call_details.get("traceback_exception")
else EMPTY_MAPPING
)
return {
**model_call_details,
**redacted_params,
**redacted_failure_fields,
"standard_logging_object": standard_logging_object_copy,
}

11
litellm/integrations/datadog/datadog.py
View file

@ -303,12 +303,21 @@ class DataDogLogger(
from litellm.litellm_core_utils.litellm_logging import (
StandardLoggingPayloadSetup,
)
from litellm.litellm_core_utils.redact_messages import (
redact_error_information,
should_redact_failed_request,
)
from litellm.litellm_core_utils.safe_json_dumps import safe_dumps
error_information: Final = StandardLoggingPayloadSetup.get_error_information(
_error_information_raw: Final = StandardLoggingPayloadSetup.get_error_information(
original_exception=original_exception,
traceback_str=traceback_str,
)
error_information: Final = (
redact_error_information(_error_information_raw)
if should_redact_failed_request(request_data)
else _error_information_raw
)
_code: Final = error_information.get("error_code") or ""
status_code: int | None = None
if _code and str(_code).strip().isdigit():

16
litellm/integrations/mlflow.py
View file

@ -3,7 +3,9 @@ import threading
from typing import TYPE_CHECKING, Any, Final
from litellm._logging import verbose_logger
from litellm.constants import REDACTED_BY_LITELLM
from litellm.integrations.custom_logger import CustomLogger
from litellm.litellm_core_utils.redact_messages import should_redact_message_logging
if TYPE_CHECKING:
from litellm.types.utils import StandardLoggingPayload
@ -88,7 +90,19 @@ class MlflowLogger(CustomLogger):
# Record exception info as event
if exception := kwargs.get("exception"):
span.add_event(SpanEvent.from_exception(exception))
if should_redact_message_logging(kwargs):
span.add_event(
SpanEvent(
name="exception",
attributes={
"exception.type": type(exception).__name__,
"exception.message": REDACTED_BY_LITELLM,
"exception.stacktrace": REDACTED_BY_LITELLM,
},
)
)
else:
span.add_event(SpanEvent.from_exception(exception))
self._extract_and_set_chat_attributes(span, kwargs, response_obj)
self._end_span_or_trace(

45
litellm/integrations/opentelemetry.py
View file

@ -10,6 +10,7 @@ from typing import TYPE_CHECKING, Any, Final, TypedDict, cast
import litellm
from litellm._logging import verbose_logger
from litellm.constants import REDACTED_BY_LITELLM
from litellm.integrations._types.open_inference import (
OpenInferenceSpanKindValues,
SpanAttributes,
@ -930,17 +931,26 @@ class OpenTelemetry(OTELGenAISemconvMixin, CustomLogger):
from litellm.litellm_core_utils.litellm_logging import (
StandardLoggingPayloadSetup,
)
from litellm.litellm_core_utils.redact_messages import (
redact_error_information,
should_redact_failed_request,
)
error_information: Final = StandardLoggingPayloadSetup.get_error_information(
redact: Final = should_redact_failed_request(request_data)
_error_information_raw: Final = StandardLoggingPayloadSetup.get_error_information(
original_exception=original_exception,
traceback_str=traceback_str,
)
error_information: Final = (
redact_error_information(_error_information_raw) if redact else _error_information_raw
)
self._record_exception_on_span(
span=parent_otel_span,
kwargs={
"exception": original_exception,
"standard_logging_object": {"error_information": error_information},
},
redact_content=redact,
)
# _record_exception_on_span only stamps when error_code is set;
@ -963,7 +973,7 @@ class OpenTelemetry(OTELGenAISemconvMixin, CustomLogger):
self.safe_set_attribute(
span=exception_logging_span,
key="exception",
value=str(original_exception),
value=REDACTED_BY_LITELLM if redact else str(original_exception),
)
self._set_team_attributes_on_span(
span=exception_logging_span,
@ -2207,25 +2217,40 @@ class OpenTelemetry(OTELGenAISemconvMixin, CustomLogger):
):
parent_otel_span.end(end_time=self._to_ns(end_time))
def _record_exception_on_span(self, span: Span, kwargs: dict):
def _record_exception_on_span(self, span: Span, kwargs: dict, redact_content: bool | None = None):
"""
Record exception information on the span using OTEL standard methods.
This extracts error information from StandardLoggingPayload and:
1. Uses span.record_exception() for the actual exception object (OTEL standard)
2. Sets structured error attributes from StandardLoggingPayloadErrorInformation
``redact_content`` overrides the message-redaction decision; when unset it is
resolved from kwargs via ``should_redact_message_logging``.
"""
try:
from litellm.integrations._types.open_inference import (
ErrorAttributes,
)
from litellm.litellm_core_utils.redact_messages import should_redact_message_logging
redact: Final = should_redact_message_logging(kwargs) if redact_content is None else redact_content
# Get the exception object if available
exception: Final = kwargs.get("exception")
# Record the exception using OTEL's standard method
if exception is not None:
span.record_exception(exception)
if redact:
span.record_exception(
exception,
attributes={
"exception.message": REDACTED_BY_LITELLM,
"exception.stacktrace": REDACTED_BY_LITELLM,
},
)
else:
span.record_exception(exception)
# Get StandardLoggingPayload for structured error information
standard_logging_payload: Final[StandardLoggingPayload | None] = kwargs.get("standard_logging_object")
@ -3571,12 +3596,22 @@ class OpenTelemetry(OTELGenAISemconvMixin, CustomLogger):
from litellm.litellm_core_utils.litellm_logging import (
StandardLoggingPayloadSetup,
)
from litellm.litellm_core_utils.redact_messages import (
redact_error_information,
should_redact_message_logging,
)
redact: Final = should_redact_message_logging({})
error_information: Final = StandardLoggingPayloadSetup.get_error_information(original_exception=exception)
error_information["error_code"] = str(status_code)
self._record_exception_on_span(
span=span,
kwargs={"standard_logging_object": {"error_information": error_information}},
kwargs={
"standard_logging_object": {
"error_information": redact_error_information(error_information) if redact else error_information
}
},
redact_content=redact,
)
def set_preprocessing_duration_attribute(self, span: Span | None, container: object) -> None:

34
litellm/integrations/otel/logger.py
View file

@ -26,6 +26,7 @@ from typing_extensions import TypedDict, Unpack
import litellm
from litellm._logging import verbose_logger
from litellm.constants import REDACTED_BY_LITELLM
from litellm.integrations.custom_logger import CustomLogger
from litellm.integrations.otel.emitter import SpanEmitter, stamp_error
from litellm.integrations.otel.mappers import resolve_mappers
@ -77,6 +78,10 @@ from litellm.integrations.otel.plumbing.providers import (
resolve_meter_provider,
)
from litellm.integrations.otel.plumbing.routing import TenantTracerCache
from litellm.litellm_core_utils.redact_messages import (
should_redact_failed_request,
should_redact_message_logging,
)
if TYPE_CHECKING:
from opentelemetry.metrics import MeterProvider
@ -99,6 +104,7 @@ def _span_error_from_exception(
*,
status_code: int | None = None,
traceback_str: str | None = None,
redact_content: bool = False,
) -> SpanError:
"""A ``SpanError`` for a proxy-level failure that never produced a
``StandardLoggingPayload`` (auth / validation / malformed-body rejections),
@ -113,9 +119,13 @@ def _span_error_from_exception(
)
return SpanError(
error_type=info.get("error_class") or info.get("error_code") or None,
message=info.get("error_message") or None,
message=REDACTED_BY_LITELLM
if redact_content and info.get("error_message")
else (info.get("error_message") or None),
code=str(status_code) if status_code is not None else (info.get("error_code") or None),
stack_trace=info.get("traceback") or None,
stack_trace=REDACTED_BY_LITELLM
if redact_content and info.get("traceback")
else (info.get("traceback") or None),
llm_provider=info.get("llm_provider") or None,
)
@ -741,7 +751,12 @@ class OpenTelemetryV2(CustomLogger):
yield span
except Exception as exc:
if is_recordable_span(span):
stamp_error(span, _span_error_from_exception(exc), record_event=False, set_status=False)
stamp_error(
span,
_span_error_from_exception(exc, redact_content=should_redact_message_logging({})),
record_event=False,
set_status=False,
)
raise
async def async_pre_call_hook(
@ -782,7 +797,9 @@ class OpenTelemetryV2(CustomLogger):
already_stamped: Final = Error.TYPE in (getattr(span, "attributes", None) or ())
stamp_error(
span,
_span_error_from_exception(exception, status_code=status_code),
_span_error_from_exception(
exception, status_code=status_code, redact_content=should_redact_message_logging({})
),
record_event=not already_stamped,
)
@ -808,7 +825,14 @@ class OpenTelemetryV2(CustomLogger):
span: Final = mcp_message_transport_span() or request_root_span() or user_api_key_dict.parent_otel_span
if span is None or not is_recordable_span(span):
return
stamp_error(span, _span_error_from_exception(original_exception, traceback_str=traceback_str))
stamp_error(
span,
_span_error_from_exception(
original_exception,
traceback_str=traceback_str,
redact_content=should_redact_failed_request(request_data),
),
)
return
def emit_guardrail_span(self, entry: "StandardLoggingGuardrailInformation") -> None:

14
litellm/litellm_core_utils/litellm_logging.py
View file

@ -3780,7 +3780,9 @@ class Logging(LiteLLMLoggingBaseClass):
start_time=start_time,
end_time=end_time,
response_obj=result,
kwargs=self.model_call_details,
kwargs=callback.redact_standard_logging_payload_from_model_call_details(
model_call_details=self.model_call_details
),
)
if callback == "langfuse":
global langFuseLogger
@ -3898,8 +3900,12 @@ class Logging(LiteLLMLoggingBaseClass):
global_callbacks=litellm._async_failure_callback,
)
result: Final = None # result sent to all loggers, init this to None incase it's not created
result: object = None # result sent to all loggers, init this to None incase it's not created
result = redact_message_input_output_from_logging(
model_call_details=(self.model_call_details if hasattr(self, "model_call_details") else {}),
result=result,
)
self.has_run_logging(event_type="async_failure")
for callback in callbacks:
try:
@ -3913,7 +3919,9 @@ class Logging(LiteLLMLoggingBaseClass):
continue
if isinstance(callback, CustomLogger): # custom logger class
await callback.async_log_failure_event(
kwargs=self.model_call_details,
kwargs=callback.redact_standard_logging_payload_from_model_call_details(
model_call_details=self.model_call_details
),
response_obj=result,
start_time=start_time,
end_time=end_time,

53
litellm/litellm_core_utils/redact_messages.py
View file

@ -11,7 +11,7 @@ import asyncio
import copy
import inspect
from collections.abc import Mapping
from typing import TYPE_CHECKING, Any, Final
from typing import TYPE_CHECKING, Any, Final, cast
import litellm
from litellm.constants import REDACTED_BY_LITELLM
@ -26,7 +26,10 @@ from litellm.llms.vertex_ai.common_utils import (
redact_vertex_ai_metadata_from_logged_object,
)
from litellm.secret_managers.main import str_to_bool
from litellm.types.utils import StandardCallbackDynamicParams
from litellm.types.utils import (
StandardCallbackDynamicParams,
StandardLoggingPayloadErrorInformation,
)
if TYPE_CHECKING:
from litellm.litellm_core_utils.litellm_logging import (
@ -183,6 +186,42 @@ def redacted_standard_logging_payload(payload: Mapping[str, object]) -> Mapping[
return _redact_standard_logging_object(payload)
def redact_error_information(
error_information: StandardLoggingPayloadErrorInformation,
) -> StandardLoggingPayloadErrorInformation:
"""
Return a copy of ``StandardLoggingPayloadErrorInformation`` with the fields that can
quote the prompt (``error_message``, ``traceback``) replaced by ``REDACTED_BY_LITELLM``
when they are non-empty strings. Every other field is carried over unchanged.
"""
redacted: Final = dict(error_information)
for field in ("error_message", "traceback"):
value: Final = redacted.get(field)
if isinstance(value, str) and value:
redacted[field] = REDACTED_BY_LITELLM
return cast(StandardLoggingPayloadErrorInformation, redacted)
def should_redact_failed_request(request_data: Mapping[str, object]) -> bool:
"""
Message-redaction decision for proxy ``async_post_call_failure_hook`` callbacks, which
only receive ``request_data`` (the Logging object is popped before hooks run).
``litellm_metadata`` is included only when present so
``get_metadata_variable_name_from_kwargs`` resolves ``metadata`` for chat routes.
"""
litellm_params: Final[dict[str, object]] = {"metadata": request_data.get("metadata")}
if "litellm_metadata" in request_data:
litellm_params["litellm_metadata"] = request_data.get("litellm_metadata")
return should_redact_message_logging(
{
"litellm_params": litellm_params,
"standard_callback_dynamic_params": {
"turn_off_message_logging": request_data.get("turn_off_message_logging")
},
}
)
def _redact_standard_logging_object(payload: Mapping[str, object]) -> dict[str, object]:
standard_logging_object: Final = copy.deepcopy(without_classifier_audit(payload))
redacted_str: Final = REDACTED_BY_LITELLM
@ -207,6 +246,14 @@ def _redact_standard_logging_object(payload: Mapping[str, object]) -> dict[str,
else:
# For other formats (empty dict, None, etc.), use simple text format
standard_logging_object["response"] = {"text": redacted_str}
if standard_logging_object.get("error_str"):
standard_logging_object["error_str"] = redacted_str
error_information: Final = standard_logging_object.get("error_information")
if isinstance(error_information, Mapping):
standard_logging_object["error_information"] = redact_error_information(
cast(StandardLoggingPayloadErrorInformation, dict(error_information))
)
return standard_logging_object
@ -272,6 +319,8 @@ def perform_redaction(model_call_details: dict, result, redact_streaming_respons
standard_logging_object: Final = model_call_details.get("standard_logging_object")
if isinstance(standard_logging_object, Mapping):
model_call_details["standard_logging_object"] = _redact_standard_logging_object(standard_logging_object)
if isinstance(model_call_details.get("traceback_exception"), str) and model_call_details["traceback_exception"]:
model_call_details["traceback_exception"] = REDACTED_BY_LITELLM
redact_vertex_ai_metadata_from_litellm_params(model_call_details)
# Redact streaming response

6
litellm/proxy/hooks/proxy_track_cost_callback.py
View file

@ -16,6 +16,10 @@ from litellm.litellm_core_utils.core_helpers import (
)
from litellm.litellm_core_utils.litellm_logging import StandardLoggingPayloadSetup
from litellm.litellm_core_utils.llm_cost_calc.guardrail_cost import guardrail_information_cost
from litellm.litellm_core_utils.redact_messages import (
redact_error_information,
should_redact_failed_request,
)
from litellm.proxy._types import UserAPIKeyAuth
from litellm.proxy.auth.auth_checks import (
get_key_object,
@ -171,6 +175,8 @@ class _ProxyDBLogger(CustomLogger):
original_exception=original_exception,
traceback_str=traceback_str,
)
if should_redact_failed_request(request_data):
_error_information = redact_error_information(_error_information)
if should_suppress_spend_log_tracebacks():
# Drop the traceback key entirely so the per-row Metadata pane in
# the UI (which renders the JSON blob verbatim) doesn't show a

164
tests/integration/observability/test_failure_redaction.py Normal file
View file

@ -0,0 +1,164 @@
import json
import uuid
from collections.abc import Iterator
from dataclasses import dataclass
from pathlib import Path
from typing import Final
import httpx
import pytest
import yaml
from integration._support.client import Gateway, JsonValue, eventually, gateway_from_environment, object_value
from integration._support.database import read_rows
from integration._support.process import owned_proxy
from integration._support.wire import Reply, Request, Wire, wire_server
@dataclass(frozen=True, slots=True)
class Rig:
proxy: Gateway
provider: Wire
sink: Wire
batches: list[Request]
def failure_events(self, model: str) -> tuple[dict[str, JsonValue], ...]:
self.batches.extend(self.sink.drain())
return tuple(
object_value(event)
for batch in self.batches
for event in json.loads(batch.body)
if model in json.dumps(event)
)
def _provider(request: Request) -> Reply:
text: Final = json.loads(request.body)["messages"][-1]["content"]
return Reply(
status=400,
body=json.dumps(
{"error": {"type": "invalid_request_error", "message": f"Unsupported content: {text}"}}
).encode(),
)
@pytest.fixture(scope="module")
def rig(tmp_path_factory: pytest.TempPathFactory) -> Iterator[Rig]:
root: Final = tmp_path_factory.mktemp("failure_redaction")
config: Final = yaml.safe_load(Path("tests/integration/proxy_config.yaml").read_text())
config["litellm_settings"].update(
{
"callbacks": ["generic_api"],
"DEFAULT_FLUSH_INTERVAL_SECONDS": 1,
"turn_off_message_logging": True,
"standard_logging_payload_excluded_fields": ["hidden_params"],
}
)
path: Final = root / "failure_redaction.yaml"
path.write_text(yaml.safe_dump(config))
with (
gateway_from_environment() as gateway,
wire_server(_provider) as provider,
wire_server(lambda _: Reply()) as sink,
owned_proxy(gateway, root, {"GENERIC_LOGGER_ENDPOINT": sink.url}, config=path) as proxy,
):
yield Rig(proxy, provider, sink, []) # mutable-ok: sink drain consumes batches, later polls keep earlier ones
def _secret_prompt() -> str:
return "confidential-prompt-" + uuid.uuid4().hex
def _single_failure_event(rig: Rig, model: str) -> dict[str, JsonValue]:
events: Final = eventually(lambda: rig.failure_events(model), lambda values: len(values) >= 1, seconds=20)
assert len(events) == 1, events
assert events[0]["status"] == "failure", events[0]
return events[0]
def _spend_row(call_id: str) -> dict[str, JsonValue]:
rows: Final = eventually(
lambda: read_rows(
'SELECT status, messages, response, proxy_server_request, metadata FROM "LiteLLM_SpendLogs" '
"WHERE request_id=%s",
(call_id,),
),
lambda values: len(values) == 1,
seconds=70,
)
return rows[0]
def _chat(rig: Rig, model: str, messages: list[JsonValue], key: str | None = None) -> httpx.Response:
return rig.proxy.request("POST", "/v1/chat/completions", {"model": model, "messages": messages}, key=key)
def _error_information(event: dict[str, JsonValue]) -> dict[str, JsonValue]:
return object_value(event["error_information"])
def test_provider_error_echoing_the_prompt_is_redacted_in_callbacks_and_spend_logs(rig: Rig) -> None:
secret: Final = _secret_prompt()
with rig.proxy.scenario() as scenario:
model: Final = scenario.model(
api_base=rig.provider.url + "/v1", api_key="synthetic-provider-key", num_retries=0
)
response: Final = _chat(rig, model, [{"role": "user", "content": secret}])
assert response.status_code == 400, response.text
assert any(secret.encode() in request.body for request in rig.provider.drain())
event: Final = _single_failure_event(rig, model)
assert secret not in json.dumps(event), json.dumps(event)
error_information: Final = _error_information(event)
assert error_information["error_class"] == "BadRequestError", error_information
assert error_information["error_code"] == "400", error_information
assert error_information["llm_provider"] == "openai", error_information
assert "hidden_params" not in event, sorted(event)
row: Final = _spend_row(response.headers["x-litellm-call-id"])
assert row["status"] == "failure", row
assert secret not in json.dumps(row, default=str), row
persisted: Final = object_value(
json.loads(row["metadata"]) if isinstance(row["metadata"], str) else row["metadata"]
)
assert object_value(persisted["error_information"])["error_class"] == "BadRequestError", persisted
def test_transformation_error_quoting_the_prompt_is_redacted_in_callbacks(rig: Rig) -> None:
secret: Final = _secret_prompt()
with rig.proxy.scenario() as scenario:
model: Final = scenario.model(
model="anthropic/claude-sonnet-5", api_base=rig.provider.url, api_key="synthetic-anthropic-key"
)
response: Final = _chat(
rig,
model,
[
{"role": "user", "content": secret},
{
"role": "assistant",
"content": None,
"tool_calls": [{"id": 12345, "type": "function", "function": {"name": None, "arguments": "{}"}}],
},
{"role": "tool", "tool_call_id": 12345, "content": secret},
],
)
assert response.status_code == 400, response.text
assert rig.provider.drain() == ()
event: Final = _single_failure_event(rig, model)
assert secret not in json.dumps(event), json.dumps(event)
error_information: Final = _error_information(event)
assert error_information["error_code"] == "400", error_information
assert error_information["error_class"], error_information
def test_proxy_only_rejection_does_not_leak_the_prompt_to_callbacks(rig: Rig) -> None:
secret: Final = _secret_prompt()
with rig.proxy.scenario() as scenario:
allowed: Final = scenario.model(api_base=rig.provider.url + "/v1", api_key="synthetic-provider-key")
denied: Final = scenario.model(api_base=rig.provider.url + "/v1", api_key="synthetic-provider-key")
key: Final = scenario.key(models=[allowed])
response: Final = _chat(rig, denied, [{"role": "user", "content": secret}], key=key)
assert response.status_code in (401, 403), response.text
assert rig.provider.drain() == ()
event: Final = _single_failure_event(rig, denied)
assert secret not in json.dumps(event), json.dumps(event)
assert "hidden_params" not in event, sorted(event)
assert _error_information(event)["error_code"] == str(response.status_code), event

43
tests/unit/integrations/datadog/test_datadog_logger_batching.py
View file

@ -1,4 +1,5 @@
import asyncio
import json
from unittest.mock import AsyncMock, Mock, patch
import httpx
@ -6,6 +7,7 @@ import pytest
from httpx import Request, Response
from pydantic import BaseModel, computed_field
import litellm
from litellm.integrations.datadog.datadog import DataDogLogger
from litellm.llms.custom_httpx.http_handler import MaskedHTTPStatusError
from litellm.types.integrations.datadog import (
@ -558,3 +560,44 @@ async def test_raised_intake_error_preserves_datadog_requeue_behavior(datadog_en
await logger.async_send_batch()
assert [event["message"] for event in logger.log_queue] == ['{"event": 0}', '{"event": 1}']
@pytest.mark.asyncio
async def test_failure_hook_redacts_exception_payload_when_redaction_on(datadog_env, monkeypatch):
monkeypatch.setattr(litellm, "turn_off_message_logging", True)
with patch("asyncio.create_task"):
logger = DataDogLogger()
secret = "secret-prompt-marker"
await logger.async_post_call_failure_hook(
request_data={"metadata": {}},
original_exception=litellm.BadRequestError(
message=f"Unsupported content: {secret}", model="gpt-4o", llm_provider="openai"
),
user_api_key_dict=type("UserKey", (), {})(),
traceback_str=f"Traceback ... {secret} ...",
)
message = json.loads(logger.log_queue[0]["message"])
assert secret not in json.dumps(message)
assert message["exception"] == "redacted-by-litellm"
assert message["traceback"] == "redacted-by-litellm"
assert message["error_class"] == "BadRequestError"
@pytest.mark.asyncio
async def test_failure_hook_keeps_exception_payload_when_redaction_off(datadog_env, monkeypatch):
monkeypatch.setattr(litellm, "turn_off_message_logging", False)
with patch("asyncio.create_task"):
logger = DataDogLogger()
secret = "secret-prompt-marker"
await logger.async_post_call_failure_hook(
request_data={"metadata": {}},
original_exception=Exception(f"boom {secret}"),
user_api_key_dict=type("UserKey", (), {})(),
traceback_str="trace",
)
message = json.loads(logger.log_queue[0]["message"])
assert message["exception"] == f"boom {secret}"

53
tests/unit/integrations/otel/test_otel_v2_logger.py
View file

@ -3238,3 +3238,56 @@ def test_provisional_close_then_payload_close_does_not_duplicate():
server.end()
llm_spans = [s for s in exporter.get_finished_spans() if s.name.startswith("chat")]
assert len(llm_spans) == 1
def test_async_post_call_failure_hook_redacts_error_text_when_gated():
"""With message redaction on, the proxy-level failure span must not carry the
prompt through error.message / the exception event, while error.type and the
provider error code stay intact."""
import litellm
from litellm.proxy._types import UserAPIKeyAuth
litellm.turn_off_message_logging = True
try:
logger, exporter = _logger()
server = logger._emitter.start_span(SpanRole.PROXY_REQUEST, LITELLM_PROXY_REQUEST_SPAN_NAME)
set_request_root_span(server)
secret = "secret-prompt-marker"
result = asyncio.run(
logger.async_post_call_failure_hook(
request_data={"metadata": {}},
original_exception=_proxy_exc(f"Unsupported content: {secret}", 400),
user_api_key_dict=UserAPIKeyAuth(),
traceback_str=f"Traceback ... {secret} ...",
)
)
server.end()
assert result is None
(span,) = exporter.get_finished_spans()
assert secret not in str(dict(span.attributes or {}))
assert secret not in str([dict(e.attributes or {}) for e in span.events])
assert span.attributes["error.message"] == "redacted-by-litellm"
assert span.attributes["error.type"] == "ProxyException"
assert span.attributes["litellm.provider.error.code"] == "400"
assert span.attributes["litellm.provider.error.stack_trace"] == "redacted-by-litellm"
finally:
litellm.turn_off_message_logging = False
def test_async_post_call_failure_hook_keeps_error_text_when_not_gated():
from litellm.proxy._types import UserAPIKeyAuth
logger, exporter = _logger()
server = logger._emitter.start_span(SpanRole.PROXY_REQUEST, LITELLM_PROXY_REQUEST_SPAN_NAME)
set_request_root_span(server)
secret = "secret-prompt-marker"
asyncio.run(
logger.async_post_call_failure_hook(
request_data={"metadata": {}},
original_exception=_proxy_exc(f"Unsupported content: {secret}", 400),
user_api_key_dict=UserAPIKeyAuth(),
traceback_str="trace",
)
)
server.end()
(span,) = exporter.get_finished_spans()
assert secret in span.attributes["error.message"]

76
tests/unit/integrations/test_mlflow.py
View file

@ -262,3 +262,79 @@ def test_mlflow_end_span_or_trace_works_with_mlflow_2x_client():
span=child_span, outputs="out", end_time_ns=1, status="OK"
)
assert client.ended_spans == [("req-2", "span-2")]
def _failure_modules():
modules = _mock_mlflow_modules()
class RecordingSpanEvent:
calls = []
def __init__(self, name, attributes):
self.name = name
self.attributes = attributes
@classmethod
def from_exception(cls, exception):
cls.calls.append(exception)
return cls("exception-from-exception", {"exception.message": str(exception)})
RecordingSpanEvent.calls = []
modules["mlflow.entities"].SpanEvent = RecordingSpanEvent
modules["_span_event_cls"] = RecordingSpanEvent
return modules
def test_mlflow_failure_event_redacts_exception_when_gated(monkeypatch):
import litellm
monkeypatch.setattr(litellm, "turn_off_message_logging", True)
modules = _failure_modules()
with patch.dict("sys.modules", modules):
from litellm.integrations.mlflow import MlflowLogger
mlflow_logger = MlflowLogger()
span = MagicMock()
mlflow_logger._start_span_or_trace = MagicMock(return_value=span)
mlflow_logger._end_span_or_trace = MagicMock()
mlflow_logger._extract_and_set_chat_attributes = MagicMock()
secret = "secret-prompt-marker"
mlflow_logger._handle_failure(
kwargs={"litellm_call_id": "x", "exception": Exception(f"boom {secret}")},
response_obj=None,
start_time=datetime.utcnow(),
end_time=datetime.utcnow(),
)
event = span.add_event.call_args.args[0]
assert event.attributes == {
"exception.type": "Exception",
"exception.message": "redacted-by-litellm",
"exception.stacktrace": "redacted-by-litellm",
}
assert modules["_span_event_cls"].calls == []
def test_mlflow_failure_event_uses_from_exception_when_not_gated(monkeypatch):
import litellm
monkeypatch.setattr(litellm, "turn_off_message_logging", False)
modules = _failure_modules()
with patch.dict("sys.modules", modules):
from litellm.integrations.mlflow import MlflowLogger
mlflow_logger = MlflowLogger()
span = MagicMock()
mlflow_logger._start_span_or_trace = MagicMock(return_value=span)
mlflow_logger._end_span_or_trace = MagicMock()
mlflow_logger._extract_and_set_chat_attributes = MagicMock()
exc = Exception("boom secret-prompt-marker")
mlflow_logger._handle_failure(
kwargs={"litellm_call_id": "x", "exception": exc},
response_obj=None,
start_time=datetime.utcnow(),
end_time=datetime.utcnow(),
)
assert modules["_span_event_cls"].calls == [exc]

80
tests/unit/integrations/test_opentelemetry.py
View file

@ -6762,3 +6762,83 @@ class TestOpenTelemetryNonInferenceUsage(unittest.TestCase):
self.assertEqual(
self._time_per_output_token_calls("aget_responses", response_obj=self.BACKGROUND_RESPONSE_OBJ), 1
)
SECRET_PROMPT = "secret-prompt-marker"
class TestOpenTelemetryFailureHookRedaction(unittest.TestCase):
def _run_hook(self, request_data, exception, redact):
original = litellm.turn_off_message_logging
litellm.turn_off_message_logging = redact
try:
exporter = InMemorySpanExporter()
provider = TracerProvider()
provider.add_span_processor(SimpleSpanProcessor(exporter))
tracer = provider.get_tracer(__name__)
otel = OpenTelemetry()
otel.tracer = tracer
server_span = tracer.start_span("Received Proxy Server Request")
user_api_key_dict = MagicMock()
user_api_key_dict.parent_otel_span = server_span
asyncio.run(
otel.async_post_call_failure_hook(
request_data=request_data,
original_exception=exception,
user_api_key_dict=user_api_key_dict,
traceback_str=f"Traceback ... {SECRET_PROMPT} ...",
)
)
finally:
litellm.turn_off_message_logging = original
finished = {s.name: s for s in exporter.get_finished_spans()}
return finished
def test_failure_hook_redacts_error_text_when_gated(self):
secret = SECRET_PROMPT
finished = self._run_hook(
{"metadata": {}},
litellm.BadRequestError(message=f"Unsupported content: {secret}", model="gpt-4o", llm_provider="openai"),
redact=True,
)
server = finished["Received Proxy Server Request"]
child = finished["Failed Proxy Server Request"]
for span in (server, child):
assert secret not in str(dict(span.attributes or {}))
for event in span.events:
assert secret not in str(dict(event.attributes or {}))
assert server.attributes["error.message"] == "redacted-by-litellm"
assert server.attributes["error.type"] == "BadRequestError"
assert child.attributes["exception"] == "redacted-by-litellm"
def test_failure_hook_keeps_error_text_when_not_gated(self):
secret = SECRET_PROMPT
finished = self._run_hook(
{"metadata": {}},
litellm.BadRequestError(message=f"Unsupported content: {secret}", model="gpt-4o", llm_provider="openai"),
redact=False,
)
child = finished["Failed Proxy Server Request"]
assert secret in child.attributes["exception"]
def test_record_error_attributes_on_span_redacts_via_global_gate(self):
original = litellm.turn_off_message_logging
litellm.turn_off_message_logging = True
try:
otel = OpenTelemetry()
span = MagicMock()
otel.record_error_attributes_on_span(
span=span,
exception=litellm.BadRequestError(
message=f"Unsupported content: {SECRET_PROMPT}", model="gpt-4o", llm_provider="openai"
),
status_code=400,
)
finally:
litellm.turn_off_message_logging = original
stamped = {call.args[0]: call.args[1] for call in span.set_attribute.call_args_list}
assert stamped.get("error.message") == "redacted-by-litellm"
assert stamped.get("error.type") == "BadRequestError"

137
tests/unit/litellm_core_utils/test_litellm_logging.py
View file

@ -9029,3 +9029,140 @@ def test_signoz_dispatch_requires_an_endpoint(monkeypatch):
logging_module._in_memory_loggers.clear()
monkeypatch.delenv("LITELLM_OTEL_V2", raising=False)
is_otel_v2_enabled.cache_clear()
class _CapturingFailureLogger(CustomLogger):
def __init__(self, **kwargs):
super().__init__(**kwargs)
self.sync_kwargs = None
self.async_kwargs = None
def log_failure_event(self, kwargs, response_obj, start_time, end_time):
self.sync_kwargs = kwargs
async def async_log_failure_event(self, kwargs, response_obj, start_time, end_time):
self.async_kwargs = kwargs
def _failure_logging_obj(secret):
return LitellmLogging(
model="gpt-4o",
messages=[{"role": "user", "content": secret}],
stream=False,
call_type="completion",
start_time=time.time(),
litellm_call_id="failure-redaction-test",
function_id="failure-redaction-test",
)
def _assert_no_secret_leaks(kwargs, secret):
payload = kwargs.get("standard_logging_object") or {}
assert secret not in json.dumps(payload, default=str)
assert secret not in str(kwargs.get("traceback_exception"))
error_information = payload.get("error_information") or {}
assert error_information.get("error_class") == "BadRequestError"
assert error_information.get("error_code") == "400"
assert error_information.get("llm_provider") == "openai"
def test_sync_failure_handler_redacts_error_text_for_custom_logger(monkeypatch):
secret = "secret-prompt-marker"
capture = _CapturingFailureLogger()
monkeypatch.setattr(litellm, "turn_off_message_logging", True)
monkeypatch.setattr(litellm, "failure_callback", [capture])
monkeypatch.setattr(litellm, "standard_logging_payload_excluded_fields", None)
logging_obj = _failure_logging_obj(secret)
logging_obj.failure_handler(
litellm.BadRequestError(message=f"Unsupported content: {secret}", model="gpt-4o", llm_provider="openai"),
f"Traceback ... {secret} ...",
)
assert capture.sync_kwargs is not None
_assert_no_secret_leaks(capture.sync_kwargs, secret)
def test_sync_failure_handler_honours_excluded_fields(monkeypatch):
capture = _CapturingFailureLogger()
monkeypatch.setattr(litellm, "failure_callback", [capture])
monkeypatch.setattr(litellm, "standard_logging_payload_excluded_fields", ["hidden_params"])
logging_obj = _failure_logging_obj("anything")
logging_obj.failure_handler(litellm.BadRequestError(message="denied", model="gpt-4o", llm_provider="openai"), "tb")
payload = capture.sync_kwargs.get("standard_logging_object") or {}
assert "hidden_params" not in payload
def test_sync_failure_handler_leaves_error_text_alone_when_redaction_off(monkeypatch):
secret = "secret-prompt-marker"
capture = _CapturingFailureLogger()
monkeypatch.setattr(litellm, "turn_off_message_logging", False)
monkeypatch.setattr(litellm, "failure_callback", [capture])
monkeypatch.setattr(litellm, "standard_logging_payload_excluded_fields", None)
logging_obj = _failure_logging_obj(secret)
logging_obj.failure_handler(
litellm.BadRequestError(message=f"Unsupported content: {secret}", model="gpt-4o", llm_provider="openai"),
f"Traceback ... {secret} ...",
)
payload = capture.sync_kwargs.get("standard_logging_object") or {}
assert secret in payload.get("error_information", {}).get("error_message", "")
assert secret in str(payload.get("error_str"))
@pytest.mark.asyncio
async def test_async_failure_handler_redacts_error_text_for_custom_logger(monkeypatch):
secret = "secret-prompt-marker"
capture = _CapturingFailureLogger()
monkeypatch.setattr(litellm, "turn_off_message_logging", True)
monkeypatch.setattr(litellm, "_async_failure_callback", [capture])
monkeypatch.setattr(litellm, "standard_logging_payload_excluded_fields", None)
logging_obj = _failure_logging_obj(secret)
await logging_obj.async_failure_handler(
litellm.BadRequestError(message=f"Unsupported content: {secret}", model="gpt-4o", llm_provider="openai"),
f"Traceback ... {secret} ...",
)
assert capture.async_kwargs is not None
_assert_no_secret_leaks(capture.async_kwargs, secret)
@pytest.mark.asyncio
async def test_async_failure_handler_honours_excluded_fields(monkeypatch):
capture = _CapturingFailureLogger()
monkeypatch.setattr(litellm, "_async_failure_callback", [capture])
monkeypatch.setattr(litellm, "standard_logging_payload_excluded_fields", ["hidden_params"])
logging_obj = _failure_logging_obj("anything")
await logging_obj.async_failure_handler(
litellm.BadRequestError(message="denied", model="gpt-4o", llm_provider="openai"), "tb"
)
payload = capture.async_kwargs.get("standard_logging_object") or {}
assert "hidden_params" not in payload
@pytest.mark.asyncio
async def test_per_callback_turn_off_redacts_error_fields_even_for_self_redacting_logger(monkeypatch):
secret = "secret-prompt-marker"
class SelfRedactingLogger(_CapturingFailureLogger):
def redacts_messages_itself(self):
return True
capture = SelfRedactingLogger(turn_off_message_logging=True)
monkeypatch.setattr(litellm, "turn_off_message_logging", False)
monkeypatch.setattr(litellm, "_async_failure_callback", [capture])
monkeypatch.setattr(litellm, "standard_logging_payload_excluded_fields", None)
logging_obj = _failure_logging_obj(secret)
await logging_obj.async_failure_handler(
litellm.BadRequestError(message=f"Unsupported content: {secret}", model="gpt-4o", llm_provider="openai"),
f"Traceback ... {secret} ...",
)
assert capture.async_kwargs is not None
_assert_no_secret_leaks(capture.async_kwargs, secret)

136
tests/unit/litellm_core_utils/test_redact_messages.py
View file

@ -1040,3 +1040,139 @@ def test_perform_redaction_drops_the_served_output_texts_from_the_callback_kwarg
details: Final = {"litellm_params": {}, SERVED_OUTPUT_TEXTS_KEY: ("Card: <CREDIT_CARD>",)}
perform_redaction(details, None)
assert SERVED_OUTPUT_TEXTS_KEY not in details
def _error_information(**overrides):
info = {
"error_code": "400",
"error_class": "BadRequestError",
"llm_provider": "openai",
"traceback": "Traceback ... secret-prompt-marker ...",
"error_message": "Unsupported content: secret-prompt-marker",
"error_rate_limit_category": None,
"normalized_error": None,
}
info.update(overrides)
return info
class TestRedactErrorInformation:
def test_replaces_message_and_traceback_keeps_everything_else(self):
from litellm.litellm_core_utils.redact_messages import redact_error_information
info = _error_information()
redacted = redact_error_information(info)
assert redacted == {
"error_code": "400",
"error_class": "BadRequestError",
"llm_provider": "openai",
"traceback": "redacted-by-litellm",
"error_message": "redacted-by-litellm",
"error_rate_limit_category": None,
"normalized_error": None,
}
def test_empty_fields_stay_empty(self):
from litellm.litellm_core_utils.redact_messages import redact_error_information
redacted = redact_error_information(_error_information(traceback="", error_message=None))
assert redacted["traceback"] == ""
assert redacted["error_message"] is None
assert "secret-prompt-marker" not in str(redacted)
def test_input_not_mutated(self):
from litellm.litellm_core_utils.redact_messages import redact_error_information
info = _error_information()
redact_error_information(info)
assert info["error_message"] == "Unsupported content: secret-prompt-marker"
assert info["traceback"] == "Traceback ... secret-prompt-marker ..."
class TestFailureRedactionOnStandardLoggingObject:
def test_redacted_standard_logging_payload_covers_error_fields(self):
payload = {
"messages": [{"role": "user", "content": "secret-prompt-marker"}],
"error_str": "Error: secret-prompt-marker",
"error_information": _error_information(),
"status": "failure",
}
redacted = redacted_standard_logging_payload(payload)
assert "secret-prompt-marker" not in str(redacted)
assert redacted["error_str"] == "redacted-by-litellm"
assert redacted["error_information"] == {
"error_code": "400",
"error_class": "BadRequestError",
"llm_provider": "openai",
"traceback": "redacted-by-litellm",
"error_message": "redacted-by-litellm",
"error_rate_limit_category": None,
"normalized_error": None,
}
def test_perform_redaction_covers_traceback_exception(self):
details = {
"litellm_params": {},
"traceback_exception": "Traceback ... secret-prompt-marker",
"standard_logging_object": {"error_str": "secret-prompt-marker"},
}
perform_redaction(details, None)
assert "secret-prompt-marker" not in str(details)
assert details["traceback_exception"] == "redacted-by-litellm"
assert details["standard_logging_object"]["error_str"] == "redacted-by-litellm"
def test_perform_redaction_leaves_empty_traceback_exception_alone(self):
details = {"litellm_params": {}, "traceback_exception": ""}
perform_redaction(details, None)
assert details["traceback_exception"] == ""
def _request_data(metadata=None, litellm_metadata=None, turn_off_message_logging=None):
data = {"metadata": metadata if metadata is not None else {}}
if litellm_metadata is not None:
data["litellm_metadata"] = litellm_metadata
if turn_off_message_logging is not None:
data["turn_off_message_logging"] = turn_off_message_logging
return data
class TestShouldRedactFailedRequest:
def test_global_on(self):
from litellm.litellm_core_utils.redact_messages import should_redact_failed_request
litellm.turn_off_message_logging = True
assert should_redact_failed_request(_request_data()) is True
def test_global_off(self):
from litellm.litellm_core_utils.redact_messages import should_redact_failed_request
assert should_redact_failed_request(_request_data()) is False
def test_enable_header_in_metadata(self):
from litellm.litellm_core_utils.redact_messages import should_redact_failed_request
request_data = _request_data(metadata={"headers": {"x-litellm-enable-message-redaction": "true"}})
assert should_redact_failed_request(request_data) is True
def test_disable_header_overrides_global_on(self):
from litellm.litellm_core_utils.redact_messages import should_redact_failed_request
litellm.turn_off_message_logging = True
request_data = _request_data(metadata={"headers": {"litellm-disable-message-redaction": "true"}})
assert should_redact_failed_request(request_data) is False
def test_enable_header_in_litellm_metadata(self):
from litellm.litellm_core_utils.redact_messages import should_redact_failed_request
request_data = _request_data(litellm_metadata={"headers": {"x-litellm-enable-message-redaction": "true"}})
assert should_redact_failed_request(request_data) is True
def test_dynamic_param_true(self):
from litellm.litellm_core_utils.redact_messages import should_redact_failed_request
assert should_redact_failed_request(_request_data(turn_off_message_logging=True)) is True
def test_dynamic_param_false_overrides_global_on(self):
from litellm.litellm_core_utils.redact_messages import should_redact_failed_request
litellm.turn_off_message_logging = True
assert should_redact_failed_request(_request_data(turn_off_message_logging=False)) is False

61
tests/unit/proxy/hooks/test_proxy_track_cost_callback.py Normal file
View file

@ -0,0 +1,61 @@
import json
from unittest.mock import AsyncMock, MagicMock
import pytest
import litellm
from litellm.proxy._types import UserAPIKeyAuth
from litellm.proxy.hooks.proxy_track_cost_callback import _ProxyDBLogger
SECRET = "secret-prompt-marker"
def _logger_and_writer():
writer = MagicMock()
writer.update_database = AsyncMock()
logger = _ProxyDBLogger(spend_writer=lambda: writer)
return logger, writer
@pytest.mark.asyncio
async def test_failure_hook_redacts_persisted_error_information(monkeypatch):
monkeypatch.setattr(litellm, "turn_off_message_logging", True)
logger, writer = _logger_and_writer()
request_data = {"metadata": {}}
await logger.async_post_call_failure_hook(
request_data=request_data,
original_exception=litellm.BadRequestError(
message=f"Unsupported content: {SECRET}", model="gpt-4o", llm_provider="openai"
),
user_api_key_dict=UserAPIKeyAuth(),
traceback_str=f"Traceback ... {SECRET} ...",
)
persisted = request_data["litellm_params"]["metadata"]["error_information"]
assert SECRET not in json.dumps(persisted)
assert persisted["error_message"] == "redacted-by-litellm"
assert persisted["traceback"] == "redacted-by-litellm"
assert persisted["error_class"] == "BadRequestError"
assert persisted["error_code"] == "400"
writer.update_database.assert_awaited_once()
@pytest.mark.asyncio
async def test_failure_hook_leaves_error_information_alone_when_redaction_off(monkeypatch):
monkeypatch.setattr(litellm, "turn_off_message_logging", False)
logger, writer = _logger_and_writer()
request_data = {"metadata": {}}
await logger.async_post_call_failure_hook(
request_data=request_data,
original_exception=litellm.BadRequestError(
message=f"Unsupported content: {SECRET}", model="gpt-4o", llm_provider="openai"
),
user_api_key_dict=UserAPIKeyAuth(),
traceback_str="trace",
)
persisted = request_data["litellm_params"]["metadata"]["error_information"]
assert SECRET in persisted["error_message"]
writer.update_database.assert_awaited_once()