authentik.sources.oauth.types.slack

  1"""Slack OAuth Views"""
  2
  3from typing import Any
  4
  5from django.http import Http404
  6
  7from authentik.sources.oauth.clients.oauth2 import OAuth2Client
  8from authentik.sources.oauth.models import OAuthSource
  9from authentik.sources.oauth.types.registry import SourceType, registry
 10from authentik.sources.oauth.views.callback import OAuthCallback
 11from authentik.sources.oauth.views.redirect import OAuthRedirect
 12
 13
 14class SlackOAuthClient(OAuth2Client):
 15    """Slack OAuth2 Client that handles Slack's nested token response.
 16
 17    Slack's oauth.v2.access returns tokens in a nested structure:
 18    {
 19        "ok": true,
 20        "access_token": "xoxb-...",  # bot token
 21        "refresh_token": "xoxe-1-...",  # bot refresh token (if rotation enabled)
 22        "authed_user": {
 23            "id": "U1234",
 24            "scope": "...",
 25            "access_token": "xoxp-...",  # user token
 26            "refresh_token": "xoxe-1-...",  # user refresh token (if rotation enabled)
 27            "token_type": "user",
 28            "expires_in": 43200
 29        }
 30    }
 31
 32    For user scopes (like admin for SCIM), we need the authed_user token.
 33    """
 34
 35    def get_access_token(self, **request_kwargs) -> dict[str, Any] | None:
 36        """Fetch access token and normalize Slack's nested response."""
 37        token = super().get_access_token(**request_kwargs)
 38        if token is None or "error" in token:
 39            return token
 40
 41        # If we have authed_user with access_token, use that (user token)
 42        # If authed_user isn't there, then we were given a bot token
 43        if "authed_user" in token and "access_token" in token.get("authed_user", {}):
 44            authed_user = token["authed_user"]
 45            token["access_token"] = authed_user["access_token"]
 46
 47            if "refresh_token" in authed_user:
 48                token["refresh_token"] = authed_user["refresh_token"]
 49            if "expires_in" in authed_user:
 50                token["expires_in"] = authed_user["expires_in"]
 51            token["id"] = authed_user.get("id")
 52
 53        # Slack returns "user", but API expects Bearer
 54        # not a password, OAuth token type
 55        token["token_type"] = "Bearer"  # nosec
 56
 57        return token
 58
 59
 60class SlackOAuthRedirect(OAuthRedirect):
 61    """Slack OAuth2 Redirect
 62
 63    Slack uses two separate scope parameters:
 64    - scope: Bot token scopes (xoxb- tokens)
 65    - user_scope: User token scopes (xoxp- tokens)
 66
 67    For user authentication and SCIM,
 68    we need scopes in user_scope, not scope.
 69    """
 70
 71    def get_additional_parameters(self, source):
 72        # Start with base user scopes for authentication
 73        user_scopes = ["openid", "email", "profile"]
 74
 75        # Add any additional scopes from the source config to user_scope
 76        # (not to scope, which is for bot tokens)
 77        if source.additional_scopes:
 78            additional = source.additional_scopes
 79            if additional.startswith("*"):
 80                additional = additional[1:]
 81            user_scopes.extend(additional.split())
 82
 83        return {
 84            "scope": [],
 85            "user_scope": user_scopes,
 86        }
 87
 88    def get_redirect_url(self, **kwargs) -> str:
 89        """Build redirect URL with Slack-specific scope handling.
 90
 91        Slack uses two separate scope parameters:
 92        - scope: Bot token scopes (xoxb- tokens)
 93        - user_scope: User token scopes (xoxp- tokens)
 94
 95        The base class adds additional_scopes to 'scope', but Slack needs them
 96        in 'user_scope'. We override completely to handle this properly.
 97        """
 98
 99        slug = kwargs.get("source_slug", "")
100        try:
101            source: OAuthSource = OAuthSource.objects.get(slug=slug)
102        except OAuthSource.DoesNotExist:
103            raise Http404(f"Unknown OAuth source '{slug}'.") from None
104        if not source.enabled:
105            raise Http404(f"source {slug} is not enabled.")
106
107        client = self.get_client(source, callback=self.get_callback_url(source))
108        # get_additional_parameters handles all scopes for Slack (both scope and user_scope)
109        params = self.get_additional_parameters(source)
110        params.update(self._try_login_hint_extract())
111        return client.get_redirect_url(params)
112
113
114class SlackOAuth2Callback(OAuthCallback):
115    """Slack OAuth2 Callback"""
116
117    client_class = SlackOAuthClient
118
119    def get_user_id(self, info: dict[str, Any]) -> str | None:
120        """Return unique identifier from Slack profile info."""
121        return info.get("sub")
122
123
124@registry.register()
125class SlackType(SourceType):
126    """Slack Type definition"""
127
128    callback_view = SlackOAuth2Callback
129    redirect_view = SlackOAuthRedirect
130    verbose_name = "Slack"
131    name = "slack"
132
133    authorization_url = "https://slack.com/oauth/v2/authorize"
134    access_token_url = "https://slack.com/api/oauth.v2.access"  # nosec
135    profile_url = "https://slack.com/api/openid.connect.userInfo"
136
137    def get_base_user_properties(self, source, info: dict[str, Any], **kwargs) -> dict[str, Any]:
138        return {
139            "username": info.get("name"),
140            "email": info.get("email"),
141            "name": info.get("name"),
142        }