[Fix] Address Greptile review: POST /organization/info auth bypass, inline imports, team access denial tests

- Add _verify_org_access to deprecated POST /organization/info endpoint
- Move get_user_object to module-level import in organization_endpoints.py
- Add tests for _verify_team_access 403 denial path
This commit is contained in:
Yuneng Jiang 2026-04-11 12:40:55 -07:00
parent 57b77fecbd
commit 218daca867
No known key found for this signature in database
2 changed files with 107 additions and 3 deletions

View file

@ -19,7 +19,7 @@ from fastapi import APIRouter, Depends, HTTPException, Request, status
from litellm._logging import verbose_proxy_logger
from litellm._uuid import uuid
from litellm.proxy._types import *
from litellm.proxy.auth.auth_checks import can_user_call_model
from litellm.proxy.auth.auth_checks import can_user_call_model, get_user_object
from litellm.proxy.auth.user_api_key_auth import user_api_key_auth
from litellm.proxy.management_endpoints.budget_management_endpoints import (
new_budget,
@ -65,7 +65,6 @@ async def _verify_org_access(
detail="You do not have access to this organization",
)
from litellm.proxy.auth.auth_checks import get_user_object
from litellm.proxy.proxy_server import proxy_logging_obj, user_api_key_cache
caller_user = await get_user_object(
@ -815,7 +814,10 @@ async def info_organization(
tags=["organization management"],
dependencies=[Depends(user_api_key_auth)],
)
async def deprecated_info_organization(data: OrganizationRequest):
async def deprecated_info_organization(
data: OrganizationRequest,
user_api_key_dict: UserAPIKeyAuth = Depends(user_api_key_auth),
):
"""
DEPRECATED: Use GET /organization/info instead
"""
@ -831,6 +833,15 @@ async def deprecated_info_organization(data: OrganizationRequest):
"error": f"Specify list of organization id's to query. Passed in={data.organizations}"
},
)
# Verify caller has access to each requested organization
for org_id in data.organizations:
await _verify_org_access(
organization_id=org_id,
user_api_key_dict=user_api_key_dict,
prisma_client=prisma_client,
)
response = await prisma_client.db.litellm_organizationtable.find_many(
where={"organization_id": {"in": data.organizations}},
include={"litellm_budget_table": True},

View file

@ -37,11 +37,13 @@ from litellm.proxy.management_endpoints.team_endpoints import (
_save_deleted_team_records,
_transform_teams_to_deleted_records,
_validate_and_populate_member_user_info,
_verify_team_access,
delete_team,
list_available_teams,
router,
team_member_add_duplication_check,
team_member_delete,
update_team,
validate_team_org_change,
)
from litellm.proxy.management_helpers.team_member_permission_checks import (
@ -6876,3 +6878,94 @@ class TestBatchResolveAccessGroupResources:
call_args = fake_find_many.call_args
assert len(call_args.kwargs["where"]["access_group_id"]["in"]) == 1
assert "ag-1" in result
@pytest.mark.asyncio
async def test_verify_team_access_denies_unauthorized_user():
"""
Test that _verify_team_access raises 403 when the caller is not a proxy admin,
not a team admin, and not an org admin for the team's organization.
"""
team_obj = LiteLLM_TeamTable(
team_id="team-123",
team_alias="test-team",
members_with_roles=[
Member(role="admin", user_id="other_admin_user"),
],
organization_id="org-456",
)
# Caller is an internal user with no admin role and not in the team
caller = UserAPIKeyAuth(
user_role=LitellmUserRoles.INTERNAL_USER,
user_id="unauthorized_user",
)
with patch(
"litellm.proxy.management_endpoints.team_endpoints._is_user_org_admin_for_team",
new_callable=AsyncMock,
return_value=False,
):
with pytest.raises(HTTPException) as exc_info:
await _verify_team_access(
team_obj=team_obj,
user_api_key_dict=caller,
)
assert exc_info.value.status_code == 403
@pytest.mark.asyncio
async def test_update_team_rejects_unauthorized_caller():
"""
Test that /team/update returns 403 when the caller is not a proxy admin,
not a team admin, and not an org admin — exercising the _verify_team_access
guard added to the update_team endpoint.
"""
from unittest.mock import Mock
from fastapi import Request
mock_request = Mock(spec=Request)
caller = UserAPIKeyAuth(
user_role=LitellmUserRoles.INTERNAL_USER,
user_id="unauthorized_user",
)
from litellm.proxy._types import UpdateTeamRequest
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_existing_team = MagicMock()
mock_existing_team.model_dump.return_value = {
"team_id": "team-123",
"team_alias": "test-team",
"members_with_roles": [
{"role": "admin", "user_id": "other_admin_user"},
],
"organization_id": "org-456",
}
mock_prisma_client.db.litellm_teamtable.find_unique = AsyncMock(
return_value=mock_existing_team
)
update_request = UpdateTeamRequest(
team_id="team-123",
team_alias="updated-alias",
)
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"