|
| 1 | +from __future__ import annotations |
| 2 | + |
| 3 | +import pytest |
| 4 | +import requests |
| 5 | +from simple_logger.logger import get_logger |
| 6 | + |
| 7 | +from tests.model_serving.maas_billing.maas_subscription.utils import ( |
| 8 | + list_api_keys, |
| 9 | + get_api_key, |
| 10 | + revoke_api_key, |
| 11 | +) |
| 12 | + |
| 13 | +LOGGER = get_logger(name=__name__) |
| 14 | + |
| 15 | + |
| 16 | +@pytest.mark.parametrize("ocp_token_for_actor", [{"type": "free"}], indirect=True) |
| 17 | +@pytest.mark.usefixtures( |
| 18 | + "maas_subscription_controller_enabled_latest", |
| 19 | + "maas_gateway_api", |
| 20 | + "maas_api_gateway_reachable", |
| 21 | +) |
| 22 | +class TestAPIKeyAuthorization: |
| 23 | + """Tests for MaaS API key admin and non-admin access control.""" |
| 24 | + |
| 25 | + @pytest.mark.tier1 |
| 26 | + def test_admin_manage_other_users_keys( |
| 27 | + self, |
| 28 | + request_session_http: requests.Session, |
| 29 | + base_url: str, |
| 30 | + admin_ocp_token: str, |
| 31 | + active_api_key_id: str, |
| 32 | + free_user_username: str, |
| 33 | + ) -> None: |
| 34 | + """Verify an admin can search for another user's keys by username and revoke them.""" |
| 35 | + list_resp, list_body = list_api_keys( |
| 36 | + request_session_http=request_session_http, |
| 37 | + base_url=base_url, |
| 38 | + ocp_user_token=admin_ocp_token, |
| 39 | + filters={"username": free_user_username, "status": ["active"]}, |
| 40 | + sort={"by": "created_at", "order": "desc"}, |
| 41 | + pagination={"limit": 50, "offset": 0}, |
| 42 | + ) |
| 43 | + assert list_resp.status_code == 200, ( |
| 44 | + f"Expected 200 on admin search for username='{free_user_username}', " |
| 45 | + f"got {list_resp.status_code}: {list_resp.text[:200]}" |
| 46 | + ) |
| 47 | + items: list[dict] = list_body.get("items") or list_body.get("data") or [] |
| 48 | + key_ids = [item["id"] for item in items] |
| 49 | + assert active_api_key_id in key_ids, ( |
| 50 | + f"Expected free user's key id={active_api_key_id} in admin search results " |
| 51 | + f"for username='{free_user_username}', found ids={key_ids}" |
| 52 | + ) |
| 53 | + LOGGER.info(f"[authz] Admin found {len(items)} active key(s) for user='{free_user_username}'") |
| 54 | + |
| 55 | + revoke_resp, revoke_body = revoke_api_key( |
| 56 | + request_session_http=request_session_http, |
| 57 | + base_url=base_url, |
| 58 | + key_id=active_api_key_id, |
| 59 | + ocp_user_token=admin_ocp_token, |
| 60 | + ) |
| 61 | + assert revoke_resp.status_code == 200, ( |
| 62 | + f"Expected 200 on admin DELETE /v1/api-keys/{active_api_key_id}, " |
| 63 | + f"got {revoke_resp.status_code}: {revoke_resp.text[:200]}" |
| 64 | + ) |
| 65 | + assert revoke_body.get("status") == "revoked", ( |
| 66 | + f"Expected status='revoked' in admin revoke response, got: {revoke_body}" |
| 67 | + ) |
| 68 | + LOGGER.info(f"[authz] Admin successfully revoked free user's key id={active_api_key_id}") |
| 69 | + |
| 70 | + @pytest.mark.tier1 |
| 71 | + def test_non_admin_cannot_access_other_users_keys( |
| 72 | + self, |
| 73 | + request_session_http: requests.Session, |
| 74 | + base_url: str, |
| 75 | + ocp_token_for_actor: str, |
| 76 | + admin_active_api_key_id: str, |
| 77 | + ) -> None: |
| 78 | + """Verify a non-admin user gets 404 when accessing another user's API key.""" |
| 79 | + get_resp, _ = get_api_key( |
| 80 | + request_session_http=request_session_http, |
| 81 | + base_url=base_url, |
| 82 | + key_id=admin_active_api_key_id, |
| 83 | + ocp_user_token=ocp_token_for_actor, |
| 84 | + ) |
| 85 | + assert get_resp.status_code == 404, ( |
| 86 | + f"Expected 404 (IDOR protection) on free user GET of admin's key id={admin_active_api_key_id}, " |
| 87 | + f"got {get_resp.status_code}: {get_resp.text[:200]}" |
| 88 | + ) |
| 89 | + LOGGER.info(f"[authz] Free user correctly received 404 on GET of admin's key id={admin_active_api_key_id}") |
| 90 | + |
| 91 | + revoke_resp, _ = revoke_api_key( |
| 92 | + request_session_http=request_session_http, |
| 93 | + base_url=base_url, |
| 94 | + key_id=admin_active_api_key_id, |
| 95 | + ocp_user_token=ocp_token_for_actor, |
| 96 | + ) |
| 97 | + assert revoke_resp.status_code == 404, ( |
| 98 | + f"Expected 404 (IDOR protection) on free user DELETE of admin's key id={admin_active_api_key_id}, " |
| 99 | + f"got {revoke_resp.status_code}: {revoke_resp.text[:200]}" |
| 100 | + ) |
| 101 | + LOGGER.info(f"[authz] Free user correctly received 404 on DELETE of admin's key id={admin_active_api_key_id}") |
| 102 | + |
| 103 | + @pytest.mark.tier1 |
| 104 | + def test_non_admin_search_only_returns_own_keys( |
| 105 | + self, |
| 106 | + request_session_http: requests.Session, |
| 107 | + base_url: str, |
| 108 | + ocp_token_for_actor: str, |
| 109 | + active_api_key_id: str, |
| 110 | + admin_active_api_key_id: str, |
| 111 | + ) -> None: |
| 112 | + """Verify a non-admin user's search results contain only their own keys.""" |
| 113 | + list_resp, list_body = list_api_keys( |
| 114 | + request_session_http=request_session_http, |
| 115 | + base_url=base_url, |
| 116 | + ocp_user_token=ocp_token_for_actor, |
| 117 | + filters={"status": ["active"]}, |
| 118 | + sort={"by": "created_at", "order": "desc"}, |
| 119 | + pagination={"limit": 50, "offset": 0}, |
| 120 | + ) |
| 121 | + assert list_resp.status_code == 200, ( |
| 122 | + f"Expected 200 on free user search, got {list_resp.status_code}: {list_resp.text[:200]}" |
| 123 | + ) |
| 124 | + items: list[dict] = list_body.get("items") or list_body.get("data") or [] |
| 125 | + key_ids = [item["id"] for item in items] |
| 126 | + |
| 127 | + assert active_api_key_id in key_ids, ( |
| 128 | + f"Expected free user's own key id={active_api_key_id} in results, found ids={key_ids}" |
| 129 | + ) |
| 130 | + assert admin_active_api_key_id not in key_ids, ( |
| 131 | + f"Admin's key id={admin_active_api_key_id} must NOT appear in free user's search results" |
| 132 | + ) |
| 133 | + LOGGER.info( |
| 134 | + f"[authz] Free user search returned {len(items)} key(s) — " |
| 135 | + f"own key present, admin's key correctly excluded" |
| 136 | + ) |
| 137 | + |
| 138 | + @pytest.mark.tier1 |
| 139 | + def test_non_admin_cannot_search_by_other_username( |
| 140 | + self, |
| 141 | + request_session_http: requests.Session, |
| 142 | + base_url: str, |
| 143 | + ocp_token_for_actor: str, |
| 144 | + admin_username: str, |
| 145 | + ) -> None: |
| 146 | + """Verify a non-admin user gets 403 when searching with another user's username as a filter.""" |
| 147 | + list_resp, _ = list_api_keys( |
| 148 | + request_session_http=request_session_http, |
| 149 | + base_url=base_url, |
| 150 | + ocp_user_token=ocp_token_for_actor, |
| 151 | + filters={"username": admin_username, "status": ["active"]}, |
| 152 | + sort={"by": "created_at", "order": "desc"}, |
| 153 | + pagination={"limit": 50, "offset": 0}, |
| 154 | + ) |
| 155 | + assert list_resp.status_code == 403, ( |
| 156 | + f"Expected 403 when free user searches by admin username='{admin_username}', " |
| 157 | + f"got {list_resp.status_code}: {list_resp.text[:200]}" |
| 158 | + ) |
| 159 | + LOGGER.info( |
| 160 | + f"[authz] Free user correctly received 403 when searching by admin username='{admin_username}'" |
| 161 | + ) |
0 commit comments