scriptoriumadmin/litellm
1
0
Fork 0
mirror of https://github.com/BerriAI/litellm.git synced 2026-09-11 22:51:28 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 154

Merge pull request #27004 from stuxf/fix/managed-resource-service-account-isolation

Browse source 
fix(proxy): isolate managed resources for service-account API keys
This commit is contained in:
yuneng-jiang 2026-05-04 18:45:55 -07:00 • committed by GitHub
commit 2f305050ce
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
11 changed files with 717 additions and 159 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

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

@ -15,6 +15,11 @@ from litellm.caching.caching import DualCache
from litellm.integrations.custom_logger import CustomLogger
from litellm.litellm_core_utils.prompt_templates.common_utils import extract_file_data
from litellm.llms.base_llm.files.transformation import BaseFileEndpoints
from litellm.llms.base_llm.managed_resources.isolation import (
build_list_page,
build_owner_filter,
can_access_resource,
)
from litellm.proxy._types import (
CallTypes,
LiteLLM_ManagedFileTable,
@ -99,6 +104,7 @@ class _PROXY_LiteLLMManagedFiles(CustomLogger, BaseFileEndpoints):
model_mappings=model_mappings,
flat_model_file_ids=list(model_mappings.values()),
created_by=user_api_key_dict.user_id,
team_id=user_api_key_dict.team_id,
updated_by=user_api_key_dict.user_id,
)
await self.internal_usage_cache.async_set_cache(
@ -114,6 +120,7 @@ class _PROXY_LiteLLMManagedFiles(CustomLogger, BaseFileEndpoints):
"model_mappings": json.dumps(model_mappings),
"flat_model_file_ids": list(model_mappings.values()),
"created_by": user_api_key_dict.user_id,
"team_id": user_api_key_dict.team_id,
"updated_by": user_api_key_dict.user_id,
}
@ -125,7 +132,7 @@ class _PROXY_LiteLLMManagedFiles(CustomLogger, BaseFileEndpoints):
db_data["storage_backend"] = hidden_params["storage_backend"]
if "storage_url" in hidden_params:
db_data["storage_url"] = hidden_params["storage_url"]
verbose_logger.debug(
f"Storage metadata: storage_backend={db_data.get('storage_backend')}, "
f"storage_url={db_data.get('storage_url')}"
@ -171,6 +178,7 @@ class _PROXY_LiteLLMManagedFiles(CustomLogger, BaseFileEndpoints):
"model_object_id": model_object_id,
"file_purpose": file_purpose,
"created_by": user_api_key_dict.user_id,
"team_id": user_api_key_dict.team_id,
"updated_by": user_api_key_dict.user_id,
"status": file_object.status,
},
@ -229,15 +237,16 @@ class _PROXY_LiteLLMManagedFiles(CustomLogger, BaseFileEndpoints):
async def can_user_call_unified_file_id(
self, unified_file_id: str, user_api_key_dict: UserAPIKeyAuth
) -> bool:
## check if the user has access to the unified file id
user_id = user_api_key_dict.user_id
managed_file = await self.prisma_client.db.litellm_managedfiletable.find_first(
where={"unified_file_id": unified_file_id}
)
if managed_file:
return managed_file.created_by == user_id
return can_access_resource(
user_api_key_dict=user_api_key_dict,
created_by=managed_file.created_by,
resource_team_id=managed_file.team_id,
)
raise HTTPException(
status_code=404,
detail=f"File not found: {unified_file_id}",
@ -246,8 +255,6 @@ class _PROXY_LiteLLMManagedFiles(CustomLogger, BaseFileEndpoints):
async def can_user_call_unified_object_id(
self, unified_object_id: str, user_api_key_dict: UserAPIKeyAuth
) -> bool:
## check if the user has access to the unified object id
user_id = user_api_key_dict.user_id
managed_object = (
await self.prisma_client.db.litellm_managedobjecttable.find_first(
where={"unified_object_id": unified_object_id}
@ -255,7 +262,11 @@ class _PROXY_LiteLLMManagedFiles(CustomLogger, BaseFileEndpoints):
)
if managed_object:
return managed_object.created_by == user_id
return can_access_resource(
user_api_key_dict=user_api_key_dict,
created_by=managed_object.created_by,
resource_team_id=managed_object.team_id,
)
raise HTTPException(
status_code=404,
detail=f"Object not found: {unified_object_id}",
@ -285,28 +296,27 @@ class _PROXY_LiteLLMManagedFiles(CustomLogger, BaseFileEndpoints):
raise Exception(
"Filtering by 'target_model_names' is not supported when using managed batches."
)
where_clause: Dict[str, Any] = {"file_purpose": "batch"}
# Filter by user who created the batch
if user_api_key_dict.user_id:
where_clause["created_by"] = user_api_key_dict.user_id
owner_filter = build_owner_filter(user_api_key_dict)
if owner_filter is None:
return build_list_page([])
where_clause: Dict[str, Any] = {"file_purpose": "batch", **owner_filter}
if after:
where_clause["id"] = {"gt": after}
# Fetch more than needed to allow for post-fetch filtering
fetch_limit = limit or 20
if target_model_names:
# Fetch extra to account for filtering
# Oversample so post-fetch model-name filtering still has enough rows.
fetch_limit = max(fetch_limit * 3, 100)
batches = await self.prisma_client.db.litellm_managedobjecttable.find_many(
where=where_clause,
take=fetch_limit,
order={"created_at": "desc"},
)
batch_objects: List[LiteLLMBatch] = []
for batch in batches:
try:
@ -314,7 +324,11 @@ class _PROXY_LiteLLMManagedFiles(CustomLogger, BaseFileEndpoints):
if len(batch_objects) >= (limit or 20):
break
batch_data = json.loads(batch.file_object) if isinstance(batch.file_object, str) else batch.file_object
batch_data = (
json.loads(batch.file_object)
if isinstance(batch.file_object, str)
else batch.file_object
)
batch_obj = LiteLLMBatch(**batch_data)
batch_obj.id = batch.unified_object_id
batch_objects.append(batch_obj)
@ -324,27 +338,29 @@ class _PROXY_LiteLLMManagedFiles(CustomLogger, BaseFileEndpoints):
f"Failed to parse batch object {batch.unified_object_id}: {e}"
)
continue
return {
"object": "list",
"data": batch_objects,
"first_id": batch_objects[0].id if batch_objects else None,
"last_id": batch_objects[-1].id if batch_objects else None,
"has_more": len(batch_objects) == (limit or 20),
}
return build_list_page(
batch_objects, has_more=len(batch_objects) == (limit or 20)
)
async def get_user_created_file_ids(
self, user_api_key_dict: UserAPIKeyAuth, model_object_ids: List[str]
) -> List[OpenAIFileObject]:
"""
Get all file ids created by the user for a list of model object ids
Get all file ids the caller is allowed to see for a list of model
object ids. Service-account keys (no user_id) are scoped to their
team via ``team_id``; admins see all matches.
Returns:
- List of OpenAIFileObject's
"""
owner_filter = build_owner_filter(user_api_key_dict)
if owner_filter is None:
return []
file_ids = await self.prisma_client.db.litellm_managedfiletable.find_many(
where={
"created_by": user_api_key_dict.user_id,
**owner_filter,
"flat_model_file_ids": {"hasSome": model_object_ids},
}
)
@ -377,11 +393,11 @@ class _PROXY_LiteLLMManagedFiles(CustomLogger, BaseFileEndpoints):
"""
Check if the user has access to a list of file IDs.
Only checks managed (unified) file IDs.
Args:
file_ids: List of file IDs to check access for
user_api_key_dict: User API key authentication details
Raises:
HTTPException: If user doesn't have access to any of the files
"""
@ -419,10 +435,10 @@ class _PROXY_LiteLLMManagedFiles(CustomLogger, BaseFileEndpoints):
### HANDLE TRANSFORMATIONS ###
# Check both completion and acompletion call types
is_completion_call = (
call_type == CallTypes.completion.value
call_type == CallTypes.completion.value
or call_type == CallTypes.acompletion.value
)
if is_completion_call:
messages = data.get("messages")
model = data.get("model", "")
@ -431,22 +447,27 @@ class _PROXY_LiteLLMManagedFiles(CustomLogger, BaseFileEndpoints):
if file_ids:
# Check user has access to all managed files
await self.check_file_ids_access(file_ids, user_api_key_dict)
# Check if any files are stored in storage backends and need base64 conversion
# This is needed for Vertex AI/Gemini which requires base64 content
is_vertex_ai = model and ("vertex_ai" in model or "gemini" in model.lower())
is_vertex_ai = model and (
"vertex_ai" in model or "gemini" in model.lower()
)
if is_vertex_ai:
await self._convert_storage_files_to_base64(
messages=messages,
file_ids=file_ids,
litellm_parent_otel_span=user_api_key_dict.parent_otel_span,
)
model_file_id_mapping = await self.get_model_file_id_mapping(
file_ids, user_api_key_dict.parent_otel_span
)
data["model_file_id_mapping"] = model_file_id_mapping
elif call_type == CallTypes.aresponses.value or call_type == CallTypes.responses.value:
elif (
call_type == CallTypes.aresponses.value
or call_type == CallTypes.responses.value
):
# Handle managed files in responses API input and tools
file_ids = []
@ -611,7 +632,9 @@ class _PROXY_LiteLLMManagedFiles(CustomLogger, BaseFileEndpoints):
if model_id is None:
model_id = cast(
Optional[str],
kwargs.get("litellm_metadata", {}).get("model_info", {}).get("id", None),
kwargs.get("litellm_metadata", {})
.get("model_info", {})
.get("id", None),
)
mapped_file_id: Optional[str] = None
if input_file_id and model_file_id_mapping and model_id:
@ -648,7 +671,7 @@ class _PROXY_LiteLLMManagedFiles(CustomLogger, BaseFileEndpoints):
) -> List[str]:
"""
Gets file ids from responses API input.
The input can be:
- A string (no files)
- A list of input items, where each item can have:
@ -656,32 +679,35 @@ class _PROXY_LiteLLMManagedFiles(CustomLogger, BaseFileEndpoints):
- content: a list that can contain items with type: "input_file" and file_id
"""
file_ids: List[str] = []
if isinstance(input, str):
return file_ids
if not isinstance(input, list):
return file_ids
for item in input:
if not isinstance(item, dict):
continue
# Check for direct input_file type
if item.get("type") == "input_file":
file_id = item.get("file_id")
if file_id:
file_ids.append(file_id)
# Check for input_file in content array
content = item.get("content")
if isinstance(content, list):
for content_item in content:
if isinstance(content_item, dict) and content_item.get("type") == "input_file":
if (
isinstance(content_item, dict)
and content_item.get("type") == "input_file"
):
file_id = content_item.get("file_id")
if file_id:
file_ids.append(file_id)
return file_ids
def get_file_ids_from_responses_tools(
@ -689,7 +715,7 @@ class _PROXY_LiteLLMManagedFiles(CustomLogger, BaseFileEndpoints):
) -> List[str]:
"""
Gets file ids from responses API tools parameter.
The tools can contain code_interpreter with container.file_ids:
[
{
@ -699,14 +725,14 @@ class _PROXY_LiteLLMManagedFiles(CustomLogger, BaseFileEndpoints):
]
"""
file_ids: List[str] = []
if not isinstance(tools, list):
return file_ids
for tool in tools:
if not isinstance(tool, dict):
continue
# Check for code_interpreter with container file_ids
if tool.get("type") == "code_interpreter":
container = tool.get("container")
@ -716,7 +742,7 @@ class _PROXY_LiteLLMManagedFiles(CustomLogger, BaseFileEndpoints):
for file_id in container_file_ids:
if isinstance(file_id, str):
file_ids.append(file_id)
return file_ids
def get_vector_store_ids_from_file_search_tools(
@ -916,10 +942,17 @@ class _PROXY_LiteLLMManagedFiles(CustomLogger, BaseFileEndpoints):
# Emit Prometheus metrics for managed file creation
prom_logger = self._get_prometheus_logger()
if prom_logger:
first_model = target_model_names_list[0] if target_model_names_list else None
first_model = (
target_model_names_list[0] if target_model_names_list else None
)
first_provider = ""
if responses:
first_provider = getattr(responses[0], "_hidden_params", {}).get("custom_llm_provider") or ""
first_provider = (
getattr(responses[0], "_hidden_params", {}).get(
"custom_llm_provider"
)
or ""
)
prom_logger.record_managed_file_created(
model=first_model or "",
api_provider=first_provider,
@ -1073,16 +1106,24 @@ class _PROXY_LiteLLMManagedFiles(CustomLogger, BaseFileEndpoints):
model_name=resolved_model_name,
)
setattr(response, file_attr, unified_file_id)
# Use llm_router credentials when available. Without credentials,
# Azure and other auth-required providers return 500/401.
file_object = None
try:
# Import module and use getattr for better testability with mocks
import litellm.proxy.proxy_server as proxy_server_module
_llm_router = getattr(proxy_server_module, 'llm_router', None)
_llm_router = getattr(
proxy_server_module, "llm_router", None
)
if _llm_router is not None and model_id:
_creds = _llm_router.get_deployment_credentials_with_provider(model_id) or {}
_creds = (
_llm_router.get_deployment_credentials_with_provider(
model_id
)
or {}
)
file_object = await litellm.afile_retrieve(
file_id=original_file_id,
**_creds,
@ -1099,7 +1140,7 @@ class _PROXY_LiteLLMManagedFiles(CustomLogger, BaseFileEndpoints):
verbose_logger.warning(
f"Failed to retrieve file object for {file_attr}={original_file_id}: {str(e)}. Storing with None and will fetch on-demand."
)
await self.store_unified_file_id(
file_id=unified_file_id,
file_object=file_object,
@ -1128,6 +1169,7 @@ class _PROXY_LiteLLMManagedFiles(CustomLogger, BaseFileEndpoints):
from litellm.litellm_core_utils.get_llm_provider_logic import (
get_llm_provider,
)
_, batch_provider, _, _ = get_llm_provider(model=model_name)
except Exception:
if "/" in model_name:
@ -1199,7 +1241,7 @@ class _PROXY_LiteLLMManagedFiles(CustomLogger, BaseFileEndpoints):
# Case 1 : This is not a managed file
if not stored_file_object:
raise Exception(f"LiteLLM Managed File object with id={file_id} not found")
# Case 2: Managed file and the file object exists in the database
# The stored file_object has the raw provider ID. Replace with the unified ID
# so callers see a consistent ID (matching Case 3 which does response.id = file_id).
@ -1217,13 +1259,21 @@ class _PROXY_LiteLLMManagedFiles(CustomLogger, BaseFileEndpoints):
)
try:
model_id, model_file_id = next(iter(stored_file_object.model_mappings.items()))
credentials = llm_router.get_deployment_credentials_with_provider(model_id) or {}
response = await litellm.afile_retrieve(file_id=model_file_id, **credentials)
model_id, model_file_id = next(
iter(stored_file_object.model_mappings.items())
)
credentials = (
llm_router.get_deployment_credentials_with_provider(model_id) or {}
)
response = await litellm.afile_retrieve(
file_id=model_file_id, **credentials
)
response.id = file_id # Replace with unified ID
return response
except Exception as e:
raise Exception(f"Failed to retrieve file {file_id} from provider: {str(e)}") from e
raise Exception(
f"Failed to retrieve file {file_id} from provider: {str(e)}"
) from e
async def afile_list(
self,
@ -1245,19 +1295,19 @@ class _PROXY_LiteLLMManagedFiles(CustomLogger, BaseFileEndpoints):
import litellm.proxy.proxy_server as proxy_server_module
# Check if the scheduler has the batch cost checking job registered
scheduler = getattr(proxy_server_module, 'scheduler', None)
scheduler = getattr(proxy_server_module, "scheduler", None)
if scheduler is None:
return False
# Check if the check_batch_cost_job exists in the scheduler
try:
job = scheduler.get_job('check_batch_cost_job')
job = scheduler.get_job("check_batch_cost_job")
if job is not None:
return True
except Exception:
# Job not found or scheduler doesn't support get_job
pass
return False
except Exception as e:
verbose_logger.warning(
@ -1265,28 +1315,26 @@ class _PROXY_LiteLLMManagedFiles(CustomLogger, BaseFileEndpoints):
)
return False
async def _get_batches_referencing_file(
self, file_id: str
) -> List[Dict[str, Any]]:
async def _get_batches_referencing_file(self, file_id: str) -> List[Dict[str, Any]]:
"""
Find batches that reference this file and still need cost tracking.
Find batches that are in non-terminal state and have not yet been processed by CheckBatchCost.
Args:
file_id: The unified file ID to check
Returns:
List of batch objects referencing this file in non-terminal state
(max 10 for error message display)
"""
# Prepare list of file IDs to check (both unified and provider IDs)
file_ids_to_check = [file_id]
# Get model-specific file IDs for this unified file ID if it's a managed file
try:
model_file_id_mapping = await self.get_model_file_id_mapping(
[file_id], litellm_parent_otel_span=None
)
if model_file_id_mapping and file_id in model_file_id_mapping:
# Add all provider file IDs for this unified file
provider_file_ids = list(model_file_id_mapping[file_id].values())
@ -1296,59 +1344,67 @@ class _PROXY_LiteLLMManagedFiles(CustomLogger, BaseFileEndpoints):
f"Could not get model file ID mapping for {file_id}: {e}. "
f"Will only check unified file ID."
)
MAX_MATCHES_TO_RETURN = 10
MAX_MATCHES_TO_RETURN = 10
batches = await self.prisma_client.db.litellm_managedobjecttable.find_many(
where={
"file_purpose": "batch",
"batch_processed": False,
"status": {"not_in": ["failed", "expired", "cancelled"]}
"status": {"not_in": ["failed", "expired", "cancelled"]},
},
take=MAX_MATCHES_TO_RETURN,
order={"created_at": "desc"},
)
referencing_batches = []
for batch in batches:
try:
# Parse the batch file_object to check for file references
batch_data = json.loads(batch.file_object) if isinstance(batch.file_object, str) else batch.file_object
batch_data = (
json.loads(batch.file_object)
if isinstance(batch.file_object, str)
else batch.file_object
)
# Extract file IDs from batch
# Batches typically reference the unified file ID in input_file_id
# Output and error files are generated by the provider
input_file_id = batch_data.get("input_file_id")
output_file_id = batch_data.get("output_file_id")
error_file_id = batch_data.get("error_file_id")
referenced_file_ids = [fid for fid in [input_file_id, output_file_id, error_file_id] if fid]
referenced_file_ids = [
fid for fid in [input_file_id, output_file_id, error_file_id] if fid
]
# Check if any referenced file ID matches the file we're trying to delete
if any(ref_id in file_ids_to_check for ref_id in referenced_file_ids):
referencing_batches.append({
"batch_id": batch.unified_object_id,
"status": batch.status,
"created_at": batch.created_at,
})
referencing_batches.append(
{
"batch_id": batch.unified_object_id,
"status": batch.status,
"created_at": batch.created_at,
}
)
except Exception as e:
verbose_logger.warning(
f"Error parsing batch object {batch.unified_object_id}: {e}"
)
continue
return referencing_batches
async def _check_file_deletion_allowed(self, file_id: str) -> None:
"""
Check if file deletion should be blocked due to batch references.
Blocks deletion if:
1. File is referenced by any batch in non-terminal state, AND
2. Batch polling is configured (user wants cost tracking)
Args:
file_id: The unified file ID to check
Raises:
HTTPException: If file deletion should be blocked
"""
@ -1356,39 +1412,45 @@ class _PROXY_LiteLLMManagedFiles(CustomLogger, BaseFileEndpoints):
if not self._is_batch_polling_enabled():
# Batch polling not configured, allow deletion
return
# Check if file is referenced by any non-terminal batches
referencing_batches = await self._get_batches_referencing_file(file_id)
if referencing_batches:
# File is referenced by non-terminal batches and polling is enabled
MAX_BATCHES_IN_ERROR = 5 # Limit batches shown in error message for readability
MAX_BATCHES_IN_ERROR = (
5 # Limit batches shown in error message for readability
)
# Show up to MAX_BATCHES_IN_ERROR in the error message
batches_to_show = referencing_batches[:MAX_BATCHES_IN_ERROR]
batch_statuses = [f"{b['batch_id']}: {b['status']}" for b in batches_to_show]
batch_statuses = [
f"{b['batch_id']}: {b['status']}" for b in batches_to_show
]
# Determine the count message
count_message = f"{len(referencing_batches)}"
if len(referencing_batches) >= 10: # MAX_MATCHES_TO_RETURN from _get_batches_referencing_file
if (
len(referencing_batches) >= 10
): # MAX_MATCHES_TO_RETURN from _get_batches_referencing_file
count_message = "10+"
error_message = (
f"Cannot delete file {file_id}. "
f"The file is referenced by {count_message} batch(es) in non-terminal state"
)
# Add specific batch details if not too many
if len(referencing_batches) <= MAX_BATCHES_IN_ERROR:
error_message += f": {', '.join(batch_statuses)}. "
else:
error_message += f" (showing {MAX_BATCHES_IN_ERROR} most recent): {', '.join(batch_statuses)}. "
error_message += (
f"To delete this file before complete cost tracking, please delete or cancel the referencing batch(es) first. "
f"Alternatively, wait for all batches to complete and for cost to be computed (batch_processed=true)."
)
# Record blocked deletion metric
prom_logger = self._get_prometheus_logger()
if prom_logger:
@ -1419,7 +1481,9 @@ class _PROXY_LiteLLMManagedFiles(CustomLogger, BaseFileEndpoints):
specific_model_file_id_mapping = model_file_id_mapping.get(file_id)
if specific_model_file_id_mapping:
# Remove conflicting keys from data to avoid duplicate keyword arguments
filtered_data = {k: v for k, v in data.items() if k not in ("model", "file_id")}
filtered_data = {
k: v for k, v in data.items() if k not in ("model", "file_id")
}
for model_id, model_file_id in specific_model_file_id_mapping.items():
delete_response = await llm_router.afile_delete(model=model_id, file_id=model_file_id, **filtered_data) # type: ignore
@ -1480,7 +1544,7 @@ class _PROXY_LiteLLMManagedFiles(CustomLogger, BaseFileEndpoints):
) -> None:
"""
Convert files stored in storage backends to base64 format for Vertex AI/Gemini.
This method checks if any managed files are stored in storage backends,
downloads them, and converts them to base64 format in the messages.
"""
@ -1488,29 +1552,29 @@ class _PROXY_LiteLLMManagedFiles(CustomLogger, BaseFileEndpoints):
for file_id in file_ids:
# Check if this is a base64 encoded unified file ID
decoded_unified_file_id = _is_base64_encoded_unified_file_id(file_id)
if not decoded_unified_file_id:
continue
# Check database for storage backend info
# IMPORTANT: The database stores the base64 encoded unified_file_id (not the decoded version)
# So we query with the original file_id (which is base64 encoded)
db_file = await self.prisma_client.db.litellm_managedfiletable.find_first(
where={"unified_file_id": file_id}
)
if not db_file or not db_file.storage_backend or not db_file.storage_url:
continue
# File is stored in a storage backend, download and convert to base64
try:
from litellm.llms.base_llm.files.storage_backend_factory import (
get_storage_backend,
)
storage_backend_name = db_file.storage_backend
storage_url = db_file.storage_url
# Get storage backend (uses same env vars as callback)
try:
storage_backend = get_storage_backend(storage_backend_name)
@ -1519,18 +1583,22 @@ class _PROXY_LiteLLMManagedFiles(CustomLogger, BaseFileEndpoints):
f"Storage backend '{storage_backend_name}' error for file {file_id}: {str(e)}"
)
continue
file_content = await storage_backend.download_file(storage_url)
# Determine content type from file object
content_type = self._get_content_type_from_file_object(db_file.file_object)
content_type = self._get_content_type_from_file_object(
db_file.file_object
)
# Convert to base64
base64_data = base64.b64encode(file_content).decode("utf-8")
base64_data_uri = f"data:{content_type};base64,{base64_data}"
# Update messages to use base64 instead of file_id
self._update_messages_with_base64_data(messages, file_id, base64_data_uri, content_type)
self._update_messages_with_base64_data(
messages, file_id, base64_data_uri, content_type
)
except Exception as e:
verbose_logger.exception(
f"Error converting file {file_id} from storage backend to base64: {str(e)}"
@ -1541,21 +1609,21 @@ class _PROXY_LiteLLMManagedFiles(CustomLogger, BaseFileEndpoints):
def _get_content_type_from_file_object(self, file_object: Optional[Any]) -> str:
"""
Determine content type from file object.
Uses the MIME type utility for consistent detection and normalization.
Args:
file_object: The file object from the database (can be dict, JSON string, or None)
Returns:
str: MIME type (defaults to "application/octet-stream" if cannot be determined)
"""
# Use utility function for detection
content_type = get_content_type_from_file_object(file_object)
# Normalize for Gemini/Vertex AI (requires image/jpeg, not image/jpg)
content_type = normalize_mime_type_for_provider(content_type, provider="gemini")
return content_type
def _update_messages_with_base64_data(
@ -1567,7 +1635,7 @@ class _PROXY_LiteLLMManagedFiles(CustomLogger, BaseFileEndpoints):
) -> None:
"""
Update messages to replace file_id with base64 data URI.
Args:
messages: List of messages to update
file_id: The file ID to replace
@ -1582,7 +1650,7 @@ class _PROXY_LiteLLMManagedFiles(CustomLogger, BaseFileEndpoints):
if element.get("type") == "file":
file_element = cast(ChatCompletionFileObject, element)
file_element_file = file_element.get("file", {})
if file_element_file.get("file_id") == file_id:
# Replace file_id with base64 data
file_element_file["file_data"] = base64_data_uri
@ -1590,7 +1658,7 @@ class _PROXY_LiteLLMManagedFiles(CustomLogger, BaseFileEndpoints):
file_element_file["format"] = content_type
# Remove file_id to ensure only file_data is used
file_element_file.pop("file_id", None)
verbose_logger.debug(
f"Converted file {file_id} from storage backend to base64 with format {content_type}"
)

20
litellm-proxy-extras/litellm_proxy_extras/migrations/20260501195714_managed_resource_team_owner/migration.sql Normal file
View file

@ -0,0 +1,20 @@
-- Adds `team_id` to managed-resource tables so service-account API
-- keys (no `user_id`) can be scoped by team instead of bypassing the
-- `created_by` filter entirely. Existing rows keep `team_id = NULL`
-- and become invisible to team-only callers — that is the intended isolation
-- outcome; backfill manually if legacy rows must remain visible.
--
-- The composite indexes match the listing query: filter by team owner, sort by
-- created_at DESC. Tables are typically small (resources per tenant, not per
-- request); a future operator with a large table can switch to
-- CREATE INDEX CONCURRENTLY in a follow-up migration.
ALTER TABLE "LiteLLM_ManagedFileTable" ADD COLUMN IF NOT EXISTS "team_id" TEXT;
ALTER TABLE "LiteLLM_ManagedObjectTable" ADD COLUMN IF NOT EXISTS "team_id" TEXT;
ALTER TABLE "LiteLLM_ManagedVectorStoreTable" ADD COLUMN IF NOT EXISTS "team_id" TEXT;
-- Index names follow Prisma's auto-generated convention so `prisma migrate diff`
-- against the schema is clean.
CREATE INDEX IF NOT EXISTS "LiteLLM_ManagedFileTable_team_id_created_at_idx" ON "LiteLLM_ManagedFileTable" ("team_id", "created_at" DESC);
CREATE INDEX IF NOT EXISTS "LiteLLM_ManagedObjectTable_team_id_created_at_idx" ON "LiteLLM_ManagedObjectTable" ("team_id", "created_at" DESC);
CREATE INDEX IF NOT EXISTS "LiteLLM_ManagedVectorStoreTable_team_id_created_at_idx" ON "LiteLLM_ManagedVectorStoreTable" ("team_id", "created_at" DESC);

16
litellm-proxy-extras/litellm_proxy_extras/schema.prisma
View file

@ -884,28 +884,32 @@ model LiteLLM_ManagedFileTable {
storage_backend String? // Storage backend name (e.g., "azure_storage", "gcs", "default")
storage_url String? // The actual storage URL where the file is stored
created_at DateTime @default(now())
created_by String?
created_by String?
team_id String? // Team that owns the resource; populated for service-account keys without a user_id so listings can isolate by team.
updated_at DateTime @updatedAt
updated_by String?
@@index([unified_file_id])
@@index([team_id, created_at(sort: Desc)])
}
model LiteLLM_ManagedObjectTable { // for batches or finetuning jobs which use the
model LiteLLM_ManagedObjectTable { // for batches or finetuning jobs which use the
id String @id @default(uuid())
unified_object_id String @unique // The base64 encoded unified file ID
model_object_id String @unique // the id returned by the backend API provider
model_object_id String @unique // the id returned by the backend API provider
file_object Json // Stores the OpenAIFileObject
file_purpose String // either 'batch' or 'fine-tune'
status String? // check if batch cost has been tracked
status String? // check if batch cost has been tracked
batch_processed Boolean @default(false) // set to true by CheckBatchCost after cost is computed
created_at DateTime @default(now())
created_by String?
team_id String?
updated_at DateTime @updatedAt
updated_by String?
updated_by String?
@@index([unified_object_id])
@@index([model_object_id])
@@index([team_id, created_at(sort: Desc)])
}
model LiteLLM_ManagedVectorStoreTable {
@ -918,10 +922,12 @@ model LiteLLM_ManagedVectorStoreTable {
storage_url String? // Storage URL (if applicable)
created_at DateTime @default(now())
created_by String?
team_id String?
updated_at DateTime @updatedAt
updated_by String?
@@index([unified_resource_id])
@@index([team_id, created_at(sort: Desc)])
}
model LiteLLM_ManagedVectorStoresTable {

33
litellm/llms/base_llm/managed_resources/base_managed_resource.py
View file

@ -18,6 +18,11 @@ from typing import (
)
from litellm import verbose_logger
from litellm.llms.base_llm.managed_resources.isolation import (
build_list_page,
build_owner_filter,
can_access_resource,
)
from litellm.proxy._types import UserAPIKeyAuth
from litellm.types.utils import SpecialEnums
@ -169,6 +174,7 @@ class BaseManagedResource(ABC, Generic[ResourceObjectType]):
"model_mappings": model_mappings,
"flat_model_resource_ids": list(model_mappings.values()),
"created_by": user_api_key_dict.user_id,
"team_id": user_api_key_dict.team_id,
"updated_by": user_api_key_dict.user_id,
}
@ -190,6 +196,7 @@ class BaseManagedResource(ABC, Generic[ResourceObjectType]):
"model_mappings": json.dumps(model_mappings),
"flat_model_resource_ids": list(model_mappings.values()),
"created_by": user_api_key_dict.user_id,
"team_id": user_api_key_dict.team_id,
"updated_by": user_api_key_dict.user_id,
}
@ -316,15 +323,17 @@ class BaseManagedResource(ABC, Generic[ResourceObjectType]):
Returns:
True if user has access, False otherwise
"""
user_id = user_api_key_dict.user_id
# Use cached method instead of direct DB query
resource = await self.get_unified_resource_id(
unified_resource_id, litellm_parent_otel_span
)
if resource:
return resource.get("created_by") == user_id
return can_access_resource(
user_api_key_dict=user_api_key_dict,
created_by=resource.get("created_by"),
resource_team_id=resource.get("team_id"),
)
return False
@ -549,11 +558,11 @@ class BaseManagedResource(ABC, Generic[ResourceObjectType]):
Returns:
Dictionary with list of resources and pagination info
"""
where_clause: Dict[str, Any] = {}
owner_filter = build_owner_filter(user_api_key_dict)
if owner_filter is None:
return build_list_page([])
# Filter by user who created the resource
if user_api_key_dict.user_id:
where_clause["created_by"] = user_api_key_dict.user_id
where_clause: Dict[str, Any] = {**owner_filter}
if after:
where_clause["id"] = {"gt": after}
@ -598,10 +607,6 @@ class BaseManagedResource(ABC, Generic[ResourceObjectType]):
)
continue
return {
"object": "list",
"data": resource_objects,
"first_id": resource_objects[0].id if resource_objects else None,
"last_id": resource_objects[-1].id if resource_objects else None,
"has_more": len(resource_objects) == (limit or 20),
}
return build_list_page(
resource_objects, has_more=len(resource_objects) == (limit or 20)
)

97
litellm/llms/base_llm/managed_resources/isolation.py Normal file
View file

@ -0,0 +1,97 @@
"""
Tenant-isolation helpers for managed file/batch/vector-store resources.
Returns a Prisma filter and an ownership check that scope managed resources
to the caller's identity: proxy admins see everything, user-keyed callers
see records they created, and service-account keys (no user_id) fall back
to the resource's owning team. Callers with no admin role and no
identifying ids are denied so an empty user_id can never select an
unscoped query.
"""
from typing import Any, Dict, List, Optional
from litellm.proxy._types import UserAPIKeyAuth
from litellm.proxy.management_endpoints.common_utils import _user_has_admin_view
def build_list_page(items: List[Any], has_more: bool = False) -> Dict[str, Any]:
"""Build the OpenAI-style paginated list response shape used by managed
file/batch/vector-store listings. ``first_id`` and ``last_id`` are
sourced from each item's ``.id`` attribute."""
return {
"object": "list",
"data": items,
"first_id": items[0].id if items else None,
"last_id": items[-1].id if items else None,
"has_more": has_more,
}
def build_owner_filter(
user_api_key_dict: UserAPIKeyAuth,
) -> Optional[Dict[str, Any]]:
"""Return a Prisma `where` fragment that scopes a managed-resource listing
to records the caller is allowed to see.
- ``{}`` means no scoping (proxy admins).
- ``{"created_by": <user_id>}`` for user-keyed callers.
- ``{"team_id": <team_id>}`` for service-account callers
that have a team but no user_id.
- ``{"OR": [...]}`` when the caller has both — listing must include
both their own resources and team-shared ones so it stays consistent
with ``can_access_resource``.
- ``None`` means deny: callers MUST skip the query rather than fall
back to an unscoped fetch.
"""
if _user_has_admin_view(user_api_key_dict):
return {}
user_id = user_api_key_dict.user_id
team_id = user_api_key_dict.team_id
if user_id is not None and team_id is not None:
return {
"OR": [
{"created_by": user_id},
{"team_id": team_id},
]
}
if user_id is not None:
return {"created_by": user_id}
if team_id is not None:
return {"team_id": team_id}
return None
def can_access_resource(
user_api_key_dict: UserAPIKeyAuth,
created_by: Optional[str],
resource_team_id: Optional[str],
) -> bool:
"""Return True iff the caller may read/modify a managed resource.
The resource's ``created_by`` and ``team_id`` fields must be non-None
to match the caller's identity — guarding against the ``None == None``
bypass that previously let service-account keys read every keyless
resource.
"""
if _user_has_admin_view(user_api_key_dict):
return True
user_id = user_api_key_dict.user_id
if user_id is not None and created_by is not None and created_by == user_id:
return True
team_id = user_api_key_dict.team_id
if (
team_id is not None
and resource_team_id is not None
and resource_team_id == team_id
):
return True
return False

4
litellm/proxy/_types.py
View file

@ -4603,6 +4603,7 @@ class LiteLLM_ManagedFileTable(LiteLLMPydanticObjectBase):
model_mappings: Dict[str, str]
flat_model_file_ids: List[str]
created_by: Optional[str] = None
team_id: Optional[str] = None
updated_by: Optional[str] = None
storage_backend: Optional[str] = None
storage_url: Optional[str] = None
@ -4613,6 +4614,8 @@ class LiteLLM_ManagedObjectTable(LiteLLMPydanticObjectBase):
model_object_id: str
file_purpose: Literal["batch", "fine-tune", "response"]
file_object: Union[LiteLLMBatch, LiteLLMFineTuningJob, ResponsesAPIResponse]
created_by: Optional[str] = None
team_id: Optional[str] = None
class LiteLLM_ManagedVectorStoreTable(LiteLLMPydanticObjectBase):
@ -4623,6 +4626,7 @@ class LiteLLM_ManagedVectorStoreTable(LiteLLMPydanticObjectBase):
model_mappings: Dict[str, str]
flat_model_resource_ids: List[str]
created_by: Optional[str] = None
team_id: Optional[str] = None
updated_by: Optional[str] = None
storage_backend: Optional[str] = None
storage_url: Optional[str] = None

16
litellm/proxy/schema.prisma
View file

@ -884,28 +884,32 @@ model LiteLLM_ManagedFileTable {
storage_backend String? // Storage backend name (e.g., "azure_storage", "gcs", "default")
storage_url String? // The actual storage URL where the file is stored
created_at DateTime @default(now())
created_by String?
created_by String?
team_id String? // Team that owns the resource; populated for service-account keys without a user_id so listings can isolate by team.
updated_at DateTime @updatedAt
updated_by String?
@@index([unified_file_id])
@@index([team_id, created_at(sort: Desc)])
}
model LiteLLM_ManagedObjectTable { // for batches or finetuning jobs which use the
model LiteLLM_ManagedObjectTable { // for batches or finetuning jobs which use the
id String @id @default(uuid())
unified_object_id String @unique // The base64 encoded unified file ID
model_object_id String @unique // the id returned by the backend API provider
model_object_id String @unique // the id returned by the backend API provider
file_object Json // Stores the OpenAIFileObject
file_purpose String // either 'batch' or 'fine-tune'
status String? // check if batch cost has been tracked
status String? // check if batch cost has been tracked
batch_processed Boolean @default(false) // set to true by CheckBatchCost after cost is computed
created_at DateTime @default(now())
created_by String?
team_id String?
updated_at DateTime @updatedAt
updated_by String?
updated_by String?
@@index([unified_object_id])
@@index([model_object_id])
@@index([team_id, created_at(sort: Desc)])
}
model LiteLLM_ManagedVectorStoreTable {
@ -918,10 +922,12 @@ model LiteLLM_ManagedVectorStoreTable {
storage_url String? // Storage URL (if applicable)
created_at DateTime @default(now())
created_by String?
team_id String?
updated_at DateTime @updatedAt
updated_by String?
@@index([unified_resource_id])
@@index([team_id, created_at(sort: Desc)])
}
model LiteLLM_ManagedVectorStoresTable {

16
schema.prisma
View file

@ -884,28 +884,32 @@ model LiteLLM_ManagedFileTable {
storage_backend String? // Storage backend name (e.g., "azure_storage", "gcs", "default")
storage_url String? // The actual storage URL where the file is stored
created_at DateTime @default(now())
created_by String?
created_by String?
team_id String? // Team that owns the resource; populated for service-account keys without a user_id so listings can isolate by team.
updated_at DateTime @updatedAt
updated_by String?
@@index([unified_file_id])
@@index([team_id, created_at(sort: Desc)])
}
model LiteLLM_ManagedObjectTable { // for batches or finetuning jobs which use the
model LiteLLM_ManagedObjectTable { // for batches or finetuning jobs which use the
id String @id @default(uuid())
unified_object_id String @unique // The base64 encoded unified file ID
model_object_id String @unique // the id returned by the backend API provider
model_object_id String @unique // the id returned by the backend API provider
file_object Json // Stores the OpenAIFileObject
file_purpose String // either 'batch' or 'fine-tune'
status String? // check if batch cost has been tracked
status String? // check if batch cost has been tracked
batch_processed Boolean @default(false) // set to true by CheckBatchCost after cost is computed
created_at DateTime @default(now())
created_by String?
team_id String?
updated_at DateTime @updatedAt
updated_by String?
updated_by String?
@@index([unified_object_id])
@@index([model_object_id])
@@index([team_id, created_at(sort: Desc)])
}
model LiteLLM_ManagedVectorStoreTable {
@ -918,10 +922,12 @@ model LiteLLM_ManagedVectorStoreTable {
storage_url String? // Storage URL (if applicable)
created_at DateTime @default(now())
created_by String?
team_id String?
updated_at DateTime @updatedAt
updated_by String?
@@index([unified_resource_id])
@@index([team_id, created_at(sort: Desc)])
}
model LiteLLM_ManagedVectorStoresTable {

69
tests/test_litellm/enterprise/proxy/test_managed_files_access_check.py
View file

@ -31,7 +31,11 @@ def _make_unified_file_id() -> str:
return base64.b64encode(raw.encode()).decode()
def _make_managed_files_instance(file_created_by: str, unified_file_id: str):
def _make_managed_files_instance(
file_created_by: str,
unified_file_id: str,
file_team_id=None,
):
"""Create a _PROXY_LiteLLMManagedFiles with a mocked DB that returns a file owned by file_created_by."""
from litellm_enterprise.proxy.hooks.managed_files import (
_PROXY_LiteLLMManagedFiles,
@ -39,6 +43,7 @@ def _make_managed_files_instance(file_created_by: str, unified_file_id: str):
mock_db_record = MagicMock()
mock_db_record.created_by = file_created_by
mock_db_record.team_id = file_team_id
mock_prisma = MagicMock()
mock_prisma.db.litellm_managedfiletable.find_first = AsyncMock(
@ -105,6 +110,57 @@ async def test_should_block_default_user_id_access():
assert exc_info.value.status_code == 403
# --- Service-account isolation: created_by/team_id checks ---
@pytest.mark.asyncio
async def test_keyless_caller_cannot_access_keyless_file():
"""A file created by a key without a user_id used to be accessible by
any other keyless caller because `None == None` was True."""
unified_file_id = _make_unified_file_id()
managed_files = _make_managed_files_instance(
file_created_by=None,
file_team_id=None,
unified_file_id=unified_file_id,
)
keyless = UserAPIKeyAuth(api_key="sk-test", parent_otel_span=None)
data = {"file_id": unified_file_id}
with pytest.raises(HTTPException) as exc_info:
await managed_files.check_managed_file_id_access(data, keyless)
assert exc_info.value.status_code == 403
@pytest.mark.asyncio
async def test_service_account_can_access_team_file():
unified_file_id = _make_unified_file_id()
managed_files = _make_managed_files_instance(
file_created_by=None,
file_team_id="team-eng",
unified_file_id=unified_file_id,
)
sa = UserAPIKeyAuth(api_key="sk-svc", team_id="team-eng", parent_otel_span=None)
data = {"file_id": unified_file_id}
assert await managed_files.check_managed_file_id_access(data, sa) is True
@pytest.mark.asyncio
async def test_service_account_blocked_from_other_team_file():
unified_file_id = _make_unified_file_id()
managed_files = _make_managed_files_instance(
file_created_by=None,
file_team_id="team-sales",
unified_file_id=unified_file_id,
)
sa = UserAPIKeyAuth(api_key="sk-svc", team_id="team-eng", parent_otel_span=None)
data = {"file_id": unified_file_id}
with pytest.raises(HTTPException) as exc_info:
await managed_files.check_managed_file_id_access(data, sa)
assert exc_info.value.status_code == 403
# --- Option C fix test: check_batch_cost bypasses managed files hook ---
@ -144,6 +200,7 @@ async def test_check_batch_cost_should_call_afile_content_directly_with_credenti
# Mock the batch response (completed, with output file)
from litellm.types.utils import LiteLLMBatch
batch_response = LiteLLMBatch(
id="batch-123",
completion_window="24h",
@ -201,9 +258,11 @@ async def test_check_batch_cost_should_call_afile_content_directly_with_credenti
# Verify the DB update writes batch_processed, status, and file_object
mock_prisma.db.litellm_managedobjecttable.update.assert_called_once()
update_call_kwargs = mock_prisma.db.litellm_managedobjecttable.update.call_args.kwargs
update_call_kwargs = (
mock_prisma.db.litellm_managedobjecttable.update.call_args.kwargs
)
assert update_call_kwargs["data"]["batch_processed"] is True
assert update_call_kwargs["data"]["status"] == "complete"
assert "file_object" in update_call_kwargs["data"], (
"file_object must be written to DB so list_batches reads updated status"
)
assert (
"file_object" in update_call_kwargs["data"]
), "file_object must be written to DB so list_batches reads updated status"

131
tests/test_litellm/llms/base_llm/test_base_managed_resource.py Normal file
View file

@ -0,0 +1,131 @@
"""
Integration tests for BaseManagedResource listing and access control.
"""
from typing import List
from unittest.mock import AsyncMock, MagicMock
import pytest
from litellm.llms.base_llm.managed_resources.base_managed_resource import (
BaseManagedResource,
)
from litellm.proxy._types import LitellmUserRoles, UserAPIKeyAuth
class _StubResource(BaseManagedResource):
"""Concrete subclass exposing the abstract surface for testing."""
@property
def resource_type(self) -> str:
return "test_resource"
@property
def table_name(self) -> str:
return "litellm_test_resource_table"
def get_unified_resource_id_format(
self, resource_object, target_model_names_list
) -> str:
return "test"
async def create_resource_for_model(
self, llm_router, model, request_data, litellm_parent_otel_span
):
return {"id": "test"}
def _make_resource(records: List = None) -> _StubResource:
cache = MagicMock()
cache.async_get_cache = AsyncMock(return_value=None)
prisma = MagicMock()
table = MagicMock()
table.find_many = AsyncMock(return_value=records or [])
prisma.db = MagicMock()
setattr(prisma.db, "litellm_test_resource_table", table)
return _StubResource(internal_usage_cache=cache, prisma_client=prisma)
@pytest.mark.asyncio
async def test_list_admin_query_is_unscoped():
resource = _make_resource()
admin = UserAPIKeyAuth(user_role=LitellmUserRoles.PROXY_ADMIN)
await resource.list_user_resources(user_api_key_dict=admin)
table = resource.prisma_client.db.litellm_test_resource_table
where = table.find_many.await_args.kwargs["where"]
assert "created_by" not in where
assert "team_id" not in where
@pytest.mark.asyncio
async def test_list_user_filters_by_user_id():
resource = _make_resource()
user = UserAPIKeyAuth(user_id="alice")
await resource.list_user_resources(user_api_key_dict=user)
where = resource.prisma_client.db.litellm_test_resource_table.find_many.await_args.kwargs[
"where"
]
assert where["created_by"] == "alice"
assert "team_id" not in where
@pytest.mark.asyncio
async def test_list_service_account_filters_by_team_id():
resource = _make_resource()
service_account = UserAPIKeyAuth(team_id="team-eng")
await resource.list_user_resources(user_api_key_dict=service_account)
where = resource.prisma_client.db.litellm_test_resource_table.find_many.await_args.kwargs[
"where"
]
assert where["team_id"] == "team-eng"
assert "created_by" not in where
@pytest.mark.asyncio
async def test_list_identity_less_caller_returns_empty_without_query():
"""A caller with no admin role and no identifying ids must NOT issue a
query — the original bug skipped the filter and returned everything."""
resource = _make_resource()
nobody = UserAPIKeyAuth()
result = await resource.list_user_resources(user_api_key_dict=nobody)
assert result == {
"object": "list",
"data": [],
"first_id": None,
"last_id": None,
"has_more": False,
}
resource.prisma_client.db.litellm_test_resource_table.find_many.assert_not_awaited()
@pytest.mark.asyncio
@pytest.mark.parametrize(
"caller_team_id,expected",
[("team-eng", True), ("team-sales", False), (None, False)],
)
async def test_can_access_uses_team_id_for_service_account(caller_team_id, expected):
cache = MagicMock()
cache.async_get_cache = AsyncMock(
return_value={
"created_by": None,
"team_id": "team-eng",
}
)
prisma = MagicMock()
resource = _StubResource(internal_usage_cache=cache, prisma_client=prisma)
caller = (
UserAPIKeyAuth(team_id=caller_team_id) if caller_team_id else UserAPIKeyAuth()
)
assert await resource.can_user_access_unified_resource_id("rid", caller) is expected

156
tests/test_litellm/llms/base_llm/test_managed_resource_isolation.py Normal file
View file

@ -0,0 +1,156 @@
"""
Tests for managed-resource tenant isolation helpers.
"""
import pytest
from litellm.llms.base_llm.managed_resources.isolation import (
build_owner_filter,
can_access_resource,
)
from litellm.proxy._types import LitellmUserRoles, UserAPIKeyAuth
# ---------------------------------------------------------------------------
# build_owner_filter
# ---------------------------------------------------------------------------
@pytest.mark.parametrize(
"role",
[LitellmUserRoles.PROXY_ADMIN, LitellmUserRoles.PROXY_ADMIN_VIEW_ONLY],
)
def test_owner_filter_admin_unscoped(role):
assert build_owner_filter(UserAPIKeyAuth(user_role=role)) == {}
def test_owner_filter_user_scoped_to_user_id():
user = UserAPIKeyAuth(user_id="alice")
assert build_owner_filter(user) == {"created_by": "alice"}
def test_owner_filter_service_account_scoped_to_team():
service_account = UserAPIKeyAuth(team_id="team-eng")
assert build_owner_filter(service_account) == {"team_id": "team-eng"}
def test_owner_filter_user_with_team_returns_or_filter():
"""List view must mirror `can_access_resource`: a user-keyed caller in a
team can also access team-shared resources, so the listing returns both
their own records and team records via an OR filter."""
user = UserAPIKeyAuth(user_id="alice", team_id="team-eng")
assert build_owner_filter(user) == {
"OR": [
{"created_by": "alice"},
{"team_id": "team-eng"},
]
}
def test_owner_filter_no_identity_returns_none():
"""A caller with no admin role and no identifying ids must be denied so
the listing path can refuse the query rather than fall through to an
unscoped fetch."""
assert build_owner_filter(UserAPIKeyAuth()) is None
# ---------------------------------------------------------------------------
# can_access_resource
# ---------------------------------------------------------------------------
@pytest.mark.parametrize(
"role",
[LitellmUserRoles.PROXY_ADMIN, LitellmUserRoles.PROXY_ADMIN_VIEW_ONLY],
)
@pytest.mark.parametrize(
"created_by,resource_team_id",
[("alice", "team-eng"), (None, None)],
)
def test_access_admin_can_read_any_resource(role, created_by, resource_team_id):
admin = UserAPIKeyAuth(user_role=role)
assert (
can_access_resource(
admin, created_by=created_by, resource_team_id=resource_team_id
)
is True
)
@pytest.mark.parametrize(
"user_id,created_by,expected",
[
("alice", "alice", True),
("alice", "bob", False),
("alice", None, False),
],
)
def test_access_user_id_match(user_id, created_by, expected):
user = UserAPIKeyAuth(user_id=user_id)
assert (
can_access_resource(user, created_by=created_by, resource_team_id=None)
is expected
)
@pytest.mark.parametrize(
"caller_team_id,resource_team_id,expected",
[
("team-eng", "team-eng", True),
("team-eng", "team-sales", False),
("team-eng", None, False),
],
)
def test_access_service_account_team_id_match(
caller_team_id, resource_team_id, expected
):
service_account = UserAPIKeyAuth(team_id=caller_team_id)
assert (
can_access_resource(
service_account, created_by=None, resource_team_id=resource_team_id
)
is expected
)
def test_access_user_can_see_team_match_when_no_user_id_match():
"""Falls through to the team check when user_id doesn't match — lets a
team member read a resource created by a sibling service account in the
same team."""
user = UserAPIKeyAuth(user_id="alice", team_id="team-eng")
assert (
can_access_resource(user, created_by="service-bot", resource_team_id="team-eng")
is True
)
def test_access_service_account_denied_user_resource_in_different_team():
service_account = UserAPIKeyAuth(team_id="team-eng")
assert (
can_access_resource(
service_account, created_by="bob", resource_team_id="team-sales"
)
is False
)
@pytest.mark.parametrize(
"created_by,resource_team_id",
[
(None, None),
("anybody", None),
(None, "any-team"),
("anybody", "any-team"),
],
)
def test_access_identity_less_caller_always_denied(created_by, resource_team_id):
"""The original `None == None` bypass — a caller with no admin role and
no identifying ids is denied against every resource regardless of how
the resource was tagged."""
nobody = UserAPIKeyAuth()
assert (
can_access_resource(
nobody, created_by=created_by, resource_team_id=resource_team_id
)
is False
)