mirror of
https://github.com/BerriAI/litellm.git
synced 2026-08-28 05:25:59 +00:00
Merge cb3d9c23db into 47cfd3f5ca
This commit is contained in:
commit
c0a01af287
2 changed files with 88 additions and 10 deletions
|
|
@ -226,17 +226,14 @@ class RouteChecks:
|
|||
route (str): The route being accessed
|
||||
|
||||
Raises:
|
||||
Exception: With user role and masked user_id information
|
||||
HTTPException: 403, with user role and masked user_id information
|
||||
"""
|
||||
user_role = "unknown"
|
||||
user_id = "unknown"
|
||||
if user_obj is not None:
|
||||
user_role = user_obj.user_role or "unknown"
|
||||
user_id = user_obj.user_id or "unknown"
|
||||
|
||||
user_role: Final = (user_obj.user_role if user_obj is not None else None) or "unknown"
|
||||
user_id: Final = (user_obj.user_id if user_obj is not None else None) or "unknown"
|
||||
masked_user_id: Final = RouteChecks._mask_user_id(user_id)
|
||||
raise Exception(
|
||||
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}"
|
||||
raise HTTPException(
|
||||
status_code=status.HTTP_403_FORBIDDEN,
|
||||
detail=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
|
||||
|
|
|
|||
|
|
@ -1,6 +1,7 @@
|
|||
import os
|
||||
from datetime import datetime
|
||||
from unittest.mock import MagicMock, patch
|
||||
from typing import Final
|
||||
from unittest.mock import AsyncMock, MagicMock, patch
|
||||
|
||||
|
||||
import pytest
|
||||
|
|
@ -57,6 +58,43 @@ def test_non_admin_config_update_route_rejected():
|
|||
assert "Your role=internal_user" in str(exc_info.value)
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"route",
|
||||
["/policies/list", "/prompts/list", "/in_product_nudges", "/config/update"],
|
||||
)
|
||||
def test_admin_only_route_denial_is_forbidden_and_carries_status(route):
|
||||
"""A valid non-admin key denied an admin-only route is an authorization
|
||||
failure, so it must surface as 403 and carry the status code the metrics
|
||||
layer reads, not an unlabelled 401.
|
||||
|
||||
Regression test for https://github.com/BerriAI/litellm/issues/37108
|
||||
"""
|
||||
user_obj = LiteLLM_UserTable(
|
||||
user_id="test_user",
|
||||
user_email="test@example.com",
|
||||
user_role=LitellmUserRoles.INTERNAL_USER.value,
|
||||
)
|
||||
valid_token = UserAPIKeyAuth(
|
||||
user_id="test_user",
|
||||
user_role=LitellmUserRoles.INTERNAL_USER.value,
|
||||
)
|
||||
request = MagicMock(spec=Request)
|
||||
request.query_params = {}
|
||||
|
||||
with pytest.raises(HTTPException) as exc_info:
|
||||
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={},
|
||||
)
|
||||
|
||||
assert exc_info.value.status_code == 403
|
||||
assert f"Route={route}" in str(exc_info.value.detail)
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"role",
|
||||
[
|
||||
|
|
@ -3521,3 +3559,46 @@ def test_agent_registry_route_gate_open_to_non_admin_roles(user_role, method, ro
|
|||
valid_token=valid_token,
|
||||
request_data={},
|
||||
)
|
||||
|
||||
|
||||
@pytest.mark.parametrize("route", ["/policies/list", "/prompts/list"])
|
||||
def test_admin_only_denial_reaches_the_client_as_403(route):
|
||||
"""Drives a real request so the status the caller sees is asserted, not just the
|
||||
status the check raises. A regression that swallowed it and re-raised 401 would
|
||||
leave the direct-call tests above green.
|
||||
"""
|
||||
from fastapi.testclient import TestClient
|
||||
|
||||
import litellm.proxy.proxy_server as ps
|
||||
from litellm.proxy.proxy_server import app
|
||||
|
||||
key: Final = "sk-non-admin-1234"
|
||||
valid_token: Final = UserAPIKeyAuth(
|
||||
token=key, key_name=key, user_id="test_user", user_role=LitellmUserRoles.INTERNAL_USER.value
|
||||
)
|
||||
user_obj: Final = LiteLLM_UserTable(
|
||||
user_id="test_user", user_email="test@example.com", user_role=LitellmUserRoles.INTERNAL_USER.value
|
||||
)
|
||||
|
||||
class _StubIdentityStore:
|
||||
def __init__(self, *args, **kwargs):
|
||||
pass
|
||||
|
||||
async def resolve(self, *args, **kwargs):
|
||||
return MagicMock(source_key=valid_token)
|
||||
|
||||
@staticmethod
|
||||
def key_from_principal(principal):
|
||||
return valid_token
|
||||
|
||||
with (
|
||||
patch.object(ps, "master_key", "sk-master-1234"),
|
||||
patch.object(ps, "prisma_client", MagicMock()),
|
||||
patch("litellm.proxy.auth.user_api_key_auth.IdentityStore", _StubIdentityStore),
|
||||
patch("litellm.proxy.auth.user_api_key_auth.get_user_object", new=AsyncMock(return_value=user_obj)),
|
||||
):
|
||||
response: Final = TestClient(app, raise_server_exceptions=False).get(
|
||||
route, headers={"Authorization": f"Bearer {key}"}
|
||||
)
|
||||
|
||||
assert response.status_code == 403, f"{route} returned {response.status_code}: {response.text[:200]}"
|
||||
|
|
|
|||
Loading…
Add table
Reference in a new issue