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

fix(proxy): keep org admins' own team memberships in other orgs visible on team list

Browse source 
An org admin listing their own teams on GET /team/list and GET /v2/team/list
only saw teams in the orgs they administer. Teams they belong to in other
orgs were dropped because the org scope and the membership scope were ANDed.
A self query now unions the two, while a query for another user keeps the
org boundary intersection.

Resolves LIT-3723

Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com>
This commit is contained in:
yassin 2026-09-14 15:27:06 +00:00
parent 9e1ed40db3
commit 58c3ccf668
2 changed files with 247 additions and 57 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

143
litellm/proxy/management_endpoints/team_endpoints.py
View file

@ -179,6 +179,7 @@ from litellm.types.proxy.management_endpoints.team_endpoints import (
if TYPE_CHECKING:
from prisma import Prisma
from prisma import models as prisma_models
from prisma import types as prisma_types
router: Final = APIRouter()
@ -4857,6 +4858,25 @@ async def _get_org_admin_org_ids(
return org_ids if org_ids else None
async def _get_user_team_ids(
user_id: str,
prisma_client: PrismaClient,
user_api_key_cache: UserApiKeyCache,
proxy_logging_obj: ProxyLogging,
) -> tuple[str, ...]:
try:
user: Final = await get_user_object(
user_id=user_id,
prisma_client=prisma_client,
user_api_key_cache=user_api_key_cache,
user_id_upsert=False,
proxy_logging_obj=proxy_logging_obj,
)
except ValueError:
return ()
return tuple(user.teams or ()) if user is not None else ()
async def _build_team_list_where_conditions(
prisma_client: PrismaClient,
team_id: str | None,
@ -4867,12 +4887,16 @@ async def _build_team_list_where_conditions(
search: str | None = None,
search_team_id_match: TeamIdSearchMatch = "exact",
org_admin_org_ids: list[str] | None = None,
own_team_ids: tuple[str, ...] = (),
user_api_key_cache: UserApiKeyCache | None = None,
proxy_logging_obj: ProxyLogging | None = None,
) -> dict[str, object] | None:
"""
Build where conditions for team list query.
An org admin listing their own teams sees the union of the teams in the
orgs they administer and `own_team_ids`, the teams they are a member of.
Returns None when the query is guaranteed to yield no results (e.g. user
has no team memberships), allowing the caller to skip the DB round-trip.
"""
@ -4895,6 +4919,11 @@ async def _build_team_list_where_conditions(
if organization_id:
where_conditions["organization_id"] = organization_id
elif org_admin_org_ids is not None and own_team_ids:
org_or_membership_scope: Final[prisma_types.LiteLLM_TeamTableWhereInput] = {
"OR": [{"organization_id": {"in": org_admin_org_ids}}, {"team_id": {"in": list(own_team_ids)}}]
}
where_conditions["AND"] = [org_or_membership_scope]
elif org_admin_org_ids is not None:
# Org admin: always scope to their orgs, even when filtering by user_id.
where_conditions["organization_id"] = {"in": org_admin_org_ids}
@ -5026,66 +5055,72 @@ async def _enforce_list_team_v2_access(
prisma_client: PrismaClient,
user_api_key_cache: UserApiKeyCache,
proxy_logging_obj: ProxyLogging,
) -> tuple[str | None, list[str] | None]:
) -> tuple[str | None, list[str] | None, tuple[str, ...]]:
"""Enforce access control for list_team_v2.
- Proxy admins and admin viewers can query any teams.
- Org admins can query teams within their organizations.
- Org admins can query teams within their organizations, plus the teams
they are a member of when listing their own teams.
- Regular users can only query their own teams.
Returns the (possibly overridden) user_id and org_admin_org_ids.
Returns the (possibly overridden) user_id, org_admin_org_ids and, for an
org admin's own query, the caller's own team ids.
"""
is_proxy_admin: Final = _user_has_admin_view(user_api_key_dict)
org_admin_org_ids: list[str] | None = None
caller_user_id: Final = user_api_key_dict.user_id
if is_proxy_admin:
return user_id, org_admin_org_ids
return user_id, None, ()
# Always check org admin status so that even own-queries see
# the full set of organisation teams, not just direct memberships.
if user_api_key_dict.user_id:
org_admin_org_ids = await _get_org_admin_org_ids(
user_id=user_api_key_dict.user_id,
org_admin_org_ids: Final = (
await _get_org_admin_org_ids(
user_id=caller_user_id,
prisma_client=prisma_client,
user_api_key_cache=user_api_key_cache,
proxy_logging_obj=proxy_logging_obj,
)
if caller_user_id
else None
)
if org_admin_org_ids is not None:
if caller_user_id and org_admin_org_ids is not None:
# Org admin: validate org_id filter if provided
if organization_id and organization_id not in org_admin_org_ids:
raise HTTPException(
status_code=403,
detail={"error": "You can only view teams within your organizations."},
)
# When the caller is an org admin querying their own teams (or no
# specific user), null out user_id so that
# _build_team_list_where_conditions scopes only by organization_id
# — org admins should see all teams in their orgs, not just teams
# they are a direct member of. Keep user_id when the org admin
# explicitly queries a *different* user's teams.
if user_id is None or user_id == user_api_key_dict.user_id:
user_id = None
is_own_query: Final = user_id is None or user_id == caller_user_id
own_team_ids: Final = (
await _get_user_team_ids(
user_id=caller_user_id,
prisma_client=prisma_client,
user_api_key_cache=user_api_key_cache,
proxy_logging_obj=proxy_logging_obj,
)
if is_own_query
else ()
)
verbose_proxy_logger.debug(
"list_team_v2: org admin access for user=%s, org_ids=%s, user_id_filter=%s",
user_api_key_dict.user_id,
caller_user_id,
org_admin_org_ids,
user_id,
None if is_own_query else user_id,
)
else:
# Not an org admin — fall back to standard route check
if not allowed_route_check_inside_route(user_api_key_dict=user_api_key_dict, requested_user_id=user_id):
raise HTTPException(
status_code=401,
detail={
"error": f"Only admin users can query all teams/other teams. Your user role={user_api_key_dict.user_role}"
},
)
# Regular user — auto-inject caller's user_id
if user_id is None:
user_id = user_api_key_dict.user_id
return None if is_own_query else user_id, org_admin_org_ids, own_team_ids
return user_id, org_admin_org_ids
# Not an org admin — fall back to standard route check
if not allowed_route_check_inside_route(user_api_key_dict=user_api_key_dict, requested_user_id=user_id):
raise HTTPException(
status_code=401,
detail={
"error": f"Only admin users can query all teams/other teams. Your user role={user_api_key_dict.user_role}"
},
)
# Regular user — auto-inject caller's user_id
return user_id if user_id is not None else caller_user_id, None, ()
@router.get(
@ -5163,7 +5198,7 @@ async def list_team_v2(
)
# --- Access control ---
user_id, org_admin_org_ids = await _enforce_list_team_v2_access(
user_id, org_admin_org_ids, own_team_ids = await _enforce_list_team_v2_access(
user_api_key_dict=user_api_key_dict,
user_id=user_id,
organization_id=organization_id,
@ -5195,6 +5230,7 @@ async def list_team_v2(
search=search,
search_team_id_match=search_team_id_match,
org_admin_org_ids=org_admin_org_ids,
own_team_ids=own_team_ids,
user_api_key_cache=user_api_key_cache,
proxy_logging_obj=proxy_logging_obj,
)
@ -5290,18 +5326,18 @@ async def _authorize_and_filter_teams(
Authorize the /team/list request and return filtered teams.
- Proxy admins: all teams (or filtered by user_id if provided).
- Org admins: teams from their orgs (scoped to user_id if provided).
- Org admins: teams from their orgs (scoped to user_id if provided), plus
the teams they are a member of when querying themselves.
- Own query (user_id matches caller): teams the user is a member of.
- Others: 401.
"""
is_proxy_admin: Final = _user_has_admin_view(user_api_key_dict)
is_own_query: Final = (
user_id is not None and user_api_key_dict.user_id is not None and user_api_key_dict.user_id == user_id
)
allowed_org_ids: list[str] | None = None
if not is_proxy_admin:
is_own_query: Final = (
user_id is not None and user_api_key_dict.user_id is not None and user_api_key_dict.user_id == user_id
)
# Check if user is an org admin (even for own queries, so they see org teams)
if user_api_key_dict.user_id is not None:
caller_user: Final = await get_user_object(
@ -5328,33 +5364,30 @@ async def _authorize_and_filter_teams(
},
)
if allowed_org_ids is not None:
# Org admin: query DB for teams in their orgs
if allowed_org_ids is not None and user_id and not is_own_query:
org_teams: Final = await _raw_team_db(TeamRepository(prisma_client)).find_many(
where={"organization_id": {"in": allowed_org_ids}},
include={"litellm_model_table": True},
)
if not user_id:
return list(org_teams)
# Filter org teams to only those where the target user is a member
return [
team
for team in org_teams
if team.members_with_roles and any(m.get("user_id") == user_id for m in team.members_with_roles)
]
elif user_id:
# Regular user: fetch all and filter by membership (Prisma can't filter JSON arrays)
response: Final = await _raw_team_db(TeamRepository(prisma_client)).find_many(
include={"litellm_model_table": True}
)
return [
team
for team in response
if team.members_with_roles and any(m.get("user_id") == user_id for m in team.members_with_roles)
]
else:
response: Final = await _raw_team_db(TeamRepository(prisma_client)).find_many(include={"litellm_model_table": True})
if allowed_org_ids is None and not user_id:
# Proxy admin: all teams
return list(await _raw_team_db(TeamRepository(prisma_client)).find_many(include={"litellm_model_table": True}))
return list(response)
# Prisma can't filter JSON arrays, so membership is filtered in Python
viewer_id: Final = user_id or user_api_key_dict.user_id
return [
team
for team in response
if (allowed_org_ids is not None and team.organization_id in allowed_org_ids)
or (team.members_with_roles and any(m.get("user_id") == viewer_id for m in team.members_with_roles))
]
@router.get("/team/list", tags=["team management"], dependencies=[Depends(user_api_key_auth)])

161
tests/test_litellm/proxy/management_endpoints/test_team_endpoints.py
View file

@ -4055,10 +4055,167 @@ async def test_list_team_v2_org_admin_own_user_id_sees_all_org_teams():
assert result["total"] == 2
assert len(result["teams"]) == 2
# Verify the where clause scopes by org only — no team_id filter
# Verify the where clause scopes by org OR own membership — no
# top-level team_id filter that would hide org teams they aren't in
where = mock_db.litellm_teamtable.find_many.call_args.kwargs["where"]
assert where["organization_id"] == {"in": ["org_A"]}
assert where["AND"] == [
{"OR": [{"organization_id": {"in": ["org_A"]}}, {"team_id": {"in": ["team_1"]}}]}
]
assert "team_id" not in where
assert "organization_id" not in where
@pytest.mark.asyncio
async def test_list_team_v2_org_admin_own_query_keeps_memberships_in_other_orgs(monkeypatch):
"""
An org admin of org_A who is only a member of a team in org_B must still
see that team when listing their own teams: the where clause must union
org scope with membership instead of intersecting them.
Regression test for LIT-3723.
"""
from litellm.proxy.common_utils.user_api_key_cache import UserApiKeyCache
from litellm.proxy.management_endpoints.team_endpoints import list_team_v2
org_admin = UserAPIKeyAuth(user_role=LitellmUserRoles.INTERNAL_USER, user_id="org_admin_user")
cache = UserApiKeyCache()
await cache.async_set_cache(
key="org_admin_user",
value=LiteLLM_UserTable(
user_id="org_admin_user",
teams=["team_in_org_A", "team_in_org_B"],
organization_memberships=[
LiteLLM_OrganizationMembershipTable(
user_id="org_admin_user",
organization_id="org_A",
user_role="org_admin",
spend=0.0,
created_at=datetime.now(),
updated_at=datetime.now(),
),
LiteLLM_OrganizationMembershipTable(
user_id="org_admin_user",
organization_id="org_B",
user_role="internal_user",
spend=0.0,
created_at=datetime.now(),
updated_at=datetime.now(),
),
],
),
model_type=LiteLLM_UserTable,
)
prisma_client = MagicMock()
prisma_client.db.litellm_teamtable.find_many = AsyncMock(return_value=[])
prisma_client.db.litellm_teamtable.count = AsyncMock(return_value=0)
prisma_client.db.litellm_verificationtoken.group_by = AsyncMock(return_value=[])
monkeypatch.setattr("litellm.proxy.proxy_server.prisma_client", prisma_client)
monkeypatch.setattr("litellm.proxy.proxy_server.user_api_key_cache", cache)
proxy_logging_obj = MagicMock()
proxy_logging_obj.service_logging_obj.async_service_success_hook = AsyncMock()
monkeypatch.setattr("litellm.proxy.proxy_server.proxy_logging_obj", proxy_logging_obj)
for own_user_id in ("org_admin_user", None):
await list_team_v2(
http_request=MagicMock(),
user_id=own_user_id,
organization_id=None,
team_id=None,
team_alias=None,
search="team",
user_api_key_dict=org_admin,
page=1,
page_size=10,
sort_by=None,
sort_order="asc",
status=None,
)
where = prisma_client.db.litellm_teamtable.find_many.call_args.kwargs["where"]
assert where["AND"] == [
{
"OR": [
{"organization_id": {"in": ["org_A"]}},
{"team_id": {"in": ["team_in_org_A", "team_in_org_B"]}},
]
}
]
assert where["OR"] == [
{"team_id": "team"},
{"team_alias": {"contains": "team", "mode": "insensitive"}},
]
assert "organization_id" not in where
assert "team_id" not in where
@pytest.mark.asyncio
async def test_list_team_v1_org_admin_own_query_keeps_memberships_in_other_orgs():
"""
/team/list: an org admin of org_A listing their own teams sees every team
in org_A plus the org_B team they are a member of, but a query for another
user stays scoped to org_A.
Regression test for LIT-3723.
"""
from litellm.proxy.common_utils.user_api_key_cache import UserApiKeyCache
from litellm.proxy.management_endpoints.team_endpoints import _authorize_and_filter_teams
org_admin = UserAPIKeyAuth(user_role=LitellmUserRoles.INTERNAL_USER, user_id="org_admin_user")
cache = UserApiKeyCache()
await cache.async_set_cache(
key="org_admin_user",
value=LiteLLM_UserTable(
user_id="org_admin_user",
teams=["team_in_org_A", "team_in_org_B"],
organization_memberships=[
LiteLLM_OrganizationMembershipTable(
user_id="org_admin_user",
organization_id="org_A",
user_role="org_admin",
spend=0.0,
created_at=datetime.now(),
updated_at=datetime.now(),
),
],
),
model_type=LiteLLM_UserTable,
)
def team(team_id, organization_id, *member_ids):
return SimpleNamespace(
team_id=team_id,
organization_id=organization_id,
members_with_roles=[{"user_id": m, "role": "user"} for m in member_ids],
)
all_teams = [
team("team_in_org_A", "org_A", "org_admin_user"),
team("other_team_in_org_A", "org_A", "other_user"),
team("team_in_org_B", "org_B", "org_admin_user", "other_user"),
team("unrelated_team_in_org_B", "org_B", "other_user"),
]
async def find_many(where=None, **kwargs):
if where is None:
return all_teams
return [t for t in all_teams if t.organization_id in where["organization_id"]["in"]]
prisma_client = MagicMock()
prisma_client.db.litellm_teamtable.find_many = AsyncMock(side_effect=find_many)
async def list_teams(user_id):
teams = await _authorize_and_filter_teams(
user_api_key_dict=org_admin,
user_id=user_id,
prisma_client=prisma_client,
user_api_key_cache=cache,
proxy_logging_obj=MagicMock(),
)
return [t.team_id for t in teams]
assert await list_teams("org_admin_user") == ["team_in_org_A", "other_team_in_org_A", "team_in_org_B"]
assert await list_teams(None) == ["team_in_org_A", "other_team_in_org_A", "team_in_org_B"]
assert await list_teams("other_user") == ["other_team_in_org_A"]
@pytest.mark.asyncio