mirror of
https://github.com/BerriAI/litellm.git
synced 2026-10-05 02:41:56 +00:00
Merge 7c0e586d78 into 426e675c49
This commit is contained in:
commit
5b6c83029c
4 changed files with 90 additions and 8 deletions
|
|
@ -43,6 +43,7 @@ if TYPE_CHECKING:
|
|||
router = APIRouter()
|
||||
|
||||
_OBJECT_PERMISSION_PAYLOAD: Final = TypeAdapter(dict[str, object])
|
||||
_TEAM_MEMBERS: Final = TypeAdapter(list[Member])
|
||||
|
||||
|
||||
def _team_table(prisma_client: PrismaClient) -> TableActions["prisma_models.LiteLLM_TeamTable"]:
|
||||
|
|
@ -105,14 +106,21 @@ async def _check_user_permission_for_project(
|
|||
if not team_id or not user_api_key_dict.user_id:
|
||||
return False
|
||||
|
||||
team = team_object
|
||||
if team is None:
|
||||
team = await _team_table(prisma_client).find_unique(where={"team_id": team_id})
|
||||
team_row: Final = (
|
||||
team_object
|
||||
if team_object is not None
|
||||
else await _team_table(prisma_client).find_unique(where={"team_id": team_id})
|
||||
)
|
||||
if team_row is None:
|
||||
return False
|
||||
|
||||
if team and team.admins:
|
||||
return user_api_key_dict.user_id in team.admins
|
||||
|
||||
return False
|
||||
raw_members: Final = team_row.members_with_roles
|
||||
members: Final = _TEAM_MEMBERS.validate_python(raw_members) if isinstance(raw_members, list) else ()
|
||||
is_role_admin: Final = any(
|
||||
member.user_id is not None and member.user_id == user_api_key_dict.user_id and member.role == "admin"
|
||||
for member in members
|
||||
)
|
||||
return is_role_admin or user_api_key_dict.user_id in (team_row.admins or [])
|
||||
|
||||
|
||||
async def _validate_team_exists(
|
||||
|
|
|
|||
|
|
@ -868,9 +868,12 @@ class LiteLLMRoutes(enum.Enum):
|
|||
"/prompt/list",
|
||||
"/prompt/info",
|
||||
"/vector_store/info",
|
||||
# Project read routes - endpoint scopes results to caller's teams (non-admin)
|
||||
# Project routes - reads scope results to caller's teams; /new and
|
||||
# /update require proxy admin or admin of the project's team in the endpoint
|
||||
"/project/list",
|
||||
"/project/info",
|
||||
"/project/new",
|
||||
"/project/update",
|
||||
# Endpoint enforces proxy-admin vs team-admin model access itself.
|
||||
"/health/test_connection",
|
||||
# Invitation routes - org/team admins checked in endpoint via _user_has_admin_privileges
|
||||
|
|
|
|||
|
|
@ -1383,3 +1383,33 @@ async def test_new_project_flag_on_access_group_model_returns_400(monkeypatch):
|
|||
|
||||
assert "prod-models" in str(exc_info.value)
|
||||
assert "expand to multiple models at request time" in str(exc_info.value)
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_check_user_permission_for_project_uses_members_with_roles():
|
||||
"""Team admins added via /team/member_add live in members_with_roles, not the legacy admins list."""
|
||||
from litellm.proxy._types import LiteLLM_TeamTable, Member
|
||||
from litellm_enterprise.proxy.management_endpoints.project_endpoints import (
|
||||
_check_user_permission_for_project,
|
||||
)
|
||||
|
||||
team = LiteLLM_TeamTable(
|
||||
team_id="team-1",
|
||||
admins=[],
|
||||
members_with_roles=[
|
||||
Member(user_id="team-admin", role="admin"),
|
||||
Member(user_id="plain-member", role="user"),
|
||||
],
|
||||
)
|
||||
|
||||
async def check(user_id: str) -> bool:
|
||||
return await _check_user_permission_for_project(
|
||||
user_api_key_dict=UserAPIKeyAuth(user_id=user_id, user_role=LitellmUserRoles.INTERNAL_USER),
|
||||
team_id="team-1",
|
||||
prisma_client=mock.MagicMock(),
|
||||
team_object=team,
|
||||
)
|
||||
|
||||
assert await check("team-admin") is True
|
||||
assert await check("plain-member") is False
|
||||
assert await check("stranger") is False
|
||||
|
|
|
|||
|
|
@ -3714,6 +3714,47 @@ def test_agent_registry_route_gate_open_to_non_admin_roles(user_role, method, ro
|
|||
valid_token=valid_token,
|
||||
request_data={},
|
||||
)
|
||||
|
||||
|
||||
@pytest.mark.parametrize("route", ["/project/new", "/project/update"])
|
||||
def test_project_write_routes_reach_endpoint_for_internal_user(route):
|
||||
"""A team admin is an internal_user at the route gate. /project/new and
|
||||
/project/update must pass it so the endpoint can apply its own proxy-admin
|
||||
or team-admin check instead of the gate 403ing every non-proxy-admin."""
|
||||
|
||||
valid_token = UserAPIKeyAuth(user_id="test_user", user_role=LitellmUserRoles.INTERNAL_USER.value)
|
||||
request = MagicMock(spec=Request)
|
||||
request.method = "POST"
|
||||
request.query_params = {}
|
||||
|
||||
RouteChecks.non_proxy_admin_allowed_routes_check(
|
||||
user_obj=LiteLLM_UserTable(user_id="test_user", user_role=LitellmUserRoles.INTERNAL_USER.value),
|
||||
_user_role=LitellmUserRoles.INTERNAL_USER.value,
|
||||
route=route,
|
||||
request=request,
|
||||
valid_token=valid_token,
|
||||
request_data={"team_id": "team-1"},
|
||||
)
|
||||
assert RouteChecks.check_route_access(route=route, allowed_routes=LiteLLMRoutes.self_managed_routes.value)
|
||||
|
||||
|
||||
def test_project_delete_route_stays_proxy_admin_only():
|
||||
valid_token = UserAPIKeyAuth(user_id="test_user", user_role=LitellmUserRoles.INTERNAL_USER.value)
|
||||
request = MagicMock(spec=Request)
|
||||
request.method = "DELETE"
|
||||
request.query_params = {}
|
||||
|
||||
with pytest.raises(Exception, match="Only proxy admin can be used"):
|
||||
RouteChecks.non_proxy_admin_allowed_routes_check(
|
||||
user_obj=LiteLLM_UserTable(user_id="test_user", user_role=LitellmUserRoles.INTERNAL_USER.value),
|
||||
_user_role=LitellmUserRoles.INTERNAL_USER.value,
|
||||
route="/project/delete",
|
||||
request=request,
|
||||
valid_token=valid_token,
|
||||
request_data={},
|
||||
)
|
||||
|
||||
|
||||
TEAM_CALLBACK_ROUTES = (
|
||||
"/team/06bda574-5ca9-43d3-beb8-3b23c2f17112/callback",
|
||||
"/team/06bda574-5ca9-43d3-beb8-3b23c2f17112/callback/langfuse",
|
||||
|
|
|
|||
Loading…
Add table
Reference in a new issue