[Fix] Derive use_substring_matching from role, not auto-set path

Substring matching for user_id and key_alias on /key/list is now gated
by caller role (admin only) rather than by whether user_id was auto-set.
This closes the gap where a non-admin explicitly passing their own
user_id still got contains-based filtering, which could surface keys
belonging to other users with overlapping IDs.

Adds three unit tests covering the substring, exact-match, and default
code paths in _build_key_filter_conditions. Restores the existing
member-team test to assert exact matching as a security regression guard.
This commit is contained in:
Ryan Crabbe 2026-03-28 16:50:09 -07:00
parent 65df501f15
commit a59acc8f8a
No known key found for this signature in database
2 changed files with 116 additions and 12 deletions

View file

@ -4280,10 +4280,12 @@ async def list_keys(
else:
admin_team_ids = None
if not user_id and user_api_key_dict.user_role not in [
use_substring_matching = user_api_key_dict.user_role in [
LitellmUserRoles.PROXY_ADMIN.value,
LitellmUserRoles.PROXY_ADMIN_VIEW_ONLY.value,
]:
]
if not user_id and not use_substring_matching:
user_id = user_api_key_dict.user_id
response = await _list_key_helper(
@ -4305,6 +4307,7 @@ async def list_keys(
status=status,
project_id=project_id,
access_group_id=access_group_id,
use_substring_matching=use_substring_matching,
)
verbose_proxy_logger.debug("Successfully prepared response")
@ -4522,6 +4525,7 @@ def _build_key_filter_conditions(
include_created_by_keys: bool = False,
project_id: Optional[str] = None,
access_group_id: Optional[str] = None,
use_substring_matching: bool = False,
) -> Dict[str, Union[str, Dict[str, Any], List[Dict[str, Any]]]]:
"""Build filter conditions for key listing.
@ -4543,15 +4547,21 @@ def _build_key_filter_conditions(
# Base conditions for user's own keys
user_condition: Dict[str, Any] = {}
if user_id and isinstance(user_id, str):
user_condition["user_id"] = {
"contains": user_id,
"mode": "insensitive",
}
if use_substring_matching:
user_condition["user_id"] = {
"contains": user_id,
"mode": "insensitive",
}
else:
user_condition["user_id"] = user_id
if key_alias and isinstance(key_alias, str):
user_condition["key_alias"] = {
"contains": key_alias,
"mode": "insensitive",
}
if use_substring_matching:
user_condition["key_alias"] = {
"contains": key_alias,
"mode": "insensitive",
}
else:
user_condition["key_alias"] = key_alias
if exclude_team_id and isinstance(exclude_team_id, str):
user_condition["team_id"] = {"not": exclude_team_id}
if organization_id and isinstance(organization_id, str):
@ -4654,6 +4664,7 @@ async def _list_key_helper(
status: Optional[str] = None,
project_id: Optional[str] = None,
access_group_id: Optional[str] = None,
use_substring_matching: bool = False,
) -> KeyListResponseObject:
"""
Helper function to list keys
@ -4689,6 +4700,7 @@ async def _list_key_helper(
include_created_by_keys=include_created_by_keys,
project_id=project_id,
access_group_id=access_group_id,
use_substring_matching=use_substring_matching,
)
# Calculate skip for pagination

View file

@ -6588,9 +6588,9 @@ async def test_build_key_filter_member_team_service_accounts():
# Should have 2 conditions: user's own keys + member team service accounts
assert len(or_conditions) == 2
# First: user's own keys (substring match)
# First: user's own keys (exact match — non-admin callers use exact matching)
user_cond = or_conditions[0]
assert user_cond["user_id"] == {"contains": user_id, "mode": "insensitive"}
assert user_cond["user_id"] == user_id
# Second: service accounts for member teams (user_id=None AND team_id in member teams)
service_account_cond = or_conditions[1]
@ -6988,6 +6988,98 @@ async def test_build_key_filter_team_id_scoped():
)
@pytest.mark.asyncio
async def test_build_key_filter_admin_substring_matching():
"""
Admin callers get substring (contains + insensitive) matching for user_id
and key_alias when use_substring_matching=True.
"""
from litellm.proxy.management_endpoints.key_management_endpoints import (
_build_key_filter_conditions,
)
user_id = "alice"
key_alias = "prod"
where = _build_key_filter_conditions(
user_id=user_id,
team_id=None,
organization_id=None,
key_alias=key_alias,
key_hash=None,
exclude_team_id=None,
admin_team_ids=None,
member_team_ids=None,
include_created_by_keys=False,
use_substring_matching=True,
)
# Single OR condition is flattened into the top-level where dict
assert where["user_id"] == {"contains": user_id, "mode": "insensitive"}
assert where["key_alias"] == {"contains": key_alias, "mode": "insensitive"}
@pytest.mark.asyncio
async def test_build_key_filter_non_admin_exact_matching():
"""
Non-admin callers get exact matching for user_id and key_alias when
use_substring_matching=False (the default). This prevents a user whose
ID is a substring of another user's ID from seeing that user's keys.
"""
from litellm.proxy.management_endpoints.key_management_endpoints import (
_build_key_filter_conditions,
)
user_id = "alice@example.com"
key_alias = "my-key"
where = _build_key_filter_conditions(
user_id=user_id,
team_id=None,
organization_id=None,
key_alias=key_alias,
key_hash=None,
exclude_team_id=None,
admin_team_ids=None,
member_team_ids=None,
include_created_by_keys=False,
use_substring_matching=False,
)
# Single OR condition is flattened into the top-level where dict
# Exact match — no contains/insensitive wrapping
assert where["user_id"] == user_id
assert where["key_alias"] == key_alias
@pytest.mark.asyncio
async def test_build_key_filter_default_is_exact_matching():
"""
The default for use_substring_matching is False, ensuring backward
compatibility callers that don't pass the flag get exact matching.
"""
from litellm.proxy.management_endpoints.key_management_endpoints import (
_build_key_filter_conditions,
)
user_id = "user-123"
where = _build_key_filter_conditions(
user_id=user_id,
team_id=None,
organization_id=None,
key_alias=None,
key_hash=None,
exclude_team_id=None,
admin_team_ids=None,
member_team_ids=None,
include_created_by_keys=False,
)
# Single OR condition is flattened into the top-level where dict
assert where["user_id"] == user_id
@pytest.mark.asyncio
async def test_get_member_team_ids():
"""