fix(proxy): require admin opt-in for request-body bedrock_tags

Caller-supplied bedrock_tags land as AWS resource tags under the proxy's
AWS identity, letting an authenticated caller forge ownership or
cost-allocation labels. Add bedrock_tags to _BANNED_REQUEST_BODY_PARAMS
so per-request tags need general_settings.allow_client_side_credentials
or configurable_clientside_auth_params on the deployment, matching the
aws_bedrock_project_id precedent. Deployment-level bedrock_tags in
litellm_params are unaffected.

Also stop an explicit empty bedrock_tags list in litellm_params from
falling through to optional_params
This commit is contained in:
mateo-berri 2026-07-20 15:00:41 -07:00
parent 249e1f8ce7
commit c2bd8699be
4 changed files with 107 additions and 1 deletions

View file

@ -215,7 +215,8 @@ class BedrockBatchesConfig(BaseAWSLLM, BaseBatchesConfig):
"roleArn": role_arn,
}
bedrock_tags = litellm_params.get("bedrock_tags") or optional_params.get("bedrock_tags")
config_bedrock_tags = litellm_params.get("bedrock_tags")
bedrock_tags = config_bedrock_tags if config_bedrock_tags is not None else optional_params.get("bedrock_tags")
if bedrock_tags is not None:
bedrock_request["tags"] = _validate_bedrock_tags(bedrock_tags)

View file

@ -273,6 +273,7 @@ _BANNED_REQUEST_BODY_PARAMS: 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",
"bedrock_tags",
# Provider-specific endpoint overrides that flow into the outbound
# request via ``optional_params``. Same threat as ``api_base``:
# ``s3_endpoint_url`` redirects Bedrock file uploads to attacker

View file

@ -298,6 +298,25 @@ def test_create_request_forwards_bedrock_tags_from_optional_params(config):
assert mock_sign.call_args.kwargs["data"]["tags"] == tags
def test_create_request_empty_litellm_params_tags_do_not_fall_through(config):
with patch.object(
config.common_utils,
"generate_unique_job_name",
return_value="litellm-batch-1",
), patch.object(config.common_utils, "sign_aws_request") as mock_sign:
mock_sign.return_value = ({}, b"{}")
config.transform_create_batch_request(
model="m",
create_batch_data={"input_file_id": "s3://b/in.jsonl"},
optional_params={"bedrock_tags": [{"key": "env", "value": "prod"}]},
litellm_params={
"aws_batch_role_arn": "arn:aws:iam::1:role/r",
"bedrock_tags": [],
},
)
assert mock_sign.call_args.kwargs["data"]["tags"] == []
def test_create_request_omits_tags_when_bedrock_tags_absent(config):
with patch.object(
config.common_utils,

View file

@ -1944,6 +1944,91 @@ class TestIsRequestBodySafeBlocksRivaUseSsl:
)
class TestIsRequestBodySafeBlocksBedrockTags:
"""``bedrock_tags`` lands as AWS resource tags on Bedrock batch jobs
created with the proxy's AWS identity, so a caller-supplied value can
forge ownership or cost-allocation labels; like
``aws_bedrock_project_id`` it is blocked without an admin opt-in."""
def test_bedrock_tags_in_request_body_is_rejected(self):
with pytest.raises(ValueError, match="bedrock_tags"):
is_request_body_safe(
request_body={
"model": "bedrock-batch-opus",
"bedrock_tags": [{"key": "application", "value": "genai-proxy"}],
},
general_settings={},
llm_router=None,
model="bedrock-batch-opus",
)
def test_admin_opt_in_proxy_wide_allows_bedrock_tags(self):
assert (
is_request_body_safe(
request_body={
"model": "bedrock-batch-opus",
"bedrock_tags": [{"key": "application", "value": "genai-proxy"}],
},
general_settings={"allow_client_side_credentials": True},
llm_router=None,
model="bedrock-batch-opus",
)
is True
)
def test_admin_opt_in_per_deployment_allows_bedrock_tags(self):
from litellm import Router
router = Router(
model_list=[
{
"model_name": "bedrock-batch-opus",
"litellm_params": {
"model": "bedrock/us.anthropic.claude-opus-4-7",
"configurable_clientside_auth_params": ["bedrock_tags"],
},
}
]
)
assert (
is_request_body_safe(
request_body={
"model": "bedrock-batch-opus",
"bedrock_tags": [{"key": "application", "value": "genai-proxy"}],
},
general_settings={},
llm_router=router,
model="bedrock-batch-opus",
)
is True
)
def test_per_deployment_opt_in_for_other_param_still_rejects_bedrock_tags(self):
from litellm import Router
router = Router(
model_list=[
{
"model_name": "bedrock-batch-opus",
"litellm_params": {
"model": "bedrock/us.anthropic.claude-opus-4-7",
"configurable_clientside_auth_params": ["api_base"],
},
}
]
)
with pytest.raises(ValueError, match="bedrock_tags"):
is_request_body_safe(
request_body={
"model": "bedrock-batch-opus",
"bedrock_tags": [{"key": "application", "value": "genai-proxy"}],
},
general_settings={},
llm_router=router,
model="bedrock-batch-opus",
)
# ── is_request_body_safe nested-config recursion (VERIA-6) ────────────────────