fix(auth): let internal users view search tools (#29542)

* fix(auth): let internal users view search tools

Internal users could not see search tools in the UI even when an admin
created them, while vector stores were visible. The Search Tools page
rendered but its list calls 403'd because the read routes were not in
internal_user_routes. Grant internal users read-only access to the
listing and provider routes; create/update/delete stay admin-only.

Resolves LIT-3150

* fix(auth): scope /search_tools/list to caller-allowed tools

Adding the read routes to internal_user_routes let any internal user
call /search_tools/list, which returned every configured tool's id,
provider, api_base, and metadata regardless of the caller's
object_permission.search_tools allowlist. The api_key was masked, so
this was metadata disclosure rather than a credential leak, but it
ignored the key/team scoping that /search already enforces.

Filter the listing through the same can_key_call_search_tool /
can_team_call_search_tool checks (exposed as a boolean
can_user_view_search_tool), so non-admin callers only see tools they
may invoke; admins still see all. Mirrors how /vector_store/list
scopes results.

* fix(types): resolve mypy errors in list_search_tools

The config and DB build loops reused one loop variable, so mypy pinned
it to the config element type (SearchToolTypedDict); its .get() calls
returned object and the DB element (SearchTool) failed the reuse
assignment. Give each loop its own variable so each gets its real type,
and coerce the config tool's SearchToolInfoTypedDict to a plain dict to
match SearchToolInfoResponse.search_tool_info.
This commit is contained in:
ryan-crabbe-berri 2026-06-02 18:36:14 -07:00 committed by GitHub
parent 08223e1ec3
commit 0a767ed14f
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
5 changed files with 348 additions and 15 deletions

View File

@ -701,6 +701,10 @@ class LiteLLMRoutes(enum.Enum):
"/v2/guardrails/list",
"/project/list",
"/project/info",
# Read-only search tool routes power the Search Tools UI page.
# Create/update/delete and test_connection stay admin-only.
"/search_tools/list",
"/search_tools/ui/available_providers",
]
+ spend_tracking_routes
+ key_management_routes

View File

@ -3428,6 +3428,29 @@ async def can_team_call_search_tool(
)
async def can_user_view_search_tool(
search_tool_name: str,
valid_token: UserAPIKeyAuth,
team_object: Optional[LiteLLM_TeamTable],
) -> bool:
"""
Boolean variant of the key + team authorization enforced on /search, used to
scope /search_tools/list so a non-admin caller only sees tools it may invoke.
"""
try:
await can_key_call_search_tool(
search_tool_name=search_tool_name,
valid_token=valid_token,
)
await can_team_call_search_tool(
search_tool_name=search_tool_name,
team_object=team_object,
)
except ProxyException:
return False
return True
async def is_valid_fallback_model(
model: str,
llm_router: Optional[Router],

View File

@ -3,12 +3,17 @@ CRUD ENDPOINTS FOR SEARCH TOOLS
"""
from datetime import datetime
from typing import Any, Dict, List, Union
from typing import Any, Dict, List, Optional, Union
from fastapi import APIRouter, Depends, HTTPException
from pydantic import BaseModel
from litellm._logging import verbose_proxy_logger
from litellm.proxy._types import (
LiteLLM_TeamTable,
LitellmUserRoles,
UserAPIKeyAuth,
)
from litellm.proxy.auth.user_api_key_auth import user_api_key_auth
from litellm.proxy.search_endpoints.search_tool_registry import SearchToolRegistry
from litellm.types.search import (
@ -41,13 +46,61 @@ def _convert_datetime_to_str(value: Union[datetime, str, None]) -> Union[str, No
return value
async def _filter_visible_search_tools(
search_tools: List[SearchToolInfoResponse],
user_api_key_dict: UserAPIKeyAuth,
) -> List[SearchToolInfoResponse]:
"""
Drop search tools the caller is not authorized to invoke, applying the same
key/team object_permission allowlists enforced on /search. Admins see all tools.
"""
if user_api_key_dict.user_role in (
LitellmUserRoles.PROXY_ADMIN,
LitellmUserRoles.PROXY_ADMIN_VIEW_ONLY,
):
return search_tools
from litellm.proxy.auth.auth_checks import (
can_user_view_search_tool,
get_team_object,
)
from litellm.proxy.proxy_server import (
prisma_client,
proxy_logging_obj,
user_api_key_cache,
)
team_object: Optional[LiteLLM_TeamTable] = None
if user_api_key_dict.team_id:
team_object = await get_team_object(
team_id=user_api_key_dict.team_id,
prisma_client=prisma_client,
user_api_key_cache=user_api_key_cache,
parent_otel_span=user_api_key_dict.parent_otel_span,
proxy_logging_obj=proxy_logging_obj,
)
visible: List[SearchToolInfoResponse] = []
for tool in search_tools:
tool_name = tool.get("search_tool_name")
if tool_name and await can_user_view_search_tool(
search_tool_name=tool_name,
valid_token=user_api_key_dict,
team_object=team_object,
):
visible.append(tool)
return visible
@router.get(
"/search_tools/list",
tags=["Search Tools"],
dependencies=[Depends(user_api_key_auth)],
response_model=ListSearchToolsResponse,
)
async def list_search_tools():
async def list_search_tools(
user_api_key_dict: UserAPIKeyAuth = Depends(user_api_key_auth),
):
"""
List all search tools that are available in the database and config file.
@ -114,22 +167,25 @@ async def list_search_tools():
f"Could not get config-defined search tools: {e}"
)
for search_tool in config_search_tools:
tool_name = search_tool.get("search_tool_name")
for config_search_tool in config_search_tools:
tool_name = config_search_tool.get("search_tool_name")
if tool_name:
litellm_params_dict = dict(search_tool.get("litellm_params", {}))
litellm_params_dict = dict(config_search_tool.get("litellm_params", {}))
masked_litellm_params_dict = _get_masked_values(
litellm_params_dict,
unmasked_length=4,
number_of_asterisks=4,
)
config_tool_info = config_search_tool.get("search_tool_info")
search_tool_configs.append(
SearchToolInfoResponse(
search_tool_id=None,
search_tool_name=tool_name,
litellm_params=masked_litellm_params_dict,
search_tool_info=search_tool.get("search_tool_info"),
search_tool_info=(
dict(config_tool_info) if config_tool_info else None
),
created_at=None,
updated_at=None,
is_from_config=True,
@ -142,8 +198,8 @@ async def list_search_tools():
if tool.get("search_tool_name") not in db_tool_names
]
for search_tool in search_tools_from_db:
litellm_params_dict = dict(search_tool.get("litellm_params", {}))
for db_search_tool in search_tools_from_db:
litellm_params_dict = dict(db_search_tool.get("litellm_params", {}))
masked_litellm_params_dict = _get_masked_values(
litellm_params_dict,
unmasked_length=4,
@ -152,17 +208,25 @@ async def list_search_tools():
search_tool_configs.append(
SearchToolInfoResponse(
search_tool_id=search_tool.get("search_tool_id"),
search_tool_name=search_tool.get("search_tool_name", ""),
search_tool_id=db_search_tool.get("search_tool_id"),
search_tool_name=db_search_tool.get("search_tool_name", ""),
litellm_params=masked_litellm_params_dict,
search_tool_info=search_tool.get("search_tool_info"),
created_at=_convert_datetime_to_str(search_tool.get("created_at")),
updated_at=_convert_datetime_to_str(search_tool.get("updated_at")),
search_tool_info=db_search_tool.get("search_tool_info"),
created_at=_convert_datetime_to_str(
db_search_tool.get("created_at")
),
updated_at=_convert_datetime_to_str(
db_search_tool.get("updated_at")
),
is_from_config=False,
)
)
return ListSearchToolsResponse(search_tools=search_tool_configs)
visible_search_tools = await _filter_visible_search_tools(
search_tool_configs, user_api_key_dict
)
return ListSearchToolsResponse(search_tools=visible_search_tools)
except Exception as e:
verbose_proxy_logger.exception(f"Error getting search tools: {e}")
raise HTTPException(status_code=500, detail=str(e))

View File

@ -2602,3 +2602,72 @@ def test_legitimate_passthrough_routes_still_classified_as_llm_route(route):
assert (
RouteChecks.is_llm_api_route(route=route) is True
), f"{route!r} should be classified as an LLM API route"
@pytest.mark.parametrize(
"route",
[
"/search_tools/list",
"/search_tools/ui/available_providers",
],
)
def test_internal_user_can_read_search_tools(route):
"""Regression for LIT-3150: internal users must be able to view search tools,
the same way they can view vector stores."""
user_obj = LiteLLM_UserTable(
user_id="test_user",
user_email="user@example.com",
user_role=LitellmUserRoles.INTERNAL_USER.value,
)
valid_token = UserAPIKeyAuth(
user_id="test_user",
user_role=LitellmUserRoles.INTERNAL_USER.value,
)
request = MagicMock(spec=Request)
request.query_params = {}
RouteChecks.non_proxy_admin_allowed_routes_check(
user_obj=user_obj,
_user_role=LitellmUserRoles.INTERNAL_USER.value,
route=route,
request=request,
valid_token=valid_token,
request_data={},
)
@pytest.mark.parametrize(
"route",
[
"/search_tools", # create
"/search_tools/abc123", # update / delete / get-by-id
"/search_tools/test_connection",
],
)
def test_internal_user_blocked_from_search_tool_writes(route):
"""Read access must not leak the search-tool management write routes to
internal users; only proxy admins create/update/delete/test them."""
user_obj = LiteLLM_UserTable(
user_id="test_user",
user_email="user@example.com",
user_role=LitellmUserRoles.INTERNAL_USER.value,
)
valid_token = UserAPIKeyAuth(
user_id="test_user",
user_role=LitellmUserRoles.INTERNAL_USER.value,
)
request = MagicMock(spec=Request)
request.query_params = {}
with pytest.raises(Exception) as exc_info:
RouteChecks.non_proxy_admin_allowed_routes_check(
user_obj=user_obj,
_user_role=LitellmUserRoles.INTERNAL_USER.value,
route=route,
request=request,
valid_token=valid_token,
request_data={},
)
assert "Only proxy admin" in str(exc_info.value)
assert f"Route={route}" in str(exc_info.value)
assert "Your role=internal_user" in str(exc_info.value)

View File

@ -1,3 +1,4 @@
import contextlib
import os
import sys
from datetime import datetime
@ -10,7 +11,12 @@ sys.path.insert(
0, os.path.abspath("../../../../..")
) # Adds the parent directory to the system path
from litellm.proxy._types import LitellmUserRoles, UserAPIKeyAuth
from litellm.proxy._types import (
LiteLLM_ObjectPermissionTable,
LiteLLM_TeamTable,
LitellmUserRoles,
UserAPIKeyAuth,
)
# Import proxy_server module first to ensure it's initialized
import litellm.proxy.proxy_server as ps
@ -603,3 +609,170 @@ async def test_list_search_tools_db_masking_sensitive_values(monkeypatch):
assert tool4["litellm_params"]["search_provider"] == "custom"
finally:
app.dependency_overrides.pop(user_api_key_auth, None)
@contextlib.contextmanager
def _mock_search_tool_backend(db_tools):
"""Patch the DB registry, prisma client, and config so /search_tools/list
returns exactly ``db_tools`` (no config-defined tools)."""
mock_registry = MagicMock()
mock_registry.get_all_search_tools_from_db = AsyncMock(return_value=db_tools)
mock_proxy_config = MagicMock()
mock_proxy_config.get_config = AsyncMock(return_value={})
mock_proxy_config.parse_search_tools = MagicMock(return_value=None)
with (
patch(
"litellm.proxy.search_endpoints.search_tool_management.SEARCH_TOOL_REGISTRY",
mock_registry,
),
patch("litellm.proxy.proxy_server.prisma_client", MagicMock()),
patch("litellm.proxy.proxy_server.proxy_config", mock_proxy_config),
):
yield
def _scoping_db_tools():
return [
{
"search_tool_id": "db-id-1",
"search_tool_name": "db-tool-1",
"litellm_params": {
"search_provider": "perplexity",
"api_key": "pplx-secret-1",
"api_base": "https://api.perplexity.ai",
},
"search_tool_info": {"description": "Perplexity"},
"created_at": datetime(2024, 1, 1),
"updated_at": datetime(2024, 1, 1),
},
{
"search_tool_id": "db-id-2",
"search_tool_name": "db-tool-2",
"litellm_params": {
"search_provider": "tavily",
"api_key": "tvly-secret-2",
"api_base": "https://api.tavily.com",
},
"search_tool_info": {"description": "Tavily"},
"created_at": datetime(2024, 1, 1),
"updated_at": datetime(2024, 1, 1),
},
{
"search_tool_id": "db-id-3",
"search_tool_name": "db-tool-3",
"litellm_params": {"search_provider": "exa", "api_key": "exa-secret-3"},
"search_tool_info": {"description": "Exa"},
"created_at": datetime(2024, 1, 1),
"updated_at": datetime(2024, 1, 1),
},
]
@contextlib.contextmanager
def _override_auth(user):
from litellm.proxy.auth.user_api_key_auth import user_api_key_auth
app.dependency_overrides[user_api_key_auth] = lambda: user
try:
yield
finally:
app.dependency_overrides.pop(user_api_key_auth, None)
@pytest.mark.asyncio
async def test_list_search_tools_scoped_to_key_object_permission():
"""
Regression: an internal user whose key is restricted to specific search tools
must only see those tools. Before the fix /search_tools/list returned every
configured tool, leaking ids, api_base, and metadata for tools it cannot call.
"""
restricted_user = UserAPIKeyAuth(
user_role=LitellmUserRoles.INTERNAL_USER,
user_id="internal_user",
object_permission=LiteLLM_ObjectPermissionTable(
object_permission_id="op-key",
search_tools=["db-tool-1"],
),
)
with (
_mock_search_tool_backend(_scoping_db_tools()),
_override_auth(restricted_user),
):
response = TestClient(app).get("/search_tools/list")
assert response.status_code == 200
tools = response.json()["search_tools"]
assert [t["search_tool_name"] for t in tools] == ["db-tool-1"]
leaked = {t["litellm_params"].get("api_base") for t in tools}
assert "https://api.tavily.com" not in leaked
@pytest.mark.asyncio
async def test_list_search_tools_unrestricted_internal_user_sees_all():
"""An internal user with no search_tools allowlist is unrestricted and sees every tool."""
unrestricted_user = UserAPIKeyAuth(
user_role=LitellmUserRoles.INTERNAL_USER, user_id="internal_user"
)
with (
_mock_search_tool_backend(_scoping_db_tools()),
_override_auth(unrestricted_user),
):
response = TestClient(app).get("/search_tools/list")
assert response.status_code == 200
names = {t["search_tool_name"] for t in response.json()["search_tools"]}
assert names == {"db-tool-1", "db-tool-2", "db-tool-3"}
@pytest.mark.asyncio
async def test_list_search_tools_scoped_to_team_object_permission():
"""A team-level search_tools allowlist also scopes the listing for a non-admin caller."""
team_member = UserAPIKeyAuth(
user_role=LitellmUserRoles.INTERNAL_USER,
user_id="internal_user",
team_id="team-1",
)
team_object = LiteLLM_TeamTable(
team_id="team-1",
object_permission=LiteLLM_ObjectPermissionTable(
object_permission_id="op-team",
search_tools=["db-tool-2"],
),
)
with (
_mock_search_tool_backend(_scoping_db_tools()),
patch(
"litellm.proxy.auth.auth_checks.get_team_object",
AsyncMock(return_value=team_object),
),
_override_auth(team_member),
):
response = TestClient(app).get("/search_tools/list")
assert response.status_code == 200
assert [t["search_tool_name"] for t in response.json()["search_tools"]] == [
"db-tool-2"
]
@pytest.mark.asyncio
async def test_list_search_tools_admin_with_restricted_key_still_sees_all():
"""Proxy admins bypass search-tool scoping even if their key carries an allowlist."""
admin_user = UserAPIKeyAuth(
user_role=LitellmUserRoles.PROXY_ADMIN,
user_id="admin_user",
object_permission=LiteLLM_ObjectPermissionTable(
object_permission_id="op-admin",
search_tools=["db-tool-1"],
),
)
with _mock_search_tool_backend(_scoping_db_tools()), _override_auth(admin_user):
response = TestClient(app).get("/search_tools/list")
assert response.status_code == 200
names = {t["search_tool_name"] for t in response.json()["search_tools"]}
assert names == {"db-tool-1", "db-tool-2", "db-tool-3"}