feat: integrate search tool access control into search endpoints

- Add access check in search() endpoint after resolving search_tool_name
- Add _get_allowed_search_tool_names() helper for computing allowed tools
- Filter list_search_tools() results based on key/team permissions
- Least privilege: empty search_tools list means no access

Co-authored-by: yuneng-jiang <yuneng-jiang@users.noreply.github.com>
This commit is contained in:
Cursor Agent 2026-03-19 01:37:51 +00:00
parent c219302bfe
commit fd975053ba
No known key found for this signature in database

View file

@ -12,6 +12,60 @@ from litellm.proxy.common_request_processing import ProxyBaseLLMRequestProcessin
router = APIRouter()
async def _get_allowed_search_tool_names(
user_api_key_dict: UserAPIKeyAuth,
) -> Optional[List[str]]:
"""
Compute the intersection of key-level and team-level search tool permissions.
Returns:
None no restriction (all tools accessible)
list only those tool names are accessible (may be empty = none)
"""
from litellm.proxy.proxy_server import prisma_client
if prisma_client is None:
return None
key_allowed: Optional[List[str]] = None
team_allowed: Optional[List[str]] = None
# Key-level permissions
if user_api_key_dict.object_permission_id is not None:
key_perm = (
await prisma_client.db.litellm_objectpermissiontable.find_unique(
where={
"object_permission_id": user_api_key_dict.object_permission_id
},
)
)
if key_perm is not None:
key_allowed = key_perm.search_tools # None means no restriction
# Team-level permissions
team_perm_id = getattr(user_api_key_dict, "team_object_permission_id", None)
if team_perm_id is not None:
team_perm = (
await prisma_client.db.litellm_objectpermissiontable.find_unique(
where={"object_permission_id": team_perm_id},
)
)
if team_perm is not None:
team_allowed = team_perm.search_tools # None means no restriction
# Combine: both None → None (no restriction)
# One set → use that set
# Both set → intersection
if key_allowed is None and team_allowed is None:
return None
if key_allowed is None:
return team_allowed
if team_allowed is None:
return key_allowed
# Both are set - return the intersection
return list(set(key_allowed) & set(team_allowed))
@router.post(
"/v1/search/{search_tool_name}",
dependencies=[Depends(user_api_key_auth)],
@ -163,6 +217,16 @@ async def search(
data["metadata"] = {}
data["metadata"]["model_group"] = search_tool_name_value
# Access control check for search tools
resolved_search_tool_name = data.get("search_tool_name")
if resolved_search_tool_name:
from litellm.proxy.auth.auth_checks import search_tool_access_check
await search_tool_access_check(
search_tool_name=resolved_search_tool_name,
valid_token=user_api_key_dict,
)
# Process request using ProxyBaseLLMRequestProcessing
processor = ProxyBaseLLMRequestProcessing(data=data)
try:
@ -258,6 +322,15 @@ async def list_search_tools(
search_tools_list.append(tool_info)
# Filter search tools based on user's permissions
allowed_names = await _get_allowed_search_tool_names(user_api_key_dict)
if allowed_names is not None:
search_tools_list = [
tool
for tool in search_tools_list
if tool.get("search_tool_name") in allowed_names
]
return {"object": "list", "data": search_tools_list}
except Exception as e:
from litellm._logging import verbose_proxy_logger