Module slack_bolt.middleware.authorization.multi_teams_authorization

Classes

class MultiTeamsAuthorization (*, authorize: Authorize, base_logger: Optional[logging.Logger] = None, user_token_resolution: str = 'authed_user', user_facing_authorize_error_message: Optional[str] = None)

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 MultiTeamsAuthorization(Authorization):
    authorize: Authorize
    user_token_resolution: str

    def __init__(
        self,
        *,
        authorize: Authorize,
        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(MultiTeamsAuthorization, 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()
        )

    def process(
        self,
        *,
        req: BoltRequest,
        resp: BoltResponse,
        next: Callable[[], BoltResponse],
    ) -> BoltResponse:
        if _is_no_auth_required(req):
            return 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 next()

        try:
            auth_result: Optional[AuthorizeResult] = None
            if self.user_token_resolution == "actor":
                auth_result = 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 = 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 is not None:
                req.context.set_authorize_result(auth_result)
                token = auth_result.bot_token or auth_result.user_token
                req.context["token"] = token
                # As App#_init_context() generates a new WebClient for this request,
                # it's safe to modify this instance.
                req.context.client.token = token  # type: ignore[union-attr]
                return 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:
                    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

Class variables

var authorize : Authorize
var user_token_resolution : str

Inherited members