diff --git a/enterprise/litellm_enterprise/proxy/management_endpoints/project_endpoints.py b/enterprise/litellm_enterprise/proxy/management_endpoints/project_endpoints.py index f6ed7767c463..01d4fdd38174 100644 --- a/enterprise/litellm_enterprise/proxy/management_endpoints/project_endpoints.py +++ b/enterprise/litellm_enterprise/proxy/management_endpoints/project_endpoints.py @@ -588,24 +588,21 @@ async def update_project( # noqa: PLR0915 param="project_id", ) - # Validate team exists and get team object for limit + permission checks - team_id_to_check = data.team_id or existing_project.team_id - team_obj_for_checks = None - if team_id_to_check is not None: - team_obj_for_checks = await _validate_team_exists( - team_id=team_id_to_check, prisma_client=prisma_client + # Permission to *edit* the project must be evaluated against the + # project's CURRENT team. Sourcing the team from `data.team_id` + # would let an admin of any team pass the check by supplying their + # own team_id, hijacking the project (VERIA-55). + target_team_id = data.team_id or existing_project.team_id + target_team_obj = None + if target_team_id is not None: + target_team_obj = await _validate_team_exists( + team_id=target_team_id, prisma_client=prisma_client ) - # Check if user has permission to update this project has_permission = await _check_user_permission_for_project( user_api_key_dict=user_api_key_dict, team_id=existing_project.team_id, prisma_client=prisma_client, - team_object=( - LiteLLM_TeamTable(**team_obj_for_checks.model_dump()) - if team_obj_for_checks - else None - ), ) if not has_permission: @@ -614,10 +611,32 @@ async def update_project( # noqa: PLR0915 detail={"error": "Only admins or team admins can update projects"}, ) + # Reassigning to a different team also requires admin rights on the + # destination team — otherwise a team admin could shed projects into + # an unsuspecting team's namespace. + if data.team_id is not None and data.team_id != existing_project.team_id: + can_assign_to_target = await _check_user_permission_for_project( + user_api_key_dict=user_api_key_dict, + team_id=data.team_id, + prisma_client=prisma_client, + team_object=( + LiteLLM_TeamTable(**target_team_obj.model_dump()) + if target_team_obj + else None + ), + ) + if not can_assign_to_target: + raise HTTPException( + status_code=403, + detail={ + "error": "Cannot reassign project to a team you are not an admin of" + }, + ) + # Validate project limits against team limits - if team_obj_for_checks is not None: + if target_team_obj is not None: _check_team_project_limits( - team_object=LiteLLM_TeamTable(**team_obj_for_checks.model_dump()), + team_object=LiteLLM_TeamTable(**target_team_obj.model_dump()), data=data, ) diff --git a/litellm/proxy/management_endpoints/key_management_endpoints.py b/litellm/proxy/management_endpoints/key_management_endpoints.py index 2485aea14f19..52dbfd1ece19 100644 --- a/litellm/proxy/management_endpoints/key_management_endpoints.py +++ b/litellm/proxy/management_endpoints/key_management_endpoints.py @@ -1168,6 +1168,42 @@ def check_org_key_rpm_tpm_limits( ) +async def _validate_caller_can_assign_key_org( + user_api_key_dict: UserAPIKeyAuth, + organization_id: str, + prisma_client: PrismaClient, +) -> None: + """Reject ``/key/update`` requests that point a key at an organization + the caller does not belong to. + + Mirrors the org-membership rule already enforced on ``/key/list`` in + ``validate_key_list_check``. Proxy admins are checked at the call site. + """ + if user_api_key_dict.user_id is None: + raise HTTPException( + status_code=status.HTTP_403_FORBIDDEN, + detail="Cannot assign a key to an organization without a user_id on the caller's token", + ) + + user_row = await prisma_client.db.litellm_usertable.find_unique( + where={"user_id": user_api_key_dict.user_id}, + include={"organization_memberships": True}, + ) + memberships = ( + getattr(user_row, "organization_memberships", None) if user_row else None + ) + member_org_ids = { + membership.organization_id + for membership in (memberships or []) + if membership.organization_id is not None + } + if organization_id not in member_org_ids: + raise HTTPException( + status_code=status.HTTP_403_FORBIDDEN, + detail=f"Caller is not a member of organization_id={organization_id}", + ) + + async def _check_org_key_limits( org_table: LiteLLM_OrganizationTable, data: Union[GenerateKeyRequest, UpdateKeyRequest], @@ -2168,10 +2204,26 @@ async def _validate_update_key_data( user_api_key_cache=user_api_key_cache, ) + # When the caller asks to change the key's organization_id, require that + # they are a member of (or a proxy admin over) the target organization. + # Without this gate, any caller could assign their key to an arbitrary + # organization_id by passing it in the request body — VERIA-55 secondary + # IDOR. The check mirrors the membership rule already used on the + # `/key/list` filter path in `validate_key_list_check`. + _existing_org_id = getattr(existing_key_row, "organization_id", None) + if ( + data.organization_id is not None + and data.organization_id != _existing_org_id + and not _is_proxy_admin + ): + await _validate_caller_can_assign_key_org( + user_api_key_dict=user_api_key_dict, + organization_id=data.organization_id, + prisma_client=prisma_client, + ) + # Check org key limits only when throughput-related fields or organization_id change - _org_id_to_check = data.organization_id or getattr( - existing_key_row, "organization_id", None - ) + _org_id_to_check = data.organization_id or _existing_org_id _throughput_fields_changed = ( data.organization_id is not None or data.tpm_limit is not None diff --git a/tests/test_litellm/proxy/management_endpoints/test_project_org_authz.py b/tests/test_litellm/proxy/management_endpoints/test_project_org_authz.py new file mode 100644 index 000000000000..bd982480d60a --- /dev/null +++ b/tests/test_litellm/proxy/management_endpoints/test_project_org_authz.py @@ -0,0 +1,195 @@ +""" +Unit tests for the VERIA-55 fixes: + +- Project update permission must be evaluated against the project's *current* + team, not a team supplied in the request body. +- Key update may not assign a key to an organization the caller is not a + member of. +""" + +from unittest.mock import AsyncMock, MagicMock + +import pytest +from fastapi import HTTPException + +from litellm.proxy._types import LitellmUserRoles, UserAPIKeyAuth + + +# --------------------------------------------------------------------------- +# /project/update — _check_user_permission_for_project +# --------------------------------------------------------------------------- + + +def _make_prisma_with_team(team_id: str, admins: list): + prisma = MagicMock() + team_row = MagicMock() + team_row.team_id = team_id + team_row.admins = admins + prisma.db.litellm_teamtable.find_unique = AsyncMock(return_value=team_row) + return prisma + + +@pytest.mark.asyncio +async def test_project_perm_check_uses_current_team_not_caller_supplied(): + """The permission check must look at the project's existing team. Even + if the caller is admin of an unrelated team, they must not pass when no + explicit team_object is forced through.""" + from enterprise.litellm_enterprise.proxy.management_endpoints.project_endpoints import ( + _check_user_permission_for_project, + ) + + # Project lives on team-A, caller is admin only of team-B. + prisma = _make_prisma_with_team(team_id="team-A", admins=["alice"]) + caller = UserAPIKeyAuth( + user_id="bob", + user_role=LitellmUserRoles.INTERNAL_USER.value, + ) + + has_perm = await _check_user_permission_for_project( + user_api_key_dict=caller, + team_id="team-A", + prisma_client=prisma, + ) + assert has_perm is False + prisma.db.litellm_teamtable.find_unique.assert_awaited_once() + + +@pytest.mark.asyncio +async def test_project_perm_check_allows_team_admin_of_existing_team(): + from enterprise.litellm_enterprise.proxy.management_endpoints.project_endpoints import ( + _check_user_permission_for_project, + ) + + prisma = _make_prisma_with_team(team_id="team-A", admins=["alice"]) + alice = UserAPIKeyAuth( + user_id="alice", + user_role=LitellmUserRoles.INTERNAL_USER.value, + ) + + has_perm = await _check_user_permission_for_project( + user_api_key_dict=alice, + team_id="team-A", + prisma_client=prisma, + ) + assert has_perm is True + + +@pytest.mark.asyncio +async def test_project_perm_check_proxy_admin_always_allowed(): + from enterprise.litellm_enterprise.proxy.management_endpoints.project_endpoints import ( + _check_user_permission_for_project, + ) + + prisma = MagicMock() + admin = UserAPIKeyAuth( + user_id="root", + user_role=LitellmUserRoles.PROXY_ADMIN.value, + ) + + has_perm = await _check_user_permission_for_project( + user_api_key_dict=admin, + team_id="team-A", + prisma_client=prisma, + ) + assert has_perm is True + # Admin shortcut should not even hit the DB. + prisma.db.litellm_teamtable.find_unique.assert_not_called() + + +# --------------------------------------------------------------------------- +# /key/update — _validate_caller_can_assign_key_org +# --------------------------------------------------------------------------- + + +def _make_prisma_with_user_orgs(user_id: str, org_ids: list): + prisma = MagicMock() + user_row = MagicMock() + user_row.organization_memberships = [ + MagicMock(organization_id=org_id) for org_id in org_ids + ] + prisma.db.litellm_usertable.find_unique = AsyncMock(return_value=user_row) + return prisma + + +@pytest.mark.asyncio +async def test_assign_key_org_allows_member(): + from litellm.proxy.management_endpoints.key_management_endpoints import ( + _validate_caller_can_assign_key_org, + ) + + prisma = _make_prisma_with_user_orgs("alice", ["org-1", "org-2"]) + caller = UserAPIKeyAuth( + user_id="alice", + user_role=LitellmUserRoles.INTERNAL_USER.value, + ) + # Should not raise. + await _validate_caller_can_assign_key_org( + user_api_key_dict=caller, + organization_id="org-2", + prisma_client=prisma, + ) + + +@pytest.mark.asyncio +async def test_assign_key_org_blocks_non_member(): + """The IDOR: caller asks to point a key at an org they don't belong to.""" + from litellm.proxy.management_endpoints.key_management_endpoints import ( + _validate_caller_can_assign_key_org, + ) + + prisma = _make_prisma_with_user_orgs("alice", ["org-1"]) + caller = UserAPIKeyAuth( + user_id="alice", + user_role=LitellmUserRoles.INTERNAL_USER.value, + ) + with pytest.raises(HTTPException) as exc_info: + await _validate_caller_can_assign_key_org( + user_api_key_dict=caller, + organization_id="someone-elses-org", + prisma_client=prisma, + ) + assert exc_info.value.status_code == 403 + assert "someone-elses-org" in str(exc_info.value.detail) + + +@pytest.mark.asyncio +async def test_assign_key_org_blocks_caller_without_user_id(): + from litellm.proxy.management_endpoints.key_management_endpoints import ( + _validate_caller_can_assign_key_org, + ) + + prisma = MagicMock() + caller = UserAPIKeyAuth( + user_role=LitellmUserRoles.INTERNAL_USER.value, + ) + with pytest.raises(HTTPException) as exc_info: + await _validate_caller_can_assign_key_org( + user_api_key_dict=caller, + organization_id="org-1", + prisma_client=prisma, + ) + assert exc_info.value.status_code == 403 + + +@pytest.mark.asyncio +async def test_assign_key_org_blocks_caller_with_no_memberships(): + from litellm.proxy.management_endpoints.key_management_endpoints import ( + _validate_caller_can_assign_key_org, + ) + + prisma = MagicMock() + user_row = MagicMock() + user_row.organization_memberships = None + prisma.db.litellm_usertable.find_unique = AsyncMock(return_value=user_row) + + caller = UserAPIKeyAuth( + user_id="alice", + user_role=LitellmUserRoles.INTERNAL_USER.value, + ) + with pytest.raises(HTTPException) as exc_info: + await _validate_caller_can_assign_key_org( + user_api_key_dict=caller, + organization_id="org-1", + prisma_client=prisma, + ) + assert exc_info.value.status_code == 403