fix(proxy): close project hijacking and key org IDOR

Two related authorization gaps in management endpoints:

1. `/project/update` evaluated permission against the team_id supplied in
   the request body. By passing `data.team_id` pointing at a team they
   admin, a caller could hijack any project — `_check_user_permission_for_project`
   was given the attacker's team_object and happily checked admin
   membership against that. Drop the team_object kwarg so the helper
   re-fetches the existing project's team. Also require admin rights on
   the destination team when reassigning a project across teams, so a
   team admin cannot shed projects into another team's namespace.

2. `/key/update` accepted any `organization_id` and only checked that
   the org existed before applying limits. A caller could thereby point
   their key at an arbitrary org. Add `_validate_caller_can_assign_key_org`
   which enforces the same membership rule already applied on the
   `/key/list` filter path (`validate_key_list_check`); proxy admins and
   no-change updates skip the check.

Tests cover both helpers in isolation: existing-team-admin allow,
unrelated-team admin deny, proxy-admin shortcut, org-member allow,
non-member deny, missing user_id deny, no-memberships deny.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

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
+        # Permission to *edit* the project must be evaluated against the
-        team_id_to_check = data.team_id or existing_project.team_id
+        # project's CURRENT team. Sourcing the team from `data.team_id`
-        team_obj_for_checks = None
+        # would let an admin of any team pass the check by supplying their
-        if team_id_to_check is not None:
+        # own team_id, hijacking the project (VERIA-55).
-            team_obj_for_checks = await _validate_team_exists(
+        target_team_id = data.team_id or existing_project.team_id
-                team_id=team_id_to_check, prisma_client=prisma_client
+        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,
             )
 

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(
+    _org_id_to_check = data.organization_id or _existing_org_id
-        existing_key_row, "organization_id", None
-    )
     _throughput_fields_changed = (
         data.organization_id is not None
         or data.tpm_limit is not None

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