fix(bedrock): prevent caller AWS identity override

litellm/llms/bedrock/batches/transformation.py

@@ -28,7 +28,11 @@ from litellm.types.llms.openai import (
 from litellm.types.utils import LiteLLMBatch, LlmProviders
 
 from ..base_aws_llm import BaseAWSLLM
-from ..common_utils import CommonBatchFilesUtils, resolve_s3_encryption_key_id
+from ..common_utils import (
+    CommonBatchFilesUtils,
+    merge_bedrock_aws_request_params,
+    resolve_s3_encryption_key_id,
+)
 
 # Bedrock batch input files are uploaded as
 # s3://bucket/litellm-bedrock-files-{model, ":" -> "-"}-{uuid4}.jsonl (see
@@ -130,10 +134,7 @@ class BedrockBatchesConfig(BaseAWSLLM, BaseBatchesConfig):
         Get the complete URL for Bedrock batch creation.
         Bedrock batch jobs are created via the model invocation job API.
         """
-        request_params: Final = {
-            **litellm_params,
-            **optional_params,
-        }  # mutable-ok: merged params are read by AWS helpers
+        request_params: Final = merge_bedrock_aws_request_params(litellm_params, optional_params)
         aws_region_name: Final = self._get_aws_region_name(request_params, model)
 
         # Bedrock model invocation job endpoint
@@ -236,10 +237,7 @@ class BedrockBatchesConfig(BaseAWSLLM, BaseBatchesConfig):
 
         # For Bedrock, we need to return a pre-signed request with AWS auth headers
         # Use common utility for AWS signing
-        request_params: Final = {
-            **litellm_params,
-            **optional_params,
-        }  # mutable-ok: merged params are read by AWS helpers
+        request_params: Final = merge_bedrock_aws_request_params(litellm_params, optional_params)
         endpoint_url: Final = (
             f"https://bedrock.{self._get_aws_region_name(request_params, model)}.amazonaws.com/model-invocation-job"
         )
@@ -395,10 +393,7 @@ class BedrockBatchesConfig(BaseAWSLLM, BaseBatchesConfig):
         endpoint_url: Final = f"https://bedrock.{region}.amazonaws.com/model-invocation-job/{encoded_arn}"
 
         # Use common utility for AWS signing
-        request_params: Final = {
-            **litellm_params,
-            **optional_params,
-        }  # mutable-ok: merged params are read by AWS helpers
+        request_params: Final = merge_bedrock_aws_request_params(litellm_params, optional_params)
         signed_headers, _ = self.common_utils.sign_aws_request(
             service_name="bedrock",
             data={},  # GET request has no body

litellm/llms/bedrock/common_utils.py

@@ -36,6 +36,44 @@ class BedrockError(BaseLLMException):
     pass
 
 
+_BEDROCK_AWS_AUTH_PARAMETER_KEYS: Final[tuple[str, ...]] = (
+    "aws_access_key_id",
+    "aws_secret_access_key",
+    "aws_session_token",
+    "aws_region_name",
+    "aws_session_name",
+    "aws_profile_name",
+    "aws_role_name",
+    "aws_web_identity_token",
+    "aws_sts_endpoint",
+    "aws_external_id",
+)
+
+
+def merge_bedrock_aws_request_params(
+    litellm_params: Mapping[str, Any],
+    optional_params: Mapping[str, Any],
+) -> dict[str, Any]:
+    """Merge deployment and request parameters without allowing auth escalation.
+
+    Deployment configuration is authoritative for AWS authentication. When a
+    deployment supplies static credentials, caller-supplied profile/role/token
+    selectors must not redirect signing to another identity available on the
+    server. Requests may still provide AWS credentials when the deployment has
+    no static credentials configured.
+    """
+    request_params: Final = {**optional_params, **litellm_params}  # mutable-ok: AWS helpers require a plain dict
+    has_static_deployment_credentials = all(
+        isinstance(litellm_params.get(key), str) and bool(litellm_params.get(key))
+        for key in ("aws_access_key_id", "aws_secret_access_key", "aws_region_name")
+    )
+    if has_static_deployment_credentials:
+        for key in _BEDROCK_AWS_AUTH_PARAMETER_KEYS:
+            if key not in litellm_params:
+                request_params.pop(key, None)
+    return request_params
+
+
 # Lazy import cache to avoid circular imports and performance impact
 _get_model_info = None
 

litellm/llms/bedrock/files/transformation.py

@@ -46,7 +46,7 @@ from litellm.types.utils import ExtractedFileData, LlmProviders, SpecialEnums
 from litellm.utils import get_llm_provider
 
 from ..base_aws_llm import BaseAWSLLM
-from ..common_utils import BedrockError, resolve_s3_encryption_key_id
+from ..common_utils import BedrockError, merge_bedrock_aws_request_params, resolve_s3_encryption_key_id
 
 # litellm_params key used to hand the SigV4-signed GET headers from
 # `transform_file_content_request` to `validate_environment` (the only hook
@@ -257,10 +257,7 @@ class BedrockFilesConfig(BaseAWSLLM, BaseFilesConfig):
         """
         Get the complete S3 URL for the file upload request
         """
-        request_params: Final = {
-            **litellm_params,
-            **optional_params,
-        }  # mutable-ok: merged params are read by AWS helpers
+        request_params: Final = merge_bedrock_aws_request_params(litellm_params, optional_params)
         bucket_name = litellm_params.get("s3_bucket_name") or os.getenv("AWS_S3_BUCKET_NAME")
         if not bucket_name:
             raise ValueError(
@@ -732,10 +729,7 @@ class BedrockFilesConfig(BaseAWSLLM, BaseFilesConfig):
         if s3_region_name:
             optional_params = {**optional_params, "aws_region_name": s3_region_name}
 
-        request_params: Final = {
-            **litellm_params,
-            **optional_params,
-        }  # mutable-ok: merged params are read by AWS helpers
+        request_params: Final = merge_bedrock_aws_request_params(litellm_params, optional_params)
 
         # Sign the request and return a pre-signed request object
         signed_headers, signed_body = self._sign_s3_request(

tests/batches_tests/test_bedrock_files_and_batches.py

@@ -389,3 +389,133 @@ def test_bedrock_batch_with_encryption_key_in_post_request():
     )
 
     print("SUCCESS: s3_encryption_key_id properly included in AWS POST request")
+
+
+def test_bedrock_file_upload_signing_uses_deployment_credentials(monkeypatch):
+    from litellm.llms.bedrock.files.transformation import BedrockFilesConfig
+
+    config = BedrockFilesConfig()
+    captured = {}
+
+    def capture_signing(**kwargs):
+        captured.update(kwargs)
+        return {}, ""
+
+    monkeypatch.setattr(config, "_sign_s3_request", capture_signing)
+
+    result = config.transform_create_file_request(
+        model="",
+        create_file_data={
+            "file": (
+                "batch.jsonl",
+                b'{"custom_id":"req-1","body":{"model":"bedrock/model"}}\n',
+                "application/jsonl",
+            ),
+            "purpose": "batch",
+        },
+        optional_params={},
+        litellm_params={
+            "s3_bucket_name": "deployment-bucket",
+            "aws_access_key_id": "deployment-access-key",
+            "aws_secret_access_key": "deployment-secret",
+            "aws_region_name": "eu-west-1",
+        },
+    )
+
+    assert "eu-west-1" in result["url"]
+    assert captured["optional_params"]["aws_access_key_id"] == "deployment-access-key"
+    assert captured["optional_params"]["aws_secret_access_key"] == "deployment-secret"
+    assert captured["optional_params"]["aws_region_name"] == "eu-west-1"
+
+
+def test_bedrock_batch_signing_uses_deployment_credentials(monkeypatch):
+    from litellm.llms.bedrock.batches.transformation import BedrockBatchesConfig
+
+    config = BedrockBatchesConfig()
+    captured = {}
+
+    def capture_signing(**kwargs):
+        captured.update(kwargs)
+        return {}, b"{}"
+
+    monkeypatch.setattr(config.common_utils, "sign_aws_request", capture_signing)
+
+    result = config.transform_create_batch_request(
+        model="us.anthropic.claude-haiku-4-5-20251001-v1:0",
+        create_batch_data={
+            "input_file_id": "s3://deployment-bucket/input.jsonl",
+            "completion_window": "24h",
+            "endpoint": "/v1/chat/completions",
+        },
+        optional_params={},
+        litellm_params={
+            "aws_access_key_id": "deployment-access-key",
+            "aws_secret_access_key": "deployment-secret",
+            "aws_region_name": "eu-west-1",
+            "aws_batch_role_arn": "arn:aws:iam::123456789012:role/bedrock-batch",
+        },
+    )
+
+    assert result["url"].startswith("https://bedrock.eu-west-1.amazonaws.com/")
+    assert captured["optional_params"]["aws_access_key_id"] == "deployment-access-key"
+    assert captured["optional_params"]["aws_secret_access_key"] == "deployment-secret"
+    assert captured["optional_params"]["aws_region_name"] == "eu-west-1"
+
+
+def test_bedrock_batch_retrieval_signing_uses_deployment_credentials(monkeypatch):
+    from litellm.llms.bedrock.batches.transformation import BedrockBatchesConfig
+
+    config = BedrockBatchesConfig()
+    captured = {}
+
+    def capture_signing(**kwargs):
+        captured.update(kwargs)
+        return {}, b""
+
+    monkeypatch.setattr(config.common_utils, "sign_aws_request", capture_signing)
+
+    result = config.transform_retrieve_batch_request(
+        batch_id="arn:aws:bedrock:eu-west-1:123456789012:model-invocation-job/job-1",
+        optional_params={},
+        litellm_params={
+            "aws_access_key_id": "deployment-access-key",
+            "aws_secret_access_key": "deployment-secret",
+            "aws_region_name": "eu-west-1",
+        },
+    )
+
+    assert result["url"].startswith("https://bedrock.eu-west-1.amazonaws.com/")
+    assert captured["optional_params"]["aws_access_key_id"] == "deployment-access-key"
+    assert captured["optional_params"]["aws_secret_access_key"] == "deployment-secret"
+    assert captured["optional_params"]["aws_region_name"] == "eu-west-1"
+
+
+def test_bedrock_deployment_credentials_block_caller_profile_override(monkeypatch):
+    from litellm.llms.bedrock.batches.transformation import BedrockBatchesConfig
+
+    config = BedrockBatchesConfig()
+    captured = {}
+
+    def capture_signing(**kwargs):
+        captured.update(kwargs)
+        return {}, b"{}"
+
+    monkeypatch.setattr(config.common_utils, "sign_aws_request", capture_signing)
+
+    config.transform_create_batch_request(
+        model="us.anthropic.claude-haiku-4-5-20251001-v1:0",
+        create_batch_data={
+            "input_file_id": "s3://deployment-bucket/input.jsonl",
+            "completion_window": "24h",
+        },
+        optional_params={"aws_profile_name": "caller-controlled-profile"},
+        litellm_params={
+            "aws_access_key_id": "deployment-access-key",
+            "aws_secret_access_key": "deployment-secret",
+            "aws_region_name": "eu-west-1",
+            "aws_batch_role_arn": "arn:aws:iam::123456789012:role/bedrock-batch",
+        },
+    )
+
+    assert "aws_profile_name" not in captured["optional_params"]
+    assert captured["optional_params"]["aws_access_key_id"] == "deployment-access-key"

tests/litellm/llms/bedrock/test_bedrock_request_credentials.py

@@ -1,95 +0,0 @@
-from litellm.llms.bedrock.batches.transformation import BedrockBatchesConfig
-from litellm.llms.bedrock.files.transformation import BedrockFilesConfig
-
-
-def test_bedrock_file_upload_signing_uses_deployment_credentials(monkeypatch):
-    config = BedrockFilesConfig()
-    captured = {}
-
-    def capture_signing(**kwargs):
-        captured.update(kwargs)
-        return {}, ""
-
-    monkeypatch.setattr(config, "_sign_s3_request", capture_signing)
-
-    result = config.transform_create_file_request(
-        model="",
-        create_file_data={
-            "file": (
-                "batch.jsonl",
-                b'{"custom_id":"req-1","body":{"model":"bedrock/model"}}\n',
-                "application/jsonl",
-            ),
-            "purpose": "batch",
-        },
-        optional_params={},
-        litellm_params={
-            "s3_bucket_name": "deployment-bucket",
-            "aws_access_key_id": "deployment-access-key",
-            "aws_secret_access_key": "deployment-secret",
-            "aws_region_name": "eu-west-1",
-        },
-    )
-
-    assert "eu-west-1" in result["url"]
-    assert captured["optional_params"]["aws_access_key_id"] == "deployment-access-key"
-    assert captured["optional_params"]["aws_secret_access_key"] == "deployment-secret"
-    assert captured["optional_params"]["aws_region_name"] == "eu-west-1"
-
-
-def test_bedrock_batch_signing_uses_deployment_credentials(monkeypatch):
-    config = BedrockBatchesConfig()
-    captured = {}
-
-    def capture_signing(**kwargs):
-        captured.update(kwargs)
-        return {}, b"{}"
-
-    monkeypatch.setattr(config.common_utils, "sign_aws_request", capture_signing)
-
-    result = config.transform_create_batch_request(
-        model="us.anthropic.claude-haiku-4-5-20251001-v1:0",
-        create_batch_data={
-            "input_file_id": "s3://deployment-bucket/input.jsonl",
-            "completion_window": "24h",
-            "endpoint": "/v1/chat/completions",
-        },
-        optional_params={},
-        litellm_params={
-            "aws_access_key_id": "deployment-access-key",
-            "aws_secret_access_key": "deployment-secret",
-            "aws_region_name": "eu-west-1",
-            "aws_batch_role_arn": "arn:aws:iam::123456789012:role/bedrock-batch",
-        },
-    )
-
-    assert result["url"].startswith("https://bedrock.eu-west-1.amazonaws.com/")
-    assert captured["optional_params"]["aws_access_key_id"] == "deployment-access-key"
-    assert captured["optional_params"]["aws_secret_access_key"] == "deployment-secret"
-    assert captured["optional_params"]["aws_region_name"] == "eu-west-1"
-
-
-def test_bedrock_batch_retrieval_signing_uses_deployment_credentials(monkeypatch):
-    config = BedrockBatchesConfig()
-    captured = {}
-
-    def capture_signing(**kwargs):
-        captured.update(kwargs)
-        return {}, b""
-
-    monkeypatch.setattr(config.common_utils, "sign_aws_request", capture_signing)
-
-    result = config.transform_retrieve_batch_request(
-        batch_id="arn:aws:bedrock:eu-west-1:123456789012:model-invocation-job/job-1",
-        optional_params={},
-        litellm_params={
-            "aws_access_key_id": "deployment-access-key",
-            "aws_secret_access_key": "deployment-secret",
-            "aws_region_name": "eu-west-1",
-        },
-    )
-
-    assert result["url"].startswith("https://bedrock.eu-west-1.amazonaws.com/")
-    assert captured["optional_params"]["aws_access_key_id"] == "deployment-access-key"
-    assert captured["optional_params"]["aws_secret_access_key"] == "deployment-secret"
-    assert captured["optional_params"]["aws_region_name"] == "eu-west-1"