scriptoriumadmin/litellm
1
0
Fork 0
mirror of https://github.com/BerriAI/litellm.git synced 2026-09-24 00:52:24 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 91

fix(bedrock/claude_platform): strip body params the AWS endpoint rejects (#31203)

Browse source 
* fix(bedrock/claude_platform): strip body params the AWS endpoint rejects

Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com>

* test(bedrock/claude_platform): assert exact bodies through a strict fake gateway for every workspace alias and auth mode

Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com>

* fix(proxy): block Claude Platform workspace id aliases in request bodies without admin opt-in

Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com>

---------

Co-authored-by: Venkata Donavalli <vdonavalli@live.com>
Co-authored-by: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com>
Co-authored-by: shrey kharbanda <shreshth@berri.ai>
This commit is contained in:
Mateo Wang 2026-09-22 09:57:05 -07:00 • committed by GitHub
parent 7056151b91
commit f3c920a06b
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
6 changed files with 597 additions and 94 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

61
litellm/llms/bedrock/claude_platform/common_utils.py
View file

@ -1,15 +1,76 @@
from collections.abc import Mapping
from typing import Final
import httpx
import litellm
from litellm._logging import verbose_logger
from litellm.llms.bedrock.base_aws_llm import BaseAWSLLM
from litellm.llms.bedrock.common_utils import BedrockError
from litellm.secret_managers.main import get_secret_str
from litellm.types.router import GenericLiteLLMParams
CLAUDE_PLATFORM_SERVICE_NAME: Final = "aws-external-anthropic"
CLAUDE_PLATFORM_BEDROCK_ROUTE: Final = "claude_platform/"
CLAUDE_PLATFORM_UNSUPPORTED_PARAMS_OVERRIDE_KEY: Final = "claude_platform_unsupported_params"
CLAUDE_PLATFORM_ON_AWS_NON_REQUEST_PARAMS: Final = frozenset(
{
"workspace_id",
"aws_workspace_id",
"anthropic_workspace_id",
"anthropic-workspace-id",
CLAUDE_PLATFORM_UNSUPPORTED_PARAMS_OVERRIDE_KEY,
}
)
CLAUDE_PLATFORM_ON_AWS_UNSUPPORTED_REQUEST_PARAMS: Final = frozenset({"context_management"})
def filter_claude_platform_request_body(
params: Mapping[str, object],
unsupported_override: frozenset[str] | None = None,
log_dropped: bool = True,
) -> dict[str, object]:
unsupported: Final = (
unsupported_override if unsupported_override is not None else CLAUDE_PLATFORM_ON_AWS_UNSUPPORTED_REQUEST_PARAMS
)
dropped_unsupported: Final = tuple(k for k in params if k in unsupported)
if dropped_unsupported and log_dropped:
verbose_logger.warning(
"bedrock/claude_platform: dropping unsupported Messages API param(s) %s from the request body; "
"the Claude Platform on AWS endpoint rejects unknown fields. The request will proceed without them.",
dropped_unsupported,
)
return {
k: v
for k, v in params.items()
if k not in CLAUDE_PLATFORM_ON_AWS_NON_REQUEST_PARAMS and k not in unsupported and not k.startswith("aws_")
}
def resolve_unsupported_override(
litellm_params: Mapping[str, object] | GenericLiteLLMParams,
optional_params: Mapping[str, object] | None = None,
log_invalid: bool = True,
) -> frozenset[str] | None:
from_optional: Final = (optional_params or {}).get(CLAUDE_PLATFORM_UNSUPPORTED_PARAMS_OVERRIDE_KEY)
raw: Final = (
from_optional
if from_optional is not None
else litellm_params.get(CLAUDE_PLATFORM_UNSUPPORTED_PARAMS_OVERRIDE_KEY)
)
if raw is None:
return None
if isinstance(raw, (list, set, frozenset, tuple)):
return frozenset(str(item) for item in raw)
if log_invalid:
verbose_logger.warning(
"bedrock/claude_platform: ignoring claude_platform_unsupported_params of type %s; "
"expected a list of param names. Using the default unsupported-param set.",
type(raw).__name__,
)
return None
def strip_claude_platform_route(model: str) -> str:
if model.startswith(CLAUDE_PLATFORM_BEDROCK_ROUTE):

41
litellm/llms/bedrock/claude_platform/messages_transformation.py
View file

@ -8,7 +8,12 @@ from litellm.llms.anthropic.experimental_pass_through.messages.transformation im
from litellm.secret_managers.main import get_secret_str
from litellm.types.router import GenericLiteLLMParams
from .common_utils import BedrockClaudePlatformMixin, strip_claude_platform_route
from .common_utils import (
BedrockClaudePlatformMixin,
filter_claude_platform_request_body,
resolve_unsupported_override,
strip_claude_platform_route,
)
class BedrockClaudePlatformMessagesConfig(BedrockClaudePlatformMixin, AnthropicMessagesConfig):
@ -46,26 +51,38 @@ class BedrockClaudePlatformMessagesConfig(BedrockClaudePlatformMixin, AnthropicM
if resolved_api_key and "x-api-key" not in headers:
headers["x-api-key"] = resolved_api_key
headers = self._update_headers_with_anthropic_beta(
headers=headers,
optional_params=optional_params,
messages=messages,
return (
self._update_headers_with_anthropic_beta(
headers=headers,
optional_params=filter_claude_platform_request_body(
optional_params,
unsupported_override=resolve_unsupported_override(
litellm_params, optional_params=optional_params, log_invalid=False
),
log_dropped=False,
),
messages=messages,
),
api_base,
)
return headers, api_base
def transform_anthropic_messages_request(
self,
model: str,
messages: list[dict],
anthropic_messages_optional_request_params: dict,
messages: list[dict[str, object]],
anthropic_messages_optional_request_params: dict[str, object],
litellm_params: GenericLiteLLMParams,
headers: dict,
) -> dict:
headers: dict[str, str],
) -> dict[str, object]:
return super().transform_anthropic_messages_request(
model=strip_claude_platform_route(model),
messages=messages,
anthropic_messages_optional_request_params=anthropic_messages_optional_request_params,
anthropic_messages_optional_request_params=filter_claude_platform_request_body(
anthropic_messages_optional_request_params,
unsupported_override=resolve_unsupported_override(
litellm_params, optional_params=anthropic_messages_optional_request_params
),
),
litellm_params=litellm_params,
headers=headers,
)

25
litellm/llms/bedrock/claude_platform/transformation.py
View file

@ -5,7 +5,11 @@ from litellm.llms.anthropic.chat.transformation import AnthropicConfig
from litellm.secret_managers.main import get_secret_str
from litellm.types.llms.openai import AllMessageValues
from .common_utils import BedrockClaudePlatformMixin
from .common_utils import (
BedrockClaudePlatformMixin,
filter_claude_platform_request_body,
resolve_unsupported_override,
)
class BedrockClaudePlatformConfig(BedrockClaudePlatformMixin, AnthropicConfig):
@ -66,6 +70,25 @@ class BedrockClaudePlatformConfig(BedrockClaudePlatformMixin, AnthropicConfig):
anthropic_headers["anthropic-workspace-id"] = workspace_id
return {**headers, **anthropic_headers}
def transform_request(
self,
model: str,
messages: list[AllMessageValues],
optional_params: dict[str, object],
litellm_params: dict[str, object],
headers: dict[str, str],
) -> dict[str, object]:
return super().transform_request(
model=model,
messages=messages,
optional_params=filter_claude_platform_request_body(
optional_params,
unsupported_override=resolve_unsupported_override(litellm_params, optional_params=optional_params),
),
litellm_params=litellm_params,
headers=headers,
)
def get_model_response_iterator(
self,
streaming_response: Any,

4
litellm/proxy/auth/auth_utils.py
View file

@ -338,6 +338,10 @@ _BANNED_REQUEST_BODY_PARAMS: Final[tuple[str, ...]] = (
# re-route the request's retention and accounting to any project
# reachable with the deployment's shared AWS credentials.
"aws_bedrock_project_id",
"workspace_id",
"aws_workspace_id",
"anthropic_workspace_id",
"anthropic-workspace-id",
"bedrock_tags",
# Provider-specific endpoint overrides that flow into the outbound
# request via ``optional_params``. Same threat as ``api_base``:

536
tests/test_litellm/llms/bedrock/test_claude_platform_provider.py
View file

@ -1,9 +1,62 @@
import json
from collections.abc import Callable, Mapping
from typing import Final
from unittest.mock import MagicMock, patch
import httpx
import pytest
from botocore.credentials import Credentials
from pydantic import BaseModel, ConfigDict, ValidationError
from litellm.llms.custom_httpx.http_handler import AsyncHTTPHandler, HTTPHandler
from litellm.types.router import GenericLiteLLMParams
WORKSPACE_ALIASES: Final = ("workspace_id", "aws_workspace_id", "anthropic_workspace_id", "anthropic-workspace-id")
class ClaudePlatformMessagesBody(BaseModel):
"""Messages API fields per https://docs.anthropic.com/en/api/messages (2026-09) minus context_management, which
the AWS endpoint rejects with 400"""
model_config = ConfigDict(extra="forbid")
model: str
messages: list[dict]
max_tokens: int
system: str | list[dict] | None = None
metadata: dict | None = None
stop_sequences: list[str] | None = None
stream: bool | None = None
temperature: float | None = None
top_k: int | None = None
top_p: float | None = None
tools: list[dict] | None = None
tool_choice: dict | None = None
thinking: dict | None = None
service_tier: str | None = None
mcp_servers: list[dict] | None = None
output_format: dict | None = None
container: str | dict | None = None
def _gateway_reject(url: str, message: str) -> httpx.Response:
return httpx.Response(
status_code=400,
json={"type": "error", "error": {"type": "invalid_request_error", "message": message}},
request=httpx.Request("POST", url),
)
def _fake_claude_platform_gateway(url: str, headers: Mapping[str, str], data: bytes | str | None) -> httpx.Response:
if "anthropic-workspace-id" not in headers:
return _gateway_reject(url, "missing anthropic-workspace-id header")
if "x-api-key" not in headers and not headers.get("Authorization", "").startswith("AWS4-HMAC-SHA256 "):
return _gateway_reject(url, "missing x-api-key or SigV4 Authorization")
try:
ClaudePlatformMessagesBody.model_validate_json(data or "{}")
except ValidationError as exc:
return _gateway_reject(url, "; ".join(f"{'.'.join(map(str, e['loc']))}: {e['msg']}" for e in exc.errors()))
return _anthropic_response(url)
def _anthropic_response(url: str) -> httpx.Response:
@ -23,15 +76,39 @@ def _anthropic_response(url: str) -> httpx.Response:
)
def _capture_request(url: str, headers: dict, data: bytes | str | None) -> dict:
def _capture_request(url: str, headers: Mapping[str, str], data: bytes | str | None) -> dict:
raw_body = data.decode("utf-8") if isinstance(data, bytes) else data or "{}"
return {
"path": httpx.URL(url).path,
"headers": headers,
"headers": httpx.Headers(dict(headers)),
"body": json.loads(raw_body),
}
GatewayResponder = Callable[[str, Mapping[str, str], bytes], httpx.Response]
def _gateway_transport(requests: list[dict], respond: GatewayResponder) -> httpx.MockTransport:
def handle(request: httpx.Request) -> httpx.Response:
requests.append(_capture_request(url=str(request.url), headers=request.headers, data=request.content))
return respond(str(request.url), request.headers, request.content)
return httpx.MockTransport(handle)
def _sync_gateway_client(
requests: list[dict],
respond: GatewayResponder = _fake_claude_platform_gateway,
) -> HTTPHandler:
return HTTPHandler(client=httpx.Client(transport=_gateway_transport(requests, respond)))
def _async_gateway_client(requests: list[dict]) -> AsyncHTTPHandler:
handler: Final = AsyncHTTPHandler()
handler.client = httpx.AsyncClient(transport=_gateway_transport(requests, _fake_claude_platform_gateway))
return handler
def test_claude_platform_builds_default_messages_url_from_region():
from litellm.llms.bedrock.claude_platform.transformation import (
BedrockClaudePlatformConfig,
@ -77,9 +154,7 @@ def test_claude_platform_uses_bedrock_subroute():
import litellm
from litellm.llms.bedrock.common_utils import BedrockModelInfo
model, provider, _, _ = litellm.get_llm_provider(
model="bedrock/claude_platform/claude-sonnet-4-6"
)
model, provider, _, _ = litellm.get_llm_provider(model="bedrock/claude_platform/claude-sonnet-4-6")
assert provider == "bedrock"
assert model == "claude_platform/claude-sonnet-4-6"
@ -177,9 +252,7 @@ def test_claude_platform_sigv4_signs_transformed_request_body():
assert signed_body == json.dumps(request_body).encode()
assert headers["Authorization"] == "signed"
mock_sign_request.assert_called_once()
assert (
mock_sign_request.call_args.kwargs["service_name"] == "aws-external-anthropic"
)
assert mock_sign_request.call_args.kwargs["service_name"] == "aws-external-anthropic"
assert mock_sign_request.call_args.kwargs["request_data"] == request_body
@ -237,7 +310,7 @@ def test_bedrock_claude_platform_messages_config_round_trips_native_body():
model="claude_platform/claude-sonnet-4-6",
messages=[{"role": "user", "content": "hello"}],
anthropic_messages_optional_request_params={"max_tokens": 10},
litellm_params={},
litellm_params=GenericLiteLLMParams(),
headers=headers,
)
@ -250,69 +323,6 @@ def test_bedrock_claude_platform_messages_config_round_trips_native_body():
}
def test_chat_completion_routes_bedrock_claude_platform_to_messages_api():
import litellm
requests = []
def mock_post(self, url, data=None, headers=None, **kwargs):
requests.append(_capture_request(url=url, headers=headers or {}, data=data))
return _anthropic_response(url)
with patch("litellm.llms.custom_httpx.http_handler.HTTPHandler.post", mock_post):
response = litellm.completion(
model="bedrock/claude_platform/claude-sonnet-4-6",
messages=[{"role": "user", "content": "hello"}],
max_tokens=10,
api_base="https://aws-external-anthropic.us-west-2.api.aws",
api_key="fake-platform-key",
workspace_id="wrkspc_test",
)
assert response.choices[0].message.content == "ok"
assert len(requests) == 1
assert requests[0]["path"] == "/v1/messages"
assert requests[0]["headers"]["x-api-key"] == "fake-platform-key"
assert requests[0]["headers"]["anthropic-workspace-id"] == "wrkspc_test"
assert requests[0]["body"]["model"] == "claude-sonnet-4-6"
@pytest.mark.asyncio
async def test_anthropic_messages_routes_bedrock_claude_platform_to_messages_api():
import litellm
requests = []
async def mock_post(self, url, data=None, headers=None, **kwargs):
requests.append(_capture_request(url=url, headers=headers or {}, data=data))
return _anthropic_response(url)
try:
with patch(
"litellm.llms.custom_httpx.http_handler.AsyncHTTPHandler.post",
new=mock_post,
):
response = await litellm.anthropic_messages(
model="bedrock/claude_platform/claude-sonnet-4-6",
messages=[{"role": "user", "content": "hello"}],
max_tokens=10,
api_base="https://aws-external-anthropic.us-west-2.api.aws",
api_key="fake-platform-key",
workspace_id="wrkspc_test",
)
finally:
await litellm.close_litellm_async_clients()
assert response["content"][0]["text"] == "ok"
assert len(requests) == 1
assert requests[0]["path"] == "/v1/messages"
assert requests[0]["headers"]["x-api-key"] == "fake-platform-key"
assert requests[0]["headers"]["anthropic-workspace-id"] == "wrkspc_test"
assert requests[0]["body"]["messages"] == [{"role": "user", "content": "hello"}]
assert requests[0]["body"]["max_tokens"] == 10
assert requests[0]["body"]["model"] == "claude-sonnet-4-6"
@pytest.mark.asyncio
async def test_anthropic_messages_bedrock_claude_platform_forwards_anthropic_beta_verbatim():
import litellm
@ -348,16 +358,380 @@ async def test_anthropic_messages_bedrock_claude_platform_forwards_anthropic_bet
]
def test_sigv4_no_duplicate_content_type_when_caller_sets_lowercase():
"""
Regression: get_anthropic_headers() supplies "content-type" (lowercase).
_sign_request() used to prepend "Content-Type" (uppercase), leaving both
keys in the dict. botocore joins them into "application/json, application/json"
in the canonical string, while the wire request sends only one value → 401.
def test_claude_platform_strips_auth_params_from_request_body():
from litellm.llms.bedrock.claude_platform.transformation import (
BedrockClaudePlatformConfig,
)
Fix: prepend with lowercase "content-type" so **headers overwrites it when
the caller already set it.
"""
config = BedrockClaudePlatformConfig()
optional_params = {
"workspace_id": "wrkspc_test",
"aws_region_name": "us-west-2",
"max_tokens": 10,
}
request_body = config.transform_request(
model="claude-sonnet-4-6",
messages=[{"role": "user", "content": "hello"}],
optional_params=optional_params,
litellm_params={},
headers={},
)
assert request_body == EXPECTED_CHAT_BODY
assert optional_params == {"workspace_id": "wrkspc_test", "aws_region_name": "us-west-2", "max_tokens": 10}
def test_claude_platform_messages_strips_auth_params_from_request_body():
import litellm
from litellm.types.utils import LlmProviders
config = litellm.ProviderConfigManager.get_provider_anthropic_messages_config(
model="claude_platform/claude-sonnet-4-6",
provider=LlmProviders.BEDROCK,
)
assert config is not None
input_params = {
"workspace_id": "wrkspc_test",
"aws_region_name": "us-west-2",
"max_tokens": 10,
}
request_body = config.transform_anthropic_messages_request(
model="claude_platform/claude-sonnet-4-6",
messages=[{"role": "user", "content": "hello"}],
anthropic_messages_optional_request_params=input_params,
litellm_params=GenericLiteLLMParams(),
headers={},
)
assert request_body == {
"model": "claude-sonnet-4-6",
"messages": [{"role": "user", "content": "hello"}],
"max_tokens": 10,
}
assert input_params == {"workspace_id": "wrkspc_test", "aws_region_name": "us-west-2", "max_tokens": 10}
def test_claude_platform_strips_unsupported_context_management_param(caplog):
import logging
from litellm.llms.bedrock.claude_platform.transformation import (
BedrockClaudePlatformConfig,
)
config = BedrockClaudePlatformConfig()
optional_params = {
"workspace_id": "wrkspc_test",
"context_management": {"edits": [{"type": "clear_tool_uses_20250919"}]},
"max_tokens": 10,
}
with caplog.at_level(logging.WARNING, logger="LiteLLM"):
request_body = config.transform_request(
model="claude-sonnet-4-6",
messages=[{"role": "user", "content": "hello"}],
optional_params=optional_params,
litellm_params={},
headers={},
)
assert "context_management" not in request_body
assert request_body["max_tokens"] == 10
assert "context_management" in optional_params
assert any(
"context_management" in record.message and record.levelno == logging.WARNING for record in caplog.records
)
def test_claude_platform_messages_strips_unsupported_context_management_param():
import litellm
from litellm.types.utils import LlmProviders
config = litellm.ProviderConfigManager.get_provider_anthropic_messages_config(
model="claude_platform/claude-sonnet-4-6",
provider=LlmProviders.BEDROCK,
)
assert config is not None
input_params = {
"context_management": {"edits": [{"type": "clear_tool_uses_20250919"}]},
"max_tokens": 10,
}
request_body = config.transform_anthropic_messages_request(
model="claude_platform/claude-sonnet-4-6",
messages=[{"role": "user", "content": "hello"}],
anthropic_messages_optional_request_params=input_params,
litellm_params=GenericLiteLLMParams(),
headers={},
)
assert "context_management" not in request_body
assert request_body["max_tokens"] == 10
assert "context_management" in input_params
@pytest.mark.parametrize("override_in", ["litellm_params", "optional_params"])
def test_claude_platform_unsupported_override_allows_context_management(override_in):
from litellm.llms.bedrock.claude_platform.transformation import (
BedrockClaudePlatformConfig,
)
config = BedrockClaudePlatformConfig()
override = {"claude_platform_unsupported_params": []}
context_management = {"edits": [{"type": "clear_tool_uses_20250919"}]}
request_body = config.transform_request(
model="claude-sonnet-4-6",
messages=[{"role": "user", "content": "hello"}],
optional_params={
"context_management": context_management,
"max_tokens": 10,
**(override if override_in == "optional_params" else {}),
},
litellm_params=override if override_in == "litellm_params" else {},
headers={},
)
assert request_body == {
"model": "claude-sonnet-4-6",
"messages": [{"role": "user", "content": [{"type": "text", "text": "hello"}]}],
"max_tokens": 10,
"context_management": context_management,
}
def test_claude_platform_unsupported_override_ignores_invalid_type():
from litellm.llms.bedrock.claude_platform import common_utils
from litellm.llms.bedrock.claude_platform.transformation import (
BedrockClaudePlatformConfig,
)
config = BedrockClaudePlatformConfig()
with patch.object(common_utils.verbose_logger, "warning") as mock_warning:
request_body = config.transform_request(
model="claude-sonnet-4-6",
messages=[{"role": "user", "content": "hello"}],
optional_params={
"context_management": {"edits": [{"type": "clear_tool_uses_20250919"}]},
"max_tokens": 10,
},
litellm_params={"claude_platform_unsupported_params": "not_a_list"},
headers={},
)
assert "context_management" not in request_body
assert request_body["max_tokens"] == 10
warned = [call.args[0] for call in mock_warning.call_args_list]
assert any("claude_platform_unsupported_params" in message for message in warned)
def test_claude_platform_messages_does_not_advertise_beta_for_stripped_context_management():
import litellm
from litellm.types.utils import LlmProviders
config = litellm.ProviderConfigManager.get_provider_anthropic_messages_config(
model="claude_platform/claude-sonnet-4-6",
provider=LlmProviders.BEDROCK,
)
assert config is not None
headers, _ = config.validate_anthropic_messages_environment(
api_key="fake-platform-key",
headers={},
model="claude_platform/claude-sonnet-4-6",
messages=[{"role": "user", "content": "hello"}],
optional_params={
"max_tokens": 10,
"context_management": {"edits": [{"type": "clear_tool_uses_20250919"}]},
},
litellm_params={"workspace_id": "wrkspc_test"},
)
assert "context-management-2025-06-27" not in headers.get("anthropic-beta", "")
def test_claude_platform_messages_override_keeps_beta_for_context_management():
import litellm
from litellm.types.utils import LlmProviders
config = litellm.ProviderConfigManager.get_provider_anthropic_messages_config(
model="claude_platform/claude-sonnet-4-6",
provider=LlmProviders.BEDROCK,
)
assert config is not None
headers, _ = config.validate_anthropic_messages_environment(
api_key="fake-platform-key",
headers={},
model="claude_platform/claude-sonnet-4-6",
messages=[{"role": "user", "content": "hello"}],
optional_params={
"max_tokens": 10,
"context_management": {"edits": [{"type": "clear_tool_uses_20250919"}]},
},
litellm_params={
"workspace_id": "wrkspc_test",
"claude_platform_unsupported_params": [],
},
)
assert "context-management-2025-06-27" in headers.get("anthropic-beta", "")
def test_claude_platform_messages_unsupported_override_allows_context_management():
import litellm
from litellm.types.utils import LlmProviders
config = litellm.ProviderConfigManager.get_provider_anthropic_messages_config(
model="claude_platform/claude-sonnet-4-6",
provider=LlmProviders.BEDROCK,
)
assert config is not None
request_body = config.transform_anthropic_messages_request(
model="claude_platform/claude-sonnet-4-6",
messages=[{"role": "user", "content": "hello"}],
anthropic_messages_optional_request_params={
"context_management": {"edits": [{"type": "clear_tool_uses_20250919"}]},
"max_tokens": 10,
},
litellm_params=GenericLiteLLMParams.model_validate({"claude_platform_unsupported_params": []}),
headers={},
)
assert "context_management" in request_body
assert request_body["max_tokens"] == 10
SIGV4_KWARGS: Final = {
"aws_region_name": "us-west-2",
"aws_access_key_id": "AKIATEST",
"aws_secret_access_key": "test-secret",
"aws_session_token": "test-token",
}
API_KEY_KWARGS: Final = {"api_key": "fake-platform-key", "aws_region_name": "us-west-2"}
EXPECTED_CHAT_BODY: Final = {
"model": "claude-sonnet-4-6",
"messages": [{"role": "user", "content": [{"type": "text", "text": "hello"}]}],
"max_tokens": 10,
}
EXPECTED_NATIVE_BODY: Final = {
"model": "claude-sonnet-4-6",
"messages": [{"role": "user", "content": "hello"}],
"max_tokens": 10,
"stream": False,
}
def _assert_gateway_accepted(request: dict, auth_kwargs: dict) -> None:
assert request["path"] == "/v1/messages"
assert request["headers"]["anthropic-workspace-id"] == "wrkspc_test"
if "api_key" in auth_kwargs:
assert request["headers"]["x-api-key"] == auth_kwargs["api_key"]
assert "Authorization" not in request["headers"]
else:
assert request["headers"]["Authorization"].startswith("AWS4-HMAC-SHA256 Credential=AKIATEST/")
assert "/us-west-2/aws-external-anthropic/aws4_request" in request["headers"]["Authorization"]
@pytest.mark.parametrize("auth_kwargs", [API_KEY_KWARGS, SIGV4_KWARGS], ids=["api_key", "sigv4"])
@pytest.mark.parametrize("workspace_alias", WORKSPACE_ALIASES)
def test_chat_completion_claude_platform_sends_exact_body_through_strict_gateway(auth_kwargs, workspace_alias):
import litellm
requests: list[dict] = []
response = litellm.completion(
model="bedrock/claude_platform/claude-sonnet-4-6",
messages=[{"role": "user", "content": "hello"}],
max_tokens=10,
client=_sync_gateway_client(requests),
**{workspace_alias: "wrkspc_test"},
**auth_kwargs,
)
assert response.choices[0].message.content == "ok"
assert len(requests) == 1, requests
assert requests[0]["body"] == EXPECTED_CHAT_BODY
_assert_gateway_accepted(requests[0], auth_kwargs)
@pytest.mark.asyncio
@pytest.mark.parametrize("auth_kwargs", [API_KEY_KWARGS, SIGV4_KWARGS], ids=["api_key", "sigv4"])
@pytest.mark.parametrize("workspace_alias", WORKSPACE_ALIASES)
async def test_anthropic_messages_claude_platform_sends_exact_body_through_strict_gateway(auth_kwargs, workspace_alias):
import litellm
requests: list[dict] = []
response = await litellm.anthropic_messages(
model="bedrock/claude_platform/claude-sonnet-4-6",
messages=[{"role": "user", "content": "hello"}],
max_tokens=10,
client=_async_gateway_client(requests),
**{workspace_alias: "wrkspc_test"},
**auth_kwargs,
)
assert response["content"][0]["text"] == "ok"
assert len(requests) == 1, requests
assert requests[0]["body"] == EXPECTED_NATIVE_BODY
_assert_gateway_accepted(requests[0], auth_kwargs)
def test_chat_completion_claude_platform_drops_context_management_and_gateway_accepts():
import litellm
requests: list[dict] = []
litellm.completion(
model="bedrock/claude_platform/claude-sonnet-4-6",
messages=[{"role": "user", "content": "hello"}],
max_tokens=10,
workspace_id="wrkspc_test",
context_management={"edits": [{"type": "clear_tool_uses_20250919"}]},
client=_sync_gateway_client(requests),
**API_KEY_KWARGS,
)
assert requests[0]["body"] == EXPECTED_CHAT_BODY
def test_chat_completion_claude_platform_override_kwarg_is_honoured_and_not_sent():
import litellm
requests: list[dict] = []
context_management = {"edits": [{"type": "clear_tool_uses_20250919"}]}
litellm.completion(
model="bedrock/claude_platform/claude-sonnet-4-6",
messages=[{"role": "user", "content": "hello"}],
max_tokens=10,
workspace_id="wrkspc_test",
context_management=context_management,
claude_platform_unsupported_params=[],
client=_sync_gateway_client(requests, respond=lambda url, headers, data: _anthropic_response(url)),
**API_KEY_KWARGS,
)
assert requests[0]["body"] == {**EXPECTED_CHAT_BODY, "context_management": context_management}
def test_fake_claude_platform_gateway_rejects_leaked_internal_fields():
leaked = json.dumps({**EXPECTED_NATIVE_BODY, "workspace_id": "wrkspc_test", "aws_region_name": "us-west-2"})
response = _fake_claude_platform_gateway(
url="https://aws-external-anthropic.us-west-2.api.aws/v1/messages",
headers={"anthropic-workspace-id": "wrkspc_test", "x-api-key": "k"},
data=leaked,
)
assert response.status_code == 400
assert response.json()["error"]["message"] == (
"workspace_id: Extra inputs are not permitted; aws_region_name: Extra inputs are not permitted"
)
def test_sigv4_no_duplicate_content_type_when_caller_sets_lowercase():
from litellm.llms.bedrock.base_aws_llm import BaseAWSLLM
llm = BaseAWSLLM()

24
tests/test_litellm/proxy/auth/test_auth_utils.py
View file

@ -2727,6 +2727,30 @@ class TestIsRequestBodySafeBlocksBedrockProjectOverride:
)
class TestIsRequestBodySafeBlocksClaudePlatformWorkspaceOverride:
@pytest.mark.parametrize(
"alias", ["workspace_id", "aws_workspace_id", "anthropic_workspace_id", "anthropic-workspace-id"]
)
def test_workspace_alias_in_request_body_is_rejected(self, alias):
with pytest.raises(ValueError, match=alias):
is_request_body_safe(
request_body={"model": "gpt-4", alias: "wrkspc_attacker"},
general_settings={},
llm_router=None,
model="gpt-4",
)
def test_admin_opt_in_proxy_wide_allows_workspace_id(self):
assert (
is_request_body_safe(
request_body={"model": "gpt-4", "workspace_id": "wrkspc_byok"},
general_settings={"allow_client_side_credentials": True},
llm_router=None,
model="gpt-4",
)
is True
)
class TestIsRequestBodySafeBlocksRustOptIn:
"""``rust`` hands the whole call to the Rust core, which signs and sends
with its own HTTP client rather than the one the deployment configured, and