authentik.sources.oauth.types.mailcow
Mailcow OAuth Views
1"""Mailcow OAuth Views""" 2 3from typing import Any 4 5from requests.exceptions import RequestException 6from structlog.stdlib import get_logger 7 8from authentik.sources.oauth.clients.oauth2 import OAuth2Client 9from authentik.sources.oauth.models import AuthorizationCodeAuthMethod 10from authentik.sources.oauth.types.registry import SourceType, registry 11from authentik.sources.oauth.views.callback import OAuthCallback 12from authentik.sources.oauth.views.redirect import OAuthRedirect 13 14LOGGER = get_logger() 15 16 17class MailcowOAuthRedirect(OAuthRedirect): 18 """Mailcow OAuth2 Redirect""" 19 20 def get_additional_parameters(self, source): # pragma: no cover 21 return { 22 "scope": ["profile"], 23 } 24 25 26class MailcowOAuth2Client(OAuth2Client): 27 """MailcowOAuth2Client, for some reason, mailcow does not like the default headers""" 28 29 def get_profile_info(self, token: dict[str, str]) -> dict[str, Any] | None: 30 "Fetch user profile information." 31 profile_url = self.source.source_type.profile_url or "" 32 if self.source.source_type.urls_customizable and self.source.profile_url: 33 profile_url = self.source.profile_url 34 response = self.session.request( 35 "get", 36 f"{profile_url}?access_token={token['access_token']}", 37 ) 38 try: 39 response.raise_for_status() 40 except RequestException as exc: 41 LOGGER.warning("Unable to fetch user profile", exc=exc, body=response.text) 42 return None 43 return response.json() 44 45 46class MailcowOAuth2Callback(OAuthCallback): 47 """Mailcow OAuth2 Callback""" 48 49 client_class = MailcowOAuth2Client 50 51 52@registry.register() 53class MailcowType(SourceType): 54 """Mailcow Type definition""" 55 56 callback_view = MailcowOAuth2Callback 57 redirect_view = MailcowOAuthRedirect 58 verbose_name = "Mailcow" 59 name = "mailcow" 60 61 urls_customizable = True 62 63 authorization_code_auth_method = AuthorizationCodeAuthMethod.POST_BODY 64 65 def get_base_user_properties(self, info: dict[str, Any], **kwargs) -> dict[str, Any]: 66 return { 67 "username": info.get("full_name"), 68 "email": info.get("email"), 69 "name": info.get("full_name"), 70 }
LOGGER =
<BoundLoggerLazyProxy(logger=None, wrapper_class=None, processors=None, context_class=None, initial_values={}, logger_factory_args=())>
18class MailcowOAuthRedirect(OAuthRedirect): 19 """Mailcow OAuth2 Redirect""" 20 21 def get_additional_parameters(self, source): # pragma: no cover 22 return { 23 "scope": ["profile"], 24 }
Mailcow OAuth2 Redirect
27class MailcowOAuth2Client(OAuth2Client): 28 """MailcowOAuth2Client, for some reason, mailcow does not like the default headers""" 29 30 def get_profile_info(self, token: dict[str, str]) -> dict[str, Any] | None: 31 "Fetch user profile information." 32 profile_url = self.source.source_type.profile_url or "" 33 if self.source.source_type.urls_customizable and self.source.profile_url: 34 profile_url = self.source.profile_url 35 response = self.session.request( 36 "get", 37 f"{profile_url}?access_token={token['access_token']}", 38 ) 39 try: 40 response.raise_for_status() 41 except RequestException as exc: 42 LOGGER.warning("Unable to fetch user profile", exc=exc, body=response.text) 43 return None 44 return response.json()
MailcowOAuth2Client, for some reason, mailcow does not like the default headers
def
get_profile_info(self, token: dict[str, str]) -> dict[str, Any] | None:
30 def get_profile_info(self, token: dict[str, str]) -> dict[str, Any] | None: 31 "Fetch user profile information." 32 profile_url = self.source.source_type.profile_url or "" 33 if self.source.source_type.urls_customizable and self.source.profile_url: 34 profile_url = self.source.profile_url 35 response = self.session.request( 36 "get", 37 f"{profile_url}?access_token={token['access_token']}", 38 ) 39 try: 40 response.raise_for_status() 41 except RequestException as exc: 42 LOGGER.warning("Unable to fetch user profile", exc=exc, body=response.text) 43 return None 44 return response.json()
Fetch user profile information.
Inherited Members
47class MailcowOAuth2Callback(OAuthCallback): 48 """Mailcow OAuth2 Callback""" 49 50 client_class = MailcowOAuth2Client
Mailcow OAuth2 Callback
client_class =
<class 'MailcowOAuth2Client'>
53@registry.register() 54class MailcowType(SourceType): 55 """Mailcow Type definition""" 56 57 callback_view = MailcowOAuth2Callback 58 redirect_view = MailcowOAuthRedirect 59 verbose_name = "Mailcow" 60 name = "mailcow" 61 62 urls_customizable = True 63 64 authorization_code_auth_method = AuthorizationCodeAuthMethod.POST_BODY 65 66 def get_base_user_properties(self, info: dict[str, Any], **kwargs) -> dict[str, Any]: 67 return { 68 "username": info.get("full_name"), 69 "email": info.get("email"), 70 "name": info.get("full_name"), 71 }
Mailcow Type definition
callback_view =
<class 'MailcowOAuth2Callback'>
redirect_view =
<class 'MailcowOAuthRedirect'>
def
get_base_user_properties(self, info: dict[str, typing.Any], **kwargs) -> dict[str, typing.Any]:
66 def get_base_user_properties(self, info: dict[str, Any], **kwargs) -> dict[str, Any]: 67 return { 68 "username": info.get("full_name"), 69 "email": info.get("email"), 70 "name": info.get("full_name"), 71 }
Get base user properties for enrollment/update