fix(proxy): keep a callback_settings reference the only entry for its backend on a key

litellm/proxy/common_utils/callback_config_validation.py

@@ -361,6 +361,9 @@ def logging_metadata_config_error(metadata: Mapping[str, object] | None) -> str
     )
     if entry_error is not None:
         return entry_error
+    collision: Final = reference_backend_collision_error(entries)
+    if collision is not None:
+        return collision
     entry_vars: Final = tuple(_entry_callback_vars(entry) for entry in entries)
     named_vars: Final = tuple(_backend_entry(entry) for entry in entries)
     return next(
@@ -376,6 +379,46 @@ def logging_metadata_config_error(metadata: Mapping[str, object] | None) -> str
     )
 
 
+def reference_backend_collision_error(logging_entries: object) -> str | None:
+    """A reference is the only entry its backend may have, since one backend's entries merge into one credential map.
+
+    Otherwise a sibling entry naming the same backend could point the definition's
+    credentials at its own host.
+    """
+    names: Final = tuple(name for name, _ in stored_callback_entries(logging_entries) if name is not None)
+    backends: Final = tuple(callback_backend_name(name) for name in names)
+    return next(
+        (
+            f"callback_settings.{name} already provides the {backend} callback for this key, "
+            f"so no other {backend} entry may sit next to it"
+            for name, backend in zip(names, backends, strict=True)
+            if named_callback_definition(name) is not None and backends.count(backend) > 1
+        ),
+        None,
+    )
+
+
+def drop_entries_shadowing_a_reference(logging_entries: object) -> tuple[Mapping[str, object], ...]:
+    """Keep a reference as the only entry for its backend, so a stored sibling cannot redirect the definition's credentials."""
+    entries: Final = tuple(
+        _STRING_KEYED_MAPPING.validate_python(entry)
+        for entry in (_object_sequence(logging_entries) or ())
+        if isinstance(entry, Mapping)
+    )
+    referenced: Final = frozenset(
+        callback_backend_name(name)
+        for name, _ in stored_callback_entries(entries)
+        if name is not None and named_callback_definition(name) is not None
+    )
+    return tuple(
+        entry
+        for entry in entries
+        if (name := _entry_callback_name(entry)) is None
+        or named_callback_definition(name) is not None
+        or callback_backend_name(name) not in referenced
+    )
+
+
 def _backend_entry(entry: object) -> tuple[str | None, Mapping[str, str]]:
     """An entry as the backend sees it: a reference carries its definition's content settings."""
     callback_name: Final = _entry_callback_name(entry)

litellm/proxy/litellm_pre_call_utils.py

@@ -77,6 +77,7 @@ from litellm.proxy._types import (
 )
 from litellm.proxy.auth.auth_utils import get_request_route
 from litellm.proxy.auth.route_checks import RouteChecks
+from litellm.proxy.common_utils.callback_config_validation import drop_entries_shadowing_a_reference
 from litellm.proxy.common_utils.callback_utils import (
     decrypt_callback_vars,
     get_metadata_variable_name_from_kwargs,
@@ -1294,7 +1295,7 @@ def get_dynamic_logging_metadata(
     # Key-based callbacks
     #########################################################################################
     if key_dynamic_logging_settings is not None:
-        for item in key_dynamic_logging_settings:
+        for item in _key_entries_for_resolution(key_dynamic_logging_settings, source="key-level"):
             callback = get_resolved_key_callback_metadata(item=item, source="key-level")
             if callback is None:
                 continue
@@ -1341,6 +1342,23 @@ def get_dynamic_logging_metadata(
     return callback_settings_obj
 
 
+def _key_entries_for_resolution(logging_entries: object, *, source: str) -> tuple[Mapping[str, object], ...]:
+    """A key's entries with those sharing a backend with a ``callback_settings`` reference dropped, so the definition's credentials only ever go where it says."""
+    kept: Final = drop_entries_shadowing_a_reference(logging_entries)
+    dropped: Final = len(_object_sequence(logging_entries) or ()) - len(kept)
+    if dropped:
+        verbose_proxy_logger.warning(
+            "Ignoring %d %s callback metadata entries that share a backend with a callback_settings reference",
+            dropped,
+            source,
+        )
+    return kept
+
+
+def _object_sequence(value: object) -> Sequence[object] | None:
+    return value if isinstance(value, Sequence) and not isinstance(value, (str, bytes)) else None
+
+
 def _callback_settings_fields(callback_settings_obj: TeamCallbackMetadata) -> Mapping[str, object]:
     callback_vars: Final = callback_settings_obj.callback_vars
     return {
@@ -1452,7 +1470,7 @@ def resolve_tenant_otel_destinations(
         return ()
     key_entries: Final = KeyAndTeamLoggingSettings.get_key_dynamic_logging_settings(user_api_key_dict)
     entries: Final = (
-        key_entries
+        _key_entries_for_resolution(key_entries, source="otel-destination")
         if key_entries is not None
         else KeyAndTeamLoggingSettings.get_team_dynamic_logging_settings(user_api_key_dict)
     )

litellm/proxy/management_endpoints/key_management_endpoints.py

@@ -973,7 +973,9 @@ def raise_on_unauthorized_callback_reference(
             raise HTTPException(status_code=400, detail={"error": overridden})
         return
     stored: Final = _admin_only_entry_keys(existing_metadata or {})
-    if _admin_only_entry_keys(metadata) == stored:
+    if _admin_only_entry_keys(metadata) == stored and (
+        not stored or _entry_keys(metadata) == _entry_keys(existing_metadata or {})
+    ):
         return
     unknown: Final = next(iter(_unknown_callback_names(metadata)), None)
     raise HTTPException(status_code=403, detail={"error": _callback_reference_denied(unknown, stored=bool(stored))})
@@ -981,7 +983,10 @@ def raise_on_unauthorized_callback_reference(
 
 def _callback_reference_denied(unknown_callback_name: str | None, *, stored: bool) -> str:
     if stored:
-        return "Only a proxy admin can change or remove a key's callback_settings reference"
+        return (
+            "Only a proxy admin can change or remove the logging entries of a key that holds a callback_settings "
+            "reference"
+        )
     if unknown_callback_name is None:
         return "Only a proxy admin can point a key at a callback_settings definition"
     return (
@@ -998,6 +1003,14 @@ def _admin_only_entry_keys(metadata: Mapping[str, object]) -> tuple[tuple[str, t
     )
 
 
+def _entry_keys(metadata: Mapping[str, object]) -> tuple[tuple[str | None, tuple[tuple[str, str], ...]], ...]:
+    """Every logging entry, since a sibling of a reference shares its backend's merged credential map."""
+    return tuple(
+        (callback_name, tuple(sorted(callback_vars.items())))
+        for callback_name, callback_vars in stored_callback_entries(metadata.get("logging"))
+    )
+
+
 def _unknown_callback_names(metadata: Mapping[str, object]) -> tuple[str, ...]:
     return tuple(
         callback_name

tests/unit/proxy/common_utils/test_callback_config_validation.py

@@ -548,7 +548,8 @@ class TestCallbackReferenceEntries:
 
         assert error is not None and "excluded_services" in error and "'langfuse'" in error
 
-    def test_an_inline_entry_cannot_contradict_the_content_setting_its_reference_inherits(self, otel_v2_on):
+    def test_an_inline_entry_cannot_change_the_content_setting_its_reference_inherits(self, otel_v2_on):
+        """The option belongs on the reference entry itself; a same-backend sibling is refused outright."""
         reference: Final = {"callback_name": "arize_space_a", "callback_type": "success", "callback_vars": {}}
         inline: Final = {
             "callback_name": "arize",
@@ -558,14 +559,28 @@ class TestCallbackReferenceEntries:
 
         error: Final = logging_metadata_config_error({"logging": [reference, inline]})
 
-        assert error is not None and "capture_message_content" in error
+        assert error is not None and "arize_space_a" in error
+        assert logging_metadata_config_error(self._entry("arize_space_a", capture_message_content="no_content")) is None
 
-    def test_an_inline_entry_may_repeat_the_content_setting_its_reference_inherits(self, otel_v2_on):
+    def test_an_inline_entry_cannot_share_the_backend_a_reference_provides(self, otel_v2_on):
+        """One backend's entries merge into one credential map, so an inline sibling could point the definition's secrets at its own host."""
         reference: Final = {"callback_name": "arize_space_a", "callback_type": "success", "callback_vars": {}}
         inline: Final = {
             "callback_name": "arize",
             "callback_type": "success",
-            "callback_vars": {"capture_message_content": "no_content"},
+            "callback_vars": {"arize_api_key": "mine", "arize_space_id": "attacker-space"},
+        }
+
+        error: Final = logging_metadata_config_error({"logging": [inline, reference]})
+
+        assert error is not None and "arize_space_a" in error and "arize" in error
+
+    def test_an_inline_entry_for_another_backend_may_sit_next_to_a_reference(self, otel_v2_on):
+        reference: Final = {"callback_name": "arize_space_a", "callback_type": "success", "callback_vars": {}}
+        inline: Final = {
+            "callback_name": "langfuse",
+            "callback_type": "success",
+            "callback_vars": {"langfuse_public_key": "pk", "langfuse_secret_key": "sk"},
         }
 
         assert logging_metadata_config_error({"logging": [reference, inline]}) is None

tests/unit/proxy/management_endpoints/test_callback_management_endpoints.py

@@ -657,3 +657,28 @@ class TestCallbackReferenceKeyUpdates:
 
     def test_a_proxy_admin_can_remove_the_reference(self):
         assert self._check({"note": "edited"}, "proxy_admin") is None
+
+    @pytest.mark.parametrize(
+        "sibling",
+        [
+            {
+                "callback_name": "arize",
+                "callback_type": "success",
+                "callback_vars": {"arize_api_key": "mine", "arize_space_id": "space-b"},
+            },
+            {
+                "callback_name": "langfuse",
+                "callback_type": "success",
+                "callback_vars": {"langfuse_public_key": "pk", "langfuse_secret_key": "sk"},
+            },
+        ],
+        ids=["same-backend", "other-backend"],
+    )
+    def test_a_non_admin_cannot_add_an_entry_next_to_the_stored_reference(self, sibling):
+        """Resending the reference unchanged and appending a sibling must not pass, since the sibling joins the reference's credential map."""
+        from fastapi import HTTPException
+
+        with pytest.raises(HTTPException) as exc:
+            self._check({**self._STORED, "logging": [*self._STORED["logging"], sibling]}, "internal_user")
+        assert exc.value.status_code == 403
+        assert "callback_settings reference" in str(exc.value.detail)

tests/unit/proxy/test_litellm_pre_call_utils.py

@@ -3017,6 +3017,34 @@ def test_key_callback_reference_keeps_the_values_the_key_adds(monkeypatch):
     assert result.callback_vars["arize_api_key"] == "first-key"
 
 
+def test_a_key_entry_sharing_the_backend_of_a_reference_cannot_redirect_its_credentials(monkeypatch):
+    """A stored inline arize entry next to an arize_space_a reference would otherwise overwrite the space the secrets go to."""
+    monkeypatch.setattr(litellm, "callback_settings", _ARIZE_REFERENCE_SETTINGS)
+    monkeypatch.setenv("LIT10061_ARIZE_API_KEY", "first-key")
+    monkeypatch.setenv("LIT10061_ARIZE_SPACE_ID", "space-a")
+    user_api_key_dict: Final = UserAPIKeyAuth(
+        api_key="test-key",
+        team_id="team-a",
+        metadata={
+            "logging": [
+                {"callback_name": "arize_space_a", "callback_type": "success", "callback_vars": {}},
+                {
+                    "callback_name": "arize",
+                    "callback_type": "success",
+                    "callback_vars": {"arize_api_key": "mine", "arize_space_id": "attacker-space"},
+                },
+            ]
+        },
+        team_metadata={},
+    )
+
+    result: Final = get_dynamic_logging_metadata(user_api_key_dict=user_api_key_dict, proxy_config=MagicMock())
+
+    assert result is not None
+    assert result.callback_vars["arize_space_id"] == "space-a"
+    assert result.callback_vars["arize_api_key"] == "first-key"
+
+
 def test_key_callback_reference_with_an_unreadable_secret_logs_nowhere(monkeypatch):
     monkeypatch.setattr(litellm, "callback_settings", _ARIZE_REFERENCE_SETTINGS)
     monkeypatch.delenv("LIT10061_ARIZE_API_KEY", raising=False)