diff --git a/slack_sdk/oauth/state_store/__init__.py b/slack_sdk/oauth/state_store/__init__.py index 15491fd1d..c3fbd0c4e 100644 --- a/slack_sdk/oauth/state_store/__init__.py +++ b/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", ] diff --git a/slack_sdk/oauth/state_store/stateless/__init__.py b/slack_sdk/oauth/state_store/stateless/__init__.py new file mode 100644 index 000000000..e123b730c --- /dev/null +++ b/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) diff --git a/tests/slack_sdk/oauth/state_store/test_stateless.py b/tests/slack_sdk/oauth/state_store/test_stateless.py new file mode 100644 index 000000000..7a36e4b1f --- /dev/null +++ b/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")