fix(anthropic): pin the federated token exchange to trusted hosts

The exchange derived its token URL from the deployment's api_base and sent the
signed assertion wherever that pointed. Any write path that could set api_base on
a WIF deployment could therefore redirect a valid assertion to a host of its
choosing, and gating each of those paths individually has no termination condition
as new ones are added.

Enforce it where the exchange is actually built instead: the host must be
api.anthropic.com or an entry in LITELLM_ANTHROPIC_WIF_ALLOWED_HOSTS, matched on
the parsed hostname so a lookalike like api.anthropic.com.evil.test cannot pass by
containing the real one. Both the sync and async call sites check before any
assertion leaves the process.

litellm/llms/anthropic/wif.py

@@ -1,7 +1,9 @@
 """Anthropic workload identity federation: exchanges an external OIDC identity
 token for a short-lived ``sk-ant-oat01`` token via the shared RFC 7523 engine."""
 
+import os
 from collections.abc import Callable, Mapping
+from itertools import chain
 from types import MappingProxyType
 from typing import Final, NoReturn, TypeVar
 from urllib.parse import urlsplit, urlunsplit
@@ -41,6 +43,14 @@ _INLINE_ENV_VAR: Final = "ANTHROPIC_IDENTITY_TOKEN"
 _DISABLE_WIF_PARAM: Final = "anthropic_disable_workload_identity_federation"
 _ACCEPTED_REF_PREFIX: Final = "oidc/"
 _CHAT_BASE_SUFFIXES: Final = ("/v1/messages", "/v1")
+# Hosts a federated exchange may talk to. api_base decides where the workload's assertion is sent
+# AND where the minted org-scoped token is presented, so anyone able to write api_base on a
+# federated deployment could otherwise redirect both. Gating each write path does not terminate:
+# a deployment, a referenced credential and a future endpoint all reach the same value. This is the
+# one place a federated exchange is built, so the trust decision is enforced here instead, and the
+# allowlist is server-owned -- read from the environment, never from a model or credential API.
+_TRUSTED_EXCHANGE_HOSTS_ENV: Final = "LITELLM_ANTHROPIC_WIF_ALLOWED_HOSTS"
+_DEFAULT_TRUSTED_EXCHANGE_HOST: Final = "api.anthropic.com"
 _REJECTED_REF_PREFIX: Final = "oidc/env_path/"
 _IDENTITY_SOURCE_PARAM: Final = "anthropic_identity_source"
 _IDENTITY_SOURCE_ENV: Final = "ANTHROPIC_IDENTITY_SOURCE"
@@ -228,7 +238,9 @@ def get_anthropic_wif_token(
     params: Final = resolve_anthropic_wif_params(litellm_params)
     if params is None:
         return None
-    result: Final = engine.get_token(build_anthropic_wif_spec(params, _token_exchange_base(api_base)))
+    exchange_base: Final = _token_exchange_base(api_base)
+    _raise_if_exchange_host_untrusted(exchange_base, model)
+    result: Final = engine.get_token(build_anthropic_wif_spec(params, exchange_base))
     return _token_from_result(result, model, params)
 
 
@@ -241,7 +253,9 @@ async def aget_anthropic_wif_token(
     params: Final = resolve_anthropic_wif_params(litellm_params)
     if params is None:
         return None
-    result: Final = await engine.aget_token(build_anthropic_wif_spec(params, _token_exchange_base(api_base)))
+    exchange_base: Final = _token_exchange_base(api_base)
+    _raise_if_exchange_host_untrusted(exchange_base, model)
+    result: Final = await engine.aget_token(build_anthropic_wif_spec(params, exchange_base))
     return _token_from_result(result, model, params)
 
 
@@ -260,6 +274,41 @@ def _token_exchange_base(api_base: str | None) -> str:
     return _strip_chat_suffix(api_base if api_base is not None else _resolve_default_api_base())
 
 
+def _trusted_exchange_hosts() -> frozenset[str]:
+    """Hostnames a federated exchange may reach: Anthropic's own, plus whatever the operator put in
+    the environment. Comma separated, case folded, entries given as a URL reduced to their host."""
+    configured: Final = os.getenv(_TRUSTED_EXCHANGE_HOSTS_ENV) or ""
+    extra: Final = (entry.strip() for entry in configured.split(",") if entry.strip())
+    return frozenset(
+        chain(
+            (_DEFAULT_TRUSTED_EXCHANGE_HOST,),
+            ((urlsplit(entry).hostname or entry.split("/")[0]).lower() for entry in extra),
+        )
+    )
+
+
+def _raise_if_exchange_host_untrusted(exchange_base: str, model: str) -> None:
+    """The federated exchange refuses any host the operator has not vouched for, whatever wrote the
+    deployment's api_base. Exact hostname match, never a substring: ``api.anthropic.com.evil.test``
+    contains the real host and must not pass."""
+    host: Final = (urlsplit(exchange_base).hostname or "").lower()
+    if host and host in _trusted_exchange_hosts():
+        return
+    raise litellm.AuthenticationError(
+        message=(
+            f"Anthropic workload identity federation refused to use host {host or exchange_base!r}. "
+            f"A federated exchange sends the workload's identity token to this host and presents the "
+            f"minted token to it, so only {_DEFAULT_TRUSTED_EXCHANGE_HOST} is trusted by default. To "
+            f"use a private Anthropic-compatible gateway, add its hostname to the "
+            f"{_TRUSTED_EXCHANGE_HOSTS_ENV} environment variable (comma separated); that is a "
+            f"decision to trust it with org-scoped credentials, so it is deliberately server-owned "
+            f"and cannot be set through the model or credential APIs."
+        ),
+        llm_provider="anthropic",
+        model=model,
+    )
+
+
 def _resolve_default_api_base() -> str:
     from litellm.llms.anthropic.common_utils import AnthropicModelInfo
 

tests/test_litellm/llms/anthropic/test_anthropic_common_utils.py

@@ -2572,6 +2572,9 @@ class TestWifTokenUrlParity:
         ],
     )
     def test_both_tiers_share_one_clean_token_url(self, monkeypatch, wif_engine, configured_base):
+        # This is about deriving one URL from many spellings of the same base, not about which
+        # hosts an operator trusts with org-scoped credentials, so the private host is allowlisted.
+        monkeypatch.setenv("LITELLM_ANTHROPIC_WIF_ALLOWED_HOSTS", "gw.example.com")
         from litellm.llms.anthropic.common_utils import AnthropicModelInfo
 
         poster, _ = wif_engine

tests/test_litellm/llms/anthropic/test_anthropic_wif.py

@@ -212,9 +212,76 @@ class TestWireProtocolExact:
         assert dict(spec.request_headers) == {}
 
 
+class TestExchangeHostTrust:
+    """A federated exchange sends the workload's identity token to api_base and presents the minted
+    org-scoped token to it, so api_base is a trust decision. Anyone able to write api_base, on the
+    deployment or on a credential it references, could otherwise redirect both, which is why this is
+    enforced where the exchange is built rather than at each write path."""
+
+    def _mint(self, api_base: str | None, monkeypatch: pytest.MonkeyPatch) -> str:
+        monkeypatch.setenv("ANTHROPIC_IDENTITY_TOKEN", "inline-jwt")
+        poster = ScriptedPoster([token_response()])
+        get_anthropic_wif_token(
+            {"anthropic_federation_rule_id": "fdrl_1", "anthropic_organization_id": "org-1"},
+            api_base,
+            "claude-sonnet-4-5",
+            make_engine(poster),
+        )
+        return poster.requests[0].url
+
+    def test_anthropic_is_trusted_without_configuration(self, monkeypatch: pytest.MonkeyPatch):
+        monkeypatch.delenv("LITELLM_ANTHROPIC_WIF_ALLOWED_HOSTS", raising=False)
+        assert self._mint("https://api.anthropic.com", monkeypatch) == "https://api.anthropic.com/v1/oauth/token"
+
+    def test_an_unlisted_host_never_receives_the_identity_token(self, monkeypatch: pytest.MonkeyPatch):
+        monkeypatch.delenv("LITELLM_ANTHROPIC_WIF_ALLOWED_HOSTS", raising=False)
+        monkeypatch.setenv("ANTHROPIC_IDENTITY_TOKEN", "inline-jwt")
+        poster = ScriptedPoster([token_response()])
+
+        with pytest.raises(litellm.AuthenticationError) as exc_info:
+            get_anthropic_wif_token(
+                {"anthropic_federation_rule_id": "fdrl_1", "anthropic_organization_id": "org-1"},
+                "https://attacker.example",
+                "claude-sonnet-4-5",
+                make_engine(poster),
+            )
+
+        assert poster.requests == [], "the exchange must be refused before anything is sent"
+        assert "attacker.example" in str(exc_info.value)
+        assert "LITELLM_ANTHROPIC_WIF_ALLOWED_HOSTS" in str(exc_info.value), (
+            "an operator running a private gateway has to be told how to allow it"
+        )
+
+    def test_a_lookalike_host_does_not_pass_on_a_substring(self, monkeypatch: pytest.MonkeyPatch):
+        monkeypatch.delenv("LITELLM_ANTHROPIC_WIF_ALLOWED_HOSTS", raising=False)
+        monkeypatch.setenv("ANTHROPIC_IDENTITY_TOKEN", "inline-jwt")
+        poster = ScriptedPoster([token_response()])
+
+        with pytest.raises(litellm.AuthenticationError):
+            get_anthropic_wif_token(
+                {"anthropic_federation_rule_id": "fdrl_1", "anthropic_organization_id": "org-1"},
+                "https://api.anthropic.com.evil.test",
+                "claude-sonnet-4-5",
+                make_engine(poster),
+            )
+
+        assert poster.requests == []
+
+    def test_an_operator_can_allow_a_private_gateway(self, monkeypatch: pytest.MonkeyPatch):
+        monkeypatch.setenv("LITELLM_ANTHROPIC_WIF_ALLOWED_HOSTS", "gateway.internal")
+        assert self._mint("https://gateway.internal", monkeypatch) == "https://gateway.internal/v1/oauth/token"
+
+
 class TestBaseUrlDerivation:
     def _mint(self, api_base: str | None, monkeypatch: pytest.MonkeyPatch) -> str:
         monkeypatch.setenv("ANTHROPIC_IDENTITY_TOKEN", "inline-jwt")
+        # These cases are about how a base is normalised into a token URL, not about which hosts an
+        # operator trusts, so the private hosts they use are allowlisted explicitly. The trust
+        # boundary itself is covered by TestExchangeHostTrust.
+        monkeypatch.setenv(
+            "LITELLM_ANTHROPIC_WIF_ALLOWED_HOSTS",
+            "gw.example.com,env.example.com,base.example.com,model.example.com",
+        )
         poster = ScriptedPoster([token_response()])
         engine = make_engine(poster)
         get_anthropic_wif_token(
@@ -407,9 +474,7 @@ class TestServiceAccountIdIsOptional:
     without it; resolution must not gate activation on it, and the wire body must omit
     the key entirely rather than send it as null."""
 
-    def test_activates_and_omits_service_account_id_when_unset(
-        self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch
-    ):
+    def test_activates_and_omits_service_account_id_when_unset(self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch):
         monkeypatch.setenv("LITELLM_OIDC_ALLOWED_CREDENTIAL_DIRS", str(tmp_path))
         token_file = write_token_file(tmp_path, "jwt-assertion-value")
         litellm_params: Final = {
@@ -484,9 +549,7 @@ class TestFileAllowlistAndSymlink:
         assert self.SECRET_CONTENT not in exc_info.value.message
         assert poster.requests == []
 
-    def test_disallowed_path_message_names_allowlist_and_env_var(
-        self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch
-    ):
+    def test_disallowed_path_message_names_allowlist_and_env_var(self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch):
         """The disallowed_path error must explain the allowlist and name the env var an
         operator would set, not surface as a bare '(disallowed_path)' code dump."""
         monkeypatch.setenv("LITELLM_OIDC_ALLOWED_CREDENTIAL_DIRS", str(tmp_path / "allowed"))