Fix plaintext JWTs leaking in debug logs (#22424)

* Fix plaintext JWTs leaking in debug logs

Wrap raw request headers in RedactedDict (dict subclass with redacted
str/repr) at the single entry point where they enter the system. This
prevents any downstream logging path from exposing Bearer tokens.

Also remove a redundant log that re-read request.headers directly,
bypassing the already-cleaned _headers variable.

* Add e2e test for JWT redaction in debug logs

* Preserve RedactedDict type through copy()
This commit is contained in:
ryan-crabbe
2026-02-28 15:44:38 -08:00
committed by GitHub
parent 7a33d51edf
commit 07def978cf
3 changed files with 87 additions and 5 deletions
+8 -5
View File
@@ -839,9 +839,14 @@ async def add_litellm_data_to_request( # noqa: PLR0915
"""
from litellm.proxy.proxy_server import llm_router, premium_user
from litellm.types.proxy.litellm_pre_call_utils import SecretFields
from litellm.types.proxy.litellm_pre_call_utils import (
RedactedDict,
SecretFields,
)
_raw_headers: Dict[str, str] = _safe_get_request_headers(request)
_raw_headers: Dict[str, str] = RedactedDict(
_safe_get_request_headers(request)
)
forward_llm_auth = False
if general_settings:
@@ -938,9 +943,7 @@ async def add_litellm_data_to_request( # noqa: PLR0915
add_provider_specific_headers_to_request(data=data, headers=_headers)
## Cache Controls
headers = request.headers
verbose_proxy_logger.debug("Request Headers: %s", headers)
cache_control_header = headers.get("Cache-Control", None)
cache_control_header = _headers.get("Cache-Control", None)
if cache_control_header:
cache_dict = parse_cache_control(cache_control_header)
data["ttl"] = cache_dict.get("s-maxage")
@@ -1,6 +1,19 @@
from typing_extensions import TypedDict
class RedactedDict(dict):
"""Dict subclass with redacted str/repr to prevent leaking in logs."""
def __repr__(self) -> str:
return "RedactedDict(REDACTED)"
def __str__(self) -> str:
return "RedactedDict(REDACTED)"
def copy(self) -> "RedactedDict":
return RedactedDict(super().copy())
class SecretFields(TypedDict):
"""
Stored in data["secret_fields"]
@@ -1713,3 +1713,69 @@ async def test_add_guardrails_from_policy_engine_policy_version_by_id():
# Clean up
policy_registry._policies = {}
policy_registry._initialized = False
@pytest.mark.asyncio
async def test_bearer_token_not_in_debug_logs():
"""
E2E regression test for the client-reported JWT leak.
Calls add_litellm_data_to_request with a Bearer token in the request
headers and captures all debug log output. Asserts the raw token never
appears in any log message — covering the exact paths the client reported:
- "Request Headers: ..."
- "receiving data: ..."
- "[PROXY] returned data from litellm_pre_call_utils: ..."
"""
import logging
from io import StringIO
from litellm.proxy.litellm_pre_call_utils import add_litellm_data_to_request
from litellm.proxy.proxy_server import ProxyConfig
secret_token = "eyJhbGciOiJSUzI1NiIsInR5cCI6IkpXVCJ9.eyJzdWIiOiIxMjM0NTY3ODkwIn0.fakesignature"
mock_request = MagicMock(spec=Request)
mock_request.headers = {
"authorization": f"Bearer {secret_token}",
"content-type": "application/json",
}
mock_request.url = MagicMock()
mock_request.url.__str__ = lambda self: "http://localhost:4000/v1/chat/completions"
mock_request.method = "POST"
mock_request.query_params = {}
data = {
"model": "gpt-4",
"messages": [{"role": "user", "content": "hi"}],
}
user_api_key_dict = UserAPIKeyAuth(api_key="sk-1234")
# Capture all debug log output from the proxy logger
log_capture = StringIO()
log_handler = logging.StreamHandler(log_capture)
log_handler.setLevel(logging.DEBUG)
logger = logging.getLogger("LiteLLM Proxy")
logger.addHandler(log_handler)
original_level = logger.level
logger.setLevel(logging.DEBUG)
try:
with patch("litellm.proxy.proxy_server.llm_router", None), \
patch("litellm.proxy.proxy_server.premium_user", True):
await add_litellm_data_to_request(
data=data,
request=mock_request,
user_api_key_dict=user_api_key_dict,
proxy_config=ProxyConfig(),
general_settings={},
)
finally:
logger.removeHandler(log_handler)
logger.setLevel(original_level)
log_output = log_capture.getvalue()
assert secret_token not in log_output, (
f"Bearer token leaked in debug logs. "
f"Found token in log output:\n{log_output[:500]}"
)