harden bedrock file bucket validation

litellm/litellm_core_utils/cloud_storage_security.py

@@ -69,7 +69,10 @@ def _validate_cloud_object_path(object_name: str) -> None:
         raise ValueError("Cloud storage object name must be relative")
     if any(ord(char) < 32 or ord(char) == 127 for char in object_name):
         raise ValueError("Cloud storage object name contains control characters")
-    if any(segment in {".", ".."} for segment in object_name.split("/")):
+    segments = object_name.split("/")
+    if any(segment in {".", ".."} for segment in segments):
+        raise ValueError("Cloud storage object name contains an invalid path segment")
+    if "" in segments[:-1]:
         raise ValueError("Cloud storage object name contains an invalid path segment")
 
 

litellm/llms/bedrock/files/handler.py

@@ -90,12 +90,10 @@ class BedrockFilesHandler(BaseAWSLLM):
         )
 
     def _get_configured_s3_bucket_name(self, optional_params: dict) -> str:
-        bucket_name = optional_params.get("s3_bucket_name") or os.getenv(
-            "AWS_S3_BUCKET_NAME"
-        )
+        bucket_name = os.getenv("AWS_S3_BUCKET_NAME")
         if not bucket_name:
             raise ValueError(
-                "S3 bucket_name is required. Set 's3_bucket_name' or AWS_S3_BUCKET_NAME."
+                "S3 bucket_name is required. Set AWS_S3_BUCKET_NAME for Bedrock file content retrieval."
             )
         return bucket_name
 

tests/test_litellm/llms/bedrock/files/test_bedrock_files_handler.py

@@ -1,4 +1,6 @@
 import base64
+import os
+from unittest.mock import patch
 
 import pytest
 
@@ -60,6 +62,13 @@ class TestBedrockFilesHandler:
                 configured_bucket_name="safe-bucket",
             )
 
+    def test_should_reject_empty_middle_path_segment(self):
+        with pytest.raises(ValueError, match="invalid path segment"):
+            self.handler._parse_s3_uri(
+                s3_uri="s3://safe-bucket/litellm-bedrock-files//secret.jsonl",
+                configured_bucket_name="safe-bucket",
+            )
+
     def test_should_extract_unified_managed_s3_uri(self):
         file_id = _encode_unified_file_id(
             "s3://safe-bucket/litellm-batch-outputs/job/output.jsonl"
@@ -83,3 +92,19 @@ class TestBedrockFilesHandler:
                 s3_uri=s3_uri,
                 configured_bucket_name="safe-bucket",
             )
+
+    def test_should_not_trust_request_s3_bucket_name_for_expected_bucket(self):
+        with patch.dict(os.environ, {"AWS_S3_BUCKET_NAME": "safe-bucket"}):
+            assert (
+                self.handler._get_configured_s3_bucket_name(
+                    {"s3_bucket_name": "attacker-bucket"}
+                )
+                == "safe-bucket"
+            )
+
+    def test_should_require_server_s3_bucket_name(self):
+        with patch.dict(os.environ, {}, clear=True):
+            with pytest.raises(ValueError, match="AWS_S3_BUCKET_NAME"):
+                self.handler._get_configured_s3_bucket_name(
+                    {"s3_bucket_name": "attacker-bucket"}
+                )