Skip to content

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

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
merged 2 commits into from
Mar 13, 2023
Merged

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

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
6c46fda
Enhance AuthorizeResult to have bot/user_scopes when they're availabl…
seratch Mar 11, 2023
1eea8f6
Fix
seratch Mar 11, 2023
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
48 changes: 33 additions & 15 deletions slack_bolt/authorization/async_authorize.py
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -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
Expand Down Expand Up @@ -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,
Expand All @@ -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
Expand All @@ -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
Expand All @@ -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:
Expand All @@ -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
Expand All @@ -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
Expand Down
48 changes: 33 additions & 15 deletions slack_bolt/authorization/authorize.py
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -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
Expand Down Expand Up @@ -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,
Expand All @@ -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
Expand All @@ -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
Expand All @@ -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:
Expand All @@ -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
Expand All @@ -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
Expand Down
23 changes: 22 additions & 1 deletion slack_bolt/authorization/authorize_result.py
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,4 +1,4 @@
from typing import Optional
from typing import Optional, List, Union

from slack_sdk.web import SlackResponse

Expand All @@ -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,
Expand All @@ -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:
Expand All @@ -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,
)
Loading