From 7e41fc8f9500db04268cdf3940d5ff82846f26d4 Mon Sep 17 00:00:00 2001 From: Ejiro Asiuwhu Date: Wed, 25 Mar 2026 01:35:44 +0100 Subject: [PATCH] feat: add api key management methods to both facades - User API key create/list/revoke on StackServerApp and AsyncStackServerApp - Team API key create/list/revoke on both facades - check_api_key validates any key and returns user_id/team_id - Tests for all sync and async API key methods --- .../python/src/stack_auth/_app.py | 187 ++++++++++ sdks/implementations/python/tests/test_app.py | 353 ++++++++++++++++++ 2 files changed, 540 insertions(+) diff --git a/sdks/implementations/python/src/stack_auth/_app.py b/sdks/implementations/python/src/stack_auth/_app.py index ace83f03d..a59c884dd 100644 --- a/sdks/implementations/python/src/stack_auth/_app.py +++ b/sdks/implementations/python/src/stack_auth/_app.py @@ -17,7 +17,14 @@ from stack_auth._constants import DEFAULT_BASE_URL from stack_auth._pagination import PaginatedResult, _PaginationMeta from stack_auth._token_store import TokenStore, TokenStoreInit, resolve_token_store from stack_auth.errors import ApiKeyError, NotFoundError +from stack_auth.models.api_keys import ( + TeamApiKey, + TeamApiKeyFirstView, + UserApiKey, + UserApiKeyFirstView, +) from stack_auth.models.contact_channels import ContactChannel +from stack_auth.models.oauth import OAuthProvider from stack_auth.models.permissions import TeamPermission from stack_auth.models.sessions import ActiveSession from stack_auth.models.teams import ServerTeam, TeamInvitation, TeamMemberProfile @@ -616,6 +623,94 @@ class StackServerApp: "POST", "/contact-channels/verify", body={"code": code} ) + # -- API keys ------------------------------------------------------------ + + def create_user_api_key( + self, + user_id: str, + *, + description: str, + expires_at_millis: Optional[int] = None, + scope: Optional[str] = None, + team_id: Optional[str] = None, + ) -> UserApiKeyFirstView: + """Create a user API key. + + Returns the key including the secret (only available at creation time). + """ + body = _build_params( + description=description, + expires_at_millis=expires_at_millis, + scope=scope, + team_id=team_id, + ) + data = self._client.request( + "POST", f"/users/{user_id}/api-keys", body=body + ) + return UserApiKeyFirstView.model_validate(data) + + def list_user_api_keys(self, user_id: str) -> list[UserApiKey]: + """List API keys for a user.""" + data = self._client.request("GET", f"/users/{user_id}/api-keys") + return [ + UserApiKey.model_validate(item) + for item in (data or {}).get("items", []) + ] + + def revoke_user_api_key(self, api_key_id: str) -> None: + """Revoke (delete) a user API key.""" + self._client.request("DELETE", f"/api-keys/{api_key_id}") + + def create_team_api_key( + self, + team_id: str, + *, + description: str, + expires_at_millis: Optional[int] = None, + scope: Optional[str] = None, + ) -> TeamApiKeyFirstView: + """Create a team API key. + + Returns the key including the secret (only available at creation time). + """ + body = _build_params( + description=description, + expires_at_millis=expires_at_millis, + scope=scope, + ) + data = self._client.request( + "POST", f"/teams/{team_id}/api-keys", body=body + ) + return TeamApiKeyFirstView.model_validate(data) + + def list_team_api_keys(self, team_id: str) -> list[TeamApiKey]: + """List API keys for a team.""" + data = self._client.request("GET", f"/teams/{team_id}/api-keys") + return [ + TeamApiKey.model_validate(item) + for item in (data or {}).get("items", []) + ] + + def revoke_team_api_key(self, api_key_id: str) -> None: + """Revoke (delete) a team API key.""" + self._client.request("DELETE", f"/api-keys/{api_key_id}") + + def check_api_key(self, api_key: str) -> dict[str, Any] | None: + """Validate an API key and return associated user/team info. + + Returns a dict with ``user_id`` and/or ``team_id``, or ``None`` + if the key is invalid. + """ + try: + data = self._client.request( + "POST", "/api-keys/check", body={"api_key": api_key} + ) + except (NotFoundError, ApiKeyError): + return None + if data is None: + return None + return data + # --------------------------------------------------------------------------- # AsyncStackServerApp (async) @@ -1141,3 +1236,95 @@ class AsyncStackServerApp: await self._client.request( "POST", "/contact-channels/verify", body={"code": code} ) + + # -- API keys ------------------------------------------------------------ + + async def create_user_api_key( + self, + user_id: str, + *, + description: str, + expires_at_millis: Optional[int] = None, + scope: Optional[str] = None, + team_id: Optional[str] = None, + ) -> UserApiKeyFirstView: + """Create a user API key. + + Returns the key including the secret (only available at creation time). + """ + body = _build_params( + description=description, + expires_at_millis=expires_at_millis, + scope=scope, + team_id=team_id, + ) + data = await self._client.request( + "POST", f"/users/{user_id}/api-keys", body=body + ) + return UserApiKeyFirstView.model_validate(data) + + async def list_user_api_keys(self, user_id: str) -> list[UserApiKey]: + """List API keys for a user.""" + data = await self._client.request( + "GET", f"/users/{user_id}/api-keys" + ) + return [ + UserApiKey.model_validate(item) + for item in (data or {}).get("items", []) + ] + + async def revoke_user_api_key(self, api_key_id: str) -> None: + """Revoke (delete) a user API key.""" + await self._client.request("DELETE", f"/api-keys/{api_key_id}") + + async def create_team_api_key( + self, + team_id: str, + *, + description: str, + expires_at_millis: Optional[int] = None, + scope: Optional[str] = None, + ) -> TeamApiKeyFirstView: + """Create a team API key. + + Returns the key including the secret (only available at creation time). + """ + body = _build_params( + description=description, + expires_at_millis=expires_at_millis, + scope=scope, + ) + data = await self._client.request( + "POST", f"/teams/{team_id}/api-keys", body=body + ) + return TeamApiKeyFirstView.model_validate(data) + + async def list_team_api_keys(self, team_id: str) -> list[TeamApiKey]: + """List API keys for a team.""" + data = await self._client.request( + "GET", f"/teams/{team_id}/api-keys" + ) + return [ + TeamApiKey.model_validate(item) + for item in (data or {}).get("items", []) + ] + + async def revoke_team_api_key(self, api_key_id: str) -> None: + """Revoke (delete) a team API key.""" + await self._client.request("DELETE", f"/api-keys/{api_key_id}") + + async def check_api_key(self, api_key: str) -> dict[str, Any] | None: + """Validate an API key and return associated user/team info. + + Returns a dict with ``user_id`` and/or ``team_id``, or ``None`` + if the key is invalid. + """ + try: + data = await self._client.request( + "POST", "/api-keys/check", body={"api_key": api_key} + ) + except (NotFoundError, ApiKeyError): + return None + if data is None: + return None + return data diff --git a/sdks/implementations/python/tests/test_app.py b/sdks/implementations/python/tests/test_app.py index d806076c1..1da38da1b 100644 --- a/sdks/implementations/python/tests/test_app.py +++ b/sdks/implementations/python/tests/test_app.py @@ -1816,3 +1816,356 @@ class TestAsyncVerifyContactChannel: body = json.loads(route.calls[0].request.content) assert body["code"] == "abc123" await app.aclose() + + +# --------------------------------------------------------------------------- +# API key test data +# --------------------------------------------------------------------------- + +USER_API_KEY_JSON = { + "id": "key-1", + "description": "My key", + "expiresAtMillis": 1700100000000, + "createdAtMillis": 1700000000000, + "isValid": True, + "userId": "user-1", + "teamId": None, +} + +USER_API_KEY_FIRST_VIEW_JSON = { + **USER_API_KEY_JSON, + "apiKey": "sk_user_secret_123", +} + +TEAM_API_KEY_JSON = { + "id": "key-2", + "description": "CI key", + "expiresAtMillis": None, + "createdAtMillis": 1700000000000, + "isValid": True, + "teamId": "team-1", +} + +TEAM_API_KEY_FIRST_VIEW_JSON = { + **TEAM_API_KEY_JSON, + "apiKey": "sk_team_secret_456", +} + + +# --------------------------------------------------------------------------- +# StackServerApp - create_user_api_key +# --------------------------------------------------------------------------- + + +class TestCreateUserApiKey: + @respx.mock + def test_create_user_api_key(self) -> None: + route = respx.post(f"{API_PREFIX}/users/user-1/api-keys").mock( + return_value=httpx.Response(200, json=USER_API_KEY_FIRST_VIEW_JSON) + ) + app = StackServerApp(project_id="proj", secret_server_key="sk") + key = app.create_user_api_key("user-1", description="My key") + assert key.id == "key-1" + assert key.api_key == "sk_user_secret_123" + assert key.user_id == "user-1" + assert key.description == "My key" + + @respx.mock + def test_create_user_api_key_with_optional_fields(self) -> None: + route = respx.post(f"{API_PREFIX}/users/user-1/api-keys").mock( + return_value=httpx.Response(200, json=USER_API_KEY_FIRST_VIEW_JSON) + ) + app = StackServerApp(project_id="proj", secret_server_key="sk") + key = app.create_user_api_key( + "user-1", + description="My key", + expires_at_millis=1700100000000, + scope="read", + team_id="team-1", + ) + assert key.id == "key-1" + import json + + body = json.loads(route.calls[0].request.content) + assert body["description"] == "My key" + assert body["expires_at_millis"] == 1700100000000 + assert body["scope"] == "read" + assert body["team_id"] == "team-1" + + +# --------------------------------------------------------------------------- +# StackServerApp - list_user_api_keys +# --------------------------------------------------------------------------- + + +class TestListUserApiKeys: + @respx.mock + def test_list_user_api_keys(self) -> None: + respx.get(f"{API_PREFIX}/users/user-1/api-keys").mock( + return_value=httpx.Response( + 200, json={"items": [USER_API_KEY_JSON]} + ) + ) + app = StackServerApp(project_id="proj", secret_server_key="sk") + keys = app.list_user_api_keys("user-1") + assert len(keys) == 1 + assert keys[0].id == "key-1" + assert keys[0].user_id == "user-1" + + @respx.mock + def test_list_user_api_keys_empty(self) -> None: + respx.get(f"{API_PREFIX}/users/user-1/api-keys").mock( + return_value=httpx.Response(200, json={"items": []}) + ) + app = StackServerApp(project_id="proj", secret_server_key="sk") + keys = app.list_user_api_keys("user-1") + assert keys == [] + + +# --------------------------------------------------------------------------- +# StackServerApp - revoke_user_api_key +# --------------------------------------------------------------------------- + + +class TestRevokeUserApiKey: + @respx.mock + def test_revoke_user_api_key(self) -> None: + route = respx.delete(f"{API_PREFIX}/api-keys/key-1").mock( + return_value=httpx.Response(200) + ) + app = StackServerApp(project_id="proj", secret_server_key="sk") + result = app.revoke_user_api_key("key-1") + assert result is None + assert route.called + + +# --------------------------------------------------------------------------- +# StackServerApp - create_team_api_key +# --------------------------------------------------------------------------- + + +class TestCreateTeamApiKey: + @respx.mock + def test_create_team_api_key(self) -> None: + route = respx.post(f"{API_PREFIX}/teams/team-1/api-keys").mock( + return_value=httpx.Response(200, json=TEAM_API_KEY_FIRST_VIEW_JSON) + ) + app = StackServerApp(project_id="proj", secret_server_key="sk") + key = app.create_team_api_key("team-1", description="CI key") + assert key.id == "key-2" + assert key.api_key == "sk_team_secret_456" + assert key.team_id == "team-1" + + +# --------------------------------------------------------------------------- +# StackServerApp - list_team_api_keys +# --------------------------------------------------------------------------- + + +class TestListTeamApiKeys: + @respx.mock + def test_list_team_api_keys(self) -> None: + respx.get(f"{API_PREFIX}/teams/team-1/api-keys").mock( + return_value=httpx.Response( + 200, json={"items": [TEAM_API_KEY_JSON]} + ) + ) + app = StackServerApp(project_id="proj", secret_server_key="sk") + keys = app.list_team_api_keys("team-1") + assert len(keys) == 1 + assert keys[0].id == "key-2" + assert keys[0].team_id == "team-1" + + +# --------------------------------------------------------------------------- +# StackServerApp - revoke_team_api_key +# --------------------------------------------------------------------------- + + +class TestRevokeTeamApiKey: + @respx.mock + def test_revoke_team_api_key(self) -> None: + route = respx.delete(f"{API_PREFIX}/api-keys/key-2").mock( + return_value=httpx.Response(200) + ) + app = StackServerApp(project_id="proj", secret_server_key="sk") + result = app.revoke_team_api_key("key-2") + assert result is None + assert route.called + + +# --------------------------------------------------------------------------- +# StackServerApp - check_api_key +# --------------------------------------------------------------------------- + + +class TestCheckApiKey: + @respx.mock + def test_check_api_key_valid(self) -> None: + respx.post(f"{API_PREFIX}/api-keys/check").mock( + return_value=httpx.Response( + 200, json={"user_id": "user-1", "team_id": "team-1"} + ) + ) + app = StackServerApp(project_id="proj", secret_server_key="sk") + result = app.check_api_key("sk_123") + assert result is not None + assert result["user_id"] == "user-1" + assert result["team_id"] == "team-1" + + @respx.mock + def test_check_api_key_invalid(self) -> None: + respx.post(f"{API_PREFIX}/api-keys/check").mock( + return_value=httpx.Response( + 200, + json={"message": "API key not valid"}, + headers={ + "x-stack-known-error": "API_KEY_NOT_VALID", + "x-stack-actual-status": "400", + }, + ) + ) + app = StackServerApp(project_id="proj", secret_server_key="sk") + result = app.check_api_key("invalid") + assert result is None + + @respx.mock + def test_check_api_key_sends_correct_body(self) -> None: + route = respx.post(f"{API_PREFIX}/api-keys/check").mock( + return_value=httpx.Response( + 200, json={"user_id": "user-1"} + ) + ) + app = StackServerApp(project_id="proj", secret_server_key="sk") + app.check_api_key("sk_123") + import json + + body = json.loads(route.calls[0].request.content) + assert body == {"api_key": "sk_123"} + + +# --------------------------------------------------------------------------- +# AsyncStackServerApp - API key methods +# --------------------------------------------------------------------------- + + +class TestAsyncCreateUserApiKey: + @respx.mock + @pytest.mark.asyncio + async def test_create_user_api_key(self) -> None: + respx.post(f"{API_PREFIX}/users/user-1/api-keys").mock( + return_value=httpx.Response(200, json=USER_API_KEY_FIRST_VIEW_JSON) + ) + app = AsyncStackServerApp(project_id="proj", secret_server_key="sk") + key = await app.create_user_api_key("user-1", description="My key") + assert key.id == "key-1" + assert key.api_key == "sk_user_secret_123" + await app.aclose() + + +class TestAsyncListUserApiKeys: + @respx.mock + @pytest.mark.asyncio + async def test_list_user_api_keys(self) -> None: + respx.get(f"{API_PREFIX}/users/user-1/api-keys").mock( + return_value=httpx.Response( + 200, json={"items": [USER_API_KEY_JSON]} + ) + ) + app = AsyncStackServerApp(project_id="proj", secret_server_key="sk") + keys = await app.list_user_api_keys("user-1") + assert len(keys) == 1 + assert keys[0].id == "key-1" + await app.aclose() + + +class TestAsyncRevokeUserApiKey: + @respx.mock + @pytest.mark.asyncio + async def test_revoke_user_api_key(self) -> None: + route = respx.delete(f"{API_PREFIX}/api-keys/key-1").mock( + return_value=httpx.Response(200) + ) + app = AsyncStackServerApp(project_id="proj", secret_server_key="sk") + result = await app.revoke_user_api_key("key-1") + assert result is None + assert route.called + await app.aclose() + + +class TestAsyncCreateTeamApiKey: + @respx.mock + @pytest.mark.asyncio + async def test_create_team_api_key(self) -> None: + respx.post(f"{API_PREFIX}/teams/team-1/api-keys").mock( + return_value=httpx.Response(200, json=TEAM_API_KEY_FIRST_VIEW_JSON) + ) + app = AsyncStackServerApp(project_id="proj", secret_server_key="sk") + key = await app.create_team_api_key("team-1", description="CI key") + assert key.id == "key-2" + assert key.api_key == "sk_team_secret_456" + await app.aclose() + + +class TestAsyncListTeamApiKeys: + @respx.mock + @pytest.mark.asyncio + async def test_list_team_api_keys(self) -> None: + respx.get(f"{API_PREFIX}/teams/team-1/api-keys").mock( + return_value=httpx.Response( + 200, json={"items": [TEAM_API_KEY_JSON]} + ) + ) + app = AsyncStackServerApp(project_id="proj", secret_server_key="sk") + keys = await app.list_team_api_keys("team-1") + assert len(keys) == 1 + assert keys[0].id == "key-2" + await app.aclose() + + +class TestAsyncRevokeTeamApiKey: + @respx.mock + @pytest.mark.asyncio + async def test_revoke_team_api_key(self) -> None: + route = respx.delete(f"{API_PREFIX}/api-keys/key-2").mock( + return_value=httpx.Response(200) + ) + app = AsyncStackServerApp(project_id="proj", secret_server_key="sk") + result = await app.revoke_team_api_key("key-2") + assert result is None + assert route.called + await app.aclose() + + +class TestAsyncCheckApiKey: + @respx.mock + @pytest.mark.asyncio + async def test_check_api_key_valid(self) -> None: + respx.post(f"{API_PREFIX}/api-keys/check").mock( + return_value=httpx.Response( + 200, json={"user_id": "user-1", "team_id": "team-1"} + ) + ) + app = AsyncStackServerApp(project_id="proj", secret_server_key="sk") + result = await app.check_api_key("sk_123") + assert result is not None + assert result["user_id"] == "user-1" + await app.aclose() + + @respx.mock + @pytest.mark.asyncio + async def test_check_api_key_invalid(self) -> None: + respx.post(f"{API_PREFIX}/api-keys/check").mock( + return_value=httpx.Response( + 200, + json={"message": "API key not valid"}, + headers={ + "x-stack-known-error": "API_KEY_NOT_VALID", + "x-stack-actual-status": "400", + }, + ) + ) + app = AsyncStackServerApp(project_id="proj", secret_server_key="sk") + result = await app.check_api_key("invalid") + assert result is None + await app.aclose()