scriptoriumadmin/litellm
1
0
Fork 0
mirror of https://github.com/BerriAI/litellm.git synced 2026-08-28 05:25:59 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 48

Merge pull request #37905 from BerriAI/litellm_fix_oauth_credential_forwarding

Browse source 
fix(proxy): stop forwarding a client Anthropic OAuth token to Bedrock and Vertex
This commit is contained in:
Mateo Wang 2026-08-22 09:03:24 -07:00 • committed by GitHub
commit d6f5ea2b37
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
8 changed files with 345 additions and 42 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

24
litellm/litellm_core_utils/get_provider_specific_headers.py
View file

@ -1,3 +1,4 @@
from collections.abc import Sequence
from typing import Final
from litellm.types.utils import ProviderSpecificHeader
@ -6,13 +7,17 @@ from litellm.types.utils import ProviderSpecificHeader
class ProviderSpecificHeaderUtils:
@staticmethod
def get_provider_specific_headers(
provider_specific_header: ProviderSpecificHeader | None,
provider_specific_header: ProviderSpecificHeader | Sequence[ProviderSpecificHeader] | None,
custom_llm_provider: str | None,
) -> dict:
"""
Get the provider specific headers for the given custom llm provider.
Supports comma-separated provider lists for headers that work across multiple providers.
Accepts either a single ProviderSpecificHeader or a sequence of them. Each entry
carries its own comma-separated provider list, so headers that are safe for several
providers and headers that are safe for exactly one can travel on the same request
without sharing a scope. Entries whose provider list does not contain
`custom_llm_provider` contribute nothing.
Returns:
Dict: The provider specific headers for the given custom llm provider
@ -20,10 +25,15 @@ class ProviderSpecificHeaderUtils:
if provider_specific_header is None or custom_llm_provider is None:
return {}
stored_providers: Final = provider_specific_header.get("custom_llm_provider", "")
provider_list: Final = [p.strip() for p in stored_providers.split(",")]
scoped_headers: Final = (
(provider_specific_header,) if isinstance(provider_specific_header, dict) else provider_specific_header
)
if custom_llm_provider in provider_list:
return provider_specific_header.get("extra_headers", {})
matched_headers: Final = {}
for scoped_header in scoped_headers:
stored_providers = scoped_header.get("custom_llm_provider", "")
provider_list = [p.strip() for p in stored_providers.split(",")]
if custom_llm_provider in provider_list:
matched_headers.update(scoped_header.get("extra_headers", {}))
return {}
return matched_headers

2
litellm/llms/custom_httpx/llm_http_handler.py
View file

@ -2056,7 +2056,7 @@ class BaseLLMHTTPHandler:
# Prepare headers
kwargs = kwargs or {}
provider_specific_header: Final = cast(
litellm.types.utils.ProviderSpecificHeader | None,
litellm.types.utils.ProviderSpecificHeader | Sequence[litellm.types.utils.ProviderSpecificHeader] | None,
kwargs.get("provider_specific_header", None),
)
provider_specific_headers: Final = ProviderSpecificHeaderUtils.get_provider_specific_headers(

8
litellm/main.py
View file

@ -5091,14 +5091,16 @@ def completion(
model_info: Final = kwargs.get("model_info", None)
proxy_server_request: Final = kwargs.get("proxy_server_request", None)
fallbacks = kwargs.get("fallbacks", None)
provider_specific_header: Final = cast(ProviderSpecificHeader | None, kwargs.get("provider_specific_header", None))
provider_specific_header: Final = cast(
ProviderSpecificHeader | Sequence[ProviderSpecificHeader] | None,
kwargs.get("provider_specific_header", None),
)
headers = kwargs.get("headers", None) or extra_headers
ensure_alternating_roles: Final[bool | None] = kwargs.get("ensure_alternating_roles", None)
user_continue_message: Final[ChatCompletionUserMessage | None] = kwargs.get("user_continue_message", None)
assistant_continue_message: ChatCompletionAssistantMessage | None = kwargs.get("assistant_continue_message", None)
if headers is None:
headers = {}
headers = {} if headers is None else dict(headers)
if extra_headers is not None:
headers.update(extra_headers)
# Inject proxy auth headers if configured

44
litellm/proxy/litellm_pre_call_utils.py
View file

@ -3044,36 +3044,36 @@ async def add_guardrails_from_policy_engine(
)
_ANTHROPIC_API_HEADER_PROVIDERS: Final = ",".join(
(LlmProviders.ANTHROPIC.value, LlmProviders.BEDROCK.value, LlmProviders.VERTEX_AI.value)
)
_ANTHROPIC_OAUTH_CREDENTIAL_PROVIDERS: Final = LlmProviders.ANTHROPIC.value
def add_provider_specific_headers_to_request(
data: dict,
headers: dict,
):
from litellm.llms.anthropic.common_utils import is_anthropic_oauth_key
anthropic_headers: Final = {}
# boolean to indicate if a header was added
added_header = False
for header in ANTHROPIC_API_HEADERS:
if header in headers:
header_value = headers[header]
anthropic_headers[header] = header_value
added_header = True
anthropic_api_headers: Final = {header: headers[header] for header in ANTHROPIC_API_HEADERS if header in headers}
anthropic_oauth_credential_headers: Final = {
header: value
for header, value in headers.items()
if header.lower() == "authorization" and is_anthropic_oauth_key(value)
}
# Check for Authorization header with Anthropic OAuth token (sk-ant-oat*)
# This needs to be handled via provider-specific headers to ensure it only
# goes to Anthropic-compatible providers, not all providers in the router
for header, value in headers.items():
if header.lower() == "authorization" and is_anthropic_oauth_key(value):
anthropic_headers[header] = value
added_header = True
break
if added_header is True:
# Anthropic headers work across multiple providers
# Store as comma-separated list so retrieval can match any of them
data["provider_specific_header"] = ProviderSpecificHeader(
custom_llm_provider=f"{LlmProviders.ANTHROPIC.value},{LlmProviders.BEDROCK.value},{LlmProviders.VERTEX_AI.value}",
extra_headers=anthropic_headers,
scoped_headers: Final = [
ProviderSpecificHeader(custom_llm_provider=providers, extra_headers=extra_headers)
for providers, extra_headers in (
(_ANTHROPIC_API_HEADER_PROVIDERS, anthropic_api_headers),
(_ANTHROPIC_OAUTH_CREDENTIAL_PROVIDERS, anthropic_oauth_credential_headers),
)
if extra_headers
]
if scoped_headers:
data["provider_specific_header"] = scoped_headers[0] if len(scoped_headers) == 1 else scoped_headers
def _add_otel_traceparent_to_data(data: dict, request: Request):

34
tests/test_litellm/litellm_core_utils/test_provider_specific_headers.py
View file

@ -112,3 +112,37 @@ class TestProviderSpecificHeaderUtils:
provider_specific_header, None
)
assert result == {}
def test_get_provider_specific_headers_scopes_each_entry_independently(self):
"""Entries in a list each carry their own provider scope."""
scoped_headers: list[ProviderSpecificHeader] = [
{
"custom_llm_provider": "anthropic,bedrock,vertex_ai",
"extra_headers": {"anthropic-beta": "context-1m-2025-08-07"},
},
{
"custom_llm_provider": "anthropic",
"extra_headers": {"authorization": "Bearer sk-ant-oat01-fake-token"},
},
]
assert ProviderSpecificHeaderUtils.get_provider_specific_headers(
scoped_headers, "anthropic"
) == {
"anthropic-beta": "context-1m-2025-08-07",
"authorization": "Bearer sk-ant-oat01-fake-token",
}
assert ProviderSpecificHeaderUtils.get_provider_specific_headers(
scoped_headers, "bedrock"
) == {"anthropic-beta": "context-1m-2025-08-07"}
assert (
ProviderSpecificHeaderUtils.get_provider_specific_headers(
scoped_headers, "openai"
)
== {}
)
def test_get_provider_specific_headers_empty_list(self):
"""An empty list of scoped entries contributes nothing."""
result = ProviderSpecificHeaderUtils.get_provider_specific_headers([], "anthropic")
assert result == {}

20
tests/test_litellm/llms/anthropic/test_anthropic_common_utils.py
View file

@ -554,7 +554,7 @@ class TestProxyOAuthHeaderForwarding:
def test_add_provider_specific_headers_forwards_oauth(self):
"""add_provider_specific_headers_to_request should forward OAuth Authorization
as a ProviderSpecificHeader scoped to Anthropic-compatible providers."""
as a ProviderSpecificHeader scoped to Anthropic and nothing else."""
from litellm.proxy.litellm_pre_call_utils import (
add_provider_specific_headers_to_request,
)
@ -569,9 +569,7 @@ class TestProxyOAuthHeaderForwarding:
assert "provider_specific_header" in data
psh = data["provider_specific_header"]
assert "anthropic" in psh["custom_llm_provider"]
assert "bedrock" in psh["custom_llm_provider"]
assert "vertex_ai" in psh["custom_llm_provider"]
assert psh["custom_llm_provider"] == "anthropic"
assert psh["extra_headers"]["authorization"] == f"Bearer {FAKE_OAUTH_TOKEN}"
def test_add_provider_specific_headers_ignores_non_oauth(self):
@ -593,7 +591,10 @@ class TestProxyOAuthHeaderForwarding:
def test_add_provider_specific_headers_combines_anthropic_and_oauth(self):
"""When both anthropic-beta and OAuth Authorization are present, both
should be included in the ProviderSpecificHeader."""
reach Anthropic."""
from litellm.litellm_core_utils.get_provider_specific_headers import (
ProviderSpecificHeaderUtils,
)
from litellm.proxy.litellm_pre_call_utils import (
add_provider_specific_headers_to_request,
)
@ -608,9 +609,12 @@ class TestProxyOAuthHeaderForwarding:
add_provider_specific_headers_to_request(data=data, headers=headers)
assert "provider_specific_header" in data
psh = data["provider_specific_header"]
assert psh["extra_headers"]["authorization"] == f"Bearer {FAKE_OAUTH_TOKEN}"
assert psh["extra_headers"]["anthropic-beta"] == "oauth-2025-04-20"
anthropic_headers = ProviderSpecificHeaderUtils.get_provider_specific_headers(
provider_specific_header=data["provider_specific_header"],
custom_llm_provider="anthropic",
)
assert anthropic_headers["authorization"] == f"Bearer {FAKE_OAUTH_TOKEN}"
assert anthropic_headers["anthropic-beta"] == "oauth-2025-04-20"
def test_clean_headers_forwards_x_api_key_when_authenticated_with_litellm_key(self):
"""clean_headers should forward x-api-key when user authenticated with x-litellm-api-key and forward_llm_provider_auth_headers=True."""

208
tests/test_litellm/proxy/test_litellm_pre_call_utils.py
View file

@ -6,6 +6,7 @@ import sys
from unittest.mock import AsyncMock, MagicMock, patch
import pytest
from botocore.credentials import Credentials
from fastapi import Request
from pydantic import ValidationError as PydanticValidationError
from starlette.datastructures import Headers
@ -26,12 +27,17 @@ from litellm.proxy.litellm_pre_call_utils import (
_update_model_if_key_alias_exists,
add_guardrails_from_policy_engine,
add_litellm_data_to_request,
add_provider_specific_headers_to_request,
check_if_token_is_service_account,
clean_headers,
)
from litellm.litellm_core_utils.get_provider_specific_headers import (
ProviderSpecificHeaderUtils,
)
from litellm.litellm_core_utils.initialize_dynamic_callback_params import (
TRUSTED_CALLBACK_VARS_FIELD,
)
from litellm.llms.bedrock.base_aws_llm import BaseAWSLLM
from litellm.types.utils import CredentialItem
sys.path.insert(
@ -6334,7 +6340,17 @@ async def test_add_litellm_data_to_request_redacts_oauth_header_from_logging_cop
assert updated["proxy_server_request"]["headers"] is updated[metadata_variable_name]["headers"]
assert updated["provider_specific_header"]["extra_headers"]["Authorization"] == _OAUTH_TOKEN
from litellm.litellm_core_utils.get_provider_specific_headers import (
ProviderSpecificHeaderUtils,
)
assert (
ProviderSpecificHeaderUtils.get_provider_specific_headers(
provider_specific_header=updated["provider_specific_header"],
custom_llm_provider="anthropic",
)["Authorization"]
== _OAUTH_TOKEN
)
@pytest.mark.asyncio
@ -7008,3 +7024,193 @@ async def test_add_litellm_data_to_request_caller_tags_empty_when_caller_sends_n
assert updated["metadata"]["tags"] == ["key-supplied"]
assert updated["metadata"]["caller_tags"] == ()
OAUTH_TOKEN = "Bearer sk-ant-oat01-fake-subscription-token-for-testing-0123456789"
GOOGLE_ACCESS_TOKEN = "Bearer ya29.fake-google-access-token-for-testing"
BEDROCK_API_KEY = "ABSKQmVkcm9ja0FQSUtleUZvclRlc3Rpbmc="
CROSS_ACCOUNT_AUTHORIZATION = "Bearer deliberately-configured-pass-through-token"
SIGV4_PREFIX = "AWS4-HMAC-SHA256"
AUTHORIZATION_HEADER_CASINGS = ["authorization", "Authorization", "AUTHORIZATION"]
LEAK_TARGET_PROVIDERS = ["bedrock", "bedrock_converse", "vertex_ai"]
BEDROCK_ENDPOINT = (
"https://bedrock-runtime.us-west-2.amazonaws.com"
"/model/us.anthropic.claude-sonnet-4-5-20250929-v1:0/invoke"
)
BEDROCK_REGION = "us-west-2"
BEDROCK_REQUEST_DATA = {"messages": [{"role": "user", "content": "Say OK"}], "max_tokens": 32}
SIGV4_OPTIONAL_PARAMS = {
"aws_access_key_id": "AKIAIOSFODNN7EXAMPLE",
"aws_secret_access_key": "wJalrXUtnFEMI/K7MDENG/bPxRfiCYEXAMPLEKEY",
"aws_region_name": BEDROCK_REGION,
}
def _client_headers(authorization_header_name: str | None = "authorization") -> dict:
headers = {
"content-type": "application/json",
"anthropic-version": "2023-06-01",
"user-agent": "claude-cli/2.1.239",
}
if authorization_header_name is not None:
headers[authorization_header_name] = OAUTH_TOKEN
return headers
def _headers_forwarded_to(client_headers: dict, custom_llm_provider: str) -> dict:
data: dict = {}
add_provider_specific_headers_to_request(data=data, headers=client_headers)
return ProviderSpecificHeaderUtils.get_provider_specific_headers(
provider_specific_header=data.get("provider_specific_header"),
custom_llm_provider=custom_llm_provider,
)
def _authorization_values(headers) -> list:
return [value for name, value in headers.items() if name.lower() == "authorization"]
def _signed_headers_for_bedrock(request_headers: dict, api_key: str | None = None) -> dict:
with patch.dict(os.environ, {"AWS_BEARER_TOKEN_BEDROCK": ""}):
signed_headers, _ = BaseAWSLLM()._sign_request(
service_name="bedrock",
headers=request_headers,
optional_params=SIGV4_OPTIONAL_PARAMS,
request_data=BEDROCK_REQUEST_DATA,
api_base=BEDROCK_ENDPOINT,
api_key=api_key,
)
return signed_headers
def _signed_headers_component(signature: str, component: str) -> str:
for part in signature.removeprefix(SIGV4_PREFIX).split(","):
name, _, value = part.strip().partition("=")
if name == component:
return value
raise AssertionError(f"{component} missing from {signature}")
@pytest.mark.parametrize("authorization_header_name", AUTHORIZATION_HEADER_CASINGS)
@pytest.mark.parametrize("custom_llm_provider", LEAK_TARGET_PROVIDERS)
def test_oauth_credential_is_never_forwarded_to_bedrock_or_vertex(
authorization_header_name, custom_llm_provider
):
"""
A client's Anthropic OAuth credential is meaningless to AWS and Google, and sending it
there both breaks the request and hands a third-party cloud a credential it should
never hold. It must not survive the pre-call path for any non-Anthropic provider.
"""
forwarded = _headers_forwarded_to(_client_headers(authorization_header_name), custom_llm_provider)
assert _authorization_values(forwarded) == []
assert OAUTH_TOKEN not in forwarded.values()
@pytest.mark.parametrize("authorization_header_name", AUTHORIZATION_HEADER_CASINGS)
def test_oauth_credential_still_reaches_anthropic_unchanged(authorization_header_name):
forwarded = _headers_forwarded_to(_client_headers(authorization_header_name), "anthropic")
assert forwarded[authorization_header_name] == OAUTH_TOKEN
assert _authorization_values(forwarded) == [OAUTH_TOKEN]
def test_oauth_credential_entry_is_scoped_to_anthropic_alone():
data: dict = {}
add_provider_specific_headers_to_request(data=data, headers=_client_headers())
scoped_headers = data["provider_specific_header"]
if not isinstance(scoped_headers, list):
scoped_headers = [scoped_headers]
credential_entries = [
entry for entry in scoped_headers if OAUTH_TOKEN in entry["extra_headers"].values()
]
assert [entry["custom_llm_provider"] for entry in credential_entries] == ["anthropic"]
def test_no_provider_specific_header_when_client_sends_nothing_anthropic():
data: dict = {}
add_provider_specific_headers_to_request(
data=data, headers={"content-type": "application/json", "authorization": "Bearer sk-a-normal-key"}
)
assert "provider_specific_header" not in data
def test_bedrock_sigv4_signature_survives_a_client_oauth_header():
forwarded = _headers_forwarded_to(_client_headers(), "bedrock")
signed = _signed_headers_for_bedrock({"Content-Type": "application/json", **forwarded})
authorizations = _authorization_values(signed)
assert len(authorizations) == 1
assert authorizations[0].startswith(SIGV4_PREFIX)
assert signed["X-Amz-Date"]
def test_bedrock_sigv4_signing_is_unchanged_by_the_client_oauth_header():
without_oauth = _signed_headers_for_bedrock(
{"Content-Type": "application/json", **_headers_forwarded_to(_client_headers(None), "bedrock")}
)
with_oauth = _signed_headers_for_bedrock(
{"Content-Type": "application/json", **_headers_forwarded_to(_client_headers(), "bedrock")}
)
assert without_oauth["Authorization"].startswith(SIGV4_PREFIX)
assert _signed_headers_component(with_oauth["Authorization"], "SignedHeaders") == (
_signed_headers_component(without_oauth["Authorization"], "SignedHeaders")
)
def test_bedrock_get_request_headers_keeps_the_sigv4_signature():
forwarded = _headers_forwarded_to(_client_headers(), "bedrock")
with patch.dict(os.environ, {"AWS_BEARER_TOKEN_BEDROCK": ""}):
prepped = BaseAWSLLM().get_request_headers(
credentials=Credentials(
SIGV4_OPTIONAL_PARAMS["aws_access_key_id"],
SIGV4_OPTIONAL_PARAMS["aws_secret_access_key"],
),
aws_region_name=BEDROCK_REGION,
extra_headers=forwarded,
endpoint_url=BEDROCK_ENDPOINT,
data=json.dumps(BEDROCK_REQUEST_DATA),
headers={"Content-Type": "application/json", **forwarded},
)
authorizations = _authorization_values(prepped.headers)
assert len(authorizations) == 1
assert authorizations[0].startswith(SIGV4_PREFIX)
def test_bedrock_api_key_deployment_keeps_its_own_bearer_token():
forwarded = _headers_forwarded_to(_client_headers(), "bedrock")
signed = _signed_headers_for_bedrock(
{"Content-Type": "application/json", **forwarded}, api_key=BEDROCK_API_KEY
)
assert _authorization_values(signed) == [f"Bearer {BEDROCK_API_KEY}"]
def test_deliberately_configured_authorization_still_overrides_sigv4():
signed = _signed_headers_for_bedrock(
{"Content-Type": "application/json", "Authorization": CROSS_ACCOUNT_AUTHORIZATION}
)
assert _authorization_values(signed) == [CROSS_ACCOUNT_AUTHORIZATION]
def test_vertex_sends_exactly_one_authorization_header():
forwarded = _headers_forwarded_to(_client_headers(), "vertex_ai")
vertex_request_headers = {
"content-type": "application/json",
"Authorization": GOOGLE_ACCESS_TOKEN,
}
vertex_request_headers.update(forwarded)
assert _authorization_values(vertex_request_headers) == [GOOGLE_ACCESS_TOKEN]

47
tests/test_litellm/test_main.py
View file

@ -2756,6 +2756,53 @@ def test_completion_default_api_base_sends_prompt_cache_breakpoint_for_gpt_5_6()
assert request_body["extra_body"]["prompt_cache_options"] == {"mode": "explicit"}
_SUBSCRIPTION_OAUTH_CREDENTIAL = "Bearer sk-ant-oat01-fake-subscription-token-for-testing-0123456789"
def _scoped_headers_for_oauth_request():
from litellm.types.utils import ProviderSpecificHeader
return [
ProviderSpecificHeader(
custom_llm_provider="anthropic,bedrock,vertex_ai",
extra_headers={"anthropic-version": "2023-06-01"},
),
ProviderSpecificHeader(
custom_llm_provider="anthropic",
extra_headers={"authorization": _SUBSCRIPTION_OAUTH_CREDENTIAL},
),
]
def _run_anthropic_hop_with_shared_headers(shared_headers):
litellm.completion(
model="anthropic/claude-3-5-sonnet-20240620",
messages=[{"role": "user", "content": "Say OK"}],
extra_headers=shared_headers,
provider_specific_header=_scoped_headers_for_oauth_request(),
api_key="sk-fake-anthropic-key",
mock_response="OK",
)
def test_completion_does_not_mutate_caller_supplied_headers():
shared_headers = {"x-tenant": "acme"}
_run_anthropic_hop_with_shared_headers(shared_headers)
assert shared_headers == {"x-tenant": "acme"}
def test_anthropic_oauth_credential_does_not_persist_into_next_provider_hop():
shared_headers = {"x-tenant": "acme"}
_run_anthropic_hop_with_shared_headers(shared_headers)
leaked = [name for name, value in shared_headers.items() if value == _SUBSCRIPTION_OAUTH_CREDENTIAL]
assert leaked == []
assert "anthropic-version" not in shared_headers
STREAM_COST_MODEL = "gpt-4o"
STREAMED_USAGE = {"prompt_tokens": 137, "completion_tokens": 42, "total_tokens": 179}