From 6b2ac7cc5b79bc173e5aea73f963a8b34089f885 Mon Sep 17 00:00:00 2001
From: mateo-berri <277851410+mateo-berri@users.noreply.github.com>
Date: Thu, 6 Aug 2026 19:25:47 -0700
Subject: [PATCH] refactor(bedrock): freeze the SSE-KMS header and key-source
 collections

The staging merge tightened the LIT002 ceiling, so the three mutable dict
literals this branch added now breach it. Build the S3 request headers as
MappingProxyType and resolve the encryption key from a tuple of sources.
---
 litellm/llms/bedrock/common_utils.py         | 10 +++----
 litellm/llms/bedrock/files/transformation.py | 28 +++++++++++---------
 2 files changed, 21 insertions(+), 17 deletions(-)

diff --git a/litellm/llms/bedrock/common_utils.py b/litellm/llms/bedrock/common_utils.py
index c5ef78b44f3..d18cb7d8734 100644
--- a/litellm/llms/bedrock/common_utils.py
+++ b/litellm/llms/bedrock/common_utils.py
@@ -1314,11 +1314,11 @@ def resolve_s3_encryption_key_id(
     Precedence: `s3_encryption_key_id` in litellm_params, then optional_params
     (client-side / request params), then the AWS_S3_ENCRYPTION_KEY_ID env var.
     """
-    for source in (litellm_params, optional_params or {}):
-        value = source.get("s3_encryption_key_id")
-        if isinstance(value, str) and value:
-            return value
-    return get_secret_str("AWS_S3_ENCRYPTION_KEY_ID")
+    candidates: Final = tuple(
+        source.get("s3_encryption_key_id") for source in (litellm_params, optional_params) if source is not None
+    )
+    explicit: Final = next((value for value in candidates if isinstance(value, str) and value), None)
+    return explicit or get_secret_str("AWS_S3_ENCRYPTION_KEY_ID")
 
 
 class CommonBatchFilesUtils:
diff --git a/litellm/llms/bedrock/files/transformation.py b/litellm/llms/bedrock/files/transformation.py
index b4c4a270bd7..728fd02f001 100644
--- a/litellm/llms/bedrock/files/transformation.py
+++ b/litellm/llms/bedrock/files/transformation.py
@@ -788,20 +788,24 @@ class BedrockFilesConfig(BaseAWSLLM, BaseFilesConfig):
 
         # Prepare headers with required S3 headers (same as s3_v2.py)
         sse_headers: Final = (
-            {
-                "x-amz-server-side-encryption": "aws:kms",
-                "x-amz-server-side-encryption-aws-kms-key-id": s3_encryption_key_id,
-            }
+            MappingProxyType(
+                {
+                    "x-amz-server-side-encryption": "aws:kms",
+                    "x-amz-server-side-encryption-aws-kms-key-id": s3_encryption_key_id,
+                }
+            )
             if s3_encryption_key_id
-            else {}
+            else MappingProxyType({})
+        )
+        request_headers: Final = MappingProxyType(
+            {
+                "Content-Type": "application/json",  # JSONL files are JSON content
+                "x-amz-content-sha256": content_hash,  # REQUIRED by S3
+                "Content-Language": "en",
+                "Cache-Control": "private, immutable, max-age=31536000, s-maxage=0",
+                **sse_headers,
+            }
         )
-        request_headers: Final = {
-            "Content-Type": "application/json",  # JSONL files are JSON content
-            "x-amz-content-sha256": content_hash,  # REQUIRED by S3
-            "Content-Language": "en",
-            "Cache-Control": "private, immutable, max-age=31536000, s-maxage=0",
-            **sse_headers,
-        }
 
         # Use requests.Request to prepare the request (same pattern as s3_v2.py)
         req: Final = requests.Request("PUT", api_base, data=content, headers=request_headers)