From d832b80403793b718a1a0b59e1a14ae27c09c7ff Mon Sep 17 00:00:00 2001 From: derhornspieler <15236687+derhornspieler@users.noreply.github.com> Date: Sun, 23 Aug 2026 19:36:58 -0400 Subject: [PATCH] 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 | 53 ++++++++++++- .../anthropic/test_anthropic_common_utils.py | 3 + .../llms/anthropic/test_anthropic_wif.py | 75 +++++++++++++++++-- 3 files changed, 123 insertions(+), 8 deletions(-) diff --git a/litellm/llms/anthropic/wif.py b/litellm/llms/anthropic/wif.py index 7900bcebcd7..0342926a598 100644 --- a/litellm/llms/anthropic/wif.py +++ b/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 diff --git a/tests/test_litellm/llms/anthropic/test_anthropic_common_utils.py b/tests/test_litellm/llms/anthropic/test_anthropic_common_utils.py index 73738d97259..52713f32374 100644 --- a/tests/test_litellm/llms/anthropic/test_anthropic_common_utils.py +++ b/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 diff --git a/tests/test_litellm/llms/anthropic/test_anthropic_wif.py b/tests/test_litellm/llms/anthropic/test_anthropic_wif.py index 56420956ae5..09770b19b2e 100644 --- a/tests/test_litellm/llms/anthropic/test_anthropic_wif.py +++ b/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"))