fix(team): make one entry own a credential family end to end

Every stored entry's callback_vars are flattened into one dict before a
request reads them, and that dict is what the exporter authenticates and
addresses with. So an entry naming only a destination is enough to
redirect a credential written somewhere else: a host on a second entry
pairs with the key pair from the first, and the request carries that key
pair to the new host. A team admin cannot read the team's masked Langfuse
secret, but could add such an entry and receive it.

Reject, for writers who are not proxy admins, an entry using a credential
family another entry already holds. Family rather than callback name,
because langfuse and langfuse_otel configure one Langfuse project and
would otherwise redirect each other, and because a destination like
dd_agent_host that no integration registry lists still pairs with the
Datadog credentials beside it.

A proxy admin already holds every credential the proxy has, so the rule
would buy nothing there and would break configs that predate it. A team
admin who does want to move a family deletes the entry holding it first,
which reveals nothing.

litellm/proxy/common_utils/callback_config_validation.py

@@ -44,6 +44,67 @@ def _langfuse_environment_error(callback_vars: Mapping[str, str]) -> str | None:
     return None
 
 
+# Which credential family a dynamic variable belongs to. The families are the
+# integrations that share one account: every langfuse_* variable configures the
+# same Langfuse project whether it rides the classic callback or the OTel one,
+# and every dd_* variable configures the same Datadog account.
+_VAR_FAMILIES: Final[Mapping[str, str]] = MappingProxyType(
+    {
+        "arize_": "Arize",
+        "dd_": "Datadog",
+        "gcs_": "GCS",
+        "humanloop_": "Humanloop",
+        "langfuse_": "Langfuse",
+        "langsmith_": "LangSmith",
+        "newrelic_": "New Relic",
+        "posthog_": "PostHog",
+        "wandb_": "Weights & Biases",
+        "weave_": "Weights & Biases",
+    }
+)
+
+
+def _family_of(var: str) -> str | None:
+    """The credential family ``var`` configures, or ``None`` if it configures none.
+
+    ``turn_off_message_logging`` and friends belong to no backend, so they carry
+    no credentials anyone could redirect.
+    """
+    return next((family for prefix, family in _VAR_FAMILIES.items() if var.startswith(prefix)), None)
+
+
+def cross_entry_family_error(
+    callback_vars: Mapping[str, str] | None,
+    stored_vars_by_entry: Sequence[Mapping[str, str]],
+) -> str | None:
+    """Reject an entry that joins a credential family another entry already holds.
+
+    Every stored entry's variables are flattened into one dict before a request
+    reads them, and the flattened dict is what the exporter authenticates and
+    addresses with. So an entry naming only a destination is enough to redirect
+    credentials that were written somewhere else: a host on a second entry pairs
+    with the key from the first, and the request carries that key to the new
+    host.
+
+    Requiring one entry to own a family end to end removes the pairing. Only the
+    writers this endpoint newly admits are held to it, because a proxy admin
+    already holds every credential the proxy has. A team admin who does want to
+    move a family deletes the entry holding it first, which reveals nothing.
+    """
+    if not callback_vars:
+        return None
+    held: Final = {family for entry in stored_vars_by_entry for family in map(_family_of, entry) if family is not None}
+    return next(
+        (
+            f"{family} is already configured by another callback entry on this team. "
+            f"Remove that entry before setting {var} here."
+            for var, family in ((v, _family_of(v)) for v in callback_vars)
+            if family is not None and family in held
+        ),
+        None,
+    )
+
+
 def logging_metadata_config_error(metadata: Mapping[str, object] | None) -> str | None:
     """Validate every ``logging`` entry of a team/key metadata payload."""
     if not metadata:

litellm/proxy/management_endpoints/team_callback_endpoints.py

@@ -29,7 +29,10 @@ from litellm.proxy._types import (
     UserAPIKeyAuth,
 )
 from litellm.proxy.auth.user_api_key_auth import user_api_key_auth
-from litellm.proxy.common_utils.callback_config_validation import callback_config_error
+from litellm.proxy.common_utils.callback_config_validation import (
+    callback_config_error,
+    cross_entry_family_error,
+)
 from litellm.proxy.common_utils.callback_utils import (
     _CALLBACK_VAR_ENCRYPTED_PREFIX,
     decrypt_callback_vars,
@@ -340,6 +343,23 @@ async def add_team_callbacks(
         if team_callback_settings is None or not isinstance(team_callback_settings, list):
             team_callback_settings = []
 
+        # One entry has to own a credential family end to end. The entries are
+        # flattened into one dict before a request reads them, so an entry
+        # naming only a destination would pair with a key written on another
+        # entry and carry it to that destination -- a key a team admin can read
+        # back nowhere. Proxy admins are exempt: they already hold every
+        # credential the proxy has.
+        if user_api_key_dict.user_role != LitellmUserRoles.PROXY_ADMIN:
+            stored_entry_vars: Final = [  # mutable-ok: read-only input to the check, never stored
+                entry.get("callback_vars") or {} for entry in team_callback_settings
+            ]
+            family_error: Final = cross_entry_family_error(data.callback_vars, stored_entry_vars)
+            if family_error is not None:
+                raise HTTPException(
+                    status_code=status.HTTP_400_BAD_REQUEST,
+                    detail=family_error,
+                )
+
         ## check if it already exists, for the same callback event
         for callback in team_callback_settings:
             if (

tests/test_litellm/proxy/management_endpoints/test_team_callback_endpoints.py

@@ -19,6 +19,7 @@ from litellm.proxy._types import (
     LitellmUserRoles,
     UserAPIKeyAuth,
 )
+from litellm.proxy.common_utils.callback_config_validation import cross_entry_family_error
 from litellm.proxy.management_endpoints.team_callback_endpoints import (
     add_team_callbacks,
     delete_team_callback,
@@ -1518,3 +1519,33 @@ async def test_proxy_admin_still_told_the_team_is_unknown():
 
     assert exc.value.status_code == 404
     assert "does not exist" in str(exc.value.detail)
+
+
+@pytest.mark.parametrize(
+    "new_vars, stored, rejected",
+    [
+        # the redirect, in every carrier a caller could pick: an entry naming
+        # only a host, pairing with a key pair written on another entry
+        ({"langfuse_host": "http://attacker.invalid"}, [{"langfuse_public_key": "pk", "langfuse_secret_key": "sk"}], True),
+        # the sibling carrier -- langfuse and langfuse_otel are one account
+        ({"langfuse_host": "http://attacker.invalid"}, [{"langfuse_host": "https://us.cloud.langfuse.com", "langfuse_secret_key": "sk"}], True),
+        # a destination variable no integration registry lists
+        ({"dd_agent_host": "attacker.invalid"}, [{"dd_api_key": "k", "dd_site": "us5.datadoghq.com"}], True),
+        # one entry owning its family end to end is the feature
+        ({"langfuse_host": "https://eu.cloud.langfuse.com", "langfuse_public_key": "pk", "langfuse_secret_key": "sk"}, [], False),
+        # a different family alongside an existing one stays fine
+        ({"gcs_bucket_name": "bucket"}, [{"langfuse_public_key": "pk", "langfuse_secret_key": "sk"}], False),
+        ({"langsmith_api_key": "k"}, [{"dd_api_key": "k"}], False),
+        # variables that configure no backend carry nothing to redirect
+        ({"turn_off_message_logging": "true"}, [{"langfuse_secret_key": "sk"}], False),
+    ],
+)
+def test_one_entry_owns_a_credential_family(new_vars, stored, rejected):
+    """A team admin must not be able to redirect a credential they cannot read.
+
+    The stored entries are flattened into one dict before a request reads them,
+    so an entry naming only a destination pairs with a key written elsewhere and
+    carries it to that destination.
+    """
+    error = cross_entry_family_error(new_vars, stored)
+    assert (error is not None) is rejected