diff --git a/enterprise/litellm_enterprise/proxy/hooks/managed_files.py b/enterprise/litellm_enterprise/proxy/hooks/managed_files.py
index c7e1b94a2ef..4899b87da7a 100644
--- a/enterprise/litellm_enterprise/proxy/hooks/managed_files.py
+++ b/enterprise/litellm_enterprise/proxy/hooks/managed_files.py
@@ -27,10 +27,13 @@ import litellm
from litellm import Router, verbose_logger
from litellm._uuid import uuid
from litellm.caching.caching import DualCache
+from litellm.constants import MAX_FILE_LIST_LIMIT
from litellm.integrations.custom_logger import CustomLogger
from litellm.litellm_core_utils.prompt_templates.common_utils import (
extract_file_metadata,
)
+from openai.types.file_deleted import FileDeleted
+
from litellm.llms.base_llm.files.transformation import BaseFileEndpoints
from litellm.llms.base_llm.managed_resources.isolation import (
build_list_page,
@@ -48,7 +51,6 @@ from litellm.proxy._types import (
from litellm.proxy.openai_files_endpoints.common_utils import (
BATCH_CREATE_HIDDEN_PARAM,
FILE_LIST_CONTINUATION_CHUNK_SIZE,
- MAX_FILE_LIST_LIMIT,
_is_base64_encoded_unified_file_id,
apply_unified_file_ids,
decode_model_from_file_id,
@@ -1787,7 +1789,7 @@ class _PROXY_LiteLLMManagedFiles(CustomLogger, BaseFileEndpoints):
litellm_parent_otel_span: Optional[Span],
llm_router: Router,
**data: Dict,
- ) -> OpenAIFileObject:
+ ) -> FileDeleted:
# Check if file deletion should be blocked due to batch references
await self._check_file_deletion_allowed(file_id)
@@ -1795,7 +1797,6 @@ class _PROXY_LiteLLMManagedFiles(CustomLogger, BaseFileEndpoints):
# file_id = convert_b64_uid_to_unified_uid(file_id)
model_file_id_mapping = await self.get_model_file_id_mapping([file_id], litellm_parent_otel_span)
- delete_response = None
specific_model_file_id_mapping = model_file_id_mapping.get(file_id)
if specific_model_file_id_mapping:
# Remove conflicting keys from data to avoid duplicate keyword arguments
@@ -1810,23 +1811,14 @@ class _PROXY_LiteLLMManagedFiles(CustomLogger, BaseFileEndpoints):
else {}
),
}
- delete_response = await llm_router.afile_delete(model=model_id, file_id=model_file_id, **delete_data)
+ await llm_router.afile_delete(model=model_id, file_id=model_file_id, **delete_data)
- stored_file_object = await self.delete_unified_file_id(file_id, litellm_parent_otel_span)
+ await self.delete_unified_file_id(file_id, litellm_parent_otel_span)
- # Record successful deletion metric only on actual success
- if stored_file_object or delete_response:
- prom_logger = self._get_prometheus_logger()
- if prom_logger:
- prom_logger.record_managed_file_deleted(result="success")
-
- if stored_file_object:
- return OpenAIFileObject.model_validate(stored_file_object).model_copy(update={"id": file_id})
- elif delete_response:
- delete_response.id = file_id
- return delete_response
- else:
- raise Exception(f"LiteLLM Managed File object with id={file_id} not found")
+ prom_logger = self._get_prometheus_logger()
+ if prom_logger:
+ prom_logger.record_managed_file_deleted(result="success")
+ return FileDeleted(id=file_id, object="file", deleted=True)
async def afile_content(
self,
diff --git a/litellm/constants.py b/litellm/constants.py
index 6b984c2673c..da86665cecb 100644
--- a/litellm/constants.py
+++ b/litellm/constants.py
@@ -53,6 +53,7 @@ S3_BOUNDED_OBJECT_KEY_HEAD_BYTES: Final = 64
S3_PREFIX_DIGEST_CHARS: Final = 16
# s3 allows 2048 bytes of combined metadata headers, which Content-Disposition counts against
MAX_S3_OBJECT_DOWNLOAD_FILENAME_BYTES: Final = 1024
+MAX_FILE_LIST_LIMIT: Final = 10000
DEFAULT_SQS_FLUSH_INTERVAL_SECONDS: Final = int(os.getenv("DEFAULT_SQS_FLUSH_INTERVAL_SECONDS", 10))
DEFAULT_NUM_WORKERS_LITELLM_PROXY: Final = int(os.getenv("DEFAULT_NUM_WORKERS_LITELLM_PROXY", 1))
budget_reservation_disabled_info_emitted = False
diff --git a/litellm/files/main.py b/litellm/files/main.py
index 218518eb3cd..1d5da29fe6f 100644
--- a/litellm/files/main.py
+++ b/litellm/files/main.py
@@ -682,6 +682,10 @@ def file_list(
)
if provider_config is not None:
litellm_params_dict: Final = get_litellm_params(**kwargs)
+ add_trusted_model_credentials_to_litellm_params(
+ litellm_params_dict=litellm_params_dict,
+ kwargs=kwargs,
+ )
litellm_params_dict["api_key"] = optional_params.api_key
litellm_params_dict["api_base"] = optional_params.api_base
diff --git a/litellm/llms/base_llm/files/transformation.py b/litellm/llms/base_llm/files/transformation.py
index 7a7088c2fb5..6d16a1cea69 100644
--- a/litellm/llms/base_llm/files/transformation.py
+++ b/litellm/llms/base_llm/files/transformation.py
@@ -1,5 +1,5 @@
from abc import ABC, abstractmethod
-from collections.abc import Iterator
+from collections.abc import Iterator, Mapping
from typing import TYPE_CHECKING, Any, Union
import httpx
@@ -160,6 +160,15 @@ class BaseFilesConfig(BaseConfig):
) -> tuple[str, dict]:
"""Transform file list request into provider-specific format."""
+ def transform_list_files_next_request(
+ self,
+ raw_response: httpx.Response,
+ optional_params: Mapping[str, object],
+ litellm_params: dict, # mutable-ok: carries provider stashes from the request transform to the response one
+ ) -> tuple[str, dict[str, str]] | None:
+ """Request for the page after `raw_response`, or None once the listing is complete."""
+ return None
+
@abstractmethod
def transform_list_files_response(
self,
@@ -258,7 +267,7 @@ class BaseFileEndpoints(ABC):
litellm_parent_otel_span: Span | None,
llm_router: Router,
**data: dict,
- ) -> OpenAIFileObject:
+ ) -> FileDeleted:
pass
@abstractmethod
diff --git a/litellm/llms/bedrock/files/transformation.py b/litellm/llms/bedrock/files/transformation.py
index 9875ac2b9c3..6e2b0c12090 100644
--- a/litellm/llms/bedrock/files/transformation.py
+++ b/litellm/llms/bedrock/files/transformation.py
@@ -1,14 +1,18 @@
import base64
import json
import os
+import posixpath
import time
+import xml.etree.ElementTree as ET
from collections.abc import Iterable, Mapping, MutableMapping, Sequence
from contextlib import suppress
+from dataclasses import dataclass
+from datetime import datetime
from functools import cache
from itertools import chain
from types import MappingProxyType
from typing import Any, Final, Literal, TypeAlias, TypedDict
-from urllib.parse import unquote
+from urllib.parse import quote, unquote, urlencode
import httpx
from httpx import Headers, Response
@@ -23,6 +27,7 @@ from litellm.files.utils import FilesAPIUtils
from litellm.litellm_core_utils.aws_partition import get_aws_dns_suffix
from litellm.litellm_core_utils.cloud_storage_security import (
BEDROCK_MANAGED_S3_BATCH_PREFIX,
+ BEDROCK_MANAGED_S3_OUTPUT_PREFIX,
BEDROCK_MANAGED_S3_PREFIXES,
BEDROCK_MANAGED_S3_UPLOAD_PREFIX,
build_managed_cloud_object_name,
@@ -62,6 +67,10 @@ from ..common_utils import BedrockError, merge_bedrock_aws_request_params, resol
S3_SIGNED_REQUEST_HEADERS_PARAM: Final = "_s3_signed_request_headers"
+LIST_FILES_PURPOSE_PARAM: Final = "_s3_list_files_purpose"
+
+LIST_FILES_LOCATION_PARAM: Final = "_s3_list_files_location"
+
class _S3DeleteContext(BaseModel):
file_id: str = Field(min_length=1)
@@ -152,6 +161,13 @@ class _BedrockS3RequestParams(BaseModel):
s3_endpoint_url: str | None = None
+@dataclass(frozen=True, slots=True)
+class _S3RequestTarget:
+ endpoint_url: str
+ aws_region_name: str
+ request_params: _BedrockS3RequestParams
+
+
class _TrustedS3ModelCredentials(BaseModel):
"""The S3 buckets the server trusts file ids against, from the deployment snapshot."""
@@ -248,6 +264,128 @@ def _validate_file_id_against_configured_buckets(
return validate_against(configured_bucket_names[-1])
+_REJECTED_FILE_ID_REQUEST_URL: Final = "https://litellm.ai"
+
+
+def _rejected_file_id(reason: ValueError) -> BedrockError:
+ message: Final = str(reason)
+ return BedrockError(
+ status_code=400,
+ message=message,
+ response=httpx.Response(
+ status_code=400,
+ text=message,
+ request=httpx.Request(method="GET", url=_REJECTED_FILE_ID_REQUEST_URL),
+ ),
+ )
+
+
+def _resolve_managed_s3_object(file_id: str, litellm_params: Mapping[str, object]) -> tuple[str, str]:
+ configured_bucket_names: Final = get_configured_s3_bucket_names(litellm_params)
+ allow_legacy_cloud_file_ids: Final = should_allow_legacy_cloud_file_ids(litellm_params)
+ try:
+ return _validate_file_id_against_configured_buckets(
+ s3_uri=extract_s3_uri_from_file_id(file_id),
+ configured_bucket_names=configured_bucket_names,
+ allow_legacy_cloud_file_ids=allow_legacy_cloud_file_ids,
+ )
+ except ValueError as reason:
+ raise _rejected_file_id(reason) from reason
+
+
+_ANY_MANAGED_LISTING_PREFIX: Final = os.path.commonprefix(BEDROCK_MANAGED_S3_PREFIXES)
+_MANAGED_LISTING_PREFIX_BY_PURPOSE: Final = MappingProxyType(
+ {
+ "batch": os.path.commonprefix((BEDROCK_MANAGED_S3_BATCH_PREFIX, BEDROCK_MANAGED_S3_UPLOAD_PREFIX)),
+ "batch_output": BEDROCK_MANAGED_S3_OUTPUT_PREFIX,
+ }
+)
+
+
+_EMPTY_LISTING_QUERY: Final = (("list-type", "2"), ("max-keys", "0"))
+
+
+def _managed_listing_prefix(configured_prefix: str, purpose: str | None) -> str | None:
+ managed_prefix: Final = _MANAGED_LISTING_PREFIX_BY_PURPOSE.get(purpose) if purpose else _ANY_MANAGED_LISTING_PREFIX
+ if managed_prefix is None:
+ return None
+ return f"{configured_prefix}/{managed_prefix}" if configured_prefix else managed_prefix
+
+
+def _listing_query(configured_prefix: str, purpose: str | None) -> tuple[tuple[str, str], ...]:
+ listing_prefix: Final = _managed_listing_prefix(configured_prefix, purpose)
+ if listing_prefix is None:
+ return _EMPTY_LISTING_QUERY
+ return (("list-type", "2"), ("prefix", listing_prefix))
+
+
+def _requested_listing_purpose(litellm_params: Mapping[str, object]) -> str | None:
+ requested_purpose: Final = litellm_params.get(LIST_FILES_PURPOSE_PARAM)
+ return requested_purpose if isinstance(requested_purpose, str) else None
+
+
+def _walked_listing_purpose(litellm_params: Mapping[str, object]) -> str | None:
+ walked_purpose: Final = litellm_params.get(LIST_FILES_LOCATION_PARAM)
+ return walked_purpose if isinstance(walked_purpose, str) else _requested_listing_purpose(litellm_params)
+
+
+def _output_location_still_unlisted(litellm_params: Mapping[str, object]) -> bool:
+ if _walked_listing_purpose(litellm_params) is not None:
+ return False
+ return _listing_bucket_name(litellm_params, "batch_output") != _listing_bucket_name(litellm_params, None)
+
+
+def _listing_bucket_name(litellm_params: Mapping[str, object], purpose: str | None) -> str:
+ if purpose != "batch_output":
+ return get_configured_s3_bucket_name(litellm_params)
+ trusted: Final = _trusted_s3_model_credentials(litellm_params)
+ return (
+ trusted.s3_output_bucket_name
+ or os.getenv("AWS_S3_OUTPUT_BUCKET_NAME")
+ or get_configured_s3_bucket_name(litellm_params)
+ )
+
+
+def _listed_object_created_at(entry: ET.Element) -> int:
+ last_modified: Final = entry.findtext("{*}LastModified")
+ if not last_modified:
+ return 0
+ return int(datetime.fromisoformat(last_modified.replace("Z", "+00:00")).timestamp())
+
+
+def _listed_managed_file(
+ entry: ET.Element,
+ bucket_name: str,
+ configured_bucket_name: str,
+ allow_legacy_cloud_file_ids: bool,
+) -> OpenAIFileObject | None:
+ object_key: Final = entry.findtext("{*}Key")
+ if not object_key:
+ return None
+ file_id: Final = f"s3://{bucket_name}/{object_key}"
+ try:
+ validate_managed_cloud_file_id(
+ file_id=file_id,
+ scheme="s3://",
+ configured_bucket_name=configured_bucket_name,
+ allowed_object_prefixes=BEDROCK_MANAGED_S3_PREFIXES,
+ allow_legacy_cloud_file_ids=allow_legacy_cloud_file_ids,
+ )
+ except ValueError:
+ return None
+ _, configured_prefix = split_configured_cloud_bucket_name(configured_bucket_name)
+ relative_key: Final = object_key[len(configured_prefix) + 1 :] if configured_prefix else object_key
+ return OpenAIFileObject(
+ id=file_id,
+ bytes=int(entry.findtext("{*}Size") or 0),
+ created_at=_listed_object_created_at(entry),
+ filename=posixpath.basename(object_key),
+ object="file",
+ purpose="batch_output" if relative_key.startswith(BEDROCK_MANAGED_S3_OUTPUT_PREFIX) else "batch",
+ status="uploaded",
+ )
+
+
def _uploaded_object_size(litellm_params: Mapping[str, object], raw_response: Response) -> int:
"""
S3 answers PutObject with an empty body, so the stored object size comes from the
@@ -1213,18 +1351,86 @@ class BedrockFilesConfig(BaseAWSLLM, BaseFilesConfig):
def transform_list_files_request(
self,
purpose: str | None,
- optional_params: dict,
- litellm_params: dict,
- ) -> tuple[str, dict]:
- raise NotImplementedError("BedrockFilesConfig does not support file listing")
+ optional_params: Mapping[str, object],
+ litellm_params: MutableMapping[str, object],
+ ) -> tuple[str, dict[str, str]]:
+ litellm_params[LIST_FILES_PURPOSE_PARAM] = purpose # rebind-ok: handed to the response transform
+ litellm_params[LIST_FILES_LOCATION_PARAM] = purpose # rebind-ok: names the location the next page walks
+ return self._signed_listing_request(purpose, optional_params, litellm_params, continuation_token=None)
+
+ def transform_list_files_next_request(
+ self,
+ raw_response: httpx.Response,
+ optional_params: Mapping[str, object],
+ litellm_params: MutableMapping[str, object],
+ ) -> tuple[str, dict[str, str]] | None:
+ if raw_response.status_code >= 400:
+ return None
+ continuation_token: Final = ET.fromstring(raw_response.content).findtext("{*}NextContinuationToken")
+ if continuation_token:
+ return self._signed_listing_request(
+ _walked_listing_purpose(litellm_params), optional_params, litellm_params, continuation_token
+ )
+ if not _output_location_still_unlisted(litellm_params):
+ return None
+ litellm_params[LIST_FILES_LOCATION_PARAM] = "batch_output" # rebind-ok: the input location is fully listed
+ return self._signed_listing_request("batch_output", optional_params, litellm_params, continuation_token=None)
+
+ def _signed_listing_request(
+ self,
+ purpose: str | None,
+ optional_params: Mapping[str, object],
+ litellm_params: MutableMapping[str, object],
+ continuation_token: str | None,
+ ) -> tuple[str, dict[str, str]]:
+ bucket_name, configured_prefix = split_configured_cloud_bucket_name(
+ _listing_bucket_name(litellm_params, purpose)
+ )
+ target: Final = self._s3_request_target(optional_params=optional_params, litellm_params=litellm_params)
+ url: Final = f"{target.endpoint_url}/{bucket_name}/"
+ listing_query: Final = _listing_query(configured_prefix, purpose)
+ continuation_query: Final = (("continuation-token", continuation_token),) if continuation_token else ()
+ query: Final[dict[str, str]] = dict( # mutable-ok: the base files contract returns the query as a dict
+ listing_query + continuation_query
+ )
+ signed_headers: Final = self._sign_s3_request_without_body(
+ method="GET",
+ api_base=f"{url}?{urlencode(query, quote_via=quote, safe='')}",
+ aws_region_name=target.aws_region_name,
+ request_params=target.request_params,
+ )
+ litellm_params[S3_SIGNED_REQUEST_HEADERS_PARAM] = signed_headers # rebind-ok: handed to validate_environment
+ return url, query
def transform_list_files_response(
self,
raw_response: httpx.Response,
logging_obj: LiteLLMLoggingObj,
- litellm_params: dict,
+ litellm_params: Mapping[str, object],
) -> list[OpenAIFileObject]:
- raise NotImplementedError("BedrockFilesConfig does not support file listing")
+ if raw_response.status_code >= 400:
+ raise BedrockError(
+ status_code=raw_response.status_code,
+ message=raw_response.text,
+ headers=raw_response.headers,
+ response=raw_response,
+ )
+ purpose: Final = _requested_listing_purpose(litellm_params)
+ configured_bucket_name: Final = _listing_bucket_name(litellm_params, _walked_listing_purpose(litellm_params))
+ allow_legacy_cloud_file_ids: Final = should_allow_legacy_cloud_file_ids(litellm_params)
+ listing: Final = ET.fromstring(raw_response.content)
+ bucket_name: Final = (
+ listing.findtext("{*}Name") or split_configured_cloud_bucket_name(configured_bucket_name)[0]
+ )
+ listed_files: Final = (
+ _listed_managed_file(entry, bucket_name, configured_bucket_name, allow_legacy_cloud_file_ids)
+ for entry in listing.iterfind("{*}Contents")
+ )
+ return [ # mutable-ok: the base files contract returns a list
+ listed_file
+ for listed_file in listed_files
+ if listed_file is not None and (purpose is None or listed_file.purpose == purpose)
+ ]
def transform_file_content_request(
self,
@@ -1255,39 +1461,54 @@ class BedrockFilesConfig(BaseAWSLLM, BaseFilesConfig):
optional_params: Mapping[str, object],
litellm_params: MutableMapping[str, object],
) -> tuple[str, dict[str, str]]:
- s3_uri: Final = extract_s3_uri_from_file_id(file_id)
- bucket_name, object_key = _validate_file_id_against_configured_buckets(
- s3_uri=s3_uri,
- configured_bucket_names=get_configured_s3_bucket_names(litellm_params),
- allow_legacy_cloud_file_ids=should_allow_legacy_cloud_file_ids(litellm_params),
+ bucket_name, object_key = _resolve_managed_s3_object(file_id=file_id, litellm_params=litellm_params)
+ target: Final = self._s3_request_target(optional_params=optional_params, litellm_params=litellm_params)
+ url: Final = f"{target.endpoint_url}/{bucket_name}/{encode_s3_object_key_for_url(object_key)}"
+ signed_headers: Final = self._sign_s3_request_without_body(
+ method=method,
+ api_base=url,
+ aws_region_name=target.aws_region_name,
+ request_params=target.request_params,
)
+ litellm_params[S3_SIGNED_REQUEST_HEADERS_PARAM] = signed_headers # rebind-ok: handed to validate_environment
+ return url, {} # mutable-ok: the base files contract returns the query as a dict
- request_params: Final = _BedrockS3RequestParams.model_validate({**litellm_params, **optional_params})
-
+ def _s3_request_target(
+ self,
+ optional_params: Mapping[str, object],
+ litellm_params: Mapping[str, object],
+ ) -> _S3RequestTarget:
+ """
+ The shared files handler passes optional_params={}, so AWS credentials and
+ region arrive via litellm_params here (unlike the upload path).
+ s3_region_name wins over aws_region_name, same priority as get_complete_file_url.
+ """
+ request_params: Final = _BedrockS3RequestParams.model_validate(
+ MappingProxyType({**litellm_params, **optional_params})
+ )
region_preference: Final = request_params.s3_region_name or request_params.aws_region_name
- region_params: Final[dict[str, str | None]] = {"aws_region_name": region_preference}
- aws_region_name: Final = self._get_aws_region_name(optional_params=region_params, model="")
-
- s3_endpoint_url: Final = (
+ aws_region_name: Final = self._get_aws_region_name(
+ optional_params={"aws_region_name": region_preference}, # mutable-ok: BaseAWSLLM takes a dict
+ model="",
+ )
+ endpoint_url: Final = (
request_params.s3_endpoint_url or f"https://s3.{aws_region_name}.{get_aws_dns_suffix(aws_region_name)}"
).rstrip("/")
- url: Final = f"{s3_endpoint_url}/{bucket_name}/{encode_s3_object_key_for_url(object_key)}"
-
- litellm_params[S3_SIGNED_REQUEST_HEADERS_PARAM] = self._sign_s3_request_without_body(
- api_base=url,
- aws_region_name=aws_region_name,
- request_params=request_params,
- method=method,
+ return _S3RequestTarget(
+ endpoint_url=endpoint_url, aws_region_name=aws_region_name, request_params=request_params
)
- return url, {}
def _sign_s3_request_without_body(
self,
+ method: Literal["GET", "DELETE"],
api_base: str,
aws_region_name: str,
request_params: _BedrockS3RequestParams,
- method: Literal["GET", "DELETE"] = "GET",
- ) -> dict[str, str]:
+ ) -> Mapping[str, str]:
+ """
+ SigV4-sign a bodiless S3 request (GetObject, DeleteObject, ListObjectsV2),
+ mirroring `_sign_s3_request` (PUT).
+ """
try:
import hashlib
@@ -1313,11 +1534,11 @@ class BedrockFilesConfig(BaseAWSLLM, BaseFilesConfig):
aws_request: Final = AWSRequest( # any-ok: botocore AWSRequest is untyped
method=method,
url=api_base,
- headers={"x-amz-content-sha256": empty_body_hash},
+ headers={"x-amz-content-sha256": empty_body_hash}, # mutable-ok: botocore AWSRequest takes a dict
)
auth: Final = S3SigV4Auth(credentials, "s3", aws_region_name) # any-ok: botocore untyped
auth.add_auth(aws_request) # any-ok: botocore request mutation is untyped
- return dict(aws_request.headers) # any-ok: botocore headers are untyped
+ return MappingProxyType(dict(aws_request.headers)) # any-ok: botocore headers are untyped
def transform_file_content_response(
self,
@@ -1330,6 +1551,7 @@ class BedrockFilesConfig(BaseAWSLLM, BaseFilesConfig):
status_code=raw_response.status_code,
message=raw_response.text,
headers=raw_response.headers,
+ response=raw_response,
)
return HttpxBinaryResponseContent(response=raw_response)
diff --git a/litellm/llms/custom_httpx/llm_http_handler.py b/litellm/llms/custom_httpx/llm_http_handler.py
index 7109e6942d1..2fe4130a310 100644
--- a/litellm/llms/custom_httpx/llm_http_handler.py
+++ b/litellm/llms/custom_httpx/llm_http_handler.py
@@ -18,7 +18,7 @@ import litellm.types
import litellm.types.utils
from litellm._logging import _redact_string, verbose_logger
from litellm.anthropic_beta_headers_manager import update_headers_with_filtered_beta
-from litellm.constants import REALTIME_WEBSOCKET_MAX_MESSAGE_SIZE_BYTES
+from litellm.constants import MAX_FILE_LIST_LIMIT, REALTIME_WEBSOCKET_MAX_MESSAGE_SIZE_BYTES
from litellm.litellm_core_utils.agentic_loop_settings import (
DEFAULT_MAX_AGENTIC_LOOPS,
validated_max_agentic_loops,
@@ -4981,15 +4981,16 @@ class BaseLLMHTTPHandler:
)
try:
- response: Final = sync_httpx_client.get(url=url, headers=headers, params=params)
+ response: Final = sync_httpx_client.get(url=url, headers=headers, params=params, timeout=timeout)
except Exception as e:
raise self._handle_error(e=e, provider_config=provider_config)
- return provider_config.transform_list_files_response(
- raw_response=response,
- logging_obj=logging_obj,
- litellm_params=litellm_params,
+ files_per_page: Final = self._files_per_listing_page(
+ response, provider_config, logging_obj, litellm_params, headers, sync_httpx_client, timeout
)
+ return [ # mutable-ok: the files contract returns the listing as a list
+ listed_file for page_files in files_per_page for listed_file in page_files
+ ]
async def async_list_files(
self,
@@ -5037,16 +5038,101 @@ class BaseLLMHTTPHandler:
)
try:
- response: Final = await async_httpx_client.get(url=url, headers=headers, params=params)
+ response: Final = await async_httpx_client.get(url=url, headers=headers, params=params, timeout=timeout)
except Exception as e:
raise self._handle_error(e=e, provider_config=provider_config)
- return provider_config.transform_list_files_response(
- raw_response=response,
- logging_obj=logging_obj,
+ files_per_page: Final = self._files_per_async_listing_page(
+ response, provider_config, logging_obj, litellm_params, headers, async_httpx_client, timeout
+ )
+ return [ # mutable-ok: the files contract returns the listing as a list
+ listed_file async for page_files in files_per_page for listed_file in page_files
+ ]
+
+ def _files_per_listing_page(
+ self,
+ first_page: httpx.Response,
+ provider_config: BaseFilesConfig,
+ logging_obj: LiteLLMLoggingObj,
+ litellm_params: dict, # mutable-ok: handed to the files contract, which types it as a dict
+ headers: dict, # mutable-ok: handed to validate_environment, which types it as a dict
+ client: HTTPHandler,
+ timeout: float | httpx.Timeout | None,
+ ) -> Iterator[list[OpenAIFileObject]]: # mutable-ok: each page arrives as the list the files contract returns
+ latest_page = first_page # rebind-ok: advances one page per loop turn
+ listed_count = 0 # rebind-ok: grows per page so the listing stops at MAX_FILE_LIST_LIMIT, OpenAI's ceiling
+ while True:
+ page_files = provider_config.transform_list_files_response(
+ raw_response=latest_page, logging_obj=logging_obj, litellm_params=litellm_params
+ )
+ yield page_files[: MAX_FILE_LIST_LIMIT - listed_count]
+ listed_count += len(page_files)
+ next_request = self._next_listing_request(latest_page, provider_config, litellm_params, listed_count)
+ if next_request is None:
+ return
+ url, params = next_request
+ next_headers = self._next_listing_page_headers(provider_config, headers, litellm_params)
+ try:
+ latest_page = client.get(url=url, headers=next_headers, params=params, timeout=timeout)
+ except Exception as e: # noqa: BLE001 # _handle_error maps every failure kind, like the first page's fetch
+ raise self._handle_error(e=e, provider_config=provider_config)
+
+ async def _files_per_async_listing_page(
+ self,
+ first_page: httpx.Response,
+ provider_config: BaseFilesConfig,
+ logging_obj: LiteLLMLoggingObj,
+ litellm_params: dict, # mutable-ok: handed to the files contract, which types it as a dict
+ headers: dict, # mutable-ok: handed to validate_environment, which types it as a dict
+ client: AsyncHTTPHandler,
+ timeout: float | httpx.Timeout | None,
+ ) -> AsyncIterator[list[OpenAIFileObject]]: # mutable-ok: each page arrives as the list the files contract returns
+ latest_page = first_page # rebind-ok: advances one page per loop turn
+ listed_count = 0 # rebind-ok: grows per page so the listing stops at MAX_FILE_LIST_LIMIT, OpenAI's ceiling
+ while True:
+ page_files = provider_config.transform_list_files_response(
+ raw_response=latest_page, logging_obj=logging_obj, litellm_params=litellm_params
+ )
+ yield page_files[: MAX_FILE_LIST_LIMIT - listed_count]
+ listed_count += len(page_files)
+ next_request = self._next_listing_request(latest_page, provider_config, litellm_params, listed_count)
+ if next_request is None:
+ return
+ url, params = next_request
+ next_headers = self._next_listing_page_headers(provider_config, headers, litellm_params)
+ try:
+ latest_page = await client.get(url=url, headers=next_headers, params=params, timeout=timeout)
+ except Exception as e: # noqa: BLE001 # _handle_error maps every failure kind, like the first page's fetch
+ raise self._handle_error(e=e, provider_config=provider_config)
+
+ def _next_listing_page_headers(
+ self,
+ provider_config: BaseFilesConfig,
+ headers: dict, # mutable-ok: handed to validate_environment, which types it as a dict
+ litellm_params: dict, # mutable-ok: handed to the files contract, which types it as a dict
+ ) -> dict: # mutable-ok: validate_environment returns the header dict the files contract types
+ return provider_config.validate_environment(
+ api_key=litellm_params.get("api_key"),
+ headers=headers,
+ model="",
+ messages=[],
+ optional_params={},
litellm_params=litellm_params,
)
+ def _next_listing_request(
+ self,
+ latest_page: httpx.Response,
+ provider_config: BaseFilesConfig,
+ litellm_params: dict, # mutable-ok: handed to the files contract, which types it as a dict
+ listed_count: int,
+ ) -> tuple[str, dict[str, str]] | None: # mutable-ok: the base files contract returns the query as a dict
+ if listed_count >= MAX_FILE_LIST_LIMIT:
+ return None
+ return provider_config.transform_list_files_next_request(
+ raw_response=latest_page, optional_params={}, litellm_params=litellm_params
+ )
+
def retrieve_file_content(
self,
file_content_request: "FileContentRequest",
diff --git a/litellm/proxy/openai_files_endpoints/common_utils.py b/litellm/proxy/openai_files_endpoints/common_utils.py
index b1f282a0978..38a907892b4 100644
--- a/litellm/proxy/openai_files_endpoints/common_utils.py
+++ b/litellm/proxy/openai_files_endpoints/common_utils.py
@@ -16,6 +16,7 @@ from typing import (
)
from litellm.batches.batch_utils import batch_cost_is_final
+from litellm.constants import MAX_FILE_LIST_LIMIT
from litellm.proxy._types import ProxyException
from litellm.repositories.table_repositories import (
ManagedFileRepository,
@@ -34,8 +35,6 @@ if TYPE_CHECKING:
from litellm.types.utils import LiteLLMBatch
-MAX_FILE_LIST_LIMIT: Final = 10000
-
FILE_LIST_CONTINUATION_CHUNK_SIZE: Final = 500
BATCH_CREATE_HIDDEN_PARAM: Final = "batch_create"
diff --git a/litellm/proxy/openai_files_endpoints/files_endpoints.py b/litellm/proxy/openai_files_endpoints/files_endpoints.py
index c315d30b8f3..07cdc33e306 100644
--- a/litellm/proxy/openai_files_endpoints/files_endpoints.py
+++ b/litellm/proxy/openai_files_endpoints/files_endpoints.py
@@ -33,6 +33,7 @@ from litellm.litellm_core_utils.cloud_storage_security import (
)
from litellm.litellm_core_utils.core_helpers import get_or_create_metadata_bucket
from litellm.llms.base_llm.files.transformation import BaseFileEndpoints
+from litellm.llms.base_llm.managed_resources.isolation import build_list_page
from litellm.proxy._types import *
from litellm.proxy.auth.user_api_key_auth import user_api_key_auth
from litellm.proxy.common_request_processing import ProxyBaseLLMRequestProcessing
@@ -90,6 +91,7 @@ from litellm.router import Router
from litellm.types.llms.openai import (
CREATE_FILE_REQUESTS_PURPOSE,
FileExpiresAfter,
+ FileListPage,
OpenAIFileObject,
OpenAIFilesPurpose,
)
@@ -97,6 +99,7 @@ from litellm.types.llms.openai import (
router: Final = APIRouter()
_MAX_BATCH_FILE_SIZE_MB_ADAPTER: Final = TypeAdapter(int | None)
+_LISTED_FILES_ADAPTER: Final = TypeAdapter(list[OpenAIFileObject])
class UploadedFileInfo(TypedDict):
@@ -1287,6 +1290,11 @@ async def delete_file(
user_api_key_dict=user_api_key_dict,
managed_files_obj=proxy_logging_obj.get_proxy_hook("managed_files"),
)
+ if is_managed_cloud_storage_uri(file_id) and user_api_key_dict.user_role != LitellmUserRoles.PROXY_ADMIN:
+ raise HTTPException(
+ status_code=403,
+ detail="Raw cloud storage file ids can only be deleted by a proxy admin key. Use the LiteLLM managed file id returned when the file was created.",
+ )
custom_llm_provider: Final = (
provider
@@ -1446,6 +1454,12 @@ async def delete_file(
)
+def _as_file_list_page(response: object) -> object:
+ if not isinstance(response, list):
+ return response
+ return FileListPage(**build_list_page(_LISTED_FILES_ADAPTER.validate_python(response)))
+
+
@router.get(
"/{provider}/v1/files",
dependencies=[Depends(user_api_key_auth)],
@@ -1524,7 +1538,7 @@ async def list_files(
if should_route and credentials is not None:
# Use model-based routing with credentials from config
- prepare_data_with_credentials(data=data, credentials=credentials)
+ prepare_data_with_credentials(data=data, credentials=credentials, include_internal_credentials=True)
response = await litellm.afile_list(
custom_llm_provider=credentials["custom_llm_provider"],
purpose=purpose,
@@ -1550,7 +1564,7 @@ async def list_files(
model_id=target_model_names_list[0],
operation_context="file list",
)
- prepare_data_with_credentials(data=data, credentials=credentials)
+ prepare_data_with_credentials(data=data, credentials=credentials, include_internal_credentials=True)
response = await litellm.afile_list(
custom_llm_provider=credentials["custom_llm_provider"],
purpose=purpose,
@@ -1592,6 +1606,7 @@ async def list_files(
status_code=500,
detail="Either 'provider' or 'target_model_names' must be provided e.g. `?target_model_names=gpt-4o`",
)
+ response = _as_file_list_page(response) # rebind-ok: each dispatch branch above binds response
## POST CALL HOOKS ###
_response: Final = await proxy_logging_obj.post_call_success_hook(
diff --git a/tests/test_litellm/enterprise/proxy/test_managed_files_hook.py b/tests/test_litellm/enterprise/proxy/test_managed_files_hook.py
index 48fceb50403..cb08e00ff65 100644
--- a/tests/test_litellm/enterprise/proxy/test_managed_files_hook.py
+++ b/tests/test_litellm/enterprise/proxy/test_managed_files_hook.py
@@ -1193,7 +1193,7 @@ async def test_afile_delete_returns_managed_id_for_stored_provider_output():
assert response.id == unified_file_id
assert response.object == "file"
- assert response.filename == stored_file.filename
+ assert response.deleted is True
assert stored_file.id == provider_file_id
router.afile_delete.assert_awaited_once_with(model="model-123", file_id=provider_file_id)
table.delete.assert_awaited_once_with(where={"unified_file_id": unified_file_id})
@@ -1730,3 +1730,100 @@ async def test_batch_retrieve_hook_does_not_claim_attribution():
managed_files.store_unified_object_id.assert_awaited_once()
assert managed_files.store_unified_object_id.await_args.kwargs["persist_attribution"] is False
+
+
+@pytest.mark.asyncio
+async def test_afile_delete_passes_trusted_model_credentials_to_router():
+ """
+ afile_delete must hand the deployment's credential snapshot to the router
+ call, since Bedrock validates the s3:// file id against the bucket in it.
+ """
+ from types import MappingProxyType
+
+ managed_files = _make_managed_files_instance()
+ unified_file_id = "unified-file-id"
+ s3_uri = "s3://my-bucket/litellm-bedrock-files/job-123/input.jsonl"
+ managed_files.get_model_file_id_mapping = AsyncMock(return_value={unified_file_id: {"model-123": s3_uri}})
+ managed_files.delete_unified_file_id = AsyncMock(return_value=_make_file_object(unified_file_id))
+
+ mock_router = MagicMock()
+ mock_router.get_deployment_credentials_with_provider = MagicMock(
+ return_value={
+ "custom_llm_provider": "bedrock",
+ "s3_bucket_name": "my-bucket",
+ "aws_region_name": "us-west-2",
+ }
+ )
+ mock_router.afile_delete = AsyncMock(return_value=MagicMock())
+
+ await managed_files.afile_delete(
+ file_id=unified_file_id,
+ litellm_parent_otel_span=None,
+ llm_router=mock_router,
+ )
+
+ call_kwargs = mock_router.afile_delete.call_args.kwargs
+ assert call_kwargs["model"] == "model-123"
+ assert call_kwargs["file_id"] == s3_uri
+ trusted_credentials = call_kwargs["_litellm_internal_model_credentials"]
+ assert isinstance(trusted_credentials, MappingProxyType)
+ assert trusted_credentials["s3_bucket_name"] == "my-bucket"
+
+
+@pytest.mark.asyncio
+async def test_afile_delete_bedrock_unified_id_end_to_end(monkeypatch):
+ """
+ Proxy repro for deleting a Bedrock batch input file by unified id: the
+ s3:// object must be removed via a SigV4-signed S3 DELETE using the
+ deployment's s3_bucket_name (no AWS_S3_BUCKET_NAME env).
+
+ Regression test for "BedrockFilesConfig does not support file deletion"
+ raised on this path.
+ """
+ import httpx
+ import respx
+
+ import litellm
+ from litellm import Router
+
+ monkeypatch.delenv("AWS_S3_BUCKET_NAME", raising=False)
+ monkeypatch.setattr(litellm, "disable_aiohttp_transport", True)
+ litellm.in_memory_llm_clients_cache.flush_cache()
+
+ router = Router(
+ model_list=[
+ {
+ "model_name": "bedrock-claude",
+ "litellm_params": {
+ "model": "bedrock/anthropic.claude-3-5-sonnet-20240620-v1:0",
+ "aws_access_key_id": "AKIAEXAMPLE",
+ "aws_secret_access_key": "secret",
+ "aws_region_name": "us-west-2",
+ "s3_bucket_name": "my-bucket",
+ },
+ "model_info": {"id": "model-123"},
+ }
+ ]
+ )
+
+ managed_files = _make_managed_files_instance()
+ unified_file_id = "unified-file-id"
+ s3_uri = "s3://my-bucket/litellm-bedrock-files/job-123/input.jsonl"
+ managed_files.get_model_file_id_mapping = AsyncMock(return_value={unified_file_id: {"model-123": s3_uri}})
+ managed_files.delete_unified_file_id = AsyncMock(return_value=_make_file_object(unified_file_id))
+
+ expected_url = "https://s3.us-west-2.amazonaws.com/my-bucket/litellm-bedrock-files/job-123/input.jsonl"
+ with respx.mock:
+ route = respx.delete(expected_url).mock(return_value=httpx.Response(204))
+
+ response = await managed_files.afile_delete(
+ file_id=unified_file_id,
+ litellm_parent_otel_span=None,
+ llm_router=router,
+ )
+
+ assert route.called
+ assert route.calls[0].request.headers["Authorization"].startswith("AWS4-HMAC-SHA256")
+ assert response.id == unified_file_id
+ assert response.model_dump() == {"id": unified_file_id, "object": "file", "deleted": True}
+ managed_files.delete_unified_file_id.assert_awaited_once_with(unified_file_id, None)
diff --git a/tests/test_litellm/llms/bedrock/files/test_bedrock_files_transformation.py b/tests/test_litellm/llms/bedrock/files/test_bedrock_files_transformation.py
index 3b01a4f2054..c609455f3d8 100644
--- a/tests/test_litellm/llms/bedrock/files/test_bedrock_files_transformation.py
+++ b/tests/test_litellm/llms/bedrock/files/test_bedrock_files_transformation.py
@@ -1948,11 +1948,13 @@ class TestBedrockFileDeletion:
def test_delete_rejects_untrusted_objects_before_signing(
self, file_id: str, message: str, monkeypatch: pytest.MonkeyPatch
) -> None:
+ from litellm.llms.bedrock.common_utils import BedrockError
from litellm.llms.bedrock.files.transformation import BedrockFilesConfig
monkeypatch.setenv("AWS_S3_BUCKET_NAME", "my-bucket")
- with pytest.raises(ValueError, match=message):
+ with pytest.raises(BedrockError, match=message) as rejection:
BedrockFilesConfig().transform_delete_file_request(file_id=file_id, optional_params={}, litellm_params={})
+ assert rejection.value.status_code == 400
class TestBedrockFileContentTransformation:
@@ -2030,11 +2032,12 @@ class TestBedrockFileContentTransformation:
assert url == self.EXPECTED_URL
def test_transform_file_content_request_rejects_foreign_bucket(self, monkeypatch):
+ from litellm.llms.bedrock.common_utils import BedrockError
from litellm.llms.bedrock.files.transformation import BedrockFilesConfig
monkeypatch.setenv("AWS_S3_BUCKET_NAME", "my-bucket")
- with pytest.raises(ValueError, match="configured storage bucket"):
+ with pytest.raises(BedrockError, match="configured storage bucket") as rejection:
BedrockFilesConfig().transform_file_content_request(
file_content_request={
"file_id": "s3://other-bucket/litellm-batch-outputs/job/x.jsonl.out"
@@ -2043,18 +2046,25 @@ class TestBedrockFileContentTransformation:
litellm_params=self._litellm_params(),
)
+ assert rejection.value.status_code == 400
+
+
def test_transform_file_content_request_rejects_unmanaged_key(self, monkeypatch):
+ from litellm.llms.bedrock.common_utils import BedrockError
from litellm.llms.bedrock.files.transformation import BedrockFilesConfig
monkeypatch.setenv("AWS_S3_BUCKET_NAME", "my-bucket")
- with pytest.raises(ValueError, match="LiteLLM-managed"):
+ with pytest.raises(BedrockError, match="LiteLLM-managed") as rejection:
BedrockFilesConfig().transform_file_content_request(
file_content_request={"file_id": "s3://my-bucket/private/x.jsonl"},
optional_params={},
litellm_params=self._litellm_params(),
)
+ assert rejection.value.status_code == 400
+
+
def test_extract_s3_uri_rejects_non_managed_file_id(self):
"""A file id that is neither an s3:// URI nor a unified id must be rejected."""
from litellm.llms.bedrock.files.transformation import (
@@ -2183,12 +2193,13 @@ class TestBedrockFileContentTransformation:
def test_rejects_bucket_outside_input_and_output(self, monkeypatch):
"""A file id whose bucket is neither the input nor the output bucket is
still rejected (SSRF / bucket-confusion guard)."""
+ from litellm.llms.bedrock.common_utils import BedrockError
from litellm.llms.bedrock.files.transformation import BedrockFilesConfig
monkeypatch.delenv("AWS_S3_BUCKET_NAME", raising=False)
monkeypatch.delenv("AWS_S3_OUTPUT_BUCKET_NAME", raising=False)
- with pytest.raises(ValueError, match="configured storage bucket"):
+ with pytest.raises(BedrockError, match="configured storage bucket") as rejection:
BedrockFilesConfig().transform_file_content_request(
file_content_request={
"file_id": "s3://other-bucket/litellm-batch-outputs/job/x.jsonl.out"
@@ -2199,6 +2210,9 @@ class TestBedrockFileContentTransformation:
),
)
+ assert rejection.value.status_code == 400
+
+
def test_sign_request_without_botocore_raises_helpful_error(self, monkeypatch):
"""A missing botocore must surface an actionable 'install boto3' error
rather than a raw import failure."""
@@ -2605,6 +2619,7 @@ def test_sign_s3_request_without_body_assumes_role_with_external_id(monkeypatch)
with patch.object(boto3, "client", return_value=FakeSTSClient()):
signed_headers = BedrockFilesConfig()._sign_s3_request_without_body(
+ method="GET",
api_base="https://s3.us-east-1.amazonaws.com/safe-bucket/litellm-bedrock-files-model-id-abc.jsonl",
aws_region_name="us-east-1",
request_params=request_params,
@@ -2612,3 +2627,1058 @@ def test_sign_s3_request_without_body_assumes_role_with_external_id(monkeypatch)
authorization = {key.lower(): value for key, value in signed_headers.items()}["authorization"]
assert "ASIAFILESGETROLE" in authorization
+
+
+def _s3_signature_for(method: str, url: str, headers: Mapping[str, str]) -> str:
+ sent = {name.lower(): value for name, value in headers.items()}
+ signed_names = sent["authorization"].split("SignedHeaders=")[1].split(",")[0].split(";")
+ request = AWSRequest(
+ method=method,
+ url=url,
+ headers={name: sent[name] for name in signed_names if name in sent},
+ )
+ request.context["timestamp"] = sent["x-amz-date"]
+ signer = S3SigV4Auth(Credentials("AKIAEXAMPLE", "secret"), "s3", "us-west-2")
+ return signer.signature(signer.string_to_sign(request, signer.canonical_request(request)), request)
+
+
+def _sent_signature(headers: Mapping[str, str]) -> str:
+ authorization = {name.lower(): value for name, value in headers.items()}["authorization"]
+ return authorization.split("Signature=")[1].strip()
+
+
+def _bedrock_s3_params() -> dict:
+ return {
+ "aws_access_key_id": "AKIAEXAMPLE",
+ "aws_secret_access_key": "secret",
+ "aws_region_name": "us-west-2",
+ }
+
+
+def _trusted_bucket_snapshot(**deployment_litellm_params) -> dict:
+ from types import MappingProxyType
+
+ from litellm.types.router import CredentialLiteLLMParams
+
+ snapshot = CredentialLiteLLMParams(**deployment_litellm_params).model_dump(exclude_none=True)
+ return {**_bedrock_s3_params(), "_litellm_internal_model_credentials": MappingProxyType(snapshot)}
+
+
+class TestBedrockFileDeletionTransformation:
+ """SigV4-signed S3 DeleteObject for LiteLLM-managed Bedrock batch files."""
+
+ S3_URI = "s3://my-bucket/litellm-bedrock-files/job-123/input.jsonl"
+ EXPECTED_URL = "https://s3.us-west-2.amazonaws.com/my-bucket/litellm-bedrock-files/job-123/input.jsonl"
+
+ def test_transform_delete_file_request_signs_s3_delete(self, monkeypatch):
+ import hashlib
+
+ from litellm.llms.bedrock.files.transformation import (
+ S3_SIGNED_REQUEST_HEADERS_PARAM,
+ BedrockFilesConfig,
+ )
+
+ monkeypatch.setenv("AWS_S3_BUCKET_NAME", "my-bucket")
+ litellm_params = _bedrock_s3_params()
+
+ url, params = BedrockFilesConfig().transform_delete_file_request(
+ file_id=self.S3_URI,
+ optional_params={},
+ litellm_params=litellm_params,
+ )
+
+ assert url == self.EXPECTED_URL
+ assert params == {}
+ signed_headers = litellm_params[S3_SIGNED_REQUEST_HEADERS_PARAM]
+ lowered = {name.lower(): value for name, value in signed_headers.items()}
+ assert lowered["x-amz-content-sha256"] == hashlib.sha256(b"").hexdigest()
+ assert "/us-west-2/s3/aws4_request" in lowered["authorization"]
+ assert _sent_signature(signed_headers) == _s3_signature_for("DELETE", url, signed_headers)
+
+ def test_transform_delete_file_request_decodes_unified_file_id(self, monkeypatch):
+ import base64
+
+ from litellm.llms.bedrock.files.transformation import BedrockFilesConfig
+ from litellm.types.utils import SpecialEnums
+
+ monkeypatch.setenv("AWS_S3_BUCKET_NAME", "my-bucket")
+ unified_file_id = SpecialEnums.LITELLM_MANAGED_FILE_COMPLETE_STR.value.format(
+ "application/json", "unified-id", "", self.S3_URI, "model-id"
+ )
+ encoded_file_id = base64.urlsafe_b64encode(unified_file_id.encode()).decode().rstrip("=")
+ litellm_params = _bedrock_s3_params()
+
+ url, _ = BedrockFilesConfig().transform_delete_file_request(
+ file_id=encoded_file_id,
+ optional_params={},
+ litellm_params=litellm_params,
+ )
+
+ assert url == self.EXPECTED_URL
+
+ def test_transform_delete_file_request_uses_trusted_snapshot_bucket(self, monkeypatch):
+ from litellm.llms.bedrock.files.transformation import BedrockFilesConfig
+
+ monkeypatch.delenv("AWS_S3_BUCKET_NAME", raising=False)
+
+ url, _ = BedrockFilesConfig().transform_delete_file_request(
+ file_id=self.S3_URI,
+ optional_params={},
+ litellm_params=_trusted_bucket_snapshot(s3_bucket_name="my-bucket"),
+ )
+
+ assert url == self.EXPECTED_URL
+
+ def test_transform_delete_file_request_rejects_foreign_bucket(self, monkeypatch):
+ from litellm.llms.bedrock.common_utils import BedrockError
+ from litellm.llms.bedrock.files.transformation import BedrockFilesConfig
+
+ monkeypatch.setenv("AWS_S3_BUCKET_NAME", "my-bucket")
+
+ with pytest.raises(BedrockError, match="configured storage bucket") as rejection:
+ BedrockFilesConfig().transform_delete_file_request(
+ file_id="s3://other-bucket/litellm-bedrock-files/job-123/input.jsonl",
+ optional_params={},
+ litellm_params=_bedrock_s3_params(),
+ )
+
+ assert rejection.value.status_code == 400
+
+
+ def test_transform_delete_file_request_rejects_unmanaged_key(self, monkeypatch):
+ from litellm.llms.bedrock.common_utils import BedrockError
+ from litellm.llms.bedrock.files.transformation import BedrockFilesConfig
+
+ monkeypatch.setenv("AWS_S3_BUCKET_NAME", "my-bucket")
+
+ with pytest.raises(BedrockError, match="LiteLLM-managed") as rejection:
+ BedrockFilesConfig().transform_delete_file_request(
+ file_id="s3://my-bucket/private/x.jsonl",
+ optional_params={},
+ litellm_params=_bedrock_s3_params(),
+ )
+
+ assert rejection.value.status_code == 400
+
+
+ def test_transform_delete_file_response_echoes_the_deleted_id(self):
+ import httpx
+
+ from litellm.llms.bedrock.files.transformation import BedrockFilesConfig
+
+ deleted = BedrockFilesConfig().transform_delete_file_response(
+ raw_response=httpx.Response(204),
+ logging_obj=MagicMock(model_call_details={"additional_args": {"file_id": self.S3_URI}}),
+ litellm_params={},
+ )
+
+ assert deleted.id == self.S3_URI
+ assert deleted.deleted is True
+ assert deleted.object == "file"
+
+ def test_transform_delete_file_response_raises_on_s3_error(self):
+ import httpx
+
+ from litellm.llms.bedrock.common_utils import BedrockError
+ from litellm.llms.bedrock.files.transformation import BedrockFilesConfig
+
+ with pytest.raises(BedrockError) as excinfo:
+ BedrockFilesConfig().transform_delete_file_response(
+ raw_response=httpx.Response(403, text="AccessDenied"),
+ logging_obj=MagicMock(),
+ litellm_params={},
+ )
+
+ assert excinfo.value.status_code == 403
+
+ def test_file_delete_end_to_end_sends_signed_delete(self, monkeypatch):
+ import httpx
+ import respx
+
+ import litellm
+
+ monkeypatch.setenv("AWS_S3_BUCKET_NAME", "my-bucket")
+
+ with respx.mock:
+ route = respx.delete(self.EXPECTED_URL).mock(return_value=httpx.Response(204))
+
+ response = litellm.file_delete(
+ file_id=self.S3_URI,
+ custom_llm_provider="bedrock",
+ **_bedrock_s3_params(),
+ )
+
+ assert route.called
+ request = route.calls[0].request
+ assert request.headers["Authorization"].startswith("AWS4-HMAC-SHA256")
+ assert _sent_signature(request.headers) == _s3_signature_for("DELETE", str(request.url), request.headers)
+ assert response.id == self.S3_URI
+ assert response.deleted is True
+
+ @pytest.mark.asyncio
+ async def test_afile_delete_end_to_end_sends_signed_delete(self, monkeypatch):
+ import httpx
+ import respx
+
+ import litellm
+
+ monkeypatch.setenv("AWS_S3_BUCKET_NAME", "my-bucket")
+ monkeypatch.setattr(litellm, "disable_aiohttp_transport", True)
+ litellm.in_memory_llm_clients_cache.flush_cache()
+
+ with respx.mock:
+ route = respx.delete(self.EXPECTED_URL).mock(return_value=httpx.Response(204))
+
+ response = await litellm.afile_delete(
+ file_id=self.S3_URI,
+ custom_llm_provider="bedrock",
+ **_bedrock_s3_params(),
+ )
+
+ assert route.called
+ request = route.calls[0].request
+ assert _sent_signature(request.headers) == _s3_signature_for("DELETE", str(request.url), request.headers)
+ assert response.id == self.S3_URI
+ assert response.deleted is True
+
+ def test_file_delete_end_to_end_answers_400_for_a_foreign_bucket(self, monkeypatch):
+ import respx
+
+ import litellm
+ from litellm.llms.bedrock.common_utils import BedrockError
+
+ monkeypatch.setenv("AWS_S3_BUCKET_NAME", "my-bucket")
+
+ with respx.mock, pytest.raises(BedrockError) as rejection:
+ litellm.file_delete(
+ file_id="s3://other-bucket/litellm-bedrock-files/job-123/input.jsonl",
+ custom_llm_provider="bedrock",
+ **_bedrock_s3_params(),
+ )
+
+ assert rejection.value.status_code == 400
+ assert "configured storage bucket" in rejection.value.message
+
+ def test_file_delete_end_to_end_answers_400_for_a_non_managed_id(self, monkeypatch):
+ import respx
+
+ import litellm
+ from litellm.llms.bedrock.common_utils import BedrockError
+
+ monkeypatch.setenv("AWS_S3_BUCKET_NAME", "my-bucket")
+
+ with respx.mock, pytest.raises(BedrockError) as rejection:
+ litellm.file_delete(file_id="file-1234567890", custom_llm_provider="bedrock", **_bedrock_s3_params())
+
+ assert rejection.value.status_code == 400
+ assert "managed LiteLLM S3 file id" in rejection.value.message
+
+ def test_file_delete_end_to_end_surfaces_the_s3_error_body(self, monkeypatch):
+ import httpx
+ import respx
+
+ import litellm
+ from litellm.llms.bedrock.common_utils import BedrockError
+
+ monkeypatch.setenv("AWS_S3_BUCKET_NAME", "my-bucket")
+
+ with respx.mock:
+ respx.delete(self.EXPECTED_URL).mock(
+ return_value=httpx.Response(403, text="AccessDenied")
+ )
+ with pytest.raises(BedrockError) as denied:
+ litellm.file_delete(file_id=self.S3_URI, custom_llm_provider="bedrock", **_bedrock_s3_params())
+
+ assert denied.value.status_code == 403
+ assert "AccessDenied" in denied.value.message
+
+
+class TestBedrockFileListTransformation:
+ """SigV4-signed S3 ListObjectsV2 over the LiteLLM-managed key prefixes."""
+
+ BUCKET_URL = "https://s3.us-west-2.amazonaws.com/my-bucket/"
+ MANAGED_QUERY = {"list-type": "2", "prefix": "litellm-b"}
+ BATCH_QUERY = {"list-type": "2", "prefix": "litellm-bedrock-files"}
+ OUTPUT_QUERY = {"list-type": "2", "prefix": "litellm-batch-outputs/"}
+ OUTPUT_BUCKET_URL = "https://s3.us-west-2.amazonaws.com/my-output-bucket/"
+ OUTPUT_BUCKET_LISTING = b"""
+
+ my-output-bucket
+ litellm-batch-outputs/
+
+ litellm-batch-outputs/job-9/input.jsonl.out
+ 2026-09-04T08:00:00.000Z
+ 70
+
+"""
+ OUTPUT_BUCKET_ID = "s3://my-output-bucket/litellm-batch-outputs/job-9/input.jsonl.out"
+ LISTING = b"""
+
+ my-bucket
+ litellm-b
+ 4
+ false
+
+ litellm-bedrock-files-model-abc.jsonl
+ 2026-09-01T10:00:00.000Z
+ 120
+
+
+ litellm-bedrock-files/job-123/input.jsonl
+ 2026-09-02T11:30:00.000Z
+ 340
+
+
+ litellm-batch-outputs/job-123/input.jsonl.out
+ 2026-09-03T12:45:00.000Z
+ 560
+
+
+ litellm-bogus/other.jsonl
+ 2026-09-03T12:45:00.000Z
+ 1
+
+"""
+ BATCH_IDS = (
+ "s3://my-bucket/litellm-bedrock-files-model-abc.jsonl",
+ "s3://my-bucket/litellm-bedrock-files/job-123/input.jsonl",
+ )
+ OUTPUT_ID = "s3://my-bucket/litellm-batch-outputs/job-123/input.jsonl.out"
+ CONTINUATION_TOKEN = "1ueGcxLPRx1Tr/XYExHnhbYLgveDs2J/wm36Hy4vbOwM="
+ FIRST_PAGE = b"""
+
+ my-bucket
+ litellm-bedrock-files
+ 1
+ 1
+ true
+ 1ueGcxLPRx1Tr/XYExHnhbYLgveDs2J/wm36Hy4vbOwM=
+
+ litellm-bedrock-files/job-1/input.jsonl
+ 2026-09-01T10:00:00.000Z
+ 10
+
+"""
+ LAST_PAGE = b"""
+
+ my-bucket
+ litellm-bedrock-files
+ 1
+ 1
+ false
+ 1ueGcxLPRx1Tr/XYExHnhbYLgveDs2J/wm36Hy4vbOwM=
+
+ litellm-bedrock-files/job-2/input.jsonl
+ 2026-09-02T10:00:00.000Z
+ 20
+
+"""
+ PAGED_IDS = (
+ "s3://my-bucket/litellm-bedrock-files/job-1/input.jsonl",
+ "s3://my-bucket/litellm-bedrock-files/job-2/input.jsonl",
+ )
+
+ def test_transform_list_files_request_signs_managed_prefix_listing(self, monkeypatch):
+ from litellm.llms.bedrock.files.transformation import (
+ LIST_FILES_PURPOSE_PARAM,
+ S3_SIGNED_REQUEST_HEADERS_PARAM,
+ BedrockFilesConfig,
+ )
+
+ monkeypatch.setenv("AWS_S3_BUCKET_NAME", "my-bucket")
+ litellm_params = _bedrock_s3_params()
+
+ url, params = BedrockFilesConfig().transform_list_files_request(
+ purpose="batch",
+ optional_params={},
+ litellm_params=litellm_params,
+ )
+
+ assert url == self.BUCKET_URL
+ assert params == self.BATCH_QUERY
+ signed_headers = litellm_params[S3_SIGNED_REQUEST_HEADERS_PARAM]
+ assert _sent_signature(signed_headers) == _s3_signature_for(
+ "GET", f"{url}?list-type=2&prefix=litellm-bedrock-files", signed_headers
+ )
+ assert litellm_params[LIST_FILES_PURPOSE_PARAM] == "batch"
+
+ def test_transform_list_files_request_scopes_to_configured_prefix(self, monkeypatch):
+ from litellm.llms.bedrock.files.transformation import (
+ S3_SIGNED_REQUEST_HEADERS_PARAM,
+ BedrockFilesConfig,
+ )
+
+ monkeypatch.setenv("AWS_S3_BUCKET_NAME", "my-bucket/LLM AI Projects")
+ litellm_params = _bedrock_s3_params()
+
+ url, params = BedrockFilesConfig().transform_list_files_request(
+ purpose=None,
+ optional_params={},
+ litellm_params=litellm_params,
+ )
+
+ assert url == self.BUCKET_URL
+ assert params == {"list-type": "2", "prefix": "LLM AI Projects/litellm-b"}
+ signed_headers = litellm_params[S3_SIGNED_REQUEST_HEADERS_PARAM]
+ assert _sent_signature(signed_headers) == _s3_signature_for(
+ "GET", f"{url}?list-type=2&prefix=LLM%20AI%20Projects%2Flitellm-b", signed_headers
+ )
+
+ def test_transform_list_files_request_uses_trusted_snapshot_bucket(self, monkeypatch):
+ from litellm.llms.bedrock.files.transformation import BedrockFilesConfig
+
+ monkeypatch.delenv("AWS_S3_BUCKET_NAME", raising=False)
+
+ url, params = BedrockFilesConfig().transform_list_files_request(
+ purpose=None,
+ optional_params={},
+ litellm_params=_trusted_bucket_snapshot(s3_bucket_name="my-bucket"),
+ )
+
+ assert url == self.BUCKET_URL
+ assert params == self.MANAGED_QUERY
+
+ def _list_response(self, purpose: str | None, listing: bytes | None = None, status_code: int = 200):
+ import httpx
+
+ from litellm.llms.bedrock.files.transformation import (
+ LIST_FILES_PURPOSE_PARAM,
+ BedrockFilesConfig,
+ )
+
+ return BedrockFilesConfig().transform_list_files_response(
+ raw_response=httpx.Response(status_code, content=listing if listing is not None else self.LISTING),
+ logging_obj=MagicMock(),
+ litellm_params={**_bedrock_s3_params(), LIST_FILES_PURPOSE_PARAM: purpose},
+ )
+
+ def test_transform_list_files_response_maps_managed_objects(self, monkeypatch):
+ from datetime import datetime, timezone
+
+ monkeypatch.setenv("AWS_S3_BUCKET_NAME", "my-bucket")
+
+ files = self._list_response(purpose=None)
+
+ assert [file.id for file in files] == [*self.BATCH_IDS, self.OUTPUT_ID]
+ assert [file.purpose for file in files] == ["batch", "batch", "batch_output"]
+ assert [file.bytes for file in files] == [120, 340, 560]
+ assert [file.filename for file in files] == [
+ "litellm-bedrock-files-model-abc.jsonl",
+ "input.jsonl",
+ "input.jsonl.out",
+ ]
+ assert files[1].created_at == int(datetime(2026, 9, 2, 11, 30, tzinfo=timezone.utc).timestamp())
+ assert {file.object for file in files} == {"file"}
+ assert {file.status for file in files} == {"uploaded"}
+
+ def test_transform_list_files_response_filters_by_purpose(self, monkeypatch):
+ monkeypatch.setenv("AWS_S3_BUCKET_NAME", "my-bucket")
+
+ assert [file.id for file in self._list_response(purpose="batch")] == list(self.BATCH_IDS)
+ assert [file.id for file in self._list_response(purpose="batch_output")] == [self.OUTPUT_ID]
+
+ def test_transform_list_files_response_scopes_to_configured_prefix(self, monkeypatch):
+ monkeypatch.setenv("AWS_S3_BUCKET_NAME", "my-bucket/team-a")
+ listing = b"""
+
+ my-bucket
+ team-a/litellm-bedrock-files/job-1/input.jsonl10
+ team-a/litellm-batch-outputs/job-1/input.jsonl.out20
+ litellm-bedrock-files/job-2/input.jsonl30
+"""
+
+ files = self._list_response(purpose=None, listing=listing)
+
+ assert [(file.id, file.purpose) for file in files] == [
+ ("s3://my-bucket/team-a/litellm-bedrock-files/job-1/input.jsonl", "batch"),
+ ("s3://my-bucket/team-a/litellm-batch-outputs/job-1/input.jsonl.out", "batch_output"),
+ ]
+
+ def test_transform_list_files_response_raises_on_s3_error(self, monkeypatch):
+ from litellm.llms.bedrock.common_utils import BedrockError
+
+ monkeypatch.setenv("AWS_S3_BUCKET_NAME", "my-bucket")
+
+ with pytest.raises(BedrockError) as excinfo:
+ self._list_response(purpose=None, listing=b"AccessDenied", status_code=403)
+
+ assert excinfo.value.status_code == 403
+
+ def test_file_list_end_to_end_sends_signed_listing(self, monkeypatch):
+ import httpx
+ import respx
+
+ import litellm
+
+ monkeypatch.setenv("AWS_S3_BUCKET_NAME", "my-bucket")
+
+ with respx.mock:
+ route = respx.get(self.BUCKET_URL, params__contains=self.BATCH_QUERY).mock(
+ return_value=httpx.Response(200, content=self.LISTING)
+ )
+
+ files = litellm.file_list(custom_llm_provider="bedrock", purpose="batch", **_bedrock_s3_params())
+
+ assert route.called
+ request = route.calls[0].request
+ assert request.headers["Authorization"].startswith("AWS4-HMAC-SHA256")
+ assert _sent_signature(request.headers) == _s3_signature_for("GET", str(request.url), request.headers)
+ assert [file.id for file in files] == list(self.BATCH_IDS)
+
+ def test_file_list_uses_trusted_snapshot_bucket_without_env(self, monkeypatch):
+ import httpx
+ import respx
+
+ import litellm
+
+ monkeypatch.delenv("AWS_S3_BUCKET_NAME", raising=False)
+
+ with respx.mock:
+ route = respx.get(self.BUCKET_URL, params__contains=self.MANAGED_QUERY).mock(
+ return_value=httpx.Response(200, content=self.LISTING)
+ )
+
+ files = litellm.file_list(
+ custom_llm_provider="bedrock",
+ **_trusted_bucket_snapshot(s3_bucket_name="my-bucket"),
+ )
+
+ assert route.called
+ assert [file.id for file in files] == [*self.BATCH_IDS, self.OUTPUT_ID]
+
+ @pytest.mark.asyncio
+ async def test_afile_list_end_to_end_sends_signed_listing(self, monkeypatch):
+ import httpx
+ import respx
+
+ import litellm
+
+ monkeypatch.setenv("AWS_S3_BUCKET_NAME", "my-bucket")
+ monkeypatch.setattr(litellm, "disable_aiohttp_transport", True)
+ litellm.in_memory_llm_clients_cache.flush_cache()
+
+ with respx.mock:
+ route = respx.get(self.BUCKET_URL, params__contains=self.OUTPUT_QUERY).mock(
+ return_value=httpx.Response(200, content=self.LISTING)
+ )
+
+ files = await litellm.afile_list(
+ custom_llm_provider="bedrock", purpose="batch_output", **_bedrock_s3_params()
+ )
+
+ assert route.called
+ request = route.calls[0].request
+ assert _sent_signature(request.headers) == _s3_signature_for("GET", str(request.url), request.headers)
+ assert [file.id for file in files] == [self.OUTPUT_ID]
+
+ def test_transform_list_files_request_narrows_prefix_to_requested_purpose(self, monkeypatch):
+ from litellm.llms.bedrock.files.transformation import BedrockFilesConfig
+
+ monkeypatch.setenv("AWS_S3_BUCKET_NAME", "my-bucket")
+ monkeypatch.delenv("AWS_S3_OUTPUT_BUCKET_NAME", raising=False)
+
+ batch_url, batch_params = BedrockFilesConfig().transform_list_files_request(
+ purpose="batch", optional_params={}, litellm_params=_bedrock_s3_params()
+ )
+ output_url, output_params = BedrockFilesConfig().transform_list_files_request(
+ purpose="batch_output", optional_params={}, litellm_params=_bedrock_s3_params()
+ )
+
+ assert (batch_url, batch_params) == (self.BUCKET_URL, self.BATCH_QUERY)
+ assert (output_url, output_params) == (self.BUCKET_URL, self.OUTPUT_QUERY)
+
+ def test_transform_list_files_request_lists_configured_output_bucket_for_batch_output(self, monkeypatch):
+ from litellm.llms.bedrock.files.transformation import (
+ S3_SIGNED_REQUEST_HEADERS_PARAM,
+ BedrockFilesConfig,
+ )
+
+ monkeypatch.delenv("AWS_S3_BUCKET_NAME", raising=False)
+ monkeypatch.delenv("AWS_S3_OUTPUT_BUCKET_NAME", raising=False)
+ litellm_params = _trusted_bucket_snapshot(
+ s3_bucket_name="my-bucket", s3_output_bucket_name="my-output-bucket/team-a"
+ )
+
+ url, params = BedrockFilesConfig().transform_list_files_request(
+ purpose="batch_output", optional_params={}, litellm_params=litellm_params
+ )
+ signed_headers = litellm_params[S3_SIGNED_REQUEST_HEADERS_PARAM]
+ input_url, input_params = BedrockFilesConfig().transform_list_files_request(
+ purpose="batch", optional_params={}, litellm_params=dict(litellm_params)
+ )
+
+ assert url == self.OUTPUT_BUCKET_URL
+ assert params == {"list-type": "2", "prefix": "team-a/litellm-batch-outputs/"}
+ assert _sent_signature(signed_headers) == _s3_signature_for(
+ "GET", f"{url}?list-type=2&prefix=team-a%2Flitellm-batch-outputs%2F", signed_headers
+ )
+ assert (input_url, input_params) == (self.BUCKET_URL, self.BATCH_QUERY)
+
+ def test_transform_list_files_request_reads_output_bucket_from_env(self, monkeypatch):
+ from litellm.llms.bedrock.files.transformation import BedrockFilesConfig
+
+ monkeypatch.setenv("AWS_S3_BUCKET_NAME", "my-bucket")
+ monkeypatch.setenv("AWS_S3_OUTPUT_BUCKET_NAME", "my-output-bucket")
+
+ url, params = BedrockFilesConfig().transform_list_files_request(
+ purpose="batch_output", optional_params={}, litellm_params=_bedrock_s3_params()
+ )
+
+ assert (url, params) == (self.OUTPUT_BUCKET_URL, self.OUTPUT_QUERY)
+
+ EMPTY_LISTING = b"""
+
+ my-bucket
+
+ 0
+ 0
+ false
+"""
+ NO_KEYS_QUERY = {"list-type": "2", "max-keys": "0"}
+
+ def test_transform_list_files_request_asks_for_no_keys_when_bedrock_never_stores_the_purpose(self, monkeypatch):
+ import httpx
+
+ from litellm.llms.bedrock.files.transformation import BedrockFilesConfig
+
+ monkeypatch.setenv("AWS_S3_BUCKET_NAME", "my-bucket")
+ litellm_params = _bedrock_s3_params()
+ config = BedrockFilesConfig()
+
+ url, params = config.transform_list_files_request(
+ purpose="user_data", optional_params={}, litellm_params=litellm_params
+ )
+ next_request = config.transform_list_files_next_request(
+ raw_response=httpx.Response(200, content=self.EMPTY_LISTING),
+ optional_params={},
+ litellm_params=litellm_params,
+ )
+
+ assert (url, params) == (self.BUCKET_URL, self.NO_KEYS_QUERY)
+ assert next_request is None
+
+ def test_file_list_never_walks_the_bucket_for_a_purpose_bedrock_never_stores(self, monkeypatch):
+ import httpx
+ import respx
+
+ import litellm
+
+ monkeypatch.setenv("AWS_S3_BUCKET_NAME", "my-bucket")
+
+ with respx.mock:
+ route = respx.get(self.BUCKET_URL, params__contains=self.NO_KEYS_QUERY).mock(
+ return_value=httpx.Response(200, content=self.EMPTY_LISTING)
+ )
+
+ files = litellm.file_list(custom_llm_provider="bedrock", purpose="user_data", **_bedrock_s3_params())
+
+ assert route.call_count == 1
+ assert "prefix" not in route.calls[0].request.url.params
+ assert files == []
+
+ def test_transform_list_files_request_lists_the_output_bucket_without_an_input_bucket(self, monkeypatch):
+ from litellm.llms.bedrock.files.transformation import BedrockFilesConfig
+
+ monkeypatch.delenv("AWS_S3_BUCKET_NAME", raising=False)
+ monkeypatch.delenv("AWS_S3_OUTPUT_BUCKET_NAME", raising=False)
+ litellm_params = _trusted_bucket_snapshot(s3_output_bucket_name="my-output-bucket")
+
+ url, params = BedrockFilesConfig().transform_list_files_request(
+ purpose="batch_output", optional_params={}, litellm_params=litellm_params
+ )
+
+ assert (url, params) == (self.OUTPUT_BUCKET_URL, self.OUTPUT_QUERY)
+ with pytest.raises(ValueError, match="s3_bucket_name"):
+ BedrockFilesConfig().transform_list_files_request(
+ purpose="batch", optional_params={}, litellm_params=dict(litellm_params)
+ )
+
+ def test_transform_list_files_response_accepts_output_bucket_objects(self, monkeypatch):
+ import httpx
+
+ from litellm.llms.bedrock.files.transformation import (
+ LIST_FILES_PURPOSE_PARAM,
+ BedrockFilesConfig,
+ )
+
+ monkeypatch.delenv("AWS_S3_BUCKET_NAME", raising=False)
+ monkeypatch.delenv("AWS_S3_OUTPUT_BUCKET_NAME", raising=False)
+
+ files = BedrockFilesConfig().transform_list_files_response(
+ raw_response=httpx.Response(200, content=self.OUTPUT_BUCKET_LISTING),
+ logging_obj=MagicMock(),
+ litellm_params={
+ **_trusted_bucket_snapshot(s3_bucket_name="my-bucket", s3_output_bucket_name="my-output-bucket"),
+ LIST_FILES_PURPOSE_PARAM: "batch_output",
+ },
+ )
+
+ assert [(file.id, file.purpose, file.bytes) for file in files] == [(self.OUTPUT_BUCKET_ID, "batch_output", 70)]
+
+ def test_file_list_batch_output_end_to_end_lists_output_bucket(self, monkeypatch):
+ import httpx
+ import respx
+
+ import litellm
+
+ monkeypatch.delenv("AWS_S3_BUCKET_NAME", raising=False)
+ monkeypatch.delenv("AWS_S3_OUTPUT_BUCKET_NAME", raising=False)
+
+ with respx.mock:
+ route = respx.get(self.OUTPUT_BUCKET_URL, params__contains=self.OUTPUT_QUERY).mock(
+ return_value=httpx.Response(200, content=self.OUTPUT_BUCKET_LISTING)
+ )
+
+ files = litellm.file_list(
+ custom_llm_provider="bedrock",
+ purpose="batch_output",
+ **_trusted_bucket_snapshot(s3_bucket_name="my-bucket", s3_output_bucket_name="my-output-bucket"),
+ )
+
+ assert route.called
+ request = route.calls[0].request
+ assert _sent_signature(request.headers) == _s3_signature_for("GET", str(request.url), request.headers)
+ assert [file.id for file in files] == [self.OUTPUT_BUCKET_ID]
+
+ def test_file_list_without_purpose_also_walks_a_separate_output_bucket(self, monkeypatch):
+ import httpx
+ import respx
+
+ import litellm
+
+ monkeypatch.delenv("AWS_S3_BUCKET_NAME", raising=False)
+ monkeypatch.delenv("AWS_S3_OUTPUT_BUCKET_NAME", raising=False)
+
+ with respx.mock:
+ input_route = respx.get(self.BUCKET_URL, params__contains=self.MANAGED_QUERY).mock(
+ return_value=httpx.Response(200, content=self.LISTING)
+ )
+ output_route = respx.get(self.OUTPUT_BUCKET_URL, params__contains=self.OUTPUT_QUERY).mock(
+ return_value=httpx.Response(200, content=self.OUTPUT_BUCKET_LISTING)
+ )
+
+ files = litellm.file_list(
+ custom_llm_provider="bedrock",
+ **_trusted_bucket_snapshot(s3_bucket_name="my-bucket", s3_output_bucket_name="my-output-bucket"),
+ )
+
+ assert (input_route.call_count, output_route.call_count) == (1, 1)
+ output_request = output_route.calls[0].request
+ assert _sent_signature(output_request.headers) == _s3_signature_for(
+ "GET", str(output_request.url), output_request.headers
+ )
+ assert [file.id for file in files] == [*self.BATCH_IDS, self.OUTPUT_ID, self.OUTPUT_BUCKET_ID]
+
+ def test_file_list_without_purpose_walks_the_output_bucket_after_the_last_input_page(self, monkeypatch):
+ import httpx
+ import respx
+
+ import litellm
+
+ monkeypatch.delenv("AWS_S3_BUCKET_NAME", raising=False)
+ monkeypatch.delenv("AWS_S3_OUTPUT_BUCKET_NAME", raising=False)
+
+ with respx.mock:
+ respx.get(self.BUCKET_URL, params__contains={"continuation-token": self.CONTINUATION_TOKEN}).mock(
+ return_value=httpx.Response(200, content=self.LAST_PAGE)
+ )
+ respx.get(self.BUCKET_URL, params__contains=self.MANAGED_QUERY).mock(
+ return_value=httpx.Response(200, content=self.FIRST_PAGE)
+ )
+ respx.get(self.OUTPUT_BUCKET_URL, params__contains=self.OUTPUT_QUERY).mock(
+ return_value=httpx.Response(200, content=self.OUTPUT_BUCKET_LISTING)
+ )
+
+ files = litellm.file_list(
+ custom_llm_provider="bedrock",
+ **_trusted_bucket_snapshot(s3_bucket_name="my-bucket", s3_output_bucket_name="my-output-bucket"),
+ )
+ requested_urls = [str(call.request.url) for call in respx.calls]
+
+ assert requested_urls == [
+ f"{self.BUCKET_URL}?list-type=2&prefix=litellm-b",
+ f"{self.BUCKET_URL}?list-type=2&prefix=litellm-b"
+ "&continuation-token=1ueGcxLPRx1Tr%2FXYExHnhbYLgveDs2J%2Fwm36Hy4vbOwM%3D",
+ f"{self.OUTPUT_BUCKET_URL}?list-type=2&prefix=litellm-batch-outputs%2F",
+ ]
+ assert [file.id for file in files] == [*self.PAGED_IDS, self.OUTPUT_BUCKET_ID]
+
+ @pytest.mark.parametrize(
+ ("purpose", "bucket_snapshot"),
+ [
+ pytest.param(None, {"s3_bucket_name": "my-bucket"}, id="outputs-share-the-input-bucket"),
+ pytest.param(
+ "batch",
+ {"s3_bucket_name": "my-bucket", "s3_output_bucket_name": "my-output-bucket"},
+ id="input-purpose-requested",
+ ),
+ ],
+ )
+ def test_file_list_leaves_the_output_bucket_alone_unless_an_unfiltered_list_needs_it(
+ self, monkeypatch, purpose, bucket_snapshot
+ ):
+ import httpx
+ import respx
+
+ import litellm
+
+ monkeypatch.delenv("AWS_S3_BUCKET_NAME", raising=False)
+ monkeypatch.delenv("AWS_S3_OUTPUT_BUCKET_NAME", raising=False)
+
+ with respx.mock:
+ input_route = respx.get(self.BUCKET_URL).mock(return_value=httpx.Response(200, content=self.LISTING))
+ output_route = respx.get(self.OUTPUT_BUCKET_URL).mock(
+ return_value=httpx.Response(200, content=self.OUTPUT_BUCKET_LISTING)
+ )
+
+ files = litellm.file_list(
+ custom_llm_provider="bedrock", purpose=purpose, **_trusted_bucket_snapshot(**bucket_snapshot)
+ )
+
+ assert (input_route.call_count, output_route.call_count) == (1, 0)
+ assert [file.id for file in files] == [*self.BATCH_IDS, *(() if purpose else (self.OUTPUT_ID,))]
+
+ def test_transform_list_files_next_request_walks_an_output_prefix_inside_the_input_bucket(self, monkeypatch):
+ import httpx
+
+ from litellm.llms.bedrock.files.transformation import (
+ S3_SIGNED_REQUEST_HEADERS_PARAM,
+ BedrockFilesConfig,
+ )
+
+ monkeypatch.delenv("AWS_S3_BUCKET_NAME", raising=False)
+ monkeypatch.delenv("AWS_S3_OUTPUT_BUCKET_NAME", raising=False)
+ litellm_params = _trusted_bucket_snapshot(s3_bucket_name="my-bucket", s3_output_bucket_name="my-bucket/out")
+ config = BedrockFilesConfig()
+ config.transform_list_files_request(purpose=None, optional_params={}, litellm_params=litellm_params)
+ litellm_params.pop(S3_SIGNED_REQUEST_HEADERS_PARAM)
+
+ output_request = config.transform_list_files_next_request(
+ raw_response=httpx.Response(200, content=self.LISTING), optional_params={}, litellm_params=litellm_params
+ )
+ after_output_request = config.transform_list_files_next_request(
+ raw_response=httpx.Response(200, content=self.LISTING), optional_params={}, litellm_params=litellm_params
+ )
+
+ assert output_request == (self.BUCKET_URL, {"list-type": "2", "prefix": "out/litellm-batch-outputs/"})
+ signed_headers = litellm_params[S3_SIGNED_REQUEST_HEADERS_PARAM]
+ assert _sent_signature(signed_headers) == _s3_signature_for(
+ "GET", f"{self.BUCKET_URL}?list-type=2&prefix=out%2Flitellm-batch-outputs%2F", signed_headers
+ )
+ assert after_output_request is None
+
+ def test_transform_list_files_next_request_signs_the_continuation_page(self, monkeypatch):
+ import httpx
+
+ from litellm.llms.bedrock.files.transformation import (
+ S3_SIGNED_REQUEST_HEADERS_PARAM,
+ BedrockFilesConfig,
+ )
+
+ monkeypatch.setenv("AWS_S3_BUCKET_NAME", "my-bucket")
+ litellm_params = _bedrock_s3_params()
+ config = BedrockFilesConfig()
+ config.transform_list_files_request(purpose="batch", optional_params={}, litellm_params=litellm_params)
+ first_signature = _sent_signature(litellm_params.pop(S3_SIGNED_REQUEST_HEADERS_PARAM))
+
+ next_request = config.transform_list_files_next_request(
+ raw_response=httpx.Response(200, content=self.FIRST_PAGE),
+ optional_params={},
+ litellm_params=litellm_params,
+ )
+
+ assert next_request == (self.BUCKET_URL, {**self.BATCH_QUERY, "continuation-token": self.CONTINUATION_TOKEN})
+ signed_headers = litellm_params[S3_SIGNED_REQUEST_HEADERS_PARAM]
+ signed_url = (
+ f"{self.BUCKET_URL}?list-type=2&prefix=litellm-bedrock-files"
+ "&continuation-token=1ueGcxLPRx1Tr%2FXYExHnhbYLgveDs2J%2Fwm36Hy4vbOwM%3D"
+ )
+ assert _sent_signature(signed_headers) == _s3_signature_for("GET", signed_url, signed_headers)
+ assert _sent_signature(signed_headers) != first_signature
+
+ @pytest.mark.parametrize(
+ ("status_code", "content"),
+ [
+ pytest.param(200, LAST_PAGE, id="last-page"),
+ pytest.param(403, b"AccessDenied", id="error-page"),
+ ],
+ )
+ def test_transform_list_files_next_request_stops_after_the_last_page(self, monkeypatch, status_code, content):
+ import httpx
+
+ from litellm.llms.bedrock.files.transformation import (
+ S3_SIGNED_REQUEST_HEADERS_PARAM,
+ BedrockFilesConfig,
+ )
+
+ monkeypatch.setenv("AWS_S3_BUCKET_NAME", "my-bucket")
+ litellm_params = _bedrock_s3_params()
+
+ next_request = BedrockFilesConfig().transform_list_files_next_request(
+ raw_response=httpx.Response(status_code, content=content),
+ optional_params={},
+ litellm_params=litellm_params,
+ )
+
+ assert next_request is None
+ assert S3_SIGNED_REQUEST_HEADERS_PARAM not in litellm_params
+
+ def _mock_paged_listing(self, respx_module):
+ import httpx
+
+ last_page = respx_module.get(
+ self.BUCKET_URL, params__contains={"continuation-token": self.CONTINUATION_TOKEN}
+ ).mock(return_value=httpx.Response(200, content=self.LAST_PAGE))
+ first_page = respx_module.get(self.BUCKET_URL, params__contains=self.BATCH_QUERY).mock(
+ return_value=httpx.Response(200, content=self.FIRST_PAGE)
+ )
+ return first_page, last_page
+
+ def _assert_paged_listing(self, first_page, last_page, files):
+ assert (first_page.call_count, last_page.call_count) == (1, 1)
+ read_timeouts = [call.request.extensions["timeout"]["read"] for call in (*first_page.calls, *last_page.calls)]
+ assert read_timeouts == [12.0, 12.0]
+ assert "continuation-token" not in str(first_page.calls[0].request.url)
+ last_request = last_page.calls[0].request
+ assert _sent_signature(last_request.headers) == _s3_signature_for(
+ "GET", str(last_request.url), last_request.headers
+ )
+ assert [file.id for file in files] == list(self.PAGED_IDS)
+
+ def test_file_list_follows_continuation_tokens_across_pages(self, monkeypatch):
+ import respx
+
+ import litellm
+
+ monkeypatch.delenv("AWS_S3_BUCKET_NAME", raising=False)
+ monkeypatch.delenv("AWS_S3_OUTPUT_BUCKET_NAME", raising=False)
+
+ with respx.mock:
+ first_page, last_page = self._mock_paged_listing(respx)
+ files = litellm.file_list(
+ custom_llm_provider="bedrock",
+ purpose="batch",
+ timeout=12,
+ **_trusted_bucket_snapshot(s3_bucket_name="my-bucket"),
+ )
+
+ self._assert_paged_listing(first_page, last_page, files)
+
+ @pytest.mark.asyncio
+ async def test_afile_list_follows_continuation_tokens_across_pages(self, monkeypatch):
+ import respx
+
+ import litellm
+
+ monkeypatch.delenv("AWS_S3_BUCKET_NAME", raising=False)
+ monkeypatch.delenv("AWS_S3_OUTPUT_BUCKET_NAME", raising=False)
+ monkeypatch.setattr(litellm, "disable_aiohttp_transport", True)
+ litellm.in_memory_llm_clients_cache.flush_cache()
+
+ with respx.mock:
+ first_page, last_page = self._mock_paged_listing(respx)
+ files = await litellm.afile_list(
+ custom_llm_provider="bedrock",
+ purpose="batch",
+ timeout=12,
+ **_trusted_bucket_snapshot(s3_bucket_name="my-bucket"),
+ )
+
+ self._assert_paged_listing(first_page, last_page, files)
+
+ OVERSIZED_PAGE_SIZE = 3000
+ OVERSIZED_PAGE_COUNT = 6
+
+ def _oversized_listing_page(self, page_index: int) -> bytes:
+ contents = "".join(
+ f"litellm-bedrock-files/page-{page_index}/obj-{index}.jsonl"
+ "2026-09-01T10:00:00.000Z1"
+ for index in range(self.OVERSIZED_PAGE_SIZE)
+ )
+ continuation = (
+ f"truepage-{page_index + 1}"
+ if page_index < self.OVERSIZED_PAGE_COUNT - 1
+ else "false"
+ )
+ return (
+ ''
+ ''
+ f"{continuation}{contents}"
+ ).encode()
+
+ def _mock_oversized_listing(self, respx_module):
+ import httpx
+
+ def page_for(request):
+ token = request.url.params.get("continuation-token", "page-0")
+ return httpx.Response(200, content=self._oversized_listing_page(int(token.removeprefix("page-"))))
+
+ return respx_module.get(self.BUCKET_URL, params__contains=self.BATCH_QUERY).mock(side_effect=page_for)
+
+ def _assert_capped_listing(self, route, files):
+ from litellm.constants import MAX_FILE_LIST_LIMIT
+
+ pages_needed = -(-MAX_FILE_LIST_LIMIT // self.OVERSIZED_PAGE_SIZE)
+ last_index = MAX_FILE_LIST_LIMIT - (pages_needed - 1) * self.OVERSIZED_PAGE_SIZE - 1
+ assert pages_needed < self.OVERSIZED_PAGE_COUNT
+ assert route.call_count == pages_needed
+ assert len(files) == MAX_FILE_LIST_LIMIT
+ assert files[-1].id == f"s3://my-bucket/litellm-bedrock-files/page-{pages_needed - 1}/obj-{last_index}.jsonl"
+
+ def test_file_list_stops_at_the_openai_listing_ceiling(self, monkeypatch):
+ import respx
+
+ import litellm
+
+ monkeypatch.delenv("AWS_S3_BUCKET_NAME", raising=False)
+ monkeypatch.delenv("AWS_S3_OUTPUT_BUCKET_NAME", raising=False)
+
+ with respx.mock:
+ route = self._mock_oversized_listing(respx)
+ files = litellm.file_list(
+ custom_llm_provider="bedrock",
+ purpose="batch",
+ **_trusted_bucket_snapshot(s3_bucket_name="my-bucket"),
+ )
+
+ self._assert_capped_listing(route, files)
+
+ @pytest.mark.asyncio
+ async def test_afile_list_stops_at_the_openai_listing_ceiling(self, monkeypatch):
+ import respx
+
+ import litellm
+
+ monkeypatch.delenv("AWS_S3_BUCKET_NAME", raising=False)
+ monkeypatch.delenv("AWS_S3_OUTPUT_BUCKET_NAME", raising=False)
+ monkeypatch.setattr(litellm, "disable_aiohttp_transport", True)
+ litellm.in_memory_llm_clients_cache.flush_cache()
+
+ with respx.mock:
+ route = self._mock_oversized_listing(respx)
+ files = await litellm.afile_list(
+ custom_llm_provider="bedrock",
+ purpose="batch",
+ **_trusted_bucket_snapshot(s3_bucket_name="my-bucket"),
+ )
+
+ self._assert_capped_listing(route, files)
+
+ def test_file_list_end_to_end_surfaces_the_s3_error_body(self, monkeypatch):
+ import httpx
+ import respx
+
+ import litellm
+ from litellm.llms.bedrock.common_utils import BedrockError
+
+ monkeypatch.setenv("AWS_S3_BUCKET_NAME", "my-bucket")
+
+ with respx.mock:
+ respx.get(self.BUCKET_URL, params__contains=self.BATCH_QUERY).mock(
+ return_value=httpx.Response(403, text="AccessDenied")
+ )
+ with pytest.raises(BedrockError) as denied:
+ litellm.file_list(custom_llm_provider="bedrock", purpose="batch", **_bedrock_s3_params())
+
+ assert denied.value.status_code == 403
+ assert "AccessDenied" in denied.value.message
diff --git a/tests/test_litellm/proxy/openai_files_endpoint/test_files_endpoint.py b/tests/test_litellm/proxy/openai_files_endpoint/test_files_endpoint.py
index 5f1e7e1fe0c..b696d9ebe5f 100644
--- a/tests/test_litellm/proxy/openai_files_endpoint/test_files_endpoint.py
+++ b/tests/test_litellm/proxy/openai_files_endpoint/test_files_endpoint.py
@@ -2441,6 +2441,63 @@ def test_list_files_resolves_wildcard_deployment_credentials(
proxy_logging_obj.post_call_failure_hook.assert_not_called()
+def test_list_files_by_model_returns_an_openai_page_for_a_provider_listing(
+ mocker: MockerFixture, monkeypatch, llm_router: Router
+):
+ import litellm.proxy.proxy_server as ps
+ from litellm.proxy._types import LitellmUserRoles
+ from litellm.types.llms.openai import FileListPage
+
+ proxy_logging_obj = setup_proxy_logging_object(monkeypatch, llm_router)
+ monkeypatch.setattr("litellm.proxy.proxy_server.master_key", None)
+ monkeypatch.setattr("litellm.proxy.proxy_server.prisma_client", None)
+ monkeypatch.setattr("litellm.proxy.proxy_server.llm_router", llm_router)
+ proxy_logging_obj.update_request_status = mocker.AsyncMock()
+ proxy_logging_obj.post_call_success_hook = mocker.AsyncMock(return_value=None)
+ proxy_logging_obj.post_call_failure_hook = mocker.AsyncMock()
+
+ listed_files = [
+ OpenAIFileObject(
+ id=f"file-{index}",
+ bytes=index,
+ created_at=index,
+ filename=f"{index}.jsonl",
+ object="file",
+ purpose="batch",
+ status="uploaded",
+ )
+ for index in (1, 2)
+ ]
+
+ async def _mock_afile_list(**kwargs):
+ return list(listed_files)
+
+ monkeypatch.setattr(litellm, "afile_list", _mock_afile_list)
+
+ app.dependency_overrides[ps.user_api_key_auth] = lambda: UserAPIKeyAuth(
+ api_key="test-key",
+ user_role=LitellmUserRoles.PROXY_ADMIN,
+ user_id="test-user",
+ )
+
+ try:
+ response = client.get(
+ "/v1/files?target_model_names=gpt-3.5-turbo",
+ headers={"Authorization": "Bearer test-key"},
+ )
+ finally:
+ app.dependency_overrides.pop(ps.user_api_key_auth, None)
+
+ assert response.status_code == 200, response.text
+ body = response.json()
+ assert body["object"] == "list"
+ assert [listed["id"] for listed in body["data"]] == ["file-1", "file-2"]
+ assert (body["first_id"], body["last_id"], body["has_more"]) == ("file-1", "file-2", False)
+ hook_response = proxy_logging_obj.post_call_success_hook.call_args.kwargs["response"]
+ assert isinstance(hook_response, FileListPage)
+ assert [listed.id for listed in hook_response.data] == ["file-1", "file-2"]
+
+
def test_list_files_model_routing_does_not_forward_custom_llm_provider_twice(
mocker: MockerFixture, monkeypatch, llm_router: Router
):
@@ -4668,6 +4725,249 @@ def test_create_file_path_traversal_filename_rejected_before_forwarding(monkeypa
assert forwarded_calls == []
+def test_list_files_target_model_names_passes_trusted_bedrock_credentials(
+ mocker: MockerFixture, monkeypatch
+):
+ """
+ GET /v1/files?target_model_names= must hand the deployment's
+ immutable credential snapshot to litellm.afile_list, since Bedrock resolves
+ the S3 bucket to list from that snapshot rather than from request params.
+ """
+ from types import MappingProxyType
+
+ import litellm.proxy.proxy_server as ps
+ from litellm.proxy._types import LitellmUserRoles
+
+ bedrock_router = Router(
+ model_list=[
+ {
+ "model_name": "bedrock-claude",
+ "litellm_params": {
+ "model": "bedrock/anthropic.claude-3-5-sonnet-20240620-v1:0",
+ "aws_access_key_id": "AKIAEXAMPLE",
+ "aws_secret_access_key": "secret",
+ "aws_region_name": "us-west-2",
+ "s3_bucket_name": "my-bucket",
+ },
+ },
+ ]
+ )
+
+ proxy_logging_obj = setup_proxy_logging_object(monkeypatch, bedrock_router)
+ monkeypatch.setattr("litellm.proxy.proxy_server.master_key", None)
+ monkeypatch.setattr("litellm.proxy.proxy_server.prisma_client", None)
+ monkeypatch.setattr("litellm.proxy.proxy_server.llm_router", bedrock_router)
+ proxy_logging_obj.update_request_status = mocker.AsyncMock()
+ proxy_logging_obj.post_call_success_hook = mocker.AsyncMock(return_value=[])
+ proxy_logging_obj.post_call_failure_hook = mocker.AsyncMock()
+
+ captured_kwargs: dict = {}
+
+ async def _mock_afile_list(**kwargs):
+ captured_kwargs.update(kwargs)
+ return []
+
+ monkeypatch.setattr(litellm, "afile_list", _mock_afile_list)
+
+ app.dependency_overrides[ps.user_api_key_auth] = lambda: UserAPIKeyAuth(
+ api_key="test-key",
+ user_role=LitellmUserRoles.PROXY_ADMIN,
+ user_id="test-user",
+ )
+
+ try:
+ response = client.get(
+ "/v1/files?target_model_names=bedrock-claude&purpose=batch",
+ headers={"Authorization": "Bearer test-key"},
+ )
+ finally:
+ app.dependency_overrides.pop(ps.user_api_key_auth, None)
+
+ assert response.status_code == 200, response.text
+ assert captured_kwargs["custom_llm_provider"] == "bedrock"
+ assert captured_kwargs["purpose"] == "batch"
+ trusted_credentials = captured_kwargs["_litellm_internal_model_credentials"]
+ assert isinstance(trusted_credentials, MappingProxyType)
+ assert trusted_credentials["s3_bucket_name"] == "my-bucket"
+ proxy_logging_obj.post_call_failure_hook.assert_not_called()
+
+
+def test_delete_file_answers_400_for_an_id_outside_the_configured_bucket(mocker: MockerFixture, monkeypatch):
+ from urllib.parse import quote
+
+ import litellm.proxy.proxy_server as ps
+ from litellm.proxy._types import LitellmUserRoles
+
+ bedrock_router = Router(
+ model_list=[
+ {
+ "model_name": "bedrock-claude",
+ "litellm_params": {
+ "model": "bedrock/anthropic.claude-3-5-sonnet-20240620-v1:0",
+ "aws_access_key_id": "AKIAEXAMPLE",
+ "aws_secret_access_key": "secret",
+ "aws_region_name": "us-west-2",
+ "s3_bucket_name": "my-bucket",
+ },
+ },
+ ]
+ )
+
+ proxy_logging_obj = setup_proxy_logging_object(monkeypatch, bedrock_router)
+ monkeypatch.setattr("litellm.proxy.proxy_server.master_key", None)
+ monkeypatch.setattr("litellm.proxy.proxy_server.prisma_client", None)
+ monkeypatch.setattr("litellm.proxy.proxy_server.llm_router", bedrock_router)
+ proxy_logging_obj.update_request_status = mocker.AsyncMock()
+ proxy_logging_obj.post_call_success_hook = mocker.AsyncMock(return_value=[])
+ proxy_logging_obj.post_call_failure_hook = mocker.AsyncMock()
+
+ app.dependency_overrides[ps.user_api_key_auth] = lambda: UserAPIKeyAuth(
+ api_key="test-key",
+ user_role=LitellmUserRoles.PROXY_ADMIN,
+ user_id="test-user",
+ )
+ foreign_file_id: Final = quote("s3://other-bucket/litellm-bedrock-files/job-123/input.jsonl", safe="")
+
+ try:
+ with respx.mock:
+ response = client.delete(
+ f"/v1/files/{foreign_file_id}?model=bedrock-claude",
+ headers={"Authorization": "Bearer test-key"},
+ )
+ finally:
+ app.dependency_overrides.pop(ps.user_api_key_auth, None)
+
+ assert response.status_code == 400, response.text
+ assert "configured storage bucket" in response.json()["error"]["message"]
+
+
+def _cloud_files_router() -> Router:
+ return Router(
+ model_list=[
+ {
+ "model_name": "bedrock-claude",
+ "litellm_params": {
+ "model": "bedrock/anthropic.claude-3-5-sonnet-20240620-v1:0",
+ "aws_access_key_id": "AKIAEXAMPLE",
+ "aws_secret_access_key": "secret",
+ "aws_region_name": "us-west-2",
+ "s3_bucket_name": "my-bucket",
+ },
+ },
+ {
+ "model_name": "vertex-gemini",
+ "litellm_params": {
+ "model": "vertex_ai/gemini-3.8-flash",
+ "vertex_project": "my-project",
+ "vertex_location": "us-central1",
+ "gcs_bucket_name": "my-gcs-bucket",
+ },
+ },
+ ]
+ )
+
+
+RAW_S3_FILE_ID: Final = "s3://my-bucket/litellm-batch-outputs/job-123/abc/input.jsonl.out"
+RAW_GCS_FILE_ID: Final = "gs://my-gcs-bucket/litellm-vertex-files/publishers/google/models/gemini-3.8-flash/abc123"
+
+
+@pytest.mark.parametrize(
+ ("route_prefix", "raw_file_id", "model_name"),
+ (
+ ("/bedrock/v1/files", RAW_S3_FILE_ID, "bedrock-claude"),
+ ("/v1/files", RAW_S3_FILE_ID, "bedrock-claude"),
+ ("/files", RAW_S3_FILE_ID, "bedrock-claude"),
+ ("/vertex_ai/v1/files", RAW_GCS_FILE_ID, "vertex-gemini"),
+ ),
+)
+def test_delete_file_answers_403_for_a_raw_cloud_id_from_a_non_admin_key(
+ mocker: MockerFixture, monkeypatch, route_prefix: str, raw_file_id: str, model_name: str
+):
+ from urllib.parse import quote
+
+ import litellm.proxy.proxy_server as ps
+ from litellm.proxy._types import LitellmUserRoles
+
+ bedrock_router = _cloud_files_router()
+ proxy_logging_obj = setup_proxy_logging_object(monkeypatch, bedrock_router)
+ monkeypatch.setattr("litellm.proxy.proxy_server.master_key", None)
+ monkeypatch.setattr("litellm.proxy.proxy_server.prisma_client", None)
+ monkeypatch.setattr("litellm.proxy.proxy_server.llm_router", bedrock_router)
+ proxy_logging_obj.update_request_status = mocker.AsyncMock()
+ proxy_logging_obj.post_call_failure_hook = mocker.AsyncMock()
+ afile_delete = mocker.AsyncMock()
+ monkeypatch.setattr(litellm, "afile_delete", afile_delete)
+
+ app.dependency_overrides[ps.user_api_key_auth] = lambda: UserAPIKeyAuth(
+ api_key="test-key",
+ user_role=LitellmUserRoles.INTERNAL_USER,
+ user_id="test-user",
+ models=["bedrock-claude", "vertex-gemini"],
+ )
+
+ try:
+ response = client.delete(
+ f"{route_prefix}/{quote(raw_file_id, safe='')}?model={model_name}",
+ headers={"Authorization": "Bearer test-key"},
+ )
+ finally:
+ app.dependency_overrides.pop(ps.user_api_key_auth, None)
+
+ assert response.status_code == 403, response.text
+ assert "proxy admin" in response.json()["error"]["message"]
+ afile_delete.assert_not_called()
+
+
+def test_delete_file_forwards_a_raw_cloud_id_from_a_proxy_admin_key(mocker: MockerFixture, monkeypatch):
+ from urllib.parse import quote
+
+ import litellm.proxy.proxy_server as ps
+ from litellm.proxy._types import LitellmUserRoles
+
+ bedrock_router = _cloud_files_router()
+ proxy_logging_obj = setup_proxy_logging_object(monkeypatch, bedrock_router)
+ monkeypatch.setattr("litellm.proxy.proxy_server.master_key", None)
+ monkeypatch.setattr("litellm.proxy.proxy_server.prisma_client", None)
+ monkeypatch.setattr("litellm.proxy.proxy_server.llm_router", bedrock_router)
+ proxy_logging_obj.update_request_status = mocker.AsyncMock()
+ proxy_logging_obj.post_call_failure_hook = mocker.AsyncMock()
+
+ captured_kwargs: dict = {}
+
+ async def _mock_afile_delete(**kwargs):
+ captured_kwargs.update(kwargs)
+ return OpenAIFileObject(
+ id=RAW_S3_FILE_ID,
+ object="file",
+ bytes=2,
+ created_at=1234567890,
+ filename="input.jsonl.out",
+ purpose="batch_output",
+ status="processed",
+ )
+
+ monkeypatch.setattr(litellm, "afile_delete", _mock_afile_delete)
+
+ app.dependency_overrides[ps.user_api_key_auth] = lambda: UserAPIKeyAuth(
+ api_key="test-key",
+ user_role=LitellmUserRoles.PROXY_ADMIN,
+ user_id="test-user",
+ )
+
+ try:
+ response = client.delete(
+ f"/bedrock/v1/files/{quote(RAW_S3_FILE_ID, safe='')}?model=bedrock-claude",
+ headers={"Authorization": "Bearer test-key"},
+ )
+ finally:
+ app.dependency_overrides.pop(ps.user_api_key_auth, None)
+
+ assert response.status_code == 200, response.text
+ assert captured_kwargs.get("file_id") == RAW_S3_FILE_ID
+ assert captured_kwargs.get("custom_llm_provider") == "bedrock"
+ proxy_logging_obj.post_call_failure_hook.assert_not_called()
+
+
def _setup_managed_file_route_answering_404(
mocker: MockerFixture, monkeypatch: pytest.MonkeyPatch, llm_router: Router
) -> None: