mirror of
https://github.com/BerriAI/litellm.git
synced 2026-10-06 02:48:13 +00:00
fix(anthropic): stop workload identity federation from reaching non-Anthropic providers
The federation exchange derives its token endpoint from the deployment's api_base and returns an Anthropic-org credential, so it must only ever run for Anthropic itself. It did not. VertexAIAnthropicConfig inherits AnthropicModelInfo.validate_environment, and the MiniMax and Tencent /v1/messages configs inherit the Anthropic validate step, so a proxy configured with the ANTHROPIC_* federation variables POSTed the workload's OIDC assertion to those providers' own hosts, confirmed against a real deployment base Eligibility is now declared per class and read from that class's own __dict__, so a subclass written for another provider inherits nothing, and the get_auth_header facades refuse to mint unless a caller states that it authenticates against Anthropic's own API. The six federation fields join the banned request-body parameters: they select which server-side secret is read and, with api_base, where it is sent, so a caller-supplied value was an exfiltration primitive for any environment variable or mounted token file. The exchange client no longer follows redirects, since only the initial token URL is validated, and a base that already ends in /v1 no longer yields a doubled /v1/v1/oauth/token
This commit is contained in:
derhornspieler
parent
61a1122420
commit
20da15edb6
13 changed files with 207 additions and 55 deletions
|
|
@ -48,7 +48,9 @@ class AnthropicBatchesConfig(BaseBatchesConfig):
|
|||
params_mapping: Final = litellm_params if isinstance(litellm_params, dict) else None
|
||||
if api_base is None and params_mapping is not None:
|
||||
api_base = params_mapping.get("api_base")
|
||||
auth_header: Final = self.anthropic_model_info.get_auth_header(api_key, api_base, litellm_params=params_mapping)
|
||||
auth_header: Final = self.anthropic_model_info.get_auth_header(
|
||||
api_key, api_base, litellm_params=params_mapping, allow_workload_identity=True
|
||||
)
|
||||
if auth_header is None:
|
||||
raise ValueError(
|
||||
"Missing Anthropic API Key - A call is being made to anthropic but no key is set either in the environment variables or via params"
|
||||
|
|
|
|||
|
|
@ -2,7 +2,7 @@ import json
|
|||
import re
|
||||
import time
|
||||
from collections.abc import Mapping, Sequence
|
||||
from typing import TYPE_CHECKING, Any, Final, NoReturn, cast
|
||||
from typing import TYPE_CHECKING, Any, ClassVar, Final, NoReturn, cast
|
||||
|
||||
import httpx
|
||||
from pydantic import ValidationError
|
||||
|
|
@ -230,6 +230,8 @@ DROP_UNSUPPORTED_SPEED_WARNING: Final = (
|
|||
|
||||
|
||||
class AnthropicConfig(AnthropicModelInfo, BaseConfig):
|
||||
_workload_identity_eligible: ClassVar[bool] = True
|
||||
|
||||
"""
|
||||
Reference: https://docs.anthropic.com/claude/reference/messages_post
|
||||
|
||||
|
|
|
|||
|
|
@ -7,7 +7,7 @@ import re
|
|||
from collections.abc import Mapping, Sequence
|
||||
from datetime import datetime, timezone
|
||||
from types import MappingProxyType
|
||||
from typing import Any, Final, Literal
|
||||
from typing import Any, ClassVar, Final, Literal
|
||||
|
||||
import httpx
|
||||
from pydantic import BaseModel, ConfigDict, TypeAdapter, ValidationError
|
||||
|
|
@ -58,6 +58,14 @@ def _strip_bedrock_id_suffixes(model: str) -> str:
|
|||
|
||||
|
||||
_SERVER_OWNED_AUTH_HEADERS: Final = frozenset({"x-api-key", "authorization"})
|
||||
_WIF_ELIGIBILITY_ATTR: Final = "_workload_identity_eligible"
|
||||
|
||||
|
||||
def config_allows_workload_identity(config: object) -> bool:
|
||||
"""A federation token is an Anthropic-org credential and its exchange POSTs the workload's OIDC
|
||||
assertion to the deployment's own host, so eligibility is declared per class and read from that
|
||||
class's own ``__dict__``: a subclass written for another provider inherits nothing."""
|
||||
return type(config).__dict__.get(_WIF_ELIGIBILITY_ATTR, False) is True
|
||||
|
||||
|
||||
def is_anthropic_oauth_key(value: str | None) -> bool:
|
||||
|
|
@ -121,6 +129,8 @@ class AnthropicError(BaseLLMException):
|
|||
|
||||
|
||||
class AnthropicModelInfo(BaseLLMModelInfo):
|
||||
_workload_identity_eligible: ClassVar[bool] = True
|
||||
|
||||
def is_cache_control_set(self, messages: list[AllMessageValues]) -> bool:
|
||||
"""
|
||||
Return if {"cache_control": ..} in message content block
|
||||
|
|
@ -723,7 +733,9 @@ class AnthropicModelInfo(BaseLLMModelInfo):
|
|||
if api_key is None:
|
||||
auth_token = AnthropicModelInfo.get_auth_token()
|
||||
wif_token: Final = (
|
||||
get_anthropic_wif_token(params_mapping, api_base, model) if api_key is None and auth_token is None else None
|
||||
get_anthropic_wif_token(params_mapping, api_base, model)
|
||||
if api_key is None and auth_token is None and config_allows_workload_identity(self)
|
||||
else None
|
||||
)
|
||||
wif_minted: Final = wif_token is not None
|
||||
resolved_api_key: Final = wif_token if wif_token is not None else api_key
|
||||
|
|
@ -821,6 +833,7 @@ class AnthropicModelInfo(BaseLLMModelInfo):
|
|||
api_base: str | None = None,
|
||||
use_bearer_for_custom_base: bool = False,
|
||||
litellm_params: Mapping[str, object] | None = None,
|
||||
allow_workload_identity: bool = False,
|
||||
) -> Mapping[str, str] | None:
|
||||
"""Resolve Anthropic credentials and return the appropriate auth header dict.
|
||||
|
||||
|
|
@ -834,6 +847,8 @@ class AnthropicModelInfo(BaseLLMModelInfo):
|
|||
static_header: Final = AnthropicModelInfo._static_auth_header(api_key, api_base, use_bearer_for_custom_base)
|
||||
if static_header is not None:
|
||||
return static_header
|
||||
if not allow_workload_identity:
|
||||
return None
|
||||
wif_token: Final = get_anthropic_wif_token(litellm_params, api_base, "")
|
||||
if wif_token is not None:
|
||||
return AnthropicModelInfo._oauth_bearer_header(wif_token)
|
||||
|
|
@ -845,12 +860,15 @@ class AnthropicModelInfo(BaseLLMModelInfo):
|
|||
api_base: str | None = None,
|
||||
use_bearer_for_custom_base: bool = False,
|
||||
litellm_params: Mapping[str, object] | None = None,
|
||||
allow_workload_identity: bool = False,
|
||||
) -> Mapping[str, str] | None:
|
||||
"""Async counterpart of get_auth_header: the WIF tier can block on a token
|
||||
exchange POST, so async callers await it off the event loop."""
|
||||
static_header: Final = AnthropicModelInfo._static_auth_header(api_key, api_base, use_bearer_for_custom_base)
|
||||
if static_header is not None:
|
||||
return static_header
|
||||
if not allow_workload_identity:
|
||||
return None
|
||||
wif_token: Final = await aget_anthropic_wif_token(litellm_params, api_base, "")
|
||||
if wif_token is not None:
|
||||
return AnthropicModelInfo._oauth_bearer_header(wif_token)
|
||||
|
|
@ -882,7 +900,9 @@ class AnthropicModelInfo(BaseLLMModelInfo):
|
|||
|
||||
def get_models(self, api_key: str | None = None, api_base: str | None = None) -> list[str]:
|
||||
api_base = AnthropicModelInfo.get_api_base(api_base)
|
||||
auth_header: Final = AnthropicModelInfo.get_auth_header(api_key, api_base)
|
||||
auth_header: Final = AnthropicModelInfo.get_auth_header(
|
||||
api_key, api_base, allow_workload_identity=config_allows_workload_identity(self)
|
||||
)
|
||||
if api_base is None or auth_header is None:
|
||||
raise ValueError(
|
||||
"ANTHROPIC_API_BASE/ANTHROPIC_BASE_URL or ANTHROPIC_API_KEY/ANTHROPIC_AUTH_TOKEN (or workload "
|
||||
|
|
|
|||
|
|
@ -1,5 +1,5 @@
|
|||
from collections.abc import AsyncIterator, Mapping, Sequence
|
||||
from typing import Any, Final
|
||||
from typing import Any, ClassVar, Final
|
||||
|
||||
import httpx
|
||||
|
||||
|
|
@ -42,6 +42,8 @@ DROP_UNSUPPORTED_ADAPTIVE_EFFORT_WARNING: Final = (
|
|||
|
||||
|
||||
class AnthropicMessagesConfig(BaseAnthropicMessagesConfig):
|
||||
_workload_identity_eligible: ClassVar[bool] = True
|
||||
|
||||
@property
|
||||
def custom_llm_provider(self) -> str | None:
|
||||
return "anthropic"
|
||||
|
|
@ -314,7 +316,8 @@ class AnthropicMessagesConfig(BaseAnthropicMessagesConfig):
|
|||
AnthropicModelInfo.get_auth_header(
|
||||
api_key,
|
||||
api_base=api_base,
|
||||
litellm_params=self._wif_litellm_params(litellm_params),
|
||||
litellm_params=litellm_params,
|
||||
allow_workload_identity=self._allows_workload_identity,
|
||||
),
|
||||
)
|
||||
return self._finalize_messages_headers(headers, optional_params), api_base
|
||||
|
|
@ -350,7 +353,8 @@ class AnthropicMessagesConfig(BaseAnthropicMessagesConfig):
|
|||
await AnthropicModelInfo.aget_auth_header(
|
||||
oauth_api_key,
|
||||
api_base=api_base,
|
||||
litellm_params=self._wif_litellm_params(litellm_params),
|
||||
litellm_params=litellm_params,
|
||||
allow_workload_identity=self._allows_workload_identity,
|
||||
),
|
||||
)
|
||||
return self._finalize_messages_headers(oauth_headers, optional_params), api_base
|
||||
|
|
@ -366,12 +370,13 @@ class AnthropicMessagesConfig(BaseAnthropicMessagesConfig):
|
|||
if merged_beta:
|
||||
headers["anthropic-beta"] = merged_beta
|
||||
|
||||
def _wif_litellm_params(self, litellm_params: dict) -> Mapping[str, object] | None: # mutable-ok: sync-contract mirror
|
||||
"""Subclasses reuse this validate step for their own /v1/messages-compatible providers,
|
||||
so an Anthropic federation token is only ever minted for Anthropic itself."""
|
||||
if self._resolved_provider != "anthropic":
|
||||
return None
|
||||
return litellm_params if isinstance(litellm_params, dict) else None
|
||||
@property
|
||||
def _allows_workload_identity(self) -> bool:
|
||||
"""Subclasses reuse this validate step for their own /v1/messages-compatible providers, so
|
||||
eligibility is declared per class and never inherited."""
|
||||
from litellm.llms.anthropic.common_utils import config_allows_workload_identity
|
||||
|
||||
return config_allows_workload_identity(self)
|
||||
|
||||
def _finalize_messages_headers(self, headers: dict, optional_params: dict) -> dict: # mutable-ok: out-param
|
||||
if "anthropic-version" not in headers:
|
||||
|
|
|
|||
|
|
@ -85,7 +85,9 @@ class AnthropicFilesHandler:
|
|||
|
||||
# Get Anthropic API credentials
|
||||
api_base = self.anthropic_model_info.get_api_base(api_base)
|
||||
auth_header: Final = await self.anthropic_model_info.aget_auth_header(api_key, api_base)
|
||||
auth_header: Final = await self.anthropic_model_info.aget_auth_header(
|
||||
api_key, api_base, allow_workload_identity=True
|
||||
)
|
||||
|
||||
if auth_header is None:
|
||||
raise ValueError("Missing Anthropic API Key")
|
||||
|
|
|
|||
|
|
@ -97,7 +97,9 @@ class AnthropicFilesConfig(BaseFilesConfig):
|
|||
params_mapping: Final = litellm_params if isinstance(litellm_params, dict) else None
|
||||
if api_base is None and params_mapping is not None:
|
||||
api_base = params_mapping.get("api_base")
|
||||
auth_header: Final = AnthropicModelInfo.get_auth_header(api_key, api_base, litellm_params=params_mapping)
|
||||
auth_header: Final = AnthropicModelInfo.get_auth_header(
|
||||
api_key, api_base, litellm_params=params_mapping, allow_workload_identity=True
|
||||
)
|
||||
if auth_header is None:
|
||||
raise ValueError(
|
||||
"Anthropic API key is required. Set ANTHROPIC_API_KEY or ANTHROPIC_AUTH_TOKEN environment variable or pass api_key parameter."
|
||||
|
|
|
|||
|
|
@ -43,6 +43,7 @@ class AnthropicSkillsConfig(BaseSkillsAPIConfig):
|
|||
api_key=litellm_params.api_key if litellm_params is not None else None,
|
||||
api_base=litellm_params.api_base if litellm_params is not None else None,
|
||||
litellm_params=MappingProxyType(dict(litellm_params)) if litellm_params is not None else None,
|
||||
allow_workload_identity=True,
|
||||
)
|
||||
if auth_header is None:
|
||||
raise ValueError("ANTHROPIC_API_KEY or ANTHROPIC_AUTH_TOKEN is required for Skills API")
|
||||
|
|
|
|||
|
|
@ -30,6 +30,7 @@ _JWT_BEARER_GRANT_TYPE: Final = "urn:ietf:params:oauth:grant-type:jwt-bearer"
|
|||
_DEFAULT_API_BASE: Final = "https://api.anthropic.com"
|
||||
_INLINE_ENV_VAR: Final = "ANTHROPIC_IDENTITY_TOKEN"
|
||||
_ACCEPTED_REF_PREFIX: Final = "oidc/"
|
||||
_CHAT_BASE_SUFFIXES: Final = ("/v1/messages", "/v1")
|
||||
_REJECTED_REF_PREFIX: Final = "oidc/env_path/"
|
||||
_WORKSPACE_HINT: Final = (
|
||||
" If the federation rule is scoped to a workspace, set ANTHROPIC_WORKSPACE_ID"
|
||||
|
|
@ -151,8 +152,11 @@ def _resolve_default_api_base() -> str:
|
|||
|
||||
def _strip_chat_suffix(base: str) -> str:
|
||||
trimmed: Final = base.rstrip("/")
|
||||
stripped: Final = trimmed.removesuffix("/v1/messages")
|
||||
return stripped if stripped == trimmed else _strip_chat_suffix(stripped)
|
||||
stripped: Final = next(
|
||||
(trimmed.removesuffix(suffix) for suffix in _CHAT_BASE_SUFFIXES if trimmed.endswith(suffix)),
|
||||
trimmed,
|
||||
)
|
||||
return trimmed if stripped == trimmed else _strip_chat_suffix(stripped)
|
||||
|
||||
|
||||
def _config_value(litellm_params: Mapping[str, object] | None, param_key: str, env_name: str) -> str | None:
|
||||
|
|
|
|||
|
|
@ -214,7 +214,9 @@ class _HttpxSyncTokenPoster:
|
|||
|
||||
with self._lock:
|
||||
if self._handler is None:
|
||||
self._handler = HTTPHandler(timeout=httpx.Timeout(timeout=30.0, connect=5.0))
|
||||
handler: Final = HTTPHandler(timeout=httpx.Timeout(timeout=30.0, connect=5.0))
|
||||
handler.client.follow_redirects = False
|
||||
self._handler = handler
|
||||
return self._handler
|
||||
|
||||
def post(self, url: str, *, content: bytes, headers: Mapping[str, str], timeout: float) -> httpx.Response:
|
||||
|
|
|
|||
|
|
@ -31,7 +31,7 @@ from litellm.types.passthrough_endpoints.pass_through_endpoints import (
|
|||
LITELLM_PASS_THROUGH_ENDPOINT_MARKER,
|
||||
)
|
||||
from litellm.types.router import CONFIGURABLE_CLIENTSIDE_AUTH_PARAMS
|
||||
from litellm.types.utils import CustomPricingLiteLLMParams
|
||||
from litellm.types.utils import CustomPricingLiteLLMParams, anthropic_wif_litellm_params
|
||||
|
||||
|
||||
def _get_request_ip_address(request: Request, use_x_forwarded_for: bool | None = False) -> str | None:
|
||||
|
|
@ -317,6 +317,11 @@ _BANNED_REQUEST_BODY_PARAMS: Final[tuple[str, ...]] = (
|
|||
# so a caller-supplied value picks a transport and a callback surface the
|
||||
# admin did not choose.
|
||||
"rust",
|
||||
# Anthropic workload-identity federation. These select which server-side secret is read
|
||||
# (``anthropic_identity_token`` resolves an ``oidc/...`` reference against the proxy's own
|
||||
# environment and filesystem) and, together with ``api_base``, where that secret is sent, so a
|
||||
# caller-supplied value is an exfiltration primitive for any env var or mounted token file.
|
||||
*sorted(anthropic_wif_litellm_params),
|
||||
# SDK-only field; also rejected outright in is_request_body_safe.
|
||||
"model_list",
|
||||
"vertex_ai_credentials",
|
||||
|
|
|
|||
|
|
@ -617,7 +617,9 @@ async def anthropic_proxy_route(
|
|||
is_streaming_request: Final = await is_streaming_request_fn(request)
|
||||
|
||||
## CREATE PASS-THROUGH
|
||||
auth_header: Final = await AnthropicModelInfo.aget_auth_header(anthropic_api_key or None)
|
||||
auth_header: Final = await AnthropicModelInfo.aget_auth_header(
|
||||
anthropic_api_key or None, allow_workload_identity=True
|
||||
)
|
||||
endpoint_func: Final = create_pass_through_route(
|
||||
endpoint=endpoint,
|
||||
target=str(updated_url),
|
||||
|
|
|
|||
|
|
@ -24,16 +24,16 @@ _WIF_PARAMS: Final[dict] = {
|
|||
}
|
||||
|
||||
|
||||
def test_wif_litellm_params_passthrough_for_anthropic() -> None:
|
||||
assert AnthropicMessagesConfig()._wif_litellm_params(_WIF_PARAMS) == _WIF_PARAMS
|
||||
def test_workload_identity_allowed_for_anthropic() -> None:
|
||||
assert AnthropicMessagesConfig()._allows_workload_identity is True
|
||||
|
||||
|
||||
def test_wif_litellm_params_blocked_for_minimax() -> None:
|
||||
assert MinimaxMessagesConfig()._wif_litellm_params(_WIF_PARAMS) is None
|
||||
def test_workload_identity_blocked_for_minimax() -> None:
|
||||
assert MinimaxMessagesConfig()._allows_workload_identity is False
|
||||
|
||||
|
||||
def test_wif_litellm_params_blocked_for_tencent() -> None:
|
||||
assert TencentAnthropicMessagesConfig()._wif_litellm_params(_WIF_PARAMS) is None
|
||||
def test_workload_identity_blocked_for_tencent() -> None:
|
||||
assert TencentAnthropicMessagesConfig()._allows_workload_identity is False
|
||||
|
||||
|
||||
def test_minimax_validate_environment_never_attaches_anthropic_wif_credential(
|
||||
|
|
@ -74,20 +74,16 @@ def test_tencent_validate_environment_never_attaches_anthropic_wif_credential(
|
|||
token_file = write_token_file(tmp_path, "jwt-assertion-value")
|
||||
litellm_params = {"anthropic_identity_token_file": str(token_file)}
|
||||
|
||||
original_api_key: Final = litellm.api_key
|
||||
litellm.api_key = None
|
||||
try:
|
||||
headers, _ = TencentAnthropicMessagesConfig().validate_anthropic_messages_environment(
|
||||
headers={},
|
||||
model="deepseek-v4-pro",
|
||||
messages=[],
|
||||
optional_params={},
|
||||
litellm_params=litellm_params,
|
||||
api_key=None,
|
||||
api_base="https://tokenhub-intl.tencentcloudmaas.com",
|
||||
)
|
||||
finally:
|
||||
litellm.api_key = original_api_key
|
||||
monkeypatch.setattr(litellm, "api_key", None)
|
||||
headers, _ = TencentAnthropicMessagesConfig().validate_anthropic_messages_environment(
|
||||
headers={},
|
||||
model="deepseek-v4-pro",
|
||||
messages=[],
|
||||
optional_params={},
|
||||
litellm_params=litellm_params,
|
||||
api_key=None,
|
||||
api_base="https://tokenhub-intl.tencentcloudmaas.com",
|
||||
)
|
||||
|
||||
assert "authorization" not in headers
|
||||
assert "x-api-key" not in headers
|
||||
|
|
@ -110,7 +106,7 @@ def test_wif_token_exchange_reaches_only_anthropic_not_minimax_or_tencent(
|
|||
engine = make_engine(poster)
|
||||
|
||||
minted: Final = get_anthropic_wif_token(
|
||||
AnthropicMessagesConfig()._wif_litellm_params(litellm_params),
|
||||
litellm_params,
|
||||
"https://api.anthropic.com",
|
||||
"claude-sonnet-4-5",
|
||||
engine,
|
||||
|
|
@ -119,14 +115,6 @@ def test_wif_token_exchange_reaches_only_anthropic_not_minimax_or_tencent(
|
|||
assert len(poster.requests) == 1
|
||||
|
||||
for config in (MinimaxMessagesConfig(), TencentAnthropicMessagesConfig()):
|
||||
assert (
|
||||
get_anthropic_wif_token(
|
||||
config._wif_litellm_params(litellm_params),
|
||||
"https://api.minimax.io/anthropic",
|
||||
"MiniMax-M2.1",
|
||||
engine,
|
||||
)
|
||||
is None
|
||||
)
|
||||
assert config._allows_workload_identity is False
|
||||
|
||||
assert len(poster.requests) == 1
|
||||
|
|
|
|||
|
|
@ -2328,7 +2328,7 @@ class TestWifResolvedApiKeyThreading:
|
|||
for name, value in WIF_ENV.items():
|
||||
monkeypatch.setenv(name, value)
|
||||
|
||||
result = await AnthropicModelInfo.aget_auth_header()
|
||||
result = await AnthropicModelInfo.aget_auth_header(allow_workload_identity=True)
|
||||
|
||||
assert result is not None
|
||||
assert result["authorization"] not in (None, "Bearer None")
|
||||
|
|
@ -2355,7 +2355,7 @@ class TestGetAuthHeaderBetas:
|
|||
for name, value in WIF_ENV.items():
|
||||
monkeypatch.setenv(name, value)
|
||||
|
||||
result = AnthropicModelInfo.get_auth_header()
|
||||
result = AnthropicModelInfo.get_auth_header(allow_workload_identity=True)
|
||||
|
||||
assert result == {
|
||||
"authorization": f"Bearer {FAKE_MINTED_TOKEN}",
|
||||
|
|
@ -2552,7 +2552,7 @@ class TestWifTokenUrlParity:
|
|||
api_key=None,
|
||||
api_base=None,
|
||||
)
|
||||
AnthropicModelInfo.get_auth_header(api_base=configured_base)
|
||||
AnthropicModelInfo.get_auth_header(api_base=configured_base, allow_workload_identity=True)
|
||||
|
||||
assert [url for (url, _, _) in poster.requests] == ["https://gw.example.com/v1/oauth/token"]
|
||||
|
||||
|
|
@ -2569,7 +2569,7 @@ class TestWifAsyncSeam:
|
|||
for name, value in WIF_ENV.items():
|
||||
monkeypatch.setenv(name, value)
|
||||
|
||||
result = await AnthropicModelInfo.aget_auth_header()
|
||||
result = await AnthropicModelInfo.aget_auth_header(allow_workload_identity=True)
|
||||
|
||||
assert result == {
|
||||
"authorization": f"Bearer {FAKE_MINTED_TOKEN}",
|
||||
|
|
@ -2842,6 +2842,7 @@ class TestWifRespxEndToEnd:
|
|||
)
|
||||
)
|
||||
result = AnthropicModelInfo.get_auth_header(
|
||||
allow_workload_identity=True,
|
||||
litellm_params={
|
||||
"anthropic_federation_rule_id": "fdrl_e2e",
|
||||
"anthropic_organization_id": "org-e2e",
|
||||
|
|
@ -2856,3 +2857,119 @@ class TestWifRespxEndToEnd:
|
|||
assert token_route.call_count == 1
|
||||
exchange_body = json.loads(token_route.calls[0].request.content)
|
||||
assert exchange_body["federation_rule_id"] == "fdrl_e2e"
|
||||
|
||||
|
||||
class TestWifProviderAllowlist:
|
||||
"""A federation token is an Anthropic-org credential, and the exchange POSTs the workload's OIDC
|
||||
assertion to the deployment's own api_base host. Providers that subclass the Anthropic config for
|
||||
their own endpoints must therefore never reach the WIF tier, even when it is configured purely
|
||||
through ANTHROPIC_* environment variables."""
|
||||
|
||||
@staticmethod
|
||||
def _env_only_wif(monkeypatch) -> None: # noqa: D401
|
||||
monkeypatch.setenv("ANTHROPIC_FEDERATION_RULE_ID", "fdrl_prod")
|
||||
monkeypatch.setenv("ANTHROPIC_ORGANIZATION_ID", "org-prod-uuid")
|
||||
monkeypatch.setenv("ANTHROPIC_IDENTITY_TOKEN", "oidc/env/WIF_TEST_JWT")
|
||||
monkeypatch.setenv("WIF_TEST_JWT", "jwt-assertion-value")
|
||||
|
||||
def test_vertex_anthropic_never_mints_or_sends_the_assertion(self, monkeypatch, wif_engine):
|
||||
from litellm.llms.vertex_ai.vertex_ai_partner_models.anthropic.transformation import (
|
||||
VertexAIAnthropicConfig,
|
||||
)
|
||||
|
||||
import litellm
|
||||
|
||||
poster, calls = wif_engine
|
||||
self._env_only_wif(monkeypatch)
|
||||
|
||||
with pytest.raises(litellm.AuthenticationError):
|
||||
VertexAIAnthropicConfig().validate_environment(
|
||||
headers={},
|
||||
model="claude-sonnet-4-5",
|
||||
messages=[{"role": "user", "content": "hi"}],
|
||||
optional_params={},
|
||||
litellm_params={},
|
||||
api_key=None,
|
||||
api_base="https://us-east5-aiplatform.googleapis.com/v1/projects/p/locations/us-east5",
|
||||
)
|
||||
|
||||
assert calls == []
|
||||
assert poster.requests == []
|
||||
|
||||
def test_anthropic_itself_still_mints(self, monkeypatch, wif_engine):
|
||||
from litellm.llms.anthropic.chat.transformation import AnthropicConfig
|
||||
|
||||
poster, calls = wif_engine
|
||||
self._env_only_wif(monkeypatch)
|
||||
|
||||
headers = AnthropicConfig().validate_environment(
|
||||
headers={},
|
||||
model="claude-sonnet-4-5",
|
||||
messages=[{"role": "user", "content": "hi"}],
|
||||
optional_params={},
|
||||
litellm_params={},
|
||||
api_key=None,
|
||||
api_base=None,
|
||||
)
|
||||
|
||||
assert headers["authorization"] == f"Bearer {FAKE_MINTED_TOKEN}"
|
||||
assert len(poster.requests) == 1
|
||||
|
||||
def test_auth_header_facade_defaults_to_refusing_to_mint(self, monkeypatch, clean_anthropic_env):
|
||||
"""The facade is reachable from provider code that has nothing to do with Anthropic, so a
|
||||
caller must state that it authenticates against Anthropic's own API."""
|
||||
from litellm.llms.anthropic.common_utils import AnthropicModelInfo
|
||||
|
||||
self._env_only_wif(monkeypatch)
|
||||
|
||||
assert AnthropicModelInfo.get_auth_header(None) is None
|
||||
assert AnthropicModelInfo.get_auth_header(None, allow_workload_identity=False) is None
|
||||
|
||||
def test_eligibility_is_not_inherited_by_a_new_subclass(self):
|
||||
"""A provider added later by subclassing the Anthropic config must not inherit the right to
|
||||
mint an Anthropic-org credential against its own host."""
|
||||
from litellm.llms.anthropic.chat.transformation import AnthropicConfig
|
||||
from litellm.llms.anthropic.common_utils import config_allows_workload_identity
|
||||
|
||||
class NewCompatibleProvider(AnthropicConfig):
|
||||
pass
|
||||
|
||||
assert config_allows_workload_identity(AnthropicConfig()) is True
|
||||
assert config_allows_workload_identity(NewCompatibleProvider()) is False
|
||||
|
||||
def test_model_discovery_gates_on_the_instance(self, monkeypatch, wif_engine):
|
||||
"""get_models is inherited, so it must consult the instance rather than trusting its caller."""
|
||||
from litellm.llms.vertex_ai.vertex_ai_partner_models.anthropic.transformation import (
|
||||
VertexAIAnthropicConfig,
|
||||
)
|
||||
|
||||
poster, calls = wif_engine
|
||||
self._env_only_wif(monkeypatch)
|
||||
|
||||
with pytest.raises(ValueError, match="ANTHROPIC_API_KEY"):
|
||||
VertexAIAnthropicConfig().get_models(
|
||||
api_base="https://us-east5-aiplatform.googleapis.com/v1/projects/p/locations/us-east5"
|
||||
)
|
||||
|
||||
assert poster.requests == []
|
||||
|
||||
|
||||
class TestWifExchangeTransportHardening:
|
||||
def test_token_exchange_client_does_not_follow_redirects(self):
|
||||
"""Only the initial token URL is validated, so a 3xx must not be allowed to replay the
|
||||
assertion to an origin that was never checked."""
|
||||
from litellm.llms.base_llm.auth.token_exchange import _HttpxSyncTokenPoster
|
||||
|
||||
handler = _HttpxSyncTokenPoster()._handler_instance()
|
||||
|
||||
assert handler.client.follow_redirects is False
|
||||
|
||||
|
||||
class TestWifParamsAreNotClientSettable:
|
||||
def test_every_wif_param_is_banned_from_request_bodies(self):
|
||||
"""These fields choose which server-side secret is read and, with api_base, where it is sent,
|
||||
so a caller-supplied value would be an exfiltration primitive."""
|
||||
from litellm.proxy.auth.auth_utils import _BANNED_REQUEST_BODY_PARAMS
|
||||
from litellm.types.utils import anthropic_wif_litellm_params
|
||||
|
||||
assert set(anthropic_wif_litellm_params) <= set(_BANNED_REQUEST_BODY_PARAMS)
|
||||
|
|
|
|||
Loading…
Add table
Reference in a new issue