Expose api key validation (#491)

* introduce ApiKey module alongside with types

* wire ApiKey module through clients

* rename api_key modules to api_keys

* extend ApiKey response model by missing fields

* pluralize ApiKeyModule, ApiKey and AsyncApiKey

* remove AI slop, actually validate API key

* make api key validation path a constant

* adapt tests

* backport | None notation to Optional[T]

* return None on invalid API key

* improve naming

* black .

* fix type

---------

Co-authored-by: Nick Holden <nick.r.holden@gmail.com>

Author: denizs

tests/test_api_keys.py

@@ -0,0 +1,50 @@
+# type: ignore
+import pytest
+
+from tests.utils.fixtures.mock_api_key import MockApiKey
+from tests.utils.syncify import syncify
+from workos.api_keys import API_KEY_VALIDATION_PATH, ApiKeys, AsyncApiKeys
+
+
+@pytest.mark.sync_and_async(ApiKeys, AsyncApiKeys)
+class TestApiKeys:
+    @pytest.fixture
+    def mock_api_key(self):
+        return MockApiKey().dict()
+
+    @pytest.fixture
+    def api_key(self):
+        return "sk_my_api_key"
+
+    def test_validate_api_key_with_valid_key(
+        self,
+        module_instance,
+        api_key,
+        mock_api_key,
+        capture_and_mock_http_client_request,
+    ):
+        response_body = {"api_key": mock_api_key}
+        request_kwargs = capture_and_mock_http_client_request(
+            module_instance._http_client, response_body, 200
+        )
+
+        api_key_details = syncify(module_instance.validate_api_key(value=api_key))
+
+        assert request_kwargs["url"].endswith(API_KEY_VALIDATION_PATH)
+        assert request_kwargs["method"] == "post"
+        assert api_key_details.id == mock_api_key["id"]
+        assert api_key_details.name == mock_api_key["name"]
+        assert api_key_details.object == "api_key"
+
+    def test_validate_api_key_with_invalid_key(
+        self,
+        module_instance,
+        mock_http_client_with_response,
+    ):
+        mock_http_client_with_response(
+            module_instance._http_client,
+            {"api_key": None},
+            200,
+        )
+
+        assert syncify(module_instance.validate_api_key(value="invalid-key")) is None

tests/test_client.py

@@ -37,6 +37,9 @@ def test_client_with_api_key_and_client_id_environment_variables(self):
         os.environ.pop("WORKOS_API_KEY")
         os.environ.pop("WORKOS_CLIENT_ID")
 
+    def test_initialize_api_keys(self, default_client):
+        assert bool(default_client.api_keys)
+
     def test_initialize_sso(self, default_client):
         assert bool(default_client.sso)
 
@@ -112,6 +115,9 @@ def test_client_with_api_key_and_client_id_environment_variables(self):
         os.environ.pop("WORKOS_API_KEY")
         os.environ.pop("WORKOS_CLIENT_ID")
 
+    def test_initialize_api_keys(self, default_client):
+        assert bool(default_client.api_keys)
+
     def test_initialize_directory_sync(self, default_client):
         assert bool(default_client.directory_sync)
 

tests/utils/fixtures/mock_api_key.py

@@ -0,0 +1,19 @@
+import datetime
+
+from workos.types.api_keys import ApiKey
+
+
+class MockApiKey(ApiKey):
+    def __init__(self, id="api_key_01234567890"):
+        now = datetime.datetime.now().isoformat()
+        super().__init__(
+            object="api_key",
+            id=id,
+            owner={"type": "organization", "id": "org_1337"},
+            name="Development API Key",
+            obfuscated_value="api_..0",
+            permissions=[],
+            last_used_at=now,
+            created_at=now,
+            updated_at=now,
+        )

workos/_base_client.py

@@ -1,21 +1,21 @@
-from abc import abstractmethod
 import os
+from abc import abstractmethod
 from typing import Optional
-from workos.__about__ import __version__
+
 from workos._client_configuration import ClientConfiguration
-from workos.fga import FGAModule
-from workos.utils._base_http_client import DEFAULT_REQUEST_TIMEOUT
-from workos.utils.http_client import HTTPClient
+from workos.api_keys import ApiKeysModule
 from workos.audit_logs import AuditLogsModule
 from workos.directory_sync import DirectorySyncModule
 from workos.events import EventsModule
+from workos.fga import FGAModule
 from workos.mfa import MFAModule
-from workos.organizations import OrganizationsModule
 from workos.organization_domains import OrganizationDomainsModule
+from workos.organizations import OrganizationsModule
 from workos.passwordless import PasswordlessModule
 from workos.portal import PortalModule
 from workos.sso import SSOModule
 from workos.user_management import UserManagementModule
+from workos.utils._base_http_client import DEFAULT_REQUEST_TIMEOUT
 from workos.webhooks import WebhooksModule
 
 
@@ -65,6 +65,10 @@ def __init__(
             else int(os.getenv("WORKOS_REQUEST_TIMEOUT", DEFAULT_REQUEST_TIMEOUT))
         )
 
+    @property
+    @abstractmethod
+    def api_keys(self) -> ApiKeysModule: ...
+
     @property
     @abstractmethod
     def audit_logs(self) -> AuditLogsModule: ...

workos/api_keys.py

@@ -0,0 +1,53 @@
+from typing import Optional, Protocol
+
+from workos.types.api_keys import ApiKey
+from workos.typing.sync_or_async import SyncOrAsync
+from workos.utils.http_client import AsyncHTTPClient, SyncHTTPClient
+from workos.utils.request_helper import REQUEST_METHOD_POST
+
+API_KEY_VALIDATION_PATH = "api_keys/validations"
+RESOURCE_OBJECT_ATTRIBUTE_NAME = "api_key"
+
+
+class ApiKeysModule(Protocol):
+    def validate_api_key(self, *, value: str) -> SyncOrAsync[Optional[ApiKey]]:
+        """Validate an API key.
+
+        Kwargs:
+            value (str): API key value
+
+        Returns:
+            Optional[ApiKey]: Returns ApiKey resource object
+                if supplied value was valid, None if it was not
+        """
+        ...
+
+
+class ApiKeys(ApiKeysModule):
+    _http_client: SyncHTTPClient
+
+    def __init__(self, http_client: SyncHTTPClient):
+        self._http_client = http_client
+
+    def validate_api_key(self, *, value: str) -> Optional[ApiKey]:
+        response = self._http_client.request(
+            API_KEY_VALIDATION_PATH, method=REQUEST_METHOD_POST, json={"value": value}
+        )
+        if response.get(RESOURCE_OBJECT_ATTRIBUTE_NAME) is None:
+            return None
+        return ApiKey.model_validate(response[RESOURCE_OBJECT_ATTRIBUTE_NAME])
+
+
+class AsyncApiKeys(ApiKeysModule):
+    _http_client: AsyncHTTPClient
+
+    def __init__(self, http_client: AsyncHTTPClient):
+        self._http_client = http_client
+
+    async def validate_api_key(self, *, value: str) -> Optional[ApiKey]:
+        response = await self._http_client.request(
+            API_KEY_VALIDATION_PATH, method=REQUEST_METHOD_POST, json={"value": value}
+        )
+        if response.get(RESOURCE_OBJECT_ATTRIBUTE_NAME) is None:
+            return None
+        return ApiKey.model_validate(response[RESOURCE_OBJECT_ATTRIBUTE_NAME])

workos/async_client.py

@@ -1,6 +1,7 @@
 from typing import Optional
 from workos.__about__ import __version__
 from workos._base_client import BaseClient
+from workos.api_keys import AsyncApiKeys
 from workos.audit_logs import AuditLogsModule
 from workos.directory_sync import AsyncDirectorySync
 from workos.events import AsyncEvents
@@ -45,6 +46,12 @@ def __init__(
             timeout=self.request_timeout,
         )
 
+    @property
+    def api_keys(self) -> AsyncApiKeys:
+        if not getattr(self, "_api_keys", None):
+            self._api_keys = AsyncApiKeys(self._http_client)
+        return self._api_keys
+
     @property
     def sso(self) -> AsyncSSO:
         if not getattr(self, "_sso", None):

workos/client.py

@@ -1,6 +1,7 @@
 from typing import Optional
 from workos.__about__ import __version__
 from workos._base_client import BaseClient
+from workos.api_keys import ApiKeys
 from workos.audit_logs import AuditLogs
 from workos.directory_sync import DirectorySync
 from workos.fga import FGA
@@ -45,6 +46,12 @@ def __init__(
             timeout=self.request_timeout,
         )
 
+    @property
+    def api_keys(self) -> ApiKeys:
+        if not getattr(self, "_api_keys", None):
+            self._api_keys = ApiKeys(self._http_client)
+        return self._api_keys
+
     @property
     def sso(self) -> SSO:
         if not getattr(self, "_sso", None):

workos/types/api_keys/__init__.py

@@ -0,0 +1 @@
+from .api_keys import ApiKey as ApiKey  # noqa: F401

workos/types/api_keys/api_keys.py

@@ -0,0 +1,20 @@
+from typing import Literal, Optional, Sequence
+
+from workos.types.workos_model import WorkOSModel
+
+
+class ApiKeyOwner(WorkOSModel):
+    type: str
+    id: str
+
+
+class ApiKey(WorkOSModel):
+    object: Literal["api_key"]
+    id: str
+    owner: ApiKeyOwner
+    name: str
+    obfuscated_value: str
+    last_used_at: Optional[str] = None
+    permissions: Sequence[str]
+    created_at: str
+    updated_at: str