Skip to content

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

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
seratch merged 1 commit into from
Apr 12, 2021
Merged

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

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
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
86 changes: 54 additions & 32 deletions slack_bolt/authorization/async_authorize.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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,
Expand All @@ -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,
Expand All @@ -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.
Expand All @@ -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
Expand Down
81 changes: 50 additions & 31 deletions slack_bolt/authorization/authorize.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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,
Expand All @@ -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,
Expand All @@ -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.
Expand All @@ -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
Expand Down
150 changes: 150 additions & 0 deletions tests/scenario_tests/test_installation_store_authorize.py
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -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()
Loading