mirror of
https://github.com/BerriAI/litellm.git
synced 2026-09-09 22:31:41 +00:00
feat: add /v2/user/info endpoint - lightweight user info with RBAC
- Add UserInfoV2Response type in _types.py (returns only user object, no keys/teams) - Add /v2/user/info endpoint handler with proper access control: - Proxy admins can query any user - Team admins can query users in their teams - Internal users can query themselves only - Returns 404 for unauthorized/not-found (not 403) - Add /v2/user/info to info_routes in LiteLLMRoutes - Add route check passthrough in route_checks.py - Add get_user_v2() method to Python client Co-authored-by: yuneng-jiang <yuneng-jiang@users.noreply.github.com>
This commit is contained in:
parent
2495579210
commit
81e3a2e421
4 changed files with 207 additions and 1 deletions
|
|
@ -484,6 +484,7 @@ class LiteLLMRoutes(enum.Enum):
|
|||
"/organization/list",
|
||||
"/team/available",
|
||||
"/user/info",
|
||||
"/v2/user/info",
|
||||
"/model/info",
|
||||
"/v1/model/info",
|
||||
"/v2/model/info",
|
||||
|
|
@ -2552,6 +2553,30 @@ class UserInfoResponse(LiteLLMPydanticObjectBase):
|
|||
teams: List
|
||||
|
||||
|
||||
class UserInfoV2Response(LiteLLMPydanticObjectBase):
|
||||
"""
|
||||
Response model for GET /v2/user/info
|
||||
|
||||
Returns ONLY the user object - no keys, no teams objects.
|
||||
This is a lightweight alternative to UserInfoResponse.
|
||||
"""
|
||||
|
||||
user_id: str
|
||||
user_email: Optional[str] = None
|
||||
user_alias: Optional[str] = None
|
||||
user_role: Optional[str] = None
|
||||
spend: float = 0.0
|
||||
max_budget: Optional[float] = None
|
||||
models: List[str] = []
|
||||
budget_duration: Optional[str] = None
|
||||
budget_reset_at: Optional[datetime] = None
|
||||
metadata: Optional[dict] = None
|
||||
created_at: Optional[datetime] = None
|
||||
updated_at: Optional[datetime] = None
|
||||
sso_user_id: Optional[str] = None
|
||||
teams: List[str] = [] # Just team IDs, not full team objects
|
||||
|
||||
|
||||
class LiteLLM_Config(LiteLLMPydanticObjectBase):
|
||||
param_name: str
|
||||
param_value: Dict
|
||||
|
|
|
|||
|
|
@ -183,6 +183,9 @@ class RouteChecks:
|
|||
user_id, valid_token.user_id
|
||||
),
|
||||
)
|
||||
elif route == "/v2/user/info":
|
||||
# handled by the endpoint itself (full RBAC in handler)
|
||||
pass
|
||||
elif route == "/model/info":
|
||||
# /model/info just shows models user has access to
|
||||
pass
|
||||
|
|
|
|||
|
|
@ -35,6 +35,18 @@ class UsersManagementClient:
|
|||
response.raise_for_status()
|
||||
return response.json()
|
||||
|
||||
def get_user_v2(self, user_id: Optional[str] = None) -> Dict[str, Any]:
|
||||
"""Get user info v2 - lightweight, returns only user object (GET /v2/user/info)"""
|
||||
url = f"{self.base_url}/v2/user/info"
|
||||
params = {"user_id": user_id} if user_id else {}
|
||||
response = requests.get(url, headers=self._get_headers(), params=params)
|
||||
if response.status_code == 401:
|
||||
raise UnauthorizedError(response.text)
|
||||
if response.status_code == 404:
|
||||
raise NotFoundError(response.text)
|
||||
response.raise_for_status()
|
||||
return response.json()
|
||||
|
||||
def create_user(self, user_data: Dict[str, Any]) -> Dict[str, Any]:
|
||||
"""Create a new user (POST /user/new)"""
|
||||
url = f"{self.base_url}/user/new"
|
||||
|
|
|
|||
|
|
@ -31,7 +31,10 @@ from litellm.proxy.management_endpoints.common_daily_activity import (
|
|||
get_daily_activity_aggregated,
|
||||
)
|
||||
from litellm.proxy.auth.auth_checks import get_team_object, get_user_object
|
||||
from litellm.proxy.management_endpoints.common_utils import _user_has_admin_view
|
||||
from litellm.proxy.management_endpoints.common_utils import (
|
||||
_is_user_team_admin,
|
||||
_user_has_admin_view,
|
||||
)
|
||||
from litellm.proxy.management_endpoints.key_management_endpoints import (
|
||||
generate_key_helper_fn,
|
||||
prepare_metadata_fields,
|
||||
|
|
@ -715,6 +718,169 @@ async def user_info(
|
|||
raise handle_exception_on_proxy(e)
|
||||
|
||||
|
||||
async def _check_user_info_v2_access(
|
||||
user_api_key_dict: UserAPIKeyAuth,
|
||||
target_user_id: str,
|
||||
) -> bool:
|
||||
"""
|
||||
Check if the caller is allowed to access the target user's info.
|
||||
|
||||
Returns True if access is allowed, False otherwise.
|
||||
|
||||
Access rules:
|
||||
1. Proxy admins / proxy admin viewers can access any user
|
||||
2. User can access their own info
|
||||
3. Team admins can access info of users in their teams
|
||||
"""
|
||||
from litellm.proxy.proxy_server import prisma_client
|
||||
|
||||
# Rule 1: Proxy admins
|
||||
if _user_has_admin_view(user_api_key_dict):
|
||||
return True
|
||||
|
||||
# Rule 2: Self-lookup
|
||||
if user_api_key_dict.user_id == target_user_id:
|
||||
return True
|
||||
|
||||
# Rule 3: Team admins can look up users in their teams
|
||||
if prisma_client is not None and user_api_key_dict.user_id is not None:
|
||||
try:
|
||||
# Get caller's teams
|
||||
caller_user = await prisma_client.db.litellm_usertable.find_unique(
|
||||
where={"user_id": user_api_key_dict.user_id}
|
||||
)
|
||||
if caller_user is not None and caller_user.teams:
|
||||
# Get teams where caller is admin
|
||||
teams = await prisma_client.db.litellm_teamtable.find_many(
|
||||
where={"team_id": {"in": caller_user.teams}}
|
||||
)
|
||||
for team in teams:
|
||||
team_obj = LiteLLM_TeamTable(**team.model_dump())
|
||||
if _is_user_team_admin(
|
||||
user_api_key_dict=user_api_key_dict, team_obj=team_obj
|
||||
):
|
||||
# Check if target user is in this team
|
||||
target_user = await prisma_client.db.litellm_usertable.find_unique(
|
||||
where={"user_id": target_user_id}
|
||||
)
|
||||
if (
|
||||
target_user is not None
|
||||
and team.team_id in (target_user.teams or [])
|
||||
):
|
||||
return True
|
||||
except Exception:
|
||||
verbose_proxy_logger.debug(
|
||||
f"Error checking team admin access for user {user_api_key_dict.user_id}"
|
||||
)
|
||||
|
||||
return False
|
||||
|
||||
|
||||
@router.get(
|
||||
"/v2/user/info",
|
||||
tags=["Internal User management"],
|
||||
dependencies=[Depends(user_api_key_auth)],
|
||||
response_model=UserInfoV2Response,
|
||||
)
|
||||
@management_endpoint_wrapper
|
||||
async def user_info_v2(
|
||||
request: Request,
|
||||
user_id: Optional[str] = fastapi.Query(
|
||||
default=None, description="User ID in the request parameters"
|
||||
),
|
||||
user_api_key_dict: UserAPIKeyAuth = Depends(user_api_key_auth),
|
||||
):
|
||||
"""
|
||||
Lightweight endpoint to get user info. Returns only the user object — no keys, no teams objects.
|
||||
|
||||
This is the v2 replacement for /user/info, designed to avoid the "god endpoint" problem
|
||||
where the old endpoint loaded all keys and teams into memory.
|
||||
|
||||
Access control:
|
||||
- Proxy admins can query any user
|
||||
- Team admins can query users within their teams
|
||||
- Internal users can only query themselves (omit user_id or pass own)
|
||||
- Returns 404 for non-existent users or unauthorized access
|
||||
|
||||
Example request:
|
||||
```
|
||||
curl -X GET 'http://localhost:4000/v2/user/info?user_id=user123' \\
|
||||
--header 'Authorization: Bearer sk-1234'
|
||||
```
|
||||
"""
|
||||
from litellm.proxy.proxy_server import prisma_client
|
||||
|
||||
try:
|
||||
if prisma_client is None:
|
||||
raise HTTPException(
|
||||
status_code=500,
|
||||
detail=CommonProxyErrors.db_not_connected_error.value,
|
||||
)
|
||||
|
||||
# Handle URL encoding for + characters
|
||||
if user_id is not None and " " in user_id:
|
||||
user_id = get_user_id_from_request(request=request)
|
||||
|
||||
# Default to self-lookup if no user_id provided
|
||||
if user_id is None:
|
||||
user_id = user_api_key_dict.user_id
|
||||
|
||||
if user_id is None:
|
||||
raise HTTPException(
|
||||
status_code=400,
|
||||
detail="user_id is required. Either pass it as a query parameter or authenticate with a user-bound key.",
|
||||
)
|
||||
|
||||
# Check access
|
||||
has_access = await _check_user_info_v2_access(
|
||||
user_api_key_dict=user_api_key_dict,
|
||||
target_user_id=user_id,
|
||||
)
|
||||
|
||||
if not has_access:
|
||||
raise HTTPException(
|
||||
status_code=404,
|
||||
detail=f"User not found: {user_id}",
|
||||
)
|
||||
|
||||
# Fetch user from DB
|
||||
user_row = await prisma_client.db.litellm_usertable.find_unique(
|
||||
where={"user_id": user_id}
|
||||
)
|
||||
|
||||
if user_row is None:
|
||||
raise HTTPException(
|
||||
status_code=404,
|
||||
detail=f"User not found: {user_id}",
|
||||
)
|
||||
|
||||
user_data = user_row.model_dump()
|
||||
|
||||
return UserInfoV2Response(
|
||||
user_id=user_data.get("user_id", user_id),
|
||||
user_email=user_data.get("user_email"),
|
||||
user_alias=user_data.get("user_alias"),
|
||||
user_role=user_data.get("user_role"),
|
||||
spend=user_data.get("spend", 0.0),
|
||||
max_budget=user_data.get("max_budget"),
|
||||
models=user_data.get("models") or [],
|
||||
budget_duration=user_data.get("budget_duration"),
|
||||
budget_reset_at=user_data.get("budget_reset_at"),
|
||||
metadata=user_data.get("metadata"),
|
||||
created_at=user_data.get("created_at"),
|
||||
updated_at=user_data.get("updated_at"),
|
||||
sso_user_id=user_data.get("sso_user_id"),
|
||||
teams=user_data.get("teams") or [],
|
||||
)
|
||||
except Exception as e:
|
||||
verbose_proxy_logger.exception(
|
||||
"litellm.proxy.proxy_server.user_info_v2(): Exception occured - {}".format(
|
||||
str(e)
|
||||
)
|
||||
)
|
||||
raise handle_exception_on_proxy(e)
|
||||
|
||||
|
||||
async def _get_user_info_for_proxy_admin(user_api_key_dict: UserAPIKeyAuth):
|
||||
"""
|
||||
Admin UI Endpoint - Returns All Teams and Keys when Proxy Admin is querying
|
||||
|
|
|
|||
Loading…
Add table
Reference in a new issue