codekingpro/portable-devtools

1from requests import Session2from requests.auth import AuthBase3from authlib.oauth1 import (4    SIGNATURE_HMAC_SHA1,5    SIGNATURE_TYPE_HEADER,6)7from authlib.common.encoding import to_native8from authlib.oauth1 import ClientAuth9from authlib.oauth1.client import OAuth1Client10from ..base_client import OAuthError11from .utils import update_session_configure12 13 14class OAuth1Auth(AuthBase, ClientAuth):15    """Signs the request using OAuth 1 (RFC5849)"""16 17    def __call__(self, req):18        url, headers, body = self.prepare(19            req.method, req.url, req.headers, req.body)20 21        req.url = to_native(url)22        req.prepare_headers(headers)23        if body:24            req.body = body25        return req26 27 28class OAuth1Session(OAuth1Client, Session):29    auth_class = OAuth1Auth30 31    def __init__(self, client_id, client_secret=None,32                 token=None, token_secret=None,33                 redirect_uri=None, rsa_key=None, verifier=None,34                 signature_method=SIGNATURE_HMAC_SHA1,35                 signature_type=SIGNATURE_TYPE_HEADER,36                 force_include_body=False, **kwargs):37        Session.__init__(self)38        update_session_configure(self, kwargs)39        OAuth1Client.__init__(40            self, session=self,41            client_id=client_id, client_secret=client_secret,42            token=token, token_secret=token_secret,43            redirect_uri=redirect_uri, rsa_key=rsa_key, verifier=verifier,44            signature_method=signature_method, signature_type=signature_type,45            force_include_body=force_include_body, **kwargs)46 47    def rebuild_auth(self, prepared_request, response):48        """When being redirected we should always strip Authorization49        header, since nonce may not be reused as per OAuth spec.50        """51        if 'Authorization' in prepared_request.headers:52            # If we get redirected to a new host, we should strip out53            # any authentication headers.54            prepared_request.headers.pop('Authorization', True)55            prepared_request.prepare_auth(self.auth)56 57    @staticmethod58    def handle_error(error_type, error_description):59        raise OAuthError(error_type, error_description)60