diff --git a/litellm/_service_logger.py b/litellm/_service_logger.py index 060b4fb878e..4461b5bc428 100644 --- a/litellm/_service_logger.py +++ b/litellm/_service_logger.py @@ -340,6 +340,21 @@ class ServiceLogging(CustomLogger): elif isinstance(error, str): error_message = error + if service == ServiceTypes.REDIS: + # redis-py annotates a failed pipeline command with the command it + # sent, key and value included, so the raw message carries cached + # prompt/response text into Datadog and OTel. Redacted here rather + # than at each call site: `redis_cache` hands this hook an exception + # from ~16 places and a new one must not be able to leak by being + # written without the redaction. + from litellm.caching.redis_cache import ( + redact_redis_error, # noqa: PLC0415 # redis_cache imports this module; a module-scope import would be circular + ) + + error_message = ( # rebind-ok: narrows the message derived above, like the three assignments that produced it + redact_redis_error(error) if isinstance(error, Exception) else error_message + ) + payload: Final = ServiceLoggerPayload( is_error=True, error=error_message, @@ -374,8 +389,11 @@ class ServiceLogging(CustomLogger): else: _otel_logger_to_use = self._resolve_otel_service_logger(callback) - if not isinstance(error, str): - error = str(error) + # `error_message`, not `str(error)`: re-deriving it here would + # undo the redaction above. Prometheus still gets the raw + # exception, which it labels by class and never reads the + # message of. + error = error_message # See the success hook: no parent gate, so background failures # are traced too. V1 no-ops without a parent; V2 emits a root. diff --git a/litellm/caching/redis_cache.py b/litellm/caching/redis_cache.py index afe6f8a1eab..59a5131f7db 100644 --- a/litellm/caching/redis_cache.py +++ b/litellm/caching/redis_cache.py @@ -16,6 +16,7 @@ import inspect import itertools import json import logging +import re import threading import time import weakref @@ -506,28 +507,72 @@ class _RedisTimeoutLogThrottle: _redis_timeout_log_throttle: Final = _RedisTimeoutLogThrottle(REDIS_TIMEOUT_LOG_INTERVAL) +# redis-py annotates every failed pipeline command with the command it sent, +# arguments included: ``Command # 1 (SET EX 60) of pipeline caused +# error: `` (``Pipeline.annotate_exception`` in redis/client.py, +# redis/asyncio/client.py and redis/cluster.py). ``uv.lock`` pins a redis-py +# whose ``annotate_exception`` does not truncate that text at all, so +# ``str(exc)`` carries the whole cache key and the serialized prompt/response +# into every sink this exception reaches. +# +# ``command`` is matched greedily so the split happens at the LAST +# " of pipeline caused error: ", which is the one redis-py wrote. A cache value +# containing that literal text therefore cannot shift the boundary and smuggle +# itself into the part that is kept. +_PIPELINE_COMMAND_ANNOTATION: Final = re.compile( + r"^Command # (?P\d+) \((?P.*)\) of pipeline caused error: (?P.*)$", + re.DOTALL, +) + +_REDACTED: Final = "" + + +def redact_redis_error(exc: BaseException) -> str: + """``str(exc)`` with any redis-py pipeline command arguments removed. + + Only the command name is kept (``SET``, ``INCRBY``, ...) because that is the + part with diagnostic value; the key and value that follow it are not. The + trailing detail is the server's own error string ("WRONGTYPE ...", "OOM + ..."), which redis never echoes argument data into, so it is preserved. + + Messages that are not pipeline annotations are returned unchanged: redis-py + builds those from the server reply alone. + """ + message: Final = str(exc) + match: Final = _PIPELINE_COMMAND_ANNOTATION.match(message) + if match is None: + return message + command_name: Final = match["command"].split(" ", 1)[0].strip() or _REDACTED + return f"Command # {match['index']} ({command_name} {_REDACTED}) of pipeline caused error: {match['detail']}" + + def log_redis_failure( logger: logging.Logger, level: int, message: str, exc: BaseException, with_traceback: bool = False ) -> None: + # Every Redis failure line in this module goes through here, so redacting + # once covers all of them -- and any call site added later. + reason: Final = redact_redis_error(exc) if isinstance(exc, RedisCircuitBreakerOpenError): - logger.debug("%s: %s", message, exc, stacklevel=2) + logger.debug("%s: %s", message, reason, stacklevel=2) return - exc_info: Final = exc if with_traceback else None + exc_info: Final = ( + (Exception, Exception(f"{type(exc).__name__}: {reason}"), exc.__traceback__) if with_traceback else None + ) if not is_redis_timeout_failure(exc): - logger.log(level, "%s: %s", message, exc, exc_info=exc_info, stacklevel=2) + logger.log(level, "%s: %s", message, reason, exc_info=exc_info, stacklevel=2) return suppressed: Final = _redis_timeout_log_throttle.admit() if suppressed is None: - logger.debug("%s: %s", message, exc, stacklevel=2) + logger.debug("%s: %s", message, reason, stacklevel=2) return if suppressed == 0: - logger.log(level, "%s: %s", message, exc, exc_info=exc_info, stacklevel=2) + logger.log(level, "%s: %s", message, reason, exc_info=exc_info, stacklevel=2) return logger.log( level, "%s: %s (%d more Redis timeouts since the previous Redis timeout line were logged at DEBUG)", message, - exc, + reason, suppressed, exc_info=exc_info, stacklevel=2, diff --git a/tests/unit/caching/test_redis_cache.py b/tests/unit/caching/test_redis_cache.py index 2bdc7b9549b..19813199487 100644 --- a/tests/unit/caching/test_redis_cache.py +++ b/tests/unit/caching/test_redis_cache.py @@ -10,6 +10,7 @@ from unittest.mock import AsyncMock, MagicMock, patch import pytest import litellm +from litellm._logging import verbose_logger from litellm._service_logger import ServiceLogging from litellm.caching.redis_cache import RedisCache, RedisCircuitBreakerOpenError @@ -1033,8 +1034,9 @@ def test_prometheus_services_logger_starts_from_the_breakers_already_built(monke monkeypatch.setattr(litellm, "service_callback", ["prometheus_system"]) gc.collect() ServiceLogging() - assert REGISTRY.get_sample_value("litellm_redis_circuit_breaker_state", {"state": "closed"}) == ( - redis_breaker_state_counts()["closed"] + assert ( + REGISTRY.get_sample_value("litellm_redis_circuit_breaker_state", {"state": "closed"}) + == (redis_breaker_state_counts()["closed"]) ) @@ -1675,6 +1677,143 @@ def test_call_stack_info_skips_native_execution_frames(): assert anthropic_messages() == "anthropic_messages <- test_call_stack_info_skips_native_execution_frames" +SENSITIVE_CACHE_KEY = "litellm:cache:customer-secret-cache-key" +SENSITIVE_CACHE_VALUE = '{"messages":[{"role":"user","content":"private customer message"}]}' + + +def _annotated_pipeline_error(key=SENSITIVE_CACHE_KEY, value=SENSITIVE_CACHE_VALUE): + """The exception redis-py raises for a failed pipeline command. + + Built with redis-py's own ``annotate_exception`` rather than a hand-written + string, so this stays honest about the message shape the library produces. + """ + import redis.asyncio as async_redis + from redis.exceptions import ResponseError + + pipeline = async_redis.Redis().pipeline(transaction=False) + error = ResponseError("WRONGTYPE Operation against a key holding the wrong kind of value") + pipeline.annotate_exception(error, 1, ("SET", key, value, "EX", "60")) + return error + + +def test_the_annotation_really_does_carry_the_key_and_value(): + """Precondition: on the pinned redis-py there is something to leak. + + How much `annotate_exception` includes is a redis-py detail: the version + `uv.lock` pins does not truncate the command at all, while later releases + cap it. The redaction is worth having either way, so this asserts the + premise only on a redis-py that actually exposes the value, and says so + rather than silently passing on one that does not. + """ + raw = str(_annotated_pipeline_error()) + if SENSITIVE_CACHE_KEY not in raw: + import redis as _redis + + pytest.skip(f"redis-py {_redis.__version__} truncates the pipeline annotation") + assert "private customer message" in raw + + +def test_redact_redis_error_drops_the_command_arguments(): + from litellm.caching.redis_cache import redact_redis_error + + redacted = redact_redis_error(_annotated_pipeline_error()) + + assert SENSITIVE_CACHE_KEY not in redacted + assert "private customer message" not in redacted + # The parts worth keeping survive: which command, and the server's reason. + assert "SET" in redacted + assert "WRONGTYPE" in redacted + + +def test_redact_redis_error_leaves_a_plain_message_alone(): + from litellm.caching.redis_cache import redact_redis_error + + assert redact_redis_error(ValueError("OOM command not allowed")) == "OOM command not allowed" + + +def test_a_value_containing_the_annotation_text_cannot_shift_the_boundary(): + """The split is on the LAST separator, the one redis-py wrote.""" + from litellm.caching.redis_cache import redact_redis_error + + hostile = "x of pipeline caused error: " + SENSITIVE_CACHE_VALUE + redacted = redact_redis_error(_annotated_pipeline_error(value=hostile)) + + assert "private customer message" not in redacted + + +def test_log_redis_failure_redacts_the_line_it_writes(caplog): + import logging as _logging + + from litellm.caching.redis_cache import log_redis_failure + + with caplog.at_level(_logging.ERROR, logger="LiteLLM"): + log_redis_failure( + verbose_logger, + _logging.ERROR, + "LiteLLM Redis Caching: async set_cache_pipeline() - Got exception from REDIS", + _annotated_pipeline_error(), + ) + + assert SENSITIVE_CACHE_KEY not in caplog.text + assert "private customer message" not in caplog.text + assert "WRONGTYPE" in caplog.text + + +@pytest.mark.parametrize("chained", [False, True]) +def test_log_redis_failure_redacts_traceback_and_exception_chain( + caplog: pytest.LogCaptureFixture, chained: bool +) -> None: + import logging + + from redis.exceptions import ResponseError + + from litellm.caching.redis_cache import log_redis_failure + + with caplog.at_level(logging.ERROR, logger="LiteLLM"): + try: + raise _annotated_pipeline_error() + except ResponseError as pipeline_error: + if chained: + try: + raise RuntimeError("Redis operation failed") from pipeline_error + except RuntimeError as wrapper: + log_redis_failure(verbose_logger, logging.ERROR, "Redis failed", wrapper, with_traceback=True) + else: + log_redis_failure(verbose_logger, logging.ERROR, "Redis failed", pipeline_error, with_traceback=True) + + assert SENSITIVE_CACHE_KEY not in caplog.text + assert "private customer message" not in caplog.text + assert "Traceback (most recent call last)" in caplog.text + assert ("RuntimeError" if chained else "ResponseError") in caplog.text + assert "Redis failed" in caplog.text + + +@pytest.mark.asyncio +async def test_service_failure_hook_redacts_before_telemetry(): + """Datadog/OTel must not receive the annotated command either.""" + from litellm._service_logger import ServiceLogging + from litellm.types.services import ServiceTypes + + dd_logger = MagicMock() + dd_logger.async_service_failure_hook = AsyncMock() + service_logging = ServiceLogging() + service_logging.dd_logger = dd_logger + + with patch.object(litellm, "service_callback", ["datadog"]): + await service_logging.async_service_failure_hook( + service=ServiceTypes.REDIS, + duration=0.1, + error=_annotated_pipeline_error(), + call_type="async_set_cache_pipeline", + ) + + kwargs = dd_logger.async_service_failure_hook.call_args.kwargs + assert SENSITIVE_CACHE_KEY not in kwargs["error"] + assert "private customer message" not in kwargs["error"] + assert SENSITIVE_CACHE_KEY not in kwargs["payload"].error + assert "private customer message" not in kwargs["payload"].error + + class _FormatCountingStr(str): format_calls = 0