codekingpro/portable-devtools
115k
1from flask import g, redirect, request, session2from ..requests_client import OAuth1Session, OAuth2Session3from ..base_client import (4 BaseApp, OAuthError,5 OAuth1Mixin, OAuth2Mixin, OpenIDMixin,6)7 8 9class FlaskAppMixin:10 @property11 def token(self):12 attr = f'_oauth_token_{self.name}'13 token = g.get(attr)14 if token:15 return token16 if self._fetch_token:17 token = self._fetch_token()18 self.token = token19 return token20 21 @token.setter22 def token(self, token):23 attr = f'_oauth_token_{self.name}'24 setattr(g, attr, token)25 26 def _get_requested_token(self, *args, **kwargs):27 return self.token28 29 def save_authorize_data(self, **kwargs):30 state = kwargs.pop('state', None)31 if state:32 self.framework.set_state_data(session, state, kwargs)33 else:34 raise RuntimeError('Missing state value')35 36 def authorize_redirect(self, redirect_uri=None, **kwargs):37 """Create a HTTP Redirect for Authorization Endpoint.38 39 :param redirect_uri: Callback or redirect URI for authorization.40 :param kwargs: Extra parameters to include.41 :return: A HTTP redirect response.42 """43 rv = self.create_authorization_url(redirect_uri, **kwargs)44 self.save_authorize_data(redirect_uri=redirect_uri, **rv)45 return redirect(rv['url'])46 47 48class FlaskOAuth1App(FlaskAppMixin, OAuth1Mixin, BaseApp):49 client_cls = OAuth1Session50 51 def authorize_access_token(self, **kwargs):52 """Fetch access token in one step.53 54 :return: A token dict.55 """56 params = request.args.to_dict(flat=True)57 state = params.get('oauth_token')58 if not state:59 raise OAuthError(description='Missing "oauth_token" parameter')60 61 data = self.framework.get_state_data(session, state)62 if not data:63 raise OAuthError(description='Missing "request_token" in temporary data')64 65 params['request_token'] = data['request_token']66 params.update(kwargs)67 self.framework.clear_state_data(session, state)68 token = self.fetch_access_token(**params)69 self.token = token70 return token71 72 73class FlaskOAuth2App(FlaskAppMixin, OAuth2Mixin, OpenIDMixin, BaseApp):74 client_cls = OAuth2Session75 76 def authorize_access_token(self, **kwargs):77 """Fetch access token in one step.78 79 :return: A token dict.80 """81 if request.method == 'GET':82 error = request.args.get('error')83 if error:84 description = request.args.get('error_description')85 raise OAuthError(error=error, description=description)86 87 params = {88 'code': request.args['code'],89 'state': request.args.get('state'),90 }91 else:92 params = {93 'code': request.form['code'],94 'state': request.form.get('state'),95 }96 97 claims_options = kwargs.pop('claims_options', None)98 state_data = self.framework.get_state_data(session, params.get('state'))99 self.framework.clear_state_data(session, params.get('state'))100 params = self._format_state_params(state_data, params)101 token = self.fetch_access_token(**params, **kwargs)102 self.token = token103 104 if 'id_token' in token and 'nonce' in state_data:105 userinfo = self.parse_id_token(token, nonce=state_data['nonce'], claims_options=claims_options)106 token['userinfo'] = userinfo107 return token108 