fix(proxy): keep a callback_settings reference apart from siblings of its credential family

A sibling entry of another backend but the same credential family, such as an
inline langfuse entry with its own host next to a reference later defined as
langfuse_otel, would have received the definition's keys through the merged
credential map. The conflict rule now covers credential families at write time
and at request time. A failure-only reference is validated with its
definition's values and the entry's event type, so destination-only settings
are refused on it like on an inline entry. A definition with an explicit
non-string callback_type fails boot instead of being skipped. An ordinary
logging entry without a callback_vars object is refused at write time, since
the runtime rejects it and would skip its callback, and the reference gate
keeps a missing callback_vars distinct from an empty one

litellm/proxy/common_utils/callback_config_validation.py

@@ -31,7 +31,7 @@ _SECRET_REFERENCE_PREFIX: Final = "os.environ/"
 _CALLBACK_TYPES: Final[frozenset[str]] = frozenset({"success", "failure", "success_and_failure"})
 _STRING_KEYED_MAPPING: Final = TypeAdapter(Mapping[str, object])
 
-LoggingEntryKey: TypeAlias = tuple[str | None, str | None, tuple[tuple[str, str], ...]]
+LoggingEntryKey: TypeAlias = tuple[str | None, str | None, tuple[tuple[str, str], ...] | None]
 
 
 @dataclass(frozen=True, slots=True)
@@ -86,12 +86,37 @@ def callback_definitions_error(callback_settings: Mapping[str, object]) -> str |
     A typo in a field or secret name would otherwise surface only on the first request
     of a key that already depends on the definition. Errors name the field, never its value.
     """
+    malformed: Final = next(
+        (error for name, raw in callback_settings.items() if (error := _malformed_callback_type_error(name, raw))),
+        None,
+    )
+    if malformed is not None:
+        return malformed
     definitions: Final = (
         definition
         for name, raw in callback_settings.items()
         if (definition := _parsed_definition(name, raw)) is not None
     )
-    return next((error for definition in definitions if (error := _definition_error(definition)) is not None), None)
+    return next(
+        (
+            definition_error
+            for definition in definitions
+            if (definition_error := _definition_error(definition)) is not None
+        ),
+        None,
+    )
+
+
+def _malformed_callback_type_error(name: str, raw: object) -> str | None:
+    """An entry that names ``callback_type`` but not as a string would be skipped as a plain setting and never serve a key."""
+    if not isinstance(raw, Mapping):
+        return None
+    setting: Final = _STRING_KEYED_MAPPING.validate_python(raw)
+    if _CALLBACK_TYPE_FIELD not in setting or isinstance(setting[_CALLBACK_TYPE_FIELD], str):
+        return None
+    return (
+        f"callback_settings.{name}.callback_type must be the name of a callback, got {setting[_CALLBACK_TYPE_FIELD]!r}"
+    )
 
 
 def _definition_error(definition: CallbackDefinition) -> str | None:
@@ -135,41 +160,79 @@ def _definition_backend_error(definition: CallbackDefinition) -> str | None:
 
 
 def callback_reference_error(logging_entries: object) -> str | None:
-    """A ``callback_settings`` reference is a name only, and the only entry for its backend.
-
-    Every entry's vars merge into one map per backend at request time, so a value on the
-    reference or on a sibling naming the same backend could point the definition's
-    credentials somewhere else.
-    """
+    """A ``callback_settings`` reference is a name only, and no other entry on the key may share its backend or credentials."""
     entries: Final = stored_callback_entries(logging_entries)
-    backends: Final = tuple(None if name is None else callback_backend_name(name) for name, _ in entries)
-    return next((error for error in _reference_errors(entries, backends) if error is not None), None)
+    conflicts: Final = reference_conflicts(logging_entries)
+    return next((error for error in _reference_errors(entries, conflicts) if error is not None), None)
 
 
 def _reference_errors(
-    entries: Sequence[tuple[str | None, Mapping[str, str]]], backends: Sequence[str | None]
+    entries: Sequence[tuple[str | None, Mapping[str, str]]], conflicts: frozenset[str]
 ) -> Iterator[str | None]:
-    for (name, callback_vars), backend in zip(entries, backends, strict=True):
-        if name is None or callback_definition(name) is None:
+    for name, callback_vars in entries:
+        if name is None:
+            continue
+        if (definition := callback_definition(name)) is None:
             continue
         if callback_vars:
             yield f"callback_settings.{name} is referenced by name only, so the entry may not carry callback_vars"
-        elif backends.count(backend) > 1:
+        elif name in conflicts:
             yield (
-                f"callback_settings.{name} already provides the {backend} callback for this key, "
-                f"so no other {backend} entry may sit next to it"
+                f"callback_settings.{name} already provides the {definition.backend} callback and its credentials "
+                "for this key, so no other entry may name that backend or carry a value of the same credential family"
             )
 
 
+def reference_conflicts(logging_entries: object) -> frozenset[str]:
+    """The references on a key that share a backend or a credential family with another entry on it.
+
+    Every entry's vars merge into one map at request time and each callback reads its
+    family's credentials from that map, so a sibling with its own ``langfuse_host`` next
+    to a ``langfuse_otel`` reference would send the definition's secrets there.
+    """
+    entries: Final = stored_callback_entries(logging_entries)
+    scopes: Final = tuple(_entry_scope(name, callback_vars) for name, callback_vars in entries)
+    return frozenset(
+        name
+        for index, ((name, _), scope) in enumerate(zip(entries, scopes, strict=True))
+        if name is not None and callback_definition(name) is not None and _overlaps_another(scope, scopes, index)
+    )
+
+
+def _overlaps_another(scope: frozenset[str], scopes: Sequence[frozenset[str]], index: int) -> bool:
+    return any(scope & other for other_index, other in enumerate(scopes) if other_index != index)
+
+
+def _entry_scope(name: str | None, callback_vars: Mapping[str, str]) -> frozenset[str]:
+    """The backend an entry runs on plus every credential family its vars configure."""
+    if name is None:
+        return frozenset()
+    definition: Final = callback_definition(name)
+    variables: Final = callback_vars if definition is None else definition.callback_vars
+    backend: Final = name if definition is None else definition.backend
+    return frozenset({backend, *(family for var in variables if (family := _family_of(var)) is not None)})
+
+
 def logging_entry_keys(logging_entries: object) -> tuple[LoggingEntryKey, ...]:
-    """Every ``logging`` entry as a comparable value: name, event type and sorted vars, so no field changes unnoticed."""
+    """Every ``logging`` entry as a comparable value: name, event type and sorted vars, so no field changes unnoticed.
+
+    A missing or malformed ``callback_vars`` is kept distinct from an empty one, since the
+    runtime rejects the former and would skip the entry's callback.
+    """
     entries: Final = _object_sequence(logging_entries) or ()
     return tuple(
-        (_entry_callback_name(entry), _entry_callback_type(entry), tuple(sorted(_entry_callback_vars(entry).items())))
-        for entry in entries
+        (_entry_callback_name(entry), _entry_callback_type(entry), _entry_callback_vars_key(entry)) for entry in entries
     )
 
 
+def _entry_callback_vars_key(entry: object) -> tuple[tuple[str, str], ...] | None:
+    match entry:
+        case {"callback_vars": Mapping()}:
+            return tuple(sorted(_entry_callback_vars(entry).items()))
+        case _:
+            return None
+
+
 def callback_config_error(
     callback_name: str | None, callback_vars: Mapping[str, str] | None, callback_type: str | None = None
 ) -> str | None:
@@ -524,9 +587,16 @@ def _logging_entry_error(entry: object) -> str | None:
     if callback_type_error is not None:
         return callback_type_error
     callback_name: Final = typed_entry.get("callback_name")
-    callback_vars_raw: Final = typed_entry.get("callback_vars")
-    if not isinstance(callback_name, str) or not isinstance(callback_vars_raw, Mapping):
+    if not isinstance(callback_name, str):
         return None
+    definition: Final = callback_definition(callback_name)
+    callback_type_raw: Final = typed_entry.get("callback_type")
+    callback_type: Final = callback_type_raw if isinstance(callback_type_raw, str) else None
+    if definition is not None:
+        return callback_config_error(definition.backend, definition.callback_vars, callback_type)
+    callback_vars_raw: Final = typed_entry.get("callback_vars")
+    if not isinstance(callback_vars_raw, Mapping):
+        return f"logging entry {callback_name!r} needs a callback_vars object, pass {{}} when it has no values"
     callback_vars: Final[Mapping[object, object]] = cast(  # cast-ok: Mapping check preserves arbitrary key types
         Mapping[object, object], callback_vars_raw
     )
@@ -540,12 +610,7 @@ def _logging_entry_error(entry: object) -> str | None:
     )
     if destination_var_error is not None:
         return destination_var_error
-    callback_type_raw: Final = typed_entry.get("callback_type")
-    return callback_config_error(
-        callback_name,
-        _entry_callback_vars(typed_entry),
-        callback_type_raw if isinstance(callback_type_raw, str) else None,
-    )
+    return callback_config_error(callback_name, _entry_callback_vars(typed_entry), callback_type)
 
 
 def _arize_sampling_rate_error(callback_name: str | None, callback_vars: Mapping[str, str]) -> str | None:

litellm/proxy/litellm_pre_call_utils.py

@@ -77,7 +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 callback_backend_name, callback_definition
+from litellm.proxy.common_utils.callback_config_validation import callback_definition, reference_conflicts
 from litellm.proxy.common_utils.callback_utils import (
     decrypt_callback_vars,
     get_metadata_variable_name_from_kwargs,
@@ -1250,9 +1250,10 @@ _STRING_KEYED_MAPPING: Final = TypeAdapter(Mapping[str, object])
 def resolved_key_callbacks(logging_entries: object, *, source: str) -> tuple[AddTeamCallback, ...]:
     """A key's ``logging`` entries as the backends see them, each ``callback_settings`` reference swapped for its definition.
 
-    A reference is honoured only while no other entry on the key names its backend: every
-    entry's vars merge into one map per backend, so a sibling could point the definition's
-    credentials at its own host. The reference's own ``callback_vars`` are never read.
+    A reference is honoured only while no other entry on the key names its backend or
+    carries a value of the same credential family: every entry's vars merge into one map,
+    so a sibling could point the definition's credentials at its own host. The reference's
+    own ``callback_vars`` are never read.
     """
     if not isinstance(logging_entries, Sequence) or isinstance(logging_entries, (str, bytes)):
         return ()
@@ -1261,27 +1262,24 @@ def resolved_key_callbacks(logging_entries: object, *, source: str) -> tuple[Add
         for item in _OBJECT_LIST_ADAPTER.validate_python(logging_entries)
         if isinstance(item, Mapping)
     )
-    backends: Final = tuple(_entry_backend(entry) for entry in entries)
+    conflicts: Final = reference_conflicts(entries)
     return tuple(
         callback
-        for entry, backend in zip(entries, backends, strict=True)
-        if (callback := _resolved_key_callback(entry, source=source, shared_backend=backends.count(backend) > 1))
-        is not None
+        for entry in entries
+        if (callback := _resolved_key_callback(entry, source=source, conflicts=conflicts)) is not None
     )
 
 
-def _entry_backend(entry: Mapping[str, object]) -> str | None:
-    callback_name: Final = entry.get("callback_name")
-    return callback_backend_name(callback_name) if isinstance(callback_name, str) else None
-
-
-def _resolved_key_callback(entry: Mapping[str, object], *, source: str, shared_backend: bool) -> AddTeamCallback | None:
+def _resolved_key_callback(
+    entry: Mapping[str, object], *, source: str, conflicts: frozenset[str]
+) -> AddTeamCallback | None:
     definition: Final = callback_definition(entry.get("callback_name"))
     if definition is None:
         return get_validated_callback_metadata(item=dict(entry), source=source)
-    if shared_backend:
+    if definition.name in conflicts:
         verbose_proxy_logger.warning(
-            "Ignoring %s callback metadata naming callback_settings.%s: another entry on this key names the %s callback",
+            "Ignoring %s callback metadata naming callback_settings.%s: another entry on this key names the %s "
+            "callback or carries a value of its credential family",
             source,
             _sanitize_for_log(definition.name),
             _sanitize_for_log(definition.backend),

tests/unit/proxy/common_utils/test_callback_config_validation.py

@@ -14,6 +14,7 @@ from litellm.proxy.common_utils.callback_config_validation import (
     cross_entry_family_error,
     logging_entry_keys,
     logging_metadata_config_error,
+    reference_conflicts,
 )
 
 
@@ -687,3 +688,83 @@ def test_logging_entry_keys_keep_a_malformed_callback_type_distinct_from_a_missi
     assert logging_entry_keys([entry]) == logging_entry_keys([{**entry, "callback_type": None}])
     assert logging_entry_keys([entry]) != logging_entry_keys([{**entry, "callback_type": []}])
     assert logging_entry_keys([entry]) != logging_entry_keys([{**entry, "callback_type": "success"}])
+
+
+@pytest.fixture
+def langfuse_team_a(monkeypatch: pytest.MonkeyPatch) -> None:
+    monkeypatch.setattr(
+        litellm,
+        "callback_settings",
+        {
+            "langfuse_team_a": {
+                "callback_type": "langfuse_otel",
+                "langfuse_public_key": "pk-team-a",
+                "langfuse_secret_key": "sk-team-a",
+                "langfuse_host": "https://langfuse.team-a.example",
+            }
+        },
+    )
+
+
+@pytest.mark.usefixtures("langfuse_team_a")
+def test_a_reference_may_not_sit_next_to_an_entry_of_the_same_credential_family() -> None:
+    """The classic langfuse callback reads langfuse_* credentials from the merged map, so a sibling with its own
+    host next to a langfuse_otel reference would receive the definition's keys."""
+    reference: Final = {"callback_name": "langfuse_team_a", "callback_type": "success"}
+    sibling: Final = {
+        "callback_name": "langfuse",
+        "callback_type": "success",
+        "callback_vars": {"langfuse_host": "https://attacker.example"},
+    }
+    other_family: Final = {
+        "callback_name": "datadog",
+        "callback_type": "success",
+        "callback_vars": {"dd_api_key": "dd"},
+    }
+    error: Final = logging_metadata_config_error({"logging": [reference, sibling]})
+    assert error is not None and "langfuse_team_a" in error
+    assert logging_metadata_config_error({"logging": [sibling, reference]}) is not None
+    assert logging_metadata_config_error({"logging": [reference, other_family]}) is None
+    assert reference_conflicts([reference, sibling]) == frozenset({"langfuse_team_a"})
+    assert reference_conflicts([reference, other_family]) == frozenset()
+
+
+@pytest.mark.usefixtures("otel_v2_on")
+def test_a_failure_only_reference_is_held_to_its_definition_settings(monkeypatch: pytest.MonkeyPatch) -> None:
+    monkeypatch.setattr(
+        litellm,
+        "callback_settings",
+        {
+            "arize_space_a": {
+                "callback_type": "arize",
+                "arize_api_key": "k",
+                "arize_space_id": "s",
+                "excluded_services": "postgres,redis",
+            }
+        },
+    )
+    failure_only: Final = {"callback_name": "arize_space_a", "callback_type": "failure"}
+    error: Final = logging_metadata_config_error({"logging": [failure_only]})
+    assert error is not None and "excluded_services" in error and "callback_type" in error
+    assert logging_metadata_config_error({"logging": [{**failure_only, "callback_type": "success"}]}) is None
+
+
+@pytest.mark.parametrize("callback_type", [None, 5, ["arize"]])
+def test_a_definition_whose_callback_type_is_not_a_name_fails_config_load(callback_type: object) -> None:
+    error: Final = callback_definitions_error({"space_a": {"callback_type": callback_type, "arize_api_key": "k"}})
+    assert error is not None and "callback_settings.space_a.callback_type" in error
+
+
+def test_an_ordinary_logging_entry_without_callback_vars_is_rejected_at_write_time() -> None:
+    """Stored as sent, the entry fails model validation at request time and its logger silently never runs."""
+    error: Final = logging_metadata_config_error(
+        {"logging": [{"callback_name": "datadog", "callback_type": "success"}]}
+    )
+    assert error is not None and "datadog" in error and "callback_vars" in error
+    assert logging_metadata_config_error({"logging": [{"callback_name": "datadog", "callback_vars": {}}]}) is None
+
+
+def test_logging_entry_keys_keep_missing_callback_vars_distinct_from_empty_ones() -> None:
+    entry: Final = {"callback_name": "datadog", "callback_type": "success"}
+    assert logging_entry_keys([entry]) != logging_entry_keys([{**entry, "callback_vars": {}}])
+    assert logging_entry_keys([{**entry, "callback_vars": "x"}]) != logging_entry_keys([{**entry, "callback_vars": {}}])

tests/unit/proxy/management_endpoints/test_key_management_endpoints.py

@@ -24305,13 +24305,23 @@ class TestCallbackReferenceWriteGate:
             {"logging": [{**_REFERENCE_ENTRY, "callback_name": "arize"}]},
             {"logging": [_REFERENCE_ENTRY, _LANGFUSE_ENTRY]},
             {"logging": [{**_LANGFUSE_ENTRY, "callback_vars": {"langfuse_host": "https://attacker.example"}}]},
+            {"logging": [_REFERENCE_ENTRY, {"callback_name": "datadog", "callback_type": "success"}]},
         ],
     )
     def test_a_non_admin_cannot_change_the_logging_of_a_key_holding_a_reference(
         self, metadata: Mapping[str, object]
     ) -> None:
         """The stored entry is compared with and without an explicit event type, since an admin may have omitted it."""
-        for stored in ({"logging": [_REFERENCE_ENTRY]}, {"logging": [{"callback_name": "arize_space_a"}]}):
+        for stored in (
+            {"logging": [_REFERENCE_ENTRY]},
+            {"logging": [{"callback_name": "arize_space_a"}]},
+            {
+                "logging": [
+                    _REFERENCE_ENTRY,
+                    {"callback_name": "datadog", "callback_type": "success", "callback_vars": {}},
+                ]
+            },
+        ):
             with pytest.raises(HTTPException) as exc:
                 raise_on_unauthorized_callback_reference(metadata, _non_admin(), existing_metadata=stored)
             assert exc.value.status_code == 403

tests/unit/proxy/test_litellm_pre_call_utils.py

@@ -11551,3 +11551,37 @@ def test_a_reference_resolves_to_the_same_otel_destination_as_the_inline_entry()
     assert len(resolved) == 1
     assert resolved[0].headers["api_key"] == "resolved-space-a-key"
     assert resolve_tenant_otel_destinations(plain_key_on_the_same_team) == ()
+
+
+def test_a_reference_is_skipped_when_a_sibling_carries_a_value_of_its_credential_family(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    """A key stored with an inline langfuse host before the name was defined as langfuse_otel must not hand the
+    definition's keys to that host through the merged credential map."""
+    monkeypatch.setattr(
+        litellm,
+        "callback_settings",
+        {
+            "langfuse_team_a": {
+                "callback_type": "langfuse_otel",
+                "langfuse_public_key": "pk-team-a",
+                "langfuse_secret_key": "sk-team-a",
+            }
+        },
+    )
+    sibling: Final = {
+        "callback_name": "langfuse",
+        "callback_type": "success",
+        "callback_vars": {"langfuse_host": "https://attacker.example"},
+    }
+    auth: Final = UserAPIKeyAuth(
+        api_key="hashed-key",
+        metadata={"logging": [{"callback_name": "langfuse_team_a", "callback_type": "success"}, sibling]},
+        team_metadata={},
+    )
+
+    result: Final = get_dynamic_logging_metadata(user_api_key_dict=auth, proxy_config=MagicMock())
+
+    assert result is not None
+    assert result.success_callback == ["langfuse"]
+    assert result.callback_vars == sibling["callback_vars"]