From 0dc3851a9177f61a3535bb4e4fd0bc1a5f2b8096 Mon Sep 17 00:00:00 2001 From: mateo-berri <277851410+mateo-berri@users.noreply.github.com> Date: Thu, 29 Oct 2026 08:57:48 -0700 Subject: [PATCH] fix(logging): keep proxy-set fields over keys parsed from a JSON log message --- litellm/_logging.py | 90 +++++++-------- litellm/proxy/common_request_processing.py | 12 +- litellm/proxy/common_utils/token_count_log.py | 24 +--- litellm/proxy/proxy_server.py | 5 +- tests/unit/proxy/conftest.py | 17 ++- .../proxy/test_common_request_processing.py | 104 ++++++++++++++++++ .../proxy/test_response_model_sanitization.py | 32 +++++- tests/unit/test_logging.py | 30 +++++ 8 files changed, 235 insertions(+), 79 deletions(-) diff --git a/litellm/_logging.py b/litellm/_logging.py index 5279cb233b4..75b354abb81 100644 --- a/litellm/_logging.py +++ b/litellm/_logging.py @@ -3,6 +3,7 @@ import contextvars import functools import inspect import itertools +import json import logging import os import re @@ -115,6 +116,17 @@ def _plain_text(value: object) -> str: return UNSERIALIZABLE_OBJECT +_BARE_LOGFMT_VALUE: Final = re.compile(r"[\w.:/@+-]+", re.ASCII) +_JSON_ESCAPED_BRACES: Final = str.maketrans({"{": "\\u007b", "}": "\\u007d"}) + + +def logfmt_value(value: object) -> str: + text: Final = _plain_text(value) + if _BARE_LOGFMT_VALUE.fullmatch(text): + return text + return json.dumps(text).translate(_JSON_ESCAPED_BRACES) + + def redact_secrets(value: str) -> str: """Public API: redact known secret/credential patterns from an arbitrary string. @@ -754,10 +766,14 @@ def _get_standard_record_attrs() -> frozenset: _STANDARD_RECORD_ATTRS: Final = _get_standard_record_attrs() _NON_EXTRA_RECORD_ATTRS: Final = _STANDARD_RECORD_ATTRS | {_REDACTED_RECORD_ATTR} -# CorrelationContextFilter is the only legitimate source for these two JSON fields; -# see JsonFormatter.format() for why they're excluded from the generic message-content -# and extra-attribute promotion paths. -_RESERVED_CORRELATION_FIELDS: Final = frozenset(("trace_id", "session_id")) +_UNPROMOTABLE_MESSAGE_KEYS: Final = frozenset(("trace_id", "session_id", "stacktrace")) + + +def _parsed_message_fields(message: str) -> Mapping[str, object]: + parsed: Final = _try_parse_json_message(message) + if parsed is not None: + return parsed + return _try_parse_embedded_python_dict(message) or {} class JsonFormatter(Formatter): @@ -769,54 +785,28 @@ class JsonFormatter(Formatter): dt: Final = datetime.fromtimestamp(record.created) return dt.isoformat() - def format(self, record): + def format(self, record: logging.LogRecord) -> str: message_str: Final = _ANSI_ESCAPE.sub("", record.getMessage()) - json_record: Final[dict[str, object]] = { - "message": message_str, - "level": record.levelname, - "timestamp": self.formatTime(record), + fixed_fields: Final = {"message": message_str, "level": record.levelname, "timestamp": self.formatTime(record)} + extra_fields: Final = { + key: value for key, value in _record_attributes(record).items() if key not in _NON_EXTRA_RECORD_ATTRS } - - # Parse embedded JSON or Python dict repr in message so sub-fields become first-class properties. - # trace_id/session_id are excluded here unconditionally (not just "if not already - # set") - CorrelationContextFilter is the only legitimate source for these two - # fields, and a message that merely happens to parse as JSON/dict (e.g. a proxy - # log line dumping raw request headers) must never be able to claim them, even on - # a record the filter hasn't stamped yet (no correlation context active for it). - parsed = _try_parse_json_message(message_str) - if parsed is None: - parsed = _try_parse_embedded_python_dict(message_str) - if parsed is not None: - for key, value in parsed.items(): - if key not in json_record and key not in _RESERVED_CORRELATION_FIELDS: - json_record[key] = value - - # Include extra attributes passed via logger.debug("msg", extra={...}) - for key, value in _record_attributes(record).items(): - if key not in _NON_EXTRA_RECORD_ATTRS and key not in json_record: - json_record[key] = value - - # trace_id/session_id are reserved: CorrelationContextFilter is the only - # legitimate source for these two fields. Without this, a message string - # that happens to parse as JSON/dict (e.g. a proxy log line dumping raw - # request headers) with a "trace_id"/"session_id" key would have already - # claimed the key at the parsed-message step above, and the extra-attributes - # loop's "key not in json_record" guard would then skip the real value - - # letting a caller-supplied header spoof another request's correlation ids. - for reserved_key in _RESERVED_CORRELATION_FIELDS: - value = getattr(record, reserved_key, None) - if value: - json_record[reserved_key] = value - - # Set component/logger only if not already supplied via extra={...} - if "component" not in json_record: - json_record["component"] = record.name - if "logger" not in json_record: - json_record["logger"] = f"{record.filename}:{record.lineno}" - - if record.exc_info: - json_record["stacktrace"] = record.exc_text or self.formatException(record.exc_info) - + stacktrace_field: Final[Mapping[str, str]] = ( + {"stacktrace": record.exc_text or self.formatException(record.exc_info)} if record.exc_info else {} + ) + logger_fields: Final = { + "component": record.name, + "logger": f"{record.filename}:{record.lineno}", + **extra_fields, + **stacktrace_field, + **fixed_fields, + } + message_fields: Final = { + key: value + for key, value in _parsed_message_fields(message_str).items() + if key not in logger_fields and key not in _UNPROMOTABLE_MESSAGE_KEYS + } + json_record: Final = {**fixed_fields, **message_fields, **logger_fields} return safe_dumps( json_record if _is_redacted(record) or not _ENABLE_SECRET_REDACTION else _redact_json_record(json_record) ) diff --git a/litellm/proxy/common_request_processing.py b/litellm/proxy/common_request_processing.py index fc29a6e3a08..897174ea6b6 100644 --- a/litellm/proxy/common_request_processing.py +++ b/litellm/proxy/common_request_processing.py @@ -34,7 +34,7 @@ from starlette.types import Receive, Scope, Send from typing_extensions import Never import litellm -from litellm._logging import redact_internal_details_from_client_message, verbose_proxy_logger +from litellm._logging import logfmt_value, redact_internal_details_from_client_message, verbose_proxy_logger from litellm._uuid import uuid from litellm.anthropic_interface.exceptions import AnthropicErrorSseFrame, anthropic_error_sse_frame from litellm.constants import ( @@ -403,7 +403,7 @@ async def _record_streaming_client_disconnect_if_needed( verbose_proxy_logger.debug( "Recorded streaming client disconnect with error_code=499 for litellm_call_id=%s", - request_data.get("litellm_call_id"), + logfmt_value(request_data.get("litellm_call_id")), ) return True @@ -492,7 +492,7 @@ async def _bill_partial_streamed_spend_on_disconnect(request_data: dict, respons if partial_responses_event is not None: verbose_proxy_logger.debug( "Billing partial streamed Responses API spend after client disconnect, litellm_call_id=%s", - sanitize_for_log(logging_obj.litellm_call_id), + logfmt_value(logging_obj.litellm_call_id), ) return await _dispatch_disconnect_billing_event(logging_obj, partial_responses_event) chunks: Final[object] = getattr(response, "chunks", None) @@ -501,7 +501,7 @@ async def _bill_partial_streamed_spend_on_disconnect(request_data: dict, respons verbose_proxy_logger.debug( "Billing partial streamed spend for %s chunks after client disconnect, litellm_call_id=%s", len(chunks), - request_data.get("litellm_call_id"), + logfmt_value(request_data.get("litellm_call_id")), ) messages: Final[object] = getattr(response, "messages", None) try: @@ -1871,7 +1871,7 @@ def log_llm_api_exception(e: Exception, litellm_call_id: str | None) -> None: verbose_proxy_logger.info( "litellm.proxy.proxy_server._handle_llm_api_exception(): client disconnected, " "upstream LLM request cancelled - litellm_call_id=%s", - litellm_call_id, + logfmt_value(litellm_call_id), extra=MappingProxyType({"litellm_call_id": litellm_call_id}), ) return @@ -1882,7 +1882,7 @@ def log_llm_api_exception(e: Exception, litellm_call_id: str | None) -> None: ) log_fn( "litellm.proxy.proxy_server._handle_llm_api_exception(): Exception occured - litellm_call_id=%s - %s", - litellm_call_id, + logfmt_value(litellm_call_id), e, extra=MappingProxyType({"litellm_call_id": litellm_call_id}), ) diff --git a/litellm/proxy/common_utils/token_count_log.py b/litellm/proxy/common_utils/token_count_log.py index 04c07a29117..46e61bbd8bb 100644 --- a/litellm/proxy/common_utils/token_count_log.py +++ b/litellm/proxy/common_utils/token_count_log.py @@ -1,5 +1,3 @@ -import json -import re from asyncio import CancelledError from collections.abc import Awaitable, Callable, Mapping from dataclasses import dataclass @@ -8,23 +6,11 @@ from typing import Final, TypeVar from fastapi import status -from litellm._logging import verbose_proxy_logger +from litellm._logging import logfmt_value, verbose_proxy_logger from litellm.proxy.common_utils.openai_error_payload import error_status_code from litellm.types.token_count import TokenCountSource from litellm.types.utils import TokenCountResponse -_BARE_LOGFMT_VALUE: Final = re.compile(r"[\w.:/@+-]+", re.ASCII) -_JSON_ESCAPED_BRACES: Final = str.maketrans({"{": "\\u007b", "}": "\\u007d"}) - - -def _logfmt_value(value: str | None) -> str: - if value is None: - return "None" - if _BARE_LOGFMT_VALUE.fullmatch(value): - return value - return json.dumps(value).translate(_JSON_ESCAPED_BRACES) - - _Counted: Final = TypeVar("_Counted") _Rejection: Final = TypeVar("_Rejection") _ProviderCount: Final = TypeVar("_ProviderCount", bound=Mapping[str, object]) @@ -59,10 +45,10 @@ def _log_token_count(request: _CountRequest, fields: _CountFields, status_code: verbose_proxy_logger.info( "token count - litellm_call_id=%s request_model=%s model_id=%s model_used=%s input_tokens=%s " "count_source=%s status_code=%s", - _logfmt_value(request.litellm_call_id), - _logfmt_value(request.request_model), - _logfmt_value(request.model_id), - _logfmt_value(fields.model_used), + logfmt_value(request.litellm_call_id), + logfmt_value(request.request_model), + logfmt_value(request.model_id), + logfmt_value(fields.model_used), fields.input_tokens, fields.count_source, status_code, diff --git a/litellm/proxy/proxy_server.py b/litellm/proxy/proxy_server.py index 091e05610ba..65c4b5411b9 100644 --- a/litellm/proxy/proxy_server.py +++ b/litellm/proxy/proxy_server.py @@ -311,6 +311,7 @@ from litellm import Router from litellm._internal_context import service_target, with_service_target from litellm._logging import ( STARTUP_OUTPUT_SETTINGS, + logfmt_value, print_startup_message, redact_string, verbose_proxy_logger, @@ -9980,7 +9981,7 @@ def _restamp_streaming_chunk_model( if not model_mismatch_logged and downstream_model != target_model: verbose_proxy_logger.debug( "litellm_call_id=%s: streaming chunk model mismatch - target=%r downstream=%r fallback_was_attempted=%s. Overriding chunk model to target.", - request_data.get("litellm_call_id"), + logfmt_value(request_data.get("litellm_call_id")), target_model, downstream_model, fallback_was_attempted, @@ -10001,7 +10002,7 @@ def _restamp_streaming_chunk_model( except Exception as e: verbose_proxy_logger.error( "litellm_call_id=%s: failed to override chunk.model=%r on chunk_type=%s. error=%s", - request_data.get("litellm_call_id"), + logfmt_value(request_data.get("litellm_call_id")), target_model, type(restamped_chunk), str(e), diff --git a/tests/unit/proxy/conftest.py b/tests/unit/proxy/conftest.py index e13eddef43b..1b58f42d25d 100644 --- a/tests/unit/proxy/conftest.py +++ b/tests/unit/proxy/conftest.py @@ -3,11 +3,13 @@ import asyncio import copy import inspect +import json import logging import os import tempfile import warnings from collections.abc import Awaitable, Callable, Iterator +from io import StringIO from typing import Dict, Final, Optional from unittest.mock import AsyncMock, MagicMock, patch @@ -18,7 +20,7 @@ from prisma.errors import ClientNotConnectedError import litellm import litellm.proxy.proxy_server -from litellm._logging import verbose_proxy_logger +from litellm._logging import JsonFormatter, verbose_proxy_logger from litellm._service_logger import ServiceTypes from litellm.integrations.otel.model.payloads import ServiceSpanData from litellm.integrations.otel.model.spans import service_span_name @@ -451,6 +453,19 @@ def token_count_log_lines( ) +@pytest.fixture +def proxy_json_log_lines() -> Iterator[Callable[[], tuple[dict[str, object], ...]]]: + stream: Final = StringIO() + handler: Final = logging.StreamHandler(stream) + handler.setFormatter(JsonFormatter()) + previous_level: Final = verbose_proxy_logger.level + verbose_proxy_logger.addHandler(handler) + verbose_proxy_logger.setLevel(logging.DEBUG) + yield lambda: tuple(json.loads(line) for line in stream.getvalue().splitlines()) + verbose_proxy_logger.removeHandler(handler) + verbose_proxy_logger.setLevel(previous_level) + + @pytest.fixture def postgres_span_names() -> Iterator[Callable[[], Awaitable[tuple[str, ...]]]]: """The ``postgres.{verb} {table}`` names OTel would render for every DB service event diff --git a/tests/unit/proxy/test_common_request_processing.py b/tests/unit/proxy/test_common_request_processing.py index fdcbcbfcd05..d777b315a22 100644 --- a/tests/unit/proxy/test_common_request_processing.py +++ b/tests/unit/proxy/test_common_request_processing.py @@ -27,8 +27,10 @@ from litellm.litellm_core_utils.bug_report import ( ) from litellm.types.router import RouterRateLimitError, RouterRateLimitErrorBasic from litellm.types.utils import CachingDetails, Delta, ModelResponseStream, StreamingChoices +from litellm._logging import verbose_proxy_logger from litellm.constants import ( CLIENT_REQUESTED_MODEL_SCOPE_KEY, + LITELLM_CLIENT_DISCONNECT_DETAIL, MAX_LITELLM_CALL_ID_LENGTH, RETURN_RAW_MODEL_NAME_METADATA_KEY, STREAM_SSE_KEEPALIVE_PING_BYTES, @@ -51,6 +53,7 @@ from litellm.proxy.common_request_processing import ( _has_attribute_error_in_chain, include_guardrail_response_requested, is_azure_model_router_request, + log_llm_api_exception, open_sse_before_first_byte, resolve_litellm_call_id, ttft_keepalive_interval, @@ -82,6 +85,8 @@ from litellm.litellm_core_utils.streaming_handler import CustomStreamWrapper from litellm.utils import ModelResponseListIterator from litellm.router_utils.add_retry_fallback_headers import prepare_response_for_header_attachment +_DICT_SHAPED_CALL_ID: Final = "{'component': 'forged', 'logger': 'x', 'litellm_call_id': 'y', 'injected': 1}" + def test_attach_guardrail_information_copies_recorded_entries_onto_model_response(): recorded = [ @@ -8765,6 +8770,73 @@ class TestStreamingClientDisconnectBilling: await stream_iter.__anext__() return response + @pytest.mark.asyncio + async def test_dict_shaped_call_id_cannot_forge_fields_on_the_partial_responses_billing_log( + self, proxy_json_log_lines: Callable[[], tuple[dict[str, object], ...]] + ) -> None: + from litellm.llms.openai.responses.transformation import OpenAIResponsesAPIConfig + from litellm.responses.streaming_iterator import ResponsesAPIStreamingIterator + + logging_obj: Final = LiteLLMLoggingObj( + model="gpt-4o-mini", + messages=[{"role": "user", "content": "tell me a story"}], + stream=True, + call_type="aresponses", + start_time=datetime.datetime.now(), + litellm_call_id=_DICT_SHAPED_CALL_ID, + function_id="test-function-id", + ) + stream: Final = ResponsesAPIStreamingIterator( + response=httpx.Response(200), + model="gpt-4o-mini", + responses_api_provider_config=OpenAIResponsesAPIConfig(), + logging_obj=logging_obj, + request_data={"model": "gpt-4o-mini", "input": "tell me a story"}, + custom_llm_provider="openai", + ) + stream._process_chunk( + json.dumps( + { + "type": "response.created", + "sequence_number": 0, + "response": { + "id": "resp_partial", + "object": "response", + "created_at": 1700000000.0, + "model": "gpt-4o-mini", + "status": "in_progress", + "output": [], + "parallel_tool_calls": True, + "tool_choice": "auto", + "tools": [], + }, + } + ) + ) + stream._process_chunk( + json.dumps( + { + "type": "response.output_text.delta", + "sequence_number": 1, + "item_id": "msg_partial", + "output_index": 0, + "content_index": 0, + "delta": "The codename is AZURE-FALCON-42", + } + ) + ) + + await _bill_partial_streamed_spend_on_disconnect({"litellm_logging_obj": logging_obj}, stream) + + (line,) = ( + line + for line in proxy_json_log_lines() + if "Billing partial streamed Responses API spend" in str(line["message"]) + ) + assert line["component"] == verbose_proxy_logger.name + assert "litellm_call_id" not in line + assert "injected" not in line + @pytest.mark.asyncio async def test_disconnect_bills_partial_streamed_spend(self): recorder = _RecordingSuccessLogger() @@ -11968,6 +12040,38 @@ class TestErrorLogCarriesCallId: assert record.litellm_call_id == call_id assert call_id in record.getMessage() + @pytest.mark.parametrize( + "exc", + [ + pytest.param(HTTPException(status_code=400, detail={"error": "Invalid model name"}), id="error_line"), + pytest.param( + HTTPException(status_code=499, detail=LITELLM_CLIENT_DISCONNECT_DETAIL), id="client_disconnect_line" + ), + ], + ) + def test_dict_shaped_call_id_cannot_forge_json_log_fields( + self, exc: HTTPException, proxy_json_log_lines: Callable[[], tuple[dict[str, object], ...]] + ) -> None: + log_llm_api_exception(exc, _DICT_SHAPED_CALL_ID) + + (line,) = proxy_json_log_lines() + assert line["component"] == verbose_proxy_logger.name + assert str(line["logger"]).startswith("common_request_processing.py:") + assert line["litellm_call_id"] == _DICT_SHAPED_CALL_ID + assert "injected" not in line + + def test_provider_error_json_in_the_exception_still_becomes_a_field_next_to_a_dict_shaped_call_id( + self, proxy_json_log_lines: Callable[[], tuple[dict[str, object], ...]] + ) -> None: + provider_error: Final = "Invalid model name passed in model=nope-model" + + log_llm_api_exception( + HTTPException(status_code=400, detail={"error": provider_error}), "{'error': 'forged', 'injected': 1}" + ) + + (line,) = proxy_json_log_lines() + assert line["error"] == provider_error + class TestAnthropicMessagesStreamErrorFrame: """A ``/v1/messages`` stream that fails after the headers are out has to say so with an diff --git a/tests/unit/proxy/test_response_model_sanitization.py b/tests/unit/proxy/test_response_model_sanitization.py index f52736453c1..701282554d7 100644 --- a/tests/unit/proxy/test_response_model_sanitization.py +++ b/tests/unit/proxy/test_response_model_sanitization.py @@ -1,11 +1,13 @@ import asyncio import json -from typing import AsyncGenerator +from collections.abc import Callable +from typing import AsyncGenerator, Final from unittest.mock import AsyncMock, MagicMock import pytest import yaml from fastapi.testclient import TestClient +from pydantic import BaseModel import litellm @@ -502,6 +504,34 @@ def test_restamp_streaming_chunk_rewrites_nested_response_model(): assert chunk.response.model == "hosted_vllm/some-model" +class _UncopyableChunk(BaseModel): + model: str + + def model_copy(self, *, update: object = None, deep: bool = False) -> "_UncopyableChunk": + raise RuntimeError("chunk cannot be copied") + + +def test_a_failed_restamp_logs_a_dict_shaped_call_id_without_forging_fields( + proxy_json_log_lines: Callable[[], tuple[dict[str, object], ...]], +): + from litellm._logging import verbose_proxy_logger + from litellm.proxy.proxy_server import _restamp_streaming_chunk_model + + chunk: Final = _UncopyableChunk(model="hosted_vllm/some-model") + result, _ = _restamp_streaming_chunk_model( + chunk=chunk, + requested_model_from_client="client-model", + request_data={"litellm_call_id": "{'component': 'forged', 'litellm_call_id': 'y', 'injected': 1}"}, + model_mismatch_logged=True, + ) + + assert result is chunk + (line,) = (line for line in proxy_json_log_lines() if "failed to override chunk.model" in str(line["message"])) + assert line["component"] == verbose_proxy_logger.name + assert "litellm_call_id" not in line + assert "injected" not in line + + def _make_response_completed_event(model: str): from litellm.types.llms.openai import ( ResponseCompletedEvent, diff --git a/tests/unit/test_logging.py b/tests/unit/test_logging.py index fcd835e93dd..56134c94601 100644 --- a/tests/unit/test_logging.py +++ b/tests/unit/test_logging.py @@ -555,6 +555,36 @@ def test_trace_id_and_session_id_cannot_be_injected_with_no_active_context(monke assert "session_id" not in cap.records[0] +def test_message_dict_keys_never_override_fields_the_logger_sets(): + lg, cap = _make_capture_logger("test.message_dict_precedence") + embedded: Final = { + "component": "forged-component", + "logger": "forged.py:1", + "stacktrace": "forged stacktrace", + "litellm_call_id": "forged-call-id", + "model_name": "gpt-x", + } + lg.handle( + lg.makeRecord( + lg.name, + logging.INFO, + "/srv/litellm/router.py", + 42, + f"Selected deployment: {embedded} for model: gpt-x", + (), + None, + extra={"litellm_call_id": "real-call-id"}, + ) + ) + + record: Final = cap.records[0] + assert record["component"] == "test.message_dict_precedence" + assert record["logger"] == "router.py:42" + assert record["litellm_call_id"] == "real-call-id" + assert "stacktrace" not in record + assert record["model_name"] == "gpt-x" + + def test_trace_id_and_session_id_are_redacted_when_credential_shaped(monkeypatch): """A caller-controlled trace_id/session_id (e.g. from x-litellm-trace-id or a W3C baggage header) that happens to look like a real credential must not reach log