diff --git a/litellm/proxy/_types.py b/litellm/proxy/_types.py index b0d31df92ce..cc44a86c691 100644 --- a/litellm/proxy/_types.py +++ b/litellm/proxy/_types.py @@ -3270,6 +3270,15 @@ class UserAPIKeyAuth(LiteLLM_VerificationTokenView): # the expected response ob user_role=LitellmUserRoles.PROXY_ADMIN, ) + @property + def is_team_service_account(self) -> bool: + return ( + self.user_id is None + and self.team_id is not None + and bool(self.metadata) + and self.metadata.get("service_account_id") is not None + ) + def user_api_key_has_admin_view(user_api_key_dict: UserAPIKeyAuth) -> bool: """Return True if the caller's role grants unscoped read access to all diff --git a/litellm/proxy/auth/route_checks.py b/litellm/proxy/auth/route_checks.py index 0a6b618805d..001ab9a30b0 100644 --- a/litellm/proxy/auth/route_checks.py +++ b/litellm/proxy/auth/route_checks.py @@ -326,7 +326,10 @@ class RouteChecks: pass elif route.startswith("/v1/mcp/") or route.startswith("/mcp-rest/"): pass # authN/authZ handled by api itself - elif RouteChecks.check_passthrough_route_access(route=route, user_api_key_dict=valid_token): + elif RouteChecks.check_passthrough_route_access(route=route, user_api_key_dict=valid_token) or ( + valid_token.is_team_service_account + and RouteChecks.check_route_access(route=route, allowed_routes=LiteLLMRoutes.key_management_routes.value) + ): pass elif valid_token.allowed_routes is not None: # check if route is in allowed_routes (exact match or prefix match) diff --git a/litellm/proxy/management_endpoints/key_management_endpoints.py b/litellm/proxy/management_endpoints/key_management_endpoints.py index 802a7c3e469..42fbe95487f 100644 --- a/litellm/proxy/management_endpoints/key_management_endpoints.py +++ b/litellm/proxy/management_endpoints/key_management_endpoints.py @@ -509,6 +509,16 @@ def _get_user_in_team(team_table: LiteLLM_TeamTableCachedObj, user_id: str | Non return None +def _get_caller_team_role( + team_table: LiteLLM_TeamTableCachedObj, + user_api_key_dict: UserAPIKeyAuth, +) -> Literal["admin", "user"] | None: + if user_api_key_dict.is_team_service_account and user_api_key_dict.team_id == team_table.team_id: + return "user" + member: Final = _get_user_in_team(team_table=team_table, user_id=user_api_key_dict.user_id) + return None if member is None else member.role + + def _calculate_key_rotation_time(rotation_interval: str) -> datetime: """ Helper function to calculate the next rotation time for a key based on the rotation interval. @@ -603,7 +613,7 @@ def _team_key_operation_team_member_check( detail=f"User={assigned_user_id} not assigned to team={team_table.team_id}", ) - team_member_object: Final = _get_user_in_team(team_table=team_table, user_id=user_api_key_dict.user_id) + caller_team_role: Final = _get_caller_team_role(team_table=team_table, user_api_key_dict=user_api_key_dict) is_admin: Final = ( user_api_key_dict.user_role is not None and user_api_key_dict.user_role == LitellmUserRoles.PROXY_ADMIN.value @@ -611,22 +621,22 @@ def _team_key_operation_team_member_check( if is_admin: return True - elif team_member_object is None: + elif caller_team_role is None: raise HTTPException( status_code=400, detail=f"User={user_api_key_dict.user_id} not assigned to team={team_table.team_id}", ) elif ( "allowed_team_member_roles" in team_key_generation - and team_member_object.role not in team_key_generation["allowed_team_member_roles"] + and caller_team_role not in team_key_generation["allowed_team_member_roles"] ): raise HTTPException( status_code=400, - detail=f"Team member role {team_member_object.role} not in allowed_team_member_roles={team_key_generation['allowed_team_member_roles']}", + detail=f"Team member role {caller_team_role} not in allowed_team_member_roles={team_key_generation['allowed_team_member_roles']}", ) TeamMemberPermissionChecks.does_team_member_have_permissions_for_endpoint( - team_member_object=team_member_object, + team_member_role=caller_team_role, team_table=team_table, route=route, ) @@ -747,6 +757,12 @@ def key_generation_check( Check if admin has restricted key creation to certain roles for teams or individuals """ + if user_api_key_dict.is_team_service_account and data.team_id != user_api_key_dict.team_id: + raise HTTPException( + status_code=403, + detail=f"Service account keys can only create keys for their own team. team_id={user_api_key_dict.team_id}", + ) + ## check if key is for team or individual is_team_key: Final = _is_team_key(data=data) _is_admin: Final = ( @@ -2223,6 +2239,10 @@ async def generate_service_account_key_fn( prisma_client=prisma_client, ) + if data.metadata is None or data.metadata.get("service_account_id") is None: + service_account_id: Final = (data.metadata or {}).get("service_account_id") or data.key_alias or str(uuid.uuid4()) + data.metadata = {**(data.metadata or {}), "service_account_id": service_account_id} # rebind-ok: stamping the generated service_account_id onto the request model so it persists on the key + verbose_proxy_logger.debug("entered /key/generate") custom_key_generate_hook: Final[Callable[..., Awaitable[Mapping[str, object]]] | None] = _custom_key_generate_hook( @@ -3837,8 +3857,10 @@ async def validate_key_team_change( detail=f"Key={key.token} has a rpm_limit={key.rpm_limit} which is greater than the team's rpm_limit={team.rpm_limit}.", ) + team_table: Final = cast(LiteLLM_TeamTableCachedObj, team) + # Check if the key's user_id is a member of the team - member_object: Final = _get_user_in_team(team_table=cast(LiteLLM_TeamTableCachedObj, team), user_id=key.user_id) + member_object: Final = _get_user_in_team(team_table=team_table, user_id=key.user_id) if key.user_id is not None: if not member_object: raise HTTPException( @@ -3854,8 +3876,11 @@ async def validate_key_team_change( team_obj=team, ) or TeamMemberPermissionChecks.does_team_member_have_permissions_for_endpoint( - team_member_object=member_object, - team_table=cast(LiteLLM_TeamTableCachedObj, team), + team_member_role=_get_caller_team_role( + team_table=team_table, + user_api_key_dict=change_initiated_by, + ), + team_table=team_table, route=KeyManagementRoutes.KEY_UPDATE.value, ) ): diff --git a/litellm/proxy/management_helpers/team_member_permission_checks.py b/litellm/proxy/management_helpers/team_member_permission_checks.py index 86c7a7bd947..a7e75caeb69 100644 --- a/litellm/proxy/management_helpers/team_member_permission_checks.py +++ b/litellm/proxy/management_helpers/team_member_permission_checks.py @@ -1,4 +1,4 @@ -from typing import Final +from typing import Final, Literal from litellm.proxy._types import ( KeyManagementRoutes, @@ -6,7 +6,6 @@ from litellm.proxy._types import ( LiteLLM_VerificationToken, LiteLLMRoutes, LitellmUserRoles, - Member, ProxyErrorTypes, ProxyException, UserAPIKeyAuth, @@ -27,7 +26,6 @@ DEFAULT_TEAM_MEMBER_PERMISSIONS: Final = BASELINE_TEAM_MEMBER_PERMISSIONS class TeamMemberPermissionChecks: @staticmethod def get_permissions_for_team_member( - team_member_object: Member, team_table: LiteLLM_TeamTableCachedObj, ) -> list[KeyManagementRoutes]: """ @@ -67,7 +65,7 @@ class TeamMemberPermissionChecks: Main handler for checking if a team member can update a key """ from litellm.proxy.management_endpoints.key_management_endpoints import ( - _get_user_in_team, + _get_caller_team_role, ) # 1. Don't execute these checks if the user role is proxy admin @@ -87,12 +85,12 @@ class TeamMemberPermissionChecks: check_db_only=True, ) - # 4. Extract `Member` object from `team_table` - key_assigned_user_in_team: Final = _get_user_in_team(team_table=team_table, user_id=user_api_key_dict.user_id) + # 4. Resolve the caller's role in the key's team (service accounts act as "user") + caller_team_role: Final = _get_caller_team_role(team_table=team_table, user_api_key_dict=user_api_key_dict) # 5. Check if the team member has permissions for the endpoint has_permission: Final = TeamMemberPermissionChecks.does_team_member_have_permissions_for_endpoint( - team_member_object=key_assigned_user_in_team, + team_member_role=caller_team_role, team_table=team_table, route=route, ) @@ -106,7 +104,7 @@ class TeamMemberPermissionChecks: @staticmethod def does_team_member_have_permissions_for_endpoint( - team_member_object: Member | None, + team_member_role: Literal["admin", "user"] | None, team_table: LiteLLM_TeamTableCachedObj, route: str, ) -> bool | None: @@ -116,13 +114,12 @@ class TeamMemberPermissionChecks: # permission checks only run for non-admin users # Non-Admin user trying to access information about a team's key - if team_member_object is None: + if team_member_role is None: return False - if team_member_object.role == "admin": + if team_member_role == "admin": return True _team_member_permissions: Final = TeamMemberPermissionChecks.get_permissions_for_team_member( - team_member_object=team_member_object, team_table=team_table, ) team_member_permissions = TeamMemberPermissionChecks._get_list_of_route_enum_as_str(_team_member_permissions) @@ -156,7 +153,7 @@ class TeamMemberPermissionChecks: from fastapi import HTTPException from litellm.proxy.management_endpoints.key_management_endpoints import ( - _get_user_in_team, + _get_caller_team_role, ) # No-op when the request does not assign any access groups. @@ -177,20 +174,19 @@ class TeamMemberPermissionChecks: ), ) - team_member_object: Final = _get_user_in_team(team_table=team_table, user_id=user_api_key_dict.user_id) + caller_team_role: Final = _get_caller_team_role(team_table=team_table, user_api_key_dict=user_api_key_dict) # Team admins always bypass (consistent with other member-permission checks). - if team_member_object is not None and team_member_object.role == "admin": + if caller_team_role == "admin": return permissions: Final = ( TeamMemberPermissionChecks._get_list_of_route_enum_as_str( TeamMemberPermissionChecks.get_permissions_for_team_member( - team_member_object=team_member_object, team_table=team_table, ) ) - if team_member_object is not None + if caller_team_role is not None else [] ) @@ -214,7 +210,7 @@ class TeamMemberPermissionChecks: Returns True if the user belongs to the team that the key is assigned to """ from litellm.proxy.management_endpoints.key_management_endpoints import ( - _get_user_in_team, + _get_caller_team_role, ) from litellm.proxy.proxy_server import prisma_client, user_api_key_cache @@ -228,9 +224,9 @@ class TeamMemberPermissionChecks: check_db_only=True, ) - # 4. Extract `Member` object from `team_table` - team_member_object: Final = _get_user_in_team(team_table=team_table, user_id=user_api_key_dict.user_id) - return team_member_object is not None + # 4. Resolve the caller's role in the key's team (service accounts act as "user") + caller_team_role: Final = _get_caller_team_role(team_table=team_table, user_api_key_dict=user_api_key_dict) + return caller_team_role is not None @staticmethod def get_all_available_team_member_permissions() -> list[str]: diff --git a/tests/test_litellm/proxy/auth/test_route_checks.py b/tests/test_litellm/proxy/auth/test_route_checks.py index 72c7011e6f5..b1405fc35cb 100644 --- a/tests/test_litellm/proxy/auth/test_route_checks.py +++ b/tests/test_litellm/proxy/auth/test_route_checks.py @@ -3967,3 +3967,72 @@ def test_auto_router_session_read_grant_rejects_other_methods_paths_and_scopes( RouteChecks.should_call_route(route, valid_token, request) assert error.value.status_code == 403 + + +@pytest.mark.parametrize("route", ["/key/generate", "/key/update"]) +def test_team_service_account_key_allowed_key_management_routes(route): + """A service account key (user_id=None, team_id set, metadata.service_account_id) + can reach key-management routes; team scoping is enforced in the handlers.""" + valid_token = UserAPIKeyAuth( + api_key="sk", + team_id="t1", + user_id=None, + metadata={"service_account_id": "ci"}, + ) + request = MagicMock(spec=Request) + request.query_params = {} + + result = RouteChecks.non_proxy_admin_allowed_routes_check( + user_obj=None, + _user_role=None, + route=route, + request=request, + valid_token=valid_token, + request_data={}, + ) + assert result is None + + +def test_team_service_account_key_rejected_for_non_key_management_route(): + """The service account carve-out does not extend past key-management routes.""" + valid_token = UserAPIKeyAuth( + api_key="sk", + team_id="t1", + user_id=None, + metadata={"service_account_id": "ci"}, + ) + request = MagicMock(spec=Request) + request.query_params = {} + + with pytest.raises(Exception, match="Only proxy admin can be used to generate, delete, update"): + RouteChecks.non_proxy_admin_allowed_routes_check( + user_obj=None, + _user_role=None, + route="/team/new", + request=request, + valid_token=valid_token, + request_data={}, + ) + + +def test_team_key_without_service_account_marker_still_rejected(): + """A team key without metadata.service_account_id is not a service account + and still cannot reach key-management routes.""" + valid_token = UserAPIKeyAuth( + api_key="sk", + team_id="t1", + user_id=None, + metadata={}, + ) + request = MagicMock(spec=Request) + request.query_params = {} + + with pytest.raises(Exception, match="Only proxy admin can be used to generate, delete, update"): + RouteChecks.non_proxy_admin_allowed_routes_check( + user_obj=None, + _user_role=None, + route="/key/generate", + request=request, + valid_token=valid_token, + request_data={}, + ) diff --git a/tests/test_litellm/proxy/management_endpoints/test_key_management_endpoints.py b/tests/test_litellm/proxy/management_endpoints/test_key_management_endpoints.py index cc0a7631b59..f865e959a78 100644 --- a/tests/test_litellm/proxy/management_endpoints/test_key_management_endpoints.py +++ b/tests/test_litellm/proxy/management_endpoints/test_key_management_endpoints.py @@ -18,6 +18,7 @@ import inspect from litellm.proxy._types import ( GenerateKeyRequest, + KeyManagementRoutes, NewUserRequest, LiteLLM_BudgetTable, LiteLLM_ObjectPermissionBase, @@ -3099,7 +3100,7 @@ async def test_validate_key_team_change_with_member_permissions(): # Verify the permission check was called with correct parameters mock_has_perms.assert_called_once_with( - team_member_object=mock_member_object, + team_member_role=mock_member_object.role, team_table=mock_team, route=KeyManagementRoutes.KEY_UPDATE.value, ) @@ -19733,3 +19734,130 @@ async def test_bulk_update_team_keys_runs_custom_key_policy_per_key(monkeypatch) assert [policy_request.operation for policy_request in received] == ["update", "update"] assert [policy_request.effective_key.max_budget for policy_request in received] == [50.0, 50.0] assert [policy_request.effective_key.team_id for policy_request in received] == ["team-abc", "team-abc"] + + +class TestServiceAccountKeyGenerationCheck: + """Service account keys (user_id=None, team_id set, metadata.service_account_id) + may only create keys for their own team.""" + + def _service_account_token(self, team_id: str) -> UserAPIKeyAuth: + return UserAPIKeyAuth( + api_key="sk-sa", + user_id=None, + team_id=team_id, + metadata={"service_account_id": "sa-1"}, + ) + + def test_other_team_denied(self): + data = GenerateKeyRequest(team_id="team-b") + with pytest.raises(HTTPException) as exc_info: + key_generation_check( + team_table=None, + user_api_key_dict=self._service_account_token(team_id="team-a"), + data=data, + route=KeyManagementRoutes.KEY_GENERATE, + ) + assert exc_info.value.status_code == 403 + + def test_personal_key_denied(self): + """team_id=None would mint a personal key; service accounts may only + create keys for their own team.""" + data = GenerateKeyRequest() + with pytest.raises(HTTPException) as exc_info: + key_generation_check( + team_table=None, + user_api_key_dict=self._service_account_token(team_id="team-a"), + data=data, + route=KeyManagementRoutes.KEY_GENERATE, + ) + assert exc_info.value.status_code == 403 + + def test_own_team_with_permission_allowed(self): + team_table = LiteLLM_TeamTableCachedObj( + team_id="team-a", + members_with_roles=[], + team_member_permissions=["/key/generate"], + ) + data = GenerateKeyRequest(team_id="team-a") + assert ( + key_generation_check( + team_table=team_table, + user_api_key_dict=self._service_account_token(team_id="team-a"), + data=data, + route=KeyManagementRoutes.KEY_GENERATE, + ) + is True + ) + + def test_own_team_without_permission_denied(self): + team_table = LiteLLM_TeamTableCachedObj( + team_id="team-a", + members_with_roles=[], + team_member_permissions=["/key/info"], + ) + data = GenerateKeyRequest(team_id="team-a") + with pytest.raises(ProxyException) as exc_info: + key_generation_check( + team_table=team_table, + user_api_key_dict=self._service_account_token(team_id="team-a"), + data=data, + route=KeyManagementRoutes.KEY_GENERATE, + ) + assert str(exc_info.value.code) == "401" + + +def _stub_service_account_generation(monkeypatch): + """Stub the DB lookups generate_service_account_key_fn needs so the test + exercises only the service_account_id stamping and user_id clearing.""" + from litellm.proxy import proxy_server + from litellm.proxy.management_endpoints import key_management_endpoints as kme + + mock_helper = AsyncMock(return_value=MagicMock()) + monkeypatch.setattr(proxy_server, "prisma_client", MagicMock()) + monkeypatch.setattr(kme, "validate_team_id_used_in_service_account_request", AsyncMock()) + monkeypatch.setattr(kme, "_common_key_generation_helper", mock_helper) + return mock_helper + + +@pytest.mark.asyncio +async def test_generate_service_account_key_stamps_service_account_id(monkeypatch): + """generate_service_account_key_fn must stamp metadata.service_account_id + (key_alias fallback) so the key is identifiable as a service account by + is_team_service_account and check_if_token_is_service_account.""" + from litellm.proxy.management_endpoints.key_management_endpoints import ( + generate_service_account_key_fn, + ) + + mock_helper = _stub_service_account_generation(monkeypatch) + data = GenerateKeyRequest(team_id="team-a", key_alias="sa-alias") + + await generate_service_account_key_fn( + data=data, + user_api_key_dict=UserAPIKeyAuth(user_role=LitellmUserRoles.PROXY_ADMIN, api_key="sk-1"), + litellm_changed_by=None, + ) + + assert data.metadata is not None + assert data.metadata["service_account_id"] == "sa-alias" + assert data.user_id is None + mock_helper.assert_awaited_once() + + +@pytest.mark.asyncio +async def test_generate_service_account_key_generates_uuid_when_no_alias(monkeypatch): + """Without key_alias, service_account_id falls back to a generated uuid.""" + from litellm.proxy.management_endpoints.key_management_endpoints import ( + generate_service_account_key_fn, + ) + + _stub_service_account_generation(monkeypatch) + data = GenerateKeyRequest(team_id="team-a") + + await generate_service_account_key_fn( + data=data, + user_api_key_dict=UserAPIKeyAuth(user_role=LitellmUserRoles.PROXY_ADMIN, api_key="sk-1"), + litellm_changed_by=None, + ) + + assert data.metadata is not None + assert data.metadata["service_account_id"] diff --git a/tests/test_litellm/proxy/management_helpers/test_team_member_permission_checks.py b/tests/test_litellm/proxy/management_helpers/test_team_member_permission_checks.py index 36c61eddbb2..55d29724e38 100644 --- a/tests/test_litellm/proxy/management_helpers/test_team_member_permission_checks.py +++ b/tests/test_litellm/proxy/management_helpers/test_team_member_permission_checks.py @@ -3,7 +3,12 @@ from unittest.mock import MagicMock import pytest -from litellm.proxy._types import KeyManagementRoutes, Member, ProxyException +from litellm.proxy._types import ( + KeyManagementRoutes, + Member, + ProxyException, + UserAPIKeyAuth, +) from litellm.proxy.management_helpers.team_member_permission_checks import ( BASELINE_TEAM_MEMBER_PERMISSIONS, TeamMemberPermissionChecks, @@ -21,22 +26,16 @@ class TestGetPermissionsForTeamMember: def test_none_permissions_returns_defaults(self): """When team_member_permissions is None, return DEFAULT_TEAM_MEMBER_PERMISSIONS.""" team = _make_team_table(None) - member = MagicMock(spec=Member) - result = TeamMemberPermissionChecks.get_permissions_for_team_member( - team_member_object=member, team_table=team - ) + result = TeamMemberPermissionChecks.get_permissions_for_team_member(team_table=team) assert set(result) == set(BASELINE_TEAM_MEMBER_PERMISSIONS) def test_empty_list_includes_baseline(self): """When team_member_permissions is [], baseline permissions are still included.""" team = _make_team_table([]) - member = MagicMock(spec=Member) - result = TeamMemberPermissionChecks.get_permissions_for_team_member( - team_member_object=member, team_table=team - ) + result = TeamMemberPermissionChecks.get_permissions_for_team_member(team_table=team) assert KeyManagementRoutes.KEY_INFO in result assert KeyManagementRoutes.KEY_HEALTH in result @@ -44,11 +43,8 @@ class TestGetPermissionsForTeamMember: def test_explicit_permissions_include_baseline(self): """When explicit permissions are set, baseline is always included.""" team = _make_team_table(["/key/generate", "/key/delete"]) - member = MagicMock(spec=Member) - result = TeamMemberPermissionChecks.get_permissions_for_team_member( - team_member_object=member, team_table=team - ) + result = TeamMemberPermissionChecks.get_permissions_for_team_member(team_table=team) assert KeyManagementRoutes.KEY_GENERATE in result assert KeyManagementRoutes.KEY_DELETE in result @@ -58,11 +54,8 @@ class TestGetPermissionsForTeamMember: def test_explicit_permissions_with_baseline_no_duplicates(self): """When explicit permissions already include baseline, no duplicates.""" team = _make_team_table(["/key/info", "/key/generate"]) - member = MagicMock(spec=Member) - result = TeamMemberPermissionChecks.get_permissions_for_team_member( - team_member_object=member, team_table=team - ) + result = TeamMemberPermissionChecks.get_permissions_for_team_member(team_table=team) # Using set ensures no duplicates from the implementation assert KeyManagementRoutes.KEY_INFO in result @@ -402,3 +395,148 @@ class TestEnforceMemberCanAssignAccessGroups: team_table=self._team(["/key/generate", self.AG_PERMISSION]), access_group_ids=["ag-1"], ) + + +class TestDoesTeamMemberHavePermissionsForEndpoint: + def _team(self, team_member_permissions, team_id="team-a"): + team = MagicMock() + team.team_id = team_id + team.team_member_permissions = team_member_permissions + return team + + def test_none_role_returns_false(self): + """A caller with no team membership is denied.""" + result = TeamMemberPermissionChecks.does_team_member_have_permissions_for_endpoint( + team_member_role=None, + team_table=self._team(["/key/update"]), + route=KeyManagementRoutes.KEY_UPDATE.value, + ) + assert result is False + + def test_admin_role_always_allowed(self): + """Team admins bypass the member permission list.""" + result = TeamMemberPermissionChecks.does_team_member_have_permissions_for_endpoint( + team_member_role="admin", + team_table=self._team([]), + route=KeyManagementRoutes.KEY_UPDATE.value, + ) + assert result is True + + def test_user_role_with_permission_allowed(self): + result = TeamMemberPermissionChecks.does_team_member_have_permissions_for_endpoint( + team_member_role="user", + team_table=self._team(["/key/update"]), + route=KeyManagementRoutes.KEY_UPDATE.value, + ) + assert result is True + + def test_user_role_without_permission_raises(self): + with pytest.raises(ProxyException) as exc: + TeamMemberPermissionChecks.does_team_member_have_permissions_for_endpoint( + team_member_role="user", + team_table=self._team(["/key/generate"]), + route=KeyManagementRoutes.KEY_UPDATE.value, + ) + assert str(exc.value.code) == "401" + assert exc.value.type == "team_member_permission_error" + + +class TestCanTeamMemberExecuteKeyManagementEndpointServiceAccount: + def _service_account_token(self, team_id: str) -> UserAPIKeyAuth: + return UserAPIKeyAuth( + api_key="sk-test", + user_id=None, + team_id=team_id, + metadata={"service_account_id": "sa-1"}, + ) + + @pytest.mark.asyncio + async def test_service_account_same_team_with_permission(self, monkeypatch): + """A service account key can manage keys in its own team when the + team grants the route via team_member_permissions.""" + from litellm.proxy.management_helpers import ( + team_member_permission_checks as module, + ) + + async def _mock_get_team_object(**kwargs): + team = MagicMock() + team.team_id = "team-a" + team.members_with_roles = [] + team.team_member_permissions = ["/key/update"] + return team + + monkeypatch.setattr(module, "get_team_object", _mock_get_team_object) + + existing_key_row = MagicMock() + existing_key_row.team_id = "team-a" + + result = await TeamMemberPermissionChecks.can_team_member_execute_key_management_endpoint( + user_api_key_dict=self._service_account_token(team_id="team-a"), + route=KeyManagementRoutes.KEY_UPDATE, + prisma_client=MagicMock(), + user_api_key_cache=MagicMock(), + existing_key_row=existing_key_row, + ) + assert result is None + + @pytest.mark.asyncio + async def test_service_account_same_team_without_permission(self, monkeypatch): + """A service account key is denied when the team's + team_member_permissions does not include the route.""" + from litellm.proxy.management_helpers import ( + team_member_permission_checks as module, + ) + + async def _mock_get_team_object(**kwargs): + team = MagicMock() + team.team_id = "team-a" + team.members_with_roles = [] + team.team_member_permissions = ["/key/generate"] + return team + + monkeypatch.setattr(module, "get_team_object", _mock_get_team_object) + + existing_key_row = MagicMock() + existing_key_row.team_id = "team-a" + + with pytest.raises(ProxyException) as exc: + await TeamMemberPermissionChecks.can_team_member_execute_key_management_endpoint( + user_api_key_dict=self._service_account_token(team_id="team-a"), + route=KeyManagementRoutes.KEY_UPDATE, + prisma_client=MagicMock(), + user_api_key_cache=MagicMock(), + existing_key_row=existing_key_row, + ) + assert str(exc.value.code) == "401" + assert exc.value.type == "team_member_permission_error" + + @pytest.mark.asyncio + async def test_service_account_different_team_denied(self, monkeypatch): + """A service account key cannot manage keys in another team, even if + that team grants the route to its members.""" + from litellm.proxy.management_helpers import ( + team_member_permission_checks as module, + ) + + async def _mock_get_team_object(**kwargs): + team = MagicMock() + team.team_id = "team-b" + team.members_with_roles = [] + team.team_member_permissions = ["/key/update"] + return team + + monkeypatch.setattr(module, "get_team_object", _mock_get_team_object) + + existing_key_row = MagicMock() + existing_key_row.team_id = "team-b" + + with pytest.raises(ProxyException) as exc: + await TeamMemberPermissionChecks.can_team_member_execute_key_management_endpoint( + user_api_key_dict=self._service_account_token(team_id="team-a"), + route=KeyManagementRoutes.KEY_UPDATE, + prisma_client=MagicMock(), + user_api_key_cache=MagicMock(), + existing_key_row=existing_key_row, + ) + assert str(exc.value.code) == "401" + assert exc.value.type == "team_member_permission_error"