scriptoriumadmin/litellm
1
0
Fork 0
mirror of https://github.com/BerriAI/litellm.git synced 2026-09-07 08:26:10 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 36

Merge 1660a8904c into a2c814654e

Browse source
This commit is contained in:
michelligabriele 2026-08-27 17:18:27 -05:00 • committed by GitHub
commit cb0f3db238
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
3 changed files with 402 additions and 26 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

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

@ -4787,6 +4787,24 @@ async def _get_org_admin_org_ids(
return org_ids if org_ids else None
def _org_admin_membership_union_conditions(
org_admin_org_ids: list[str], # mutable-ok: passed straight into a Prisma `in` filter
user_team_ids: list[str], # mutable-ok: passed straight into a Prisma `in` filter
) -> list[dict[str, object]]: # mutable-ok: Prisma OR clauses are lists of dict-shaped filters
"""(teams in the admin's orgs) OR (teams the caller is a member of).
Built for a self/bare org-admin query so teams in orgs where the caller is
only a member stay visible (LIT-3723). An org admin with no memberships
still matches their org teams, so this never collapses to an empty result.
"""
union_or: Final[list[dict[str, object]]] = [ # mutable-ok: Prisma query filters are dict-shaped
{"organization_id": {"in": org_admin_org_ids}} # mutable-ok: Prisma query filters are dict-shaped
]
if user_team_ids:
union_or.append({"team_id": {"in": user_team_ids}}) # mutable-ok: Prisma query filters are dict-shaped
return union_or
async def _build_team_list_where_conditions(
prisma_client: PrismaClient,
team_id: str | None,
@ -4797,6 +4815,7 @@ 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,
union_org_admin_membership: bool = False,
user_api_key_cache: UserApiKeyCache | None = None,
proxy_logging_obj: ProxyLogging | None = None,
) -> dict[str, object] | None:
@ -4823,10 +4842,23 @@ async def _build_team_list_where_conditions(
{"team_alias": {"contains": search, "mode": "insensitive"}},
]
# Self/bare org-admin query: union org teams with the caller's own
# memberships (LIT-3723) instead of intersecting. Gated off when an
# explicit org filter is supplied (it takes precedence) or for the
# deleted-table path (keeps current behavior).
union_org_ids: Final = (
org_admin_org_ids
if union_org_admin_membership
and org_admin_org_ids is not None
and not organization_id
and not use_deleted_table
else None
)
if organization_id:
where_conditions["organization_id"] = organization_id
elif org_admin_org_ids is not None:
# Org admin: always scope to their orgs, even when filtering by user_id.
elif org_admin_org_ids is not None and union_org_ids is None:
# Org admin cross-user query: scope strictly to their orgs.
where_conditions["organization_id"] = {"in": org_admin_org_ids}
if user_id:
@ -4852,6 +4884,20 @@ async def _build_team_list_where_conditions(
if use_deleted_table:
where_conditions["members"] = {"has": user_id}
elif union_org_ids is not None:
# Self/bare org admin: (teams in my orgs) OR (teams I'm a member
# of), so teams in orgs where I'm only a member stay visible
# (LIT-3723). An org admin with no memberships must still see org
# teams, so do NOT early-return None here.
union_or: Final = _org_admin_membership_union_conditions(union_org_ids, user_team_ids)
search_or: Final = where_conditions.pop("OR", None) # set above from `search`
if search_or is not None:
where_conditions["AND"] = [ # mutable-ok: Prisma query filters are dict-shaped
{"OR": search_or}, # mutable-ok: Prisma query filters are dict-shaped
{"OR": union_or}, # mutable-ok: Prisma query filters are dict-shaped
]
else:
where_conditions["OR"] = union_or
else:
# When user_id is provided, filter by that user's direct team
# memberships. For org admins the access control gate in
@ -4956,20 +5002,24 @@ 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, bool]: # mutable-ok: org-id list flows into a Prisma `in` filter
"""Enforce access control for list_team_v2.
- Proxy admins and admin viewers can query any teams.
- Org admins can query teams within their organizations.
- 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
union_org_admin_membership — a flag set for a self/bare org-admin query
that tells the where-builder to union org teams with the caller's own
memberships instead of intersecting them.
"""
is_proxy_admin: Final = _user_has_admin_view(user_api_key_dict)
org_admin_org_ids: list[str] | None = None
union_org_admin_membership = False # rebind-ok: set when the caller is a self/bare org-admin query
if is_proxy_admin:
return user_id, org_admin_org_ids
return user_id, org_admin_org_ids, union_org_admin_membership
# Always check org admin status so that even own-queries see
# the full set of organisation teams, not just direct memberships.
@ -4988,14 +5038,17 @@ async def _enforce_list_team_v2_access(
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.
# Self/bare org-admin query: union the admin's org teams with the
# caller's own memberships, so teams in orgs where they are only a
# member (not org admin) stay visible (LIT-3723). Keep user_id =
# caller for the membership lookup instead of nulling it; the
# where-builder unions on the flag. Keep user_id (and leave the flag
# False) when the org admin explicitly queries a *different* user's
# teams — that path stays org-scoped (intersection), the #25904
# cross-user boundary.
if user_id is None or user_id == user_api_key_dict.user_id:
user_id = None
user_id = user_api_key_dict.user_id
union_org_admin_membership = True
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,
@ -5015,7 +5068,7 @@ async def _enforce_list_team_v2_access(
if user_id is None:
user_id = user_api_key_dict.user_id
return user_id, org_admin_org_ids
return user_id, org_admin_org_ids, union_org_admin_membership
@router.get(
@ -5093,7 +5146,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, union_org_admin_membership = await _enforce_list_team_v2_access(
user_api_key_dict=user_api_key_dict,
user_id=user_id,
organization_id=organization_id,
@ -5125,6 +5178,7 @@ async def list_team_v2(
search=search,
search_team_id_match=search_team_id_match,
org_admin_org_ids=org_admin_org_ids,
union_org_admin_membership=union_org_admin_membership,
user_api_key_cache=user_api_key_cache,
proxy_logging_obj=proxy_logging_obj,
)
@ -5223,13 +5277,14 @@ async def _authorize_and_filter_teams(
- Others: 401.
"""
is_proxy_admin: Final = _user_has_admin_view(user_api_key_dict)
# Bound unconditionally: the org-admin branch below reads it outside the
# `not is_proxy_admin` guard.
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(
@ -5256,22 +5311,28 @@ async def _authorize_and_filter_teams(
},
)
if allowed_org_ids is not None:
# Org admin: query DB for teams in their orgs
# An org admin's *own* query falls through to the membership path below:
# org-admin status must not clamp a user's own team list to their admin
# orgs, or teams in orgs where they are only a member disappear (LIT-3723).
is_org_admin_self_query: Final = allowed_org_ids is not None and bool(user_id) and is_own_query
if allowed_org_ids is not None and not is_org_admin_self_query:
# Org admin bare or cross-user query: query DB for teams in their orgs
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
# Cross-user query: keep the org boundary (intersection) — #25904.
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)
# Regular user (or org admin querying themselves): 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}
)

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

@ -3811,6 +3811,7 @@ async def test_list_team_v2_org_admin_sees_org_teams():
organization_id=None,
team_id=None,
team_alias=None,
search=None,
user_api_key_dict=mock_user_api_key_dict,
page=1,
page_size=10,
@ -3823,9 +3824,12 @@ async def test_list_team_v2_org_admin_sees_org_teams():
assert len(result["teams"]) == 1
assert result["teams"][0].members_count == 1
# Verify org-scoped where clause
# Bare org-admin query is scoped to the admin's org via the union
# branch (LIT-3723). With no direct memberships the union collapses to
# the org-scope branch only.
where = mock_db.litellm_teamtable.find_many.call_args.kwargs["where"]
assert where["organization_id"] == {"in": ["org_A"]}
assert where["OR"] == [{"organization_id": {"in": ["org_A"]}}]
assert "organization_id" not in where
@pytest.mark.asyncio
@ -3911,6 +3915,7 @@ async def test_list_team_v2_org_admin_own_user_id_sees_all_org_teams():
organization_id=None,
team_id=None,
team_alias=None,
search=None,
user_api_key_dict=mock_user_api_key_dict,
page=1,
page_size=10,
@ -3922,9 +3927,18 @@ 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
# Self/bare org-admin query unions org-scoped teams with the caller's
# own memberships (LIT-3723). The org-scope branch still guarantees the
# admin sees *all* teams in their org (regression #30215); the team_id
# branch only widens the result and never clamps it. No top-level
# organization_id/team_id key that could hide an org team the admin
# isn't a direct member of.
where = mock_db.litellm_teamtable.find_many.call_args.kwargs["where"]
assert where["organization_id"] == {"in": ["org_A"]}
assert where["OR"] == [
{"organization_id": {"in": ["org_A"]}},
{"team_id": {"in": ["team_1"]}},
]
assert "organization_id" not in where
assert "team_id" not in where

301
tests/test_litellm/proxy/management_endpoints/test_team_list_org_admin_scope.py Normal file
View file

@ -0,0 +1,301 @@
"""
Tests for org-admin team-list scoping (LIT-3723).
A user who is org admin of one org must still see teams they belong to in
*other* orgs (where they are only a member). PR #25904 clamped a self-query to
the caller's org-admin orgs, dropping those memberships. The fix is strictly
additive: a self/bare org-admin query unions org teams with the caller's own
memberships, while a cross-user org-admin query keeps the org-boundary
intersection (#25904).
These tests are self-contained — they mock `get_user_object`, `TeamRepository`,
and the access-gate helpers rather than touching the shared seed fixture in
tests/proxy_behavior/management/ (which ~8 suites parametrize over).
"""
import os
import sys
from datetime import datetime, timezone
from unittest.mock import AsyncMock, MagicMock, patch
import pytest
sys.path.insert(0, os.path.abspath("../../../"))
from litellm.proxy._types import (
LiteLLM_OrganizationMembershipTable,
LiteLLM_UserTable,
LitellmUserRoles,
UserAPIKeyAuth,
)
_MOD = "litellm.proxy.management_endpoints.team_endpoints"
_NOW = datetime.now(timezone.utc)
# ---------------------------------------------------------------------------
# Fixtures / helpers
# ---------------------------------------------------------------------------
class _FakeTeam:
"""Lightweight stand-in for a Prisma team row.
`members_with_roles` is a list of plain dicts so `m.get("user_id")` (the
production filter) works without pydantic coercion.
"""
def __init__(self, team_id, organization_id, member_user_ids):
self.team_id = team_id
self.organization_id = organization_id
self.members_with_roles = [{"user_id": uid, "role": "user"} for uid in member_user_ids]
def _user_with_teams(user_id="x", teams=None) -> LiteLLM_UserTable:
return LiteLLM_UserTable(user_id=user_id, teams=teams or [])
def _org_admin_caller(user_id="x", org_id="org-a") -> LiteLLM_UserTable:
return LiteLLM_UserTable(
user_id=user_id,
organization_memberships=[
LiteLLM_OrganizationMembershipTable(
user_id=user_id,
organization_id=org_id,
user_role=LitellmUserRoles.ORG_ADMIN.value,
created_at=_NOW,
updated_at=_NOW,
)
],
)
def _internal_key(user_id="x") -> UserAPIKeyAuth:
return UserAPIKeyAuth(user_id=user_id, user_role=LitellmUserRoles.INTERNAL_USER.value)
def _make_team_repo(all_teams):
"""Return a TeamRepository factory whose `.table.find_many` honors `where`.
No `where` (self-query full scan) returns every team; an
`organization_id: {in: [...]}` filter returns only teams in those orgs.
"""
async def _find_many(*args, **kwargs):
where = kwargs.get("where")
if not where:
return list(all_teams)
org_ids = where.get("organization_id", {}).get("in", [])
return [t for t in all_teams if t.organization_id in org_ids]
instance = MagicMock()
instance.table.find_many = AsyncMock(side_effect=_find_many)
return MagicMock(return_value=instance)
async def _build_where(**overrides):
from litellm.proxy.management_endpoints.team_endpoints import (
_build_team_list_where_conditions,
)
kwargs = dict(
prisma_client=MagicMock(),
team_id=None,
team_alias=None,
organization_id=None,
user_id="x",
use_deleted_table=False,
search=None,
org_admin_org_ids=["org-a"],
union_org_admin_membership=True,
user_api_key_cache=MagicMock(),
proxy_logging_obj=MagicMock(),
)
kwargs.update(overrides)
return await _build_team_list_where_conditions(**kwargs)
# ---------------------------------------------------------------------------
# _build_team_list_where_conditions — union semantics
# ---------------------------------------------------------------------------
class TestBuildWhereOrgAdminUnion:
@pytest.mark.asyncio
async def test_self_query_emits_union(self):
"""Self/bare org admin → OR(org teams, memberships), no sibling org clamp."""
user = _user_with_teams("x", ["team-2"]) # team-2 lives in org-b
with patch(f"{_MOD}.get_user_object", new_callable=AsyncMock, return_value=user):
where = await _build_where()
assert where is not None
assert where["OR"] == [
{"organization_id": {"in": ["org-a"]}},
{"team_id": {"in": ["team-2"]}},
]
# No sibling organization_id clamp — that would re-AND the intersection.
assert "organization_id" not in where
@pytest.mark.asyncio
async def test_self_query_no_memberships_keeps_org_teams(self):
"""Org admin with no memberships still sees org teams; does NOT return None."""
user = _user_with_teams("x", [])
with patch(f"{_MOD}.get_user_object", new_callable=AsyncMock, return_value=user):
where = await _build_where()
assert where is not None
assert where["OR"] == [{"organization_id": {"in": ["org-a"]}}]
@pytest.mark.asyncio
async def test_cross_user_query_keeps_intersection(self):
"""union flag off → sibling org clamp AND membership filter (the #25904 boundary)."""
target = _user_with_teams("target", ["team-2"])
with patch(f"{_MOD}.get_user_object", new_callable=AsyncMock, return_value=target):
where = await _build_where(user_id="target", union_org_admin_membership=False)
assert where is not None
assert where["organization_id"] == {"in": ["org-a"]}
assert where["team_id"] == {"in": ["team-2"]}
assert "OR" not in where
@pytest.mark.asyncio
async def test_search_co_exists_with_union(self):
"""search OR and union OR are both preserved, ANDed at the top level."""
user = _user_with_teams("x", ["team-2"])
with patch(f"{_MOD}.get_user_object", new_callable=AsyncMock, return_value=user):
where = await _build_where(search="foo")
assert where is not None
assert where["AND"] == [
{
"OR": [
{"team_id": "foo"},
{"team_alias": {"contains": "foo", "mode": "insensitive"}},
]
},
{
"OR": [
{"organization_id": {"in": ["org-a"]}},
{"team_id": {"in": ["team-2"]}},
]
},
]
assert "OR" not in where
# ---------------------------------------------------------------------------
# _enforce_list_team_v2_access — flag + boundary gates
# ---------------------------------------------------------------------------
class TestEnforceListTeamV2Access:
async def _call(self, user_id, organization_id=None, org_admin_org_ids=["org-a"]):
from litellm.proxy.management_endpoints.team_endpoints import (
_enforce_list_team_v2_access,
)
with patch(
f"{_MOD}._get_org_admin_org_ids",
new_callable=AsyncMock,
return_value=org_admin_org_ids,
):
return await _enforce_list_team_v2_access(
user_api_key_dict=_internal_key("x"),
user_id=user_id,
organization_id=organization_id,
prisma_client=MagicMock(),
user_api_key_cache=MagicMock(),
proxy_logging_obj=MagicMock(),
)
@pytest.mark.asyncio
async def test_bare_self_query_sets_flag(self):
uid, orgs, union = await self._call(user_id=None)
assert (uid, orgs, union) == ("x", ["org-a"], True)
@pytest.mark.asyncio
async def test_explicit_self_query_sets_flag(self):
uid, orgs, union = await self._call(user_id="x")
assert (uid, orgs, union) == ("x", ["org-a"], True)
@pytest.mark.asyncio
async def test_cross_user_query_clears_flag(self):
uid, orgs, union = await self._call(user_id="other")
assert (uid, orgs, union) == ("other", ["org-a"], False)
@pytest.mark.asyncio
async def test_foreign_org_filter_is_403(self):
from fastapi import HTTPException
with pytest.raises(HTTPException) as exc:
await self._call(user_id=None, organization_id="org-z")
assert exc.value.status_code == 403
@pytest.mark.asyncio
async def test_non_org_admin_bare_is_401(self):
from fastapi import HTTPException
with patch(f"{_MOD}.allowed_route_check_inside_route", return_value=False):
with pytest.raises(HTTPException) as exc:
await self._call(user_id=None, org_admin_org_ids=None)
assert exc.value.status_code == 401
# ---------------------------------------------------------------------------
# _authorize_and_filter_teams (v1) — self-query union vs cross-user intersection
# ---------------------------------------------------------------------------
class TestAuthorizeAndFilterTeams:
async def _call(self, user_id, all_teams):
from litellm.proxy.management_endpoints.team_endpoints import (
_authorize_and_filter_teams,
)
caller = _org_admin_caller("x", "org-a")
with (
patch(f"{_MOD}.get_user_object", new_callable=AsyncMock, return_value=caller),
patch(f"{_MOD}.TeamRepository", _make_team_repo(all_teams)),
):
return await _authorize_and_filter_teams(
user_api_key_dict=_internal_key("x"),
user_id=user_id,
prisma_client=MagicMock(),
user_api_key_cache=MagicMock(),
proxy_logging_obj=MagicMock(),
)
@pytest.mark.asyncio
async def test_self_query_includes_cross_org_membership(self):
"""Org admin of org-a, member of a team in org-b, self-query → both teams."""
team_a = _FakeTeam("team-1", "org-a", member_user_ids=["x"])
team_b = _FakeTeam("team-2", "org-b", member_user_ids=["x"])
other = _FakeTeam("team-3", "org-c", member_user_ids=["someone-else"])
result = await self._call(user_id="x", all_teams=[team_a, team_b, other])
ids = {t.team_id for t in result}
assert ids == {"team-1", "team-2"} # org-b membership survives
@pytest.mark.asyncio
async def test_cross_user_query_keeps_org_boundary(self):
"""Cross-user query → only org-a teams the target is a member of."""
team_a = _FakeTeam("team-1", "org-a", member_user_ids=["target"])
team_b = _FakeTeam("team-2", "org-b", member_user_ids=["target"])
result = await self._call(user_id="target", all_teams=[team_a, team_b])
ids = {t.team_id for t in result}
assert ids == {"team-1"} # team-2 (org-b) is NOT leaked
@pytest.mark.asyncio
async def test_bare_query_returns_all_org_teams(self):
"""Bare org-admin query (no user_id) → every team in the admin's orgs."""
team_a1 = _FakeTeam("team-1", "org-a", member_user_ids=["someone"])
team_a2 = _FakeTeam("team-2", "org-a", member_user_ids=[])
team_b = _FakeTeam("team-3", "org-b", member_user_ids=["x"])
result = await self._call(user_id=None, all_teams=[team_a1, team_a2, team_b])
ids = {t.team_id for t in result}
assert ids == {"team-1", "team-2"} # all org-a teams, no org-b