fix(logging): redact nested callback input and provider response fields

This commit is contained in:
moyai-devin-berriai[bot] 2026-10-11 05:51:58 +00:00 • committed by GitHub
parent 2ab80ba4fa
commit 4a581a0470
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
4 changed files with 517 additions and 9 deletions

View file

@ -14,6 +14,8 @@ from collections.abc import Mapping
from dataclasses import replace
from typing import TYPE_CHECKING, Any, Final
from pydantic import TypeAdapter
import litellm
from litellm.constants import REDACTED_BY_LITELLM
from litellm.integrations.custom_logger import CustomLogger
@ -30,6 +32,8 @@ from litellm.secret_managers.main import str_to_bool
from litellm.types.router import BaselineRouteStamp
from litellm.types.utils import StandardCallbackDynamicParams
_REQUEST_BODY: Final = TypeAdapter(Mapping[object, object])
if TYPE_CHECKING:
from litellm.litellm_core_utils.litellm_logging import (
Logging as _LiteLLMLoggingObject,
@ -66,8 +70,14 @@ def redact_streaming_responses_for_custom_logger(model_call_details: dict, custo
return {**model_call_details, **redacted_entries}
def _redacted_streaming_response_copy(streaming_response):
redacted_response: Final = copy.deepcopy(streaming_response)
def _redacted_streaming_response_copy(streaming_response: object) -> object:
hidden_params: Final = getattr(streaming_response, "_hidden_params", None)
memo: Final = (
{id(hidden_params): {**_REQUEST_BODY.validate_python(hidden_params)}}
if isinstance(hidden_params, Mapping)
else {}
)
redacted_response: Final = copy.deepcopy(streaming_response, memo)
_redact_streaming_response(redacted_response)
return redacted_response
@ -108,6 +118,7 @@ def _redact_choice_content(choice):
choice.message.reasoning_content = REDACTED_BY_LITELLM
if hasattr(choice.message, "thinking_blocks"):
choice.message.thinking_blocks = None
choice.message.provider_specific_fields = None # rebind-ok: scrub the caller-owned logging copy in place
_redact_tool_calls(getattr(choice.message, "tool_calls", None))
_redact_function_call(getattr(choice.message, "function_call", None))
elif isinstance(choice, litellm.utils.StreamingChoices):
@ -117,6 +128,7 @@ def _redact_choice_content(choice):
choice.delta.reasoning_content = REDACTED_BY_LITELLM
if hasattr(choice.delta, "thinking_blocks"):
choice.delta.thinking_blocks = None
choice.delta.provider_specific_fields = None # rebind-ok: scrub the caller-owned logging copy in place
_redact_tool_calls(getattr(choice.delta, "tool_calls", None))
_redact_function_call(getattr(choice.delta, "function_call", None))
@ -237,6 +249,7 @@ def _redact_model_response_dict_choices(choices, redacted_str: str):
choice["message"]["thinking_blocks"] = None
if "audio" in choice["message"]:
choice["message"]["audio"] = None
choice["message"]["provider_specific_fields"] = None
_redact_tool_calls_dict(choice["message"])
elif "delta" in choice and isinstance(choice["delta"], dict):
if choice["delta"].get("content") is not None:
@ -247,6 +260,7 @@ def _redact_model_response_dict_choices(choices, redacted_str: str):
choice["delta"]["thinking_blocks"] = None
if "audio" in choice["delta"]:
choice["delta"]["audio"] = None
choice["delta"]["provider_specific_fields"] = None
_redact_tool_calls_dict(choice["delta"])
elif choice.get("text") is not None:
choice["text"] = redacted_str
@ -261,6 +275,22 @@ def _redacted_baseline_metadata(metadata: Mapping[str, object]) -> Mapping[str,
return {**metadata, "_autorouter_baseline_route": replace(route, request_parameters=None)}
def _redacted_proxy_request(request: Mapping[str, object]) -> Mapping[str, object]:
raw_body: Final = request.get("body")
if not isinstance(raw_body, Mapping):
return without_classifier_audit(request)
body: Final = _REQUEST_BODY.validate_python(raw_body)
return {
**without_classifier_audit(request),
"body": {
**body,
**({"messages": [{"role": "user", "content": REDACTED_BY_LITELLM}]} if "messages" in body else {}),
**({"prompt": ""} if "prompt" in body else {}),
**({"input": ""} if "input" in body else {}),
},
}
def redacted_litellm_params(params: Mapping[str, object]) -> dict[str, object]:
request: Final = params.get("proxy_server_request")
return {
@ -270,7 +300,7 @@ def redacted_litellm_params(params: Mapping[str, object]) -> dict[str, object]:
for key, value in params.items()
if key in ("metadata", "litellm_metadata") and isinstance(value, Mapping)
},
**({"proxy_server_request": without_classifier_audit(request)} if isinstance(request, Mapping) else {}),
**({"proxy_server_request": _redacted_proxy_request(request)} if isinstance(request, Mapping) else {}),
}
@ -278,7 +308,7 @@ def perform_redaction(model_call_details: dict, result, redact_streaming_respons
"""
Performs the actual redaction on the logging object and result.
redact_streaming_responses=False skips the in-place redaction of the shared streaming
redact_streaming_responses=False skips replacing the shared streaming
response entries; per-callback redaction hands each opted-out callback its own redacted
copy via redact_streaming_responses_for_custom_logger instead.
"""
@ -298,7 +328,10 @@ def perform_redaction(model_call_details: dict, result, redact_streaming_respons
# Redact streaming response
if redact_streaming_responses and model_call_details.get("stream", False) is True:
for _streaming_key in ("complete_streaming_response", "async_complete_streaming_response"):
_redact_streaming_response(model_call_details.get(_streaming_key))
if model_call_details.get(_streaming_key) is not None:
model_call_details[_streaming_key] = ( # rebind-ok: isolate the logging copy
_redacted_streaming_response_copy(model_call_details[_streaming_key])
)
# Redact result
if result is not None:

View file

@ -0,0 +1,47 @@
import json
import os
from collections.abc import Mapping
from datetime import datetime
from typing import Final
import httpx
from pydantic import BaseModel
from litellm.integrations.custom_logger import CustomLogger
def _payload(kwargs: Mapping[str, object], response_obj: object) -> str:
params: Final = kwargs.get("litellm_params")
request: Final = params.get("proxy_server_request") if isinstance(params, Mapping) else None
standard: Final = kwargs.get("standard_logging_object")
return json.dumps(
{
"request": request,
"response": response_obj.model_dump() if isinstance(response_obj, BaseModel) else response_obj,
"standard_response": standard.get("response") if isinstance(standard, Mapping) else None,
},
default=str,
)
class SyncRedactionProbe(CustomLogger):
def log_success_event(
self, kwargs: Mapping[str, object], response_obj: object, start_time: datetime, end_time: datetime
) -> None:
with httpx.Client(trust_env=False) as client:
response: Final = client.post(os.environ["REDACTION_PROBE_URL"], content=_payload(kwargs, response_obj))
response.raise_for_status()
class RedactionProbe(CustomLogger):
async def async_log_success_event(
self, kwargs: Mapping[str, object], response_obj: object, start_time: datetime, end_time: datetime
) -> None:
async with httpx.AsyncClient(trust_env=False) as client:
response: Final = await client.post(
os.environ["REDACTION_PROBE_URL"], content=_payload(kwargs, response_obj)
)
response.raise_for_status()
probe: Final = RedactionProbe()

View file

@ -0,0 +1,369 @@
import asyncio
import json
from collections.abc import Iterator
from concurrent.futures import ThreadPoolExecutor
from pathlib import Path
from typing import Final
import pytest
import yaml
from integration._support.client import Gateway, eventually, gateway_from_environment
from integration._support.process import owned_proxy
from integration._support.redaction_probe import RedactionProbe, SyncRedactionProbe
from integration._support.wire import Reply, Request, Wire, wire_server
import litellm
from litellm.litellm_core_utils.logging_worker import GLOBAL_LOGGING_WORKER
INPUT: Final = "private-input-redaction-canary"
REASONING: Final = "private-reasoning-redaction-canary"
OUTPUT: Final = "private-output-redaction-canary"
pytestmark = pytest.mark.timeout(240)
def _provider(request: Request) -> Reply:
if request.method == "GET" and request.target == "/v1/models":
return Reply(body=b'{"object":"list","data":[]}')
body: Final = json.loads(request.body)
if request.target == "/v1/embeddings":
assert body["input"] == INPUT, body
return Reply(
body=json.dumps(
{
"object": "list",
"model": body["model"],
"data": [{"object": "embedding", "index": 0, "embedding": [0.25, 0.5]}],
"usage": {"prompt_tokens": 10, "total_tokens": 10},
}
).encode()
)
if request.target == "/v1/completions":
assert body["prompt"] == INPUT, body
return Reply(
body=json.dumps(
{
"id": "cmpl-redaction-probe",
"object": "text_completion",
"created": 1,
"model": body["model"],
"choices": [{"text": OUTPUT, "index": 0, "finish_reason": "stop", "logprobs": None}],
"usage": {"prompt_tokens": 10, "completion_tokens": 5, "total_tokens": 15},
}
).encode()
)
if request.target == "/v1/responses":
assert body["input"] == INPUT, body
return Reply(
body=json.dumps(
{
"id": "resp-redaction-probe",
"object": "response",
"created_at": 1,
"model": body["model"],
"status": "completed",
"output": [
{
"id": "msg-redaction-probe",
"type": "message",
"role": "assistant",
"status": "completed",
"content": [{"type": "output_text", "text": OUTPUT, "annotations": []}],
}
],
"usage": {"input_tokens": 10, "output_tokens": 5, "total_tokens": 15},
}
).encode()
)
if request.target == "/v1/messages":
assert body["messages"] == [{"role": "user", "content": INPUT}], body
return Reply(
body=json.dumps(
{
"id": "msg-redaction-probe",
"type": "message",
"role": "assistant",
"model": body["model"],
"content": [
{"type": "thinking", "thinking": REASONING, "signature": "synthetic-signature"},
{"type": "text", "text": OUTPUT},
],
"stop_reason": "end_turn",
"stop_sequence": None,
"usage": {"input_tokens": 10, "output_tokens": 5},
}
).encode()
)
assert body["messages"] == [{"role": "user", "content": INPUT}], body
payload: Final = {
"id": "chatcmpl-redaction-" + body.get("user", "probe"),
"object": "chat.completion",
"created": 1,
"model": "gpt-4o-mini",
"choices": [
{
"index": 0,
"message": {
"role": "assistant",
"content": OUTPUT,
"provider_specific_fields": {
"reasoning_content": REASONING,
"thinking_blocks": [{"thinking": REASONING}],
"reasoningContentBlocks": [{"reasoningText": {"text": REASONING}}],
},
},
"finish_reason": "stop",
}
],
"usage": {"prompt_tokens": 10, "completion_tokens": 5, "total_tokens": 15},
}
if body.get("stream"):
choice: Final = payload["choices"][0]
chunk: Final = {
**payload,
"object": "chat.completion.chunk",
"choices": [{"index": 0, "delta": choice["message"], "finish_reason": None}],
}
finish: Final = {
**payload,
"object": "chat.completion.chunk",
"choices": [{"index": 0, "delta": {}, "finish_reason": "stop"}],
}
return Reply(
content_type="text/event-stream",
chunks=tuple(f"data: {json.dumps(event)}\n\n".encode() for event in (chunk, finish))
+ (b"data: [DONE]\n\n",),
)
return Reply(body=json.dumps(payload).encode())
@pytest.fixture(scope="module")
def redaction_proxy(tmp_path_factory: pytest.TempPathFactory) -> Iterator[tuple[Gateway, Wire, Wire]]:
tmp_path: Final = tmp_path_factory.mktemp("custom-logger-redaction")
with (
wire_server(_provider) as provider,
wire_server(lambda request: Reply()) as sink,
gateway_from_environment() as rig,
):
config: Final = tmp_path / "redaction.yaml"
config.write_text(
yaml.safe_dump(
{
"model_list": [
{
"model_name": name,
"litellm_params": {
"model": model,
"api_base": provider.url if model.startswith("anthropic/") else provider.url + "/v1",
"api_key": "synthetic-provider-key",
},
}
for name, model in (
("redaction-probe", "openai/gpt-4o-mini"),
("redaction-text", "openai/gpt-3.5-turbo-instruct"),
("redaction-embedding", "openai/text-embedding-3-large"),
("redaction-responses", "openai/gpt-5.6-sol"),
("redaction-messages", "anthropic/claude-sonnet-5-5"),
)
],
"litellm_settings": {
"callbacks": ["integration._support.redaction_probe.probe"],
"enable_redis_auth_cache": True,
"cache": True,
"cache_params": {
"type": "redis",
"host": "os.environ/REDIS_HOST",
"port": "os.environ/REDIS_PORT",
},
},
"general_settings": {
"master_key": "os.environ/LITELLM_MASTER_KEY",
"database_url": "os.environ/DATABASE_URL",
"disable_model_info_refresh": True,
},
}
)
)
with owned_proxy(
rig, tmp_path, {"REDACTION_PROBE_URL": sink.url, "STORE_MODEL_IN_DB": "False"}, config=config, workers=2
) as proxy:
yield proxy, provider, sink
@pytest.mark.parametrize("redact", [False, True])
@pytest.mark.parametrize("stream", [False, True])
def test_custom_logger_redacts_nested_input_and_provider_reasoning(
redaction_proxy: tuple[Gateway, Wire, Wire], redact: bool, stream: bool
) -> None:
proxy, provider, sink = redaction_proxy
response: Final = proxy.request(
"POST",
"/v1/chat/completions",
{
"model": "redaction-probe",
"messages": [{"role": "user", "content": INPUT}],
"stream": stream,
"cache": {"no-cache": True},
},
headers={"x-litellm-enable-message-redaction": "true"} if redact else {},
)
assert response.status_code == 200, response.text
first: Final = (
json.loads(next(line[6:] for line in response.text.splitlines() if line.startswith("data: ")))
if stream
else response.json()
)
message: Final = first["choices"][0]["delta" if stream else "message"]
assert message["content"] == OUTPUT, message
assert REASONING in json.dumps(message["provider_specific_fields"]), message
received: Final = eventually(sink.drain, lambda values: len(values) == 1, seconds=20)
callback: Final = json.loads(received[0].body)
body: Final = callback["request"]["body"]
assert "proxy_server_request" not in body, body
assert len(tuple(request for request in provider.drain() if request.method == "POST")) == 1
logged: Final = json.dumps(callback)
if redact:
assert INPUT not in logged and REASONING not in logged and OUTPUT not in logged, callback
else:
assert INPUT in logged and REASONING in logged and OUTPUT in logged, callback
@pytest.mark.parametrize("redact", [False, True])
@pytest.mark.parametrize(
("route", "field", "model"),
[("/v1/completions", "prompt", "redaction-text"), ("/v1/embeddings", "input", "redaction-embedding")],
)
def test_custom_logger_redacts_prompt_and_embedding_input(
redaction_proxy: tuple[Gateway, Wire, Wire], route: str, field: str, model: str, redact: bool
) -> None:
proxy, provider, sink = redaction_proxy
response: Final = proxy.request(
"POST",
route,
{"model": model, field: INPUT, "cache": {"no-cache": True}},
headers={"x-litellm-enable-message-redaction": "true"} if redact else {},
)
assert response.status_code == 200, response.text
if field == "input":
assert response.json()["data"][0]["embedding"] == [0.25, 0.5]
else:
assert response.json()["choices"][0]["text"] == OUTPUT
received: Final = eventually(sink.drain, bool, seconds=20)
assert len(received) == 1, received
callback: Final = json.loads(received[0].body)
assert "proxy_server_request" not in callback["request"]["body"]
assert len(tuple(request for request in provider.drain() if request.method == "POST")) == 1
assert (INPUT in json.dumps(callback)) is not redact, callback
@pytest.mark.parametrize("redact", [False, True])
@pytest.mark.parametrize("native_messages", [False, True])
def test_native_messages_and_responses_logging_redaction(
redaction_proxy: tuple[Gateway, Wire, Wire], native_messages: bool, redact: bool
) -> None:
proxy, provider, sink = redaction_proxy
body: Final = (
{"model": "redaction-messages", "messages": [{"role": "user", "content": INPUT}], "max_tokens": 64}
if native_messages
else {"model": "redaction-responses", "input": INPUT}
)
response: Final = proxy.request(
"POST",
"/v1/messages" if native_messages else "/v1/responses",
{**body, "cache": {"no-cache": True}},
headers={"x-litellm-enable-message-redaction": "true"} if redact else {},
)
assert response.status_code == 200, response.text
if native_messages:
assert response.json()["content"][1]["text"] == OUTPUT
assert response.json()["content"][0]["thinking"] == REASONING
else:
assert response.json()["output"][0]["content"][0]["text"] == OUTPUT
received: Final = eventually(sink.drain, bool, seconds=20)
assert len(received) == 1, received
logged: Final = received[0].body.decode()
assert len(tuple(request for request in provider.drain() if request.method == "POST")) == 1
assert (INPUT in logged) is not redact, logged
assert (OUTPUT in logged) is not redact, logged
if native_messages:
assert (REASONING in logged) is not redact, logged
def test_concurrent_redacted_and_unredacted_callbacks_remain_isolated(
redaction_proxy: tuple[Gateway, Wire, Wire],
) -> None:
proxy, provider, sink = redaction_proxy
users: Final = tuple(f"redaction-chaos-{index}" for index in range(4))
def send(index: int):
return proxy.request(
"POST",
"/v1/chat/completions",
{
"model": "redaction-probe",
"messages": [{"role": "user", "content": INPUT}],
"user": users[index],
"cache": {"no-cache": True},
},
headers={"x-litellm-enable-message-redaction": "true"} if index % 2 else {},
)
with ThreadPoolExecutor(max_workers=4) as pool:
responses: Final = tuple(pool.map(send, range(4)))
for response in responses:
assert response.status_code == 200, response.text
assert response.json()["choices"][0]["message"]["content"] == OUTPUT
assert REASONING in json.dumps(response.json()["choices"][0]["message"])
batches: Final[list[Request]] = [] # mutable-ok: drain consumes the sink queue, so polls retain previous events
def observed() -> tuple[Request, ...]:
batches.extend(sink.drain())
return tuple(batches)
events: Final = eventually(observed, lambda values: len(values) == 4, seconds=20)
callbacks: Final = tuple(json.loads(event.body) for event in events)
assert {callback["request"]["body"]["user"] for callback in callbacks} == set(users)
for callback in callbacks:
index: Final = users.index(callback["request"]["body"]["user"])
logged: Final = json.dumps(callback)
assert (INPUT in logged) is (index % 2 == 0), callback
assert (REASONING in logged) is (index % 2 == 0), callback
assert (OUTPUT in logged) is (index % 2 == 0), callback
assert len(tuple(request for request in provider.drain() if request.method == "POST")) == 4
@pytest.mark.asyncio
@pytest.mark.parametrize("asynchronous", [False, True])
@pytest.mark.parametrize("redact", [False, True])
async def test_sdk_success_callback_redaction_over_http(
monkeypatch: pytest.MonkeyPatch, asynchronous: bool, redact: bool
) -> None:
with wire_server(_provider) as provider, wire_server(lambda request: Reply()) as sink:
monkeypatch.setenv("REDACTION_PROBE_URL", sink.url)
messages: Final = [{"role": "user", "content": INPUT}]
request: Final = {"body": {"messages": messages, "model": "gpt-4o-mini"}}
callback: Final = RedactionProbe().async_log_success_event if asynchronous else SyncRedactionProbe()
parameters: Final = {
"model": "openai/gpt-4o-mini",
"messages": messages,
"api_base": provider.url + "/v1",
"api_key": "synthetic-sdk-provider-key",
"proxy_server_request": request,
"turn_off_message_logging": redact,
"success_callback": [callback],
}
response: Final = (
await litellm.acompletion(**parameters)
if asynchronous
else await asyncio.to_thread(litellm.completion, **parameters)
)
assert response.choices[0].message.content == OUTPUT
assert REASONING in json.dumps(response.choices[0].message.provider_specific_fields)
await GLOBAL_LOGGING_WORKER.flush()
delivered: Final = await asyncio.to_thread(eventually, sink.drain, bool, seconds=20)
assert len(delivered) == 1, delivered
logged: Final = delivered[0].body.decode()
assert request["body"]["messages"] == messages
if redact:
assert INPUT not in logged and REASONING not in logged and OUTPUT not in logged, logged
else:
assert INPUT in logged and REASONING in logged and OUTPUT in logged, logged

View file

@ -6,6 +6,7 @@ but litellm_params["litellm_metadata"] is None.
"""
import asyncio
import copy
import importlib
import json
import os
@ -36,6 +37,7 @@ from litellm.responses.main import mock_responses_api_response
from litellm.types.router import BaselineRouteStamp
from litellm.types.utils import (
ModelResponse,
ModelResponseStream,
ResponsesAPIResponse,
StandardLoggingPayload,
TextCompletionResponse,
@ -43,6 +45,60 @@ from litellm.types.utils import (
from tests._vcr_conftest_common import install_live_call_probe, record_vcr_outcome
@pytest.mark.parametrize("as_dict", [False, True])
def test_redaction_removes_nested_callback_content_without_mutating_originals(as_dict: bool) -> None:
secret: Final = "private-callback-content"
part: Final = {
"role": "assistant",
"content": secret,
"provider_specific_fields": {
"reasoning_content": secret,
"thinking_blocks": [{"thinking": secret}],
"reasoningContentBlocks": [{"reasoningText": {"text": secret}}],
"citations": [{"cited_text": secret}],
},
}
response: Final = ModelResponse(choices=[{"index": 0, "message": part}])
original: Final = response.model_dump() if as_dict else response
snapshot: Final = copy.deepcopy(original)
request: Final = {
"url": "http://localhost/v1/chat/completions",
"body": {"model": "redaction-model", "messages": [{"role": "user", "content": secret}]},
}
details: Final = {"litellm_params": {"proxy_server_request": request}}
redacted: Final = perform_redaction(details, original)
assert secret not in str(redacted)
assert secret not in str(details)
assert original == snapshot
assert request["body"]["messages"] == [{"role": "user", "content": secret}]
assert details["litellm_params"]["proxy_server_request"]["body"]["model"] == "redaction-model"
@pytest.mark.parametrize("key", ["complete_streaming_response", "async_complete_streaming_response"])
def test_streaming_provider_reasoning_is_redacted_only_in_callback_copy(key: str) -> None:
response: Final = ModelResponseStream(
choices=[
litellm.utils.StreamingChoices(
delta=litellm.utils.Delta(
content="private-answer",
provider_specific_fields={"thinking_blocks": [{"thinking": "private-reasoning"}]},
)
)
]
)
response._hidden_params["opaque"] = MappingProxyType({"transport": "metadata"})
details: Final = {"stream": True, key: response}
perform_redaction(details, None)
assert details[key].choices[0].delta.provider_specific_fields is None
assert response.choices[0].delta.provider_specific_fields == {
"thinking_blocks": [{"thinking": "private-reasoning"}]
}
assert response.choices[0].delta.content == "private-answer"
assert details[key]._hidden_params["opaque"] is response._hidden_params["opaque"]
@pytest.fixture(autouse=True)
def _reset_global_redaction():
"""Ensure the global setting is off for every test."""
@ -473,8 +529,9 @@ class TestPerformRedaction:
perform_redaction(details, None)
tool_call = streaming_response.choices[0].delta.tool_calls[0]
tool_call = details["complete_streaming_response"].choices[0].delta.tool_calls[0]
assert tool_call.function.arguments == "redacted-by-litellm"
assert streaming_response.choices[0].delta.tool_calls[0].function.arguments == '{"city": "sensitive-city"}'
def test_redacts_tool_call_arguments_in_standard_logging_object(self):
details = {
@ -752,9 +809,11 @@ class TestPerformRedaction:
perform_redaction(details, response)
assert response.choices[0].message.content == "redacted-by-litellm"
assert getattr(response, "vertex_ai_grounding_metadata") == []
assert "vertex_ai_grounding_metadata" not in response._hidden_params
redacted: Final = details["complete_streaming_response"]
assert redacted.choices[0].message.content == "redacted-by-litellm"
assert getattr(redacted, "vertex_ai_grounding_metadata") == []
assert "vertex_ai_grounding_metadata" not in redacted._hidden_params
assert response.choices[0].message.content == "sensitive answer"
def test_redacts_vertex_provider_metadata_from_metadata_hidden_params(self):
"""Streaming success_handler copies _hidden_params into metadata before redaction."""