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

Merge pull request #41086 from BerriAI/litellm_team_list_org_admin_membership_lit3723

Browse source 
fix(proxy): keep org admins' own team memberships in other orgs visible on team list
This commit is contained in:
Yassin Kortam 2026-09-14 13:57:08 -07:00 • committed by GitHub
commit 4123b4bc2b
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
5 changed files with 308 additions and 63 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

8
litellm/proxy/auth/auth_checks.py
View file

@ -123,6 +123,7 @@ from litellm.repositories.table_repositories import (
from litellm.repositories.team_repository import TeamRepository
from litellm.repositories.user_repository import UserRepository
from litellm.router import Router
from litellm.types.proxy.auth.auth_checks import UserNotFoundError
from litellm.types.proxy.model_access_group_budget import ModelAccessGroupBudget
from litellm.utils import get_utc_datetime
@ -2375,13 +2376,6 @@ async def _backfill_null_user_email(
return updated_row
class UserNotFoundError(ValueError):
"""The user row is provably absent, as opposed to merely unreadable, so a caller that reads a missing row as no user-level limits can key on it without also swallowing a database that would not answer."""
def __init__(self, user_id: str) -> None:
super().__init__(f"User doesn't exist in db. 'user_id'={user_id}. Create user via `/user/new` call.")
@log_db_metrics
async def get_user_object(
user_id: str | None,

2
litellm/proxy/auth/resolvers/grants.py
View file

@ -28,11 +28,11 @@ from litellm.proxy._types import (
)
from litellm.proxy.auth.auth_checks import (
TeamNotFoundError,
UserNotFoundError,
get_team_membership,
get_team_object,
get_user_object,
)
from litellm.types.proxy.auth.auth_checks import UserNotFoundError
if TYPE_CHECKING:
from litellm.proxy._types import Span

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

@ -156,6 +156,7 @@ from litellm.repositories.verification_token_repository import (
VerificationTokenRepository,
)
from litellm.router import Router
from litellm.types.proxy.auth.auth_checks import UserNotFoundError
from litellm.types.proxy.management_endpoints.common_daily_activity import (
SpendAnalyticsPaginatedResponse,
)
@ -179,6 +180,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 +4859,26 @@ async def _get_org_admin_org_ids(
return org_ids if org_ids else None
async def _get_user_team_ids_from_db(
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,
check_db_only=True,
)
except UserNotFoundError:
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 +4889,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 +4921,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 +5057,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_from_db(
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,
_sanitize_for_log(caller_user_id),
org_admin_org_ids,
user_id,
_sanitize_for_log(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 +5200,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 +5232,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,
)
@ -5291,17 +5329,16 @@ async def _authorize_and_filter_teams(
- Proxy admins: all teams (or filtered by user_id if provided).
- Org admins: teams from their orgs (scoped to user_id if provided).
- Own query (user_id matches caller): teams the user is a member of.
- Own query (user_id matches caller): teams the user is a member of, across all orgs.
- 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 +5365,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 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 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
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)
]
@router.get("/team/list", tags=["team management"], dependencies=[Depends(user_api_key_auth)])

8
litellm/types/proxy/auth/auth_checks.py Normal file
View file

@ -0,0 +1,8 @@
"""Failure values raised by `litellm/proxy/auth/auth_checks.py`. Kept free of `litellm` imports so any proxy module can import them without joining the `litellm.proxy` import cycle."""
class UserNotFoundError(ValueError):
"""The user row is provably absent, as opposed to merely unreadable, so a caller that reads a missing row as no user-level limits can key on it without also swallowing a database that would not answer."""
def __init__(self, user_id: str) -> None:
super().__init__(f"User doesn't exist in db. 'user_id'={user_id}. Create user via `/user/new` call.")

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

@ -3937,6 +3937,7 @@ async def test_list_team_v2_org_admin_sees_org_teams():
mock_db.litellm_teamtable.find_many = AsyncMock(return_value=[mock_team])
mock_db.litellm_teamtable.count = AsyncMock(return_value=1)
mock_db.litellm_verificationtoken.group_by = AsyncMock(return_value=[])
mock_db.litellm_usertable.find_unique = AsyncMock(return_value=mock_user)
result = await list_team_v2(
http_request=mock_request,
@ -4036,6 +4037,7 @@ async def test_list_team_v2_org_admin_own_user_id_sees_all_org_teams():
)
mock_db.litellm_teamtable.count = AsyncMock(return_value=2)
mock_db.litellm_verificationtoken.group_by = AsyncMock(return_value=[])
mock_db.litellm_usertable.find_unique = AsyncMock(return_value=mock_user)
# UI sends the caller's own user_id for non-Admin roles
result = await list_team_v2(
@ -4055,10 +4057,217 @@ 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
def _team_where_matches(team, where) -> bool:
for key, cond in where.items():
if key == "AND":
if not all(_team_where_matches(team, c) for c in cond):
return False
elif key == "OR":
if not any(_team_where_matches(team, c) for c in cond):
return False
else:
value = getattr(team, key)
if not isinstance(cond, dict):
if value != cond:
return False
elif "in" in cond and value not in cond["in"]:
return False
elif "contains" in cond and cond["contains"].lower() not in (value or "").lower():
return False
return True
def _org_membership(user_id: str, organization_id: str, user_role: str) -> LiteLLM_OrganizationMembershipTable:
return LiteLLM_OrganizationMembershipTable(
user_id=user_id,
organization_id=organization_id,
user_role=user_role,
spend=0.0,
created_at=datetime.now(),
updated_at=datetime.now(),
)
@pytest.mark.asyncio
async def test_list_team_v2_org_admin_own_query_keeps_memberships_in_other_orgs(monkeypatch):
"""
/v2/team/list: an org admin of org_A who is a member of a team in org_B
gets that team back on a self query (with and without user_id, with and
without search), alongside every org_A team. The membership half of the
union comes from the DB, so a stale cached user object cannot hide it.
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 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"],
organization_memberships=[
_org_membership("org_admin_user", "org_A", "org_admin"),
_org_membership("org_admin_user", "org_B", "internal_user"),
],
),
model_type=LiteLLM_UserTable,
)
await cache.async_set_cache(
key="other_user",
value=LiteLLM_UserTable(
user_id="other_user",
teams=["other_team_in_org_A", "team_in_org_B", "unrelated_team_in_org_B"],
organization_memberships=[_org_membership("other_user", "org_B", "internal_user")],
),
model_type=LiteLLM_UserTable,
)
def team(team_id, organization_id, *member_ids):
return LiteLLM_TeamTable(
team_id=team_id,
team_alias=team_id,
organization_id=organization_id,
members_with_roles=[Member(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):
return [t for t in all_teams if where is None or _team_where_matches(t, where)]
async def count(where=None, **kwargs):
return len(await find_many(where))
prisma_client = MagicMock()
prisma_client.db.litellm_teamtable.find_many = AsyncMock(side_effect=find_many)
prisma_client.db.litellm_teamtable.count = AsyncMock(side_effect=count)
prisma_client.db.litellm_verificationtoken.group_by = AsyncMock(return_value=[])
prisma_client.db.litellm_usertable.find_unique = AsyncMock(
return_value=LiteLLM_UserTable(
user_id="org_admin_user",
teams=["team_in_org_A", "team_in_org_B"],
organization_memberships=[
_org_membership("org_admin_user", "org_A", "org_admin"),
_org_membership("org_admin_user", "org_B", "internal_user"),
],
)
)
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)
async def list_teams(user_id, search=None):
result = await list_team_v2(
http_request=MagicMock(),
user_id=user_id,
organization_id=None,
team_id=None,
team_alias=None,
search=search,
user_api_key_dict=org_admin,
page=1,
page_size=10,
sort_by=None,
sort_order="asc",
status=None,
)
assert result["total"] == len(result["teams"])
return [t.team_id for t in result["teams"]]
own_view = ["team_in_org_A", "other_team_in_org_A", "team_in_org_B"]
assert await list_teams("org_admin_user") == own_view
assert await list_teams(None) == own_view
assert await list_teams("org_admin_user", search="team_in_org_B") == ["team_in_org_B"]
assert await list_teams("other_user") == ["other_team_in_org_A"]
prisma_client.db.litellm_usertable.find_unique.assert_awaited_with(
where={"user_id": "org_admin_user"}, include={"organization_memberships": True}
)
prisma_client.db.litellm_usertable.find_unique.side_effect = RuntimeError("db down")
with pytest.raises(ValueError, match="db down"):
await list_teams("org_admin_user")
@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
they belong to, including the org_B one. The bare admin listing stays the
org_A view and 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=[_org_membership("org_admin_user", "org_A", "org_admin")],
),
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", "team_in_org_B"]
assert await list_teams(None) == ["team_in_org_A", "other_team_in_org_A"]
assert await list_teams("other_user") == ["other_team_in_org_A"]
@pytest.mark.asyncio