fix(proxy): require an unforgeable marker to trust litellm_metadata

resolve_success_event_metadata_variable_name treated any non-empty
litellm_metadata as authoritative, but a caller can populate it with
unrelated content (e.g. {"marker": true}) on a route where metadata is
the field the proxy actually wrote authenticated tags/identity into,
causing both rate-limit hooks to find no tag and admit past the
configured limit.

add_user_api_key_auth_to_request_metadata unconditionally stamps a
user_api_key_auth marker into whichever bucket it resolves as
authoritative, overwriting anything a caller pre-populated there.
Requiring that marker's presence instead of mere truthiness can't be
forged onto the wrong side.

Veria AI finding, surfaced on PR #38347 but the vulnerable function is
this PR's own; ported the same fix pattern already hardened on #38292.
This commit is contained in:
Deepanshu 2026-08-27 20:22:53 -04:00
parent 9813335651
commit 41be4a8782
2 changed files with 49 additions and 5 deletions

View file

@ -213,13 +213,19 @@ def resolve_success_event_metadata_variable_name(
real, populated `metadata` dict for a standard (non
LITELLM_METADATA_ROUTES) request, so the key-presence check always picks
`litellm_metadata` there and silently reads no tags/identity at all.
Requiring the value to actually be a populated dict, matching
`_get_request_tags`'s own truthiness check in litellm_logging.py, only
A plain truthiness check on `litellm_metadata` isn't enough either: a
caller can populate it with unrelated, non-empty content on a route where
`metadata` is the field the proxy actually wrote identity into, and
truthiness alone would still misresolve to the caller-controlled bucket.
`add_user_api_key_auth_to_request_metadata` (litellm_pre_call_utils.py)
unconditionally stamps a `user_api_key_auth` key into whichever bucket it
resolved as authoritative, overwriting anything a caller pre-populated
there -- so requiring that marker's presence, not mere truthiness, only
ever prefers `litellm_metadata` when it is genuinely the field the proxy
wrote identity/tags into (LITELLM_METADATA_ROUTES pre-seed it before
admission runs, so it is always a populated dict by success time there)."""
wrote identity/tags into."""
litellm_metadata: Final = litellm_params_for_metadata.get("litellm_metadata")
if isinstance(litellm_metadata, Mapping) and litellm_metadata:
if isinstance(litellm_metadata, Mapping) and "user_api_key_auth" in litellm_metadata:
return "litellm_metadata"
return "metadata"

View file

@ -16,6 +16,7 @@ from litellm.proxy.hooks.tag_rate_limits_shared import (
fixed_length_identity,
order_tags_for_identity_resolution,
partition_key,
resolve_success_event_metadata_variable_name,
)
from litellm.types.router import TagRateLimitEntry, TagRateLimitScope
@ -124,6 +125,43 @@ def test_extract_key_hash_reads_metadata_when_it_is_the_authoritative_field():
assert extract_key_hash(request_kwargs, "metadata") == "real-hash"
# ---------------------------------------------------------------------------
# resolve_success_event_metadata_variable_name -- veria-ai finding surfaced on
# PR #38347: a caller-supplied, non-empty litellm_metadata must not be
# selected over the metadata the proxy actually wrote authenticated tags into
# ---------------------------------------------------------------------------
def test_resolve_success_event_metadata_variable_name_ignores_caller_forged_non_empty_litellm_metadata():
"""A caller-supplied litellm_metadata with unrelated content (no
user_api_key_auth marker) must not be picked over metadata, even though
it is a non-empty dict."""
litellm_params_for_metadata = {"litellm_metadata": {"marker": True}}
assert resolve_success_event_metadata_variable_name(litellm_params_for_metadata) == "metadata"
def test_resolve_success_event_metadata_variable_name_selects_litellm_metadata_when_server_written():
"""A genuinely server-populated litellm_metadata (LITELLM_METADATA_ROUTES)
always carries the user_api_key_auth marker stamped by
add_user_api_key_auth_to_request_metadata."""
litellm_params_for_metadata = {
"litellm_metadata": {"user_api_key_auth": object(), "tags": ["team_id:t1"]}
}
assert resolve_success_event_metadata_variable_name(litellm_params_for_metadata) == "litellm_metadata"
def test_resolve_success_event_metadata_variable_name_defaults_to_metadata_when_litellm_metadata_absent():
assert resolve_success_event_metadata_variable_name({}) == "metadata"
def test_resolve_success_event_metadata_variable_name_defaults_to_metadata_when_litellm_metadata_none():
assert resolve_success_event_metadata_variable_name({"litellm_metadata": None}) == "metadata"
def test_resolve_success_event_metadata_variable_name_defaults_to_metadata_when_litellm_metadata_empty():
assert resolve_success_event_metadata_variable_name({"litellm_metadata": {}}) == "metadata"
# ---------------------------------------------------------------------------
# entry_applies -- enabled_for / disabled_for / apply_to_key_alias / apply_to_models
# ---------------------------------------------------------------------------