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.models import (
 19    GroupOAuthSourceConnection,
 20    OAuthSource,
 21    UserOAuthSourceConnection,
 22)
 23from authentik.sources.oauth.views.base import OAuthClientMixin
 24
 25LOGGER = get_logger()
 26
 27
 28class OAuthCallback(OAuthClientMixin, View):
 29    "Base OAuth callback view."
 30
 31    source: OAuthSource
 32    token: dict | None = None
 33
 34    def dispatch(self, request: HttpRequest, *_, **kwargs) -> HttpResponse:
 35        """View Get handler"""
 36        slug = kwargs.get("source_slug", "")
 37        try:
 38            self.source = OAuthSource.objects.get(slug=slug)
 39        except OAuthSource.DoesNotExist:
 40            raise Http404(f"Unknown OAuth source '{slug}'.") from None
 41
 42        if not self.source.enabled:
 43            raise Http404(f"Source {slug} is not enabled.")
 44        client = self.get_client(self.source, callback=self.get_callback_url(self.source))
 45        # Fetch access token
 46        self.token = client.get_access_token()
 47        if self.token is None:
 48            return self.handle_login_failure("Could not retrieve token.")
 49        if "error" in self.token:
 50            return self.handle_login_failure(self.token["error"])
 51        # Fetch profile info
 52        try:
 53            raw_info = client.get_profile_info(self.token)
 54            if raw_info is None:
 55                return self.handle_login_failure("Could not retrieve profile.")
 56        except JSONDecodeError as exc:
 57            Event.new(
 58                EventAction.CONFIGURATION_ERROR,
 59                message="Failed to JSON-decode profile.",
 60                raw_profile=exc.doc,
 61            ).from_http(self.request)
 62            return self.handle_login_failure("Could not retrieve profile.")
 63        identifier = self.get_user_id(info=raw_info)
 64        if identifier is None:
 65            return self.handle_login_failure("Could not determine id.")
 66        sfm = OAuthSourceFlowManager(
 67            source=self.source,
 68            request=self.request,
 69            identifier=identifier,
 70            user_info={
 71                "info": raw_info,
 72                "client": client,
 73                "token": self.token,
 74            },
 75            policy_context={
 76                "oauth_userinfo": raw_info,
 77            },
 78        )
 79        return sfm.get_flow(
 80            raw_info=raw_info,
 81            access_token=self.token.get("access_token"),
 82            refresh_token=self.token.get("refresh_token"),
 83            expires=self.token.get("expires_in"),
 84        )
 85
 86    def get_callback_url(self, source: OAuthSource) -> str:
 87        "Return callback url if different than the current url."
 88        return ""
 89
 90    def get_error_redirect(self, source: OAuthSource, reason: str) -> str:
 91        "Return url to redirect on login failure."
 92        return settings.LOGIN_URL
 93
 94    def get_user_id(self, info: dict[str, Any]) -> str | None:
 95        """Return unique identifier from the profile info."""
 96        if "id" in info:
 97            return info["id"]
 98        return None
 99
100    def handle_login_failure(self, reason: str) -> HttpResponse:
101        "Message user and redirect on error."
102        LOGGER.warning("Authentication Failure", reason=reason)
103        messages.error(
104            self.request,
105            _(
106                "Authentication failed: {reason}".format_map(
107                    {
108                        "reason": reason,
109                    }
110                )
111            ),
112        )
113        return redirect(self.get_error_redirect(self.source, reason))
114
115
116class OAuthSourceFlowManager(SourceFlowManager):
117    """Flow manager for oauth sources"""
118
119    user_connection_type = UserOAuthSourceConnection
120    group_connection_type = GroupOAuthSourceConnection
121
122    def update_user_connection(
123        self,
124        connection: UserOAuthSourceConnection,
125        access_token: str | None = None,
126        refresh_token: str | None = None,
127        expires_in: int | None = None,
128        **_,
129    ) -> UserOAuthSourceConnection:
130        """Set the access_token and refresh_token on the connection"""
131        connection.access_token = access_token
132        connection.refresh_token = refresh_token
133        connection.expires = now() + timedelta(seconds=expires_in) if expires_in else now()
134        return connection