authentik.sources.oauth.types.registry

  1"""Source type manager"""
  2
  3from enum import Enum
  4from typing import Any
  5
  6from django.http.request import HttpRequest
  7from django.templatetags.static import static
  8from django.urls.base import reverse
  9from structlog.stdlib import get_logger
 10
 11from authentik.flows.challenge import Challenge, RedirectChallenge
 12from authentik.sources.oauth.models import AuthorizationCodeAuthMethod, OAuthSource, PKCEMethod
 13from authentik.sources.oauth.views.callback import OAuthCallback
 14from authentik.sources.oauth.views.redirect import OAuthRedirect
 15
 16LOGGER = get_logger()
 17
 18
 19class RequestKind(Enum):
 20    """Enum of OAuth Request types"""
 21
 22    CALLBACK = "callback"
 23    REDIRECT = "redirect"
 24
 25
 26class SourceType:
 27    """Source type, allows overriding of urls and views per type"""
 28
 29    callback_view = OAuthCallback
 30    redirect_view = OAuthRedirect
 31    name: str = "default"
 32    verbose_name: str = "Default source type"
 33
 34    urls_customizable = False
 35
 36    request_token_url: str | None = None
 37    authorization_url: str | None = None
 38    access_token_url: str | None = None
 39    profile_url: str | None = None
 40    oidc_well_known_url: str | None = None
 41    oidc_jwks_url: str | None = None
 42    pkce: PKCEMethod = PKCEMethod.NONE
 43
 44    authorization_code_auth_method: AuthorizationCodeAuthMethod = (
 45        AuthorizationCodeAuthMethod.BASIC_AUTH
 46    )
 47
 48    def icon_url(self) -> str:
 49        """Get Icon URL for login"""
 50        return static(f"authentik/sources/{self.name}.svg")
 51
 52    def login_challenge(self, source: OAuthSource, request: HttpRequest) -> Challenge:
 53        """Allow types to return custom challenges"""
 54        return RedirectChallenge(
 55            data={
 56                "to": reverse(
 57                    "authentik_sources_oauth:oauth-client-login",
 58                    kwargs={"source_slug": source.slug},
 59                ),
 60            }
 61        )
 62
 63    def get_base_user_properties(
 64        self, source: OAuthSource, info: dict[str, Any], **kwargs
 65    ) -> dict[str, Any | dict[str, Any]]:
 66        """Get base user properties for enrollment/update"""
 67        return info
 68
 69    def get_base_group_properties(
 70        self, source: OAuthSource, group_id: str, **kwargs
 71    ) -> dict[str, Any | dict[str, Any]]:
 72        """Get base group properties for creation/update"""
 73        return {
 74            "name": group_id,
 75        }
 76
 77
 78class SourceTypeRegistry:
 79    """Registry to hold all Source types."""
 80
 81    def __init__(self) -> None:
 82        self.__sources: list[type[SourceType]] = []
 83
 84    def register(self):
 85        """Class decorator to register classes inline."""
 86
 87        def inner_wrapper(cls):
 88            self.__sources.append(cls)
 89            return cls
 90
 91        return inner_wrapper
 92
 93    def get(self):
 94        """Get a list of all source types"""
 95        return self.__sources
 96
 97    def get_name_tuple(self):
 98        """Get list of tuples of all registered names"""
 99        return [(x.name, x.verbose_name) for x in self.__sources]
100
101    def find_type(self, type_name: str) -> type[SourceType]:
102        """Find type based on source"""
103        found_type = None
104        for src_type in self.__sources:
105            if src_type.name == type_name:
106                return src_type
107        if not found_type:
108            found_type = SourceType
109            LOGGER.warning(
110                "no matching type found, using default",
111                wanted=type_name,
112                have=[x.name for x in self.__sources],
113            )
114        return found_type
115
116    def find(self, type_name: str, kind: RequestKind) -> type[OAuthCallback | OAuthRedirect]:
117        """Find fitting Source Type"""
118        found_type = self.find_type(type_name)
119        if kind == RequestKind.CALLBACK:
120            return found_type.callback_view
121        if kind == RequestKind.REDIRECT:
122            return found_type.redirect_view
123        raise ValueError
124
125
126registry = SourceTypeRegistry()