fix(proxy): gate get_spend_routes on key update/regenerate and auth

Address BrickerP review on #47112: refuse truthy get_spend_routes on
/key/update and /key/regenerate for non-Enterprise, and require a
truthy permission value in route checks so false no longer opens
global spend routes.
This commit is contained in:
er-s-an 2026-10-18 07:38:13 +08:00
parent 7f54c36ee1
commit 6b3e4a478e
4 changed files with 147 additions and 15 deletions

View file

@ -297,6 +297,15 @@ class RouteChecks:
f"Only proxy admin can be used to generate, delete, update info for new keys/users/teams. Route={route}. Your role={user_role}. Your user_id={masked_user_id}"
)
@staticmethod
def _has_truthy_permission(permissions: object | None, name: str) -> bool:
"""True when `name` is opted in with a truthy value (dict) or listed (sequence)."""
if isinstance(permissions, dict):
return bool(permissions.get(name))
if isinstance(permissions, (list, tuple, set)):
return name in permissions
return False
@staticmethod
def non_proxy_admin_allowed_routes_check(
user_obj: LiteLLM_UserTable | None,
@ -356,7 +365,10 @@ class RouteChecks:
elif (
route in LiteLLMRoutes.global_spend_tracking_routes.value
and getattr(valid_token, "permissions", None) is not None
and "get_spend_routes" in getattr(valid_token, "permissions", [])
and RouteChecks._has_truthy_permission(
getattr(valid_token, "permissions", None),
"get_spend_routes",
)
):
pass
elif route == "/lens" or route.startswith("/lens/"):

View file

@ -3518,6 +3518,28 @@ async def _owner_may_change_key_budget(
)
return True
def _refuse_get_spend_routes_unless_premium(
permissions: object | None,
premium_user: bool,
) -> None:
"""Block minting truthy `get_spend_routes` on non-Enterprise proxies.
Used by `/key/generate`, `/key/update`, and `/key/regenerate`. An explicit
`false` is allowed so callers can clear or set the flag without a licence.
"""
if not isinstance(permissions, dict):
return
if permissions.get("get_spend_routes") and premium_user is not True:
raise HTTPException(
status_code=status.HTTP_403_FORBIDDEN,
detail={
"error": (
"get_spend_routes permission is only available for "
f"LiteLLM Enterprise users. {CommonProxyErrors.not_premium_user.value}"
)
},
)
async def _validate_update_key_data(
data: UpdateKeyRequest,
@ -3533,6 +3555,8 @@ async def _validate_update_key_data(
# Reject NaN/±inf spend before it can reach the DB / spend counter.
validate_finite_spend(data.spend)
if "permissions" in data.model_fields_set:
_refuse_get_spend_routes_unless_premium(data.permissions, premium_user)
validate_budget_duration(data.budget_duration)
_is_proxy_admin: Final = user_api_key_dict.user_role == LitellmUserRoles.PROXY_ADMIN.value
@ -5262,20 +5286,7 @@ async def generate_key_helper_fn(
parsed_permissions = json.loads(saved_token["permissions"])
# Refuse only when the permission is requested (truthy). An explicit
# false must not trip a substring check on the serialized JSON.
if (
isinstance(parsed_permissions, dict)
and parsed_permissions.get("get_spend_routes")
and premium_user is not True
):
raise HTTPException(
status_code=status.HTTP_403_FORBIDDEN,
detail={
"error": (
"get_spend_routes permission is only available for "
f"LiteLLM Enterprise users. {CommonProxyErrors.not_premium_user.value}"
)
},
)
_refuse_get_spend_routes_unless_premium(parsed_permissions, premium_user)
saved_token["permissions"] = parsed_permissions
if isinstance(saved_token["model_max_budget"], str):
saved_token["model_max_budget"] = json.loads(saved_token["model_max_budget"])
@ -6370,6 +6381,8 @@ async def regenerate_key_fn(
data=data,
user_api_key_dict=user_api_key_dict,
)
if "permissions" in data.model_fields_set:
_refuse_get_spend_routes_unless_premium(data.permissions, premium_user)
# Mirror /key/generate's post-handle_key_type recheck so a
# non-admin can't elevate via a key_type preset that the
# regenerate flow would otherwise carry through unchecked.

View file

@ -2174,6 +2174,35 @@ def test_get_spend_routes_permission_keeps_access_for_internal_user(route):
)
@pytest.mark.parametrize("route", GLOBAL_SPEND_ROUTES)
def test_get_spend_routes_false_permission_denies_internal_user(route):
"""Explicit get_spend_routes=false must not open global spend routes."""
user_obj = LiteLLM_UserTable(
user_id="internal_user_false_perm",
user_email="user@example.com",
user_role=LitellmUserRoles.INTERNAL_USER.value,
)
valid_token = UserAPIKeyAuth(
user_id="internal_user_false_perm",
user_role=LitellmUserRoles.INTERNAL_USER.value,
permissions={"get_spend_routes": False},
)
request = MagicMock(spec=Request)
request.query_params = {}
with pytest.raises(Exception):
RouteChecks.non_proxy_admin_allowed_routes_check(
user_obj=user_obj,
_user_role=LitellmUserRoles.INTERNAL_USER.value,
route=route,
request=request,
valid_token=valid_token,
request_data={},
)
@pytest.mark.parametrize("route", ["/audit", "/audit/some-log-id"])
def test_proxy_admin_viewer_can_access_audit_logs(route):
"""

View file

@ -1035,6 +1035,84 @@ async def test_generate_key_helper_fn_get_spend_routes_false_allowed_non_premium
assert result.get("token") or result.get("key") or result.get("token_id")
@pytest.mark.asyncio
async def test_validate_update_key_data_get_spend_routes_non_premium_returns_403(
monkeypatch,
):
"""`/key/update` must refuse truthy get_spend_routes on OSS. (#47112 review)"""
from fastapi import HTTPException
from litellm.proxy._types import (
LiteLLM_VerificationToken,
LitellmUserRoles,
UpdateKeyRequest,
UserAPIKeyAuth,
)
from litellm.proxy.management_endpoints.key_management_endpoints import (
_validate_update_key_data,
)
monkeypatch.setattr(
"litellm.proxy.management_endpoints.key_management_endpoints._require_prisma_client",
lambda client: client or object(),
)
# Short-circuit unrelated validators that need a real DB / router.
monkeypatch.setattr(
"litellm.proxy.management_endpoints.key_management_endpoints._enforce_allowed_routes_update_permission",
lambda **kwargs: None,
)
monkeypatch.setattr(
"litellm.proxy.management_endpoints.key_management_endpoints.check_allowed_passthrough_routes_caller_permission",
lambda *a, **k: None,
)
monkeypatch.setattr(
"litellm.proxy.management_endpoints.key_management_endpoints.check_permissions_caller_permission",
lambda **kwargs: None,
)
monkeypatch.setattr(
"litellm.proxy.management_endpoints.key_management_endpoints.check_disable_global_guardrails_caller_permission",
lambda *a, **k: None,
)
monkeypatch.setattr(
"litellm.proxy.management_endpoints.key_management_endpoints._validate_caller_can_change_key_ownership",
lambda **kwargs: None,
)
monkeypatch.setattr(
"litellm.proxy.management_endpoints.key_management_endpoints.common_key_access_checks",
lambda **kwargs: None,
)
monkeypatch.setattr(
"litellm.proxy.management_endpoints.key_management_endpoints.TeamMemberPermissionChecks.can_team_member_execute_key_management_endpoint",
AsyncMock(return_value=None),
)
monkeypatch.setattr(
"litellm.proxy.management_endpoints.key_management_endpoints._acting_as_team_admin_for_key_update",
AsyncMock(return_value=False),
)
data = UpdateKeyRequest(key="sk-test", permissions={"get_spend_routes": True})
existing = LiteLLM_VerificationToken(token="hashed", user_id="u1")
user = UserAPIKeyAuth(
user_id="admin",
user_role=LitellmUserRoles.PROXY_ADMIN.value,
)
with pytest.raises(HTTPException) as exc_info:
await _validate_update_key_data(
data=data,
existing_key_row=existing,
user_api_key_dict=user,
llm_router=None,
premium_user=False,
prisma_client=MagicMock(),
user_api_key_cache=MagicMock(),
)
assert exc_info.value.status_code == 403
assert "get_spend_routes" in str(exc_info.value.detail)
@pytest.mark.asyncio
async def test_generate_key_helper_fn_with_budget_fallbacks(monkeypatch):
"""Regression: /key/generate must accept `budget_fallbacks` end-to-end.