chore(team): close authz bypass via the available-team check

Two paths previously treated ``_is_available_team`` as a blanket authorization bypass — the function was meant to let standard users self-join a public team but was wired into the broader admin gate without bounding the action being performed. Three concrete exposures resulted: 1. ``/team/member_add``: the bypass let an unprivileged caller add themselves as a Team Admin, or add an arbitrary other ``user_id`` into the team. 2. ``/team/permissions_update``: the same bypass let any authenticated user overwrite a team's ``team_member_permissions`` array, mutating the access policy for every member. 3. (Read endpoint ``/team/permissions_list`` is unchanged — it leaks read-only policy state to non-members but is out of scope of the advisory's recommendation; tracking separately.) This commit: - Splits ``_validate_team_member_add_permissions`` into early-return admin checks followed by an available-team self-join branch that enforces ``member.user_id == caller.user_id`` AND ``member.role == "user"`` for every member entry in the request. The bulk shape (``member: List[Member]``) is checked the same way, so a list with one valid self-entry plus one ``role=admin`` entry is rejected. Email-only members are rejected on the self-join path: matching by ``user_id`` is the only safe primitive at pre-validation time (resolving email→user_id earlier would let unauthenticated callers probe user existence). - Removes the ``_is_available_team`` clause from ``update_team_member_permissions`` entirely. Only proxy / team / org admins can update permission policies. Tests: - Update the two existing ``_validate_team_member_add_permissions`` unit tests to pass the new ``data`` argument. - Add six regression tests covering the privesc shape (role=admin), the cross-user-injection shape (other user_id), the no-caller-uid fail-closed case, the email-only rejection, and the bulk shape. - ``test_team_endpoints.py`` 133/133 pass.
2026-10-08 03:08:45 +00:00 · 2026-04-30 03:23:46 +00:00 · 2026-04-30 03:23:46 +00:00 · e4c4d9832f
commit e4c4d9832f
parent c0211d7d1b
2 changed files with 300 additions and 25 deletions
--- a/litellm/proxy/management_endpoints/team_endpoints.py
+++ b/litellm/proxy/management_endpoints/team_endpoints.py
@ -2003,21 +2003,34 @@ def team_member_add_duplication_check(
 async def _validate_team_member_add_permissions(
    user_api_key_dict: UserAPIKeyAuth,
    complete_team_data: LiteLLM_TeamTable,
+    data: TeamMemberAddRequest,
 ) -> None:
-    """Validate if user has permission to add members to the team."""
+    """Validate if user has permission to add members to the team.
+
+    Standard users can self-join an *available team*, but the bypass
+    must not be allowed to escalate them to ``role=admin`` or to add
+    other users into the team.  When access is granted via the
+    available-team bypass we therefore enforce that every member in
+    the request matches the caller's own ``user_id`` and is being
+    added with ``role="user"``.
+    """
    if (
-        hasattr(user_api_key_dict, "user_role")
-        and user_api_key_dict.user_role != LitellmUserRoles.PROXY_ADMIN.value
-        and not _is_user_team_admin(
-            user_api_key_dict=user_api_key_dict, team_obj=complete_team_data
-        )
-        and not await _is_user_org_admin_for_team(
-            user_api_key_dict=user_api_key_dict, team_obj=complete_team_data
-        )
-        and not _is_available_team(
-            team_id=complete_team_data.team_id,
-            user_api_key_dict=user_api_key_dict,
-        )
+        getattr(user_api_key_dict, "user_role", None)
+        == LitellmUserRoles.PROXY_ADMIN.value
+    ):
+        return
+    if _is_user_team_admin(
+        user_api_key_dict=user_api_key_dict, team_obj=complete_team_data
+    ):
+        return
+    if await _is_user_org_admin_for_team(
+        user_api_key_dict=user_api_key_dict, team_obj=complete_team_data
+    ):
+        return
+
+    if not _is_available_team(
+        team_id=complete_team_data.team_id,
+        user_api_key_dict=user_api_key_dict,
    ):
        raise HTTPException(
            status_code=403,
@ -2029,6 +2042,34 @@ async def _validate_team_member_add_permissions(
            },
        )

+    # Available-team self-join: caller may add only themselves, only as a
+    # standard user.  Enforce that here so the bypass cannot be used as a
+    # privilege-escalation or cross-user-injection primitive.
+    members = data.member if isinstance(data.member, list) else [data.member]
+    caller_user_id = getattr(user_api_key_dict, "user_id", None)
+    for member in members:
+        if getattr(member, "role", "user") != "user":
+            raise HTTPException(
+                status_code=403,
+                detail={
+                    "error": (
+                        "Available-team self-join cannot assign 'admin' role. "
+                        "Only proxy/team/org admins can add admins to a team."
+                    )
+                },
+            )
+        member_user_id = getattr(member, "user_id", None)
+        if not caller_user_id or not member_user_id or member_user_id != caller_user_id:
+            raise HTTPException(
+                status_code=403,
+                detail={
+                    "error": (
+                        "Available-team self-join can only add the caller "
+                        "(user_id must match the authenticated user's user_id)."
+                    )
+                },
+            )
+

 async def _process_team_members(
    data: TeamMemberAddRequest,
@ -2384,6 +2425,7 @@ async def team_member_add(
    await _validate_team_member_add_permissions(
        user_api_key_dict=user_api_key_dict,
        complete_team_data=complete_team_data,
+        data=data,
    )

    # Validate and populate user_email/user_id for members before processing
@ -4697,6 +4739,8 @@ async def update_team_member_permissions(

    complete_team_data = LiteLLM_TeamTable(**existing_team_row.model_dump())

+    # Available-team self-join must NOT grant write access to team-wide
+    # permission policies; only proxy/team/org admins can update them.
    if (
        hasattr(user_api_key_dict, "user_role")
        and user_api_key_dict.user_role != LitellmUserRoles.PROXY_ADMIN.value
@ -4706,16 +4750,12 @@ async def update_team_member_permissions(
        and not await _is_user_org_admin_for_team(
            user_api_key_dict=user_api_key_dict, team_obj=complete_team_data
        )
-        and not _is_available_team(
-            team_id=complete_team_data.team_id,
-            user_api_key_dict=user_api_key_dict,
-        )
    ):
        raise HTTPException(
            status_code=403,
            detail={
                "error": "Call not allowed. User not proxy admin OR team admin. route={}, team_id={}".format(
-                    "/team/member_add",
+                    "/team/permissions_update",
                    complete_team_data.team_id,
                )
            },
--- a/tests/test_litellm/proxy/management_endpoints/test_team_endpoints.py
+++ b/tests/test_litellm/proxy/management_endpoints/test_team_endpoints.py
@ -968,6 +968,20 @@ def test_add_new_models_to_team():
    )


+def _make_team_member_add_request(
+    member_user_id: Optional[str] = "regular-user",
+    role: str = "user",
+    team_id: str = "test-team-123",
+):
+    """Build a TeamMemberAddRequest with one Member entry for tests below."""
+    from litellm.proxy._types import Member, TeamMemberAddRequest
+
+    return TeamMemberAddRequest(
+        team_id=team_id,
+        member=Member(role=role, user_id=member_user_id),
+    )
+
+
@pytest.mark.asyncio
 async def test_validate_team_member_add_permissions_admin():
    """
@ -977,17 +991,15 @@ async def test_validate_team_member_add_permissions_admin():
        _validate_team_member_add_permissions,
    )

-    # Create admin user
    admin_user = UserAPIKeyAuth(user_role=LitellmUserRoles.PROXY_ADMIN)

-    # Create mock team
    team = MagicMock(spec=LiteLLM_TeamTable)
    team.team_id = "test-team-123"

-    # Should not raise any exception for admin
    await _validate_team_member_add_permissions(
        user_api_key_dict=admin_user,
        complete_team_data=team,
+        data=_make_team_member_add_request(member_user_id="any-user", role="admin"),
    )


@ -1000,20 +1012,17 @@ async def test_validate_team_member_add_permissions_non_admin():
        _validate_team_member_add_permissions,
    )

-    # Create non-admin user
    regular_user = UserAPIKeyAuth(
        user_id="regular-user",
        user_role=LitellmUserRoles.INTERNAL_USER,
        team_id="different-team",
    )

-    # Create mock team
    team = MagicMock(spec=LiteLLM_TeamTable)
    team.team_id = "test-team-123"
    team.members_with_roles = []
    team.organization_id = None

-    # Mock the helper functions to return False
    with (
        patch(
            "litellm.proxy.management_endpoints.team_endpoints._is_user_team_admin",
@ -1024,17 +1033,243 @@ async def test_validate_team_member_add_permissions_non_admin():
            return_value=False,
        ),
    ):
-        # Should raise HTTPException for non-admin
        with pytest.raises(HTTPException) as exc_info:
            await _validate_team_member_add_permissions(
                user_api_key_dict=regular_user,
                complete_team_data=team,
+                data=_make_team_member_add_request(),
            )

        assert exc_info.value.status_code == 403
        assert "not proxy admin OR team admin" in str(exc_info.value.detail)


+# ── VERIA-56 regression tests for _is_available_team self-join enforcement ───
+
+
+@pytest.mark.asyncio
+async def test_available_team_self_join_with_caller_user_id_allowed():
+    """A standard user adding themselves to an available team with role=user
+    is the only legitimate use of the available-team bypass."""
+    from litellm.proxy.management_endpoints.team_endpoints import (
+        _validate_team_member_add_permissions,
+    )
+
+    user = UserAPIKeyAuth(
+        user_id="alice",
+        user_role=LitellmUserRoles.INTERNAL_USER,
+    )
+
+    team = MagicMock(spec=LiteLLM_TeamTable)
+    team.team_id = "public-team"
+    team.members_with_roles = []
+    team.organization_id = None
+
+    with (
+        patch(
+            "litellm.proxy.management_endpoints.team_endpoints._is_user_team_admin",
+            return_value=False,
+        ),
+        patch(
+            "litellm.proxy.management_endpoints.team_endpoints._is_available_team",
+            return_value=True,
+        ),
+    ):
+        await _validate_team_member_add_permissions(
+            user_api_key_dict=user,
+            complete_team_data=team,
+            data=_make_team_member_add_request(member_user_id="alice", role="user"),
+        )
+
+
+@pytest.mark.asyncio
+async def test_available_team_self_join_blocks_admin_role():
+    """Privesc shape from VERIA-56: caller adds themselves with role=admin
+    via the available-team bypass.  Must be rejected."""
+    from litellm.proxy.management_endpoints.team_endpoints import (
+        _validate_team_member_add_permissions,
+    )
+
+    user = UserAPIKeyAuth(user_id="alice", user_role=LitellmUserRoles.INTERNAL_USER)
+    team = MagicMock(spec=LiteLLM_TeamTable)
+    team.team_id = "public-team"
+    team.members_with_roles = []
+    team.organization_id = None
+
+    with (
+        patch(
+            "litellm.proxy.management_endpoints.team_endpoints._is_user_team_admin",
+            return_value=False,
+        ),
+        patch(
+            "litellm.proxy.management_endpoints.team_endpoints._is_available_team",
+            return_value=True,
+        ),
+        pytest.raises(HTTPException) as exc_info,
+    ):
+        await _validate_team_member_add_permissions(
+            user_api_key_dict=user,
+            complete_team_data=team,
+            data=_make_team_member_add_request(member_user_id="alice", role="admin"),
+        )
+
+    assert exc_info.value.status_code == 403
+    assert "admin" in str(exc_info.value.detail).lower()
+
+
+@pytest.mark.asyncio
+async def test_available_team_self_join_blocks_other_user_id():
+    """Cross-user-injection shape from VERIA-56: caller adds someone else
+    via the available-team bypass.  Must be rejected."""
+    from litellm.proxy.management_endpoints.team_endpoints import (
+        _validate_team_member_add_permissions,
+    )
+
+    user = UserAPIKeyAuth(user_id="alice", user_role=LitellmUserRoles.INTERNAL_USER)
+    team = MagicMock(spec=LiteLLM_TeamTable)
+    team.team_id = "public-team"
+    team.members_with_roles = []
+    team.organization_id = None
+
+    with (
+        patch(
+            "litellm.proxy.management_endpoints.team_endpoints._is_user_team_admin",
+            return_value=False,
+        ),
+        patch(
+            "litellm.proxy.management_endpoints.team_endpoints._is_available_team",
+            return_value=True,
+        ),
+        pytest.raises(HTTPException) as exc_info,
+    ):
+        await _validate_team_member_add_permissions(
+            user_api_key_dict=user,
+            complete_team_data=team,
+            data=_make_team_member_add_request(
+                member_user_id="bob-victim", role="user"
+            ),
+        )
+
+    assert exc_info.value.status_code == 403
+
+
+@pytest.mark.asyncio
+async def test_available_team_self_join_blocks_when_caller_has_no_user_id():
+    """If the auth context has no user_id we cannot prove self-join, so the
+    bypass must fail closed."""
+    from litellm.proxy.management_endpoints.team_endpoints import (
+        _validate_team_member_add_permissions,
+    )
+
+    user = UserAPIKeyAuth(user_role=LitellmUserRoles.INTERNAL_USER)  # no user_id
+    team = MagicMock(spec=LiteLLM_TeamTable)
+    team.team_id = "public-team"
+    team.members_with_roles = []
+    team.organization_id = None
+
+    with (
+        patch(
+            "litellm.proxy.management_endpoints.team_endpoints._is_user_team_admin",
+            return_value=False,
+        ),
+        patch(
+            "litellm.proxy.management_endpoints.team_endpoints._is_available_team",
+            return_value=True,
+        ),
+        pytest.raises(HTTPException) as exc_info,
+    ):
+        await _validate_team_member_add_permissions(
+            user_api_key_dict=user,
+            complete_team_data=team,
+            data=_make_team_member_add_request(member_user_id="alice", role="user"),
+        )
+
+    assert exc_info.value.status_code == 403
+
+
+@pytest.mark.asyncio
+async def test_available_team_self_join_blocks_email_only_member():
+    """An email-only member entry can't be safely self-join-validated; the
+    caller must use their own user_id explicitly."""
+    from litellm.proxy._types import Member, TeamMemberAddRequest
+    from litellm.proxy.management_endpoints.team_endpoints import (
+        _validate_team_member_add_permissions,
+    )
+
+    user = UserAPIKeyAuth(user_id="alice", user_role=LitellmUserRoles.INTERNAL_USER)
+    team = MagicMock(spec=LiteLLM_TeamTable)
+    team.team_id = "public-team"
+    team.members_with_roles = []
+    team.organization_id = None
+
+    data = TeamMemberAddRequest(
+        team_id="public-team",
+        member=Member(role="user", user_email="alice@example.com"),
+    )
+
+    with (
+        patch(
+            "litellm.proxy.management_endpoints.team_endpoints._is_user_team_admin",
+            return_value=False,
+        ),
+        patch(
+            "litellm.proxy.management_endpoints.team_endpoints._is_available_team",
+            return_value=True,
+        ),
+        pytest.raises(HTTPException) as exc_info,
+    ):
+        await _validate_team_member_add_permissions(
+            user_api_key_dict=user,
+            complete_team_data=team,
+            data=data,
+        )
+
+    assert exc_info.value.status_code == 403
+
+
+@pytest.mark.asyncio
+async def test_available_team_self_join_blocks_admin_role_in_member_list():
+    """Bulk shape: list of members where one has role=admin must be rejected
+    even if the caller's own entry is correct."""
+    from litellm.proxy._types import Member, TeamMemberAddRequest
+    from litellm.proxy.management_endpoints.team_endpoints import (
+        _validate_team_member_add_permissions,
+    )
+
+    user = UserAPIKeyAuth(user_id="alice", user_role=LitellmUserRoles.INTERNAL_USER)
+    team = MagicMock(spec=LiteLLM_TeamTable)
+    team.team_id = "public-team"
+    team.members_with_roles = []
+    team.organization_id = None
+
+    data = TeamMemberAddRequest(
+        team_id="public-team",
+        member=[
+            Member(role="user", user_id="alice"),
+            Member(role="admin", user_id="alice"),
+        ],
+    )
+
+    with (
+        patch(
+            "litellm.proxy.management_endpoints.team_endpoints._is_user_team_admin",
+            return_value=False,
+        ),
+        patch(
+            "litellm.proxy.management_endpoints.team_endpoints._is_available_team",
+            return_value=True,
+        ),
+        pytest.raises(HTTPException) as exc_info,
+    ):
+        await _validate_team_member_add_permissions(
+            user_api_key_dict=user,
+            complete_team_data=team,
+            data=data,
+        )
+
+    assert exc_info.value.status_code == 403
+
+
@pytest.mark.asyncio
 async def test_process_team_members_single_member():
    """