scriptoriumadmin/litellm
1
0
Fork 0
mirror of https://github.com/BerriAI/litellm.git synced 2026-11-11 08:38:24 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 280

feat(proxy)!: only let clients pick S3 buckets the admin allowed (#47496)

Browse source 
* feat(proxy): let admins allow specific S3 buckets and endpoints in request bodies, block bucket settings otherwise

Client requests can no longer pick s3_bucket_name, s3_output_bucket_name or allow_legacy_cloud_file_ids unless the
value is on general_settings.allowed_client_side_s3_buckets (buckets) or the admin opted in the usual way.
s3_endpoint_url, already blocked, can now be allowed per value through general_settings.allowed_client_side_s3_endpoint_urls

* refactor(proxy): drive the client-side S3 allowlist from one typed field-to-setting map

* fix(proxy): treat a null client-side S3 allowlist in config as empty instead of failing the request
This commit is contained in:
ryan-crabbe-berri 2026-11-05 18:50:15 -08:00 • committed by GitHub
parent 5085022580
commit 48ff826cbc
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
31 changed files with 506 additions and 649 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

13
enterprise/litellm_enterprise/proxy/common_utils/check_batch_cost.py
View file

@ -8,7 +8,6 @@ import re
from collections.abc import Callable, Mapping, Sequence
from dataclasses import replace as dataclasses_replace
from datetime import datetime, timedelta
from types import MappingProxyType
from typing import TYPE_CHECKING, Final, List, Literal, Optional, Protocol, Tuple, cast
from pydantic import TypeAdapter, ValidationError
@ -766,7 +765,6 @@ class CheckBatchCost:
credentials: Final = self.llm_router.registry().credentials(model_id, form="with_provider") or {}
_file_content = await afile_content(
file_id=cast(str, raw_output_file_id),
_litellm_internal_model_credentials=MappingProxyType(dict(credentials)),
**cast(FileContentCallOptions, credentials),
)
@ -867,10 +865,7 @@ class CheckBatchCost:
error_file_failed_requests: Final = await count_error_file_failed_requests(
response,
custom_llm_provider=batch_file_provider,
litellm_params={
**credentials,
"_litellm_internal_model_credentials": MappingProxyType(dict(credentials)),
},
litellm_params=credentials,
)
batch_result: Final = (
output_file_result
@ -1029,11 +1024,7 @@ class CheckBatchCost:
custom_llm_provider=deployment_info.litellm_params.custom_llm_provider,
)
cost_provider: Final = cast(BatchFileProvider, llm_provider)
credentials: Final = self.llm_router.registry().credentials(model_id, form="with_provider") or {}
file_access: Final = {
**credentials,
"_litellm_internal_model_credentials": MappingProxyType(dict(credentials)),
}
file_access: Final = self.llm_router.registry().credentials(model_id, form="with_provider") or {}
output_content: Final = (
await fetch_batch_managed_file_content(
response.output_file_id, custom_llm_provider=cost_provider, litellm_params=file_access

30
enterprise/litellm_enterprise/proxy/hooks/managed_files.py
View file

@ -127,7 +127,6 @@ class _ManagedFileRetrieve(Protocol):
self,
*,
file_id: str,
_litellm_internal_model_credentials: Mapping[str, object] | None = None,
**kwargs: Unpack[FileRetrieveCallOptions],
) -> OpenAIFileObject: ...
@ -400,7 +399,7 @@ class _RouterFileCallKwargs(TypedDict, total=False):
_MODEL_FILE_ID_MAPPING: Final = TypeAdapter(dict[str, str])
_FILE_UPLOAD_ONLY_FIELDS: Final = frozenset({"file", "purpose", "expires_after", "extra_body"})
_FILE_DELETE_RESERVED_FIELDS: Final = frozenset({"model", "file_id", "_litellm_internal_model_credentials"})
_FILE_DELETE_RESERVED_FIELDS: Final = frozenset({"model", "file_id"})
@dataclass(frozen=True, slots=True)
@ -459,15 +458,7 @@ async def _delete_provider_file(
provider_file_id: str,
request_data: Mapping[str, object],
) -> None:
credentials: Final = llm_router.registry().credentials(model_id, form="with_provider")
delete_data: Final = {
**{k: v for k, v in request_data.items() if k not in _FILE_DELETE_RESERVED_FIELDS},
**(
{"_litellm_internal_model_credentials": MappingProxyType(dict(credentials))}
if credentials is not None
else {}
),
}
delete_data: Final = {k: v for k, v in request_data.items() if k not in _FILE_DELETE_RESERVED_FIELDS}
await llm_router.afile_delete(
model=model_id,
file_id=provider_file_id,
@ -1160,7 +1151,6 @@ class _PROXY_LiteLLMManagedFiles(CustomLogger, BaseFileEndpoints):
*,
provider_file_id: str,
call_options: Mapping[str, object],
internal_model_credentials: Mapping[str, object] | None = None,
) -> OpenAIFileObject:
"""Retrieve provider file details with transient retries and SDK retries disabled."""
retrieve_file: Final = cast(_ManagedFileRetrieve, litellm.afile_retrieve)
@ -1172,14 +1162,8 @@ class _PROXY_LiteLLMManagedFiles(CustomLogger, BaseFileEndpoints):
if attempt > 0:
await self._sleep(_PROVIDER_FILE_RETRIEVE_RETRY_DELAYS_SECONDS[attempt - 1])
try:
if internal_model_credentials is None:
return await retrieve_file(
file_id=provider_file_id,
**retrieve_options,
)
return await retrieve_file(
file_id=provider_file_id,
_litellm_internal_model_credentials=MappingProxyType(dict(internal_model_credentials)),
**retrieve_options,
)
except Exception as error:
@ -1221,7 +1205,6 @@ class _PROXY_LiteLLMManagedFiles(CustomLogger, BaseFileEndpoints):
self._afile_retrieve_with_retries(
provider_file_id=provider_file_id,
call_options=router_credentials,
internal_model_credentials=router_credentials,
),
timeout=BATCH_OUTPUT_FILE_LOOKUP_TIMEOUT_SECONDS,
)
@ -2866,15 +2849,6 @@ class _PROXY_LiteLLMManagedFiles(CustomLogger, BaseFileEndpoints):
exception_dict = {}
for model_id, provider_file_id in specific_model_file_id_mapping.items():
try:
# Cloud-storage providers (e.g. Bedrock S3) validate file ids
# against the deployment's configured bucket, which they only
# trust from this immutable server-side snapshot, never from
# request params.
credentials = llm_router.registry().credentials(model_id, form="with_provider")
if credentials is not None:
data["_litellm_internal_model_credentials"] = cast(Dict, MappingProxyType(dict(credentials)))
else:
data.pop("_litellm_internal_model_credentials", None)
return cast(
HttpxBinaryResponseContent,
await llm_router.afile_content(

4
litellm/batches/batch_utils.py
View file

@ -641,9 +641,11 @@ def extract_file_access_credentials(
"bucket_name",
"s3_endpoint_url",
"s3_region_name",
"s3_bucket_name",
"s3_output_bucket_name",
"allow_legacy_cloud_file_ids",
"timeout",
"max_retries",
"_litellm_internal_model_credentials",
*AWS_CREDENTIAL_KWARGS_KEYS,
# A federated deployment holds no api_key, so without these the fetch that reads a
# finished batch's output has nothing to authenticate with and its cost is never billed.

2
litellm/batches/main.py
View file

@ -22,7 +22,6 @@ from openai.types.batch import BatchRequestCounts
import litellm
from litellm._logging import verbose_logger
from litellm.litellm_core_utils.get_litellm_params import add_trusted_model_credentials_to_litellm_params
from litellm.litellm_core_utils.litellm_logging import Logging as LiteLLMLoggingObj
from litellm.llms.anthropic.batches.handler import AnthropicBatchesHandler
from litellm.llms.azure.batches.handler import AzureBatchesAPI
@ -585,7 +584,6 @@ def retrieve_batch(
custom_llm_provider=custom_llm_provider,
**kwargs,
)
add_trusted_model_credentials_to_litellm_params(litellm_params, kwargs)
if litellm_logging_obj is not None:
litellm_logging_obj.update_from_kwargs(
kwargs=kwargs,

17
litellm/files/main.py
View file

@ -38,7 +38,6 @@ import litellm
from litellm import get_secret_str
from litellm.files.streaming import FileContentStreamingResponse
from litellm.files.types import FileContentProvider, FileContentStreamingResult, FileRetrieveProvider
from litellm.litellm_core_utils.get_litellm_params import add_trusted_model_credentials_to_litellm_params
from litellm.litellm_core_utils.get_llm_provider_logic import get_llm_provider
from litellm.litellm_core_utils.litellm_logging import Logging as LiteLLMLoggingObj
from litellm.llms.azure.common_utils import get_azure_credentials
@ -422,10 +421,6 @@ def file_retrieve(
)
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
@ -544,10 +539,6 @@ def file_delete(
pass
optional_params: Final = GenericLiteLLMParams(**kwargs)
litellm_params_dict: Final = get_litellm_params(**kwargs)
add_trusted_model_credentials_to_litellm_params(
litellm_params_dict=litellm_params_dict,
kwargs=kwargs,
)
### TIMEOUT LOGIC ###
timeout = optional_params.timeout or kwargs.get("request_timeout", 600) or 600
# set timeout for 10 minutes by default
@ -737,10 +728,6 @@ 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
@ -890,10 +877,6 @@ def file_content(
try:
optional_params: Final = GenericLiteLLMParams(**kwargs)
litellm_params_dict: Final = get_litellm_params(**kwargs)
add_trusted_model_credentials_to_litellm_params(
litellm_params_dict=litellm_params_dict,
kwargs=kwargs,
)
### TIMEOUT LOGIC ###
timeout = optional_params.timeout or kwargs.get("request_timeout", 600) or 600
client: Final = kwargs.get("client")

11
litellm/litellm_core_utils/cloud_storage_security.py
View file

@ -2,12 +2,10 @@ import posixpath
import re
from collections.abc import Mapping, Sequence
from itertools import accumulate, repeat
from types import MappingProxyType
from typing import Final
from urllib.parse import quote, unquote
from litellm._uuid import uuid
from litellm.litellm_core_utils.core_helpers import as_str_mapping
VERTEX_AI_MANAGED_GCS_PREFIX: Final = "litellm-vertex-files/"
BEDROCK_MANAGED_S3_BATCH_PREFIX: Final = "litellm-bedrock-files-"
@ -19,7 +17,6 @@ BEDROCK_MANAGED_S3_PREFIXES: Final = (
BEDROCK_MANAGED_S3_OUTPUT_PREFIX,
)
MANAGED_CLOUD_STORAGE_SCHEMES: Final = ("s3://", "gs://")
_MAPPING_PROXY_TYPE: Final[type] = type(MappingProxyType({}))
MAX_FILE_ID_DECODE_PASSES: Final = 8
@ -138,13 +135,7 @@ def encode_s3_object_key_for_url(object_key: str) -> str:
def should_allow_legacy_cloud_file_ids(
litellm_params: Mapping[str, object] | None = None,
) -> bool:
value = None
if isinstance(litellm_params, Mapping):
trusted_model_credentials: Final = litellm_params.get("_litellm_internal_model_credentials")
if isinstance(trusted_model_credentials, _MAPPING_PROXY_TYPE):
trusted_mapping: Final = as_str_mapping(trusted_model_credentials)
value = trusted_mapping.get("allow_legacy_cloud_file_ids") if trusted_mapping is not None else None
value: Final = litellm_params.get("allow_legacy_cloud_file_ids") if litellm_params is not None else None
if isinstance(value, bool):
return value
if isinstance(value, str):

22
litellm/litellm_core_utils/get_litellm_params.py
View file

@ -1,7 +1,6 @@
import reprlib
from collections.abc import Mapping, MutableMapping
from collections.abc import Mapping
from dataclasses import dataclass, fields
from types import MappingProxyType
from typing import Final, cast
from pydantic import TypeAdapter, ValidationError
@ -61,6 +60,9 @@ OPTIONAL_KWARGS_KEYS: Final = (
"client_side_timeout",
"gcs_bucket_name",
"bucket_name",
"s3_bucket_name",
"s3_output_bucket_name",
"allow_legacy_cloud_file_ids",
"s3_endpoint_url",
"s3_region_name",
"s3_access_key_id",
@ -292,19 +294,3 @@ def get_litellm_params(
litellm_params[key] = kwargs[key]
return litellm_params
def add_trusted_model_credentials_to_litellm_params(
litellm_params_dict: MutableMapping[str, object], kwargs: Mapping[str, object]
) -> None:
"""
Carry the immutable server-side credential snapshot into litellm_params.
get_litellm_params has a fixed signature, so callers that need the snapshot to
survive into the logging object and the downstream file read have to re-add it. Only
a MappingProxyType is accepted, since providers resolve trusted configuration such
as a Bedrock file bucket from it and must not read a request-supplied mapping.
"""
trusted_model_credentials: Final = kwargs.get("_litellm_internal_model_credentials")
if isinstance(trusted_model_credentials, MappingProxyType):
litellm_params_dict["_litellm_internal_model_credentials"] = trusted_model_credentials

59
litellm/llms/bedrock/files/transformation.py
View file

@ -229,15 +229,6 @@ class _S3RequestTarget:
request_params: _BedrockS3RequestParams
class _TrustedS3ModelCredentials(LiteLLMBaseModel):
"""The S3 buckets the server trusts file ids against, from the deployment snapshot."""
model_config = ConfigDict(extra="ignore")
s3_bucket_name: str | None = None
s3_output_bucket_name: str | None = None
def extract_s3_uri_from_file_id(file_id: str) -> str:
"""
Resolve a Bedrock file id to its S3 URI.
@ -261,27 +252,31 @@ def extract_s3_uri_from_file_id(file_id: str) -> str:
raise ValueError("file_id must be a managed LiteLLM S3 file id")
_S3_BUCKET_REQUIRED_ERROR: Final = "S3 bucket_name is required. Set 's3_bucket_name' in proxy config or AWS_S3_BUCKET_NAME for Bedrock file content retrieval."
_S3_BUCKET_REQUIRED_ERROR: Final = (
"S3 bucket_name is required. Pass 's3_bucket_name', set it in proxy config, "
"or set AWS_S3_BUCKET_NAME for Bedrock file content retrieval."
)
def _trusted_s3_model_credentials(litellm_params: Mapping[str, object]) -> _TrustedS3ModelCredentials:
trusted_model_credentials: Final = litellm_params.get("_litellm_internal_model_credentials")
if not isinstance(trusted_model_credentials, MappingProxyType):
return _TrustedS3ModelCredentials()
snapshot: Final[dict[str, object]] = {}
snapshot.update(trusted_model_credentials) # any-ok: untyped snapshot
return _TrustedS3ModelCredentials.model_validate(snapshot)
def _s3_param(litellm_params: Mapping[str, object], key: str, env_var: str) -> str | None:
value: Final = litellm_params.get(key)
return value if isinstance(value, str) and value else os.getenv(env_var)
def _input_bucket_name(litellm_params: Mapping[str, object]) -> str | None:
return _s3_param(litellm_params, "s3_bucket_name", "AWS_S3_BUCKET_NAME")
def _output_bucket_name(litellm_params: Mapping[str, object]) -> str | None:
return _s3_param(litellm_params, "s3_output_bucket_name", "AWS_S3_OUTPUT_BUCKET_NAME")
def get_configured_s3_bucket_name(litellm_params: Mapping[str, object]) -> str:
"""
Resolve the server-configured S3 bucket for Bedrock file operations.
Only trusts the immutable server-side credential snapshot or the
environment; never a request-supplied param, since the bucket is what
`validate_managed_cloud_file_id` checks file ids against.
Resolve the S3 bucket for Bedrock file operations: the call's ``s3_bucket_name``,
else ``AWS_S3_BUCKET_NAME``. File ids are validated against this bucket.
"""
bucket_name: Final = _trusted_s3_model_credentials(litellm_params).s3_bucket_name or os.getenv("AWS_S3_BUCKET_NAME")
bucket_name: Final = _input_bucket_name(litellm_params)
if not bucket_name:
raise ValueError(_S3_BUCKET_REQUIRED_ERROR)
return bucket_name
@ -289,16 +284,13 @@ def get_configured_s3_bucket_name(litellm_params: Mapping[str, object]) -> str:
def get_configured_s3_bucket_names(litellm_params: Mapping[str, object]) -> tuple[str, ...]:
"""
Resolve the server-configured S3 buckets a Bedrock file id may live in.
Resolve the S3 buckets a Bedrock file id may live in.
Bedrock batch outputs land in ``s3_output_bucket_name`` when it differs from
the input bucket, so retrieval validates against both. Same trust rules as
``get_configured_s3_bucket_name``: only the immutable credential snapshot or
the environment, never a request param.
the input bucket, so retrieval validates against both.
"""
trusted: Final = _trusted_s3_model_credentials(litellm_params)
input_bucket: Final = trusted.s3_bucket_name or os.getenv("AWS_S3_BUCKET_NAME")
output_bucket: Final = trusted.s3_output_bucket_name or os.getenv("AWS_S3_OUTPUT_BUCKET_NAME")
input_bucket: Final = _input_bucket_name(litellm_params)
output_bucket: Final = _output_bucket_name(litellm_params)
buckets: Final = tuple(dict.fromkeys(bucket for bucket in (input_bucket, output_bucket) if bucket))
if not buckets:
raise ValueError(_S3_BUCKET_REQUIRED_ERROR)
@ -514,12 +506,7 @@ def _output_location_still_unlisted(litellm_params: Mapping[str, object]) -> boo
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)
)
return _output_bucket_name(litellm_params) or get_configured_s3_bucket_name(litellm_params)
def _listed_object_created_at(entry: ET.Element) -> int:

8
litellm/proxy/_types.py
View file

@ -2813,6 +2813,14 @@ class ConfigGeneralSettings(LiteLLMPydanticObjectBase):
description="How long a blocked source address, or source address and username, stays blocked. Every attempt from a blocked key, right or wrong, is refused with 429 before the password is checked; the block is not extended by refused attempts. Set under `general_settings` in config.yaml. Defaults to 300",
)
allowed_routes: list | None = Field(None, description="Proxy API Endpoints you want users to be able to access")
allowed_client_side_s3_buckets: tuple[str, ...] | None = Field(
None,
description="S3 bucket names a client may send as `s3_bucket_name` or `s3_output_bucket_name` in a request body, e.g. on /v1/batches. Any other bucket is rejected unless client-side credentials are allowed",
)
allowed_client_side_s3_endpoint_urls: tuple[str, ...] | None = Field(
None,
description="Exact `s3_endpoint_url` values a client may send in a request body. Any other value is rejected unless client-side credentials are allowed",
)
reject_clientside_metadata_tags: bool | None = Field(
None,
description="When set to True, rejects requests that contain client-side 'metadata.tags' to prevent users from influencing budgets by sending different tags. Tags can only be inherited from the API key metadata.",

62
litellm/proxy/auth/auth_utils.py
View file

@ -7,10 +7,10 @@ from functools import lru_cache
from itertools import accumulate, chain, repeat
from logging import Logger
from types import MappingProxyType
from typing import Any, Final, Protocol
from typing import Any, Final, Literal, Protocol, TypeAlias
from fastapi import HTTPException, Request, status
from pydantic import PositiveInt, TypeAdapter, ValidationError
from pydantic import BaseModel, ConfigDict, PositiveInt, TypeAdapter, ValidationError
import litellm
from litellm import Router, constants, provider_list
@ -442,6 +442,9 @@ _BANNED_REQUEST_BODY_PARAMS: Final[tuple[str, ...]] = (
# S3; ``sagemaker_base_url`` redirects all SageMaker traffic;
# ``deployment_url`` redirects SAP deployments.
"s3_endpoint_url",
"s3_bucket_name",
"s3_output_bucket_name",
"allow_legacy_cloud_file_ids",
"sagemaker_base_url",
"deployment_url",
# NVIDIA Riva fields consumed by the audio-transcription handler
@ -479,6 +482,48 @@ _BANNED_REQUEST_BODY_PARAMS: Final[tuple[str, ...]] = (
)
_ClientSideAllowlistSetting: TypeAlias = Literal[
"allowed_client_side_s3_buckets", "allowed_client_side_s3_endpoint_urls"
]
_PROXY_ALLOWLIST_SETTING_BY_PARAM: Final[Mapping[str, _ClientSideAllowlistSetting]] = MappingProxyType(
{
"s3_bucket_name": "allowed_client_side_s3_buckets",
"s3_output_bucket_name": "allowed_client_side_s3_buckets",
"s3_endpoint_url": "allowed_client_side_s3_endpoint_urls",
}
)
class _ClientSideStorageAllowlist(BaseModel):
model_config = ConfigDict(frozen=True)
allowed_client_side_s3_buckets: tuple[str, ...] | None = None
allowed_client_side_s3_endpoint_urls: tuple[str, ...] | None = None
def admits(self, setting: _ClientSideAllowlistSetting, value: object) -> bool:
match setting:
case "allowed_client_side_s3_buckets":
return value in (self.allowed_client_side_s3_buckets or ())
case "allowed_client_side_s3_endpoint_urls":
return value in (self.allowed_client_side_s3_endpoint_urls or ())
def _banned_param_error(param: str) -> ValueError:
setting: Final = _PROXY_ALLOWLIST_SETTING_BY_PARAM.get(param)
allowlist_hint: Final = (
"" if setting is None else f" To allow specific values only, list them in `general_settings.{setting}`."
)
return ValueError(
f"Rejected Request: {param} is not allowed in request body. "
"Clientside passthrough requires explicit admin opt-in via "
"either `general_settings.allow_client_side_credentials = true` "
"(proxy-wide) or `configurable_clientside_auth_params` on the "
"deployment in your proxy config.yaml. "
f"Relevant Issue: https://huntr.com/bounties/4001e1a2-7b7a-4776-a3ae-e6692ec3d997{allowlist_hint}",
)
def _check_banned_params(
body: Mapping[str, object],
general_settings: dict,
@ -508,6 +553,10 @@ def _check_banned_params(
# Proxy-wide opt-in: every banned param is permitted, exit
# entirely so the rest of the loop doesn't waste work.
return
if (
setting := _PROXY_ALLOWLIST_SETTING_BY_PARAM.get(param)
) is not None and _ClientSideStorageAllowlist.model_validate(general_settings).admits(setting, body[param]):
continue
if (
_allow_model_level_clientside_configurable_parameters(
model=model,
@ -523,14 +572,7 @@ def _check_banned_params(
# ``api_base`` with an unallowed ``langfuse_host`` is still
# rejected for the second field.
continue
raise ValueError(
f"Rejected Request: {param} is not allowed in request body. "
"Clientside passthrough requires explicit admin opt-in via "
"either `general_settings.allow_client_side_credentials = true` "
"(proxy-wide) or `configurable_clientside_auth_params` on the "
"deployment in your proxy config.yaml. "
"Relevant Issue: https://huntr.com/bounties/4001e1a2-7b7a-4776-a3ae-e6692ec3d997",
)
raise _banned_param_error(param)
_FALLBACK_FIELDS: Final[tuple[str, ...]] = (

8
litellm/proxy/batches_endpoints/endpoints.py
View file

@ -48,7 +48,6 @@ from litellm.proxy.openai_files_endpoints.common_utils import ( # noqa: F401 #
BATCH_CREATE_HIDDEN_PARAM,
_is_base64_encoded_unified_file_id, # pyright: ignore[reportPrivateUsage,reportUnusedImport] # backwards-compatible package export
add_deployment_model_info,
add_internal_model_credentials,
apply_team_provider_credentials,
authorize_model_for_key,
batch_cost_poller_is_active,
@ -750,13 +749,6 @@ async def retrieve_batch(
detail={"error": "LLM Router not initialized. Ensure models added to proxy."},
)
if unified_batch_id:
add_internal_model_credentials(
data=data,
llm_router=llm_router,
model_id=unified_model_id,
)
response = await llm_router.aretrieve_batch(**data)
set_hidden_param(response, "unified_batch_id", unified_batch_id)
if unified_batch_id:

30
litellm/proxy/openai_files_endpoints/common_utils.py
View file

@ -741,31 +741,6 @@ def apply_team_provider_credentials(
prepare_data_with_credentials(data=data, credentials={**credentials, **explicit_request_credentials(data)})
def add_internal_model_credentials(
data: dict,
llm_router: "Router",
model_id: str | None,
) -> None:
"""
Attach the deployment's immutable server-side credential snapshot to a router-routed
batch call (in-place).
Cost accounting for a completed batch reads the batch's output file, and the Bedrock
file config resolves its bucket only from this snapshot, never from a request param,
because the bucket is what managed file ids are validated against. Without it that
read fails and the batch's cost is never recorded.
"""
if model_id is None:
return
try:
credentials: Final = llm_router.registry().credentials(model_id, form="with_provider")
except Exception: # noqa: BLE001 # the snapshot only enables cost accounting; a batch whose deployment no longer resolves must still be retrievable
return
if credentials is None:
return
data["_litellm_internal_model_credentials"] = MappingProxyType(dict(credentials))
def add_deployment_model_info(
data: dict,
llm_router: Optional["Router"],
@ -791,7 +766,6 @@ def prepare_data_with_credentials(
data: dict,
credentials: dict,
file_id: str | None = None,
include_internal_credentials: bool = False,
) -> None:
"""
Update data dictionary with model credentials (in-place).
@ -800,12 +774,8 @@ def prepare_data_with_credentials(
data: Data dictionary to update
credentials: Credentials from router
file_id: Optional original file_id to set (for decoded file IDs)
include_internal_credentials: Preserve an immutable server-side snapshot
for code paths that must distinguish proxy config from request params.
"""
data.update(credentials)
if include_internal_credentials:
data["_litellm_internal_model_credentials"] = MappingProxyType(dict(credentials))
data.pop("custom_llm_provider", None)
if file_id is not None:

1
litellm/proxy/openai_files_endpoints/file_content_streaming_handler.py
View file

@ -43,7 +43,6 @@ class FileContentStreamingHandler:
data=resolved_streaming_data,
credentials=credentials,
file_id=original_file_id,
include_internal_credentials=True,
)
resolved_streaming_data.pop("model", None)
resolved_streaming_provider: Final = cast(str, credentials["custom_llm_provider"])

9
litellm/proxy/openai_files_endpoints/files_endpoints.py
View file

@ -77,7 +77,6 @@ from litellm.proxy.openai_files_endpoints.batch_guardrails import (
)
from litellm.proxy.openai_files_endpoints.common_utils import ( # noqa: F401 # legacy module exports
_is_base64_encoded_unified_file_id, # pyright: ignore[reportPrivateUsage,reportUnusedImport] # backwards-compatible package export
add_internal_model_credentials,
apply_team_provider_credentials,
authorize_model_for_key,
deployment_drop_params,
@ -1091,7 +1090,6 @@ async def get_file_content(
model: Final = cast(str | None, data.get("model"))
if model:
add_internal_model_credentials(data=data, llm_router=llm_router, model_id=model)
response = await llm_router.afile_content(
**{
"model": model,
@ -1180,7 +1178,6 @@ async def get_file_content(
data=data,
credentials=credentials,
file_id=original_file_id, # Use decoded file ID if from encoded ID
include_internal_credentials=True,
)
response = await litellm.afile_content(
custom_llm_provider=credentials["custom_llm_provider"],
@ -1354,7 +1351,6 @@ async def get_file(
data=data,
credentials=credentials,
file_id=original_file_id,
include_internal_credentials=True,
)
response = await litellm.afile_retrieve(
@ -1568,7 +1564,6 @@ async def delete_file(
data=data,
credentials=credentials,
file_id=original_file_id,
include_internal_credentials=True,
)
response = await litellm.afile_delete(
@ -1973,7 +1968,7 @@ async def list_files(
)
elif should_route and credentials is not None:
# Use model-based routing with credentials from config
prepare_data_with_credentials(data=data, credentials=credentials, include_internal_credentials=True)
prepare_data_with_credentials(data=data, credentials=credentials)
response = await litellm.afile_list(
custom_llm_provider=credentials["custom_llm_provider"],
purpose=purpose,
@ -1997,7 +1992,7 @@ async def list_files(
user_api_key_dict=user_api_key_dict,
operation_context="file list",
)
prepare_data_with_credentials(data=data, credentials=credentials, include_internal_credentials=True)
prepare_data_with_credentials(data=data, credentials=credentials)
response = await litellm.afile_list(
custom_llm_provider=credentials["custom_llm_provider"],
purpose=purpose,

1
litellm/types/router.py
View file

@ -427,6 +427,7 @@ class CredentialLiteLLMParams(LiteLLMBaseModel):
## OBJECT STORAGE (files / batches) ##
gcs_bucket_name: str | None = None
bucket_name: str | None = None
allow_legacy_cloud_file_ids: bool | None = None
## AWS BEDROCK / SAGEMAKER ##
aws_access_key_id: str | None = None

29
tests/unit/batches/test_batch_utils.py
View file

@ -16,7 +16,6 @@ deterministic stand-ins so the arithmetic under test is the only variable.
import asyncio, json, time
import logging
from types import MappingProxyType
from typing import Final, Literal
import httpx
@ -1692,17 +1691,21 @@ async def test_calculate_batch_cost_and_usage_anthropic_end_to_end():
assert result.models == ["claude-sonnet-4-5"]
def test_extract_credentials_forwards_the_trusted_model_credential_snapshot():
"""Bedrock resolves a batch's output bucket only from the immutable server-side
snapshot, never from a request param, so cost accounting on the retrieve path cannot
read the output file unless this key is forwarded. Without it the accounting raises
"S3 bucket_name is required" for a bucket the deployment has configured, and the
batch's cost is never recorded."""
snapshot = MappingProxyType({"s3_bucket_name": "configured-bucket", "aws_region_name": "us-east-1"})
def test_extract_credentials_forwards_the_deployment_bucket_settings() -> None:
"""Cost accounting on the retrieve path reads the batch's output file from the
deployment's Bedrock buckets. Without these keys the read raises "S3 bucket_name is
required" for a bucket the deployment has configured, and the batch's cost is never
recorded."""
params: Final = {
"s3_bucket_name": "configured-bucket",
"s3_output_bucket_name": "output-bucket",
"allow_legacy_cloud_file_ids": True,
"model": "bedrock/us.anthropic.claude-haiku-4-5-20251001-v1:0",
}
credentials = bu._extract_file_access_credentials({"_litellm_internal_model_credentials": snapshot})
credentials: Final = bu._extract_file_access_credentials(params)
assert credentials["_litellm_internal_model_credentials"] is snapshot
assert credentials == {key: value for key, value in params.items() if key != "model"}
def test_extract_credentials_forwards_the_deployment_aws_credentials():
@ -1736,8 +1739,6 @@ async def test_output_file_content_bedrock_reads_with_deployment_aws_credentials
return type("R", (), {"content": b""})()
monkeypatch.setattr(files_main, "afile_content", fake_afile_content)
snapshot = MappingProxyType({"s3_bucket_name": "configured-bucket", "aws_region_name": "us-west-2"})
await bu._fetch_batch_output_file_content(
_batch("s3://configured-bucket/litellm-batch-outputs/job-1/out.jsonl.out"),
custom_llm_provider="bedrock",
@ -1746,7 +1747,7 @@ async def test_output_file_content_bedrock_reads_with_deployment_aws_credentials
"aws_secret_access_key": "secret-deployment",
"aws_session_token": "token-deployment",
"aws_region_name": "us-west-2",
"_litellm_internal_model_credentials": snapshot,
"s3_bucket_name": "configured-bucket",
"model": "bedrock/us.anthropic.claude-haiku-4-5-20251001-v1:0",
},
)
@ -1757,7 +1758,7 @@ async def test_output_file_content_bedrock_reads_with_deployment_aws_credentials
assert captured["aws_secret_access_key"] == "secret-deployment"
assert captured["aws_session_token"] == "token-deployment"
assert captured["aws_region_name"] == "us-west-2"
assert captured["_litellm_internal_model_credentials"] is snapshot
assert captured["s3_bucket_name"] == "configured-bucket"
assert "model" not in captured

32
tests/unit/batches/test_main.py
View file

@ -732,34 +732,30 @@ def test_resolve_timeout__httpx_timeout_returns_float_read():
assert resolved == 99.0
def test_retrieve__forwards_trusted_model_credentials_into_litellm_params(seams):
def test_retrieve__keeps_the_deployment_bucket_settings_in_litellm_params(seams: Seams) -> None:
"""The batch's cost is computed by reading its output file after the retrieve, and
Bedrock resolves that bucket only from this immutable snapshot. get_litellm_params has
a fixed signature that drops it, so without re-adding it here the snapshot never
reaches the logging object and cost accounting fails on a bucket that is configured."""
snapshot = MappingProxyType({"s3_bucket_name": "configured-bucket"})
that read finds the Bedrock bucket in these litellm_params. Dropping them fails cost
accounting on a bucket the deployment has configured."""
logging_obj = MagicMock()
bm.retrieve_batch(
batch_id="batch-1",
custom_llm_provider="openai",
litellm_logging_obj=logging_obj,
_litellm_internal_model_credentials=snapshot,
s3_bucket_name="configured-bucket",
s3_output_bucket_name="output-bucket",
allow_legacy_cloud_file_ids=True,
)
litellm_params = logging_obj.update_from_kwargs.call_args.kwargs["litellm_params"]
assert litellm_params["_litellm_internal_model_credentials"] is snapshot
def test_retrieve__omits_trusted_model_credentials_when_not_supplied(seams):
"""A retrieve with no snapshot must not invent an empty one, which would read as a
configured bucket of nothing."""
logging_obj = MagicMock()
bm.retrieve_batch(batch_id="batch-1", custom_llm_provider="openai", litellm_logging_obj=logging_obj)
litellm_params = logging_obj.update_from_kwargs.call_args.kwargs["litellm_params"]
assert "_litellm_internal_model_credentials" not in litellm_params
assert {
key: litellm_params[key]
for key in ("s3_bucket_name", "s3_output_bucket_name", "allow_legacy_cloud_file_ids")
} == {
"s3_bucket_name": "configured-bucket",
"s3_output_bucket_name": "output-bucket",
"allow_legacy_cloud_file_ids": True,
}
# =========================================================================== #

5
tests/unit/enterprise/proxy/hooks/test_managed_files.py
View file

@ -1980,7 +1980,6 @@ async def test_store_batch_output_file_retries_model_name_provider_fetch():
assert all(
call.kwargs["custom_llm_provider"] == "bedrock"
and call.kwargs["max_retries"] == 0
and "_litellm_internal_model_credentials" not in call.kwargs
for call in retrieve.call_args_list
)
assert sleep_calls == [0.5, 1.0]
@ -2022,7 +2021,6 @@ async def test_store_batch_output_file_marks_model_name_fallback_after_four_tran
assert all(
call.kwargs["custom_llm_provider"] == "bedrock"
and call.kwargs["max_retries"] == 0
and "_litellm_internal_model_credentials" not in call.kwargs
for call in retrieve.call_args_list
)
assert sleep_calls == [0.5, 1.0, 2.0]
@ -3656,7 +3654,6 @@ async def test_afile_retrieve_refreshes_marked_fallback_without_router_from_mode
retrieve.assert_awaited_once()
assert retrieve.await_args.kwargs["custom_llm_provider"] == "bedrock"
assert retrieve.await_args.kwargs["max_retries"] == 0
assert "_litellm_internal_model_credentials" not in retrieve.await_args.kwargs
updated_row: Final = managed_file_table.rows["unified-output"]
assert updated_row.file_object == response
assert updated_row.model_mappings == {"bedrock/model-x": "provider-output"}
@ -3700,7 +3697,6 @@ async def test_afile_retrieve_uses_model_name_when_router_does_not_know_deployme
)
assert retrieve.await_args.kwargs["custom_llm_provider"] == "bedrock"
assert retrieve.await_args.kwargs["max_retries"] == 0
assert "_litellm_internal_model_credentials" not in retrieve.await_args.kwargs
@pytest.mark.asyncio
@ -3828,7 +3824,6 @@ async def test_afile_retrieve_case3_uses_default_provider_for_unknown_router_dep
assert retrieve.await_args.kwargs["file_id"] == "provider-output"
assert retrieve.await_args.kwargs["max_retries"] == 0
assert "custom_llm_provider" not in retrieve.await_args.kwargs
assert "_litellm_internal_model_credentials" not in retrieve.await_args.kwargs
@pytest.mark.asyncio

129
tests/unit/enterprise/proxy/test_managed_files_hook.py
View file

@ -1165,45 +1165,56 @@ async def test_should_skip_non_file_unified_id_on_output_file_id():
@pytest.mark.asyncio
async def test_afile_content_passes_trusted_model_credentials_to_router():
"""
afile_content must hand the deployment's credential snapshot to the router
call as an immutable server-side mapping. Cloud-storage providers (Bedrock
S3) validate file ids against the bucket in that snapshot, so without it
unified-id content retrieval only works when AWS_S3_BUCKET_NAME is set.
"""
from types import MappingProxyType
async def test_afile_content_reads_from_the_deployment_bucket(monkeypatch: pytest.MonkeyPatch) -> None:
"""Unified-id content retrieval must validate and fetch the s3:// file against the
deployment's s3_bucket_name, with no AWS_S3_BUCKET_NAME env set."""
import httpx
import respx
import litellm
from litellm import Router
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()
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"},
}
],
num_retries=0,
)
managed_files = _make_managed_files_instance()
unified_file_id = "unified-file-id"
s3_uri = "s3://my-bucket/litellm-batch-outputs/job-123/input.jsonl.out"
managed_files.get_unified_file_id = AsyncMock(return_value=None)
managed_files.get_model_file_id_mapping = AsyncMock(
return_value={unified_file_id: {"model-123": s3_uri}}
)
managed_files.get_model_file_id_mapping = AsyncMock(return_value={unified_file_id: {"model-123": s3_uri}})
mock_router = MagicMock()
mock_router.registry.return_value.credentials = MagicMock(
return_value={
"custom_llm_provider": "bedrock",
"s3_bucket_name": "my-bucket",
"aws_region_name": "us-west-2",
}
)
mock_router.afile_content = AsyncMock(return_value=MagicMock())
with respx.mock:
route = respx.get(
"https://s3.us-west-2.amazonaws.com/my-bucket/litellm-batch-outputs/job-123/input.jsonl.out"
).mock(return_value=httpx.Response(200, content=b'{"recordId":"req-1"}'))
await managed_files.afile_content(
file_id=unified_file_id,
litellm_parent_otel_span=None,
llm_router=mock_router,
)
response = await managed_files.afile_content(
file_id=unified_file_id,
litellm_parent_otel_span=None,
llm_router=router,
)
call_kwargs = mock_router.afile_content.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"
assert route.calls[0].request.headers["Authorization"].startswith("AWS4-HMAC-SHA256")
assert response.content == b'{"recordId":"req-1"}'
def _managed_deletion_file_id(provider_file_id):
@ -1273,7 +1284,6 @@ async def test_afile_delete_bedrock_uses_deployment_bucket_and_signed_s3_delete(
file_id=unified_file_id,
litellm_parent_otel_span=None,
llm_router=router,
_litellm_internal_model_credentials={"s3_bucket_name": "request-bucket"},
)
assert len(route.calls) == 1
@ -1299,7 +1309,6 @@ async def test_afile_delete_returns_managed_id_for_stored_provider_output():
file_id=unified_file_id,
litellm_parent_otel_span=None,
llm_router=router,
_litellm_internal_model_credentials={"s3_bucket_name": "request-bucket"},
)
assert response.id == unified_file_id
@ -1928,45 +1937,6 @@ async def test_post_call_hook_leaves_litellm_executed_batches_untouched(llm_batc
assert response.id == original_id
@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_unified_file_id = AsyncMock(return_value=None)
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.registry.return_value.credentials = 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):
"""
@ -2291,7 +2261,7 @@ class _FakeFilesRouter:
self.delete_gate = delete_gate
self.provider = provider
self.upload_count = 0
self.deletes: tuple[tuple[str, str, Mapping[str, object], Mapping[str, str]], ...] = ()
self.deletes: tuple[tuple[str, str, Mapping[str, object]], ...] = ()
self.deletes_marked_internal: tuple[bool, ...] = ()
self.delete_parent_request_ids: tuple[str | None, ...] = ()
@ -2328,14 +2298,13 @@ class _FakeFilesRouter:
model: str,
file_id: str,
litellm_metadata: Mapping[str, object],
_litellm_internal_model_credentials: Mapping[str, str],
litellm_call_id: str | None = None,
) -> None:
if self.delete_gate is not None:
await self.delete_gate.wait()
if isinstance(litellm_metadata, dict):
litellm_metadata.update({"model_info": {"id": model}})
self.deletes = (*self.deletes, (model, file_id, litellm_metadata, _litellm_internal_model_credentials))
self.deletes = (*self.deletes, (model, file_id, litellm_metadata))
self.deletes_marked_internal = (*self.deletes_marked_internal, is_internal_call.get())
self.delete_parent_request_ids = (*self.delete_parent_request_ids, internal_parent_request_id.get())
if file_id in self.gone_file_ids:
@ -2399,15 +2368,7 @@ async def test_acreate_file_keeps_the_upload_the_managed_id_carries_when_two_tar
assert carried == ("wildcard-dep", "file-upload-1")
assert model_mappings == {"wildcard-dep": "file-upload-1"}
assert flat_ids == ["file-upload-1"]
assert router.deletes == (
(
"wildcard-dep",
"file-upload-2",
_REQUEST_METADATA,
{"custom_llm_provider": "openai", "api_key": "key-of-wildcard-dep"},
),
)
assert isinstance(router.deletes[0][3], MappingProxyType)
assert router.deletes == (("wildcard-dep", "file-upload-2", _REQUEST_METADATA),)
@pytest.mark.asyncio

132
tests/unit/llms/bedrock/files/test_bedrock_files_handler.py
View file

@ -1,12 +1,15 @@
import base64
import os
from types import MappingProxyType
from unittest.mock import MagicMock, patch
from typing import Final
from unittest.mock import patch
import httpx
import pytest
import litellm.files.main as files_main
from litellm.llms.bedrock.files.handler import BedrockFilesHandler
from litellm.llms.custom_httpx.http_handler import HTTPHandler
from litellm.types.utils import SpecialEnums
@ -123,88 +126,67 @@ class TestBedrockFilesHandler:
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_call_s3_bucket_name_wins_over_env_bucket(self):
with patch.dict(os.environ, {"AWS_S3_BUCKET_NAME": "env-bucket"}):
assert self.handler._get_configured_s3_bucket_name({"s3_bucket_name": "call-bucket"}) == "call-bucket"
def test_should_trust_proxy_config_s3_bucket_name_for_expected_bucket(self):
trusted_credentials = MappingProxyType({"s3_bucket_name": "safe-bucket"})
def test_env_bucket_is_used_when_the_call_names_none(self):
with patch.dict(os.environ, {"AWS_S3_BUCKET_NAME": "env-bucket"}):
assert self.handler._get_configured_s3_bucket_name({}) == "env-bucket"
with patch.dict(os.environ, {}, clear=True):
assert (
self.handler._get_configured_s3_bucket_name(
{
"s3_bucket_name": "attacker-bucket",
"_litellm_internal_model_credentials": trusted_credentials,
}
)
== "safe-bucket"
)
def test_should_not_trust_user_supplied_internal_credentials_dict(self):
def test_missing_bucket_in_call_and_env_is_rejected(self):
with patch.dict(os.environ, {}, clear=True):
with pytest.raises(ValueError, match="S3 bucket_name is required"):
self.handler._get_configured_s3_bucket_name(
{
"_litellm_internal_model_credentials": {
"s3_bucket_name": "attacker-bucket"
}
}
)
def test_should_require_server_s3_bucket_name(self):
with patch.dict(os.environ, {}, clear=True):
with pytest.raises(ValueError, match="S3 bucket_name is required"):
self.handler._get_configured_s3_bucket_name(
{"s3_bucket_name": "attacker-bucket"}
)
self.handler._get_configured_s3_bucket_name({})
def test_should_forward_trusted_model_credentials_to_bedrock_provider_config():
trusted_credentials = MappingProxyType({"s3_bucket_name": "safe-bucket"})
mock_response = MagicMock()
def _echo_request_url_client() -> HTTPHandler:
return HTTPHandler(
client=httpx.Client(
transport=httpx.MockTransport(lambda request: httpx.Response(200, content=str(request.url).encode()))
)
)
with patch.object(
files_main.base_llm_http_handler,
"retrieve_file_content",
return_value=mock_response,
) as mock_retrieve_file_content:
response = files_main.file_content(
file_id="s3://safe-bucket/litellm-bedrock-files/file.jsonl",
_SDK_AWS_CREDENTIALS: Final = MappingProxyType(
{
"aws_access_key_id": "AKIASDKCALLER",
"aws_secret_access_key": "sdk-caller-secret",
"aws_region_name": "us-west-2",
}
)
def test_sdk_file_content_reads_from_the_bucket_passed_to_the_call(monkeypatch: pytest.MonkeyPatch) -> None:
monkeypatch.delenv("AWS_S3_BUCKET_NAME", raising=False)
monkeypatch.delenv("AWS_S3_OUTPUT_BUCKET_NAME", raising=False)
response: Final = files_main.file_content(
file_id="s3://sdk-bucket/litellm-bedrock-files/batch.jsonl",
custom_llm_provider="bedrock",
s3_bucket_name="sdk-bucket",
client=_echo_request_url_client(),
**_SDK_AWS_CREDENTIALS,
)
assert httpx.URL(response.content.decode()).path == "/sdk-bucket/litellm-bedrock-files/batch.jsonl"
def test_sdk_file_content_rejects_a_file_outside_the_bucket_passed_to_the_call(
monkeypatch: pytest.MonkeyPatch,
) -> None:
monkeypatch.delenv("AWS_S3_BUCKET_NAME", raising=False)
monkeypatch.delenv("AWS_S3_OUTPUT_BUCKET_NAME", raising=False)
with pytest.raises(Exception, match="configured storage bucket"):
files_main.file_content(
file_id="s3://other-bucket/litellm-bedrock-files/batch.jsonl",
custom_llm_provider="bedrock",
_litellm_internal_model_credentials=trusted_credentials,
s3_bucket_name="sdk-bucket",
client=_echo_request_url_client(),
**_SDK_AWS_CREDENTIALS,
)
assert response is mock_response
litellm_params = mock_retrieve_file_content.call_args.kwargs["litellm_params"]
assert litellm_params["_litellm_internal_model_credentials"] is trusted_credentials
assert "s3_bucket_name" not in litellm_params
def test_should_forward_trusted_model_credentials_to_retrieve_provider_config():
trusted_credentials = MappingProxyType({"allow_legacy_cloud_file_ids": True})
mock_response = MagicMock()
with patch.object(
files_main.base_llm_http_handler,
"retrieve_file",
return_value=mock_response,
) as mock_retrieve_file:
response = files_main.file_retrieve(
file_id="gs://safe-bucket/private/file.jsonl",
custom_llm_provider="vertex_ai",
_litellm_internal_model_credentials=trusted_credentials,
)
assert response is mock_response
litellm_params = mock_retrieve_file.call_args.kwargs["litellm_params"]
assert litellm_params["_litellm_internal_model_credentials"] is trusted_credentials
@pytest.mark.asyncio
async def test_afile_content_assumes_role_with_external_id(monkeypatch):
@ -249,7 +231,7 @@ async def test_afile_content_assumes_role_with_external_id(monkeypatch):
return FakeS3Client()
optional_params = {
"_litellm_internal_model_credentials": MappingProxyType({"s3_bucket_name": "safe-bucket"}),
"s3_bucket_name": "safe-bucket",
"aws_region_name": "us-east-1",
"aws_access_key_id": "AKIAFILESDOWNLOADCALLER",
"aws_secret_access_key": "pod-caller-secret",
@ -285,7 +267,7 @@ async def test_afile_content_builds_the_s3_client_with_the_s3_pair_when_it_diffe
return {"Body": FakeS3Body()}
optional_params = {
"_litellm_internal_model_credentials": MappingProxyType({"s3_bucket_name": "safe-bucket"}),
"s3_bucket_name": "safe-bucket",
"aws_region_name": "us-east-1",
"aws_access_key_id": "AKIABEDROCKONLY",
"aws_secret_access_key": "bedrock-only-secret",

103
tests/unit/llms/bedrock/files/test_bedrock_files_transformation.py
View file

@ -9,7 +9,6 @@ from contextlib import AsyncExitStack, closing
from datetime import datetime, timezone
from io import BytesIO
from pathlib import Path
from types import MappingProxyType
from typing import Final
from unittest.mock import MagicMock
from urllib.parse import unquote, urlparse
@ -2351,7 +2350,7 @@ class TestBedrockFileContentTransformation:
monkeypatch.delenv("AWS_S3_BUCKET_NAME", raising=False)
monkeypatch.delenv("AWS_S3_OUTPUT_BUCKET_NAME", raising=False)
litellm_params: Final = _trusted_bucket_snapshot(
litellm_params: Final = _deployment_bucket_params(
s3_bucket_name="in-bucket/pfx",
s3_output_bucket_name="out-bucket/outpfx",
)
@ -2541,7 +2540,7 @@ class TestBedrockFileContentTransformation:
def test_transform_file_content_request_requires_configured_bucket(
self, monkeypatch
):
"""Without a server-configured bucket (env or snapshot), the request must fail
"""Without a bucket (on the call or in env), the request must fail
before any S3 call rather than guessing a bucket from the file id."""
from litellm.llms.bedrock.files.transformation import BedrockFilesConfig
@ -2566,30 +2565,24 @@ class TestBedrockFileContentTransformation:
litellm_params=self._litellm_params(),
)
def _trusted(self, **deployment_litellm_params) -> dict:
"""Build the trusted snapshot the way the proxy does: deployment
litellm_params funneled through ``CredentialLiteLLMParams`` (the strict
allowlist ``get_deployment_credentials_with_provider`` applies) before
retrieval ever sees them. Injecting a raw ``MappingProxyType`` would
bypass that filter and hide whether a bucket field actually survives
into the snapshot in production."""
from types import MappingProxyType
def _deployment_params(self, **deployment_litellm_params) -> dict:
"""Deployment litellm_params funneled through ``CredentialLiteLLMParams``
(the strict allowlist ``get_deployment_credentials_with_provider`` applies)
the way the proxy hands them to retrieval, so a bucket field dropped from
that allowlist fails these tests."""
from litellm.types.router import CredentialLiteLLMParams
snapshot = CredentialLiteLLMParams(**deployment_litellm_params).model_dump(
credentials = CredentialLiteLLMParams(**deployment_litellm_params).model_dump(
exclude_none=True
)
params = self._litellm_params()
params["_litellm_internal_model_credentials"] = MappingProxyType(snapshot)
return params
return {**self._litellm_params(), **credentials}
def test_retrieves_from_distinct_output_bucket(self, monkeypatch):
"""Batch outputs can land in a separate s3_output_bucket_name. Retrieval
must validate the file id against the output bucket too, not just the
input bucket, or the very outputs the feature serves are unreachable.
The snapshot is built through the production credential filter, so this
fails if s3_output_bucket_name is dropped from that allowlist."""
The params go through the production credential filter, so this fails
if s3_output_bucket_name is dropped from that allowlist."""
from litellm.llms.bedrock.files.transformation import BedrockFilesConfig
monkeypatch.delenv("AWS_S3_BUCKET_NAME", raising=False)
@ -2600,7 +2593,7 @@ class TestBedrockFileContentTransformation:
"file_id": "s3://out-bucket/litellm-batch-outputs/job/in.jsonl.out"
},
optional_params={},
litellm_params=self._trusted(
litellm_params=self._deployment_params(
s3_bucket_name="in-bucket", s3_output_bucket_name="out-bucket"
),
)
@ -2611,8 +2604,8 @@ class TestBedrockFileContentTransformation:
)
def test_output_bucket_falls_back_to_env(self, monkeypatch):
"""The output bucket resolves from AWS_S3_OUTPUT_BUCKET_NAME when not in
the trusted snapshot, mirroring the input-bucket env fallback."""
"""The output bucket resolves from AWS_S3_OUTPUT_BUCKET_NAME when the
call has none, mirroring the input-bucket env fallback."""
from litellm.llms.bedrock.files.transformation import BedrockFilesConfig
monkeypatch.setenv("AWS_S3_BUCKET_NAME", "in-bucket")
@ -2644,7 +2637,7 @@ class TestBedrockFileContentTransformation:
"file_id": "s3://in-bucket/litellm-batch-outputs/job/in.jsonl.out"
},
optional_params={},
litellm_params=self._trusted(
litellm_params=self._deployment_params(
s3_bucket_name="in-bucket", s3_output_bucket_name="out-bucket"
),
)
@ -2669,7 +2662,7 @@ class TestBedrockFileContentTransformation:
"file_id": "s3://other-bucket/litellm-batch-outputs/job/x.jsonl.out"
},
optional_params={},
litellm_params=self._trusted(
litellm_params=self._deployment_params(
s3_bucket_name="in-bucket", s3_output_bucket_name="out-bucket"
),
)
@ -2694,22 +2687,16 @@ class TestBedrockFileContentTransformation:
litellm_params=self._litellm_params(),
)
def test_bucket_resolved_from_trusted_model_credentials(self, monkeypatch):
"""Per-model s3_bucket_name must be honored via the server-side credential snapshot."""
from types import MappingProxyType
def test_bucket_resolved_from_call_s3_bucket_name(self, monkeypatch):
"""An s3_bucket_name on the call is honored without AWS_S3_BUCKET_NAME."""
from litellm.llms.bedrock.files.transformation import BedrockFilesConfig
monkeypatch.delenv("AWS_S3_BUCKET_NAME", raising=False)
litellm_params = self._litellm_params()
litellm_params["_litellm_internal_model_credentials"] = MappingProxyType(
{"s3_bucket_name": "my-bucket"}
)
url, _ = BedrockFilesConfig().transform_file_content_request(
file_content_request={"file_id": self.S3_URI},
optional_params={},
litellm_params=litellm_params,
litellm_params={**self._litellm_params(), "s3_bucket_name": "my-bucket"},
)
assert url == self.EXPECTED_URL
@ -3226,13 +3213,13 @@ def _bedrock_s3_params() -> dict:
}
def _trusted_bucket_snapshot(**deployment_litellm_params) -> dict:
from types import MappingProxyType
def _deployment_bucket_params(**deployment_litellm_params: object) -> dict[str, object]:
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)}
credentials: Final = CredentialLiteLLMParams.model_validate(deployment_litellm_params).model_dump(
exclude_none=True
)
return {**_bedrock_s3_params(), **credentials}
class TestBedrockFileDeletionTransformation:
@ -3287,7 +3274,7 @@ class TestBedrockFileDeletionTransformation:
assert url == self.EXPECTED_URL
def test_transform_delete_file_request_uses_trusted_snapshot_bucket(self, monkeypatch):
def test_transform_delete_file_request_uses_call_bucket(self, monkeypatch):
from litellm.llms.bedrock.files.transformation import BedrockFilesConfig
monkeypatch.delenv("AWS_S3_BUCKET_NAME", raising=False)
@ -3295,7 +3282,7 @@ class TestBedrockFileDeletionTransformation:
url, _ = BedrockFilesConfig().transform_delete_file_request(
file_id=self.S3_URI,
optional_params={},
litellm_params=_trusted_bucket_snapshot(s3_bucket_name="my-bucket"),
litellm_params=_deployment_bucket_params(s3_bucket_name="my-bucket"),
)
assert url == self.EXPECTED_URL
@ -3595,7 +3582,7 @@ class TestBedrockFileListTransformation:
"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):
def test_transform_list_files_request_uses_call_bucket(self, monkeypatch):
from litellm.llms.bedrock.files.transformation import BedrockFilesConfig
monkeypatch.delenv("AWS_S3_BUCKET_NAME", raising=False)
@ -3603,7 +3590,7 @@ class TestBedrockFileListTransformation:
url, params = BedrockFilesConfig().transform_list_files_request(
purpose=None,
optional_params={},
litellm_params=_trusted_bucket_snapshot(s3_bucket_name="my-bucket"),
litellm_params=_deployment_bucket_params(s3_bucket_name="my-bucket"),
)
assert url == self.BUCKET_URL
@ -3696,7 +3683,7 @@ class TestBedrockFileListTransformation:
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):
def test_file_list_uses_call_bucket_without_env(self, monkeypatch):
import httpx
import respx
@ -3711,7 +3698,7 @@ class TestBedrockFileListTransformation:
files = litellm.file_list(
custom_llm_provider="bedrock",
**_trusted_bucket_snapshot(s3_bucket_name="my-bucket"),
**_deployment_bucket_params(s3_bucket_name="my-bucket"),
)
assert route.called
@ -3766,7 +3753,7 @@ class TestBedrockFileListTransformation:
monkeypatch.delenv("AWS_S3_BUCKET_NAME", raising=False)
monkeypatch.delenv("AWS_S3_OUTPUT_BUCKET_NAME", raising=False)
litellm_params = _trusted_bucket_snapshot(
litellm_params = _deployment_bucket_params(
s3_bucket_name="my-bucket", s3_output_bucket_name="my-output-bucket/team-a"
)
@ -3852,7 +3839,7 @@ class TestBedrockFileListTransformation:
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")
litellm_params = _deployment_bucket_params(s3_output_bucket_name="my-output-bucket")
url, params = BedrockFilesConfig().transform_list_files_request(
purpose="batch_output", optional_params={}, litellm_params=litellm_params
@ -3879,7 +3866,7 @@ class TestBedrockFileListTransformation:
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"),
**_deployment_bucket_params(s3_bucket_name="my-bucket", s3_output_bucket_name="my-output-bucket"),
LIST_FILES_PURPOSE_PARAM: "batch_output",
},
)
@ -3903,7 +3890,7 @@ class TestBedrockFileListTransformation:
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"),
**_deployment_bucket_params(s3_bucket_name="my-bucket", s3_output_bucket_name="my-output-bucket"),
)
assert route.called
@ -3930,7 +3917,7 @@ class TestBedrockFileListTransformation:
files = litellm.file_list(
custom_llm_provider="bedrock",
**_trusted_bucket_snapshot(s3_bucket_name="my-bucket", s3_output_bucket_name="my-output-bucket"),
**_deployment_bucket_params(s3_bucket_name="my-bucket", s3_output_bucket_name="my-output-bucket"),
)
assert (input_route.call_count, output_route.call_count) == (1, 1)
@ -3962,7 +3949,7 @@ class TestBedrockFileListTransformation:
files = litellm.file_list(
custom_llm_provider="bedrock",
**_trusted_bucket_snapshot(s3_bucket_name="my-bucket", s3_output_bucket_name="my-output-bucket"),
**_deployment_bucket_params(s3_bucket_name="my-bucket", s3_output_bucket_name="my-output-bucket"),
)
requested_urls = [str(call.request.url) for call in respx.calls]
@ -3975,7 +3962,7 @@ class TestBedrockFileListTransformation:
assert [file.id for file in files] == [*self.PAGED_IDS, self.OUTPUT_BUCKET_ID]
@pytest.mark.parametrize(
("purpose", "bucket_snapshot"),
("purpose", "bucket_params"),
[
pytest.param(None, {"s3_bucket_name": "my-bucket"}, id="outputs-share-the-input-bucket"),
pytest.param(
@ -3986,7 +3973,7 @@ class TestBedrockFileListTransformation:
],
)
def test_file_list_leaves_the_output_bucket_alone_unless_an_unfiltered_list_needs_it(
self, monkeypatch, purpose, bucket_snapshot
self, monkeypatch, purpose, bucket_params
):
import httpx
import respx
@ -4003,7 +3990,7 @@ class TestBedrockFileListTransformation:
)
files = litellm.file_list(
custom_llm_provider="bedrock", purpose=purpose, **_trusted_bucket_snapshot(**bucket_snapshot)
custom_llm_provider="bedrock", purpose=purpose, **_deployment_bucket_params(**bucket_params)
)
assert (input_route.call_count, output_route.call_count) == (1, 0)
@ -4019,7 +4006,7 @@ class TestBedrockFileListTransformation:
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")
litellm_params = _deployment_bucket_params(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)
@ -4130,7 +4117,7 @@ class TestBedrockFileListTransformation:
custom_llm_provider="bedrock",
purpose="batch",
timeout=12,
**_trusted_bucket_snapshot(s3_bucket_name="my-bucket"),
**_deployment_bucket_params(s3_bucket_name="my-bucket"),
)
self._assert_paged_listing(first_page, last_page, files)
@ -4152,7 +4139,7 @@ class TestBedrockFileListTransformation:
custom_llm_provider="bedrock",
purpose="batch",
timeout=12,
**_trusted_bucket_snapshot(s3_bucket_name="my-bucket"),
**_deployment_bucket_params(s3_bucket_name="my-bucket"),
)
self._assert_paged_listing(first_page, last_page, files)
@ -4209,7 +4196,7 @@ class TestBedrockFileListTransformation:
files = litellm.file_list(
custom_llm_provider="bedrock",
purpose="batch",
**_trusted_bucket_snapshot(s3_bucket_name="my-bucket"),
**_deployment_bucket_params(s3_bucket_name="my-bucket"),
)
self._assert_capped_listing(route, files)
@ -4230,7 +4217,7 @@ class TestBedrockFileListTransformation:
files = await litellm.afile_list(
custom_llm_provider="bedrock",
purpose="batch",
**_trusted_bucket_snapshot(s3_bucket_name="my-bucket"),
**_deployment_bucket_params(s3_bucket_name="my-bucket"),
)
self._assert_capped_listing(route, files)
@ -4321,7 +4308,7 @@ def test_transform_file_content_request_signs_with_the_s3_pair_from_litellm_para
litellm_params = {
**_SPLIT_IDENTITY_PARAMS,
"_litellm_internal_model_credentials": MappingProxyType({"s3_bucket_name": "safe-bucket"}),
"s3_bucket_name": "safe-bucket",
}
BedrockFilesConfig().transform_file_content_request(
file_content_request={"file_id": "s3://safe-bucket/litellm-bedrock-files-model-id-abc.jsonl"},

6
tests/unit/llms/vertex_ai/files/test_vertex_ai_files_handler.py
View file

@ -3,7 +3,6 @@ Test Vertex AI files handler functionality
"""
import re
from types import MappingProxyType
import pytest
from unittest.mock import AsyncMock, patch
@ -62,15 +61,14 @@ class TestVertexAIFilesHandler:
configured_bucket_name="test-bucket",
)
def test_extract_bucket_and_object_from_file_id_allows_trusted_legacy_flag(self):
def test_extract_bucket_and_object_from_file_id_allows_legacy_flag_on_the_call(self):
file_id = "gs%3A%2F%2Ftest-bucket%2Ftest-file.txt"
trusted_credentials = MappingProxyType({"allow_legacy_cloud_file_ids": True})
bucket_name, object_path = self.handler._extract_bucket_and_object_from_file_id(
file_id=file_id,
configured_bucket_name="test-bucket",
litellm_params={
"_litellm_internal_model_credentials": trusted_credentials,
"allow_legacy_cloud_file_ids": True,
},
)

33
tests/unit/llms/vertex_ai/files/test_vertex_ai_files_transformation.py
View file

@ -6,8 +6,8 @@ Includes tests for Vertex AI batch output transformation to OpenAI format.
import json
import urllib.parse
from collections.abc import Mapping, Sequence
from typing import Final
from types import MappingProxyType
from typing import Final
from urllib.parse import parse_qs, urlparse
import httpx
@ -72,46 +72,24 @@ class TestParseGcsUri:
litellm_params={"gcs_bucket_name": "my-bucket"},
)
def test_should_reject_request_supplied_legacy_flag(self, config):
with pytest.raises(ValueError, match="LiteLLM-managed"):
config._parse_gcs_uri(
"gs://my-bucket/private/object.txt",
litellm_params={
"gcs_bucket_name": "my-bucket",
"allow_legacy_cloud_file_ids": True,
},
)
def test_should_allow_legacy_object_path_with_trusted_server_flag(self, config):
trusted_credentials = MappingProxyType({"allow_legacy_cloud_file_ids": True})
def test_should_allow_legacy_object_path_when_the_call_sets_the_flag(self, config):
bucket, encoded = config._parse_gcs_uri(
"gs://my-bucket/private/object.txt",
litellm_params={
"gcs_bucket_name": "my-bucket",
"_litellm_internal_model_credentials": trusted_credentials,
"allow_legacy_cloud_file_ids": True,
},
)
assert bucket == "my-bucket"
assert encoded == urllib.parse.quote("private/object.txt", safe="")
def test_should_reject_user_supplied_legacy_flag_snapshot(self, config):
with pytest.raises(ValueError, match="LiteLLM-managed"):
config._parse_gcs_uri(
"gs://my-bucket/private/object.txt",
litellm_params={
"gcs_bucket_name": "my-bucket",
"_litellm_internal_model_credentials": {"allow_legacy_cloud_file_ids": True},
},
)
def test_should_keep_configured_prefix_for_legacy_object_path(self, config):
trusted_credentials = MappingProxyType({"allow_legacy_cloud_file_ids": True})
bucket, encoded = config._parse_gcs_uri(
"gs://my-bucket/team-a/private/object.txt",
litellm_params={
"gcs_bucket_name": "my-bucket/team-a",
"_litellm_internal_model_credentials": trusted_credentials,
"allow_legacy_cloud_file_ids": True,
},
)
@ -119,13 +97,12 @@ class TestParseGcsUri:
assert encoded == urllib.parse.quote("team-a/private/object.txt", safe="")
def test_should_reject_legacy_object_outside_configured_prefix(self, config):
trusted_credentials = MappingProxyType({"allow_legacy_cloud_file_ids": True})
with pytest.raises(ValueError, match="configured storage prefix"):
config._parse_gcs_uri(
"gs://my-bucket/team-b/private/object.txt",
litellm_params={
"gcs_bucket_name": "my-bucket/team-a",
"_litellm_internal_model_credentials": trusted_credentials,
"allow_legacy_cloud_file_ids": True,
},
)

118
tests/unit/proxy/auth/test_auth_utils.py
View file

@ -5311,6 +5311,124 @@ def test_get_end_user_id_from_request_body_backwards_compatibility():
assert get_end_user_id_from_request_body(request_body) == expected_end_user_id
_BUCKET_ALLOWLIST: Final = {"allowed_client_side_s3_buckets": ["team-batch-bucket"]}
_ENDPOINT_ALLOWLIST: Final = {"allowed_client_side_s3_endpoint_urls": ["https://s3.us-west-2.amazonaws.com"]}
def _batches_body_is_safe(body: dict[str, object], general_settings: dict[str, object]) -> bool:
return is_request_body_safe(
request_body=body,
general_settings=general_settings,
llm_router=None,
model="",
route="/v1/batches",
)
@pytest.mark.parametrize(
"field,value",
[
("s3_bucket_name", "team-batch-bucket"),
("s3_output_bucket_name", "team-batch-bucket"),
("allow_legacy_cloud_file_ids", True),
("s3_endpoint_url", "https://s3.us-west-2.amazonaws.com"),
],
)
def test_storage_setting_in_request_body_is_rejected_without_an_allowlist(field: str, value: object) -> None:
with pytest.raises(ValueError, match="Rejected Request") as error:
_batches_body_is_safe({field: value}, {})
assert field in str(error.value)
@pytest.mark.parametrize("field", ["s3_bucket_name", "s3_output_bucket_name"])
def test_bucket_on_the_proxy_allowlist_is_accepted(field: str) -> None:
assert _batches_body_is_safe({field: "team-batch-bucket"}, _BUCKET_ALLOWLIST) is True
@pytest.mark.parametrize("field", ["s3_bucket_name", "s3_output_bucket_name"])
def test_bucket_off_the_proxy_allowlist_is_rejected_and_names_the_setting(field: str) -> None:
with pytest.raises(ValueError, match=field) as error:
_batches_body_is_safe({field: "someone-elses-bucket"}, _BUCKET_ALLOWLIST)
assert "general_settings.allowed_client_side_s3_buckets" in str(error.value)
@pytest.mark.parametrize(
"general_settings,accepted_body,rejected_field",
[
(
{**_BUCKET_ALLOWLIST, "allowed_client_side_s3_endpoint_urls": None},
{"s3_bucket_name": "team-batch-bucket"},
"s3_endpoint_url",
),
(
{**_ENDPOINT_ALLOWLIST, "allowed_client_side_s3_buckets": None},
{"s3_endpoint_url": "https://s3.us-west-2.amazonaws.com"},
"s3_bucket_name",
),
],
)
def test_allowlist_left_null_in_config_is_treated_as_empty(
general_settings: dict[str, object], accepted_body: dict[str, object], rejected_field: str
) -> None:
assert _batches_body_is_safe(accepted_body, general_settings) is True
with pytest.raises(ValueError, match=f"{rejected_field} is not allowed"):
_batches_body_is_safe({rejected_field: "team-batch-bucket"}, general_settings)
def test_endpoint_on_the_proxy_allowlist_is_accepted() -> None:
assert _batches_body_is_safe({"s3_endpoint_url": "https://s3.us-west-2.amazonaws.com"}, _ENDPOINT_ALLOWLIST) is True
def test_endpoint_off_the_proxy_allowlist_is_rejected_and_names_the_setting() -> None:
with pytest.raises(ValueError, match="s3_endpoint_url") as error:
_batches_body_is_safe({"s3_endpoint_url": "https://attacker.example"}, _ENDPOINT_ALLOWLIST)
assert "general_settings.allowed_client_side_s3_endpoint_urls" in str(error.value)
def test_bucket_allowlist_admits_neither_an_endpoint_nor_the_legacy_file_flag() -> None:
with pytest.raises(ValueError, match="s3_endpoint_url"):
_batches_body_is_safe(
{"s3_bucket_name": "team-batch-bucket", "s3_endpoint_url": "team-batch-bucket"},
_BUCKET_ALLOWLIST,
)
with pytest.raises(ValueError, match="allow_legacy_cloud_file_ids"):
_batches_body_is_safe(
{"s3_bucket_name": "team-batch-bucket", "allow_legacy_cloud_file_ids": True},
_BUCKET_ALLOWLIST,
)
def test_endpoint_allowlist_does_not_admit_a_bucket() -> None:
with pytest.raises(ValueError, match="s3_bucket_name"):
_batches_body_is_safe({"s3_bucket_name": "https://s3.us-west-2.amazonaws.com"}, _ENDPOINT_ALLOWLIST)
def test_bucket_nested_in_extra_body_is_checked_against_the_allowlist() -> None:
assert _batches_body_is_safe({"extra_body": {"s3_bucket_name": "team-batch-bucket"}}, _BUCKET_ALLOWLIST) is True
with pytest.raises(ValueError, match="s3_bucket_name"):
_batches_body_is_safe({"extra_body": {"s3_bucket_name": "someone-elses-bucket"}}, _BUCKET_ALLOWLIST)
def test_proxy_wide_client_side_credentials_still_admit_any_bucket() -> None:
assert (
_batches_body_is_safe(
{"s3_bucket_name": "any-bucket", "allow_legacy_cloud_file_ids": True},
{"allow_client_side_credentials": True},
)
is True
)
def test_other_blocked_fields_do_not_point_at_the_storage_allowlist() -> None:
with pytest.raises(ValueError, match="api_base") as error:
_batches_body_is_safe({"api_base": "https://attacker.example"}, _BUCKET_ALLOWLIST)
assert "allowed_client_side" not in str(error.value)
def _request_with_content_length(content_length: int) -> Request:
return Request(
{

5
tests/unit/proxy/batches_endpoints/test_endpoints.py
View file

@ -1723,11 +1723,6 @@ async def test_retrieve__unified_batch_id_routes_to_router(retrieve_harness):
assert retrieve_harness.router_aretrieve.call_count == 1
retrieve_harness.litellm_aretrieve.assert_not_called()
# Credentials are resolved for the deployment behind the unified id so the batch's
# output file can be read for cost accounting. This id resolves to nothing here, and
# the retrieve must still serve the batch rather than fail on the lookup.
retrieve_harness.creds_resolver.assert_called_once_with("gpt-4o-mini", form="with_provider")
# router receives the (still-encoded) batch id verbatim - this layer does
# not decode it for the unified path.
assert retrieve_harness.router_kwargs() == {"batch_id": "batch-unified-blob"}

47
tests/unit/proxy/common_utils/test_check_batch_cost.py
View file

@ -362,17 +362,19 @@ class TestCheckBatchCost:
assert update_data["status"] == "complete"
@pytest.mark.asyncio
async def test_output_fetch_passes_deployment_credentials_as_trusted_snapshot(
self, check_batch_cost_instance, mock_prisma_client, mock_llm_router
async def test_output_fetch_reads_from_the_deployment_bucket(
self, check_batch_cost_instance, mock_prisma_client, mock_llm_router, monkeypatch
):
"""Bedrock resolves the output bucket ONLY from the immutable snapshot kwarg.
"""Without the deployment's s3_bucket_name the cost poller cannot read a Bedrock
batch's output file, and every completed Bedrock batch stays unbilled."""
import httpx
Spreading the credentials as plain kwargs is not enough: get_litellm_params drops
s3_bucket_name, so without _litellm_internal_model_credentials the cost poller
cannot read the output file and every completed Bedrock batch stays unbilled.
"""
from types import MappingProxyType
from unittest.mock import patch
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()
mock_prisma_client.db.litellm_managedobjecttable.update_many = AsyncMock(return_value=1)
mock_prisma_client.db.litellm_managedobjecttable.update = AsyncMock()
@ -386,7 +388,7 @@ class TestCheckBatchCost:
mock_response = MagicMock()
mock_response.status = "completed"
mock_response.output_file_id = "file-output-123"
mock_response.output_file_id = "s3://configured-batch-bucket/litellm-batch-outputs/job-1/input.jsonl.out"
mock_response.model_dump_json.return_value = '{"id":"batch-1","status":"completed"}'
mock_llm_router.aretrieve_batch = AsyncMock(return_value=mock_response)
@ -395,6 +397,8 @@ class TestCheckBatchCost:
"custom_llm_provider": "bedrock",
"s3_bucket_name": "configured-batch-bucket",
"aws_region_name": "us-east-1",
"aws_access_key_id": "AKIAEXAMPLE",
"aws_secret_access_key": "secret",
}
)
@ -404,12 +408,10 @@ class TestCheckBatchCost:
mock_deployment.model_info.model_dump.return_value = {}
mock_llm_router.registry.return_value.deployment = MagicMock(return_value=mock_deployment)
mock_file_content = MagicMock()
mock_file_content.content = b'{"recordId":"req-1"}'
decoded_id = "llm_model_id,model-123;llm_batch_id,batch-456;"
with (
respx.mock(assert_all_called=False) as s3,
patch(
"litellm.proxy.openai_files_endpoints.common_utils._is_base64_encoded_unified_file_id",
side_effect=[decoded_id, None],
@ -422,11 +424,6 @@ class TestCheckBatchCost:
"litellm.proxy.openai_files_endpoints.common_utils.get_batch_id_from_unified_batch_id",
return_value="batch-456",
),
patch(
"litellm.files.main.afile_content",
new_callable=AsyncMock,
return_value=mock_file_content,
) as mock_afile_content,
patch(
"litellm.batches.batch_utils.get_file_content_as_dictionary",
return_value=[{"recordId": "req-1"}],
@ -447,17 +444,15 @@ class TestCheckBatchCost:
mock_logging_obj = MagicMock()
mock_logging_obj.async_success_handler = AsyncMock()
mock_logging_cls.return_value = mock_logging_obj
s3.get(url__regex=r"^https://s3\.us-east-1\.amazonaws\.com/").mock(
return_value=httpx.Response(200, content=b'{"recordId":"req-1"}')
)
await check_batch_cost_instance.check_batch_cost()
mock_afile_content.assert_awaited()
passed_kwargs = mock_afile_content.await_args[1]
snapshot = passed_kwargs.get("_litellm_internal_model_credentials")
assert snapshot is not None, "cost poller must pass the trusted credential snapshot"
assert isinstance(snapshot, MappingProxyType), (
"snapshot must be a MappingProxyType; a plain dict is rejected by get_configured_s3_bucket_name"
)
assert snapshot["s3_bucket_name"] == "configured-batch-bucket"
assert [str(call.request.url) for call in s3.calls] == [
"https://s3.us-east-1.amazonaws.com/configured-batch-bucket/litellm-batch-outputs/job-1/input.jsonl.out"
]
@pytest.mark.asyncio
async def test_poller_prices_with_deployment_registered_batch_rates(

68
tests/unit/proxy/openai_files_endpoint/test_files_common_utils.py
View file

@ -396,74 +396,6 @@ async def test_a_caller_that_handed_off_accounting_still_leaves_the_marker_alone
assert data["status"] == "complete"
# =========================================================================== #
# add_internal_model_credentials - the snapshot that lets a completed
# batch's output file be read, and therefore its cost be recorded
# =========================================================================== #
def test_add_internal_model_credentials_attaches_an_immutable_snapshot():
"""Cost accounting for a completed batch reads its output file, and Bedrock resolves
that bucket only from this snapshot. It must be immutable so nothing downstream can
redirect the bucket that managed file ids are validated against."""
from litellm.proxy.openai_files_endpoints.common_utils import (
add_internal_model_credentials,
)
router = route_entry_points_to_stubs(MagicMock())
router.get_deployment_credentials_with_provider = MagicMock(
return_value={"s3_bucket_name": "configured-bucket", "aws_region_name": "us-east-1"}
)
data = {"batch_id": "unified-batch-id"}
add_internal_model_credentials(data=data, llm_router=router, model_id="deployment-1")
snapshot = data["_litellm_internal_model_credentials"]
assert snapshot["s3_bucket_name"] == "configured-bucket"
assert isinstance(snapshot, MappingProxyType)
with pytest.raises(TypeError):
snapshot["s3_bucket_name"] = "attacker-bucket"
router.get_deployment_credentials_with_provider.assert_called_once_with(model_id="deployment-1")
@pytest.mark.parametrize(
"model_id, credentials",
[(None, {"s3_bucket_name": "b"}), ("deployment-1", None)],
ids=["no-model-id", "deployment-has-no-credentials"],
)
def test_add_internal_model_credentials_is_a_noop_without_a_resolvable_deployment(model_id, credentials):
"""An unroutable batch must be left alone rather than given an empty snapshot, which
would look like a configured bucket of nothing."""
from litellm.proxy.openai_files_endpoints.common_utils import (
add_internal_model_credentials,
)
router = route_entry_points_to_stubs(MagicMock())
router.get_deployment_credentials_with_provider = MagicMock(return_value=credentials)
data = {"batch_id": "unified-batch-id"}
add_internal_model_credentials(data=data, llm_router=router, model_id=model_id)
assert "_litellm_internal_model_credentials" not in data
def test_add_internal_model_credentials_survives_a_failing_deployment_lookup():
"""The snapshot only enables cost accounting, so a batch whose deployment no longer
resolves, which happens when a model group is removed while batches are in flight,
must still be retrievable rather than failing the request on the lookup."""
from litellm.proxy.openai_files_endpoints.common_utils import (
add_internal_model_credentials,
)
router = route_entry_points_to_stubs(MagicMock())
router.get_deployment_credentials_with_provider = MagicMock(side_effect=KeyError("deployment-gone"))
data = {"batch_id": "unified-batch-id"}
add_internal_model_credentials(data=data, llm_router=router, model_id="deployment-gone")
assert data == {"batch_id": "unified-batch-id"}
from openai.types.batch import BatchRequestCounts
from litellm.proxy.openai_files_endpoints.common_utils import (

80
tests/unit/proxy/openai_files_endpoint/test_files_endpoint.py
View file

@ -4128,28 +4128,28 @@ def test_require_managed_files_rejects_raw_provider_file_id(
mock_call.assert_not_called()
def test_get_file_content_model_routed_attaches_trusted_model_credentials(monkeypatch):
"""A managed batch output id routes by model, and that branch must build the snapshot.
The managed-files pre-call hook sets data["model"] for any id carrying
llm_output_file_id, so batch output retrieval always takes the model-routed branch
and never reaches managed_files_obj.afile_content. Bedrock resolves its output
bucket only from _litellm_internal_model_credentials, so without the snapshot every
Bedrock batch output retrieval fails with "S3 bucket_name is required".
"""
def test_get_file_content_model_routed_reads_from_the_deployment_bucket(
monkeypatch: pytest.MonkeyPatch,
) -> None:
"""A managed batch output id routes by model, so Bedrock batch output retrieval only
works when the deployment's s3_bucket_name reaches the SDK. Without it every
retrieval fails with "S3 bucket_name is required"."""
import base64
from types import MappingProxyType
import litellm.proxy.proxy_server as ps
from litellm.proxy._types import LitellmUserRoles
from litellm.types.utils import SpecialEnums
monkeypatch.setattr(litellm, "disable_aiohttp_transport", True)
router = Router(
model_list=[
{
"model_name": "anthropic.batch.claude-4.5-haiku",
"litellm_params": {
"model": "bedrock/anthropic.claude-haiku-4-5-20251001-v1:0",
"aws_access_key_id": "AKIAEXAMPLE",
"aws_secret_access_key": "secret",
"aws_region_name": "us-east-1",
"s3_bucket_name": "configured-batch-bucket",
},
@ -4173,27 +4173,11 @@ def test_get_file_content_model_routed_attaches_trusted_model_credentials(monkey
monkeypatch.setattr("litellm.proxy.proxy_server.master_key", None)
monkeypatch.setattr("litellm.proxy.proxy_server.llm_router", router)
# One frozen snapshot per call rather than one dict merged across calls, so a second
# invocation is visible instead of silently overwriting the first.
calls: list[MappingProxyType] = []
async def _mock_router_afile_content(**kwargs):
calls.append(MappingProxyType(dict(kwargs)))
return HttpxBinaryResponseContent(
response=httpx.Response(
status_code=200,
content=b'{"recordId":"req-1"}',
headers={"content-type": "application/octet-stream"},
)
)
monkeypatch.setattr(router, "afile_content", _mock_router_afile_content)
unified_id = SpecialEnums.LITELLM_MANAGED_FILE_COMPLETE_STR.value.format(
"application/jsonl",
"unified-output-id",
"anthropic.batch.claude-4.5-haiku",
"llm_output_file_id,s3://configured-batch-bucket/out/batch.jsonl",
"s3://configured-batch-bucket/litellm-batch-outputs/job/output.jsonl",
"bedrock-batch-deployment-id",
)
encoded_id = base64.urlsafe_b64encode(unified_id.encode()).decode().rstrip("=")
@ -4204,21 +4188,22 @@ def test_get_file_content_model_routed_attaches_trusted_model_credentials(monkey
user_id="test-user",
)
try:
response = client.get(
f"/v1/files/{encoded_id}/content",
headers={"Authorization": "Bearer test-key", "custom-llm-provider": "bedrock"},
)
with respx.mock(assert_all_called=False) as s3:
s3_read: Final = s3.get(url__regex=r"https://.*amazonaws\.com/.*").mock(
return_value=httpx.Response(200, content=b'{"recordId":"req-1"}')
)
response = client.get(
f"/v1/files/{encoded_id}/content",
headers={"Authorization": "Bearer test-key", "custom-llm-provider": "bedrock"},
)
finally:
app.dependency_overrides.pop(ps.user_api_key_auth, None)
assert response.status_code == 200, response.text
assert len(calls) == 1, f"expected exactly one routed retrieval, got {len(calls)}"
snapshot = calls[0].get("_litellm_internal_model_credentials")
assert snapshot is not None, "model-routed branch must attach the trusted credential snapshot"
assert isinstance(
snapshot, MappingProxyType
), "snapshot must be a MappingProxyType; a plain dict is rejected by get_configured_s3_bucket_name"
assert snapshot["s3_bucket_name"] == "configured-batch-bucket"
assert response.content == b'{"recordId":"req-1"}'
assert [call.request.url.path for call in s3_read.calls] == [
"/configured-batch-bucket/litellm-batch-outputs/job/output.jsonl"
]
def _unified_managed_file_id() -> str:
@ -5471,16 +5456,11 @@ 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=<bedrock model> 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
def test_list_files_target_model_names_passes_the_deployment_bucket(
mocker: MockerFixture, monkeypatch: pytest.MonkeyPatch
) -> None:
"""GET /v1/files?target_model_names=<bedrock model> lists the deployment's bucket,
so its s3_bucket_name must reach litellm.afile_list."""
import litellm.proxy.proxy_server as ps
from litellm.proxy._types import LitellmUserRoles
@ -5532,9 +5512,7 @@ def test_list_files_target_model_names_passes_trusted_bedrock_credentials(
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"
assert captured_kwargs["s3_bucket_name"] == "my-bucket"
proxy_logging_obj.post_call_failure_hook.assert_not_called()

40
tests/unit/proxy/test_model_based_routing_files_batches.py
View file

@ -6,8 +6,6 @@ Tests the model-based routing ID encoding/decoding used by the batch
and file proxy endpoints.
"""
from types import MappingProxyType
from litellm.proxy.openai_files_endpoints.common_utils import (
decode_model_from_file_id,
encode_file_id_with_model,
@ -54,40 +52,24 @@ class TestEncodeFileIdWithModel:
class TestPrepareDataWithCredentials:
def test_preserves_trusted_internal_credentials_snapshot(self):
data = {"file_id": "file-abc"}
def test_deployment_bucket_settings_override_the_request_and_reach_the_sdk_as_kwargs(self):
data = {"file_id": "file-abc", "s3_bucket_name": "request-bucket"}
credentials = {
"custom_llm_provider": "bedrock",
"s3_bucket_name": "safe-bucket",
"s3_bucket_name": "deployment-bucket",
"s3_output_bucket_name": "deployment-output-bucket",
"allow_legacy_cloud_file_ids": True,
}
prepare_data_with_credentials(
data=data,
credentials=credentials,
include_internal_credentials=True,
)
prepare_data_with_credentials(data=data, credentials=credentials, file_id="decoded-file-id")
assert data["s3_bucket_name"] == "safe-bucket"
assert "custom_llm_provider" not in data
assert isinstance(
data["_litellm_internal_model_credentials"], type(MappingProxyType({}))
)
assert (
data["_litellm_internal_model_credentials"]["s3_bucket_name"]
== "safe-bucket"
)
def test_does_not_add_internal_credentials_by_default(self):
data = {"file_id": "file-abc"}
credentials = {
"custom_llm_provider": "bedrock",
"s3_bucket_name": "safe-bucket",
assert data == {
"file_id": "decoded-file-id",
"s3_bucket_name": "deployment-bucket",
"s3_output_bucket_name": "deployment-output-bucket",
"allow_legacy_cloud_file_ids": True,
}
prepare_data_with_credentials(data=data, credentials=credentials)
assert "_litellm_internal_model_credentials" not in data
class TestRoundTrip:
"""Tests for encode -> decode round-trip integrity."""

37
tests/unit/test_azure_ad_token_credential_resolution.py
View file

@ -149,11 +149,11 @@ class TestRouterCredentialResolutionS3OutputBucket:
"""Same strict-dump trap as azure_ad_token (#30235), for Bedrock batch
file retrieval (#26335). Bedrock batch outputs land in a per-model
``s3_output_bucket_name`` when it differs from the input bucket. The
file-content retrieval path validates a file id against the buckets in the
trusted credential snapshot, and that snapshot is built by round-tripping
the deployment's ``litellm_params`` through ``CredentialLiteLLMParams``. If
the field is undeclared it is dropped, so the output bucket never reaches
retrieval and output-bucket file ids are rejected as foreign."""
file-content retrieval path validates a file id against the buckets the
proxy passes along, and those are built by round-tripping the deployment's
``litellm_params`` through ``CredentialLiteLLMParams``. If the field is
undeclared it is dropped, so the output bucket never reaches retrieval and
output-bucket file ids are rejected as foreign."""
def test_credentials_preserve_s3_output_bucket_name(self):
from litellm import Router
@ -185,6 +185,33 @@ class TestRouterCredentialResolutionS3OutputBucket:
)
assert credentials.get("s3_bucket_name") == "in-bucket"
def test_credentials_preserve_allow_legacy_cloud_file_ids(self):
"""An admin who sets allow_legacy_cloud_file_ids on a deployment expects
legacy (pre-managed-prefix) file ids in its bucket to stay readable
through the proxy's model-routed file endpoints."""
from litellm import Router
deployment_id = "bedrock-batch-legacy-ids-fixed-uuid"
router = Router(
model_list=[
{
"model_name": "bedrock-batch-legacy-ids",
"litellm_params": {
"model": "bedrock/anthropic.claude-3-sonnet-20240229-v1:0",
"s3_bucket_name": "in-bucket",
"allow_legacy_cloud_file_ids": True,
"aws_region_name": "us-west-2",
},
"model_info": {"id": deployment_id},
}
]
)
credentials = router.get_deployment_credentials_with_provider(model_id=deployment_id)
assert credentials is not None
assert credentials.get("allow_legacy_cloud_file_ids") is True
def test_credentials_without_output_bucket_unaffected(self):
"""A deployment that configures only the input bucket keeps it and does
not gain a phantom output bucket in the resolved credentials."""

14
ui/litellm-dashboard/src/lib/http/schema.d.ts generated vendored
View file

@ -30791,6 +30791,16 @@ export interface components {
* @description If True, lets keys address Responses API ids that this proxy did not issue (raw provider ids, or ids issued before response-id encryption was configured). Such an id carries no owner, so no ownership check can run on it; ids this proxy did issue keep full ownership enforcement. Off by default, in which case an unrecognized response id is rejected with 403
*/
allow_unmanaged_response_ids?: boolean | null;
/**
* Allowed Client Side S3 Buckets
* @description S3 bucket names a client may send as `s3_bucket_name` or `s3_output_bucket_name` in a request body, e.g. on /v1/batches. Any other bucket is rejected unless client-side credentials are allowed
*/
allowed_client_side_s3_buckets?: string[] | null;
/**
* Allowed Client Side S3 Endpoint Urls
* @description Exact `s3_endpoint_url` values a client may send in a request body. Any other value is rejected unless client-side credentials are allowed
*/
allowed_client_side_s3_endpoint_urls?: string[] | null;
/**
* Allowed File Extensions
* @description the only file extensions (e.g. ['.jsonl', '.pdf', '.txt']) accepted on /v1/files uploads, for any purpose, matched case-insensitively against the uploaded filename. Files with any other extension, or none, are rejected. An empty list rejects every upload. Unset means no allowlist is applied
@ -36286,6 +36296,8 @@ export interface components {
* @default false
*/
allow_client_keepalive_override: boolean | null;
/** Allow Legacy Cloud File Ids */
allow_legacy_cloud_file_ids?: boolean | null;
/** Annotation Cost Per Page */
annotation_cost_per_page?: number | null;
/** Annotation Cost Per Page Batches */
@ -52381,6 +52393,8 @@ export interface components {
* @default false
*/
allow_client_keepalive_override: boolean | null;
/** Allow Legacy Cloud File Ids */
allow_legacy_cloud_file_ids?: boolean | null;
/** Annotation Cost Per Page */
annotation_cost_per_page?: number | null;
/** Annotation Cost Per Page Batches */