Enable installation_store authorize to fallback to bots (prep for #254)

Author: seratch

slack_bolt/authorization/async_authorize.py

@@ -93,6 +93,7 @@ async def __call__(
 class AsyncInstallationStoreAuthorize(AsyncAuthorize):
     authorize_result_cache: Dict[str, AuthorizeResult]
     find_installation_available: Optional[bool]
+    find_bot_available: Optional[bool]
 
     def __init__(
         self,
@@ -110,6 +111,7 @@ def __init__(
         self.cache_enabled = cache_enabled
         self.authorize_result_cache = {}
         self.find_installation_available = None
+        self.find_bot_available = None
 
     async def __call__(
         self,
@@ -124,12 +126,15 @@ async def __call__(
             self.find_installation_available = hasattr(
                 self.installation_store, "async_find_installation"
             )
+        if self.find_bot_available is None:
+            self.find_bot_available = hasattr(self.installation_store, "async_find_bot")
 
         bot_token: Optional[str] = None
         user_token: Optional[str] = None
 
         if not self.bot_only and self.find_installation_available:
-            # since v1.1, this is the default way
+            # Since v1.1, this is the default way.
+            # If you want to use find_bot / delete_bot only, you can set bot_only as True.
             try:
                 # Note that this is the latest information for the org/workspace.
                 # The installer may not be the user associated with this incoming request.
@@ -140,47 +145,64 @@ async def __call__(
                     team_id=team_id,
                     is_enterprise_install=context.is_enterprise_install,
                 )
-                if installation is None:
-                    self._debug_log_for_not_found(enterprise_id, team_id)
-                    return None
-
-                if installation.user_id != user_id:
-                    # First off, remove the user token as the installer is a different user
-                    installation.user_token = None
-                    installation.user_scopes = []
-
-                    # try to fetch the request user's installation
-                    # to reflect the user's access token if exists
-                    user_installation = (
-                        await self.installation_store.async_find_installation(
-                            enterprise_id=enterprise_id,
-                            team_id=team_id,
-                            user_id=user_id,
-                            is_enterprise_install=context.is_enterprise_install,
+
+                if installation is not None:
+                    if installation.user_id != user_id:
+                        # First off, remove the user token as the installer is a different user
+                        installation.user_token = None
+                        installation.user_scopes = []
+
+                        # try to fetch the request user's installation
+                        # to reflect the user's access token if exists
+                        user_installation = (
+                            await self.installation_store.async_find_installation(
+                                enterprise_id=enterprise_id,
+                                team_id=team_id,
+                                user_id=user_id,
+                                is_enterprise_install=context.is_enterprise_install,
+                            )
                         )
+                        if user_installation is not None:
+                            # Overwrite the installation with the one for this user
+                            installation = user_installation
+
+                    bot_token, user_token = (
+                        installation.bot_token,
+                        installation.user_token,
                     )
-                    if user_installation is not None:
-                        # Overwrite the installation with the one for this user
-                        installation = user_installation
 
-                bot_token, user_token = installation.bot_token, installation.user_token
             except NotImplementedError as _:
                 self.find_installation_available = False
 
-        if self.bot_only or not self.find_installation_available:
-            # Use find_bot to get bot value (legacy)
-            bot: Optional[Bot] = await self.installation_store.async_find_bot(
-                enterprise_id=enterprise_id,
-                team_id=team_id,
-                is_enterprise_install=context.is_enterprise_install,
+        if (
+            # If you intentionally use only find_bot / delete_bot,
+            self.bot_only
+            # If find_installation method is not available,
+            or not self.find_installation_available
+            # If find_installation did not return data and find_bot method is available,
+            or (
+                self.find_bot_available is True
+                and bot_token is None
+                and user_token is None
             )
-            if bot is None:
-                self._debug_log_for_not_found(enterprise_id, team_id)
-                return None
-            bot_token, user_token = bot.bot_token, None
+        ):
+            try:
+                bot: Optional[Bot] = await self.installation_store.async_find_bot(
+                    enterprise_id=enterprise_id,
+                    team_id=team_id,
+                    is_enterprise_install=context.is_enterprise_install,
+                )
+                if bot is not None:
+                    bot_token = bot.bot_token
+            except NotImplementedError as _:
+                self.find_bot_available = False
+            except Exception as e:
+                self.logger.info(f"Failed to call find_bot method: {e}")
 
         token: Optional[str] = bot_token or user_token
         if token is None:
+            # No valid token was found
+            self._debug_log_for_not_found(enterprise_id, team_id)
             return None
 
         # Check cache to see if the bot object already exists

slack_bolt/authorization/authorize.py

@@ -96,6 +96,7 @@ class InstallationStoreAuthorize(Authorize):
     authorize_result_cache: Dict[str, AuthorizeResult]
     bot_only: bool
     find_installation_available: bool
+    find_bot_available: bool
 
     def __init__(
         self,
@@ -115,6 +116,7 @@ def __init__(
         self.find_installation_available = hasattr(
             installation_store, "find_installation"
         )
+        self.find_bot_available = hasattr(installation_store, "find_bot")
 
     def __call__(
         self,
@@ -129,7 +131,8 @@ def __call__(
         user_token: Optional[str] = None
 
         if not self.bot_only and self.find_installation_available:
-            # since v1.1, this is the default way
+            # Since v1.1, this is the default way.
+            # If you want to use find_bot / delete_bot only, you can set bot_only as True.
             try:
                 # Note that this is the latest information for the org/workspace.
                 # The installer may not be the user associated with this incoming request.
@@ -140,45 +143,61 @@ def __call__(
                     team_id=team_id,
                     is_enterprise_install=context.is_enterprise_install,
                 )
-                if installation is None:
-                    self._debug_log_for_not_found(enterprise_id, team_id)
-                    return None
-
-                if installation.user_id != user_id:
-                    # First off, remove the user token as the installer is a different user
-                    installation.user_token = None
-                    installation.user_scopes = []
-
-                    # try to fetch the request user's installation
-                    # to reflect the user's access token if exists
-                    user_installation = self.installation_store.find_installation(
-                        enterprise_id=enterprise_id,
-                        team_id=team_id,
-                        user_id=user_id,
-                        is_enterprise_install=context.is_enterprise_install,
+                if installation is not None:
+                    if installation.user_id != user_id:
+                        # First off, remove the user token as the installer is a different user
+                        installation.user_token = None
+                        installation.user_scopes = []
+
+                        # try to fetch the request user's installation
+                        # to reflect the user's access token if exists
+                        user_installation = self.installation_store.find_installation(
+                            enterprise_id=enterprise_id,
+                            team_id=team_id,
+                            user_id=user_id,
+                            is_enterprise_install=context.is_enterprise_install,
+                        )
+                        if user_installation is not None:
+                            # Overwrite the installation with the one for this user
+                            installation = user_installation
+
+                    bot_token, user_token = (
+                        installation.bot_token,
+                        installation.user_token,
                     )
-                    if user_installation is not None:
-                        # Overwrite the installation with the one for this user
-                        installation = user_installation
 
-                bot_token, user_token = installation.bot_token, installation.user_token
             except NotImplementedError as _:
                 self.find_installation_available = False
 
-        if self.bot_only or not self.find_installation_available:
-            # Use find_bot to get bot value (legacy)
-            bot: Optional[Bot] = self.installation_store.find_bot(
-                enterprise_id=enterprise_id,
-                team_id=team_id,
-                is_enterprise_install=context.is_enterprise_install,
+        if (
+            # If you intentionally use only find_bot / delete_bot,
+            self.bot_only
+            # If find_installation method is not available,
+            or not self.find_installation_available
+            # If find_installation did not return data and find_bot method is available,
+            or (
+                self.find_bot_available is True
+                and bot_token is None
+                and user_token is None
             )
-            if bot is None:
-                self._debug_log_for_not_found(enterprise_id, team_id)
-                return None
-            bot_token, user_token = bot.bot_token, None
+        ):
+            try:
+                bot: Optional[Bot] = self.installation_store.find_bot(
+                    enterprise_id=enterprise_id,
+                    team_id=team_id,
+                    is_enterprise_install=context.is_enterprise_install,
+                )
+                if bot is not None:
+                    bot_token = bot.bot_token
+            except NotImplementedError as _:
+                self.find_bot_available = False
+            except Exception as e:
+                self.logger.info(f"Failed to call find_bot method: {e}")
 
         token: Optional[str] = bot_token or user_token
         if token is None:
+            # No valid token was found
+            self._debug_log_for_not_found(enterprise_id, team_id)
             return None
 
         # Check cache to see if the bot object already exists

tests/scenario_tests/test_installation_store_authorize.py

@@ -0,0 +1,150 @@
+import json
+from time import time
+from typing import Optional
+from urllib.parse import quote
+
+from slack_sdk import WebClient
+from slack_sdk.oauth import InstallationStore
+from slack_sdk.oauth.installation_store import Installation, Bot
+from slack_sdk.signature import SignatureVerifier
+
+from slack_bolt import BoltRequest
+from slack_bolt.app import App
+from tests.mock_web_api_server import (
+    setup_mock_web_api_server,
+    cleanup_mock_web_api_server,
+    assert_auth_test_count,
+)
+from tests.utils import remove_os_env_temporarily, restore_os_env
+
+valid_token = "xoxb-valid"
+valid_user_token = "xoxp-valid"
+
+
+class MyInstallationStore(InstallationStore):
+    def find_bot(
+        self,
+        *,
+        enterprise_id: Optional[str],
+        team_id: Optional[str],
+        is_enterprise_install: Optional[bool] = False,
+    ) -> Optional[Bot]:
+        return Bot(
+            app_id="A111",
+            enterprise_id="E111",
+            team_id="T111",
+            bot_token=valid_token,
+            bot_id="B111",
+            bot_user_id="W111",
+            bot_scopes=["commands"],
+            installed_at=time(),
+        )
+
+    def find_installation(
+        self,
+        *,
+        enterprise_id: Optional[str],
+        team_id: Optional[str],
+        user_id: Optional[str] = None,
+        is_enterprise_install: Optional[bool] = False,
+    ) -> Optional[Installation]:
+        return None
+
+
+class TestInstallationStoreAuthorize:
+    signing_secret = "secret"
+    mock_api_server_base_url = "http://localhost:8888"
+    signature_verifier = SignatureVerifier(signing_secret)
+    web_client = WebClient(
+        token=valid_token,
+        base_url=mock_api_server_base_url,
+    )
+
+    def setup_method(self):
+        self.old_os_env = remove_os_env_temporarily()
+        setup_mock_web_api_server(self)
+
+    def teardown_method(self):
+        cleanup_mock_web_api_server(self)
+        restore_os_env(self.old_os_env)
+
+    def generate_signature(self, body: str, timestamp: str):
+        return self.signature_verifier.generate_signature(
+            body=body,
+            timestamp=timestamp,
+        )
+
+    def build_headers(self, timestamp: str, body: str):
+        return {
+            "content-type": ["application/x-www-form-urlencoded"],
+            "x-slack-signature": [self.generate_signature(body, timestamp)],
+            "x-slack-request-timestamp": [timestamp],
+        }
+
+    def build_valid_request(self) -> BoltRequest:
+        timestamp = str(int(time()))
+        return BoltRequest(
+            body=raw_body, headers=self.build_headers(timestamp, raw_body)
+        )
+
+    def test_success(self):
+        app = App(
+            client=self.web_client,
+            installation_store=MyInstallationStore(),
+            signing_secret=self.signing_secret,
+        )
+        app.action("a")(simple_listener)
+
+        request = self.build_valid_request()
+        response = app.dispatch(request)
+        assert response.status == 200
+        assert response.body == ""
+        assert_auth_test_count(self, 1)
+
+
+body = {
+    "type": "block_actions",
+    "user": {
+        "id": "W99999",
+        "username": "primary-owner",
+        "name": "primary-owner",
+        "team_id": "T111",
+    },
+    "api_app_id": "A111",
+    "token": "verification_token",
+    "container": {
+        "type": "message",
+        "message_ts": "111.222",
+        "channel_id": "C111",
+        "is_ephemeral": True,
+    },
+    "trigger_id": "111.222.valid",
+    "team": {
+        "id": "T111",
+        "domain": "workspace-domain",
+        "enterprise_id": "E111",
+        "enterprise_name": "Sandbox Org",
+    },
+    "channel": {"id": "C111", "name": "test-channel"},
+    "response_url": "https://hooks.slack.com/actions/T111/111/random-value",
+    "actions": [
+        {
+            "action_id": "a",
+            "block_id": "b",
+            "text": {"type": "plain_text", "text": "Button", "emoji": True},
+            "value": "click_me_123",
+            "type": "button",
+            "action_ts": "1596530385.194939",
+        }
+    ],
+}
+
+raw_body = f"payload={quote(json.dumps(body))}"
+
+
+def simple_listener(ack, body, payload, action):
+    assert body["trigger_id"] == "111.222.valid"
+    assert body["actions"][0] == payload
+    assert payload == action
+    assert action["action_id"] == "a"
+    ack()