fix(cache): fail closed for incomplete auth identities

This commit is contained in:
Maximilian Rink 2026-10-10 17:17:56 +02:00
parent 207a58e51a
commit db5bf1ca86
4 changed files with 307 additions and 67 deletions

View file

@ -410,7 +410,7 @@ class Cache:
)
return "".join(f"{field}: {value}" for field, value in scope_values if value is not None)
def get_cache_key(self, **kwargs: object) -> str:
def get_cache_key(self, **kwargs: object) -> str | None:
"""
Get the cache key for the given arguments.
@ -421,11 +421,18 @@ class Cache:
str: The cache key generated from the arguments, or None if no cache key could be generated.
"""
cache_key = ""
authenticated_caller, authenticated_namespace = self._get_authenticated_cache_namespace(**kwargs)
if authenticated_caller and authenticated_namespace is None:
# An authenticated request without a stable server-side identity must
# never fall back to a shared or client-selected cache key.
return None
# verbose_logger.debug("\nGetting Cache key. Kwargs: %s", kwargs)
preset_cache_key: Final = self.get_preset_cache_key_from_kwargs(**kwargs)
if preset_cache_key is not None:
verbose_logger.debug("\nReturning preset cache key: %s", preset_cache_key)
if authenticated_caller:
return self._add_namespace_to_cache_key(preset_cache_key, **kwargs)
return preset_cache_key
combined_kwargs: Final = ModelParamHelper.get_all_llm_api_params()
@ -459,6 +466,8 @@ class Cache:
)
hashed_cache_key = Cache._get_hashed_cache_key(cache_key + reasoning_summary_material)
hashed_cache_key = self._add_namespace_to_cache_key(hashed_cache_key, **kwargs)
if hashed_cache_key is None:
return None
verbose_logger.debug(
"\nCreated cache key: %s (source material length: %d)",
hashed_cache_key,
@ -587,7 +596,7 @@ class Cache:
self,
hash_hex: str,
**kwargs: object, # kwargs-ok: cache request metadata is extensible
) -> str:
) -> str | None:
"""
If a redis namespace is provided, add it to the cache key
@ -622,29 +631,9 @@ class Cache:
)
if isinstance(source, Mapping)
)
authenticated_namespace: str | None = None # rebind-ok: populated once while scanning trusted auth metadata
try:
from litellm.proxy._types import UserAPIKeyAuth
except ImportError:
class UserAPIKeyAuth: # pragma: no cover - proxy types are available in proxy deployments
pass
for metadata in metadata_sources:
auth_object: object | None = metadata.get("user_api_key_auth")
if isinstance(auth_object, UserAPIKeyAuth):
identity_fields = {
"api_key_hash": getattr(auth_object, "token", None),
"team_id": getattr(auth_object, "team_id", None),
"project_id": getattr(auth_object, "project_id", None),
"org_id": getattr(auth_object, "org_id", None),
"user_id": getattr(auth_object, "user_id", None),
"end_user_id": getattr(auth_object, "end_user_id", None),
"user_role": getattr(auth_object, "user_role", None),
}
identity = json.dumps(identity_fields, sort_keys=True, default=str, separators=(",", ":"))
authenticated_namespace = "caller:" + hashlib.sha256(identity.encode("utf-8")).hexdigest()[:32]
break
authenticated_caller, authenticated_namespace = self._get_authenticated_cache_namespace(**kwargs)
if authenticated_caller and authenticated_namespace is None:
return None
requested_namespace_value: Final = dynamic_cache_control.get("namespace")
requested_namespace: Final[str | None] = (
requested_namespace_value if isinstance(requested_namespace_value, str) else None
@ -662,10 +651,71 @@ class Cache:
else requested_namespace or metadata_namespace or self.namespace
)
if namespace:
hash_hex = f"{namespace}:{hash_hex}" # rebind-ok: namespace prefix is the final cache key representation
prefix = f"{namespace}:"
if not hash_hex.startswith(prefix):
hash_hex = f"{prefix}{hash_hex}" # rebind-ok: namespace prefix is the final cache key representation
verbose_logger.debug("Final hashed key: %s", hash_hex)
return hash_hex
def _get_authenticated_cache_namespace(self, **kwargs: object) -> tuple[bool, str | None]:
"""Return whether trusted auth metadata exists and its stable cache namespace.
``UserAPIKeyAuth.token`` is the proxy's server-side hashed token identity.
API-key-only objects without it cannot be safely separated, so callers
must bypass cache reads and writes rather than use a shared fallback.
"""
litellm_params_value: Final = kwargs.get("litellm_params")
litellm_params: Final[Mapping[str, object]] = (
cast(Mapping[str, object], litellm_params_value) if isinstance(litellm_params_value, Mapping) else {}
)
metadata_sources: Final[tuple[Mapping[str, object], ...]] = tuple(
cast(Mapping[str, object], source)
for source in (
kwargs.get("metadata"),
kwargs.get("litellm_metadata"),
litellm_params.get("metadata"),
litellm_params.get("litellm_metadata"),
)
if isinstance(source, Mapping)
)
try:
from litellm.proxy._types import UserAPIKeyAuth
except ImportError:
class UserAPIKeyAuth: # pragma: no cover - proxy types are available in proxy deployments
pass
for metadata in metadata_sources:
auth_object: object | None = metadata.get("user_api_key_auth")
if not isinstance(auth_object, UserAPIKeyAuth):
continue
token = getattr(auth_object, "token", None)
if not isinstance(token, str) or not token:
return True, None
identity_fields = {
"api_key_hash": token,
"team_id": getattr(auth_object, "team_id", None),
"project_id": getattr(auth_object, "project_id", None),
"org_id": getattr(auth_object, "org_id", None),
"user_id": getattr(auth_object, "user_id", None),
"end_user_id": getattr(auth_object, "end_user_id", None),
"user_role": getattr(auth_object, "user_role", None),
}
identity = json.dumps(identity_fields, sort_keys=True, default=str, separators=(",", ":"))
return True, "caller:" + hashlib.sha256(identity.encode("utf-8")).hexdigest()[:32]
return False, None
def _prepare_explicit_cache_key(self, explicit_key: object, **kwargs: object) -> str | None:
"""Validate and scope a caller-provided cache key before backend access."""
if not isinstance(explicit_key, str):
return None
authenticated_caller, authenticated_namespace = self._get_authenticated_cache_namespace(**kwargs)
if not authenticated_caller:
return explicit_key
if authenticated_namespace is None:
return None
return self._add_namespace_to_cache_key(explicit_key, **kwargs)
def generate_streaming_content(self, content):
chunk_size: Final = 5 # Adjust the chunk size as needed
for i in range(0, len(content), chunk_size):
@ -759,7 +809,7 @@ class Cache:
return
with response_cache_phase("get"):
if "cache_key" in kwargs:
cache_key = kwargs["cache_key"]
cache_key = self._prepare_explicit_cache_key(kwargs["cache_key"], **kwargs)
else:
cache_key = self.get_cache_key(**kwargs)
if cache_key is not None:
@ -792,7 +842,7 @@ class Cache:
with response_cache_phase("get"):
if "cache_key" in kwargs:
cache_key = kwargs["cache_key"]
cache_key = self._prepare_explicit_cache_key(kwargs["cache_key"], **kwargs)
else:
cache_key = self.get_cache_key(**kwargs)
if cache_key is not None:
@ -813,31 +863,27 @@ class Cache:
"""
Common implementation across sync + async add_cache functions
"""
try:
if "cache_key" in kwargs:
cache_key = kwargs["cache_key"]
else:
cache_key = self.get_cache_key(**kwargs)
if cache_key is not None:
if isinstance(result, BaseModel):
result = result.model_dump_json(exclude_unset=True)
if "cache_key" in kwargs:
cache_key = self._prepare_explicit_cache_key(kwargs["cache_key"], **kwargs)
else:
cache_key = self.get_cache_key(**kwargs)
if cache_key is not None:
if isinstance(result, BaseModel):
result = result.model_dump_json(exclude_unset=True)
## DEFAULT TTL ##
if self.ttl is not None:
kwargs["ttl"] = self.ttl
## Get Cache-Controls ##
_cache_kwargs: Final = kwargs.get("cache", None)
if isinstance(_cache_kwargs, dict):
for k, v in _cache_kwargs.items():
if k == "ttl":
kwargs["ttl"] = v
## DEFAULT TTL ##
if self.ttl is not None:
kwargs["ttl"] = self.ttl
## Get Cache-Controls ##
_cache_kwargs: Final = kwargs.get("cache", None)
if isinstance(_cache_kwargs, dict):
for k, v in _cache_kwargs.items():
if k == "ttl":
kwargs["ttl"] = v
cached_data: Final = {"timestamp": time.time(), "response": result}
return cache_key, cached_data, kwargs
else:
raise Exception("cache key is None")
except Exception as e:
raise e
cached_data: Final = {"timestamp": time.time(), "response": result}
return cache_key, cached_data, kwargs
return None
def add_cache(self, result, **kwargs):
"""
@ -854,7 +900,10 @@ class Cache:
if self.should_use_cache(**kwargs) is not True:
return
with response_cache_phase("set"):
cache_key, cached_data, kwargs = self._add_cache_logic(result=result, **kwargs)
cache_entry = self._add_cache_logic(result=result, **kwargs)
if cache_entry is None:
return
cache_key, cached_data, kwargs = cache_entry
self.cache.set_cache(cache_key, cached_data, **kwargs)
except Exception as e:
self._log_add_cache_failure(e)
@ -878,7 +927,10 @@ class Cache:
# high traffic - fill in results in memory and then flush
await self.batch_cache_write(result, **kwargs)
else:
cache_key, cached_data, kwargs = self._add_cache_logic(result=result, **kwargs)
cache_entry = self._add_cache_logic(result=result, **kwargs)
if cache_entry is None:
return
cache_key, cached_data, kwargs = cache_entry
if dynamic_cache_object is not None:
await dynamic_cache_object.async_set_cache(cache_key, cached_data, **kwargs)
else:
@ -989,7 +1041,7 @@ class Cache:
input: str,
kwargs: dict,
idx_in_result_data: int = 0,
) -> tuple[str, dict, dict]:
) -> tuple[str, dict, dict] | None:
preset_cache_key: Final = self.get_cache_key(**{**kwargs, "input": input})
kwargs["cache_key"] = preset_cache_key
embedding_response: Final = result.data[idx_in_result_data]
@ -1013,11 +1065,10 @@ class Cache:
prompt_tokens_details=prompt_tokens_details,
)
cache_key, cached_data, kwargs = self._add_cache_logic(
return self._add_cache_logic(
result=embedding_dict,
**kwargs,
)
return cache_key, cached_data, kwargs
async def async_add_cache_pipeline(self, result, dynamic_cache_object: BaseCache | None = None, **kwargs):
"""
@ -1046,16 +1097,16 @@ class Cache:
cache_list: Final = []
if isinstance(kwargs["input"], list):
for idx, i in enumerate(kwargs["input"]):
(
cache_key,
cached_data,
kwargs,
) = self.add_embedding_response_to_cache(result, i, kwargs, idx)
cache_entry = self.add_embedding_response_to_cache(result, i, kwargs, idx)
if cache_entry is None:
return
cache_key, cached_data, kwargs = cache_entry
cache_list.append((cache_key, cached_data))
elif isinstance(kwargs["input"], str):
cache_key, cached_data, kwargs = self.add_embedding_response_to_cache(
result, kwargs["input"], kwargs
)
cache_entry = self.add_embedding_response_to_cache(result, kwargs["input"], kwargs)
if cache_entry is None:
return
cache_key, cached_data, kwargs = cache_entry
cache_list.append((cache_key, cached_data))
if dynamic_cache_object is not None:
@ -1087,7 +1138,10 @@ class Cache:
return False
async def batch_cache_write(self, result, **kwargs):
cache_key, cached_data, kwargs = self._add_cache_logic(result=result, **kwargs)
cache_entry = self._add_cache_logic(result=result, **kwargs)
if cache_entry is None:
return
cache_key, cached_data, kwargs = cache_entry
await self.cache.batch_cache_write(cache_key, cached_data, **kwargs)
async def ping(self):

View file

@ -93,10 +93,12 @@ class _PROXY_BatchRedisRequests(CustomLogger):
- return redis cache request
"""
try: # never block execution
if litellm.cache is None:
return None
cache_key: str | None = None
if "cache_key" in kwargs:
cache_key = kwargs["cache_key"]
elif litellm.cache is not None:
cache_key = litellm.cache._prepare_explicit_cache_key(kwargs["cache_key"], **kwargs)
else:
cache_key = litellm.cache.get_cache_key(
*args, **kwargs
) # returns "<cache_key_name>:<hash>" - we pass redis_namespace in async_pre_call_hook. Done to avoid rewriting the async_set_cache logic

View file

@ -0,0 +1,52 @@
import time
from typing import Final
import pytest
import litellm
from litellm.caching.caching import Cache
from litellm.caching.in_memory_cache import InMemoryCache
from litellm.proxy._types import UserAPIKeyAuth
from litellm.proxy.hooks.batch_redis_get import PROXY_BatchRedisRequests
from litellm.types.caching import LiteLLMCacheType
class _BackendSpy:
def __init__(self) -> None:
self.keys: list[str] = []
async def async_get_cache(self, key: str, *args, **kwargs):
self.keys.append(key)
return {"timestamp": time.time(), "response": {"key": key}}
def _caller(token: str | None) -> UserAPIKeyAuth:
return UserAPIKeyAuth(api_key=f"raw-{token}").model_copy(update={"token": token})
@pytest.mark.asyncio
async def test_batch_redis_hook_scopes_explicit_keys_and_fails_closed(monkeypatch: pytest.MonkeyPatch) -> None:
cache: Final = Cache(type=LiteLLMCacheType.LOCAL)
backend = _BackendSpy()
cache.cache = backend
monkeypatch.setattr(litellm, "cache", cache)
hook = PROXY_BatchRedisRequests()
hook.in_memory_cache = InMemoryCache()
request_a = {"metadata": {"user_api_key_auth": _caller("hashed-a")}, "cache_key": "shared"}
request_b = {"metadata": {"user_api_key_auth": _caller("hashed-b")}, "cache_key": "shared"}
first_a = await hook.async_get_cache(**request_a)
second_a = await hook.async_get_cache(**request_a)
first_b = await hook.async_get_cache(**request_b)
missing = await hook.async_get_cache(
metadata={"user_api_key_auth": _caller(None)},
cache_key="shared",
)
assert first_a == second_a
assert first_a != first_b
assert missing is None
assert len(backend.keys) == 2
assert backend.keys[0] != backend.keys[1]
assert all(key.startswith("caller:") for key in backend.keys)

View file

@ -3,6 +3,7 @@ import hashlib
import json
import logging
import re
import time
import traceback
import uuid
from datetime import timedelta
@ -53,6 +54,137 @@ def test_authenticated_proxy_callers_get_distinct_exact_cache_keys() -> None:
assert key_a != key_b
def test_authenticated_callers_without_server_token_fail_closed_for_all_key_forms() -> None:
cache: Final = Cache(type=LiteLLMCacheType.LOCAL, namespace="qa")
caller_a: Final = UserAPIKeyAuth(api_key="raw-a", team_id="team-a", user_id="user-a").model_copy(
update={"token": None}
)
caller_b: Final = UserAPIKeyAuth(api_key="raw-b", team_id="team-a", user_id="user-a").model_copy(
update={"token": None}
)
for caller in (caller_a, caller_b):
metadata = {"user_api_key_auth": caller, "redis_namespace": "client-selected"}
assert cache.get_cache_key(model="gpt-4.1-mini", messages=messages, metadata=metadata) is None
assert (
cache.get_cache_key(
model="gpt-4.1-mini",
messages=messages,
metadata=metadata,
litellm_params={"preset_cache_key": "shared"},
)
is None
)
def test_explicit_cache_keys_are_idempotently_scoped_to_valid_callers() -> None:
cache: Final = Cache(type=LiteLLMCacheType.LOCAL, namespace="qa")
caller_a: Final = UserAPIKeyAuth(api_key="raw-a", token="hashed-a", team_id="team-a")
caller_b: Final = UserAPIKeyAuth(api_key="raw-b", token="hashed-b", team_id="team-a")
metadata_a = {"user_api_key_auth": caller_a, "redis_namespace": "client-selected"}
metadata_b = {"user_api_key_auth": caller_b, "redis_namespace": "client-selected"}
key_a = cache.get_cache_key(model="gpt-4.1-mini", messages=messages, metadata=metadata_a)
assert key_a is not None
explicit_a = cache._prepare_explicit_cache_key("shared", metadata=metadata_a)
assert explicit_a is not None
assert explicit_a != cache._prepare_explicit_cache_key("shared", metadata=metadata_b)
assert cache._prepare_explicit_cache_key(explicit_a, metadata=metadata_a) == explicit_a
class _CacheBackendSpy:
def __init__(self) -> None:
self.get_calls = 0
self.set_calls = 0
def get_cache(self, *args, **kwargs):
self.get_calls += 1
return {"timestamp": time.time(), "response": {"value": "cached"}}
def set_cache(self, *args, **kwargs):
self.set_calls += 1
async def async_get_cache(self, *args, **kwargs):
self.get_calls += 1
return {"timestamp": time.time(), "response": {"value": "cached"}}
async def async_set_cache(self, *args, **kwargs):
self.set_calls += 1
def _missing_token_cache_request() -> dict[str, object]:
return {
"model": "gpt-4.1-mini",
"messages": messages,
"metadata": {
"user_api_key_auth": UserAPIKeyAuth(api_key="raw-a", team_id="team-a").model_copy(
update={"token": None}
),
},
}
def test_missing_token_skips_sync_backend_access_for_explicit_key(caplog) -> None:
cache: Final = Cache(type=LiteLLMCacheType.LOCAL)
backend = _CacheBackendSpy()
cache.cache = backend
request = _missing_token_cache_request()
assert cache.get_cache(**request, cache_key="shared") is None
cache.add_cache("response", **request, cache_key="shared")
assert backend.get_calls == 0
assert backend.set_calls == 0
assert "exception in add_cache" not in caplog.text
@pytest.mark.asyncio
async def test_missing_token_skips_async_backend_access_for_explicit_key() -> None:
cache: Final = Cache(type=LiteLLMCacheType.LOCAL)
backend = _CacheBackendSpy()
cache.cache = backend
request = _missing_token_cache_request()
assert await cache.async_get_cache(**request, cache_key="shared") is None
await cache.async_add_cache("response", **request, cache_key="shared")
assert backend.get_calls == 0
assert backend.set_calls == 0
def test_valid_token_explicit_key_reaches_sync_backend() -> None:
cache: Final = Cache(type=LiteLLMCacheType.LOCAL)
backend = _CacheBackendSpy()
cache.cache = backend
request = {
"model": "gpt-4.1-mini",
"messages": messages,
"metadata": {"user_api_key_auth": UserAPIKeyAuth(api_key="raw-a", token="hashed-a")},
}
cache.add_cache("response", **request, cache_key="shared")
assert cache.get_cache(**request, cache_key="shared") == {"value": "cached"}
assert backend.set_calls == 1
assert backend.get_calls == 1
@pytest.mark.asyncio
async def test_valid_token_explicit_key_reaches_async_backend() -> None:
cache: Final = Cache(type=LiteLLMCacheType.LOCAL)
backend = _CacheBackendSpy()
cache.cache = backend
request = {
"model": "gpt-4.1-mini",
"messages": messages,
"metadata": {"user_api_key_auth": UserAPIKeyAuth(api_key="raw-a", token="hashed-a")},
}
await cache.async_add_cache("response", **request, cache_key="shared")
assert await cache.async_get_cache(**request, cache_key="shared") == {"value": "cached"}
assert backend.set_calls == 1
assert backend.get_calls == 1
@pytest.fixture
def preserve_litellm_set_verbose(monkeypatch: pytest.MonkeyPatch) -> None:
monkeypatch.setattr(litellm, "set_verbose", litellm.set_verbose)