Merge 326966b7ac into add7dea9aa

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()

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:

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"}}