diff --git a/litellm/constants.py b/litellm/constants.py index 97631b9997a..d9d112c3be9 100644 --- a/litellm/constants.py +++ b/litellm/constants.py @@ -2392,3 +2392,7 @@ HARNESS_SNAPSHOT_SKIP_DIRS: Final = frozenset( ) DEFAULT_TOOL_LOOP_MAX_ROUNDS: Final = 20 + +# Sentinel for "this lookup path did not resolve", distinct from any caller-supplied +# default, which may itself be None. +NESTED_VALUE_MISSING_SENTINEL: Final = object() diff --git a/litellm/litellm_core_utils/dot_notation_indexing.py b/litellm/litellm_core_utils/dot_notation_indexing.py index 818cf1b9695..f11e92fc357 100644 --- a/litellm/litellm_core_utils/dot_notation_indexing.py +++ b/litellm/litellm_core_utils/dot_notation_indexing.py @@ -26,6 +26,8 @@ additional_drop_params to remove nested fields from optional parameters. from collections.abc import Mapping from typing import Any, Final, TypeVar +from litellm.constants import NESTED_VALUE_MISSING_SENTINEL + T = TypeVar("T") @@ -54,29 +56,40 @@ def get_nested_value(data: Mapping[str, object], key_path: str, default: T | Non if not key_path: return default - # Remove metadata. prefix if it exists - key_path = key_path.replace("metadata.", "", 1) if key_path.startswith("metadata.") else key_path - # Split the key path into parts, respecting escaped dots (\.) # Use a temporary placeholder, split on unescaped dots, then restore placeholder: Final = "\x00" - parts = key_path.replace("\\.", placeholder).split(".") - parts = [p.replace(placeholder, ".") for p in parts] - # Traverse through the dictionary - current: Any = data - for part in parts: - try: - current = current[part] - except (KeyError, TypeError): - return default + def _resolve(path: str) -> object: + parts: Final = path.replace("\\.", placeholder).split(".") + current: Any = data + for part in parts: + part = part.replace(placeholder, ".") + try: + current = current[part] + except (KeyError, TypeError): + return NESTED_VALUE_MISSING_SENTINEL + return current + + # A leading "metadata." has always been stripped before resolving, so + # `roles_jwt_field: "metadata.roles"` reads a top-level `roles` claim. Keep + # that spelling authoritative so no token changes team, and fall back to the + # path as written only when the legacy one is absent, which is what makes a + # genuinely nested `metadata` object readable instead of silently falling back + # to the defaults JWT auth applies when a claim is missing. + without_prefix: Final = key_path[len("metadata.") :] if key_path.startswith("metadata.") else "" + legacy_resolved: Final = _resolve(without_prefix) if without_prefix else NESTED_VALUE_MISSING_SENTINEL + resolved: Final[Any] = _resolve(key_path) if legacy_resolved is NESTED_VALUE_MISSING_SENTINEL else legacy_resolved + + if resolved is NESTED_VALUE_MISSING_SENTINEL: + return default # If default is None, we can return any type if default is None: - return current + return resolved # Otherwise, ensure the type matches the default - return current if isinstance(current, type(default)) else default + return resolved if isinstance(resolved, type(default)) else default def _parse_path_segments(path: str) -> list: diff --git a/tests/unit/litellm_core_utils/test_dot_notation_indexing.py b/tests/unit/litellm_core_utils/test_dot_notation_indexing.py index 7c670f1dac9..faa67bb4dd5 100644 --- a/tests/unit/litellm_core_utils/test_dot_notation_indexing.py +++ b/tests/unit/litellm_core_utils/test_dot_notation_indexing.py @@ -26,6 +26,60 @@ class TestGetNestedValue: data = {"a": {"b": {"c": "value"}}} assert get_nested_value(data, "a.b.c") == "value" + def test_nested_metadata_object_is_reachable(self): + """A token that really nests claims under `metadata` resolves. + + The prefix used to be stripped before traversal, so `metadata.roles` was + looked up as `roles` and a genuinely nested claim read as missing. JWT auth + then fell back to its defaults, which for get_team_id means `team_id_default` + rather than the team the token named. + """ + assert get_nested_value({"metadata": {"roles": ["admin"]}}, "metadata.roles") == ["admin"] + + def test_top_level_claim_keeps_precedence_over_nested_metadata(self): + """A token carrying both shapes must resolve to the same team it always did. + + Stripping the prefix before traversal meant the top-level claim won, so + that stays authoritative. Reading the nested value first would silently + move an existing deployment onto a different team. + """ + data = {"metadata": {"roles": ["nested"]}, "roles": ["top-level"]} + assert get_nested_value(data, "metadata.roles") == ["top-level"] + + def test_metadata_prefix_still_falls_back_to_top_level(self): + """The legacy spelling keeps working when there is no nested object.""" + assert get_nested_value({"roles": ["admin"]}, "metadata.roles") == ["admin"] + + def test_metadata_prefix_falls_back_to_a_default(self): + """With neither shape present the caller's default comes back unchanged.""" + assert get_nested_value({"other": 1}, "metadata.roles", "fallback") == "fallback" + + def test_metadata_prefix_applies_only_once(self): + """Only a leading `metadata.` is treated as a prefix. + + `metadata.metadata.x` strips one segment, so a key literally named + `metadata` inside a `metadata` object is still reachable. + """ + data = {"metadata": {"metadata": {"team_id": "team-1"}}} + assert get_nested_value(data, "metadata.metadata.team_id") == "team-1" + + def test_nested_metadata_respects_the_default_type(self): + """The caller's default still governs type coercion on the nested read.""" + data = {"metadata": {"team_id": "team-1"}} + assert get_nested_value(data, "metadata.team_id", "fallback") == "team-1" + assert get_nested_value(data, "metadata.team_id", 1234) == 1234, ( + "a default of another type must make the nested value unusable here, " + "exactly as for a non-prefixed path" + ) + + def test_bare_metadata_key_is_untouched(self): + """A top-level key named `metadata` resolves as written.""" + assert get_nested_value({"metadata": "value"}, "metadata") == "value" + + def test_non_dict_metadata_is_not_read_as_a_nested_claim(self): + """A string `metadata` claim cannot hold a nested object, so the read falls through.""" + assert get_nested_value({"metadata": "not-a-dict"}, "metadata.roles", "fallback") == "fallback" + def test_missing_key_returns_default(self): """Test that missing keys return the default value.""" data = {"a": {"b": "value"}}