From 6dcbac88b895f9d72eee2c0cccda88ad81316194 Mon Sep 17 00:00:00 2001 From: yucheng-berri Date: Thu, 2 Jul 2026 22:08:53 -0700 Subject: [PATCH] fix(proxy): admin-gate `permissions` on /user/new and /user/update (LIT-4138) (#31998) `NewUserRequest` and `UpdateUserRequest` inherit `permissions` from `GenerateRequestBase`. `/user/new` passes the field into `generate_key_helper_fn` which persists it on the auto-created key, so an org admin who lands on `/user/new` (the route allowlist accepts org_admin callers when the request body names an org where they hold that membership) can mint a key with proxy-wide capabilities such as `get_spend_routes`. This wires the existing `_check_permissions_caller_permission` helper into `new_user` and `_update_single_user_helper`. The helper's presence check keys on `data.model_fields_set`, so an omitted field flows through untouched and an explicit `{}` / `null` from a non-admin is rejected 403 the same as any other value. `_update_single_user_helper` is shared by `/user/update` and `/user/bulk_update`, so both paths inherit the gate. Tests in `tests/test_litellm/proxy/management_endpoints/test_internal_user_endpoints.py`: - test_new_user_non_admin_permissions_non_empty_rejected - test_new_user_non_admin_permissions_explicit_empty_rejected - test_new_user_non_admin_omits_permissions_succeeds (control) - test_new_user_admin_can_set_permissions (control) - test_update_single_user_non_admin_permissions_rejected - test_update_single_user_non_admin_permissions_explicit_empty_rejected The four attack-vector tests fail on the pre-fix HEAD and pass on this commit. Full mapped test file (79 tests) green. --- .../internal_user_endpoints.py | 11 + .../key_management_endpoints.py | 2 +- .../test_internal_user_endpoints.py | 253 ++++++++++++++++++ 3 files changed, 265 insertions(+), 1 deletion(-) diff --git a/litellm/proxy/management_endpoints/internal_user_endpoints.py b/litellm/proxy/management_endpoints/internal_user_endpoints.py index 9374aa3180b..d98dcf984b2 100644 --- a/litellm/proxy/management_endpoints/internal_user_endpoints.py +++ b/litellm/proxy/management_endpoints/internal_user_endpoints.py @@ -39,6 +39,7 @@ from litellm.proxy.management_endpoints.common_utils import ( validate_finite_spend, ) from litellm.proxy.management_endpoints.key_management_endpoints import ( + _check_permissions_caller_permission, generate_key_helper_fn, prepare_metadata_fields, ) @@ -440,6 +441,11 @@ async def new_user( detail=f"Only proxy admins can create administrative users (proxy_admin, proxy_admin_viewer). Attempted to create user with role: {data.user_role}. Your role: {user_api_key_dict.user_role}", ) + _check_permissions_caller_permission( + data=data, + user_api_key_dict=user_api_key_dict, + ) + data_json = data.json() # type: ignore data_json = _update_internal_new_user_params(data_json, data) _hash_password_in_dict(data_json) @@ -1198,6 +1204,11 @@ async def _update_single_user_helper( if not user_request.user_id and not user_request.user_email: raise ValueError("Either user_id or user_email must be provided") + _check_permissions_caller_permission( + data=user_request, + user_api_key_dict=user_api_key_dict, + ) + data_json: dict = user_request.model_dump(exclude_unset=True) non_default_values = _update_internal_user_params(data_json=data_json, data=user_request) _hash_password_in_dict(non_default_values) diff --git a/litellm/proxy/management_endpoints/key_management_endpoints.py b/litellm/proxy/management_endpoints/key_management_endpoints.py index f0473edded5..ac27a8102b6 100644 --- a/litellm/proxy/management_endpoints/key_management_endpoints.py +++ b/litellm/proxy/management_endpoints/key_management_endpoints.py @@ -580,7 +580,7 @@ def _check_permissions_caller_permission( return raise HTTPException( status_code=403, - detail={"error": "Only proxy admins can set `permissions` on a key."}, + detail={"error": "Only proxy admins can set `permissions`."}, ) diff --git a/tests/test_litellm/proxy/management_endpoints/test_internal_user_endpoints.py b/tests/test_litellm/proxy/management_endpoints/test_internal_user_endpoints.py index 56eeea82223..ce2d04f0d26 100644 --- a/tests/test_litellm/proxy/management_endpoints/test_internal_user_endpoints.py +++ b/tests/test_litellm/proxy/management_endpoints/test_internal_user_endpoints.py @@ -1000,6 +1000,259 @@ async def test_new_user_non_admin_cannot_create_admin(mocker): assert str(LitellmUserRoles.PROXY_ADMIN_VIEW_ONLY) in str(exc_info2.value.message) +@pytest.mark.asyncio +async def test_new_user_non_admin_permissions_non_empty_rejected(mocker): + """`new_user` rejects a non-admin when `permissions` is present in the + request body. `/user/new` propagates the value into the auto-created + key via `generate_key_helper_fn`.""" + from litellm.proxy.management_endpoints.internal_user_endpoints import new_user + + mock_prisma_client = mocker.MagicMock() + + async def mock_count(*args, **kwargs): + return 5 + + mock_prisma_client.db.litellm_usertable.count = mock_count + + async def mock_check(*_args, **_kwargs): + return None + + mocker.patch( + "litellm.proxy.management_endpoints.internal_user_endpoints._check_duplicate_user_email", + mock_check, + ) + mocker.patch( + "litellm.proxy.management_endpoints.internal_user_endpoints._check_duplicate_user_id", + mock_check, + ) + mock_license_check = mocker.MagicMock() + mock_license_check.is_over_limit.return_value = False + mocker.patch("litellm.proxy.proxy_server.prisma_client", mock_prisma_client) + mocker.patch("litellm.proxy.proxy_server._license_check", mock_license_check) + + data = NewUserRequest( + user_email="alice@example.com", + user_role=LitellmUserRoles.INTERNAL_USER, + permissions={"get_spend_routes": True}, + ) + caller = UserAPIKeyAuth( + user_id="org-admin", user_role=LitellmUserRoles.ORG_ADMIN + ) + + with pytest.raises(ProxyException) as exc_info: + await new_user(data=data, user_api_key_dict=caller) + assert str(exc_info.value.code) == "403" + assert "permissions" in str(exc_info.value.message) + + +@pytest.mark.asyncio +async def test_new_user_non_admin_permissions_explicit_empty_rejected(mocker): + """`new_user` rejects a non-admin when `permissions` is present as + `{}` in the request body. The value matches the model default but + `model_fields_set` distinguishes the two.""" + from litellm.proxy.management_endpoints.internal_user_endpoints import new_user + + mock_prisma_client = mocker.MagicMock() + + async def mock_count(*args, **kwargs): + return 5 + + mock_prisma_client.db.litellm_usertable.count = mock_count + + async def mock_check(*_args, **_kwargs): + return None + + mocker.patch( + "litellm.proxy.management_endpoints.internal_user_endpoints._check_duplicate_user_email", + mock_check, + ) + mocker.patch( + "litellm.proxy.management_endpoints.internal_user_endpoints._check_duplicate_user_id", + mock_check, + ) + mock_license_check = mocker.MagicMock() + mock_license_check.is_over_limit.return_value = False + mocker.patch("litellm.proxy.proxy_server.prisma_client", mock_prisma_client) + mocker.patch("litellm.proxy.proxy_server._license_check", mock_license_check) + + data = NewUserRequest( + user_email="alice@example.com", + user_role=LitellmUserRoles.INTERNAL_USER, + permissions={}, + ) + assert "permissions" in data.model_fields_set + caller = UserAPIKeyAuth( + user_id="org-admin", user_role=LitellmUserRoles.ORG_ADMIN + ) + + with pytest.raises(ProxyException) as exc_info: + await new_user(data=data, user_api_key_dict=caller) + assert str(exc_info.value.code) == "403" + assert "permissions" in str(exc_info.value.message) + + +@pytest.mark.asyncio +async def test_new_user_non_admin_omits_permissions_succeeds(mocker): + """`new_user` does not fire the permissions gate when `permissions` + is absent from the request body. The model-level default `{}` is not + in `model_fields_set`.""" + from litellm.proxy.management_endpoints.internal_user_endpoints import new_user + + mock_prisma_client = mocker.MagicMock() + + async def mock_count(*args, **kwargs): + return 5 + + mock_prisma_client.db.litellm_usertable.count = mock_count + + async def mock_check(*_args, **_kwargs): + return None + + mocker.patch( + "litellm.proxy.management_endpoints.internal_user_endpoints._check_duplicate_user_email", + mock_check, + ) + mocker.patch( + "litellm.proxy.management_endpoints.internal_user_endpoints._check_duplicate_user_id", + mock_check, + ) + mock_license_check = mocker.MagicMock() + mock_license_check.is_over_limit.return_value = False + mocker.patch("litellm.proxy.proxy_server.prisma_client", mock_prisma_client) + mocker.patch("litellm.proxy.proxy_server._license_check", mock_license_check) + + stub_response = {"user_id": "alice", "key": "sk-alice", "expires": None} + + async def stub_helper(**_kwargs): + return stub_response + + mocker.patch( + "litellm.proxy.management_endpoints.internal_user_endpoints.generate_key_helper_fn", + stub_helper, + ) + + data = NewUserRequest( + user_email="alice@example.com", + user_role=LitellmUserRoles.INTERNAL_USER, + ) + assert "permissions" not in data.model_fields_set + caller = UserAPIKeyAuth( + user_id="org-admin", user_role=LitellmUserRoles.ORG_ADMIN + ) + + result = await new_user(data=data, user_api_key_dict=caller) + assert result is not None + + +@pytest.mark.asyncio +async def test_new_user_admin_can_set_permissions(mocker): + """`new_user` accepts a PROXY_ADMIN caller for any shape of + `permissions` in the request body.""" + from litellm.proxy.management_endpoints.internal_user_endpoints import new_user + + mock_prisma_client = mocker.MagicMock() + + async def mock_count(*args, **kwargs): + return 5 + + mock_prisma_client.db.litellm_usertable.count = mock_count + + async def mock_check(*_args, **_kwargs): + return None + + mocker.patch( + "litellm.proxy.management_endpoints.internal_user_endpoints._check_duplicate_user_email", + mock_check, + ) + mocker.patch( + "litellm.proxy.management_endpoints.internal_user_endpoints._check_duplicate_user_id", + mock_check, + ) + mock_license_check = mocker.MagicMock() + mock_license_check.is_over_limit.return_value = False + mocker.patch("litellm.proxy.proxy_server.prisma_client", mock_prisma_client) + mocker.patch("litellm.proxy.proxy_server._license_check", mock_license_check) + + async def stub_helper(**_kwargs): + return {"user_id": "alice", "key": "sk-alice", "expires": None} + + mocker.patch( + "litellm.proxy.management_endpoints.internal_user_endpoints.generate_key_helper_fn", + stub_helper, + ) + + admin = UserAPIKeyAuth(user_id="admin", user_role=LitellmUserRoles.PROXY_ADMIN) + for permissions_value in ({"get_spend_routes": True}, {}, None): + data = NewUserRequest( + user_email=f"alice-{permissions_value}@example.com", + user_role=LitellmUserRoles.INTERNAL_USER, + permissions=permissions_value, + ) + result = await new_user(data=data, user_api_key_dict=admin) + assert result is not None + + +@pytest.mark.asyncio +async def test_update_single_user_non_admin_permissions_rejected(mocker): + """`_update_single_user_helper` rejects a non-admin when `permissions` + is present in the request body. Covers both `/user/update` and + `/user/bulk_update`, which share this helper.""" + from fastapi import HTTPException + + from litellm.proxy._types import UpdateUserRequest + from litellm.proxy.management_endpoints.internal_user_endpoints import ( + _update_single_user_helper, + ) + + mock_prisma_client = mocker.MagicMock() + mocker.patch("litellm.proxy.proxy_server.prisma_client", mock_prisma_client) + mocker.patch("litellm.proxy.proxy_server.litellm_proxy_admin_name", "admin") + + data = UpdateUserRequest( + user_id="alice", + permissions={"get_spend_routes": True}, + ) + caller = UserAPIKeyAuth( + user_id="org-admin", user_role=LitellmUserRoles.ORG_ADMIN + ) + + with pytest.raises(HTTPException) as exc_info: + await _update_single_user_helper( + user_request=data, user_api_key_dict=caller + ) + assert exc_info.value.status_code == 403 + assert "permissions" in str(exc_info.value.detail) + + +@pytest.mark.asyncio +async def test_update_single_user_non_admin_permissions_explicit_empty_rejected(mocker): + """`_update_single_user_helper` rejects a non-admin when `permissions` + is present as `{}` in the request body.""" + from fastapi import HTTPException + + from litellm.proxy._types import UpdateUserRequest + from litellm.proxy.management_endpoints.internal_user_endpoints import ( + _update_single_user_helper, + ) + + mock_prisma_client = mocker.MagicMock() + mocker.patch("litellm.proxy.proxy_server.prisma_client", mock_prisma_client) + mocker.patch("litellm.proxy.proxy_server.litellm_proxy_admin_name", "admin") + + data = UpdateUserRequest(user_id="alice", permissions={}) + assert "permissions" in data.model_fields_set + caller = UserAPIKeyAuth( + user_id="org-admin", user_role=LitellmUserRoles.ORG_ADMIN + ) + + with pytest.raises(HTTPException) as exc_info: + await _update_single_user_helper( + user_request=data, user_api_key_dict=caller + ) + assert exc_info.value.status_code == 403 + assert "permissions" in str(exc_info.value.detail) + + @pytest.mark.asyncio async def test_user_info_url_encoding_plus_character(mocker): """