Enhance AuthorizeResult to have bot/user_scopes & resolve user_id for user token (#855)

Enhance AuthorizeResult to have bot/user_scopes when they're available (optional)
Also, this commit fixes the bug where authorize_result.user_id is not resolved when both bot and user tokens exist.

Author: seratch

slack_bolt/authorization/async_authorize.py

@@ -1,5 +1,5 @@
 from logging import Logger
-from typing import Optional, Callable, Awaitable, Dict, Any
+from typing import Optional, Callable, Awaitable, Dict, Any, List
 
 from slack_sdk.errors import SlackApiError
 from slack_sdk.oauth.installation_store import Bot, Installation
@@ -156,14 +156,18 @@ async def __call__(
 
         bot_token: Optional[str] = None
         user_token: Optional[str] = None
+        bot_scopes: Optional[List[str]] = None
+        user_scopes: Optional[List[str]] = None
+        latest_bot_installation: Optional[Installation] = None
+        this_user_installation: Optional[Installation] = None
 
         if not self.bot_only and self.find_installation_available:
             # 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.
-                latest_installation: Optional[Installation] = await self.installation_store.async_find_installation(
+                latest_bot_installation = await self.installation_store.async_find_installation(
                     enterprise_id=enterprise_id,
                     team_id=team_id,
                     is_enterprise_install=context.is_enterprise_install,
@@ -173,20 +177,20 @@ async def __call__(
                 # The example use cases are:
                 # - The app's installation requires both bot and user tokens
                 # - The app has two installation paths 1) bot installation 2) individual user authorization
-                this_user_installation: Optional[Installation] = None
-
-                if latest_installation is not None:
+                if latest_bot_installation is not None:
                     # Save the latest bot token
-                    bot_token = latest_installation.bot_token  # this still can be None
-                    user_token = latest_installation.user_token  # this still can be None
+                    bot_token = latest_bot_installation.bot_token  # this still can be None
+                    user_token = latest_bot_installation.user_token  # this still can be None
+                    bot_scopes = latest_bot_installation.bot_scopes  # this still can be None
+                    user_scopes = latest_bot_installation.user_scopes  # this still can be None
 
-                    if latest_installation.user_id != user_id:
+                    if latest_bot_installation.user_id != user_id:
                         # First off, remove the user token as the installer is a different user
                         user_token = None
-                        latest_installation.user_token = None
-                        latest_installation.user_refresh_token = None
-                        latest_installation.user_token_expires_at = None
-                        latest_installation.user_scopes = []
+                        latest_bot_installation.user_token = None
+                        latest_bot_installation.user_refresh_token = None
+                        latest_bot_installation.user_token_expires_at = None
+                        latest_bot_installation.user_scopes = []
 
                         # try to fetch the request user's installation
                         # to reflect the user's access token if exists
@@ -198,26 +202,32 @@ async def __call__(
                         )
                         if this_user_installation is not None:
                             user_token = this_user_installation.user_token
-                            if latest_installation.bot_token is None:
+                            user_scopes = this_user_installation.user_scopes
+                            if latest_bot_installation.bot_token is None:
                                 # If latest_installation has a bot token, we never overwrite the value
                                 bot_token = this_user_installation.bot_token
+                                bot_scopes = this_user_installation.bot_scopes
 
                             # If token rotation is enabled, running rotation may be needed here
                             refreshed = await self._rotate_and_save_tokens_if_necessary(this_user_installation)
                             if refreshed is not None:
                                 user_token = refreshed.user_token
-                                if latest_installation.bot_token is None:
+                                user_scopes = refreshed.user_scopes
+                                if latest_bot_installation.bot_token is None:
                                     # If latest_installation has a bot token, we never overwrite the value
                                     bot_token = refreshed.bot_token
+                                    bot_scopes = refreshed.bot_scopes
 
                     # If token rotation is enabled, running rotation may be needed here
-                    refreshed = await self._rotate_and_save_tokens_if_necessary(latest_installation)
+                    refreshed = await self._rotate_and_save_tokens_if_necessary(latest_bot_installation)
                     if refreshed is not None:
                         bot_token = refreshed.bot_token
+                        bot_scopes = refreshed.bot_scopes
                         if this_user_installation is None:
                             # Only when we don't have `this_user_installation` here,
                             # the `user_token` is for the user associated with this request
                             user_token = refreshed.user_token
+                            user_scopes = refreshed.user_scopes
 
             except NotImplementedError as _:
                 self.find_installation_available = False
@@ -238,6 +248,7 @@ async def __call__(
                 )
                 if bot is not None:
                     bot_token = bot.bot_token
+                    bot_scopes = bot.bot_scopes
                     if bot.bot_refresh_token is not None:
                         # Token rotation
                         if self.token_rotator is None:
@@ -249,6 +260,7 @@ async def __call__(
                         if refreshed is not None:
                             await self.installation_store.async_save_bot(refreshed)
                             bot_token = refreshed.bot_token
+                            bot_scopes = refreshed.bot_scopes
 
             except NotImplementedError as _:
                 self.find_bot_available = False
@@ -267,10 +279,16 @@ async def __call__(
 
         try:
             auth_test_api_response = await context.client.auth_test(token=token)
+            user_auth_test_response = None
+            if user_token is not None and token != user_token:
+                user_auth_test_response = await context.client.auth_test(token=user_token)
             authorize_result = AuthorizeResult.from_auth_test_response(
                 auth_test_response=auth_test_api_response,
+                user_auth_test_response=user_auth_test_response,
                 bot_token=bot_token,
                 user_token=user_token,
+                bot_scopes=bot_scopes,
+                user_scopes=user_scopes,
             )
             if self.cache_enabled:
                 self.authorize_result_cache[token] = authorize_result

slack_bolt/authorization/authorize.py

@@ -1,5 +1,5 @@
 from logging import Logger
-from typing import Optional, Callable, Dict, Any
+from typing import Optional, Callable, Dict, Any, List
 
 from slack_sdk.errors import SlackApiError
 from slack_sdk.oauth import InstallationStore
@@ -156,14 +156,18 @@ def __call__(
 
         bot_token: Optional[str] = None
         user_token: Optional[str] = None
+        bot_scopes: Optional[List[str]] = None
+        user_scopes: Optional[List[str]] = None
+        latest_bot_installation: Optional[Installation] = None
+        this_user_installation: Optional[Installation] = None
 
         if not self.bot_only and self.find_installation_available:
             # 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.
-                latest_installation: Optional[Installation] = self.installation_store.find_installation(
+                latest_bot_installation = self.installation_store.find_installation(
                     enterprise_id=enterprise_id,
                     team_id=team_id,
                     is_enterprise_install=context.is_enterprise_install,
@@ -173,20 +177,20 @@ def __call__(
                 # The example use cases are:
                 # - The app's installation requires both bot and user tokens
                 # - The app has two installation paths 1) bot installation 2) individual user authorization
-                this_user_installation: Optional[Installation] = None
-
-                if latest_installation is not None:
+                if latest_bot_installation is not None:
                     # Save the latest bot token
-                    bot_token = latest_installation.bot_token  # this still can be None
-                    user_token = latest_installation.user_token  # this still can be None
+                    bot_token = latest_bot_installation.bot_token  # this still can be None
+                    user_token = latest_bot_installation.user_token  # this still can be None
+                    bot_scopes = latest_bot_installation.bot_scopes  # this still can be None
+                    user_scopes = latest_bot_installation.user_scopes  # this still can be None
 
-                    if latest_installation.user_id != user_id:
+                    if latest_bot_installation.user_id != user_id:
                         # First off, remove the user token as the installer is a different user
                         user_token = None
-                        latest_installation.user_token = None
-                        latest_installation.user_refresh_token = None
-                        latest_installation.user_token_expires_at = None
-                        latest_installation.user_scopes = []
+                        latest_bot_installation.user_token = None
+                        latest_bot_installation.user_refresh_token = None
+                        latest_bot_installation.user_token_expires_at = None
+                        latest_bot_installation.user_scopes = []
 
                         # try to fetch the request user's installation
                         # to reflect the user's access token if exists
@@ -198,26 +202,32 @@ def __call__(
                         )
                         if this_user_installation is not None:
                             user_token = this_user_installation.user_token
-                            if latest_installation.bot_token is None:
+                            user_scopes = this_user_installation.user_scopes
+                            if latest_bot_installation.bot_token is None:
                                 # If latest_installation has a bot token, we never overwrite the value
                                 bot_token = this_user_installation.bot_token
+                                bot_scopes = this_user_installation.bot_scopes
 
                             # If token rotation is enabled, running rotation may be needed here
                             refreshed = self._rotate_and_save_tokens_if_necessary(this_user_installation)
                             if refreshed is not None:
                                 user_token = refreshed.user_token
-                                if latest_installation.bot_token is None:
+                                user_scopes = refreshed.user_scopes
+                                if latest_bot_installation.bot_token is None:
                                     # If latest_installation has a bot token, we never overwrite the value
                                     bot_token = refreshed.bot_token
+                                    bot_scopes = refreshed.bot_scopes
 
                     # If token rotation is enabled, running rotation may be needed here
-                    refreshed = self._rotate_and_save_tokens_if_necessary(latest_installation)
+                    refreshed = self._rotate_and_save_tokens_if_necessary(latest_bot_installation)
                     if refreshed is not None:
                         bot_token = refreshed.bot_token
+                        bot_scopes = refreshed.bot_scopes
                         if this_user_installation is None:
                             # Only when we don't have `this_user_installation` here,
                             # the `user_token` is for the user associated with this request
                             user_token = refreshed.user_token
+                            user_scopes = refreshed.user_scopes
 
             except NotImplementedError as _:
                 self.find_installation_available = False
@@ -238,6 +248,7 @@ def __call__(
                 )
                 if bot is not None:
                     bot_token = bot.bot_token
+                    bot_scopes = bot.bot_scopes
                     if bot.bot_refresh_token is not None:
                         # Token rotation
                         if self.token_rotator is None:
@@ -249,6 +260,7 @@ def __call__(
                         if refreshed is not None:
                             self.installation_store.save_bot(refreshed)
                             bot_token = refreshed.bot_token
+                            bot_scopes = refreshed.bot_scopes
 
             except NotImplementedError as _:
                 self.find_bot_available = False
@@ -267,10 +279,16 @@ def __call__(
 
         try:
             auth_test_api_response = context.client.auth_test(token=token)
+            user_auth_test_response = None
+            if user_token is not None and token != user_token:
+                user_auth_test_response = context.client.auth_test(token=user_token)
             authorize_result = AuthorizeResult.from_auth_test_response(
                 auth_test_response=auth_test_api_response,
+                user_auth_test_response=user_auth_test_response,
                 bot_token=bot_token,
                 user_token=user_token,
+                bot_scopes=bot_scopes,
+                user_scopes=user_scopes,
             )
             if self.cache_enabled:
                 self.authorize_result_cache[token] = authorize_result

slack_bolt/authorization/authorize_result.py

@@ -1,4 +1,4 @@
-from typing import Optional
+from typing import Optional, List, Union
 
 from slack_sdk.web import SlackResponse
 
@@ -11,8 +11,10 @@ class AuthorizeResult(dict):
     bot_id: Optional[str]
     bot_user_id: Optional[str]
     bot_token: Optional[str]
+    bot_scopes: Optional[List[str]]  # since v1.17
     user_id: Optional[str]
     user_token: Optional[str]
+    user_scopes: Optional[List[str]]  # since v1.17
 
     def __init__(
         self,
@@ -23,9 +25,11 @@ def __init__(
         bot_user_id: Optional[str] = None,
         bot_id: Optional[str] = None,
         bot_token: Optional[str] = None,
+        bot_scopes: Optional[Union[List[str], str]] = None,
         # user
         user_id: Optional[str] = None,
         user_token: Optional[str] = None,
+        user_scopes: Optional[Union[List[str], str]] = None,
     ):
         """
         Args:
@@ -34,39 +38,56 @@ def __init__(
             bot_user_id: Bot user's User ID starting with either `U` or `W`
             bot_id: Bot ID starting with `B`
             bot_token: Bot user access token starting with `xoxb-`
+            bot_scopes: The scopes associated with the bot token
             user_id: The request user ID
             user_token: User access token starting with `xoxp-`
+            user_scopes: The scopes associated wth the user token
         """
         self["enterprise_id"] = self.enterprise_id = enterprise_id
         self["team_id"] = self.team_id = team_id
         # bot
         self["bot_user_id"] = self.bot_user_id = bot_user_id
         self["bot_id"] = self.bot_id = bot_id
         self["bot_token"] = self.bot_token = bot_token
+        if bot_scopes is not None and isinstance(bot_scopes, str):
+            bot_scopes = [scope.strip() for scope in bot_scopes.split(",")]
+        self["bot_scopes"] = self.bot_scopes = bot_scopes  # type: ignore
         # user
         self["user_id"] = self.user_id = user_id
         self["user_token"] = self.user_token = user_token
+        if user_scopes is not None and isinstance(user_scopes, str):
+            user_scopes = [scope.strip() for scope in user_scopes.split(",")]
+        self["user_scopes"] = self.user_scopes = user_scopes  # type: ignore
 
     @classmethod
     def from_auth_test_response(
         cls,
         *,
         bot_token: Optional[str] = None,
         user_token: Optional[str] = None,
+        bot_scopes: Optional[Union[List[str], str]] = None,
+        user_scopes: Optional[Union[List[str], str]] = None,
         auth_test_response: SlackResponse,
+        user_auth_test_response: Optional[SlackResponse] = None,
     ) -> "AuthorizeResult":
         bot_user_id: Optional[str] = (  # type:ignore
             auth_test_response.get("user_id") if auth_test_response.get("bot_id") is not None else None
         )
         user_id: Optional[str] = (  # type:ignore
             auth_test_response.get("user_id") if auth_test_response.get("bot_id") is None else None
         )
+        # Since v1.28, user_id can be set when user_token w/ its auth.test response exists
+        if user_id is None and user_auth_test_response is not None:
+            user_id: Optional[str] = user_auth_test_response.get("user_id")  # type:ignore
+
         return AuthorizeResult(
             enterprise_id=auth_test_response.get("enterprise_id"),
             team_id=auth_test_response.get("team_id"),
             bot_id=auth_test_response.get("bot_id"),
             bot_user_id=bot_user_id,
+            bot_scopes=bot_scopes,
             user_id=user_id,
             bot_token=bot_token,
             user_token=user_token,
+            user_scopes=user_scopes,
         )