fix(proxy): self-updates cannot lift user-level ceilings or reassign membership

Generated-by: termag

litellm/proxy/management_endpoints/internal_user_endpoints.py

@@ -1513,15 +1513,11 @@ async def _update_single_user_helper(
         # because `_update_internal_user_params` drops empty values, and `object_permission: {}` is
         # precisely the clear-my-own-ceiling case this must refuse.
         _sent_fields: Final = user_request.fields_set() if hasattr(user_request, "fields_set") else set()
-        _protected_fields: Final = ("max_budget", "model_max_budget", "soft_budget", "spend", "object_permission")
-        for _field in _protected_fields:
-            if _field in non_default_values or _field in _sent_fields:
-                raise HTTPException(
-                    status_code=403,
-                    detail={
-                        "error": f"Non-admin users cannot modify '{_field}' on their own record. Contact your proxy admin."
-                    },
-                )
+        _reject_ceiling_self_update(
+            sent_fields=_sent_fields,
+            non_default_values=non_default_values,
+            user_api_key_dict=user_api_key_dict,
+        )
 
     existing_metadata: Final = (
         cast(dict, getattr(existing_user_row, "metadata", {}) or {}) if existing_user_row is not None else {}
@@ -1641,6 +1637,56 @@ async def _update_single_user_helper(
     return response
 
 
+_SELF_UPDATE_PROTECTED_FIELDS: Final[tuple[str, ...]] = (
+    "max_budget",
+    "model_max_budget",
+    "soft_budget",
+    "spend",
+    "object_permission",
+    # User-level ceilings sit above per-key self-service sizing, so a
+    # self-write that lifts them is an escalation path.
+    "budget_duration",
+    "rpm_limit",
+    "tpm_limit",
+    "max_parallel_requests",
+    "models",
+    # Membership authority: user_object.teams / organizations feed
+    # GrantResolver.resolve and scope /key/list, team-model visibility and
+    # BYOK team scoping, so a self-write would grant real team access while
+    # the real gate (/team/member_add) stays admin-only (R1-M3).
+    "teams",
+    "organizations",
+)
+
+
+def _reject_ceiling_self_update(
+    *,
+    sent_fields: set[str],
+    non_default_values: dict,
+    user_api_key_dict: UserAPIKeyAuth,
+) -> None:
+    """Raise 403 when a non-admin self-update touches a user-level ceiling.
+
+    object_permission is a ceiling on what this human may reach, so a
+    self-write is an escalation path: an empty grant list means "no
+    restriction" and would lift a restriction an admin placed on them.
+    Checked against the fields the caller actually SENT, because
+    `_update_internal_user_params` drops empty values, and
+    `object_permission: {}` is precisely the clear-my-own-ceiling case
+    this must refuse.
+    """
+    if user_api_key_dict.user_role == LitellmUserRoles.PROXY_ADMIN.value:
+        return
+    for _field in _SELF_UPDATE_PROTECTED_FIELDS:
+        if _field in non_default_values or _field in sent_fields:
+            raise HTTPException(
+                status_code=403,
+                detail={
+                    "error": f"Non-admin users cannot modify '{_field}' on their own record. Contact your proxy admin."
+                },
+            )
+
+
 def can_user_call_user_update(
     user_api_key_dict: UserAPIKeyAuth,
     user_info: LiteLLM_UserTable,

tests/unit/proxy/management_endpoints/test_internal_user_endpoints.py

@@ -5211,3 +5211,106 @@ async def test_user_update_without_password_revokes_nothing(_admin_prisma, mocke
     await _update_single_user_helper(user_request=user_request, user_api_key_dict=admin_caller)
 
     revoke_mock.assert_not_awaited()
+
+
+# ---------------------------------------------------------------------------
+# /user/update self-update ceiling fields (fork hardening R2-17)
+# ---------------------------------------------------------------------------
+
+
+@pytest.mark.parametrize(
+    "field,value",
+    [
+        ("rpm_limit", 1000000),
+        ("tpm_limit", 1000000),
+        ("max_parallel_requests", 1000),
+        ("budget_duration", "1s"),
+        ("models", []),
+    ],
+)
+def test_user_update_self_write_of_ceiling_fields_rejected(field, value):
+    """A non-admin self-update must not lift user-wide ceilings; those sit
+    above per-key settings and were admin-imposed."""
+    from fastapi import HTTPException
+    from litellm.proxy._types import LitellmUserRoles, UpdateUserRequest, UserAPIKeyAuth
+    from litellm.proxy.management_endpoints.internal_user_endpoints import (
+        _reject_ceiling_self_update,
+    )
+
+    data = UpdateUserRequest(user_id="self", **{field: value})
+    with pytest.raises(HTTPException) as exc:
+        _reject_ceiling_self_update(
+            sent_fields=set(data.fields_set()),
+            non_default_values={field: value},
+            user_api_key_dict=UserAPIKeyAuth(
+                user_id="self", user_role=LitellmUserRoles.INTERNAL_USER
+            ),
+        )
+    assert int(getattr(exc.value, "status_code", 0)) == 403
+    assert f"'{field}'" in str(exc.value.detail)
+
+
+def test_user_update_self_write_of_innocuous_field_allowed():
+    """Guardrail on the tightening: user_alias self-edits keep flowing."""
+    from litellm.proxy._types import LitellmUserRoles, UpdateUserRequest, UserAPIKeyAuth
+    from litellm.proxy.management_endpoints.internal_user_endpoints import (
+        _reject_ceiling_self_update,
+    )
+
+    data = UpdateUserRequest(user_id="self", user_alias="me")
+    _reject_ceiling_self_update(
+        sent_fields=set(data.fields_set()),
+        non_default_values={"user_alias": "me"},
+        user_api_key_dict=UserAPIKeyAuth(
+            user_id="self", user_role=LitellmUserRoles.INTERNAL_USER
+        ),
+    )
+
+
+def test_user_update_admin_self_write_unrestricted():
+    """Proxy admins still edit every field on their own record."""
+    from litellm.proxy._types import LitellmUserRoles, UpdateUserRequest, UserAPIKeyAuth
+    from litellm.proxy.management_endpoints.internal_user_endpoints import (
+        _reject_ceiling_self_update,
+    )
+
+    data = UpdateUserRequest(user_id="admin", rpm_limit=1000000, models=[])
+    _reject_ceiling_self_update(
+        sent_fields=set(data.fields_set()),
+        non_default_values={"rpm_limit": 1000000, "models": []},
+        user_api_key_dict=UserAPIKeyAuth(
+            user_id="admin", user_role=LitellmUserRoles.PROXY_ADMIN
+        ),
+    )
+
+
+# ---------------------------------------------------------------------------
+# /user/update membership self-grant (fork hardening R1-M3)
+# ---------------------------------------------------------------------------
+
+
+@pytest.mark.asyncio
+async def test_user_update_self_write_of_membership_rejected():
+    """A non-admin self-write of `teams` / `organizations` must be rejected:
+    user_object.teams is GrantResolver's membership authority and scopes
+    /key/list, team-model visibility and BYOK scoping."""
+    from fastapi import HTTPException
+    from litellm.proxy._types import LitellmUserRoles, UpdateUserRequest, UserAPIKeyAuth
+    from litellm.proxy.management_endpoints.internal_user_endpoints import (
+        _reject_ceiling_self_update,
+    )
+    for field, value in (
+        ("teams", ["victim-team"]),
+        ("organizations", ["victim-org"]),
+    ):
+        data = UpdateUserRequest(user_id="self", **{field: value})
+        with pytest.raises(HTTPException) as exc:
+            _reject_ceiling_self_update(
+                sent_fields=set(data.fields_set()),
+                non_default_values={field: value},
+                user_api_key_dict=UserAPIKeyAuth(
+                    user_id="self", user_role=LitellmUserRoles.INTERNAL_USER
+                ),
+            )
+        assert int(getattr(exc.value, "status_code", 0)) == 403
+        assert f"'{field}'" in str(exc.value.detail)