codekingpro/portable-devtools

1import time2import logging3from authlib.common.urls import urlparse4from authlib.consts import default_user_agent5from authlib.common.security import generate_token6from .errors import (7    MismatchingStateError,8    MissingRequestTokenError,9    MissingTokenError,10)11 12log = logging.getLogger(__name__)13 14 15class BaseApp:16    client_cls = None17    OAUTH_APP_CONFIG = None18 19    def request(self, method, url, token=None, **kwargs):20        raise NotImplementedError()21 22    def get(self, url, **kwargs):23        """Invoke GET http request.24 25        If ``api_base_url`` configured, shortcut is available::26 27            client.get('users/lepture')28        """29        return self.request('GET', url, **kwargs)30 31    def post(self, url, **kwargs):32        """Invoke POST http request.33 34        If ``api_base_url`` configured, shortcut is available::35 36            client.post('timeline', json={'text': 'Hi'})37        """38        return self.request('POST', url, **kwargs)39 40    def patch(self, url, **kwargs):41        """Invoke PATCH http request.42 43        If ``api_base_url`` configured, shortcut is available::44 45            client.patch('profile', json={'name': 'Hsiaoming Yang'})46        """47        return self.request('PATCH', url, **kwargs)48 49    def put(self, url, **kwargs):50        """Invoke PUT http request.51 52        If ``api_base_url`` configured, shortcut is available::53 54            client.put('profile', json={'name': 'Hsiaoming Yang'})55        """56        return self.request('PUT', url, **kwargs)57 58    def delete(self, url, **kwargs):59        """Invoke DELETE http request.60 61        If ``api_base_url`` configured, shortcut is available::62 63            client.delete('posts/123')64        """65        return self.request('DELETE', url, **kwargs)66 67 68class _RequestMixin:69    def _get_requested_token(self, request):70        if self._fetch_token and request:71            return self._fetch_token(request)72 73    def _send_token_request(self, session, method, url, token, kwargs):74        request = kwargs.pop('request', None)75        withhold_token = kwargs.get('withhold_token')76        if self.api_base_url and not url.startswith(('https://', 'http://')):77            url = urlparse.urljoin(self.api_base_url, url)78 79        if withhold_token:80            return session.request(method, url, **kwargs)81 82        if token is None:83            token = self._get_requested_token(request)84 85        if token is None:86            raise MissingTokenError()87 88        session.token = token89        return session.request(method, url, **kwargs)90 91 92class OAuth1Base:93    client_cls = None94 95    def __init__(96            self, framework, name=None, fetch_token=None,97            client_id=None, client_secret=None,98            request_token_url=None, request_token_params=None,99            access_token_url=None, access_token_params=None,100            authorize_url=None, authorize_params=None,101            api_base_url=None, client_kwargs=None, user_agent=None, **kwargs):102        self.framework = framework103        self.name = name104        self.client_id = client_id105        self.client_secret = client_secret106        self.request_token_url = request_token_url107        self.request_token_params = request_token_params108        self.access_token_url = access_token_url109        self.access_token_params = access_token_params110        self.authorize_url = authorize_url111        self.authorize_params = authorize_params112        self.api_base_url = api_base_url113        self.client_kwargs = client_kwargs or {}114 115        self._fetch_token = fetch_token116        self._user_agent = user_agent or default_user_agent117        self._kwargs = kwargs118 119    def _get_oauth_client(self):120        session = self.client_cls(self.client_id, self.client_secret, **self.client_kwargs)121        session.headers['User-Agent'] = self._user_agent122        return session123 124 125class OAuth1Mixin(_RequestMixin, OAuth1Base):126    def request(self, method, url, token=None, **kwargs):127        with self._get_oauth_client() as session:128            return self._send_token_request(session, method, url, token, kwargs)129 130    def create_authorization_url(self, redirect_uri=None, **kwargs):131        """Generate the authorization url and state for HTTP redirect.132 133        :param redirect_uri: Callback or redirect URI for authorization.134        :param kwargs: Extra parameters to include.135        :return: dict136        """137        if not self.authorize_url:138            raise RuntimeError('Missing "authorize_url" value')139 140        if self.authorize_params:141            kwargs.update(self.authorize_params)142 143        with self._get_oauth_client() as client:144            client.redirect_uri = redirect_uri145            params = self.request_token_params or {}146            request_token = client.fetch_request_token(self.request_token_url, **params)147            log.debug(f'Fetch request token: {request_token!r}')148            url = client.create_authorization_url(self.authorize_url, **kwargs)149            state = request_token['oauth_token']150        return {'url': url, 'request_token': request_token, 'state': state}151 152    def fetch_access_token(self, request_token=None, **kwargs):153        """Fetch access token in one step.154 155        :param request_token: A previous request token for OAuth 1.156        :param kwargs: Extra parameters to fetch access token.157        :return: A token dict.158        """159        with self._get_oauth_client() as client:160            if request_token is None:161                raise MissingRequestTokenError()162            # merge request token with verifier163            token = {}164            token.update(request_token)165            token.update(kwargs)166            client.token = token167            params = self.access_token_params or {}168            token = client.fetch_access_token(self.access_token_url, **params)169        return token170 171 172class OAuth2Base:173    client_cls = None174 175    def __init__(176            self, framework, name=None, fetch_token=None, update_token=None,177            client_id=None, client_secret=None,178            access_token_url=None, access_token_params=None,179            authorize_url=None, authorize_params=None,180            api_base_url=None, client_kwargs=None, server_metadata_url=None,181            compliance_fix=None, client_auth_methods=None, user_agent=None, **kwargs):182        self.framework = framework183        self.name = name184        self.client_id = client_id185        self.client_secret = client_secret186        self.access_token_url = access_token_url187        self.access_token_params = access_token_params188        self.authorize_url = authorize_url189        self.authorize_params = authorize_params190        self.api_base_url = api_base_url191        self.client_kwargs = client_kwargs or {}192 193        self.compliance_fix = compliance_fix194        self.client_auth_methods = client_auth_methods195        self._fetch_token = fetch_token196        self._update_token = update_token197        self._user_agent = user_agent or default_user_agent198 199        self._server_metadata_url = server_metadata_url200        self.server_metadata = kwargs201 202    def _on_update_token(self, token, refresh_token=None, access_token=None):203        raise NotImplementedError()204 205    def _get_oauth_client(self, **metadata):206        client_kwargs = {}207        client_kwargs.update(self.client_kwargs)208        client_kwargs.update(metadata)209 210        if self.authorize_url:211            client_kwargs['authorization_endpoint'] = self.authorize_url212        if self.access_token_url:213            client_kwargs['token_endpoint'] = self.access_token_url214 215        session = self.client_cls(216            client_id=self.client_id,217            client_secret=self.client_secret,218            update_token=self._on_update_token,219            **client_kwargs220        )221        if self.client_auth_methods:222            for f in self.client_auth_methods:223                session.register_client_auth_method(f)224 225        if self.compliance_fix:226            self.compliance_fix(session)227 228        session.headers['User-Agent'] = self._user_agent229        return session230 231    @staticmethod232    def _format_state_params(state_data, params):233        if state_data is None:234            raise MismatchingStateError()235 236        code_verifier = state_data.get('code_verifier')237        if code_verifier:238            params['code_verifier'] = code_verifier239 240        redirect_uri = state_data.get('redirect_uri')241        if redirect_uri:242            params['redirect_uri'] = redirect_uri243        return params244 245    @staticmethod246    def _create_oauth2_authorization_url(client, authorization_endpoint, **kwargs):247        rv = {}248        if client.code_challenge_method:249            code_verifier = kwargs.get('code_verifier')250            if not code_verifier:251                code_verifier = generate_token(48)252                kwargs['code_verifier'] = code_verifier253            rv['code_verifier'] = code_verifier254            log.debug(f'Using code_verifier: {code_verifier!r}')255 256        scope = kwargs.get('scope', client.scope)257        if scope and 'openid' in scope.split():258            # this is an OpenID Connect service259            nonce = kwargs.get('nonce')260            if not nonce:261                nonce = generate_token(20)262                kwargs['nonce'] = nonce263            rv['nonce'] = nonce264 265        url, state = client.create_authorization_url(266            authorization_endpoint, **kwargs)267        rv['url'] = url268        rv['state'] = state269        return rv270 271 272class OAuth2Mixin(_RequestMixin, OAuth2Base):273    def _on_update_token(self, token, refresh_token=None, access_token=None):274        if callable(self._update_token):275            self._update_token(276                token,277                refresh_token=refresh_token,278                access_token=access_token,279            )280        self.framework.update_token(281            token,282            refresh_token=refresh_token,283            access_token=access_token,284        )285 286    def request(self, method, url, token=None, **kwargs):287        metadata = self.load_server_metadata()288        with self._get_oauth_client(**metadata) as session:289            return self._send_token_request(session, method, url, token, kwargs)290 291    def load_server_metadata(self):292        if self._server_metadata_url and '_loaded_at' not in self.server_metadata:293            with self.client_cls(**self.client_kwargs) as session:294                resp = session.request('GET', self._server_metadata_url, withhold_token=True)295                resp.raise_for_status()296                metadata = resp.json()297 298            metadata['_loaded_at'] = time.time()299            self.server_metadata.update(metadata)300        return self.server_metadata301 302    def create_authorization_url(self, redirect_uri=None, **kwargs):303        """Generate the authorization url and state for HTTP redirect.304 305        :param redirect_uri: Callback or redirect URI for authorization.306        :param kwargs: Extra parameters to include.307        :return: dict308        """309        metadata = self.load_server_metadata()310        authorization_endpoint = self.authorize_url or metadata.get('authorization_endpoint')311 312        if not authorization_endpoint:313            raise RuntimeError('Missing "authorize_url" value')314 315        if self.authorize_params:316            kwargs.update(self.authorize_params)317 318 319        with self._get_oauth_client(**metadata) as client:320            if redirect_uri is not None:321                client.redirect_uri = redirect_uri322            return self._create_oauth2_authorization_url(323                client, authorization_endpoint, **kwargs)324 325    def fetch_access_token(self, redirect_uri=None, **kwargs):326        """Fetch access token in the final step.327 328        :param redirect_uri: Callback or Redirect URI that is used in329                             previous :meth:`authorize_redirect`.330        :param kwargs: Extra parameters to fetch access token.331        :return: A token dict.332        """333        metadata = self.load_server_metadata()334        token_endpoint = self.access_token_url or metadata.get('token_endpoint')335        with self._get_oauth_client(**metadata) as client:336            if redirect_uri is not None:337                client.redirect_uri = redirect_uri338            params = {}339            if self.access_token_params:340                params.update(self.access_token_params)341            params.update(kwargs)342            token = client.fetch_token(token_endpoint, **params)343            return token344