feat(bedrock): opt-in pin_deployment_credentials to ignore caller aws_* overrides

litellm/__init__.py

@@ -240,6 +240,7 @@ use_chat_completions_url_for_anthropic_messages: bool = bool(
 # Or via `litellm_settings.strip_anthropic_total_tokens: true` in
 # config.yaml.
 strip_anthropic_total_tokens: bool = False
+pin_deployment_credentials: bool = bool(os.getenv("LITELLM_PIN_DEPLOYMENT_CREDENTIALS", False))
 route_all_chat_openai_to_responses: bool = (
     os.getenv("LITELLM_ROUTE_ALL_CHAT_OPENAI_TO_RESPONSES", "false").lower() == "true"
 )  # When True, routes all OpenAI /chat/completions requests through the Responses API bridge

litellm/constants.py

@@ -1202,6 +1202,20 @@ bedrock_embedding_models: set = set(
     ]
 )
 
+AWS_PINNED_CREDENTIAL_PARAMS: frozenset = frozenset(
+    {
+        "aws_access_key_id",
+        "aws_secret_access_key",
+        "aws_session_token",
+        "aws_session_name",
+        "aws_profile_name",
+        "aws_role_name",
+        "aws_web_identity_token",
+        "aws_sts_endpoint",
+        "aws_external_id",
+    }
+)
+
 known_tokenizer_config = {
     "mistralai/Mistral-7B-Instruct-v0.1": {
         "tokenizer": {

litellm/router.py

@@ -55,6 +55,7 @@ from litellm.caching.caching import (
     RedisClusterCache,
 )
 from litellm.constants import (
+    AWS_PINNED_CREDENTIAL_PARAMS,
     DEFAULT_HEALTH_CHECK_INTERVAL,
     DEFAULT_HEALTH_CHECK_STALENESS_MULTIPLIER,
     DEFAULT_MAX_LRU_CACHE_SIZE,
@@ -2935,6 +2936,30 @@ class Router:
         if "tool_choice" not in kwargs and dep_params.get("tool_choice") is not None:
             kwargs["tool_choice"] = dep_params["tool_choice"]
 
+    @staticmethod
+    def _pin_deployment_aws_credentials(deployment: dict, kwargs: dict) -> None:
+        """
+        When ``litellm.pin_deployment_credentials`` is enabled, treat the
+        deployment's AWS credential/identity params as authoritative by dropping
+        any caller-supplied ``aws_*`` credential override from the request
+        kwargs. The config values then flow through unchanged when deployment
+        ``litellm_params`` are merged in at the call site.
+
+        Only applies to Bedrock/SageMaker deployments, the providers that consume
+        the ``aws_*`` passthrough.
+        """
+        if not litellm.pin_deployment_credentials:
+            return
+        litellm_params = deployment.get("litellm_params", {})
+        model = litellm_params.get("model")
+        if not isinstance(model, str):
+            return
+        provider = litellm_params.get("custom_llm_provider") or (model.split("/", 1)[0] if "/" in model else "")
+        if provider != "bedrock" and not provider.startswith("sagemaker"):
+            return
+        for param in AWS_PINNED_CREDENTIAL_PARAMS:
+            kwargs.pop(param, None)
+
     def _update_kwargs_with_deployment(
         self,
         deployment: dict,
@@ -2948,6 +2973,7 @@ class Router:
         - Merges tools from deployment with request (proxy-configured tools + request tools).
         """
         self._merge_tools_from_deployment(deployment=deployment, kwargs=kwargs)
+        self._pin_deployment_aws_credentials(deployment=deployment, kwargs=kwargs)
 
         model_info = deployment.get("model_info", {}).copy()
         deployment_litellm_model_name = deployment["litellm_params"]["model"]

tests/test_litellm/test_router_pin_deployment_credentials.py

@@ -0,0 +1,109 @@
+"""Regression tests for litellm.pin_deployment_credentials (issue #32892).
+
+When enabled, caller-supplied aws_* credential/identity overrides in the request
+body must be ignored so the deployment-configured values are used.
+"""
+
+import pytest
+
+import litellm
+from litellm import Router
+
+CONFIG_ROLE = "arn:aws:iam::111111111111:role/config-role"
+CALLER_ROLE = "arn:aws:iam::222222222222:role/caller-role"
+
+
+def _bedrock_deployment() -> dict:
+    return {
+        "model_name": "claude",
+        "litellm_params": {
+            "model": "bedrock/anthropic.claude-3-sonnet-20240229-v1:0",
+            "aws_role_name": CONFIG_ROLE,
+            "aws_session_name": "config-session",
+            "aws_region_name": "us-east-1",
+        },
+    }
+
+
+def _merged(deployment: dict, kwargs: dict) -> dict:
+    """Mirror the router call-site merge: config litellm_params first, caller kwargs last."""
+    return {**deployment["litellm_params"], **kwargs}
+
+
+@pytest.fixture(autouse=True)
+def _reset_flag():
+    original = litellm.pin_deployment_credentials
+    yield
+    litellm.pin_deployment_credentials = original
+
+
+def test_pinned_credentials_ignore_caller_override():
+    litellm.pin_deployment_credentials = True
+    deployment = _bedrock_deployment()
+    kwargs = {
+        "aws_role_name": CALLER_ROLE,
+        "aws_access_key_id": "AKIA-caller",
+        "aws_secret_access_key": "caller-secret",
+    }
+
+    Router._pin_deployment_aws_credentials(deployment=deployment, kwargs=kwargs)
+
+    assert "aws_role_name" not in kwargs
+    assert "aws_access_key_id" not in kwargs
+    assert "aws_secret_access_key" not in kwargs
+
+    merged = _merged(deployment, kwargs)
+    assert merged["aws_role_name"] == CONFIG_ROLE
+    # A credential param the caller tried to inject but config never set must not leak through.
+    assert "aws_access_key_id" not in merged
+
+
+def test_region_is_not_pinned():
+    litellm.pin_deployment_credentials = True
+    deployment = _bedrock_deployment()
+    kwargs = {"aws_role_name": CALLER_ROLE, "aws_region_name": "us-west-2"}
+
+    Router._pin_deployment_aws_credentials(deployment=deployment, kwargs=kwargs)
+
+    assert "aws_region_name" in kwargs
+    assert _merged(deployment, kwargs)["aws_region_name"] == "us-west-2"
+
+
+def test_disabled_keeps_caller_override():
+    litellm.pin_deployment_credentials = False
+    deployment = _bedrock_deployment()
+    kwargs = {"aws_role_name": CALLER_ROLE}
+
+    Router._pin_deployment_aws_credentials(deployment=deployment, kwargs=kwargs)
+
+    assert kwargs["aws_role_name"] == CALLER_ROLE
+    assert _merged(deployment, kwargs)["aws_role_name"] == CALLER_ROLE
+
+
+def test_non_aws_provider_untouched():
+    litellm.pin_deployment_credentials = True
+    deployment = {
+        "model_name": "gpt-4o",
+        "litellm_params": {"model": "openai/gpt-4o", "api_key": "fake"},
+    }
+    kwargs = {"aws_role_name": CALLER_ROLE}
+
+    Router._pin_deployment_aws_credentials(deployment=deployment, kwargs=kwargs)
+
+    assert kwargs["aws_role_name"] == CALLER_ROLE
+
+
+def test_end_to_end_via_update_kwargs_with_deployment():
+    litellm.pin_deployment_credentials = True
+    router = Router(model_list=[_bedrock_deployment()])
+    deployment = (router.get_model_list(model_name="claude") or [])[0]
+    kwargs = {
+        "model": "claude",
+        "messages": [{"role": "user", "content": "hi"}],
+        "aws_role_name": CALLER_ROLE,
+    }
+
+    router._update_kwargs_with_deployment(deployment=deployment, kwargs=kwargs)
+
+    assert "aws_role_name" not in kwargs
+    assert _merged(deployment, kwargs)["aws_role_name"] == CONFIG_ROLE