codekingpro/portable-devtools

1"""2    flask_security.oauth_provider3    ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~4 5    Class and methods to create providers for oauth_glue.6    Example providers for github and google.7 8    :copyright: (c) 2024-2024 by J. Christopher Wagner (jwag).9    :license: MIT, see LICENSE for more details.10 11"""12 13from __future__ import annotations14import collections.abc as cabc15import sys16 17try:18    from authlib.integrations.flask_client import OAuth19    from authlib.integrations.base_client.errors import (20        OAuthError,21    )22except ImportError:  # pragma: no cover23    pass24 25import typing as t26 27from flask import redirect28 29from .utils import (30    config_value as cv,31    do_flash,32    get_message,33    get_url,34    url_for_security,35)36 37if t.TYPE_CHECKING:  # pragma: no cover38    from flask.typing import ResponseValue39 40if sys.version_info >= (3, 9):41    OauthCbType = cabc.Callable[["OAuth", t.Any], tuple[str, t.Any]]42else:43    OauthCbType = t.Callable[["OAuth", t.Any], t.Tuple[str, t.Any]]  # pragma: no cover44 45 46class FsOAuthProvider:47    """48    Subclass this or instantiate to add new oauth providers.49 50    Subclassing allows for customizing additional aspects of the oauth flow51    in particular - a custom error path for oauth flow state mismatches and52    other errors thrown by authlib.53 54    Call security.oauthglue.register_provider_ext(myproviderclass("myprovider"))55 56    :param name: a name for provider - must match what was passed if this57     is already registered with Oauth.58    :param registration_info: This dict is passed directly to Oauth as59     part of registration - not needed if provider already registered with60     Oauth61    :param fetch_identity_cb: Call back from response to oauth flow.62    """63 64    def __init__(65        self,66        name: str,67        registration_info: dict[str, t.Any] | None = None,68        fetch_identity_cb: OauthCbType | None = None,69    ):70        self.name = name71        self._registration_info = registration_info or {}72        self._fetch_identity_cb = fetch_identity_cb73 74    def authlib_config(self) -> dict[str, t.Any]:75        """Return dict with authlib configuration.76        This is called as part of provider registration."""77        return self._registration_info78 79    def fetch_identity_cb(self, oauth: OAuth, token: t.Any) -> tuple[str, t.Any]:80        """This callback is called when the oauth81        redirect happens. It must take the response from the provider and return82        a tuple of <user_model_field_name, value> - which will be used83        to look up the user in the datastore."""84        if not self._fetch_identity_cb:  # pragma no cover85            raise NotImplementedError86        return self._fetch_identity_cb(oauth, token)87 88    def oauth_response_failure(self, e: OAuthError) -> ResponseValue:89        """Called if authlib authorize_access_token throws an error.90 91        N.B. flashing doesn't seem to work in some cases - if the session92        cookie has samesite='strict' and it is the first registration.93        """94        m, c = get_message(95            "OAUTH_HANDSHAKE_ERROR", exerror=e.error, exdesc=e.description96        )97        if cv("REDIRECT_BEHAVIOR") == "spa":98            return redirect(get_url(cv("LOGIN_ERROR_VIEW"), qparams={c: m}))99        do_flash(m, c)100        return redirect(url_for_security("login"))101 102 103class GitHubFsOauthProvider(FsOAuthProvider):104    def authlib_config(self):105        return dict(106            access_token_url="https://github.com/login/oauth/access_token",107            access_token_params=None,108            authorize_url="https://github.com/login/oauth/authorize",109            authorize_params=None,110            api_base_url="https://api.github.com/",111            client_kwargs={"scope": "user:email"},112        )113 114    def fetch_identity_cb(self, oauth, token):115        resp = oauth.github.get("user", token=token)116        profile = resp.json()117        return "email", profile["email"]118 119 120class GoogleFsOauthProvider(FsOAuthProvider):121    def authlib_config(self):122        return dict(123            server_metadata_url="https://accounts.google.com/"124            ".well-known/openid-configuration",125            client_kwargs={"scope": "openid email profile"},126        )127 128    def fetch_identity_cb(self, oauth, token):  # pragma no cover129        profile = token["userinfo"]130        return "email", profile["email"]131