mirror of
https://github.com/BerriAI/litellm.git
synced 2026-09-12 23:01:41 +00:00
fix(proxy): stop forwarding a client Anthropic OAuth token to Bedrock and Vertex
add_provider_specific_headers_to_request tagged the client's Authorization header with the same provider list as anthropic-beta and anthropic-version, so an sk-ant-oat subscription token was sent to AWS Bedrock and Google Vertex AI as well. On Bedrock it replaced the SigV4 signature, or the deployment's own API key, and AWS answered 403 "Invalid API Key format". On Vertex it went out as a second Authorization header next to the Google one and Google answered 401 ACCESS_TOKEN_TYPE_UNSUPPORTED. The credential and those API headers need different scopes, so a request can now carry more than one ProviderSpecificHeader entry. The API headers keep the provider list they already had and the credential gets its own entry scoped to anthropic alone. get_provider_specific_headers takes either a single entry or a sequence and merges only the entries whose provider list matches, so callers that pass one entry keep working unchanged. Bedrock SigV4 signing and the deliberate extra_headers Authorization pass-through in _sign_request are left alone.
This commit is contained in:
mateo-berri
parent
3029f7eb84
commit
48aba5f103
8 changed files with 312 additions and 40 deletions
|
|
@ -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
|
||||
|
|
|
|||
|
|
@ -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(
|
||||
|
|
|
|||
|
|
@ -5091,7 +5091,10 @@ 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)
|
||||
|
|
|
|||
|
|
@ -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):
|
||||
|
|
|
|||
|
|
@ -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 == {}
|
||||
|
|
|
|||
|
|
@ -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."""
|
||||
|
|
|
|||
|
|
@ -0,0 +1,211 @@
|
|||
"""A client-supplied Anthropic OAuth credential must only ever reach Anthropic.
|
||||
|
||||
The proxy forwards a caller's ``Authorization: Bearer sk-ant-oat...`` upstream so an
|
||||
Anthropic subscription keeps working through LiteLLM. That credential is meaningless to
|
||||
AWS Bedrock and Google Vertex AI, and sending it there both breaks the request and hands
|
||||
a third-party cloud a credential it has no business holding. These tests pin the scope of
|
||||
that credential from the proxy pre-call path all the way into the headers each provider
|
||||
actually signs and sends.
|
||||
"""
|
||||
|
||||
import json
|
||||
import os
|
||||
import sys
|
||||
from unittest.mock import patch
|
||||
|
||||
import pytest
|
||||
from botocore.credentials import Credentials
|
||||
|
||||
sys.path.insert(0, os.path.abspath("../../.."))
|
||||
|
||||
from litellm.litellm_core_utils.get_provider_specific_headers import (
|
||||
ProviderSpecificHeaderUtils,
|
||||
)
|
||||
from litellm.llms.bedrock.base_aws_llm import BaseAWSLLM
|
||||
from litellm.proxy.litellm_pre_call_utils import (
|
||||
add_provider_specific_headers_to_request,
|
||||
)
|
||||
|
||||
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
|
||||
):
|
||||
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]
|
||||
|
|
@ -6334,7 +6334,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
|
||||
|
|
|
|||
Loading…
Add table
Reference in a new issue