Module slack_sdk.oauth

Modules for implementing the Slack OAuth flow

https://slack.dev/python-slack-sdk/oauth/

"""Modules for implementing the Slack OAuth flow

https://slack.dev/python-slack-sdk/oauth/
"""
from .authorize_url_generator import AuthorizeUrlGenerator
from .authorize_url_generator import OpenIDConnectAuthorizeUrlGenerator
from .installation_store import InstallationStore
from .redirect_uri_page_renderer import RedirectUriPageRenderer
from .state_store import OAuthStateStore
from .state_utils import OAuthStateUtils

__all__ = [
    "AuthorizeUrlGenerator",
    "OpenIDConnectAuthorizeUrlGenerator",
    "InstallationStore",
    "RedirectUriPageRenderer",
    "OAuthStateStore",
    "OAuthStateUtils",
]

Sub-modules

slack_sdk.oauth.authorize_url_generator

slack_sdk.oauth.installation_store

slack_sdk.oauth.redirect_uri_page_renderer

slack_sdk.oauth.state_store

OAuth state parameter data store …

slack_sdk.oauth.state_utils

slack_sdk.oauth.token_rotation

Classes

class AuthorizeUrlGenerator (*, client_id: str, redirect_uri: Optional[str] = None, scopes: Optional[Sequence[str]] = None, user_scopes: Optional[Sequence[str]] = None, authorization_url: str = 'https://slack.com/oauth/v2/authorize')

Expand source code

class AuthorizeUrlGenerator:
    def __init__(
        self,
        *,
        client_id: str,
        redirect_uri: Optional[str] = None,
        scopes: Optional[Sequence[str]] = None,
        user_scopes: Optional[Sequence[str]] = None,
        authorization_url: str = "https://slack.com/oauth/v2/authorize",
    ):
        self.client_id = client_id
        self.redirect_uri = redirect_uri
        self.scopes = scopes
        self.user_scopes = user_scopes
        self.authorization_url = authorization_url

    def generate(self, state: str, team: Optional[str] = None) -> str:
        scopes = ",".join(self.scopes) if self.scopes else ""
        user_scopes = ",".join(self.user_scopes) if self.user_scopes else ""
        url = (
            f"{self.authorization_url}?"
            f"state={state}&"
            f"client_id={self.client_id}&"
            f"scope={scopes}&"
            f"user_scope={user_scopes}"
        )
        if self.redirect_uri is not None:
            url += f"&redirect_uri={self.redirect_uri}"
        if team is not None:
            url += f"&team={team}"
        return url

Methods

def generate(self, state: str, team: Optional[str] = None) ‑> str

Expand source code

def generate(self, state: str, team: Optional[str] = None) -> str:
    scopes = ",".join(self.scopes) if self.scopes else ""
    user_scopes = ",".join(self.user_scopes) if self.user_scopes else ""
    url = (
        f"{self.authorization_url}?"
        f"state={state}&"
        f"client_id={self.client_id}&"
        f"scope={scopes}&"
        f"user_scope={user_scopes}"
    )
    if self.redirect_uri is not None:
        url += f"&redirect_uri={self.redirect_uri}"
    if team is not None:
        url += f"&team={team}"
    return url

class InstallationStore

The installation store interface.

The minimum required methods are:

• save(installation)
• find_installation(enterprise_id, team_id, user_id, is_enterprise_install)

If you would like to properly handle app uninstallations and token revocations, the following methods should be implemented.

• delete_installation(enterprise_id, team_id, user_id)
• delete_all(enterprise_id, team_id)

If your app needs only bot scope installations, the simpler way to implement would be:

• save(installation)
• find_bot(enterprise_id, team_id, is_enterprise_install)
• delete_bot(enterprise_id, team_id)
• delete_all(enterprise_id, team_id)

Expand source code

class InstallationStore:
    """The installation store interface.

    The minimum required methods are:

    * save(installation)
    * find_installation(enterprise_id, team_id, user_id, is_enterprise_install)

    If you would like to properly handle app uninstallations and token revocations,
    the following methods should be implemented.

    * delete_installation(enterprise_id, team_id, user_id)
    * delete_all(enterprise_id, team_id)

    If your app needs only bot scope installations, the simpler way to implement would be:

    * save(installation)
    * find_bot(enterprise_id, team_id, is_enterprise_install)
    * delete_bot(enterprise_id, team_id)
    * delete_all(enterprise_id, team_id)
    """

    @property
    def logger(self) -> Logger:
        raise NotImplementedError()

    def save(self, installation: Installation):
        """Saves an installation data"""
        raise NotImplementedError()

    def save_bot(self, bot: Bot):
        """Saves a bot installation data"""
        raise NotImplementedError()

    def find_bot(
        self,
        *,
        enterprise_id: Optional[str],
        team_id: Optional[str],
        is_enterprise_install: Optional[bool] = False,
    ) -> Optional[Bot]:
        """Finds a bot scope installation per workspace / org"""
        raise NotImplementedError()

    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]:
        """Finds a relevant installation for the given IDs.
        If the user_id is absent, this method may return the latest installation in the workspace / org.
        """
        raise NotImplementedError()

    def delete_bot(
        self,
        *,
        enterprise_id: Optional[str],
        team_id: Optional[str],
    ) -> None:
        """Deletes a bot scope installation per workspace / org"""
        raise NotImplementedError()

    def delete_installation(
        self,
        *,
        enterprise_id: Optional[str],
        team_id: Optional[str],
        user_id: Optional[str] = None,
    ) -> None:
        """Deletes an installation that matches the given IDs"""
        raise NotImplementedError()

    def delete_all(
        self,
        *,
        enterprise_id: Optional[str],
        team_id: Optional[str],
    ):
        """Deletes all installation data for the given workspace / org"""
        self.delete_bot(enterprise_id=enterprise_id, team_id=team_id)
        self.delete_installation(enterprise_id=enterprise_id, team_id=team_id)

Subclasses

• AmazonS3InstallationStore
• CacheableInstallationStore
• FileInstallationStore
• SQLAlchemyInstallationStore
• SQLite3InstallationStore

Instance variables

var logger : logging.Logger

Expand source code

@property
def logger(self) -> Logger:
    raise NotImplementedError()

Methods

def delete_all(self, *, enterprise_id: Optional[str], team_id: Optional[str])

Deletes all installation data for the given workspace / org

Expand source code

def delete_all(
    self,
    *,
    enterprise_id: Optional[str],
    team_id: Optional[str],
):
    """Deletes all installation data for the given workspace / org"""
    self.delete_bot(enterprise_id=enterprise_id, team_id=team_id)
    self.delete_installation(enterprise_id=enterprise_id, team_id=team_id)

def delete_bot(self, *, enterprise_id: Optional[str], team_id: Optional[str]) ‑> None

Deletes a bot scope installation per workspace / org

Expand source code

def delete_bot(
    self,
    *,
    enterprise_id: Optional[str],
    team_id: Optional[str],
) -> None:
    """Deletes a bot scope installation per workspace / org"""
    raise NotImplementedError()

def delete_installation(self, *, enterprise_id: Optional[str], team_id: Optional[str], user_id: Optional[str] = None) ‑> None

Deletes an installation that matches the given IDs

Expand source code

def delete_installation(
    self,
    *,
    enterprise_id: Optional[str],
    team_id: Optional[str],
    user_id: Optional[str] = None,
) -> None:
    """Deletes an installation that matches the given IDs"""
    raise NotImplementedError()

def find_bot(self, *, enterprise_id: Optional[str], team_id: Optional[str], is_enterprise_install: Optional[bool] = False) ‑> Optional[Bot]

Finds a bot scope installation per workspace / org

Expand source code

def find_bot(
    self,
    *,
    enterprise_id: Optional[str],
    team_id: Optional[str],
    is_enterprise_install: Optional[bool] = False,
) -> Optional[Bot]:
    """Finds a bot scope installation per workspace / org"""
    raise NotImplementedError()

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]

Finds a relevant installation for the given IDs. If the user_id is absent, this method may return the latest installation in the workspace / org.

Expand source code

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]:
    """Finds a relevant installation for the given IDs.
    If the user_id is absent, this method may return the latest installation in the workspace / org.
    """
    raise NotImplementedError()

def save(self, installation: Installation)

Saves an installation data

Expand source code

def save(self, installation: Installation):
    """Saves an installation data"""
    raise NotImplementedError()

def save_bot(self, bot: Bot)

Saves a bot installation data

Expand source code

def save_bot(self, bot: Bot):
    """Saves a bot installation data"""
    raise NotImplementedError()

class OAuthStateStore

Expand source code

class OAuthStateStore:
    @property
    def logger(self) -> Logger:
        raise NotImplementedError()

    def issue(self, *args, **kwargs) -> str:
        raise NotImplementedError()

    def consume(self, state: str) -> bool:
        raise NotImplementedError()

Subclasses

• AmazonS3OAuthStateStore
• FileOAuthStateStore
• SQLAlchemyOAuthStateStore
• SQLite3OAuthStateStore

Instance variables

var logger : logging.Logger

Expand source code

@property
def logger(self) -> Logger:
    raise NotImplementedError()

Methods

def consume(self, state: str) ‑> bool

Expand source code

def consume(self, state: str) -> bool:
    raise NotImplementedError()

def issue(self, *args, **kwargs) ‑> str

Expand source code

def issue(self, *args, **kwargs) -> str:
    raise NotImplementedError()

class OAuthStateUtils (*, cookie_name: str = 'slack-app-oauth-state', expiration_seconds: int = 600)

Expand source code

class OAuthStateUtils:
    cookie_name: str
    expiration_seconds: int

    default_cookie_name: str = "slack-app-oauth-state"
    default_expiration_seconds: int = 60 * 10  # 10 minutes

    def __init__(
        self,
        *,
        cookie_name: str = default_cookie_name,
        expiration_seconds: int = default_expiration_seconds,
    ):
        self.cookie_name = cookie_name
        self.expiration_seconds = expiration_seconds

    def build_set_cookie_for_new_state(self, state: str) -> str:
        return f"{self.cookie_name}={state}; " "Secure; " "HttpOnly; " "Path=/; " f"Max-Age={self.expiration_seconds}"

    def build_set_cookie_for_deletion(self) -> str:
        return f"{self.cookie_name}=deleted; " "Secure; " "HttpOnly; " "Path=/; " "Expires=Thu, 01 Jan 1970 00:00:00 GMT"

    def is_valid_browser(
        self,
        state: Optional[str],
        request_headers: Dict[str, Union[str, Sequence[str]]],
    ) -> bool:
        if state is None or request_headers is None or request_headers.get("cookie", None) is None:
            return False
        cookies = request_headers["cookie"]
        if isinstance(cookies, str):
            cookies = [cookies]
        for cookie in cookies:
            values = cookie.split(";")
            for value in values:
                if value.strip() == f"{self.cookie_name}={state}":
                    return True
        return False

Class variables

var cookie_name : str

var default_cookie_name : str

var default_expiration_seconds : int

var expiration_seconds : int

Methods

def build_set_cookie_for_deletion(self) ‑> str

Expand source code

def build_set_cookie_for_deletion(self) -> str:
    return f"{self.cookie_name}=deleted; " "Secure; " "HttpOnly; " "Path=/; " "Expires=Thu, 01 Jan 1970 00:00:00 GMT"

def build_set_cookie_for_new_state(self, state: str) ‑> str

Expand source code

def build_set_cookie_for_new_state(self, state: str) -> str:
    return f"{self.cookie_name}={state}; " "Secure; " "HttpOnly; " "Path=/; " f"Max-Age={self.expiration_seconds}"

def is_valid_browser(self, state: Optional[str], request_headers: Dict[str, Union[str, Sequence[str]]]) ‑> bool

Expand source code

def is_valid_browser(
    self,
    state: Optional[str],
    request_headers: Dict[str, Union[str, Sequence[str]]],
) -> bool:
    if state is None or request_headers is None or request_headers.get("cookie", None) is None:
        return False
    cookies = request_headers["cookie"]
    if isinstance(cookies, str):
        cookies = [cookies]
    for cookie in cookies:
        values = cookie.split(";")
        for value in values:
            if value.strip() == f"{self.cookie_name}={state}":
                return True
    return False

class OpenIDConnectAuthorizeUrlGenerator (*, client_id: str, redirect_uri: str, scopes: Optional[Sequence[str]] = None, authorization_url: str = 'https://slack.com/openid/connect/authorize')

Refer to https://openid.net/specs/openid-connect-core-1_0.html

Expand source code

class OpenIDConnectAuthorizeUrlGenerator:
    """Refer to https://openid.net/specs/openid-connect-core-1_0.html"""

    def __init__(
        self,
        *,
        client_id: str,
        redirect_uri: str,
        scopes: Optional[Sequence[str]] = None,
        authorization_url: str = "https://slack.com/openid/connect/authorize",
    ):
        self.client_id = client_id
        self.redirect_uri = redirect_uri
        self.scopes = scopes
        self.authorization_url = authorization_url

    def generate(self, state: str, nonce: Optional[str] = None, team: Optional[str] = None) -> str:
        scopes = ",".join(self.scopes) if self.scopes else ""
        url = (
            f"{self.authorization_url}?"
            "response_type=code&"
            f"state={state}&"
            f"client_id={self.client_id}&"
            f"scope={scopes}&"
            f"redirect_uri={self.redirect_uri}"
        )
        if team is not None:
            url += f"&team={team}"
        if nonce is not None:
            url += f"&nonce={nonce}"
        return url

Methods

def generate(self, state: str, nonce: Optional[str] = None, team: Optional[str] = None) ‑> str

Expand source code

def generate(self, state: str, nonce: Optional[str] = None, team: Optional[str] = None) -> str:
    scopes = ",".join(self.scopes) if self.scopes else ""
    url = (
        f"{self.authorization_url}?"
        "response_type=code&"
        f"state={state}&"
        f"client_id={self.client_id}&"
        f"scope={scopes}&"
        f"redirect_uri={self.redirect_uri}"
    )
    if team is not None:
        url += f"&team={team}"
    if nonce is not None:
        url += f"&nonce={nonce}"
    return url

class RedirectUriPageRenderer (*, install_path: str, redirect_uri_path: str, success_url: Optional[str] = None, failure_url: Optional[str] = None)

Expand source code

class RedirectUriPageRenderer:
    def __init__(
        self,
        *,
        install_path: str,
        redirect_uri_path: str,
        success_url: Optional[str] = None,
        failure_url: Optional[str] = None,
    ):
        self.install_path = install_path
        self.redirect_uri_path = redirect_uri_path
        self.success_url = success_url
        self.failure_url = failure_url

    def render_success_page(
        self,
        app_id: str,
        team_id: Optional[str],
        is_enterprise_install: Optional[bool] = None,
        enterprise_url: Optional[str] = None,
    ) -> str:
        url = self.success_url
        if url is None:
            if is_enterprise_install is True and enterprise_url is not None and app_id is not None:
                url = f"{enterprise_url}manage/organization/apps/profile/{app_id}/workspaces/add"
            elif team_id is None or app_id is None:
                url = "slack://open"
            else:
                url = f"slack://app?team={team_id}&id={app_id}"
        browser_url = f"https://app.slack.com/client/{team_id}"

        return f"""
<html>
<head>
<meta http-equiv="refresh" content="0; URL={html.escape(url)}">
<style>
body {{
  padding: 10px 15px;
  font-family: verdana;
  text-align: center;
}}
</style>
</head>
<body>
<h2>Thank you!</h2>
<p>Redirecting to the Slack App... click <a href="{html.escape(url)}">here</a>. If you use the browser version of Slack, click <a href="{html.escape(browser_url)}" target="_blank">this link</a> instead.</p>
</body>
</html>
"""  # noqa: E501

    def render_failure_page(self, reason: str) -> str:
        return f"""
<html>
<head>
<style>
body {{
  padding: 10px 15px;
  font-family: verdana;
  text-align: center;
}}
</style>
</head>
<body>
<h2>Oops, Something Went Wrong!</h2>
<p>Please try again from <a href="{html.escape(self.install_path)}">here</a> or contact the app owner (reason: {html.escape(reason)})</p>
</body>
</html>
"""  # noqa: E501

Methods

def render_failure_page(self, reason: str) ‑> str

Expand source code

    def render_failure_page(self, reason: str) -> str:
        return f"""
<html>
<head>
<style>
body {{
  padding: 10px 15px;
  font-family: verdana;
  text-align: center;
}}
</style>
</head>
<body>
<h2>Oops, Something Went Wrong!</h2>
<p>Please try again from <a href="{html.escape(self.install_path)}">here</a> or contact the app owner (reason: {html.escape(reason)})</p>
</body>
</html>
"""  # noqa: E501

def render_success_page(self, app_id: str, team_id: Optional[str], is_enterprise_install: Optional[bool] = None, enterprise_url: Optional[str] = None) ‑> str

Expand source code

    def render_success_page(
        self,
        app_id: str,
        team_id: Optional[str],
        is_enterprise_install: Optional[bool] = None,
        enterprise_url: Optional[str] = None,
    ) -> str:
        url = self.success_url
        if url is None:
            if is_enterprise_install is True and enterprise_url is not None and app_id is not None:
                url = f"{enterprise_url}manage/organization/apps/profile/{app_id}/workspaces/add"
            elif team_id is None or app_id is None:
                url = "slack://open"
            else:
                url = f"slack://app?team={team_id}&id={app_id}"
        browser_url = f"https://app.slack.com/client/{team_id}"

        return f"""
<html>
<head>
<meta http-equiv="refresh" content="0; URL={html.escape(url)}">
<style>
body {{
  padding: 10px 15px;
  font-family: verdana;
  text-align: center;
}}
</style>
</head>
<body>
<h2>Thank you!</h2>
<p>Redirecting to the Slack App... click <a href="{html.escape(url)}">here</a>. If you use the browser version of Slack, click <a href="{html.escape(browser_url)}" target="_blank">this link</a> instead.</p>
</body>
</html>
"""  # noqa: E501