Merge c4fa1c5180 into acd084ab38

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.

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 <key> <value> EX 60) of pipeline caused
+# error: <server 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<index>\d+) \((?P<command>.*)\) of pipeline caused error: (?P<detail>.*)$",
+    re.DOTALL,
+)
+
+_REDACTED: Final = "<redacted>"
+
+
+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,

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