fix(proxy): re-validate user_id after /user/info re-parses query (#27009)

* fix(proxy): re-validate user_id ownership after /user/info re-parses query

The route-level access check in `RouteChecks.non_proxy_admin_allowed_routes_check`
reads `request.query_params.get("user_id")`, which decodes literal `+` to
spaces. The endpoint then re-parses the raw query string with `urllib.unquote`
in `get_user_id_from_request` to preserve `+` characters (so plus-addressed
emails work as user_ids). Those two paths produce different ids: a caller
who registered a user_id containing a literal space could pass the route
check and then read another user's row by sending the encoded `+` form.

Add `_enforce_user_info_access` and call it after `_normalize_user_info_user_id`
returns the final id. Proxy admin / view-only admin still bypass; everyone
else must match the resolved user_id (or have no user_id, which falls back
to the caller's own id later in the handler).

Tests cover the admin bypass, owner-match path, and the cross-user lookup
that this change blocks.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

* fix(proxy): apply user_info ownership check to PROXY_ADMIN_VIEW_ONLY

`_enforce_user_info_access` was bypassing both PROXY_ADMIN and
PROXY_ADMIN_VIEW_ONLY, but the upstream route check in
`RouteChecks.non_proxy_admin_allowed_routes_check` only treats
PROXY_ADMIN as a true admin for the `/user/info` route — view-only
admins go through the `user_id == valid_token.user_id` enforcement
along with regular users. Mirroring that asymmetry left the same
encoded-`+` bypass open for view-only admins whose user_id contains a
literal space.

Drop the PROXY_ADMIN_VIEW_ONLY exemption so the post-decode re-check
matches the upstream rule. Update tests: a view-only admin must now
be blocked from cross-user lookups but still allowed to read their
own row.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

---------

Co-authored-by: yuneng-jiang <yuneng@berri.ai>
Co-authored-by: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
This commit is contained in:
stuxf 2026-05-01 17:38:14 -07:00 committed by GitHub
parent b80246971b
commit e8818d69e0
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
2 changed files with 146 additions and 5 deletions

View file

@ -619,6 +619,40 @@ def _normalize_user_info_user_id(
return user_id
def _enforce_user_info_access(
user_id: Optional[str], user_api_key_dict: UserAPIKeyAuth
) -> None:
"""Re-validate that the caller may read the resolved ``user_id`` after
URL-decoding has been finalized.
The route-level check in ``RouteChecks.non_proxy_admin_allowed_routes_check``
runs against ``request.query_params``, which decodes a literal ``+`` to a
space. ``_normalize_user_info_user_id`` then re-parses the raw query with
``unquote`` so the endpoint can return rows for user_ids that contain ``+``
(e.g. plus-addressed emails). That asymmetry let an attacker who registered
a username with a literal space pass the route check and then read another
user's row by sending the encoded ``+`` form. Re-checking ownership here
closes the gap without changing the supported user_id grammar.
"""
if user_id is None:
return
# Only true proxy admin bypasses ownership. PROXY_ADMIN_VIEW_ONLY is
# subject to the same `user_id == valid_token.user_id` rule that
# `RouteChecks.non_proxy_admin_allowed_routes_check` applies upstream
# for the `/user/info` route.
if user_api_key_dict.user_role == LitellmUserRoles.PROXY_ADMIN:
return
if user_id == user_api_key_dict.user_id:
return
raise HTTPException(
status_code=status.HTTP_403_FORBIDDEN,
detail=(
f"key not allowed to access this user's info. user_id={user_id}, "
f"key's user_id={user_api_key_dict.user_id}"
),
)
async def _get_user_info_teams(
prisma_client: Any,
user_id: Optional[str],
@ -733,6 +767,7 @@ async def user_info( # noqa: PLR0915
try:
user_id = _normalize_user_info_user_id(request=request, user_id=user_id)
_enforce_user_info_access(user_id=user_id, user_api_key_dict=user_api_key_dict)
if prisma_client is None:
raise Exception(

View file

@ -2050,11 +2050,13 @@ async def test_delete_user_rejects_org_admin_deleting_outside_scope(mocker):
assert exc.value.status_code == 403
# Critical: no delete_many calls should have executed.
assert not hasattr(
mock_prisma_client.db.litellm_verificationtoken.delete_many, "mock_calls"
) or len(
mock_prisma_client.db.litellm_verificationtoken.delete_many.mock_calls
) == 0
assert (
not hasattr(
mock_prisma_client.db.litellm_verificationtoken.delete_many, "mock_calls"
)
or len(mock_prisma_client.db.litellm_verificationtoken.delete_many.mock_calls)
== 0
)
@pytest.mark.asyncio
@ -2732,3 +2734,107 @@ class TestGetUserIdFromRequestValidation:
request = self._make_request(f"user_id={exact_id}")
result = get_user_id_from_request(request)
assert result == exact_id
# ---------------------------------------------------------------------------
# VERIA-60: /user/info post-decode re-authorization
# ---------------------------------------------------------------------------
def test_enforce_user_info_access_admin_bypass():
"""Proxy admins must always be allowed past the re-check."""
from litellm.proxy._types import LitellmUserRoles, UserAPIKeyAuth
from litellm.proxy.management_endpoints.internal_user_endpoints import (
_enforce_user_info_access,
)
admin = UserAPIKeyAuth(
user_id="admin", user_role=LitellmUserRoles.PROXY_ADMIN.value
)
# Should not raise even when querying a different user
_enforce_user_info_access(user_id="someone_else", user_api_key_dict=admin)
def test_enforce_user_info_access_view_only_admin_blocked_from_other_users():
"""PROXY_ADMIN_VIEW_ONLY is not a true admin for /user/info — the upstream
route check applies the same `user_id == valid_token.user_id` rule, so the
re-check here must mirror that and deny cross-user lookups."""
import pytest
from fastapi import HTTPException
from litellm.proxy._types import LitellmUserRoles, UserAPIKeyAuth
from litellm.proxy.management_endpoints.internal_user_endpoints import (
_enforce_user_info_access,
)
viewer = UserAPIKeyAuth(
user_id="viewer",
user_role=LitellmUserRoles.PROXY_ADMIN_VIEW_ONLY.value,
)
with pytest.raises(HTTPException) as exc_info:
_enforce_user_info_access(user_id="someone_else", user_api_key_dict=viewer)
assert exc_info.value.status_code == 403
def test_enforce_user_info_access_view_only_admin_can_read_own():
from litellm.proxy._types import LitellmUserRoles, UserAPIKeyAuth
from litellm.proxy.management_endpoints.internal_user_endpoints import (
_enforce_user_info_access,
)
viewer = UserAPIKeyAuth(
user_id="viewer",
user_role=LitellmUserRoles.PROXY_ADMIN_VIEW_ONLY.value,
)
_enforce_user_info_access(user_id="viewer", user_api_key_dict=viewer)
def test_enforce_user_info_access_owner_allowed():
from litellm.proxy._types import LitellmUserRoles, UserAPIKeyAuth
from litellm.proxy.management_endpoints.internal_user_endpoints import (
_enforce_user_info_access,
)
user = UserAPIKeyAuth(
user_id="alice", user_role=LitellmUserRoles.INTERNAL_USER.value
)
_enforce_user_info_access(user_id="alice", user_api_key_dict=user)
def test_enforce_user_info_access_no_user_id_allowed():
"""No user_id in query → handler resolves to caller's own id later, so
this branch must not raise."""
from litellm.proxy._types import LitellmUserRoles, UserAPIKeyAuth
from litellm.proxy.management_endpoints.internal_user_endpoints import (
_enforce_user_info_access,
)
user = UserAPIKeyAuth(
user_id="alice", user_role=LitellmUserRoles.INTERNAL_USER.value
)
_enforce_user_info_access(user_id=None, user_api_key_dict=user)
def test_enforce_user_info_access_blocks_cross_user_lookup():
"""A non-admin caller may not query another user's row, even if URL
re-parsing produced a user_id that differs from the one the route check
saw (the VERIA-60 bypass)."""
import pytest
from fastapi import HTTPException
from litellm.proxy._types import LitellmUserRoles, UserAPIKeyAuth
from litellm.proxy.management_endpoints.internal_user_endpoints import (
_enforce_user_info_access,
)
attacker = UserAPIKeyAuth(
user_id="attacker space", # original (URL-decoded) id seen by route check
user_role=LitellmUserRoles.INTERNAL_USER.value,
)
with pytest.raises(HTTPException) as exc_info:
# Re-parsed id (with literal '+') belongs to the victim
_enforce_user_info_access(user_id="victim+target", user_api_key_dict=attacker)
assert exc_info.value.status_code == 403
assert "key not allowed to access this user's info" in str(exc_info.value.detail)