diff --git a/litellm/proxy/management_endpoints/key_management_endpoints.py b/litellm/proxy/management_endpoints/key_management_endpoints.py index d82febea8fe..c140de98191 100644 --- a/litellm/proxy/management_endpoints/key_management_endpoints.py +++ b/litellm/proxy/management_endpoints/key_management_endpoints.py @@ -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 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 c32d84b111d..0ff276953c6 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,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(): """