test(proxy): route gate + update_team for org-admin team writes (#27294)

- Remove inline comments from self_managed_routes team entries.
- Assert /team/update|delete|block|unblock pass non_proxy_admin gate without organization_id.
- Assert update_team succeeds for org admin with team_id only; 403 when org-admin check fails.

Co-authored-by: Cursor <cursoragent@cursor.com>
This commit is contained in:
Milan 2026-05-08 20:11:42 +03:00
parent 8e7178aaf8
commit b08515d7a3
No known key found for this signature in database
3 changed files with 171 additions and 4 deletions

View file

@ -691,10 +691,6 @@ class LiteLLMRoutes(enum.Enum):
"/team/member_add",
"/team/member_delete",
"/team/member_update",
# Team lifecycle writes: handlers call _verify_team_access (proxy admin,
# team admin, or org admin of the team's org). Route-level org-admin
# detection only inspects organization_id in the body, which these
# payloads often omit, so they must be self-managed like member_*.
"/team/update",
"/team/delete",
"/team/block",

View file

@ -1949,6 +1949,43 @@ def test_available_roles_accessible_to_non_admin_users(user_role):
)
@pytest.mark.parametrize(
"route,request_data",
[
("/team/update", {"team_id": "team-abc"}),
("/team/delete", {"team_ids": ["team-abc"]}),
("/team/block", {"team_id": "team-abc"}),
("/team/unblock", {"team_id": "team-abc"}),
],
)
def test_team_write_routes_pass_route_gate_without_organization_id(
route, request_data
):
"""
Team lifecycle routes are self-managed: the route gate must not require
organization_id in the body (org-admin detection uses that field; handlers
enforce _verify_team_access). Regression: GitHub #27294.
"""
role = LitellmUserRoles.INTERNAL_USER.value
user_obj = LiteLLM_UserTable(
user_id="route_gate_user",
user_email="rg@example.com",
user_role=role,
)
valid_token = UserAPIKeyAuth(user_id="route_gate_user", user_role=role)
request = MagicMock(spec=Request)
request.query_params = {}
RouteChecks.non_proxy_admin_allowed_routes_check(
user_obj=user_obj,
_user_role=role,
route=route,
request=request,
valid_token=valid_token,
request_data=request_data,
)
# ── _user_is_org_admin tests ──────────────────────────────────────────────────
from datetime import datetime

View file

@ -7532,6 +7532,140 @@ async def test_update_team_rejects_unauthorized_caller():
assert exc_info.value.code == "403"
@pytest.mark.asyncio
async def test_update_team_org_admin_succeeds_with_team_id_only():
"""
Org admin of the team's org may call /team/update with only team_id in the
body once the route gate allows the request (self_managed_routes). Handler
grants access when _is_user_org_admin_for_team is true.
"""
from unittest.mock import Mock
from fastapi import Request
from litellm.proxy._types import UpdateTeamRequest
mock_request = Mock(spec=Request)
caller = UserAPIKeyAuth(
user_role=LitellmUserRoles.INTERNAL_USER,
user_id="org-admin-team-id-only",
)
update_request = UpdateTeamRequest(team_id="tid-org-admin-only", max_budget=10.0)
mock_existing_team = MagicMock()
mock_existing_team.model_dump.return_value = {
"team_id": "tid-org-admin-only",
"team_alias": "t",
"members_with_roles": [{"user_id": "other_user", "role": "admin"}],
"organization_id": "org-scope-1",
"max_budget": 5.0,
"soft_budget": None,
"model_id": None,
"object_permission_id": None,
}
mock_updated = MagicMock()
mock_updated.team_id = "tid-org-admin-only"
mock_updated.organization_id = "org-scope-1"
mock_updated.max_budget = 10.0
mock_updated.litellm_model_table = None
mock_updated.model_dump.return_value = {
"team_id": "tid-org-admin-only",
"organization_id": "org-scope-1",
"max_budget": 10.0,
}
with (
patch("litellm.proxy.proxy_server.prisma_client") as mock_prisma,
patch("litellm.proxy.proxy_server.llm_router"),
patch("litellm.proxy.proxy_server.user_api_key_cache") as mock_cache,
patch("litellm.proxy.proxy_server.proxy_logging_obj"),
patch("litellm.proxy.proxy_server.litellm_proxy_admin_name", "admin"),
patch(
"litellm.proxy.management_endpoints.team_endpoints._is_user_org_admin_for_team",
new_callable=AsyncMock,
return_value=True,
),
patch(
"litellm.proxy.management_endpoints.team_endpoints.get_org_object",
new_callable=AsyncMock,
return_value=None,
),
patch(
"litellm.proxy.auth.auth_checks._cache_team_object",
new_callable=AsyncMock,
),
patch(
"litellm.proxy.proxy_server.create_audit_log_for_update", new_callable=AsyncMock
),
):
mock_prisma.db.litellm_teamtable.find_unique = AsyncMock(
return_value=mock_existing_team
)
mock_prisma.db.litellm_teamtable.update = AsyncMock(return_value=mock_updated)
mock_prisma.jsonify_team_object = lambda db_data: db_data
mock_cache.async_get_cache = AsyncMock(return_value=None)
mock_cache.async_set_cache = AsyncMock()
result = await update_team(
data=update_request,
http_request=mock_request,
user_api_key_dict=caller,
)
assert result is not None
assert result["data"].max_budget == 10.0
@pytest.mark.asyncio
async def test_update_team_org_admin_wrong_org_still_forbidden():
"""Org admin path in handler still rejects teams outside the admin's org."""
from unittest.mock import Mock
from fastapi import Request
from litellm.proxy._types import UpdateTeamRequest
mock_request = Mock(spec=Request)
caller = UserAPIKeyAuth(
user_role=LitellmUserRoles.INTERNAL_USER,
user_id="org-admin-wrong-org",
)
update_request = UpdateTeamRequest(team_id="tid-other-org", max_budget=10.0)
mock_existing_team = MagicMock()
mock_existing_team.model_dump.return_value = {
"team_id": "tid-other-org",
"team_alias": "other",
"members_with_roles": [{"user_id": "other_user", "role": "admin"}],
"organization_id": "org-b",
}
with (
patch("litellm.proxy.proxy_server.prisma_client") as mock_prisma_client,
patch("litellm.proxy.proxy_server.llm_router"),
patch("litellm.proxy.proxy_server.user_api_key_cache"),
patch("litellm.proxy.proxy_server.proxy_logging_obj"),
patch("litellm.proxy.proxy_server.litellm_proxy_admin_name", "admin"),
patch(
"litellm.proxy.management_endpoints.team_endpoints._is_user_org_admin_for_team",
new_callable=AsyncMock,
return_value=False,
),
):
mock_prisma_client.db.litellm_teamtable.find_unique = AsyncMock(
return_value=mock_existing_team
)
with pytest.raises(ProxyException) as exc_info:
await update_team(
data=update_request,
http_request=mock_request,
user_api_key_dict=caller,
)
assert exc_info.value.code == "403"
# ----- /team/{team_id}/members/me -----