feat(oauth): add StatelessOAuthStateStore using JWT

slack_sdk/oauth/state_store/__init__.py

@@ -6,8 +6,10 @@
 # from .amazon_s3_state_store import AmazonS3OAuthStateStore
 from .file import FileOAuthStateStore
 from .state_store import OAuthStateStore
+from .stateless import StatelessOAuthStateStore
 
 __all__ = [
     "FileOAuthStateStore",
     "OAuthStateStore",
+    "StatelessOAuthStateStore",
 ]

slack_sdk/oauth/state_store/stateless/__init__.py

@@ -0,0 +1,81 @@
+import base64
+import hashlib
+import hmac
+import json
+import logging
+import time
+from logging import Logger
+
+from ..async_state_store import AsyncOAuthStateStore
+from ..state_store import OAuthStateStore
+
+
+class StatelessOAuthStateStore(OAuthStateStore, AsyncOAuthStateStore):
+    def __init__(
+        self,
+        *,
+        expiration_seconds: int,
+        signing_secret: str,
+        logger: Logger = logging.getLogger(__name__),
+    ):
+        self.expiration_seconds = expiration_seconds
+        self.signing_secret = signing_secret
+        self._logger = logger
+
+    @property
+    def logger(self) -> Logger:
+        if self._logger is None:
+            self._logger = logging.getLogger(__name__)
+        return self._logger
+
+    @staticmethod
+    def _b64url_encode(data: bytes) -> str:
+        return base64.urlsafe_b64encode(data).rstrip(b"=").decode("utf-8")
+
+    @staticmethod
+    def _b64url_decode(data: str) -> bytes:
+        padded = data + "=" * (4 - len(data) % 4)
+        return base64.urlsafe_b64decode(padded)
+
+    def _sign(self, message: str) -> str:
+        return self._b64url_encode(
+            hmac.new(
+                self.signing_secret.encode("utf-8"),
+                message.encode("utf-8"),
+                hashlib.sha256,
+            ).digest()
+        )
+
+    def issue(self, *args, **kwargs) -> str:
+        header = self._b64url_encode(json.dumps({"alg": "HS256", "typ": "JWT"}, separators=(",", ":")).encode())
+        payload = self._b64url_encode(
+            json.dumps({"exp": int(time.time()) + self.expiration_seconds}, separators=(",", ":")).encode()
+        )
+        signature = self._sign(f"{header}.{payload}")
+        return f"{header}.{payload}.{signature}"
+
+    def consume(self, state: str) -> bool:
+        try:
+            parts = state.split(".")
+            if len(parts) != 3:
+                return False
+            header, payload, signature = parts
+            expected_signature = self._sign(f"{header}.{payload}")
+            if not hmac.compare_digest(signature, expected_signature):
+                self.logger.warning("Invalid JWT signature for state parameter")
+                return False
+            claims = json.loads(self._b64url_decode(payload))
+            exp = claims.get("exp")
+            if exp is None or time.time() > exp:
+                self.logger.warning("Expired or missing exp claim in state JWT")
+                return False
+            return True
+        except Exception as e:
+            self.logger.warning(f"Failed to validate state JWT: {e}")
+            return False
+
+    async def async_issue(self, *args, **kwargs) -> str:
+        return self.issue(*args, **kwargs)
+
+    async def async_consume(self, state: str) -> bool:
+        return self.consume(state)

tests/slack_sdk/oauth/state_store/test_stateless.py

@@ -0,0 +1,65 @@
+import unittest
+
+from slack_sdk.oauth.state_store import StatelessOAuthStateStore
+
+
+class TestStateless(unittest.TestCase):
+    def setUp(self):
+        self.store = StatelessOAuthStateStore(
+            expiration_seconds=10,
+            signing_secret="test-signing-secret",
+        )
+
+    def tearDown(self):
+        pass
+
+    def test_instance(self):
+        self.assertIsNotNone(self.store)
+
+    def test_issue_returns_jwt_format(self):
+        state = self.store.issue()
+        self.assertEqual(len(state.split(".")), 3)
+
+    def test_issue_and_consume(self):
+        state = self.store.issue()
+        result = self.store.consume(state)
+        self.assertTrue(result)
+
+    def test_consume_is_repeatable(self):
+        state = self.store.issue()
+        self.assertTrue(self.store.consume(state))
+        self.assertTrue(self.store.consume(state))
+
+    def test_invalid_state_format(self):
+        self.assertFalse(self.store.consume("not-a-jwt"))
+
+    def test_tampered_signature(self):
+        state = self.store.issue()
+        parts = state.split(".")
+        parts[2] = "invalidsignature"
+        self.assertFalse(self.store.consume(".".join(parts)))
+
+    def test_tampered_payload(self):
+        state = self.store.issue()
+        parts = state.split(".")
+        parts[1] = parts[1][:-2] + "AA"
+        self.assertFalse(self.store.consume(".".join(parts)))
+
+    def test_wrong_signing_secret(self):
+        store2 = StatelessOAuthStateStore(
+            expiration_seconds=10,
+            signing_secret="different-secret",
+        )
+        state = self.store.issue()
+        self.assertFalse(store2.consume(state))
+
+    def test_expired_state(self):
+        expired_store = StatelessOAuthStateStore(
+            expiration_seconds=-1,
+            signing_secret="test-signing-secret",
+        )
+        state = expired_store.issue()
+        self.assertFalse(self.store.consume(state))
+
+    def test_kwargs(self):
+        self.store.issue(foo=123, bar="baz")