Module slack_bolt.middleware.authorization.async_multi_teams_authorization
Classes
-
A middleware can process request data before other middleware and listener functions.
Multi-workspace authorization.
Args
authorize
- The function to authorize incoming requests from Slack.
base_logger
- The base logger
user_token_resolution
- "authed_user" or "actor"
user_facing_authorize_error_message
- The user-facing error message when installation is not found
Expand source code
class AsyncMultiTeamsAuthorization(AsyncAuthorization): authorize: AsyncAuthorize user_token_resolution: str def __init__( self, authorize: AsyncAuthorize, base_logger: Optional[Logger] = None, user_token_resolution: str = "authed_user", user_facing_authorize_error_message: Optional[str] = None, ): """Multi-workspace authorization. Args: authorize: The function to authorize incoming requests from Slack. base_logger: The base logger user_token_resolution: "authed_user" or "actor" user_facing_authorize_error_message: The user-facing error message when installation is not found """ self.authorize = authorize self.logger = get_bolt_logger(AsyncMultiTeamsAuthorization, base_logger=base_logger) self.user_token_resolution = user_token_resolution self.user_facing_authorize_error_message = ( user_facing_authorize_error_message or _build_user_facing_authorize_error_message() ) async def async_process( self, *, req: AsyncBoltRequest, resp: BoltResponse, # As this method is not supposed to be invoked by bolt-python users, # the naming conflict with the built-in one affects # only the internals of this method next: Callable[[], Awaitable[BoltResponse]], ) -> BoltResponse: if _is_no_auth_required(req): return await next() if _is_no_auth_test_call_required(req): req.context.set_authorize_result( AuthorizeResult( enterprise_id=req.context.enterprise_id, team_id=req.context.team_id, user_id=req.context.user_id, ) ) return await next() try: auth_result: Optional[AuthorizeResult] = None if self.user_token_resolution == "actor": auth_result = await self.authorize( context=req.context, enterprise_id=req.context.enterprise_id, team_id=req.context.team_id, user_id=req.context.user_id, actor_enterprise_id=req.context.actor_enterprise_id, actor_team_id=req.context.actor_team_id, actor_user_id=req.context.actor_user_id, ) else: auth_result = await self.authorize( context=req.context, enterprise_id=req.context.enterprise_id, team_id=req.context.team_id, user_id=req.context.user_id, ) if auth_result: req.context.set_authorize_result(auth_result) token = auth_result.bot_token or auth_result.user_token req.context["token"] = token # As AsyncApp#_init_context() generates a new AsyncWebClient for this request, # it's safe to modify this instance. req.context.client.token = token return await next() else: # This situation can arise if: # * A developer installed the app from the "Install to Workspace" button in Slack app config page # * The InstallationStore failed to save or deleted the installation for this workspace self.logger.error( "Although the app should be installed into this workspace, " "the AuthorizeResult (returned value from authorize) for it was not found." ) if req.context.response_url is not None: await req.context.respond(self.user_facing_authorize_error_message) # type: ignore[misc] return BoltResponse(status=200, body="") return _build_user_facing_error_response(self.user_facing_authorize_error_message) except SlackApiError as e: self.logger.error(f"Failed to authorize with the given token ({e})") return _build_user_facing_error_response(self.user_facing_authorize_error_message)
Ancestors
- AsyncAuthorization
- AsyncMiddleware
- abc.ABC
Class variables
Inherited members