scriptoriumadmin/litellm
1
0
Fork 0
mirror of https://github.com/BerriAI/litellm.git synced 2026-09-21 00:21:49 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 74

feat(keys): let team service account keys use key management endpoints for their own team

Browse source 
Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com>
This commit is contained in:
shivam 2026-09-11 23:54:35 +00:00 • committed by ryan
parent 8fc9c46d1a
commit a052da6974
7 changed files with 415 additions and 47 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

9
litellm/proxy/_types.py
View file

@ -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

5
litellm/proxy/auth/route_checks.py
View file

@ -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)

41
litellm/proxy/management_endpoints/key_management_endpoints.py
View file

@ -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,
)
):

36
litellm/proxy/management_helpers/team_member_permission_checks.py
View file

@ -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]:

69
tests/test_litellm/proxy/auth/test_route_checks.py
View file

@ -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={},
)

130
tests/test_litellm/proxy/management_endpoints/test_key_management_endpoints.py
View file

@ -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"]

172
tests/test_litellm/proxy/management_helpers/test_team_member_permission_checks.py
View file

@ -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"