codekingpro/portable-devtools

1from starlette.datastructures import URL2from starlette.responses import RedirectResponse3from ..base_client import OAuthError4from ..base_client import BaseApp5from ..base_client.async_app import AsyncOAuth1Mixin, AsyncOAuth2Mixin6from ..base_client.async_openid import AsyncOpenIDMixin7from ..httpx_client import AsyncOAuth1Client, AsyncOAuth2Client8 9 10class StarletteAppMixin:11    async def save_authorize_data(self, request, **kwargs):12        state = kwargs.pop('state', None)13        if state:14            if self.framework.cache:15                session = None16            else:17                session = request.session18            await self.framework.set_state_data(session, state, kwargs)19        else:20            raise RuntimeError('Missing state value')21 22    async def authorize_redirect(self, request, redirect_uri=None, **kwargs):23        """Create a HTTP Redirect for Authorization Endpoint.24 25        :param request: HTTP request instance from Starlette view.26        :param redirect_uri: Callback or redirect URI for authorization.27        :param kwargs: Extra parameters to include.28        :return: A HTTP redirect response.29        """30 31        # Handle Starlette >= 0.26.0 where redirect_uri may now be a URL and not a string32        if redirect_uri and isinstance(redirect_uri, URL):33            redirect_uri = str(redirect_uri)34        rv = await self.create_authorization_url(redirect_uri, **kwargs)35        await self.save_authorize_data(request, redirect_uri=redirect_uri, **rv)36        return RedirectResponse(rv['url'], status_code=302)37 38 39class StarletteOAuth1App(StarletteAppMixin, AsyncOAuth1Mixin, BaseApp):40    client_cls = AsyncOAuth1Client41 42    async def authorize_access_token(self, request, **kwargs):43        params = dict(request.query_params)44        state = params.get('oauth_token')45        if not state:46            raise OAuthError(description='Missing "oauth_token" parameter')47 48        data = await self.framework.get_state_data(request.session, state)49        if not data:50            raise OAuthError(description='Missing "request_token" in temporary data')51 52        params['request_token'] = data['request_token']53        params.update(kwargs)54        await self.framework.clear_state_data(request.session, state)55        return await self.fetch_access_token(**params)56 57 58class StarletteOAuth2App(StarletteAppMixin, AsyncOAuth2Mixin, AsyncOpenIDMixin, BaseApp):59    client_cls = AsyncOAuth2Client60 61    async def authorize_access_token(self, request, **kwargs):62        error = request.query_params.get('error')63        if error:64            description = request.query_params.get('error_description')65            raise OAuthError(error=error, description=description)66 67        params = {68            'code': request.query_params.get('code'),69            'state': request.query_params.get('state'),70        }71 72        if self.framework.cache:73            session = None74        else:75            session = request.session76 77        claims_options = kwargs.pop('claims_options', None)78        state_data = await self.framework.get_state_data(session, params.get('state'))79        await self.framework.clear_state_data(session, params.get('state'))80        params = self._format_state_params(state_data, params)81        token = await self.fetch_access_token(**params, **kwargs)82 83        if 'id_token' in token and 'nonce' in state_data:84            userinfo = await self.parse_id_token(token, nonce=state_data['nonce'], claims_options=claims_options)85            token['userinfo'] = userinfo86        return token87