diff --git a/litellm/proxy/management_endpoints/key_management_endpoints.py b/litellm/proxy/management_endpoints/key_management_endpoints.py index 831922ec3f..c140de9819 100644 --- a/litellm/proxy/management_endpoints/key_management_endpoints.py +++ b/litellm/proxy/management_endpoints/key_management_endpoints.py @@ -4181,13 +4181,13 @@ async def list_keys( user_api_key_dict: UserAPIKeyAuth = Depends(user_api_key_auth), page: int = Query(1, description="Page number", ge=1), size: int = Query(10, description="Page size", ge=1, le=100), - user_id: Optional[str] = Query(None, description="Filter keys by user ID"), + user_id: Optional[str] = Query(None, description="Filter keys by user ID. Supports partial matching (substring, case-insensitive)."), team_id: Optional[str] = Query(None, description="Filter keys by team ID"), organization_id: Optional[str] = Query( None, description="Filter keys by organization ID" ), key_hash: Optional[str] = Query(None, description="Filter keys by key hash"), - key_alias: Optional[str] = Query(None, description="Filter keys by key alias"), + key_alias: Optional[str] = Query(None, description="Filter keys by key alias. Supports partial matching (substring, case-insensitive)."), return_full_object: bool = Query(False, description="Return full key object"), include_team_keys: bool = Query( False, description="Include all keys for teams that user is an admin of." @@ -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,9 +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"] = user_id + 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"] = key_alias + 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): @@ -4648,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 @@ -4683,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 diff --git a/tests/test_litellm/proxy/management_endpoints/test_key_management_endpoints.py b/tests/test_litellm/proxy/management_endpoints/test_key_management_endpoints.py index 12ec79d3e0..0ff276953c 100644 --- a/tests/test_litellm/proxy/management_endpoints/test_key_management_endpoints.py +++ b/tests/test_litellm/proxy/management_endpoints/test_key_management_endpoints.py @@ -6588,7 +6588,7 @@ 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 + # First: user's own keys (exact match — non-admin callers use exact matching) user_cond = or_conditions[0] assert user_cond["user_id"] == user_id @@ -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(): """