fix(proxy): let a client AWS identity skip the s3 bucket list only when Bedrock signs with it

An admitted aws_role_name, aws_web_identity_token or aws_profile_name now skips the
bucket list only when it decides the credentials get_credentials picks, on every
deployment of the model group. A web identity token without a role and a session
name does not, since get_credentials then ignores it. A profile does not when the
deployment or the environment sets a role, since the role branch runs first. In
those cases Bedrock still reads with the proxy's credentials, so the bucket list
applies

BaseAWSLLM.identity_choosing_params mirrors the branch order of get_credentials, and
both read the same AWS_* environment fallbacks through one helper

Ported to lts/1.108.x, where the model group is read with Router.get_model_list

litellm/llms/bedrock/base_aws_llm.py

@@ -232,6 +232,10 @@ class SignsRequestsWithAWS:
     pass
 
 
+def _or_env(value: str | None, env_var: str) -> str | None:
+    return value if value is not None else os.getenv(env_var)
+
+
 class BaseAWSLLM(SignsRequestsWithAWS):
     # Process-wide IAM credential cache (shared across instances — Bedrock passthrough is per-request).
     # Storage is in-process memory only: no Redis backend unless attached elsewhere. Entry TTL: static
@@ -346,6 +350,19 @@ class BaseAWSLLM(SignsRequestsWithAWS):
     def _is_auth_with_aws_profile(aws_profile_name: str | None) -> bool:
         return aws_profile_name is not None
 
+    @classmethod
+    def identity_choosing_params(cls, params: AwsAuthParams) -> tuple[str, ...]:
+        token: Final = _or_env(params.aws_web_identity_token, "AWS_WEB_IDENTITY_TOKEN")
+        role: Final = _or_env(params.aws_role_name, "AWS_ROLE_NAME")
+        session: Final = _or_env(params.aws_session_name, "AWS_SESSION_NAME")
+        if cls._is_auth_with_web_identity_token(token, role, session):
+            return ("aws_web_identity_token", "aws_role_name")
+        if cls._is_auth_with_aws_role(role):
+            return ("aws_role_name",)
+        if cls._is_auth_with_aws_profile(_or_env(params.aws_profile_name, "AWS_PROFILE_NAME")):
+            return ("aws_profile_name",)
+        return ()
+
     @staticmethod
     def _is_auth_with_aws_session_token_tuple(
         aws_access_key_id: str | None,
@@ -398,7 +415,7 @@ class BaseAWSLLM(SignsRequestsWithAWS):
             aws_sts_endpoint,
             aws_external_id,
         ) = tuple(
-            value if value is not None else os.getenv(env_var)
+            _or_env(value, env_var)
             for value, env_var in (
                 (aws_access_key_id, "AWS_ACCESS_KEY_ID"),
                 (aws_secret_access_key, "AWS_SECRET_ACCESS_KEY"),

litellm/proxy/auth/auth_utils.py

@@ -31,13 +31,20 @@ from litellm.litellm_core_utils.url_utils import (
     validate_url,
 )
 from litellm.llms.azure.passthrough.transformation import azure_router_model_in_endpoint
+from litellm.llms.bedrock.base_aws_llm import BaseAWSLLM
 from litellm.llms.nvidia_nim.passthrough.transformation import nvidia_nim_model_group_in_path
 from litellm.proxy._types import *
 from litellm.proxy.common_utils.http_parsing_utils import extract_nested_form_metadata
+from litellm.types.llms.bedrock import AWS_AUTH_PARAM_KEYS, AwsAuthParams
 from litellm.types.passthrough_endpoints.pass_through_endpoints import (
     LITELLM_PASS_THROUGH_ENDPOINT_MARKER,
 )
-from litellm.types.router import CONFIGURABLE_CLIENTSIDE_AUTH_PARAMS, Deployment, server_owned_wif_fields_present
+from litellm.types.router import (
+    CONFIGURABLE_CLIENTSIDE_AUTH_PARAMS,
+    Deployment,
+    DeploymentTypedDict,
+    server_owned_wif_fields_present,
+)
 from litellm.types.router import reject_server_owned_wif_params as _reject_server_owned_wif_params
 from litellm.types.utils import CustomPricingLiteLLMParams, oauth_token_exchange_litellm_params
 from litellm.types.videos.main import DecodedVideoId
@@ -242,16 +249,36 @@ def _sends_a_value(body: Mapping[str, object], param: str) -> bool:
     return isinstance(value, str) and bool(value)
 
 
+def _client_aws_fields(body: Mapping[str, object]) -> Mapping[str, object]:
+    return {key: body[key] for key in AWS_AUTH_PARAM_KEYS if _sends_a_value(body, key)}
+
+
+def _identity_choosing_params(client_fields: Mapping[str, object], deployment: DeploymentTypedDict) -> tuple[str, ...]:
+    return BaseAWSLLM.identity_choosing_params(
+        AwsAuthParams.model_validate({**deployment["litellm_params"], **client_fields})
+    )
+
+
 def deployment_admits_a_client_aws_identity(body: Mapping[str, object], llm_router: Router | None) -> bool:
     model: Final = body.get("model")
-    if not isinstance(model, str):
+    if not isinstance(model, str) or llm_router is None:
         return False
-    return any(
-        _allow_model_level_clientside_configurable_parameters(
-            model=model, param=param, request_body_value=body[param], llm_router=llm_router
-        )
+    admitted: Final = frozenset(
+        param
         for param in _CLIENT_AWS_IDENTITY_PARAMS
         if _sends_a_value(body, param)
+        and _allow_model_level_clientside_configurable_parameters(
+            model=model, param=param, request_body_value=body[param], llm_router=llm_router
+        )
+    )
+    deployments: Final = llm_router.get_model_list(model_name=model) or ()
+    client_fields: Final = _client_aws_fields(body)
+    return (
+        bool(admitted)
+        and bool(deployments)
+        and all(
+            not admitted.isdisjoint(_identity_choosing_params(client_fields, deployment)) for deployment in deployments
+        )
     )
 
 

tests/unit/proxy/test_common_request_processing.py

@@ -419,6 +419,9 @@ _S3_VALUES_ON_NON_MEDIA_ROUTES: Final = (
 )
 
 
+_AWS_IDENTITY_ENV_VARS: Final = ("AWS_WEB_IDENTITY_TOKEN", "AWS_ROLE_NAME", "AWS_SESSION_NAME", "AWS_PROFILE_NAME")
+
+
 async def _send_s3_media(
     route_type: _S3MediaRoute,
     path: str,
@@ -427,6 +430,7 @@ async def _send_s3_media(
     general_settings: dict[str, object],
     managed_files: CustomLogger | None = None,
     client_auth_params: tuple[str, ...] = (),
+    deployments_aws_params: tuple[Mapping[str, str], ...] = (MappingProxyType({}),),
 ) -> object:
     from litellm.proxy.proxy_server import ProxyConfig as RealProxyConfig
 
@@ -442,8 +446,10 @@ async def _send_s3_media(
                     "model": "bedrock/us.amazon.nova-2-lite-v1:0",
                     "mock_response": "a cat",
                     "configurable_clientside_auth_params": list(client_auth_params),
+                    **deployment_aws_params,
                 },
             }
+            for deployment_aws_params in deployments_aws_params
         ]
     )
     return await ProxyBaseLLMRequestProcessing(data={"model": "nova", **body}).base_process_llm_request(
@@ -842,16 +848,27 @@ class TestProxyBaseLLMRequestProcessing:
 
     @pytest.mark.asyncio
     @pytest.mark.parametrize(
-        ("param", "value"),
+        ("param", "value", "deployments_aws_params"),
         [
-            pytest.param("aws_role_name", "arn:aws:iam::111122223333:role/reader", id="role"),
-            pytest.param("aws_web_identity_token", "oidc-token", id="web-identity-token"),
-            pytest.param("aws_profile_name", "reader", id="profile"),
+            pytest.param("aws_role_name", "arn:aws:iam::111122223333:role/reader", ({},), id="role"),
+            pytest.param(
+                "aws_web_identity_token",
+                "oidc-token",
+                ({"aws_role_name": "arn:aws:iam::111122223333:role/web", "aws_session_name": "proxy"},),
+                id="web-identity-token-with-the-deployment-role-and-session",
+            ),
+            pytest.param("aws_profile_name", "reader", ({},), id="profile"),
         ],
     )
     async def test_non_admin_s3_media_read_with_a_client_aws_identity_the_deployment_admits_needs_no_listed_bucket(
-        self, param: str, value: str
+        self,
+        param: str,
+        value: str,
+        deployments_aws_params: tuple[dict[str, str], ...],
+        monkeypatch: pytest.MonkeyPatch,
     ) -> None:
+        for env_var in _AWS_IDENTITY_ENV_VARS:
+            monkeypatch.delenv(env_var, raising=False)
         response: Final = await _send_s3_media(
             "acompletion",
             "/v1/chat/completions",
@@ -859,10 +876,58 @@ class TestProxyBaseLLMRequestProcessing:
             LitellmUserRoles.INTERNAL_USER,
             {},
             client_auth_params=(param,),
+            deployments_aws_params=deployments_aws_params,
         )
 
         assert "a cat" in str(response)
 
+    @pytest.mark.asyncio
+    @pytest.mark.parametrize(
+        ("identity", "admitted", "deployments_aws_params"),
+        [
+            pytest.param(
+                {"aws_web_identity_token": "oidc-token"},
+                ("aws_web_identity_token",),
+                ({},),
+                id="web-identity-token-without-a-role-and-session",
+            ),
+            pytest.param(
+                {"aws_profile_name": "reader"},
+                ("aws_profile_name",),
+                ({"aws_role_name": "arn:aws:iam::111122223333:role/deployment"},),
+                id="profile-under-a-deployment-role",
+            ),
+            pytest.param(
+                {"aws_profile_name": "reader"},
+                ("aws_profile_name",),
+                ({}, {"aws_role_name": "arn:aws:iam::111122223333:role/deployment"}),
+                id="profile-under-a-role-on-one-deployment-of-the-group",
+            ),
+        ],
+    )
+    async def test_non_admin_s3_media_is_refused_when_the_admitted_client_aws_field_does_not_pick_the_credentials(
+        self,
+        identity: dict[str, str],
+        admitted: tuple[str, ...],
+        deployments_aws_params: tuple[dict[str, str], ...],
+        monkeypatch: pytest.MonkeyPatch,
+    ) -> None:
+        for env_var in _AWS_IDENTITY_ENV_VARS:
+            monkeypatch.delenv(env_var, raising=False)
+        with pytest.raises(HTTPException) as refused:
+            await _send_s3_media(
+                "acompletion",
+                "/v1/chat/completions",
+                {**_chat_image("s3://media-bucket/dir/cat.png"), **identity},
+                LitellmUserRoles.INTERNAL_USER,
+                {},
+                client_auth_params=admitted,
+                deployments_aws_params=deployments_aws_params,
+            )
+
+        assert refused.value.status_code == status.HTTP_403_FORBIDDEN, refused.value.detail
+        assert "general_settings.allowed_client_side_s3_buckets" in str(refused.value.detail)
+
     @pytest.mark.asyncio
     @pytest.mark.parametrize(
         ("uri", "identity", "admitted", "named_in_refusal"),