From 82c4a74774ecfd40adeebf62a08baf75d685abfa Mon Sep 17 00:00:00 2001 From: Avani-prajapati Date: Wed, 20 May 2026 14:39:38 +0530 Subject: [PATCH] fix(logging): add CredentialScrubberFilter to strip secrets from all log output (CWE-312) 117 CodeQL findings show secrets logged in plain text across the codebase, including the encryption key itself. Rather than patching 117 call sites, attach a CredentialScrubberFilter directly to verbose_logger, verbose_proxy_logger, and verbose_router_logger. The filter uses a compiled key-name regex to redact any key=value / key:value pattern whose key matches known secret names (api_key, encryption_key, redis_password, litellm_key, etc.), complementing the existing value-pattern SecretRedactionFilter. --- litellm/_logging.py | 43 +++++ .../test_litellm/test_credential_scrubber.py | 182 ++++++++++++++++++ 2 files changed, 225 insertions(+) create mode 100644 tests/test_litellm/test_credential_scrubber.py diff --git a/litellm/_logging.py b/litellm/_logging.py index 6b99f50e014..4241f3e724b 100644 --- a/litellm/_logging.py +++ b/litellm/_logging.py @@ -1,6 +1,7 @@ import ast import logging import os +import re import sys from datetime import datetime from logging import Formatter @@ -429,3 +430,45 @@ def _is_debugging_on() -> bool: Returns True if debugging is on """ return verbose_logger.isEnabledFor(logging.DEBUG) or set_verbose is True + + +_SECRET_KEY_RE = re.compile( + r"(?i)" + r"(api[_\-]?key|secret[_\-]?key|access[_\-]?key|auth[_\-]?token|" + r"password|passwd|token|bearer|credential|private[_\-]?key|" + r"encryption[_\-]?key|master[_\-]?key|redis[_\-]?password|" + r"client[_\-]?secret|aws[_\-]?secret|gcp[_\-]?key|litellm[_\-]?key)" + r'(["\']?\s*[:=]\s*["\']?)([^\s\'"&,}\]]{6,})' +) +_REDACTED = "[REDACTED]" + + +def _scrub_secrets(text: str) -> str: + """Replace secret values in log text with [REDACTED].""" + return _SECRET_KEY_RE.sub(lambda m: m.group(1) + m.group(2) + _REDACTED, text) + + +class CredentialScrubberFilter(logging.Filter): + """Logging filter that redacts credential values from all log records.""" + + def filter(self, record: logging.LogRecord) -> bool: + if record.msg and isinstance(record.msg, str): + record.msg = _scrub_secrets(record.msg) + if record.args: + if isinstance(record.args, dict): + record.args = { + k: _scrub_secrets(str(v)) if isinstance(v, str) else v + for k, v in record.args.items() + } + elif isinstance(record.args, tuple): + record.args = tuple( + _scrub_secrets(str(a)) if isinstance(a, str) else a + for a in record.args + ) + return True + + +_credential_scrubber = CredentialScrubberFilter() +verbose_logger.addFilter(_credential_scrubber) +verbose_proxy_logger.addFilter(_credential_scrubber) +verbose_router_logger.addFilter(_credential_scrubber) diff --git a/tests/test_litellm/test_credential_scrubber.py b/tests/test_litellm/test_credential_scrubber.py new file mode 100644 index 00000000000..1b1d0659397 --- /dev/null +++ b/tests/test_litellm/test_credential_scrubber.py @@ -0,0 +1,182 @@ +import logging + +import pytest + + +class TestScrubSecrets: + """Tests for the _scrub_secrets() helper.""" + + def test_api_key_value_redacted(self): + from litellm._logging import _scrub_secrets + + result = _scrub_secrets("api_key=sk-abcdef123456789") + assert "sk-abcdef" not in result + assert "[REDACTED]" in result + + def test_aws_secret_key_redacted(self): + from litellm._logging import _scrub_secrets + + result = _scrub_secrets("aws_secret_key: AKIAIOSFODNN7EXAMPLE") + assert "AKIAIOSFODNN7EXAMPLE" not in result + assert "[REDACTED]" in result + + def test_encryption_key_redacted(self): + from litellm._logging import _scrub_secrets + + result = _scrub_secrets("encryption_key = my-very-secret-key-123") + assert "my-very-secret-key-123" not in result + assert "[REDACTED]" in result + + def test_bearer_token_redacted(self): + # CredentialScrubberFilter matches key=value / key:value format. + # "Authorization: Bearer " (space-separated) is handled by the + # existing SecretRedactionFilter; here we test the key:value log pattern. + from litellm._logging import _scrub_secrets + + result = _scrub_secrets("auth_token: eyJhbGciOiJSUzI1NiJ9abcdef") + assert "eyJhbGciOiJSUzI1NiJ9abcdef" not in result + assert "[REDACTED]" in result + + def test_non_secret_field_not_redacted(self): + from litellm._logging import _scrub_secrets + + result = _scrub_secrets("model=gpt-4o, endpoint=https://api.openai.com") + assert "gpt-4o" in result + assert "api.openai.com" in result + + def test_short_value_not_redacted(self): + # Values < 6 chars are not secrets (avoids redacting booleans/short flags) + from litellm._logging import _scrub_secrets + + result = _scrub_secrets("api_key=abc") + assert "abc" in result + + +class TestCredentialScrubberFilter: + """Tests for CredentialScrubberFilter.filter() — covers every branch.""" + + def _make_record(self, msg, args=None): + # Always construct with args=() then set record.args directly. + # Passing a dict to LogRecord.__init__ as args triggers a KeyError on + # Python 3.13 (args[0] lookup on a str-keyed dict). + record = logging.LogRecord( + name="test", + level=logging.DEBUG, + pathname="", + lineno=0, + msg=msg, + args=(), + exc_info=None, + ) + if args is not None: + record.args = args # type: ignore[assignment] + return record + + def test_string_msg_with_secret_is_redacted(self): + from litellm._logging import CredentialScrubberFilter + + f = CredentialScrubberFilter() + record = self._make_record("api_key=sk-secret123456789") + f.filter(record) + assert "sk-secret123456789" not in record.msg + assert "[REDACTED]" in record.msg + + def test_non_string_msg_untouched(self): + # Branch: msg exists but is not a str — must not crash + from litellm._logging import CredentialScrubberFilter + + f = CredentialScrubberFilter() + record = self._make_record(12345) + f.filter(record) + assert record.msg == 12345 + + def test_none_msg_untouched(self): + # Branch: record.msg is falsy + from litellm._logging import CredentialScrubberFilter + + f = CredentialScrubberFilter() + record = self._make_record(None) + f.filter(record) + assert record.msg is None + + def test_dict_args_secret_redacted(self): + # Branch: record.args is a dict whose string value contains a key=value secret + from litellm._logging import CredentialScrubberFilter + + f = CredentialScrubberFilter() + record = self._make_record("config %s", {"info": "api_key=sk-secret123456789"}) + f.filter(record) + assert "sk-secret123456789" not in str(record.args) + + def test_dict_args_non_string_value_untouched(self): + # Branch: record.args is a dict with a non-string value — must pass through + from litellm._logging import CredentialScrubberFilter + + f = CredentialScrubberFilter() + record = self._make_record("config %s", {"count": 42}) + f.filter(record) + assert record.args == {"count": 42} # type: ignore[comparison-overlap] + + def test_tuple_args_secret_redacted(self): + # Branch: record.args is a tuple whose string element contains key=value secret + from litellm._logging import CredentialScrubberFilter + + f = CredentialScrubberFilter() + record = self._make_record("config: %s", ("api_key=sk-secret123456789",)) + f.filter(record) + assert "sk-secret123456789" not in str(record.args) + + def test_tuple_args_mixed_types_non_string_passthrough(self): + # Branch: tuple contains non-string elements — ints/None pass through untouched + from litellm._logging import CredentialScrubberFilter + + f = CredentialScrubberFilter() + record = self._make_record( + "vals=%s %s %s", (42, None, "api_key=sk-secret123456789") + ) + f.filter(record) + args = record.args + assert isinstance(args, tuple) + assert args[0] == 42 + assert args[1] is None + assert "sk-secret123456789" not in str(args[2]) + + def test_no_args_no_crash(self): + # Branch: record.args is falsy (empty tuple) + from litellm._logging import CredentialScrubberFilter + + f = CredentialScrubberFilter() + record = self._make_record("plain message with no args") + f.filter(record) + assert record.msg == "plain message with no args" + + def test_filter_always_returns_true(self): + # Filter must never drop log records + from litellm._logging import CredentialScrubberFilter + + f = CredentialScrubberFilter() + record = self._make_record("api_key=sk-secret123456789") + result = f.filter(record) + assert result is True + + def test_filter_registered_on_verbose_logger(self): + import litellm._logging as log_module + + filter_types = [type(f).__name__ for f in log_module.verbose_logger.filters] + assert "CredentialScrubberFilter" in filter_types + + def test_filter_registered_on_verbose_proxy_logger(self): + import litellm._logging as log_module + + filter_types = [ + type(f).__name__ for f in log_module.verbose_proxy_logger.filters + ] + assert "CredentialScrubberFilter" in filter_types + + def test_filter_registered_on_verbose_router_logger(self): + import litellm._logging as log_module + + filter_types = [ + type(f).__name__ for f in log_module.verbose_router_logger.filters + ] + assert "CredentialScrubberFilter" in filter_types