mirror of
https://github.com/BerriAI/litellm.git
synced 2026-10-08 03:08:45 +00:00
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.
This commit is contained in:
parent
dfbbda4f19
commit
6dcbac88b8
3 changed files with 265 additions and 1 deletions
|
|
@ -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)
|
||||
|
|
|
|||
|
|
@ -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`."},
|
||||
)
|
||||
|
||||
|
||||
|
|
|
|||
|
|
@ -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):
|
||||
"""
|
||||
|
|
|
|||
Loading…
Add table
Reference in a new issue