Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
37 changes: 24 additions & 13 deletions litellm/proxy/management_endpoints/key_management_endpoints.py
Original file line number Diff line number Diff line change
Expand Up @@ -530,26 +530,33 @@ def _check_allowed_routes_caller_permission(
allowed_routes: Optional[list],
user_api_key_dict: UserAPIKeyAuth,
*,
allowed_routes_was_provided: bool = False,
allow_safe_presets: bool = False,
) -> None:
"""
Only proxy admins may set `allowed_routes` on a key.

`allowed_routes` overrides the standard role-based route gate in
RouteChecks.non_proxy_admin_allowed_routes_check, so the field is
restricted to admins. Non-admins must instead use `key_type` to pick a
preset bucket — that path goes through `handle_key_type` and re-enters
this function with `allow_safe_presets=True`, which lets the derived
`llm_api_routes` / `info_routes` values through. Raw-body call sites
leave `allow_safe_presets=False` so non-admins can't write those values
directly.
Require PROXY_ADMIN when `allowed_routes` is present in the request body,
unless the caller went through the `key_type` preset flow.

Raw-body call sites pass
`allowed_routes_was_provided="allowed_routes" in data.model_fields_set` so a
caller that omits the field (model default flows through) is distinct from
one that sends any explicit value.

Post-`handle_key_type` call sites pass `allow_safe_presets=True` with the
values derived by `handle_key_type`; those values are not from the request
body, so `allowed_routes_was_provided` stays False and the safe-preset
carve-out below accepts any list of tokens in
`_NON_ADMIN_SAFE_ALLOWED_ROUTES_PRESETS`.
"""
# Empty list is the default on GenerateKeyRequest — treat as "not set".
if not allowed_routes:
if not allowed_routes_was_provided and not allowed_routes:
return
if user_api_key_dict.user_role == LitellmUserRoles.PROXY_ADMIN.value:
return
if allow_safe_presets and all(r in _NON_ADMIN_SAFE_ALLOWED_ROUTES_PRESETS for r in allowed_routes):
if (
allow_safe_presets
and allowed_routes
and all(r in _NON_ADMIN_SAFE_ALLOWED_ROUTES_PRESETS for r in allowed_routes)
):
return
raise HTTPException(
status_code=403,
Expand Down Expand Up @@ -1566,6 +1573,7 @@ async def generate_key_fn(
_check_allowed_routes_caller_permission(
allowed_routes=data.allowed_routes,
user_api_key_dict=user_api_key_dict,
allowed_routes_was_provided="allowed_routes" in data.model_fields_set,
)
_check_passthrough_routes_caller_permission(
data=data,
Expand Down Expand Up @@ -1736,6 +1744,7 @@ async def generate_service_account_key_fn(
_check_allowed_routes_caller_permission(
allowed_routes=data.allowed_routes,
user_api_key_dict=user_api_key_dict,
allowed_routes_was_provided="allowed_routes" in data.model_fields_set,
)
_check_passthrough_routes_caller_permission(
data=data,
Expand Down Expand Up @@ -2231,6 +2240,7 @@ async def _validate_update_key_data(
_check_allowed_routes_caller_permission(
allowed_routes=data.allowed_routes,
user_api_key_dict=user_api_key_dict,
allowed_routes_was_provided="allowed_routes" in data.model_fields_set,
)
_check_passthrough_routes_caller_permission(
data=data,
Expand Down Expand Up @@ -4551,6 +4561,7 @@ async def regenerate_key_fn(
_check_allowed_routes_caller_permission(
allowed_routes=data.allowed_routes,
user_api_key_dict=user_api_key_dict,
allowed_routes_was_provided="allowed_routes" in data.model_fields_set,
)
_check_passthrough_routes_caller_permission(
data=data,
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -11141,6 +11141,249 @@ async def test_non_admin_update_key_with_allowed_routes_rejected(self):
assert str(exc_info.value.code) == "403"
Comment thread
greptile-apps[bot] marked this conversation as resolved.
assert "allowed_routes" in str(exc_info.value.message)

@pytest.mark.asyncio
async def test_non_admin_update_key_explicit_empty_allowed_routes_rejected(self):
"""`update_key_fn` rejects a non-admin when `allowed_routes` 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.key_management_endpoints import (
update_key_fn,
)

data = UpdateKeyRequest(key="sk-test", allowed_routes=[])
assert "allowed_routes" in data.model_fields_set
user_api_key_dict = UserAPIKeyAuth(
user_id="internal-user-123",
user_role=LitellmUserRoles.INTERNAL_USER,
)
mock_prisma_client = AsyncMock()

with (
patch("litellm.proxy.proxy_server.prisma_client", mock_prisma_client),
patch("litellm.proxy.proxy_server.user_api_key_cache", MagicMock()),
patch("litellm.proxy.proxy_server.user_custom_key_update", None),
patch("litellm.proxy.proxy_server.llm_router", None),
patch("litellm.proxy.proxy_server.premium_user", True),
patch("litellm.proxy.proxy_server.proxy_logging_obj", MagicMock()),
patch(
"litellm.proxy.management_endpoints.key_management_endpoints._get_and_validate_existing_key",
new_callable=AsyncMock,
return_value=MagicMock(),
),
):
with pytest.raises(ProxyException) as exc_info:
await update_key_fn(
request=MagicMock(),
data=data,
user_api_key_dict=user_api_key_dict,
litellm_changed_by=None,
)
assert str(exc_info.value.code) == "403"
assert "allowed_routes" in str(exc_info.value.message)

@pytest.mark.asyncio
async def test_non_admin_update_key_explicit_null_allowed_routes_rejected(self):
"""`update_key_fn` rejects a non-admin when `allowed_routes` is
present as `null` in the request body."""
from litellm.proxy.management_endpoints.key_management_endpoints import (
update_key_fn,
)

data = UpdateKeyRequest(key="sk-test", allowed_routes=None)
assert "allowed_routes" in data.model_fields_set
user_api_key_dict = UserAPIKeyAuth(
user_id="internal-user-123",
user_role=LitellmUserRoles.INTERNAL_USER,
)
mock_prisma_client = AsyncMock()

with (
patch("litellm.proxy.proxy_server.prisma_client", mock_prisma_client),
patch("litellm.proxy.proxy_server.user_api_key_cache", MagicMock()),
patch("litellm.proxy.proxy_server.user_custom_key_update", None),
patch("litellm.proxy.proxy_server.llm_router", None),
patch("litellm.proxy.proxy_server.premium_user", True),
patch("litellm.proxy.proxy_server.proxy_logging_obj", MagicMock()),
patch(
"litellm.proxy.management_endpoints.key_management_endpoints._get_and_validate_existing_key",
new_callable=AsyncMock,
return_value=MagicMock(),
),
):
with pytest.raises(ProxyException) as exc_info:
await update_key_fn(
request=MagicMock(),
data=data,
user_api_key_dict=user_api_key_dict,
litellm_changed_by=None,
)
assert str(exc_info.value.code) == "403"
assert "allowed_routes" in str(exc_info.value.message)

@pytest.mark.asyncio
async def test_non_admin_regenerate_key_explicit_empty_allowed_routes_rejected(self):
"""`regenerate_key_fn` rejects a non-admin when `allowed_routes` is
present as `[]` in the request body."""
from litellm.proxy._types import RegenerateKeyRequest
from litellm.proxy.management_endpoints.key_management_endpoints import (
regenerate_key_fn,
)

data = RegenerateKeyRequest(key="sk-test", allowed_routes=[])
assert "allowed_routes" in data.model_fields_set
user_api_key_dict = UserAPIKeyAuth(
user_id="internal-user-123",
user_role=LitellmUserRoles.INTERNAL_USER,
)

with patch("litellm.proxy.proxy_server.premium_user", True):
with pytest.raises(ProxyException) as exc_info:
await regenerate_key_fn(
key=None,
data=data,
user_api_key_dict=user_api_key_dict,
litellm_changed_by=None,
)
assert str(exc_info.value.code) == "403"
assert "allowed_routes" in str(exc_info.value.message)

@pytest.mark.asyncio
async def test_non_admin_regenerate_key_allowed_routes_rejected_before_enterprise_gate(self):
"""`regenerate_key_fn` runs `_check_allowed_routes_caller_permission`
before the `premium_user` check, so a non-premium proxy still returns
the allowed_routes rejection (403) rather than the enterprise-license
error (500) when a non-admin sends `allowed_routes`."""
from litellm.proxy._types import RegenerateKeyRequest
from litellm.proxy.management_endpoints.key_management_endpoints import (
regenerate_key_fn,
)

data = RegenerateKeyRequest(key="sk-test", allowed_routes=["/*"])
user_api_key_dict = UserAPIKeyAuth(
user_id="internal-user-123",
user_role=LitellmUserRoles.INTERNAL_USER,
)

with patch("litellm.proxy.proxy_server.premium_user", False):
with pytest.raises(ProxyException) as exc_info:
await regenerate_key_fn(
key=None,
data=data,
user_api_key_dict=user_api_key_dict,
litellm_changed_by=None,
)
assert str(exc_info.value.code) == "403"
assert "allowed_routes" in str(exc_info.value.message)
assert "Enterprise" not in str(exc_info.value.message)

@pytest.mark.asyncio
async def test_non_admin_generate_key_explicit_empty_allowed_routes_rejected(self):
"""`generate_key_fn` rejects a non-admin when `allowed_routes` is
present as `[]` in the request body. The value matches the model
default but `model_fields_set` distinguishes the two, so the
explicit-empty case on the create path is caught."""
data = GenerateKeyRequest(key_alias="plain-key", allowed_routes=[])
assert "allowed_routes" in data.model_fields_set
user_api_key_dict = UserAPIKeyAuth(
user_id="internal-user-123",
user_role=LitellmUserRoles.INTERNAL_USER,
)
mock_prisma_client = AsyncMock()

with (
patch("litellm.proxy.proxy_server.prisma_client", mock_prisma_client),
patch("litellm.proxy.proxy_server.user_api_key_cache", MagicMock()),
patch("litellm.proxy.proxy_server.user_custom_key_generate", None),
patch(
"litellm.proxy.management_endpoints.key_management_endpoints._common_key_generation_helper",
new_callable=AsyncMock,
return_value=MagicMock(),
),
):
with pytest.raises(ProxyException) as exc_info:
await generate_key_fn(
data=data,
user_api_key_dict=user_api_key_dict,
litellm_changed_by=None,
)
assert str(exc_info.value.code) == "403"
assert "allowed_routes" in str(exc_info.value.message)

def test_helper_accepts_derived_safe_preset_for_non_admin(self):
"""`_check_allowed_routes_caller_permission` accepts a non-admin
when `allow_safe_presets=True` and `allowed_routes` is entirely
composed of `_NON_ADMIN_SAFE_ALLOWED_ROUTES_PRESETS` tokens. This
is the shape the post-`handle_key_type` recheck at line 914 uses
after deriving `["llm_api_routes"]` from `key_type=llm_api`."""
from litellm.proxy.management_endpoints.key_management_endpoints import (
_check_allowed_routes_caller_permission,
)

_check_allowed_routes_caller_permission(
allowed_routes=["llm_api_routes"],
user_api_key_dict=UserAPIKeyAuth(
user_id="internal-user-123",
user_role=LitellmUserRoles.INTERNAL_USER,
),
allow_safe_presets=True,
)
_check_allowed_routes_caller_permission(
allowed_routes=["info_routes"],
user_api_key_dict=UserAPIKeyAuth(
user_id="internal-user-123",
user_role=LitellmUserRoles.INTERNAL_USER,
),
allow_safe_presets=True,
)

def test_helper_rejects_derived_unsafe_preset_for_non_admin(self):
"""`_check_allowed_routes_caller_permission` rejects a non-admin
when `allow_safe_presets=True` but the derived value is outside
`_NON_ADMIN_SAFE_ALLOWED_ROUTES_PRESETS` (for example
`["management_routes"]` from `key_type=management`)."""
from fastapi import HTTPException

from litellm.proxy.management_endpoints.key_management_endpoints import (
_check_allowed_routes_caller_permission,
)

with pytest.raises(HTTPException) as exc_info:
_check_allowed_routes_caller_permission(
allowed_routes=["management_routes"],
user_api_key_dict=UserAPIKeyAuth(
user_id="internal-user-123",
user_role=LitellmUserRoles.INTERNAL_USER,
),
allow_safe_presets=True,
)
assert exc_info.value.status_code == 403
assert "allowed_routes" in str(exc_info.value.detail)

def test_helper_rejects_when_provided_and_none_without_typeerror(self):
"""`_check_allowed_routes_caller_permission` returns a 403 (not a
TypeError from iterating `None`) when a caller ever combines
`allowed_routes_was_provided=True` with `allowed_routes=None` and
`allow_safe_presets=True`. Pins the `and allowed_routes` guard on
the safe-preset branch."""
from fastapi import HTTPException

from litellm.proxy.management_endpoints.key_management_endpoints import (
_check_allowed_routes_caller_permission,
)

with pytest.raises(HTTPException) as exc_info:
_check_allowed_routes_caller_permission(
allowed_routes=None,
user_api_key_dict=UserAPIKeyAuth(
user_id="internal-user-123",
user_role=LitellmUserRoles.INTERNAL_USER,
),
allowed_routes_was_provided=True,
allow_safe_presets=True,
)
assert exc_info.value.status_code == 403
assert "allowed_routes" in str(exc_info.value.detail)


def test_jinja_prompt_manager_is_sandboxed():
"""
Expand Down
Loading