authentik.sources.oauth.views.callback

  1"""OAuth Callback Views"""
  2
  3from datetime import timedelta
  4from json import JSONDecodeError
  5from typing import Any
  6
  7from django.conf import settings
  8from django.contrib import messages
  9from django.http import Http404, HttpRequest, HttpResponse
 10from django.shortcuts import redirect
 11from django.utils.timezone import now
 12from django.utils.translation import gettext as _
 13from django.views.generic import View
 14from structlog.stdlib import get_logger
 15
 16from authentik.core.sources.flow_manager import SourceFlowManager
 17from authentik.events.models import Event, EventAction
 18from authentik.sources.oauth.clients.base import BaseOAuthClient
 19from authentik.sources.oauth.models import (
 20    GroupOAuthSourceConnection,
 21    OAuthSource,
 22    UserOAuthSourceConnection,
 23)
 24from authentik.sources.oauth.views.base import OAuthClientMixin
 25
 26LOGGER = get_logger()
 27
 28
 29class OAuthCallback(OAuthClientMixin, View):
 30    "Base OAuth callback view."
 31
 32    source: OAuthSource
 33    token: dict[str, Any] | None = None
 34
 35    def dispatch(self, request: HttpRequest, *_, **kwargs) -> HttpResponse:
 36        """View Get handler"""
 37        slug = kwargs.get("source_slug", "")
 38        try:
 39            self.source = OAuthSource.objects.get(slug=slug)
 40        except OAuthSource.DoesNotExist:
 41            raise Http404(f"Unknown OAuth source '{slug}'.") from None
 42
 43        if not self.source.enabled:
 44            raise Http404(f"Source {slug} is not enabled.")
 45        client = self.get_client(self.source, callback=self.get_callback_url(self.source))
 46        # Fetch access token
 47        self.token = client.get_access_token()
 48        if self.token is None:
 49            return self.handle_login_failure("Could not retrieve token.")
 50        if "error" in self.token:
 51            return self.handle_login_failure(self.token["error"])
 52        # Fetch profile info
 53        try:
 54            res = self.redirect_flow_manager(client)
 55        except ValueError as exc:
 56            # if we're authenticated and not in a source stage and this new flag is enabled,
 57            # just continue
 58            if self.request.user.is_authenticated:
 59                pass
 60            return self.handle_login_failure(exc.args[0])
 61        return res
 62
 63    def redirect_flow_manager(self, client: BaseOAuthClient) -> HttpResponse:
 64        try:
 65            raw_info = client.get_profile_info(self.token)
 66            if raw_info is None:
 67                raise ValueError("Could not retrieve profile.")
 68        except JSONDecodeError as exc:
 69            Event.new(
 70                EventAction.CONFIGURATION_ERROR,
 71                message="Failed to JSON-decode profile.",
 72                raw_profile=exc.doc,
 73            ).from_http(self.request)
 74            raise ValueError("Could not retrieve profile.") from None
 75        identifier = self.get_user_id(info=raw_info)
 76        if identifier is None:
 77            raise ValueError("Could not determine id.")
 78        sfm = OAuthSourceFlowManager(
 79            source=self.source,
 80            request=self.request,
 81            identifier=identifier,
 82            user_info={
 83                "info": raw_info,
 84                "client": client,
 85                "token": self.token,
 86            },
 87            policy_context={
 88                "oauth_userinfo": raw_info,
 89            },
 90        )
 91        return sfm.get_flow(
 92            raw_info=raw_info,
 93            access_token=self.token.get("access_token"),
 94            refresh_token=self.token.get("refresh_token"),
 95            expires=self.token.get("expires_in"),
 96        )
 97
 98    def get_callback_url(self, source: OAuthSource) -> str:
 99        "Return callback url if different than the current url."
100        return ""
101
102    def get_error_redirect(self, source: OAuthSource, reason: str) -> str:
103        "Return url to redirect on login failure."
104        return settings.LOGIN_URL
105
106    def get_user_id(self, info: dict[str, Any]) -> str | None:
107        """Return unique identifier from the profile info."""
108        if "id" in info:
109            return str(info["id"])
110        return None
111
112    def handle_login_failure(self, reason: str) -> HttpResponse:
113        "Message user and redirect on error."
114        LOGGER.warning("Authentication Failure", reason=reason)
115        messages.error(
116            self.request,
117            _(
118                "Authentication failed: {reason}".format_map(
119                    {
120                        "reason": reason,
121                    }
122                )
123            ),
124        )
125        return redirect(self.get_error_redirect(self.source, reason))
126
127
128class OAuthSourceFlowManager(SourceFlowManager):
129    """Flow manager for oauth sources"""
130
131    user_connection_type = UserOAuthSourceConnection
132    group_connection_type = GroupOAuthSourceConnection
133
134    def update_user_connection(
135        self,
136        connection: UserOAuthSourceConnection,
137        access_token: str | None = None,
138        refresh_token: str | None = None,
139        expires_in: int | None = None,
140        **_,
141    ) -> UserOAuthSourceConnection:
142        """Set the access_token and refresh_token on the connection"""
143        connection.access_token = access_token
144        connection.refresh_token = refresh_token
145        connection.expires = now() + timedelta(seconds=expires_in) if expires_in else now()
146        return connection