diff --git a/litellm/proxy/search_endpoints/endpoints.py b/litellm/proxy/search_endpoints/endpoints.py index 2676682c59d..7f195342dc8 100644 --- a/litellm/proxy/search_endpoints/endpoints.py +++ b/litellm/proxy/search_endpoints/endpoints.py @@ -139,38 +139,17 @@ async def search( search_tool_name_value: Final = data["search_tool_name"] # Authorization check: verify key can access this search tool - from litellm.proxy.auth.auth_checks import ( - can_key_call_search_tool, - can_team_call_search_tool, - get_team_object, + from litellm.proxy.search_endpoints.search_tool_access import ( + authorize_search_tool, + resolve_allowlist_teams, ) try: - # Check key-level access - await can_key_call_search_tool( + await authorize_search_tool( search_tool_name=search_tool_name_value, - valid_token=user_api_key_dict, + user_api_key_dict=user_api_key_dict, + teams=await resolve_allowlist_teams(user_api_key_dict), ) - - # Check team-level access if key is associated with a team - if user_api_key_dict.team_id: - from litellm.proxy.proxy_server import ( - prisma_client, - proxy_logging_obj, - user_api_key_cache, - ) - - team_object: Final = 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, - ) - await can_team_call_search_tool( - search_tool_name=search_tool_name_value, - team_object=team_object, - ) except Exception as e: verbose_proxy_logger.error("Search tool authorization failed for %s: %s", search_tool_name_value, e) raise diff --git a/litellm/proxy/search_endpoints/search_tool_access.py b/litellm/proxy/search_endpoints/search_tool_access.py new file mode 100644 index 00000000000..54776a0e836 --- /dev/null +++ b/litellm/proxy/search_endpoints/search_tool_access.py @@ -0,0 +1,179 @@ +""" +Shared team scoping for the search tool authorization checks. +""" + +import asyncio +from collections.abc import Sequence +from typing import Final, Literal, Protocol + +from fastapi import HTTPException + +from litellm._logging import verbose_proxy_logger +from litellm.constants import UI_SESSION_TOKEN_TEAM_ID +from litellm.proxy._types import LiteLLM_TeamTable, ProxyException, UserAPIKeyAuth + + +class TeamObjectLookup(Protocol): + async def __call__(self, team_id: str, user_api_key_dict: UserAPIKeyAuth) -> LiteLLM_TeamTable: ... + + +class SessionTeamIdsLookup(Protocol): + async def __call__(self, user_api_key_dict: UserAPIKeyAuth) -> Sequence[str]: ... + + +async def team_object_from_db(team_id: str, user_api_key_dict: UserAPIKeyAuth) -> LiteLLM_TeamTable: + from litellm.proxy.auth.auth_checks import get_team_object + from litellm.proxy.proxy_server import ( + prisma_client, + proxy_logging_obj, + user_api_key_cache, + ) + + return await get_team_object( + team_id=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, + ) + + +async def session_team_ids_from_db(user_api_key_dict: UserAPIKeyAuth) -> Sequence[str]: + from litellm.proxy.auth.auth_checks import get_user_object + from litellm.proxy.proxy_server import ( + prisma_client, + proxy_logging_obj, + user_api_key_cache, + ) + + user_object: Final = await get_user_object( + user_id=user_api_key_dict.user_id, + prisma_client=prisma_client, + user_api_key_cache=user_api_key_cache, + user_id_upsert=False, + parent_otel_span=user_api_key_dict.parent_otel_span, + proxy_logging_obj=proxy_logging_obj, + ) + if user_object is None: + raise ValueError("Cannot resolve the teams of a dashboard session key that carries no user id.") + return tuple(user_object.teams or ()) + + +async def _load_session_team( + team_id: str, + user_api_key_dict: UserAPIKeyAuth, + lookup_team_object: TeamObjectLookup, +) -> LiteLLM_TeamTable | HTTPException: + try: + return await lookup_team_object(team_id, user_api_key_dict) + except HTTPException as exc: + verbose_proxy_logger.warning( + "Search tool scoping: dropping team %s for dashboard session user %s: %s", + team_id, + user_api_key_dict.user_id, + exc.detail, + ) + return exc + + +async def _session_teams( + user_api_key_dict: UserAPIKeyAuth, + lookup_team_object: TeamObjectLookup, + lookup_session_team_ids: SessionTeamIdsLookup, +) -> tuple[LiteLLM_TeamTable, ...]: + team_ids: Final = await lookup_session_team_ids(user_api_key_dict) + loaded: Final = await asyncio.gather( + *(_load_session_team(team_id, user_api_key_dict, lookup_team_object) for team_id in team_ids) + ) + teams: Final = tuple(entry for entry in loaded if isinstance(entry, LiteLLM_TeamTable)) + if loaded and not teams: + raise next(entry for entry in loaded if isinstance(entry, HTTPException)) + return teams + + +async def resolve_allowlist_teams( + user_api_key_dict: UserAPIKeyAuth, + lookup_team_object: TeamObjectLookup = team_object_from_db, + lookup_session_team_ids: SessionTeamIdsLookup = session_team_ids_from_db, +) -> tuple[LiteLLM_TeamTable, ...]: + """ + The teams whose object_permission allowlists scope this caller. + + A virtual key names exactly one team and that lookup still surfaces its failure, so a key + pointing at a team that does not exist is rejected rather than treated as unscoped. + + An Admin UI session key instead carries UI_SESSION_TOKEN_TEAM_ID, a reserved sentinel that + never has a row in LiteLLM_TeamTable (`/team/new` rejects it as a team id), so looking it up + would raise 404 for every dashboard caller. It resolves to the real teams backing the session + user, the same identity the MCP dashboard surfaces resolve, which keeps the team allowlists + binding on the dashboard instead of dropping them. + + Resolving that set never fails open. Loading the session user surfaces its own failure, and one + unloadable team is dropped only because `/team/delete` leaves its id behind on the user row, so + a stale membership is ordinary and dropping it can only narrow a union the other teams still + scope. Once no team survives there is no union left to narrow, so the first failure is raised + rather than handing back the empty set that means "belongs to no team". + """ + team_id: Final = user_api_key_dict.team_id + if not team_id: + return () + if team_id != UI_SESSION_TOKEN_TEAM_ID: + return (await lookup_team_object(team_id, user_api_key_dict),) + return await _session_teams(user_api_key_dict, lookup_team_object, lookup_session_team_ids) + + +async def _team_denial(search_tool_name: str, team: LiteLLM_TeamTable) -> ProxyException | None: + from litellm.proxy.auth.auth_checks import can_team_call_search_tool + + try: + await can_team_call_search_tool(search_tool_name=search_tool_name, team_object=team) + except ProxyException as denial: + return denial + return None + + +async def authorize_search_tool( + search_tool_name: str, + user_api_key_dict: UserAPIKeyAuth, + teams: Sequence[LiteLLM_TeamTable], +) -> Literal[True]: + """ + Enforce the key and team object_permission allowlists for one search tool. + + The caller is authorized when its key permits the tool and at least one of its teams does. + A caller with no teams is unrestricted at the team level, matching a key that belongs to no + team, and is still bound by its key allowlist. + """ + from litellm.proxy.auth.auth_checks import ( + can_key_call_search_tool, + can_team_call_search_tool, + ) + + await can_key_call_search_tool(search_tool_name=search_tool_name, valid_token=user_api_key_dict) + if not teams: + return await can_team_call_search_tool(search_tool_name=search_tool_name, team_object=None) + + denials: Final = await asyncio.gather(*(_team_denial(search_tool_name, team) for team in teams)) + if any(denial is None for denial in denials): + return True + raise next(denial for denial in denials if denial is not None) + + +async def can_view_search_tool( + search_tool_name: str, + user_api_key_dict: UserAPIKeyAuth, + teams: Sequence[LiteLLM_TeamTable], +) -> bool: + """ + Boolean variant of authorize_search_tool, so a listing surface shows exactly the tools the + caller may invoke. + """ + try: + await authorize_search_tool( + search_tool_name=search_tool_name, + user_api_key_dict=user_api_key_dict, + teams=teams, + ) + except ProxyException: + return False + return True diff --git a/tests/test_litellm/proxy/search_endpoints/test_endpoints.py b/tests/test_litellm/proxy/search_endpoints/test_endpoints.py new file mode 100644 index 00000000000..cdd9c6ba2ae --- /dev/null +++ b/tests/test_litellm/proxy/search_endpoints/test_endpoints.py @@ -0,0 +1,255 @@ +import contextlib +import os +import sys +from unittest.mock import AsyncMock, MagicMock, patch + +import pytest +from fastapi import HTTPException +from fastapi.testclient import TestClient + +sys.path.insert(0, os.path.abspath("../../../../..")) + +from litellm.constants import UI_SESSION_TOKEN_TEAM_ID +from litellm.proxy._types import ( + LiteLLM_ObjectPermissionTable, + LiteLLM_TeamTable, + LiteLLM_UserTable, + LitellmUserRoles, + UserAPIKeyAuth, +) +from litellm.proxy.proxy_server import app + +SEARCH_TOOLS = [ + { + "search_tool_name": "db-tool-1", + "litellm_params": {"search_provider": "perplexity", "api_key": "pplx-secret-1"}, + }, + { + "search_tool_name": "db-tool-2", + "litellm_params": {"search_provider": "tavily", "api_key": "tvly-secret-2"}, + }, + { + "search_tool_name": "db-tool-3", + "litellm_params": {"search_provider": "exa", "api_key": "exa-secret-3"}, + }, +] + + +@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) + + +@contextlib.contextmanager +def _mock_search_backend(lookup_team_object, session_team_ids=()): + """Patch the router, prisma client, and downstream request processing so /search reaches + the authorization checks and, if they pass, returns without calling a provider.""" + router = MagicMock() + router.search_tools = SEARCH_TOOLS + processor = MagicMock() + processor.return_value.base_process_llm_request = AsyncMock(return_value={"object": "search", "results": []}) + session_user = LiteLLM_UserTable(user_id="internal_user", teams=list(session_team_ids)) + with ( + patch("litellm.proxy.proxy_server.llm_router", router), + patch("litellm.proxy.proxy_server.prisma_client", MagicMock()), + patch("litellm.proxy.auth.auth_checks.get_team_object", lookup_team_object), + patch("litellm.proxy.auth.auth_checks.get_user_object", AsyncMock(return_value=session_user)), + patch( + "litellm.proxy.search_endpoints.endpoints.ProxyBaseLLMRequestProcessing", + processor, + ), + ): + yield + + +def _team_ids_looked_up(lookup: AsyncMock) -> list[str]: + return [awaited.kwargs["team_id"] for awaited in lookup.await_args_list] + + +def _dashboard_session_key(search_tools: list[str] | None) -> UserAPIKeyAuth: + return UserAPIKeyAuth( + user_role=LitellmUserRoles.INTERNAL_USER, + user_id="internal_user", + team_id=UI_SESSION_TOKEN_TEAM_ID, + object_permission=( + LiteLLM_ObjectPermissionTable(object_permission_id="op-key", search_tools=search_tools) + if search_tools is not None + else None + ), + ) + + +def _team_allowing(team_id: str, search_tools: list[str]) -> LiteLLM_TeamTable: + return LiteLLM_TeamTable( + team_id=team_id, + object_permission=LiteLLM_ObjectPermissionTable( + object_permission_id=f"op-{team_id}", + search_tools=search_tools, + ), + ) + + +def _lookup_returning(*teams: LiteLLM_TeamTable) -> AsyncMock: + by_id = {team.team_id: team for team in teams} + + async def lookup(**kwargs): + team_id = kwargs["team_id"] + if team_id not in by_id: + raise HTTPException( + status_code=404, + detail={"error": f"Team doesn't exist in db. Team={team_id}."}, + ) + return by_id[team_id] + + return AsyncMock(side_effect=lookup) + + +def _team_lookup_fails_with_404(team_id: str) -> AsyncMock: + return AsyncMock( + side_effect=HTTPException( + status_code=404, + detail={"error": f"Team doesn't exist in db. Team={team_id}."}, + ) + ) + + +@pytest.mark.asyncio +async def test_search_dashboard_session_is_scoped_by_the_users_real_team_allowlist(): + """ + Regression: the Admin UI session key carries the reserved team id ``litellm-dashboard``, which + has no row in LiteLLM_TeamTable. Treating that as "no team" dropped team-level authorization + entirely, so a dashboard session with no key allowlist could invoke every tool on the proxy. + The session must be scoped by the allowlists of the teams its user actually belongs to. + """ + lookup = _lookup_returning(_team_allowing("team-1", ["db-tool-2"])) + + with ( + _mock_search_backend(lookup, session_team_ids=["team-1"]), + _override_auth(_dashboard_session_key(None)), + ): + blocked = TestClient(app).post("/search/db-tool-1", json={"query": "litellm"}) + allowed = TestClient(app).post("/search/db-tool-2", json={"query": "litellm"}) + + assert blocked.status_code == 403 + assert "db-tool-1" in blocked.text + assert allowed.status_code == 200 + assert _team_ids_looked_up(lookup) == ["team-1", "team-1"] + + +@pytest.mark.asyncio +async def test_search_dashboard_session_is_permitted_by_any_of_the_users_teams(): + """A user on several teams may invoke a tool any one of those teams allowlists.""" + lookup = _lookup_returning( + _team_allowing("team-1", ["db-tool-2"]), + _team_allowing("team-2", ["db-tool-3"]), + ) + + with ( + _mock_search_backend(lookup, session_team_ids=["team-1", "team-2"]), + _override_auth(_dashboard_session_key(None)), + ): + allowed = TestClient(app).post("/search/db-tool-3", json={"query": "litellm"}) + blocked = TestClient(app).post("/search/db-tool-1", json={"query": "litellm"}) + + assert allowed.status_code == 200 + assert blocked.status_code == 403 + + +@pytest.mark.asyncio +async def test_search_dashboard_session_key_does_not_look_up_the_ui_team(): + """ + Regression: resolving ``litellm-dashboard`` as a real team raised 404, so no non-admin could + invoke a search tool from the dashboard. + """ + ui_team_is_not_a_real_team = _team_lookup_fails_with_404(UI_SESSION_TOKEN_TEAM_ID) + + with ( + _mock_search_backend(ui_team_is_not_a_real_team), + _override_auth(_dashboard_session_key(["db-tool-3"])), + ): + response = TestClient(app).post("/search/db-tool-3", json={"query": "litellm"}) + + assert response.status_code == 200 + ui_team_is_not_a_real_team.assert_not_awaited() + + +@pytest.mark.asyncio +async def test_search_dashboard_session_denied_when_none_of_the_users_teams_load(): + """ + Regression: a dashboard session whose every team failed to load resolved to an empty team set, + which reads as "belongs to no team" and skips team authorization altogether. A user who could + invoke nothing must not become a user who can invoke everything. + """ + every_team_is_gone = _team_lookup_fails_with_404("team-1") + + with ( + _mock_search_backend(every_team_is_gone, session_team_ids=["team-1", "team-2"]), + _override_auth(_dashboard_session_key(None)), + ): + response = TestClient(app).post("/search/db-tool-1", json={"query": "litellm"}) + + assert response.status_code == 404 + assert every_team_is_gone.await_count == 2 + + +@pytest.mark.asyncio +async def test_search_dashboard_session_key_still_bound_by_its_key_allowlist(): + """ + Resolving the session through its user's teams must not widen access: a dashboard session + whose key allowlists only one tool is still refused every other tool. + """ + lookup = _lookup_returning(_team_allowing("team-1", [])) + + with ( + _mock_search_backend(lookup, session_team_ids=["team-1"]), + _override_auth(_dashboard_session_key(["db-tool-3"])), + ): + response = TestClient(app).post("/search/db-tool-1", json={"query": "litellm"}) + + assert response.status_code == 403 + assert "db-tool-1" in response.text + + +@pytest.mark.asyncio +async def test_search_real_team_allowlist_still_blocks_a_tool_it_does_not_permit(): + """A caller with a real team is still resolved and scoped by that team's allowlist.""" + team_member = UserAPIKeyAuth( + user_role=LitellmUserRoles.INTERNAL_USER, + user_id="internal_user", + team_id="team-1", + ) + lookup = _lookup_returning(_team_allowing("team-1", ["db-tool-2"])) + + with _mock_search_backend(lookup), _override_auth(team_member): + blocked = TestClient(app).post("/search/db-tool-1", json={"query": "litellm"}) + allowed = TestClient(app).post("/search/db-tool-2", json={"query": "litellm"}) + + assert blocked.status_code == 403 + assert allowed.status_code == 200 + assert _team_ids_looked_up(lookup) == ["team-1", "team-1"] + + +@pytest.mark.asyncio +async def test_search_missing_real_team_is_still_rejected(): + """ + A caller whose real team cannot be resolved must not fall through to "no team", which + would drop that team's allowlist and let it call tools it may not. + """ + team_member = UserAPIKeyAuth( + user_role=LitellmUserRoles.INTERNAL_USER, + user_id="internal_user", + team_id="deleted-team", + ) + lookup = _team_lookup_fails_with_404("deleted-team") + + with _mock_search_backend(lookup), _override_auth(team_member): + response = TestClient(app).post("/search/db-tool-1", json={"query": "litellm"}) + + assert response.status_code == 404 + assert _team_ids_looked_up(lookup) == ["deleted-team"] diff --git a/tests/test_litellm/proxy/search_endpoints/test_search_tool_access.py b/tests/test_litellm/proxy/search_endpoints/test_search_tool_access.py new file mode 100644 index 00000000000..e4dcabb5db8 --- /dev/null +++ b/tests/test_litellm/proxy/search_endpoints/test_search_tool_access.py @@ -0,0 +1,217 @@ +import os +import sys +from unittest.mock import AsyncMock + +import pytest +from fastapi import HTTPException + +sys.path.insert(0, os.path.abspath("../../../../..")) + +from litellm.constants import UI_SESSION_TOKEN_TEAM_ID +from litellm.proxy._types import ( + LiteLLM_ObjectPermissionTable, + LiteLLM_TeamTable, + LitellmUserRoles, + ProxyException, + UserAPIKeyAuth, +) +from litellm.proxy.search_endpoints.search_tool_access import ( + authorize_search_tool, + resolve_allowlist_teams, + session_team_ids_from_db, +) + + +def _team(team_id: str, search_tools: list[str] | None = None) -> LiteLLM_TeamTable: + return LiteLLM_TeamTable( + team_id=team_id, + object_permission=( + LiteLLM_ObjectPermissionTable(object_permission_id=f"op-{team_id}", search_tools=search_tools) + if search_tools is not None + else None + ), + ) + + +def _dashboard_session(search_tools: list[str] | None = None) -> UserAPIKeyAuth: + return UserAPIKeyAuth( + user_role=LitellmUserRoles.INTERNAL_USER, + user_id="internal_user", + team_id=UI_SESSION_TOKEN_TEAM_ID, + object_permission=( + LiteLLM_ObjectPermissionTable(object_permission_id="op-key", search_tools=search_tools) + if search_tools is not None + else None + ), + ) + + +def _key_on_team(team_id: str) -> UserAPIKeyAuth: + return UserAPIKeyAuth( + user_role=LitellmUserRoles.INTERNAL_USER, + user_id="internal_user", + team_id=team_id, + ) + + +def _lookup_failing_with_404(team_id: str) -> AsyncMock: + return AsyncMock( + side_effect=HTTPException( + status_code=404, + detail={"error": f"Team doesn't exist in db. Team={team_id}."}, + ) + ) + + +@pytest.mark.asyncio +async def test_resolve_returns_no_teams_when_the_key_has_none(): + lookup = AsyncMock() + session_team_ids = AsyncMock() + + assert await resolve_allowlist_teams(_key_on_team(""), lookup, session_team_ids) == () + lookup.assert_not_awaited() + session_team_ids.assert_not_awaited() + + +@pytest.mark.asyncio +async def test_resolve_looks_up_a_real_team_and_never_widens_to_the_session_path(): + team = _team("team-1") + lookup = AsyncMock(return_value=team) + session_team_ids = AsyncMock() + + assert await resolve_allowlist_teams(_key_on_team("team-1"), lookup, session_team_ids) == (team,) + assert [awaited.args[0] for awaited in lookup.await_args_list] == ["team-1"] + session_team_ids.assert_not_awaited() + + +@pytest.mark.asyncio +async def test_resolve_propagates_a_real_team_lookup_failure(): + """A key naming a team that does not exist must be rejected, not treated as unscoped.""" + lookup = _lookup_failing_with_404("deleted-team") + + with pytest.raises(HTTPException) as exc_info: + await resolve_allowlist_teams(_key_on_team("deleted-team"), lookup, AsyncMock()) + + assert exc_info.value.status_code == 404 + + +@pytest.mark.asyncio +async def test_resolve_maps_the_session_sentinel_onto_the_users_real_teams(): + """The sentinel itself is never looked up; the session user's own teams are.""" + teams = (_team("team-1"), _team("team-2")) + lookup = AsyncMock(side_effect=teams) + session_team_ids = AsyncMock(return_value=["team-1", "team-2"]) + + assert await resolve_allowlist_teams(_dashboard_session(), lookup, session_team_ids) == teams + assert [awaited.args[0] for awaited in lookup.await_args_list] == ["team-1", "team-2"] + + +@pytest.mark.asyncio +async def test_resolve_drops_only_the_session_team_that_cannot_be_loaded(): + """Dropping an unloadable team can only narrow the union, so the rest still bind.""" + loadable = _team("team-2") + + async def lookup_impl(team_id, _user_api_key_dict): + if team_id == "team-1": + raise HTTPException(status_code=404, detail={"error": "Team doesn't exist in db. Team=team-1."}) + return loadable + + session_team_ids = AsyncMock(return_value=["team-1", "team-2"]) + + resolved = await resolve_allowlist_teams(_dashboard_session(), AsyncMock(side_effect=lookup_impl), session_team_ids) + + assert resolved == (loadable,) + + +@pytest.mark.asyncio +async def test_resolve_denies_a_session_when_no_team_can_be_loaded(): + """ + The collapse case. Dropping teams narrows a union only while one survives; once none do there + is nothing left to narrow, and returning () would read as 'belongs to no team' and go + unrestricted. The failure has to surface instead. + """ + lookup = _lookup_failing_with_404("team-1") + session_team_ids = AsyncMock(return_value=["team-1", "team-2"]) + + with pytest.raises(HTTPException) as exc_info: + await resolve_allowlist_teams(_dashboard_session(), lookup, session_team_ids) + + assert exc_info.value.status_code == 404 + assert lookup.await_count == 2 + + +@pytest.mark.asyncio +async def test_resolve_surfaces_a_failure_to_resolve_the_session_users_teams(): + """A session user whose own record cannot be read is not a session user with no teams.""" + session_team_ids = AsyncMock(side_effect=ValueError("User doesn't exist in db. 'user_id'=internal_user.")) + + with pytest.raises(ValueError, match="User doesn't exist in db"): + await resolve_allowlist_teams(_dashboard_session(), AsyncMock(), session_team_ids) + + +@pytest.mark.asyncio +async def test_resolve_returns_no_teams_for_a_session_user_on_no_teams(): + """The one empty result that is genuinely unscoped, and the reason the collapse needs its own case.""" + lookup = AsyncMock() + + assert await resolve_allowlist_teams(_dashboard_session(), lookup, AsyncMock(return_value=[])) == () + lookup.assert_not_awaited() + + +@pytest.mark.asyncio +async def test_session_team_ids_do_not_swallow_a_failure_to_load_the_session_user(monkeypatch): + """Resolution reads the session user directly so a DB failure cannot arrive as an empty team list.""" + monkeypatch.setattr("litellm.proxy.proxy_server.prisma_client", None) + + with pytest.raises(Exception, match="No db connected"): + await session_team_ids_from_db(_dashboard_session()) + + +@pytest.mark.asyncio +async def test_authorize_permits_a_tool_any_one_team_allowlists(): + assert ( + await authorize_search_tool( + search_tool_name="db-tool-2", + user_api_key_dict=_dashboard_session(), + teams=(_team("team-1", ["db-tool-1"]), _team("team-2", ["db-tool-2"])), + ) + is True + ) + + +@pytest.mark.asyncio +async def test_authorize_denies_a_tool_no_team_allowlists(): + with pytest.raises(ProxyException) as exc_info: + await authorize_search_tool( + search_tool_name="db-tool-3", + user_api_key_dict=_dashboard_session(), + teams=(_team("team-1", ["db-tool-1"]), _team("team-2", ["db-tool-2"])), + ) + + assert exc_info.value.code == "403" + assert "db-tool-3" in exc_info.value.message + + +@pytest.mark.asyncio +async def test_authorize_is_unrestricted_at_team_level_with_no_teams(): + assert ( + await authorize_search_tool( + search_tool_name="db-tool-3", + user_api_key_dict=_dashboard_session(), + teams=(), + ) + is True + ) + + +@pytest.mark.asyncio +async def test_authorize_enforces_the_key_allowlist_before_any_team_grant(): + """A team grant never widens a key past its own allowlist.""" + with pytest.raises(ProxyException) as exc_info: + await authorize_search_tool( + search_tool_name="db-tool-3", + user_api_key_dict=_dashboard_session(["db-tool-1"]), + teams=(_team("team-1", ["db-tool-3"]),), + ) + + assert exc_info.value.code == "403"