ai-workspace-services/litellm
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix(spend-logs): redact echoed prompts in error_information (LIT-2992) (#27689)

Browse Source 
Provider validation errors (e.g. OpenAI RateLimitError carrying 178
pydantic errors each with their own 'input': [...]) were stored verbatim
in LiteLLM_SpendLogs.metadata.error_information.error_message via
str(original_exception), producing rows >12 MB.

Sanitize before metadata is serialized:
- redact 'input'/'messages' values in both error_message and traceback
  when store_prompts_in_spend_logs is False (back-door leak paths)
- always apply the MAX_STRING_LENGTH_PROMPT_IN_DB size cap to
  error_message and traceback (DB-storage safeguard)

Value scanning uses a parser-based balanced-bracket walk that respects
string quoting, so multi-modal payloads ('messages': [{'content': [...]}])
and user text containing literal brackets ("secret[123") are handled
correctly instead of leaking past a depth-1 regex.

Scoped to the spend-log path so OTEL/Datadog/etc. callbacks still
receive the untruncated error per LITELLM_TRUNCATION_DB_SAFEGUARD_NOTE.

Co-authored-by: Yassin Kortam <yassinkortam@g.ucla.edu>
Co-authored-by: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
This commit is contained in:
Yassin Kortam 2026-05-13 22:11:24 -07:00 committed by GitHub
parent 117b036ed6
commit de1747dca8
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
3 changed files with 830 additions and 216 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

445
litellm/proxy/hooks/proxy_track_cost_callback.py
View File

@ -23,8 +23,14 @@ from litellm.proxy.spend_tracking.spend_log_error_logger import (
should_suppress_spend_log_tracebacks,
spend_log_error,
)
from litellm.proxy.spend_tracking.spend_tracking_utils import (
_sanitize_error_information_for_spend_logs,
)
from litellm.proxy.utils import ProxyUpdateSpend
from litellm.types.utils import StandardLoggingPayload
from litellm.types.utils import (
StandardLoggingPayload,
StandardLoggingPayloadErrorInformation,
)
from litellm.utils import get_end_user_id_for_cost_tracking
@ -34,35 +40,35 @@ class _ProxyDBLogger(CustomLogger):
kwargs, response_obj, start_time, end_time
)
async def async_post_call_failure_hook(
self,
request_data: dict,
original_exception: Exception,
user_api_key_dict: UserAPIKeyAuth,
traceback_str: Optional[str] = None,
):
try:
await _release_budget_reservation(
budget_reservation=user_api_key_dict.budget_reservation
)
except Exception:
verbose_proxy_logger.exception(
"Failed to release budget reservation during failure handling"
)
try:
await _invalidate_budget_reservation_counters(
budget_reservation=user_api_key_dict.budget_reservation
)
if user_api_key_dict.budget_reservation is not None:
user_api_key_dict.budget_reservation["finalized"] = True
except Exception:
verbose_proxy_logger.exception(
"Failed to invalidate budget reservation counters after failure release failed"
)
request_route = user_api_key_dict.request_route
if _ProxyDBLogger._should_track_errors_in_db() is False:
return
async def async_post_call_failure_hook(
self,
request_data: dict,
original_exception: Exception,
user_api_key_dict: UserAPIKeyAuth,
traceback_str: Optional[str] = None,
):
try:
await _release_budget_reservation(
budget_reservation=user_api_key_dict.budget_reservation
)
except Exception:
verbose_proxy_logger.exception(
"Failed to release budget reservation during failure handling"
)
try:
await _invalidate_budget_reservation_counters(
budget_reservation=user_api_key_dict.budget_reservation
)
if user_api_key_dict.budget_reservation is not None:
user_api_key_dict.budget_reservation["finalized"] = True
except Exception:
verbose_proxy_logger.exception(
"Failed to invalidate budget reservation counters after failure release failed"
)
request_route = user_api_key_dict.request_route
if _ProxyDBLogger._should_track_errors_in_db() is False:
return
elif request_route is not None and not (
RouteChecks.is_llm_api_route(route=request_route)
or RouteChecks.is_info_route(route=request_route)
@ -89,6 +95,13 @@ class _ProxyDBLogger(CustomLogger):
# ``.get("traceback")`` / truthy checks, and the TypedDict marks
# the field as optional, so omitting is type-safe.
_error_information.pop("traceback", None)
# Strip echoed request input + apply DB-size cap before storing in
# the spend-log metadata column (LIT-2992). Result is never None
# here because the input above is constructed non-None.
_error_information = cast(
StandardLoggingPayloadErrorInformation,
_sanitize_error_information_for_spend_logs(_error_information),
)
_metadata["error_information"] = _error_information
_metadata = await _ProxyDBLogger._enrich_failure_metadata_with_key_info(
@ -184,64 +197,64 @@ class _ProxyDBLogger(CustomLogger):
f"kwargs stream: {kwargs.get('stream', None)} + complete streaming response: {kwargs.get('complete_streaming_response', None)}"
)
parent_otel_span = _get_parent_otel_span_from_kwargs(kwargs=kwargs)
litellm_params = kwargs.get("litellm_params", {}) or {}
end_user_id = get_end_user_id_for_cost_tracking(litellm_params)
metadata = get_litellm_metadata_from_kwargs(kwargs=kwargs)
budget_reservation = _get_budget_reservation_from_metadata(
metadata=metadata
)
user_id = cast(Optional[str], metadata.get("user_api_key_user_id", None))
team_id = cast(Optional[str], metadata.get("user_api_key_team_id", None))
org_id = cast(Optional[str], metadata.get("user_api_key_org_id", None))
litellm_params = kwargs.get("litellm_params", {}) or {}
end_user_id = get_end_user_id_for_cost_tracking(litellm_params)
metadata = get_litellm_metadata_from_kwargs(kwargs=kwargs)
budget_reservation = _get_budget_reservation_from_metadata(
metadata=metadata
)
user_id = cast(Optional[str], metadata.get("user_api_key_user_id", None))
team_id = cast(Optional[str], metadata.get("user_api_key_team_id", None))
org_id = cast(Optional[str], metadata.get("user_api_key_org_id", None))
key_alias = cast(Optional[str], metadata.get("user_api_key_alias", None))
end_user_max_budget = metadata.get("user_api_end_user_max_budget", None)
sl_object: Optional[StandardLoggingPayload] = kwargs.get(
"standard_logging_object", None
)
response_cost = (
sl_object.get("response_cost", None)
if sl_object is not None
else kwargs.get("response_cost", None)
)
tags = _get_request_tags_for_cost_tracking(
sl_object=sl_object,
metadata=metadata,
)
if response_cost is not None:
user_api_key = metadata.get("user_api_key", None)
response_cost = (
sl_object.get("response_cost", None)
if sl_object is not None
else kwargs.get("response_cost", None)
)
tags = _get_request_tags_for_cost_tracking(
sl_object=sl_object,
metadata=metadata,
)
if response_cost is not None:
user_api_key = metadata.get("user_api_key", None)
if kwargs.get("cache_hit", False) is True:
response_cost = 0.0
verbose_proxy_logger.debug(
f"Cache Hit: response_cost {response_cost}, for user_id {user_id}"
)
verbose_proxy_logger.debug(
f"user_api_key {user_api_key}, user_id {user_id}, team_id {team_id}, end_user_id {end_user_id}"
)
if _should_track_cost_callback(
user_api_key=user_api_key,
verbose_proxy_logger.debug(
f"user_api_key {user_api_key}, user_id {user_id}, team_id {team_id}, end_user_id {end_user_id}"
)
if _should_track_cost_callback(
user_api_key=user_api_key,
user_id=user_id,
team_id=team_id,
end_user_id=end_user_id,
):
## UPDATE DATABASE
await _update_database_and_spend_counters(
proxy_logging_obj=proxy_logging_obj,
increment_spend_counters=increment_spend_counters,
user_api_key=user_api_key,
user_id=user_id,
end_user_id=end_user_id,
team_id=team_id,
org_id=org_id,
kwargs=kwargs,
completion_response=completion_response,
start_time=start_time,
end_time=end_time,
response_cost=response_cost,
budget_reservation=budget_reservation,
request_tags=tags,
)
end_user_id=end_user_id,
):
## UPDATE DATABASE
await _update_database_and_spend_counters(
proxy_logging_obj=proxy_logging_obj,
increment_spend_counters=increment_spend_counters,
user_api_key=user_api_key,
user_id=user_id,
end_user_id=end_user_id,
team_id=team_id,
org_id=org_id,
kwargs=kwargs,
completion_response=completion_response,
start_time=start_time,
end_time=end_time,
response_cost=response_cost,
budget_reservation=budget_reservation,
request_tags=tags,
)
# update cache (fire-and-forget for backward compat:
# cached object fields, soft budget alerts, etc.)
@ -261,15 +274,15 @@ class _ProxyDBLogger(CustomLogger):
token=user_api_key,
key_alias=key_alias,
end_user_id=end_user_id,
response_cost=response_cost,
max_budget=end_user_max_budget,
)
elif budget_reservation is not None:
await _release_budget_reservation(
budget_reservation=budget_reservation
)
response_cost=response_cost,
max_budget=end_user_max_budget,
)
elif budget_reservation is not None:
await _release_budget_reservation(
budget_reservation=budget_reservation
)
else:
await _release_budget_reservation(budget_reservation=budget_reservation)
await _release_budget_reservation(budget_reservation=budget_reservation)
# Non-model call types (health checks, afile_delete) have no model or standard_logging_object.
# Use .get() for "stream" to avoid KeyError on health checks.
if sl_object is None and not kwargs.get("model"):
@ -396,7 +409,7 @@ class _ProxyDBLogger(CustomLogger):
return
def _should_track_cost_callback(
def _should_track_cost_callback(
user_api_key: Optional[str],
user_id: Optional[str],
team_id: Optional[str],
@ -417,135 +430,135 @@ def _should_track_cost_callback(
or end_user_id is not None
):
return True
return False
def _get_budget_reservation_from_metadata(metadata: dict) -> Optional[dict]:
metadata_budget_reservation = metadata.get("user_api_key_budget_reservation")
if isinstance(metadata_budget_reservation, dict):
return metadata_budget_reservation
user_api_key_auth_obj = metadata.get("user_api_key_auth")
if user_api_key_auth_obj is None:
return None
if isinstance(user_api_key_auth_obj, dict):
budget_reservation = user_api_key_auth_obj.get("budget_reservation")
return budget_reservation if isinstance(budget_reservation, dict) else None
return getattr(user_api_key_auth_obj, "budget_reservation", None)
def _get_request_tags_for_cost_tracking(
sl_object: Optional[StandardLoggingPayload],
metadata: dict,
) -> Optional[List[str]]:
if sl_object is not None:
request_tags = sl_object.get("request_tags", None)
if isinstance(request_tags, list):
return request_tags
metadata_tags = metadata.get("tags", None)
if isinstance(metadata_tags, list):
return metadata_tags
return None
async def _update_database_and_spend_counters(
proxy_logging_obj: Any,
increment_spend_counters: Any,
user_api_key: Optional[str],
user_id: Optional[str],
end_user_id: Optional[str],
team_id: Optional[str],
org_id: Optional[str],
kwargs: dict,
completion_response: Optional[Union[litellm.ModelResponse, Any]],
start_time: Any,
end_time: Any,
response_cost: float,
budget_reservation: Optional[dict],
request_tags: Optional[List[str]] = None,
) -> None:
try:
await proxy_logging_obj.db_spend_update_writer.update_database(
token=user_api_key,
response_cost=response_cost,
user_id=user_id,
end_user_id=end_user_id,
team_id=team_id,
kwargs=kwargs,
completion_response=completion_response,
start_time=start_time,
end_time=end_time,
org_id=org_id,
)
except Exception:
if budget_reservation is not None:
try:
await _release_budget_reservation(budget_reservation=budget_reservation)
except Exception:
verbose_proxy_logger.exception(
"Failed to release budget reservation after database update failed"
)
try:
await _invalidate_budget_reservation_counters(
budget_reservation=budget_reservation
)
except Exception:
verbose_proxy_logger.exception(
"Failed to invalidate budget reservation counters after release failed"
)
raise
try:
await increment_spend_counters(
token=user_api_key,
team_id=team_id,
user_id=user_id,
response_cost=response_cost,
org_id=org_id,
budget_reservation=budget_reservation,
end_user_id=end_user_id,
tags=request_tags,
)
except Exception:
if budget_reservation is not None:
try:
await _invalidate_budget_reservation_counters(
budget_reservation=budget_reservation
)
except Exception:
verbose_proxy_logger.exception(
"Failed to invalidate budget reservation counters after spend counter update failed"
)
finally:
budget_reservation["finalized"] = True
raise
async def _release_budget_reservation(budget_reservation: Optional[dict]) -> None:
if budget_reservation is None:
return
from litellm.proxy.spend_tracking.budget_reservation import (
release_budget_reservation,
)
await release_budget_reservation(
budget_reservation=budget_reservation,
)
async def _invalidate_budget_reservation_counters(
budget_reservation: Optional[dict],
) -> None:
if budget_reservation is None:
return
from litellm.proxy.spend_tracking.budget_reservation import (
invalidate_budget_reservation_counters,
)
await invalidate_budget_reservation_counters(
budget_reservation=budget_reservation,
)
return False
def _get_budget_reservation_from_metadata(metadata: dict) -> Optional[dict]:
metadata_budget_reservation = metadata.get("user_api_key_budget_reservation")
if isinstance(metadata_budget_reservation, dict):
return metadata_budget_reservation
user_api_key_auth_obj = metadata.get("user_api_key_auth")
if user_api_key_auth_obj is None:
return None
if isinstance(user_api_key_auth_obj, dict):
budget_reservation = user_api_key_auth_obj.get("budget_reservation")
return budget_reservation if isinstance(budget_reservation, dict) else None
return getattr(user_api_key_auth_obj, "budget_reservation", None)
def _get_request_tags_for_cost_tracking(
sl_object: Optional[StandardLoggingPayload],
metadata: dict,
) -> Optional[List[str]]:
if sl_object is not None:
request_tags = sl_object.get("request_tags", None)
if isinstance(request_tags, list):
return request_tags
metadata_tags = metadata.get("tags", None)
if isinstance(metadata_tags, list):
return metadata_tags
return None
async def _update_database_and_spend_counters(
proxy_logging_obj: Any,
increment_spend_counters: Any,
user_api_key: Optional[str],
user_id: Optional[str],
end_user_id: Optional[str],
team_id: Optional[str],
org_id: Optional[str],
kwargs: dict,
completion_response: Optional[Union[litellm.ModelResponse, Any]],
start_time: Any,
end_time: Any,
response_cost: float,
budget_reservation: Optional[dict],
request_tags: Optional[List[str]] = None,
) -> None:
try:
await proxy_logging_obj.db_spend_update_writer.update_database(
token=user_api_key,
response_cost=response_cost,
user_id=user_id,
end_user_id=end_user_id,
team_id=team_id,
kwargs=kwargs,
completion_response=completion_response,
start_time=start_time,
end_time=end_time,
org_id=org_id,
)
except Exception:
if budget_reservation is not None:
try:
await _release_budget_reservation(budget_reservation=budget_reservation)
except Exception:
verbose_proxy_logger.exception(
"Failed to release budget reservation after database update failed"
)
try:
await _invalidate_budget_reservation_counters(
budget_reservation=budget_reservation
)
except Exception:
verbose_proxy_logger.exception(
"Failed to invalidate budget reservation counters after release failed"
)
raise
try:
await increment_spend_counters(
token=user_api_key,
team_id=team_id,
user_id=user_id,
response_cost=response_cost,
org_id=org_id,
budget_reservation=budget_reservation,
end_user_id=end_user_id,
tags=request_tags,
)
except Exception:
if budget_reservation is not None:
try:
await _invalidate_budget_reservation_counters(
budget_reservation=budget_reservation
)
except Exception:
verbose_proxy_logger.exception(
"Failed to invalidate budget reservation counters after spend counter update failed"
)
finally:
budget_reservation["finalized"] = True
raise
async def _release_budget_reservation(budget_reservation: Optional[dict]) -> None:
if budget_reservation is None:
return
from litellm.proxy.spend_tracking.budget_reservation import (
release_budget_reservation,
)
await release_budget_reservation(
budget_reservation=budget_reservation,
)
async def _invalidate_budget_reservation_counters(
budget_reservation: Optional[dict],
) -> None:
if budget_reservation is None:
return
from litellm.proxy.spend_tracking.budget_reservation import (
invalidate_budget_reservation_counters,
)
await invalidate_budget_reservation_counters(
budget_reservation=budget_reservation,
)

179
litellm/proxy/spend_tracking/spend_tracking_utils.py
View File

@ -1,6 +1,7 @@
import hashlib
import json
import os
import re
import secrets
from datetime import datetime
from datetime import datetime as dt
@ -33,6 +34,7 @@ from litellm.types.utils import (
StandardLoggingMCPToolCall,
StandardLoggingModelInformation,
StandardLoggingPayload,
StandardLoggingPayloadErrorInformation,
StandardLoggingVectorStoreRequest,
VectorStoreSearchResponse,
)
@ -689,6 +691,183 @@ def _sanitize_request_body_for_spend_logs_payload(
}
# Quoted-key form: ``"input"`` / ``'messages'`` / ``"prompt"`` followed by
# ``:``. Covers JSON bodies and Python dict-reprs in provider error strings.
# ``prompt`` is included for ``/v1/completions``-style payloads where the user
# input lives under a top-level ``prompt`` key rather than ``messages``.
_ERROR_MESSAGE_PROMPT_LEAK_KEYS = ("input", "messages", "prompt")
# Assignment-style keys: Pydantic v2 validation errors render the offending
# value as ``input_value=<repr>`` inside ``[type=..., input_value=...,
# input_type=...]``. The same prompt body that would appear under an
# ``"input"`` JSON key is echoed here as a Python repr, so we redact it
# under the same store_prompts_in_spend_logs gate.
_ERROR_MESSAGE_ASSIGN_LEAK_KEYS = ("input_value",)
_SENSITIVE_KEY_START_PATTERN = re.compile(
r"(?:"
r"['\"](?:" + "|".join(_ERROR_MESSAGE_PROMPT_LEAK_KEYS) + r")['\"]\s*:\s*"
r"|"
r"\b(?:" + "|".join(_ERROR_MESSAGE_ASSIGN_LEAK_KEYS) + r")\s*=\s*"
r")"
)
def _scan_quoted_string_end(text: str, start: int, quote: str) -> int:
"""
Given ``text[start] == quote`` (``'`` or ``"``), return the index just
past the matching close quote, honoring backslash escapes. Returns
``-1`` if unterminated.
"""
n = len(text)
i = start + 1
while i < n:
c = text[i]
if c == "\\":
i += 2
continue
if c == quote:
return i + 1
i += 1
return -1
def _scan_balanced_value_end(text: str, start: int) -> int:
"""
Given ``text[start]`` is ``[``, ``{``, ``'`` or ``"``, return the index
just past the matching close, accounting for nested brackets and
quoted strings (with escape sequences). Returns ``-1`` if the
structure is unterminated.
Implemented iteratively (no self-recursion): the bracket scanner
inlines a quote-skip helper rather than re-entering itself, since
JSON-style values cannot contain another bracket *as a first char*
inside a quoted string — only the quote-skip case can occur.
"""
n = len(text)
if start >= n:
return -1
first = text[start]
if first in ("'", '"'):
return _scan_quoted_string_end(text, start, first)
if first == "[":
close = "]"
elif first == "{":
close = "}"
else:
return -1
depth = 0
i = start
while i < n:
c = text[i]
if c in ("'", '"'):
end = _scan_quoted_string_end(text, i, c)
if end == -1:
return -1
i = end
continue
if c == first:
depth += 1
elif c == close:
depth -= 1
if depth == 0:
return i + 1
i += 1
return -1
def _redact_prompt_leaks_in_error_string(text: str) -> str:
"""
Strip echoed request input from provider error strings.
Provider validation errors (e.g. OpenAI ``RateLimitError`` carrying 178
pydantic validation errors, each with its own ``'input': [...]`` field)
embed the full request body in their message. When prompts must not be
stored in spend logs, that echo is a back-door leak.
Two leak shapes are handled:
- Quoted-key form — ``"<key>": <value>`` where ``key`` is ``input``,
``messages`` or ``prompt`` (covers JSON bodies, Python dict-reprs,
and ``/v1/completions`` payloads).
- Assignment form — ``input_value=<value>`` from Pydantic v2 validation
errors, which render the offending value as a Python repr inside
``[type=..., input_value=..., input_type=...]``.
The value scan understands nested ``[]`` / ``{}`` and quoted strings,
so multi-modal payloads (``'messages': [{'content': [{...}]}]``) and
user text containing brackets (``"secret[123"``) are handled correctly.
"""
if not text:
return text
redaction = f'"{REDACTED_BY_LITELM_STRING}"'
out: List[str] = []
n = len(text)
pos = 0
while pos < n:
m = _SENSITIVE_KEY_START_PATTERN.search(text, pos)
if not m:
out.append(text[pos:])
break
out.append(text[pos : m.end()])
v_start = m.end()
if v_start >= n:
break
first = text[v_start]
if first in ("[", "{", "'", '"'):
v_end = _scan_balanced_value_end(text, v_start)
if v_end == -1:
# Unterminated value — redact through the rest of the string
# so a malformed leak can't slip past.
out.append(redaction)
pos = n
break
out.append(redaction)
pos = v_end
else:
# Unquoted scalar (number, null, bare identifier) — not a leak
# carrier, leave intact and resume after the key match.
pos = v_start
return "".join(out)
def _sanitize_error_information_for_spend_logs(
error_information: Optional[StandardLoggingPayloadErrorInformation],
) -> Optional[StandardLoggingPayloadErrorInformation]:
"""
Sanitize ``error_information`` before it lands in ``LiteLLM_SpendLogs.metadata``.
Provider errors are stored verbatim via ``str(original_exception)``; those
strings can echo the full request body, producing multi-megabyte spend-log
rows.
- Always: cap ``error_message`` and ``traceback`` with the existing
``MAX_STRING_LENGTH_PROMPT_IN_DB`` DB-storage safeguard.
- When ``store_prompts_in_spend_logs`` is False: additionally redact
``'input'`` / ``'messages'`` / ``'prompt'`` values *and* Pydantic v2
``input_value=...`` assignments inside both ``error_message`` and
``traceback`` so prompts cannot leak through either field.
Scoped to the spend-log path — OTEL/Datadog/etc. callbacks still receive
the untruncated error per ``LITELLM_TRUNCATION_DB_SAFEGUARD_NOTE``.
"""
if error_information is None:
return None
sanitized = cast(dict, {**error_information})
if not _should_store_prompts_and_responses_in_spend_logs():
for field in ("error_message", "traceback"):
value = sanitized.get(field)
if isinstance(value, str):
sanitized[field] = _redact_prompt_leaks_in_error_string(value)
sanitized = _sanitize_request_body_for_spend_logs_payload(sanitized)
return cast(StandardLoggingPayloadErrorInformation, sanitized)
def _convert_to_json_serializable_dict(
obj: Any, visited: Optional[set] = None, max_depth: int = 20
) -> Any:

422
tests/test_litellm/proxy/spend_tracking/test_spend_tracking_utils.py
View File

@ -30,6 +30,8 @@ from litellm.proxy.spend_tracking.spend_tracking_utils import (
_get_spend_logs_metadata,
_get_vector_store_request_for_spend_logs_payload,
_is_master_key,
_redact_prompt_leaks_in_error_string,
_sanitize_error_information_for_spend_logs,
_sanitize_request_body_for_spend_logs_payload,
_should_store_prompts_and_responses_in_spend_logs,
get_logging_payload,
@ -1587,3 +1589,423 @@ def test_proxy_server_request_payload_excludes_secret_fields(mock_should_store):
), "secret_fields must never appear in the spend-log proxy_server_request column"
assert parsed["model"] == "gpt-4"
assert parsed["messages"] == [{"role": "user", "content": "hello"}]
# ---------------------------------------------------------------------------
# LIT-2992: error_information sanitization for spend logs
# ---------------------------------------------------------------------------
def test_redact_prompt_leaks_strips_input_value_python_repr():
# OpenAI-style pydantic validation error: each entry carries its own
# 'input': [...] field echoing the full conversation.
error_text = (
"OpenAIException - {'error': {'message': \"1 validation error:\\n "
"{'type': 'string_type', 'loc': ('body', 'input', 'str'), "
"'msg': 'Input should be a valid string', "
"'input': [{'role': 'user', 'content': 'super-secret-prompt'}]}"
'"}}'
)
redacted = _redact_prompt_leaks_in_error_string(error_text)
assert "super-secret-prompt" not in redacted
assert REDACTED_BY_LITELM_STRING in redacted
# Surrounding context (error class, msg, loc) is preserved.
assert "string_type" in redacted
assert "Input should be a valid string" in redacted
def test_redact_prompt_leaks_strips_input_value_json():
error_text = (
'{"error":{"message":"validation failed",'
'"input":[{"role":"user","content":"top-secret-content"}]}}'
)
redacted = _redact_prompt_leaks_in_error_string(error_text)
assert "top-secret-content" not in redacted
assert REDACTED_BY_LITELM_STRING in redacted
def test_redact_prompt_leaks_strips_messages_value():
error_text = '{"error":{"messages":[{"role":"user","content":"leak"}]}}'
redacted = _redact_prompt_leaks_in_error_string(error_text)
assert "leak" not in redacted
assert REDACTED_BY_LITELM_STRING in redacted
def test_redact_prompt_leaks_preserves_prose_mentions():
# The word "input" / "messages" in prose (not as a key) must not be
# redacted — only quoted-key matches.
error_text = "Rate limit exceeded. Reduce input size and retry."
assert _redact_prompt_leaks_in_error_string(error_text) == error_text
def test_redact_prompt_leaks_empty_string():
assert _redact_prompt_leaks_in_error_string("") == ""
@patch(
"litellm.proxy.spend_tracking.spend_tracking_utils._should_store_prompts_and_responses_in_spend_logs"
)
def test_sanitize_error_information_redacts_when_not_storing_prompts(
mock_should_store,
):
mock_should_store.return_value = False
error_info = {
"error_code": "429",
"error_class": "RateLimitError",
"llm_provider": "openai",
"traceback": "Traceback (most recent call last):\n File ...",
"error_message": (
'OpenAIException - {"error":{"message":"validation failed",'
'"input":[{"role":"user","content":"leaked-prompt-content"}]}}'
),
}
sanitized = _sanitize_error_information_for_spend_logs(error_info)
assert sanitized is not None
assert "leaked-prompt-content" not in sanitized["error_message"]
assert REDACTED_BY_LITELM_STRING in sanitized["error_message"]
# Non-leaking fields untouched.
assert sanitized["error_code"] == "429"
assert sanitized["error_class"] == "RateLimitError"
assert sanitized["llm_provider"] == "openai"
@patch(
"litellm.proxy.spend_tracking.spend_tracking_utils._should_store_prompts_and_responses_in_spend_logs"
)
def test_sanitize_error_information_skips_redaction_when_storing_prompts(
mock_should_store,
):
mock_should_store.return_value = True
error_info = {
"error_code": "429",
"error_class": "RateLimitError",
"llm_provider": "openai",
"traceback": "",
"error_message": (
'OpenAIException - {"error":{"input":[{"role":"user","content":"kept"}]}}'
),
}
sanitized = _sanitize_error_information_for_spend_logs(error_info)
assert sanitized is not None
# User opted in via store_prompts_in_spend_logs — no key-level redaction.
assert "kept" in sanitized["error_message"]
assert REDACTED_BY_LITELM_STRING not in sanitized["error_message"]
@patch(
"litellm.proxy.spend_tracking.spend_tracking_utils._should_store_prompts_and_responses_in_spend_logs"
)
def test_sanitize_error_information_caps_size_regardless_of_prompt_flag(
mock_should_store,
):
# The DB-storage cap must apply even when prompt storage is enabled, so a
# provider error that echoes a multi-MB body can't blow up a single row.
from litellm.constants import MAX_STRING_LENGTH_PROMPT_IN_DB
mock_should_store.return_value = True
huge_error = "x" * (MAX_STRING_LENGTH_PROMPT_IN_DB * 10)
error_info = {
"error_code": "500",
"error_class": "InternalServerError",
"llm_provider": "openai",
"traceback": "x" * (MAX_STRING_LENGTH_PROMPT_IN_DB * 10),
"error_message": huge_error,
}
sanitized = _sanitize_error_information_for_spend_logs(error_info)
assert sanitized is not None
assert len(sanitized["error_message"]) < len(huge_error)
assert LITELLM_TRUNCATED_PAYLOAD_FIELD in sanitized["error_message"]
assert LITELLM_TRUNCATED_PAYLOAD_FIELD in sanitized["traceback"]
def test_sanitize_error_information_none_passthrough():
assert _sanitize_error_information_for_spend_logs(None) is None
@patch(
"litellm.proxy.spend_tracking.spend_tracking_utils._should_store_prompts_and_responses_in_spend_logs"
)
def test_sanitize_error_information_reproduces_lit_2992(mock_should_store):
# Mirrors the reproduced row body from LIT-2992 — a RateLimitError whose
# message embeds 178 pydantic validation errors, each carrying a full
# 'input': [...] echo of the conversation.
mock_should_store.return_value = False
huge_conversation_blob = "user-conversation-history-" * 5000
validation_entries = []
for _ in range(50):
validation_entries.append(
"{'type': 'string_type', 'loc': ('body', 'input', 'str'), "
"'msg': 'Input should be a valid string', "
f"'input': [{{'role': 'user', 'content': '{huge_conversation_blob}'}}]}}"
)
error_message = (
"litellm.RateLimitError: RateLimitError: OpenAIException - "
'{"error":{"message":"' + "\\n ".join(validation_entries) + '"}}'
)
error_info = {
"error_code": "429",
"error_class": "RateLimitError",
"llm_provider": "openai",
"traceback": "",
"error_message": error_message,
}
sanitized = _sanitize_error_information_for_spend_logs(error_info)
assert sanitized is not None
assert huge_conversation_blob not in sanitized["error_message"]
# The structural fields that aid debugging remain.
assert "RateLimitError" in sanitized["error_message"]
assert "string_type" in sanitized["error_message"]
def test_redact_prompt_leaks_handles_nested_multimodal_content():
# Multi-modal payload: 'content' is itself a list. The depth-1 regex
# would stop at the inner '['; the parser-based scanner must walk
# through balanced nested brackets.
error_text = (
'{"error":{"messages":[{"role":"user",'
'"content":[{"type":"text","text":"top-secret-multimodal"}]}]}}'
)
redacted = _redact_prompt_leaks_in_error_string(error_text)
assert "top-secret-multimodal" not in redacted
assert REDACTED_BY_LITELM_STRING in redacted
def test_redact_prompt_leaks_handles_bracket_in_prompt_text():
# Prompt text contains a literal '[' — the depth-1 regex would close
# the outer ']' prematurely. The parser must respect string quoting.
error_text = (
'{"error":{"input":[{"role":"user","content":"secret[123 still secret"}]}}'
)
redacted = _redact_prompt_leaks_in_error_string(error_text)
assert "secret[123" not in redacted
assert "still secret" not in redacted
assert REDACTED_BY_LITELM_STRING in redacted
def test_redact_prompt_leaks_handles_escaped_quote_in_prompt_text():
# Prompt with an escaped quote inside a JSON string must not break
# value scanning.
error_text = '{"error":{"input":[{"role":"user","content":"she said \\"hi[\\""}]}}'
redacted = _redact_prompt_leaks_in_error_string(error_text)
assert "she said" not in redacted
assert REDACTED_BY_LITELM_STRING in redacted
def test_redact_prompt_leaks_handles_nested_input_python_repr():
# Python dict-repr with nested list inside 'input' — single quotes.
error_text = (
"validation error: {'input': [{'role': 'user', "
"'content': [{'type': 'text', 'text': 'leaked-nested-text'}]}]}"
)
redacted = _redact_prompt_leaks_in_error_string(error_text)
assert "leaked-nested-text" not in redacted
assert REDACTED_BY_LITELM_STRING in redacted
def test_redact_prompt_leaks_handles_unterminated_value():
# If a value never closes (malformed error string), redact through to
# the end rather than leaving the prompt content reachable.
error_text = '{"input":[{"role":"user","content":"never-closes-leaked'
redacted = _redact_prompt_leaks_in_error_string(error_text)
assert "never-closes-leaked" not in redacted
assert REDACTED_BY_LITELM_STRING in redacted
@patch(
"litellm.proxy.spend_tracking.spend_tracking_utils._should_store_prompts_and_responses_in_spend_logs"
)
def test_sanitize_error_information_redacts_traceback_when_not_storing_prompts(
mock_should_store,
):
# If a Python exception bubbles up with the request body embedded in
# its repr (e.g. ValueError(f"bad request: {body}")), the traceback
# column would carry the prompt unredacted. Verify the redaction
# covers the traceback field too.
mock_should_store.return_value = False
error_info = {
"error_code": "500",
"error_class": "ValueError",
"llm_provider": "",
"traceback": (
'Traceback (most recent call last):\n File "x.py", line 1, in <module>\n'
' raise ValueError({"input":[{"role":"user","content":"tb-leaked-prompt"}]})\n'
"ValueError: invalid request"
),
"error_message": "invalid request",
}
sanitized = _sanitize_error_information_for_spend_logs(error_info)
assert sanitized is not None
assert "tb-leaked-prompt" not in sanitized["traceback"]
assert REDACTED_BY_LITELM_STRING in sanitized["traceback"]
# Surrounding traceback frames remain so the error stays debuggable.
assert "Traceback (most recent call last):" in sanitized["traceback"]
assert "ValueError: invalid request" in sanitized["traceback"]
@patch(
"litellm.proxy.spend_tracking.spend_tracking_utils._should_store_prompts_and_responses_in_spend_logs"
)
def test_sanitize_error_information_skips_traceback_redaction_when_storing_prompts(
mock_should_store,
):
mock_should_store.return_value = True
error_info = {
"error_code": "500",
"error_class": "ValueError",
"llm_provider": "",
"traceback": (
'raise ValueError({"input":[{"role":"user","content":"tb-kept"}]})'
),
"error_message": "invalid request",
}
sanitized = _sanitize_error_information_for_spend_logs(error_info)
assert sanitized is not None
assert "tb-kept" in sanitized["traceback"]
assert REDACTED_BY_LITELM_STRING not in sanitized["traceback"]
def test_redact_prompt_leaks_strips_prompt_key_completions_payload():
# /v1/completions echoes the user input under the top-level 'prompt' key
# rather than 'messages'. Without 'prompt' coverage the body would survive
# the redactor when store_prompts_in_spend_logs is False.
error_text = (
'{"error":{"message":"validation failed",'
'"prompt":"super-secret-completion-text"}}'
)
redacted = _redact_prompt_leaks_in_error_string(error_text)
assert "super-secret-completion-text" not in redacted
assert REDACTED_BY_LITELM_STRING in redacted
def test_redact_prompt_leaks_strips_prompt_key_python_repr():
error_text = (
"{'model': 'gpt-3.5-turbo-instruct', "
"'prompt': 'leaked-completion-prompt-body'}"
)
redacted = _redact_prompt_leaks_in_error_string(error_text)
assert "leaked-completion-prompt-body" not in redacted
assert REDACTED_BY_LITELM_STRING in redacted
def test_redact_prompt_leaks_preserves_prompt_substring_keys():
# 'prompt_tokens' / 'prompt_token_count' / etc. are not the leak key —
# the matcher requires a closing quote before ':' so substrings shouldn't
# trigger redaction.
error_text = '{"usage":{"prompt_tokens":42,"completion_tokens":7}}'
assert _redact_prompt_leaks_in_error_string(error_text) == error_text
def test_redact_prompt_leaks_strips_pydantic_input_value_list():
# Pydantic v2 validation error format — the offending value is rendered
# as a Python repr after `input_value=`. The full repr can carry the
# entire request body and must be redacted under the same gate as the
# quoted-key form.
error_text = (
"1 validation error for ChatCompletionRequest\n"
"messages\n"
" Input should be a valid list "
"[type=list_type, input_value=['secret-pydantic-prompt'], input_type=str]"
)
redacted = _redact_prompt_leaks_in_error_string(error_text)
assert "secret-pydantic-prompt" not in redacted
assert REDACTED_BY_LITELM_STRING in redacted
# Surrounding pydantic context is preserved so the error stays debuggable.
assert "list_type" in redacted
assert "input_type=str" in redacted
def test_redact_prompt_leaks_strips_pydantic_input_value_dict():
error_text = (
"[type=dict_type, "
"input_value={'role': 'user', 'content': 'leaked-dict-content'}, "
"input_type=dict]"
)
redacted = _redact_prompt_leaks_in_error_string(error_text)
assert "leaked-dict-content" not in redacted
assert REDACTED_BY_LITELM_STRING in redacted
assert "input_type=dict" in redacted
def test_redact_prompt_leaks_strips_pydantic_input_value_quoted_string():
error_text = "[type=string_type, input_value='leaked-string-value', input_type=str]"
redacted = _redact_prompt_leaks_in_error_string(error_text)
assert "leaked-string-value" not in redacted
assert REDACTED_BY_LITELM_STRING in redacted
def test_redact_prompt_leaks_pydantic_input_value_scalar_left_intact():
# Bare numeric / bool / None scalars in input_value are not prompt
# carriers; leaving them untouched keeps the validation error readable.
error_text = "[type=int_type, input_value=42, input_type=int]"
assert _redact_prompt_leaks_in_error_string(error_text) == error_text
def test_redact_prompt_leaks_input_value_substring_does_not_match():
# Word-boundary anchoring on `input_value` so similarly named keys (e.g.
# `my_input_value=` from another stack frame) are not mis-redacted.
error_text = "frame: my_input_value=42 elsewhere"
assert _redact_prompt_leaks_in_error_string(error_text) == error_text
def test_redact_prompt_leaks_combined_quoted_key_and_pydantic_assignment():
# A real OpenAI/pydantic error often carries BOTH forms in the same
# string — quoted JSON 'input' echoed once, then pydantic 'input_value='
# echoed per validation entry. Both must be redacted in one pass.
error_text = (
'{"error":{"input":[{"role":"user","content":"leak-via-json"}]}} '
"[type=list_type, input_value=['leak-via-pydantic'], input_type=str]"
)
redacted = _redact_prompt_leaks_in_error_string(error_text)
assert "leak-via-json" not in redacted
assert "leak-via-pydantic" not in redacted
assert redacted.count(REDACTED_BY_LITELM_STRING) >= 2
@patch(
"litellm.proxy.spend_tracking.spend_tracking_utils._should_store_prompts_and_responses_in_spend_logs"
)
def test_sanitize_error_information_redacts_pydantic_assignment_form(
mock_should_store,
):
# End-to-end: a pydantic-style error that lands in error_message must be
# redacted under the spend-log path, not just the regex-level helper.
mock_should_store.return_value = False
error_info = {
"error_code": "422",
"error_class": "ValidationError",
"llm_provider": "openai",
"traceback": "",
"error_message": (
"1 validation error for ChatCompletionRequest\n"
"messages\n"
" Field required "
"[type=missing, input_value={'prompt': 'leaked-via-pydantic-msg'}, "
"input_type=dict]"
),
}
sanitized = _sanitize_error_information_for_spend_logs(error_info)
assert sanitized is not None
assert "leaked-via-pydantic-msg" not in sanitized["error_message"]
assert REDACTED_BY_LITELM_STRING in sanitized["error_message"]