Add async M2M token auth

stytch/consumer/api/m2m.py

@@ -118,7 +118,6 @@ def authenticate_token(
         ] = m2m_authorization.perform_authorization_check,
     ) -> Optional[M2MJWTClaims]:
         """Validates a M2M JWT locally.
-        Note: There is no async version of this since we make no network calls.
 
         Fields:
           - access_token: The ID of the client.
@@ -162,4 +161,60 @@ def authenticate_token(
             custom_claims=custom_claims,
         )
 
+    async def authenticate_token_async(
+        self,
+        access_token: str,
+        required_scopes: Optional[List[str]] = None,
+        max_token_age: Optional[int] = None,
+        scope_authorization_func: Callable[
+            [m2m_authorization.AuthorizationCheckParams], bool
+        ] = m2m_authorization.perform_authorization_check,
+    ) -> Optional[M2MJWTClaims]:
+        """Validates a M2M JWT locally without blocking the event loop.
+
+        Uses asyncio.to_thread() for the JWKS fetch, which may make a network call
+        if the signing key is not cached.
+
+        Fields:
+          - access_token: The ID of the client.
+          - required_scopes: A list of scopes the token must have to be valid.
+          - max_token_age: The maximum possible lifetime in seconds for the token to be valid.
+          - scope_authorization_func: A function to check if the token has the required scopes. This defaults to
+            a function that assumes scopes are either direct string matches or written in the form "action:resource".
+            See the documentation for `m2m_authorization.perform_authorization_check` for more information.
+        """  # noqa
+
+        _scope_claim = "scope"
+        generic_claims = await jwt_helpers.authenticate_jwt_local_async(
+            project_id=self.project_id,
+            jwks_client=self.jwks_client,
+            jwt=access_token,
+            max_token_age_seconds=max_token_age,
+            base_url=self.api_base.base_url,
+        )
+        if generic_claims is None:
+            return None
+
+        scope = generic_claims.untyped_claims[_scope_claim]
+        scopes = [s for s in scope.split(" ") if len(s) > 0]
+        required_scopes = required_scopes or []
+
+        is_authorized = scope_authorization_func(
+            m2m_authorization.AuthorizationCheckParams(
+                has_scopes=scopes,
+                required_scopes=required_scopes,
+            )
+        )
+        if not is_authorized:
+            return None
+
+        custom_claims = {
+            k: v for k, v in generic_claims.untyped_claims.items() if k != _scope_claim
+        }
+        return M2MJWTClaims(
+            client_id=generic_claims.reserved_claims["sub"],
+            scopes=scopes,
+            custom_claims=custom_claims,
+        )
+
     # ENDMANUAL(m2m.authenticate_token)

stytch/consumer/api/test/test_m2m_authenticate_token_async.py

@@ -0,0 +1,98 @@
+#!/usr/bin/env python3
+
+import asyncio
+import time
+import unittest
+from unittest.mock import AsyncMock, MagicMock, patch
+
+from stytch.consumer.api.m2m import M2M
+from stytch.shared.jwt_helpers import GenericClaims
+
+FAKE_JWT = "fake.jwt.token"
+FAKE_PROJECT_ID = "project-test-abc123"
+
+FAKE_GENERIC_CLAIMS = GenericClaims(
+    reserved_claims={"sub": "m2m-client-test-123", "exp": 9999999999},
+    untyped_claims={"scope": "read:docs write:docs"},
+)
+
+
+def _make_m2m() -> M2M:
+    mock_api_base = MagicMock()
+    mock_api_base.base_url = "https://test.stytch.com/"
+    return M2M(
+        api_base=mock_api_base,
+        sync_client=MagicMock(),
+        async_client=MagicMock(),
+        jwks_client=MagicMock(),
+        project_id=FAKE_PROJECT_ID,
+    )
+
+
+class TestM2MAuthenticateTokenAsync(unittest.IsolatedAsyncioTestCase):
+    def setUp(self) -> None:
+        self.m2m = _make_m2m()
+
+    def test_is_coroutine_function(self) -> None:
+        self.assertTrue(asyncio.iscoroutinefunction(self.m2m.authenticate_token_async))
+
+    @patch(
+        "stytch.consumer.api.m2m.jwt_helpers.authenticate_jwt_local_async",
+        new_callable=AsyncMock,
+    )
+    async def test_returns_none_for_invalid_jwt(self, mock_jwt) -> None:
+        mock_jwt.return_value = None
+        result = await self.m2m.authenticate_token_async(access_token=FAKE_JWT)
+        self.assertIsNone(result)
+
+    @patch(
+        "stytch.consumer.api.m2m.jwt_helpers.authenticate_jwt_local_async",
+        new_callable=AsyncMock,
+    )
+    async def test_returns_claims_for_valid_jwt(self, mock_jwt) -> None:
+        mock_jwt.return_value = FAKE_GENERIC_CLAIMS
+        result = await self.m2m.authenticate_token_async(access_token=FAKE_JWT)
+        self.assertIsNotNone(result)
+        if result is not None:
+            self.assertEqual(result.client_id, "m2m-client-test-123")
+            self.assertEqual(result.scopes, ["read:docs", "write:docs"])
+
+    @patch(
+        "stytch.consumer.api.m2m.jwt_helpers.authenticate_jwt_local_async",
+        new_callable=AsyncMock,
+    )
+    async def test_returns_none_when_required_scope_missing(self, mock_jwt) -> None:
+        mock_jwt.return_value = FAKE_GENERIC_CLAIMS
+        result = await self.m2m.authenticate_token_async(
+            access_token=FAKE_JWT, required_scopes=["admin"]
+        )
+        self.assertIsNone(result)
+
+    async def test_is_non_blocking_jwt_verification(self) -> None:
+        DELAY = 0.1
+        N = 5
+
+        async def slow_authenticate_jwt_local_async(**kwargs) -> GenericClaims:
+            await asyncio.sleep(DELAY)
+            return FAKE_GENERIC_CLAIMS
+
+        with patch(
+            "stytch.consumer.api.m2m.jwt_helpers.authenticate_jwt_local_async",
+            side_effect=slow_authenticate_jwt_local_async,
+        ):
+            start = time.monotonic()
+            results = await asyncio.gather(
+                *[
+                    self.m2m.authenticate_token_async(access_token=FAKE_JWT)
+                    for _ in range(N)
+                ]
+            )
+            elapsed = time.monotonic() - start
+
+        # All N calls should interleave at the await point, completing in ~DELAY total
+        self.assertLess(elapsed, DELAY * 2)
+        self.assertEqual(len(results), N)
+
+
+if __name__ == "__main__":
+    unittest.main()

stytch/version.py

@@ -1 +1 @@
-__version__ = "15.0.0"
+__version__ = "15.1.0"