fix(files): constrain cloud storage file paths (VERIA-45, VERIA-59)

litellm/files/main.py

@@ -10,6 +10,7 @@
 import time
 import uuid as uuid_module
 from functools import partial
+from types import MappingProxyType
 from typing import Any, Coroutine, Dict, Literal, Optional, Union, cast
 
 import httpx
@@ -85,6 +86,16 @@ def _should_sdk_support_streaming(
 #################################################
 
 
+def _add_trusted_model_credentials_to_litellm_params(
+    litellm_params_dict: Dict[str, Any], kwargs: Dict[str, Any]
+) -> None:
+    trusted_model_credentials = kwargs.get("_litellm_internal_model_credentials")
+    if isinstance(trusted_model_credentials, type(MappingProxyType({}))):
+        litellm_params_dict["_litellm_internal_model_credentials"] = (
+            trusted_model_credentials
+        )
+
+
 @client
 async def acreate_file(
     file: FileTypes,
@@ -373,6 +384,10 @@ def file_retrieve(
             )
             if provider_config is not None:
                 litellm_params_dict = 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
 
@@ -497,6 +512,10 @@ def file_delete(
             pass
         optional_params = GenericLiteLLMParams(**kwargs)
         litellm_params_dict = 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
@@ -846,6 +865,10 @@ def file_content(
     try:
         optional_params = GenericLiteLLMParams(**kwargs)
         litellm_params_dict = 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 = kwargs.get("client")
@@ -993,6 +1016,7 @@ def file_content(
                 vertex_location=vertex_ai_location,
                 timeout=timeout,
                 max_retries=optional_params.max_retries,
+                litellm_params=litellm_params_dict,
             )
         elif custom_llm_provider == "bedrock":
             response = bedrock_files_instance.file_content(

litellm/integrations/gcs_bucket/gcs_bucket.py

@@ -6,12 +6,14 @@
 from litellm._uuid import uuid
 from datetime import datetime, timedelta, timezone
 from typing import TYPE_CHECKING, Any, Dict, List, Optional, Tuple
-from urllib.parse import quote
 
 from litellm._logging import verbose_logger
 from litellm.constants import LITELLM_ASYNCIO_QUEUE_MAXSIZE
 from litellm.integrations.additional_logging_utils import AdditionalLoggingUtils
 from litellm.integrations.gcs_bucket.gcs_bucket_base import GCSBucketBase
+from litellm.litellm_core_utils.cloud_storage_security import (
+    sanitize_cloud_object_component,
+)
 from litellm.proxy._types import CommonProxyErrors
 from litellm.types.integrations.base_health_check import IntegrationHealthCheckStatus
 from litellm.types.integrations.gcs_bucket import *
@@ -335,7 +337,11 @@ def _get_object_name(
         _litellm_params = kwargs.get("litellm_params", None) or {}
         _metadata = _litellm_params.get("metadata", None) or {}
         if "gcs_log_id" in _metadata:
-            object_name = _metadata["gcs_log_id"]
+            safe_log_id = sanitize_cloud_object_component(
+                _metadata.get("gcs_log_id"), fallback=""
+            )
+            if safe_log_id:
+                object_name = f"{current_date}/custom-{uuid.uuid4().hex}-{safe_log_id}"
 
         return object_name
 
@@ -367,8 +373,7 @@ async def get_request_response_payload(
                     request_date_str=date_str,
                     response_id=request_id,
                 )
-                encoded_object_name = quote(object_name, safe="")
-                response = await self.download_gcs_object(encoded_object_name)
+                response = await self.download_gcs_object(object_name)
 
                 if response is not None:
                     loaded_response = json.loads(response)

litellm/integrations/gcs_bucket/gcs_bucket_base.py

@@ -11,6 +11,10 @@
 
 from litellm._logging import verbose_logger
 from litellm.integrations.custom_batch_logger import CustomBatchLogger
+from litellm.litellm_core_utils.cloud_storage_security import (
+    encode_gcs_object_name_for_url,
+    split_configured_cloud_bucket_name,
+)
 from litellm.llms.custom_httpx.http_handler import (
     get_async_httpx_client,
     httpxSpecialProvider,
@@ -133,8 +137,8 @@ def _handle_folders_in_bucket_name(
             - Returns: bucket_name="my-bucket", object_name="my-folder/dev/my-object"
 
         """
-        if "/" in bucket_name:
-            bucket_name, prefix = bucket_name.split("/", 1)
+        bucket_name, prefix = split_configured_cloud_bucket_name(bucket_name)
+        if prefix:
             object_name = f"{prefix}/{object_name}"
             return bucket_name, object_name
         return bucket_name, object_name
@@ -248,6 +252,7 @@ async def download_gcs_object(self, object_name: str, **kwargs):
                 bucket_name=bucket_name,
                 object_name=object_name,
             )
+            object_name = encode_gcs_object_name_for_url(object_name)
 
             url = f"https://storage.googleapis.com/storage/v1/b/{bucket_name}/o/{object_name}?alt=media"
 
@@ -288,6 +293,7 @@ async def delete_gcs_object(self, object_name: str, **kwargs):
                 bucket_name=bucket_name,
                 object_name=object_name,
             )
+            object_name = encode_gcs_object_name_for_url(object_name)
 
             url = f"https://storage.googleapis.com/storage/v1/b/{bucket_name}/o/{object_name}"
 
@@ -334,10 +340,11 @@ async def _log_json_data_on_gcs(
             bucket_name=bucket_name,
             object_name=object_name,
         )
+        encoded_object_name = encode_gcs_object_name_for_url(object_name)
 
         response = await self.async_httpx_client.post(
             headers=headers,
-            url=f"https://storage.googleapis.com/upload/storage/v1/b/{bucket_name}/o?uploadType=media&name={object_name}",
+            url=f"https://storage.googleapis.com/upload/storage/v1/b/{bucket_name}/o?uploadType=media&name={encoded_object_name}",
             data=json_logged_payload,
         )
 

litellm/litellm_core_utils/cloud_storage_security.py

@@ -0,0 +1,175 @@
+import posixpath
+import re
+from types import MappingProxyType
+from typing import Any, Mapping, Optional, Sequence, Tuple, cast
+from urllib.parse import quote, unquote
+
+from litellm._uuid import uuid
+
+VERTEX_AI_MANAGED_GCS_PREFIX = "litellm-vertex-files/"
+BEDROCK_MANAGED_S3_BATCH_PREFIX = "litellm-bedrock-files-"
+BEDROCK_MANAGED_S3_UPLOAD_PREFIX = "litellm-bedrock-files/"
+BEDROCK_MANAGED_S3_OUTPUT_PREFIX = "litellm-batch-outputs/"
+BEDROCK_MANAGED_S3_PREFIXES = (
+    BEDROCK_MANAGED_S3_BATCH_PREFIX,
+    BEDROCK_MANAGED_S3_UPLOAD_PREFIX,
+    BEDROCK_MANAGED_S3_OUTPUT_PREFIX,
+)
+_MAPPING_PROXY_TYPE: type = type(MappingProxyType({}))
+
+_SAFE_OBJECT_COMPONENT_PATTERN = re.compile(r"[^A-Za-z0-9._-]+")
+
+
+def sanitize_cloud_object_component(
+    value: Optional[str], fallback: str = "file"
+) -> str:
+    if not isinstance(value, str):
+        return fallback
+
+    component = posixpath.basename(value.replace("\\", "/")).strip()
+    if component in {"", ".", ".."}:
+        return fallback
+
+    component = "".join(
+        "_" if ord(char) < 32 or ord(char) == 127 else char for char in component
+    )
+    component = _SAFE_OBJECT_COMPONENT_PATTERN.sub("_", component)
+    component = component.strip("._")
+    if not component:
+        return fallback
+    return component[:255]
+
+
+def sanitize_cloud_object_path(value: Optional[str], fallback: str = "file") -> str:
+    if not isinstance(value, str):
+        return fallback
+
+    segments = []
+    for segment in value.replace("\\", "/").split("/"):
+        sanitized_segment = sanitize_cloud_object_component(segment, fallback="")
+        if sanitized_segment:
+            segments.append(sanitized_segment)
+
+    if not segments:
+        return fallback
+    return "/".join(segments)
+
+
+def build_managed_cloud_object_name(
+    prefix: str, filename: Optional[str], fallback_filename: str = "file"
+) -> str:
+    safe_filename = sanitize_cloud_object_component(
+        filename, fallback=fallback_filename
+    )
+    return f"{prefix}{uuid.uuid4().hex}-{safe_filename}"
+
+
+def _validate_cloud_object_path(object_name: str) -> None:
+    if not object_name:
+        raise ValueError("Cloud storage object name is required")
+    if object_name.startswith("/"):
+        raise ValueError("Cloud storage object name must be relative")
+    if any(ord(char) < 32 or ord(char) == 127 for char in object_name):
+        raise ValueError("Cloud storage object name contains control characters")
+    segments = object_name.split("/")
+    if any(segment in {".", ".."} for segment in segments):
+        raise ValueError("Cloud storage object name contains an invalid path segment")
+    if "" in segments[:-1]:
+        raise ValueError("Cloud storage object name contains an invalid path segment")
+
+
+def split_configured_cloud_bucket_name(bucket_name: str) -> Tuple[str, str]:
+    if not isinstance(bucket_name, str) or not bucket_name.strip():
+        raise ValueError("Cloud storage bucket name is required")
+
+    bucket_name = bucket_name.strip()
+    if "://" in bucket_name or "?" in bucket_name or "#" in bucket_name:
+        raise ValueError(
+            "Cloud storage bucket name must not include a URI scheme or query"
+        )
+    if any(ord(char) < 32 or ord(char) == 127 for char in bucket_name):
+        raise ValueError("Cloud storage bucket name contains control characters")
+
+    bucket, _, prefix = bucket_name.partition("/")
+    if not bucket:
+        raise ValueError("Cloud storage bucket name is required")
+    if "\\" in bucket:
+        raise ValueError("Cloud storage bucket name contains an invalid separator")
+
+    prefix = prefix.strip("/")
+    if prefix:
+        _validate_cloud_object_path(prefix)
+
+    return bucket, prefix
+
+
+def encode_gcs_object_name_for_url(object_name: str) -> str:
+    return quote(unquote(object_name), safe="")
+
+
+def encode_s3_object_key_for_url(object_key: str) -> str:
+    return quote(unquote(object_key), safe="/")
+
+
+def should_allow_legacy_cloud_file_ids(
+    litellm_params: Optional[Mapping[str, Any]] = None,
+) -> bool:
+    value = None
+    if isinstance(litellm_params, Mapping):
+        trusted_model_credentials = litellm_params.get(
+            "_litellm_internal_model_credentials"
+        )
+        if isinstance(trusted_model_credentials, _MAPPING_PROXY_TYPE):
+            value = cast(Mapping[str, Any], trusted_model_credentials).get(
+                "allow_legacy_cloud_file_ids"
+            )
+
+    if isinstance(value, bool):
+        return value
+    if isinstance(value, str):
+        return value.strip().lower() in {"1", "true", "yes", "on"}
+    return False
+
+
+def validate_managed_cloud_file_id(
+    file_id: str,
+    scheme: str,
+    configured_bucket_name: str,
+    allowed_object_prefixes: Sequence[str],
+    allow_legacy_cloud_file_ids: bool = False,
+) -> Tuple[str, str]:
+    decoded_file_id = unquote(file_id)
+    if not decoded_file_id.startswith(scheme):
+        raise ValueError(f"file_id must be a {scheme} URI")
+
+    full_path = decoded_file_id[len(scheme) :]
+    if "/" not in full_path:
+        raise ValueError("file_id must include a cloud storage object name")
+
+    bucket_name, object_name = full_path.split("/", 1)
+    configured_bucket, configured_prefix = split_configured_cloud_bucket_name(
+        configured_bucket_name
+    )
+    if bucket_name != configured_bucket:
+        raise ValueError("file_id bucket does not match the configured storage bucket")
+
+    _validate_cloud_object_path(object_name)
+    allowed_prefixes = tuple(allowed_object_prefixes)
+    if configured_prefix:
+        allowed_prefixes = tuple(
+            f"{configured_prefix.rstrip('/')}/{prefix}" for prefix in allowed_prefixes
+        )
+
+    if object_name.startswith(allowed_prefixes):
+        return bucket_name, object_name
+
+    if allow_legacy_cloud_file_ids:
+        if configured_prefix and not object_name.startswith(
+            f"{configured_prefix.rstrip('/')}/"
+        ):
+            raise ValueError(
+                "file_id object does not match the configured storage prefix"
+            )
+        return bucket_name, object_name
+
+    raise ValueError("file_id must reference a LiteLLM-managed storage object")

litellm/litellm_core_utils/initialize_dynamic_callback_params.py

@@ -37,8 +37,6 @@ def validate_no_callback_env_reference(
     "langfuse_secret_key",
     "langfuse_host",
     "langfuse_prompt_version",
-    "gcs_bucket_name",
-    "gcs_path_service_account",
     "langsmith_api_key",
     "langsmith_project",
     "langsmith_base_url",
@@ -57,20 +55,27 @@ def validate_no_callback_env_reference(
     "lunary_public_key",
 ]
 
+_request_blocked_callback_params = {
+    "gcs_bucket_name",
+    "gcs_path_service_account",
+}
+
 
 def initialize_standard_callback_dynamic_params(
     kwargs: Optional[Dict] = None,
 ) -> StandardCallbackDynamicParams:
     """
     Initialize the standard callback dynamic params from the kwargs
 
-    checks if langfuse_secret_key, gcs_bucket_name in kwargs and sets the corresponding attributes in StandardCallbackDynamicParams
+    checks supported request callback params in kwargs and sets the corresponding attributes in StandardCallbackDynamicParams
     """
 
     standard_callback_dynamic_params = StandardCallbackDynamicParams()
     if kwargs:
         # 1. Check top-level kwargs
         for param in _supported_callback_params:
+            if param in _request_blocked_callback_params:
+                continue
             if param in kwargs:
                 _param_value = kwargs.get(param)
                 validate_no_callback_env_reference(
@@ -86,6 +91,8 @@ def initialize_standard_callback_dynamic_params(
 
         if isinstance(metadata, dict):
             for param in _supported_callback_params:
+                if param in _request_blocked_callback_params:
+                    continue
                 if param not in standard_callback_dynamic_params and param in metadata:
                     _param_value = metadata.get(param)
                     validate_no_callback_env_reference(