<s> """<STR_LIT>""" <EOL> import datetime <EOL> import time <EOL> import random <EOL> import urllib . request , urllib . parse , urllib . error <EOL> import urllib . parse <EOL> import atom . http_core <EOL> try : <EOL> import simplejson <EOL> except ImportError : <EOL> try : <EOL> from django . utils import simplejson <EOL> except ImportError : <EOL> import json as simplejson <EOL> try : <EOL> from urllib . parse import parse_qsl <EOL> except ImportError : <EOL> from cgi import parse_qsl <EOL> __author__ = '<STR_LIT>' <EOL> PROGRAMMATIC_AUTH_LABEL = '<STR_LIT>' <EOL> AUTHSUB_AUTH_LABEL = '<STR_LIT>' <EOL> OAUTH2_AUTH_LABEL = '<STR_LIT>' <EOL> AUTH_SCOPES = { <EOL> '<STR_LIT>' : ( <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' ) , <EOL> '<STR_LIT>' : ( <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' ) , <EOL> '<STR_LIT>' : ( <EOL> '<STR_LIT>' , ) , <EOL> '<STR_LIT>' : ( <EOL> '<STR_LIT>' , ) , <EOL> '<STR_LIT>' : ( <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' ) , <EOL> '<STR_LIT>' : ( <EOL> '<STR_LIT>' , ) , <EOL> '<STR_LIT>' : ( <EOL> '<STR_LIT>' , ) , <EOL> '<STR_LIT>' : ( <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' ) , <EOL> '<STR_LIT>' : ( <EOL> '<STR_LIT>' , ) , <EOL> '<STR_LIT>' : ( <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' ) , <EOL> '<STR_LIT>' : ( <EOL> '<STR_LIT>' , ) , <EOL> '<STR_LIT>' : ( <EOL> '<STR_LIT>' , ) , <EOL> '<STR_LIT>' : ( <EOL> '<STR_LIT>' , ) , <EOL> '<STR_LIT>' : ( <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' ) , <EOL> '<STR_LIT>' : ( <EOL> '<STR_LIT>' , ) , <EOL> '<STR_LIT>' : ( <EOL> '<STR_LIT>' , ) , <EOL> '<STR_LIT>' : ( <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' ) , <EOL> '<STR_LIT>' : ( <EOL> '<STR_LIT>' , ) , <EOL> '<STR_LIT:code>' : ( <EOL> '<STR_LIT>' , ) } <EOL> class Error ( Exception ) : <EOL> pass <EOL> class UnsupportedTokenType ( Error ) : <EOL> """<STR_LIT>""" <EOL> pass <EOL> class OAuth2AccessTokenError ( Error ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , error_message ) : <EOL> self . error_message = error_message <EOL> def generate_client_login_request_body ( email , password , service , source , <EOL> account_type = '<STR_LIT>' , captcha_token = None , <EOL> captcha_response = None ) : <EOL> """<STR_LIT>""" <EOL> request_fields = { '<STR_LIT>' : email , <EOL> '<STR_LIT>' : password , <EOL> '<STR_LIT>' : account_type , <EOL> '<STR_LIT>' : service , <EOL> '<STR_LIT:source>' : source } <EOL> if captcha_token and captcha_response : <EOL> request_fields [ '<STR_LIT>' ] = captcha_token <EOL> request_fields [ '<STR_LIT>' ] = captcha_response <EOL> return urllib . parse . urlencode ( request_fields ) <EOL> GenerateClientLoginRequestBody = generate_client_login_request_body <EOL> def get_client_login_token_string ( http_body ) : <EOL> """<STR_LIT>""" <EOL> for response_line in http_body . splitlines ( ) : <EOL> if response_line . decode ( ) . startswith ( '<STR_LIT>' ) : <EOL> return response_line [ <NUM_LIT:5> : ] . decode ( "<STR_LIT:ascii>" ) <EOL> return None <EOL> GetClientLoginTokenString = get_client_login_token_string <EOL> def get_captcha_challenge ( http_body , <EOL> captcha_base_url = '<STR_LIT>' ) : <EOL> """<STR_LIT>""" <EOL> contains_captcha_challenge = False <EOL> captcha_parameters = { } <EOL> for response_line in http_body . splitlines ( ) : <EOL> if response_line . startswith ( '<STR_LIT>' ) : <EOL> contains_captcha_challenge = True <EOL> elif response_line . startswith ( '<STR_LIT>' ) : <EOL> captcha_parameters [ '<STR_LIT>' ] = response_line [ <NUM_LIT> : ] <EOL> elif response_line . startswith ( '<STR_LIT>' ) : <EOL> captcha_parameters [ '<STR_LIT:url>' ] = '<STR_LIT>' % ( captcha_base_url , <EOL> response_line [ <NUM_LIT:11> : ] ) <EOL> if contains_captcha_challenge : <EOL> return captcha_parameters <EOL> else : <EOL> return None <EOL> GetCaptchaChallenge = get_captcha_challenge <EOL> class ClientLoginToken ( object ) : <EOL> def __init__ ( self , token_string ) : <EOL> if isinstance ( token_string , str ) : <EOL> self . token_string = token_string <EOL> else : <EOL> self . token_string = token_string . decode ( ) <EOL> def modify_request ( self , http_request ) : <EOL> http_request . headers [ '<STR_LIT>' ] = '<STR_LIT>' % ( PROGRAMMATIC_AUTH_LABEL , <EOL> self . token_string ) <EOL> ModifyRequest = modify_request <EOL> def _to_uri ( str_or_uri ) : <EOL> if isinstance ( str_or_uri , str ) : <EOL> return atom . http_core . Uri . parse_uri ( str_or_uri ) <EOL> return str_or_uri <EOL> def generate_auth_sub_url ( next , scopes , secure = False , session = True , <EOL> request_url = atom . http_core . parse_uri ( <EOL> '<STR_LIT>' ) , <EOL> domain = '<STR_LIT:default>' , scopes_param_prefix = '<STR_LIT>' ) : <EOL> """<STR_LIT>""" <EOL> if isinstance ( next , str ) : <EOL> next = atom . http_core . Uri . parse_uri ( next ) <EOL> if isinstance ( scopes , ( str , atom . http_core . Uri ) ) : <EOL> scopes = ( scopes , ) <EOL> scopes_string = '<STR_LIT:U+0020>' . join ( [ str ( scope ) for scope in scopes ] ) <EOL> next . query [ scopes_param_prefix ] = scopes_string <EOL> if isinstance ( request_url , str ) : <EOL> request_url = atom . http_core . Uri . parse_uri ( request_url ) <EOL> request_url . query [ '<STR_LIT>' ] = str ( next ) <EOL> request_url . query [ '<STR_LIT>' ] = scopes_string <EOL> if session : <EOL> request_url . query [ '<STR_LIT>' ] = '<STR_LIT:1>' <EOL> else : <EOL> request_url . query [ '<STR_LIT>' ] = '<STR_LIT:0>' <EOL> if secure : <EOL> request_url . query [ '<STR_LIT>' ] = '<STR_LIT:1>' <EOL> else : <EOL> request_url . query [ '<STR_LIT>' ] = '<STR_LIT:0>' <EOL> request_url . query [ '<STR_LIT>' ] = domain <EOL> return request_url <EOL> def auth_sub_string_from_url ( url , scopes_param_prefix = '<STR_LIT>' ) : <EOL> """<STR_LIT>""" <EOL> if isinstance ( url , str ) : <EOL> url = atom . http_core . Uri . parse_uri ( url ) <EOL> if '<STR_LIT>' not in url . query : <EOL> return ( None , None ) <EOL> token = url . query [ '<STR_LIT>' ] <EOL> scopes = None <EOL> if scopes_param_prefix in url . query : <EOL> scopes = tuple ( url . query [ scopes_param_prefix ] . split ( '<STR_LIT:U+0020>' ) ) <EOL> return ( token , scopes ) <EOL> AuthSubStringFromUrl = auth_sub_string_from_url <EOL> def auth_sub_string_from_body ( http_body ) : <EOL> """<STR_LIT>""" <EOL> for response_line in http_body . splitlines ( ) : <EOL> if response_line . startswith ( '<STR_LIT>' ) : <EOL> return response_line [ <NUM_LIT:6> : ] <EOL> return None <EOL> class AuthSubToken ( object ) : <EOL> def __init__ ( self , token_string , scopes = None ) : <EOL> self . token_string = token_string <EOL> self . scopes = scopes or [ ] <EOL> def modify_request ( self , http_request ) : <EOL> """<STR_LIT>""" <EOL> http_request . headers [ '<STR_LIT>' ] = '<STR_LIT>' % ( AUTHSUB_AUTH_LABEL , <EOL> self . token_string ) <EOL> ModifyRequest = modify_request <EOL> def from_url ( str_or_uri ) : <EOL> """<STR_LIT>""" <EOL> token_and_scopes = auth_sub_string_from_url ( str_or_uri ) <EOL> return AuthSubToken ( token_and_scopes [ <NUM_LIT:0> ] , token_and_scopes [ <NUM_LIT:1> ] ) <EOL> from_url = staticmethod ( from_url ) <EOL> FromUrl = from_url <EOL> def _upgrade_token ( self , http_body ) : <EOL> """<STR_LIT>""" <EOL> self . token_string = auth_sub_string_from_body ( http_body ) <EOL> def build_auth_sub_data ( http_request , timestamp , nonce ) : <EOL> """<STR_LIT>""" <EOL> return '<STR_LIT>' % ( http_request . method , str ( http_request . uri ) , <EOL> str ( timestamp ) , nonce ) <EOL> def generate_signature ( data , rsa_key ) : <EOL> """<STR_LIT>""" <EOL> import base64 <EOL> try : <EOL> from . tlslite . utils import keyfactory <EOL> except ImportError : <EOL> try : <EOL> from gdata . tlslite . utils import keyfactory <EOL> except ImportError : <EOL> from . tlslite . tlslite . utils import keyfactory <EOL> private_key = keyfactory . parsePrivateKey ( rsa_key ) <EOL> signed = private_key . hashAndSign ( data ) <EOL> if hasattr ( base64 , '<STR_LIT>' ) : <EOL> return base64 . b64encode ( signed ) <EOL> else : <EOL> return base64 . encodestring ( signed ) . replace ( '<STR_LIT:\n>' , '<STR_LIT>' ) <EOL> class SecureAuthSubToken ( AuthSubToken ) : <EOL> def __init__ ( self , token_string , rsa_private_key , scopes = None ) : <EOL> self . token_string = token_string <EOL> self . scopes = scopes or [ ] <EOL> self . rsa_private_key = rsa_private_key <EOL> def from_url ( str_or_uri , rsa_private_key ) : <EOL> """<STR_LIT>""" <EOL> token_and_scopes = auth_sub_string_from_url ( str_or_uri ) <EOL> return SecureAuthSubToken ( token_and_scopes [ <NUM_LIT:0> ] , rsa_private_key , <EOL> token_and_scopes [ <NUM_LIT:1> ] ) <EOL> from_url = staticmethod ( from_url ) <EOL> FromUrl = from_url <EOL> def modify_request ( self , http_request ) : <EOL> """<STR_LIT>""" <EOL> timestamp = str ( int ( time . time ( ) ) ) <EOL> nonce = '<STR_LIT>' . join ( [ str ( random . randint ( <NUM_LIT:0> , <NUM_LIT:9> ) ) for i in range ( <NUM_LIT:15> ) ] ) <EOL> data = build_auth_sub_data ( http_request , timestamp , nonce ) <EOL> signature = generate_signature ( data , self . rsa_private_key ) <EOL> http_request . headers [ '<STR_LIT>' ] = ( <EOL> '<STR_LIT>' % ( AUTHSUB_AUTH_LABEL , <EOL> self . token_string , data , signature ) ) <EOL> ModifyRequest = modify_request <EOL> RSA_SHA1 = '<STR_LIT>' <EOL> HMAC_SHA1 = '<STR_LIT>' <EOL> def build_oauth_base_string ( http_request , consumer_key , nonce , signaure_type , <EOL> timestamp , version , next = '<STR_LIT>' , token = None , <EOL> verifier = None ) : <EOL> """<STR_LIT>""" <EOL> params = http_request . uri . query . copy ( ) <EOL> params [ '<STR_LIT>' ] = consumer_key <EOL> params [ '<STR_LIT>' ] = nonce <EOL> params [ '<STR_LIT>' ] = signaure_type <EOL> params [ '<STR_LIT>' ] = str ( timestamp ) <EOL> if next is not None : <EOL> params [ '<STR_LIT>' ] = str ( next ) <EOL> if token is not None : <EOL> params [ '<STR_LIT>' ] = token <EOL> if version is not None : <EOL> params [ '<STR_LIT>' ] = version <EOL> if verifier is not None : <EOL> params [ '<STR_LIT>' ] = verifier <EOL> sorted_keys = None <EOL> try : <EOL> sorted_keys = sorted ( params . keys ( ) ) <EOL> except NameError : <EOL> sorted_keys = list ( params . keys ( ) ) <EOL> sorted_keys . sort ( ) <EOL> pairs = [ ] <EOL> for key in sorted_keys : <EOL> pairs . append ( '<STR_LIT>' % ( urllib . parse . quote ( key , safe = '<STR_LIT>' ) , <EOL> urllib . parse . quote ( params [ key ] , safe = '<STR_LIT>' ) ) ) <EOL> all_parameters = urllib . parse . quote ( '<STR_LIT:&>' . join ( pairs ) , safe = '<STR_LIT>' ) <EOL> normailzed_host = http_request . uri . host . lower ( ) <EOL> normalized_scheme = ( http_request . uri . scheme or '<STR_LIT:http>' ) . lower ( ) <EOL> non_default_port = None <EOL> if ( http_request . uri . port is not None <EOL> and ( ( normalized_scheme == '<STR_LIT>' and http_request . uri . port != <NUM_LIT> ) <EOL> or ( normalized_scheme == '<STR_LIT:http>' and http_request . uri . port != <NUM_LIT> ) ) ) : <EOL> non_default_port = http_request . uri . port <EOL> path = http_request . uri . path or '<STR_LIT:/>' <EOL> request_path = None <EOL> if not path . startswith ( '<STR_LIT:/>' ) : <EOL> path = '<STR_LIT>' % path <EOL> if non_default_port is not None : <EOL> request_path = urllib . parse . quote ( '<STR_LIT>' % ( <EOL> normalized_scheme , normailzed_host , non_default_port , path ) , safe = '<STR_LIT>' ) <EOL> else : <EOL> request_path = urllib . parse . quote ( '<STR_LIT>' % ( <EOL> normalized_scheme , normailzed_host , path ) , safe = '<STR_LIT>' ) <EOL> base_string = '<STR_LIT:&>' . join ( ( http_request . method . upper ( ) , request_path , <EOL> all_parameters ) ) <EOL> return base_string <EOL> def generate_hmac_signature ( http_request , consumer_key , consumer_secret , <EOL> timestamp , nonce , version , next = '<STR_LIT>' , <EOL> token = None , token_secret = None , verifier = None ) : <EOL> import hmac <EOL> import base64 <EOL> base_string = build_oauth_base_string ( <EOL> http_request , consumer_key , nonce , HMAC_SHA1 , timestamp , version , <EOL> next , token , verifier = verifier ) <EOL> hash_key = None <EOL> hashed = None <EOL> if token_secret is not None : <EOL> hash_key = '<STR_LIT>' % ( urllib . parse . quote ( consumer_secret , safe = '<STR_LIT>' ) , <EOL> urllib . parse . quote ( token_secret , safe = '<STR_LIT>' ) ) <EOL> else : <EOL> hash_key = '<STR_LIT>' % urllib . parse . quote ( consumer_secret , safe = '<STR_LIT>' ) <EOL> try : <EOL> import hashlib <EOL> hashed = hmac . new ( hash_key , base_string , hashlib . sha1 ) <EOL> except ImportError : <EOL> import sha <EOL> hashed = hmac . new ( hash_key , base_string , sha ) <EOL> if hasattr ( base64 , '<STR_LIT>' ) : <EOL> return base64 . b64encode ( hashed . digest ( ) ) <EOL> else : <EOL> return base64 . encodestring ( hashed . digest ( ) ) . replace ( '<STR_LIT:\n>' , '<STR_LIT>' ) <EOL> def generate_rsa_signature ( http_request , consumer_key , rsa_key , <EOL> timestamp , nonce , version , next = '<STR_LIT>' , <EOL> token = None , token_secret = None , verifier = None ) : <EOL> import base64 <EOL> try : <EOL> from . tlslite . utils import keyfactory <EOL> except ImportError : <EOL> try : <EOL> from gdata . tlslite . utils import keyfactory <EOL> except ImportError : <EOL> from . tlslite . tlslite . utils import keyfactory <EOL> base_string = build_oauth_base_string ( <EOL> http_request , consumer_key , nonce , RSA_SHA1 , timestamp , version , <EOL> next , token , verifier = verifier ) <EOL> private_key = keyfactory . parsePrivateKey ( rsa_key ) <EOL> signed = private_key . hashAndSign ( base_string ) <EOL> if hasattr ( base64 , '<STR_LIT>' ) : <EOL> return base64 . b64encode ( signed ) <EOL> else : <EOL> return base64 . encodestring ( signed ) . replace ( '<STR_LIT:\n>' , '<STR_LIT>' ) <EOL> def generate_auth_header ( consumer_key , timestamp , nonce , signature_type , <EOL> signature , version = '<STR_LIT:1.0>' , next = None , token = None , <EOL> verifier = None ) : <EOL> """<STR_LIT>""" <EOL> params = { <EOL> '<STR_LIT>' : consumer_key , <EOL> '<STR_LIT>' : version , <EOL> '<STR_LIT>' : nonce , <EOL> '<STR_LIT>' : str ( timestamp ) , <EOL> '<STR_LIT>' : signature_type , <EOL> '<STR_LIT>' : signature } <EOL> if next is not None : <EOL> params [ '<STR_LIT>' ] = str ( next ) <EOL> if token is not None : <EOL> params [ '<STR_LIT>' ] = token <EOL> if verifier is not None : <EOL> params [ '<STR_LIT>' ] = verifier <EOL> pairs = [ <EOL> '<STR_LIT>' % ( <EOL> k , urllib . parse . quote ( v , safe = '<STR_LIT>' ) ) for k , v in params . items ( ) ] <EOL> return '<STR_LIT>' % ( '<STR_LIT:U+002CU+0020>' . join ( pairs ) ) <EOL> REQUEST_TOKEN_URL = '<STR_LIT>' <EOL> ACCESS_TOKEN_URL = '<STR_LIT>' <EOL> def generate_request_for_request_token ( <EOL> consumer_key , signature_type , scopes , rsa_key = None , consumer_secret = None , <EOL> auth_server_url = REQUEST_TOKEN_URL , next = '<STR_LIT>' , version = '<STR_LIT:1.0>' ) : <EOL> """<STR_LIT>""" <EOL> request = atom . http_core . HttpRequest ( auth_server_url , '<STR_LIT:POST>' ) <EOL> if scopes : <EOL> request . uri . query [ '<STR_LIT>' ] = '<STR_LIT:U+0020>' . join ( scopes ) <EOL> timestamp = str ( int ( time . time ( ) ) ) <EOL> nonce = '<STR_LIT>' . join ( [ str ( random . randint ( <NUM_LIT:0> , <NUM_LIT:9> ) ) for i in range ( <NUM_LIT:15> ) ] ) <EOL> signature = None <EOL> if signature_type == HMAC_SHA1 : <EOL> signature = generate_hmac_signature ( <EOL> request , consumer_key , consumer_secret , timestamp , nonce , version , <EOL> next = next ) <EOL> elif signature_type == RSA_SHA1 : <EOL> signature = generate_rsa_signature ( <EOL> request , consumer_key , rsa_key , timestamp , nonce , version , next = next ) <EOL> else : <EOL> return None <EOL> request . headers [ '<STR_LIT>' ] = generate_auth_header ( <EOL> consumer_key , timestamp , nonce , signature_type , signature , version , <EOL> next ) <EOL> request . headers [ '<STR_LIT>' ] = '<STR_LIT:0>' <EOL> return request <EOL> def generate_request_for_access_token ( <EOL> request_token , auth_server_url = ACCESS_TOKEN_URL ) : <EOL> """<STR_LIT>""" <EOL> http_request = atom . http_core . HttpRequest ( auth_server_url , '<STR_LIT:POST>' ) <EOL> http_request . headers [ '<STR_LIT>' ] = '<STR_LIT:0>' <EOL> return request_token . modify_request ( http_request ) <EOL> def oauth_token_info_from_body ( http_body ) : <EOL> """<STR_LIT>""" <EOL> token = None <EOL> token_secret = None <EOL> for pair in http_body . split ( '<STR_LIT:&>' ) : <EOL> if pair . startswith ( '<STR_LIT>' ) : <EOL> token = urllib . parse . unquote ( pair [ len ( '<STR_LIT>' ) : ] ) <EOL> if pair . startswith ( '<STR_LIT>' ) : <EOL> token_secret = urllib . parse . unquote ( pair [ len ( '<STR_LIT>' ) : ] ) <EOL> return ( token , token_secret ) <EOL> def hmac_token_from_body ( http_body , consumer_key , consumer_secret , <EOL> auth_state ) : <EOL> token_value , token_secret = oauth_token_info_from_body ( http_body ) <EOL> token = OAuthHmacToken ( consumer_key , consumer_secret , token_value , <EOL> token_secret , auth_state ) <EOL> return token <EOL> def rsa_token_from_body ( http_body , consumer_key , rsa_private_key , <EOL> auth_state ) : <EOL> token_value , token_secret = oauth_token_info_from_body ( http_body ) <EOL> token = OAuthRsaToken ( consumer_key , rsa_private_key , token_value , <EOL> token_secret , auth_state ) <EOL> return token <EOL> DEFAULT_DOMAIN = '<STR_LIT:default>' <EOL> OAUTH_AUTHORIZE_URL = '<STR_LIT>' <EOL> def generate_oauth_authorization_url ( <EOL> token , next = None , hd = DEFAULT_DOMAIN , hl = None , btmpl = None , <EOL> auth_server = OAUTH_AUTHORIZE_URL ) : <EOL> """<STR_LIT>""" <EOL> uri = atom . http_core . Uri . parse_uri ( auth_server ) <EOL> uri . query [ '<STR_LIT>' ] = token <EOL> uri . query [ '<STR_LIT>' ] = hd <EOL> if next is not None : <EOL> uri . query [ '<STR_LIT>' ] = str ( next ) <EOL> if hl is not None : <EOL> uri . query [ '<STR_LIT>' ] = hl <EOL> if btmpl is not None : <EOL> uri . query [ '<STR_LIT>' ] = btmpl <EOL> return uri <EOL> def oauth_token_info_from_url ( url ) : <EOL> """<STR_LIT>""" <EOL> if isinstance ( url , str ) : <EOL> url = atom . http_core . Uri . parse_uri ( url ) <EOL> token = None <EOL> verifier = None <EOL> if '<STR_LIT>' in url . query : <EOL> token = urllib . parse . unquote ( url . query [ '<STR_LIT>' ] ) <EOL> if '<STR_LIT>' in url . query : <EOL> verifier = urllib . parse . unquote ( url . query [ '<STR_LIT>' ] ) <EOL> return ( token , verifier ) <EOL> def authorize_request_token ( request_token , url ) : <EOL> """<STR_LIT>""" <EOL> token , verifier = oauth_token_info_from_url ( url ) <EOL> request_token . token = token <EOL> request_token . verifier = verifier <EOL> request_token . auth_state = AUTHORIZED_REQUEST_TOKEN <EOL> return request_token <EOL> AuthorizeRequestToken = authorize_request_token <EOL> def upgrade_to_access_token ( request_token , server_response_body ) : <EOL> """<STR_LIT>""" <EOL> token , token_secret = oauth_token_info_from_body ( server_response_body ) <EOL> request_token . token = token <EOL> request_token . token_secret = token_secret <EOL> request_token . auth_state = ACCESS_TOKEN <EOL> request_token . next = None <EOL> request_token . verifier = None <EOL> return request_token <EOL> UpgradeToAccessToken = upgrade_to_access_token <EOL> REQUEST_TOKEN = <NUM_LIT:1> <EOL> AUTHORIZED_REQUEST_TOKEN = <NUM_LIT:2> <EOL> ACCESS_TOKEN = <NUM_LIT:3> <EOL> class OAuthHmacToken ( object ) : <EOL> SIGNATURE_METHOD = HMAC_SHA1 <EOL> def __init__ ( self , consumer_key , consumer_secret , token , token_secret , <EOL> auth_state , next = None , verifier = None ) : <EOL> self . consumer_key = consumer_key <EOL> self . consumer_secret = consumer_secret <EOL> self . token = token <EOL> self . token_secret = token_secret <EOL> self . auth_state = auth_state <EOL> self . next = next <EOL> self . verifier = verifier <EOL> def generate_authorization_url ( <EOL> self , google_apps_domain = DEFAULT_DOMAIN , language = None , btmpl = None , <EOL> auth_server = OAUTH_AUTHORIZE_URL ) : <EOL> """<STR_LIT>""" <EOL> return generate_oauth_authorization_url ( <EOL> self . token , hd = google_apps_domain , hl = language , btmpl = btmpl , <EOL> auth_server = auth_server ) <EOL> GenerateAuthorizationUrl = generate_authorization_url <EOL> def modify_request ( self , http_request ) : <EOL> """<STR_LIT>""" <EOL> timestamp = str ( int ( time . time ( ) ) ) <EOL> nonce = '<STR_LIT>' . join ( [ str ( random . randint ( <NUM_LIT:0> , <NUM_LIT:9> ) ) for i in range ( <NUM_LIT:15> ) ] ) <EOL> signature = generate_hmac_signature ( <EOL> http_request , self . consumer_key , self . consumer_secret , timestamp , <EOL> nonce , version = '<STR_LIT:1.0>' , next = self . __next__ , token = self . token , <EOL> token_secret = self . token_secret , verifier = self . verifier ) <EOL> http_request . headers [ '<STR_LIT>' ] = generate_auth_header ( <EOL> self . consumer_key , timestamp , nonce , HMAC_SHA1 , signature , <EOL> version = '<STR_LIT:1.0>' , next = self . __next__ , token = self . token , <EOL> verifier = self . verifier ) <EOL> return http_request <EOL> ModifyRequest = modify_request <EOL> class OAuthRsaToken ( OAuthHmacToken ) : <EOL> SIGNATURE_METHOD = RSA_SHA1 <EOL> def __init__ ( self , consumer_key , rsa_private_key , token , token_secret , <EOL> auth_state , next = None , verifier = None ) : <EOL> self . consumer_key = consumer_key <EOL> self . rsa_private_key = rsa_private_key <EOL> self . token = token <EOL> self . token_secret = token_secret <EOL> self . auth_state = auth_state <EOL> self . next = next <EOL> self . verifier = verifier <EOL> def modify_request ( self , http_request ) : <EOL> """<STR_LIT>""" <EOL> timestamp = str ( int ( time . time ( ) ) ) <EOL> nonce = '<STR_LIT>' . join ( [ str ( random . randint ( <NUM_LIT:0> , <NUM_LIT:9> ) ) for i in range ( <NUM_LIT:15> ) ] ) <EOL> signature = generate_rsa_signature ( <EOL> http_request , self . consumer_key , self . rsa_private_key , timestamp , <EOL> nonce , version = '<STR_LIT:1.0>' , next = self . __next__ , token = self . token , <EOL> token_secret = self . token_secret , verifier = self . verifier ) <EOL> http_request . headers [ '<STR_LIT>' ] = generate_auth_header ( <EOL> self . consumer_key , timestamp , nonce , RSA_SHA1 , signature , <EOL> version = '<STR_LIT:1.0>' , next = self . __next__ , token = self . token , <EOL> verifier = self . verifier ) <EOL> return http_request <EOL> ModifyRequest = modify_request <EOL> class TwoLeggedOAuthHmacToken ( OAuthHmacToken ) : <EOL> def __init__ ( self , consumer_key , consumer_secret , requestor_id ) : <EOL> self . requestor_id = requestor_id <EOL> OAuthHmacToken . __init__ ( <EOL> self , consumer_key , consumer_secret , None , None , ACCESS_TOKEN , <EOL> next = None , verifier = None ) <EOL> def modify_request ( self , http_request ) : <EOL> """<STR_LIT>""" <EOL> http_request . uri . query [ '<STR_LIT>' ] = self . requestor_id <EOL> return OAuthHmacToken . modify_request ( self , http_request ) <EOL> ModifyRequest = modify_request <EOL> class TwoLeggedOAuthRsaToken ( OAuthRsaToken ) : <EOL> def __init__ ( self , consumer_key , rsa_private_key , requestor_id ) : <EOL> self . requestor_id = requestor_id <EOL> OAuthRsaToken . __init__ ( <EOL> self , consumer_key , rsa_private_key , None , None , ACCESS_TOKEN , <EOL> next = None , verifier = None ) <EOL> def modify_request ( self , http_request ) : <EOL> """<STR_LIT>""" <EOL> http_request . uri . query [ '<STR_LIT>' ] = self . requestor_id <EOL> return OAuthRsaToken . modify_request ( self , http_request ) <EOL> ModifyRequest = modify_request <EOL> class OAuth2Token ( object ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , client_id , client_secret , scope , user_agent , <EOL> auth_uri = '<STR_LIT>' , <EOL> token_uri = '<STR_LIT>' , <EOL> access_token = None , refresh_token = None ) : <EOL> """<STR_LIT>""" <EOL> self . client_id = client_id <EOL> self . client_secret = client_secret <EOL> self . scope = scope <EOL> self . user_agent = user_agent <EOL> self . auth_uri = auth_uri <EOL> self . token_uri = token_uri <EOL> self . access_token = access_token <EOL> self . refresh_token = refresh_token <EOL> self . _invalid = False <EOL> @ property <EOL> def invalid ( self ) : <EOL> """<STR_LIT>""" <EOL> return getattr ( self , '<STR_LIT>' , False ) <EOL> def _refresh ( self , request ) : <EOL> """<STR_LIT>""" <EOL> body = urllib . parse . urlencode ( { <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : self . client_id , <EOL> '<STR_LIT>' : self . client_secret , <EOL> '<STR_LIT>' : self . refresh_token <EOL> } ) <EOL> headers = { <EOL> '<STR_LIT>' : self . user_agent , <EOL> } <EOL> http_request = atom . http_core . HttpRequest ( <EOL> uri = self . token_uri , method = '<STR_LIT:POST>' , headers = headers ) <EOL> http_request . add_body_part ( <EOL> body , mime_type = '<STR_LIT>' ) <EOL> response = request ( http_request ) <EOL> body = response . read ( ) <EOL> if response . status == <NUM_LIT:200> : <EOL> self . _extract_tokens ( body ) <EOL> else : <EOL> self . _invalid = True <EOL> return response <EOL> def _extract_tokens ( self , body ) : <EOL> d = simplejson . loads ( body ) <EOL> self . access_token = d [ '<STR_LIT>' ] <EOL> self . refresh_token = d . get ( '<STR_LIT>' , self . refresh_token ) <EOL> if '<STR_LIT>' in d : <EOL> self . token_expiry = datetime . timedelta ( <EOL> seconds = int ( d [ '<STR_LIT>' ] ) ) + datetime . datetime . now ( ) <EOL> else : <EOL> self . token_expiry = None <EOL> def generate_authorize_url ( self , redirect_uri = '<STR_LIT>' , response_type = '<STR_LIT:code>' , <EOL> access_type = '<STR_LIT>' , ** kwargs ) : <EOL> """<STR_LIT>""" <EOL> self . redirect_uri = redirect_uri <EOL> query = { <EOL> '<STR_LIT>' : response_type , <EOL> '<STR_LIT>' : self . client_id , <EOL> '<STR_LIT>' : redirect_uri , <EOL> '<STR_LIT>' : self . scope , <EOL> '<STR_LIT>' : access_type <EOL> } <EOL> query . update ( kwargs ) <EOL> parts = list ( urllib . parse . urlparse ( self . auth_uri ) ) <EOL> query . update ( dict ( parse_qsl ( parts [ <NUM_LIT:4> ] ) ) ) <EOL> parts [ <NUM_LIT:4> ] = urllib . parse . urlencode ( query ) <EOL> return urllib . parse . urlunparse ( parts ) <EOL> def get_access_token ( self , code ) : <EOL> """<STR_LIT>""" <EOL> if not ( isinstance ( code , str ) or isinstance ( code , str ) ) : <EOL> code = code [ '<STR_LIT:code>' ] <EOL> body = urllib . parse . urlencode ( { <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : self . client_id , <EOL> '<STR_LIT>' : self . client_secret , <EOL> '<STR_LIT:code>' : code , <EOL> '<STR_LIT>' : self . redirect_uri , <EOL> '<STR_LIT>' : self . scope <EOL> } ) <EOL> headers = { <EOL> '<STR_LIT>' : self . user_agent , <EOL> } <EOL> http_client = atom . http_core . HttpClient ( ) <EOL> http_request = atom . http_core . HttpRequest ( uri = self . token_uri , method = '<STR_LIT:POST>' , <EOL> headers = headers ) <EOL> http_request . add_body_part ( data = body , mime_type = '<STR_LIT>' ) <EOL> response = http_client . request ( http_request ) <EOL> body = response . read ( ) <EOL> if response . status == <NUM_LIT:200> : <EOL> self . _extract_tokens ( body ) <EOL> return self <EOL> else : <EOL> error_msg = '<STR_LIT>' % response . status <EOL> try : <EOL> d = simplejson . loads ( body ) <EOL> if '<STR_LIT:error>' in d : <EOL> error_msg = d [ '<STR_LIT:error>' ] <EOL> except : <EOL> pass <EOL> raise OAuth2AccessTokenError ( error_msg ) <EOL> def authorize ( self , client ) : <EOL> """<STR_LIT>""" <EOL> client . auth_token = self <EOL> request_orig = client . http_client . request <EOL> def new_request ( http_request ) : <EOL> response = request_orig ( http_request ) <EOL> if response . status == <NUM_LIT> : <EOL> refresh_response = self . _refresh ( request_orig ) <EOL> if self . _invalid : <EOL> return refresh_response <EOL> else : <EOL> self . modify_request ( http_request ) <EOL> return request_orig ( http_request ) <EOL> else : <EOL> return response <EOL> client . http_client . request = new_request <EOL> return client <EOL> def modify_request ( self , http_request ) : <EOL> """<STR_LIT>""" <EOL> http_request . headers [ '<STR_LIT>' ] = '<STR_LIT>' % ( OAUTH2_AUTH_LABEL , self . access_token ) <EOL> return http_request <EOL> ModifyRequest = modify_request <EOL> def _join_token_parts ( * args ) : <EOL> """<STR_LIT>""" <EOL> return '<STR_LIT:|>' . join ( [ urllib . parse . quote_plus ( a or '<STR_LIT>' ) for a in args ] ) <EOL> def _split_token_parts ( blob ) : <EOL> """<STR_LIT>""" <EOL> return [ urllib . parse . unquote_plus ( part ) or None for part in blob . split ( '<STR_LIT:|>' ) ] <EOL> def token_to_blob ( token ) : <EOL> """<STR_LIT>""" <EOL> if isinstance ( token , ClientLoginToken ) : <EOL> return _join_token_parts ( '<STR_LIT>' , token . token_string ) <EOL> elif isinstance ( token , SecureAuthSubToken ) : <EOL> return _join_token_parts ( '<STR_LIT>' , token . token_string , token . rsa_private_key , <EOL> * token . scopes ) <EOL> elif isinstance ( token , AuthSubToken ) : <EOL> return _join_token_parts ( '<STR_LIT>' , token . token_string , * token . scopes ) <EOL> elif isinstance ( token , TwoLeggedOAuthRsaToken ) : <EOL> return _join_token_parts ( <EOL> '<STR_LIT>' , token . consumer_key , token . rsa_private_key , token . requestor_id ) <EOL> elif isinstance ( token , TwoLeggedOAuthHmacToken ) : <EOL> return _join_token_parts ( <EOL> '<STR_LIT>' , token . consumer_key , token . consumer_secret , token . requestor_id ) <EOL> elif isinstance ( token , OAuthRsaToken ) : <EOL> return _join_token_parts ( <EOL> '<STR_LIT>' , token . consumer_key , token . rsa_private_key , token . token , <EOL> token . token_secret , str ( token . auth_state ) , token . __next__ , <EOL> token . verifier ) <EOL> elif isinstance ( token , OAuthHmacToken ) : <EOL> return _join_token_parts ( <EOL> '<STR_LIT>' , token . consumer_key , token . consumer_secret , token . token , <EOL> token . token_secret , str ( token . auth_state ) , token . __next__ , <EOL> token . verifier ) <EOL> elif isinstance ( token , OAuth2Token ) : <EOL> return _join_token_parts ( <EOL> '<STR_LIT>' , token . client_id , token . client_secret , token . scope , <EOL> token . user_agent , token . auth_uri , token . token_uri , <EOL> token . access_token , token . refresh_token ) <EOL> else : <EOL> raise UnsupportedTokenType ( <EOL> '<STR_LIT>' % type ( token ) ) <EOL> TokenToBlob = token_to_blob <EOL> def token_from_blob ( blob ) : <EOL> """<STR_LIT>""" <EOL> parts = _split_token_parts ( blob ) <EOL> if parts [ <NUM_LIT:0> ] == '<STR_LIT>' : <EOL> return ClientLoginToken ( parts [ <NUM_LIT:1> ] ) <EOL> elif parts [ <NUM_LIT:0> ] == '<STR_LIT>' : <EOL> return AuthSubToken ( parts [ <NUM_LIT:1> ] , parts [ <NUM_LIT:2> : ] ) <EOL> elif parts [ <NUM_LIT:0> ] == '<STR_LIT>' : <EOL> return SecureAuthSubToken ( parts [ <NUM_LIT:1> ] , parts [ <NUM_LIT:2> ] , parts [ <NUM_LIT:3> : ] ) <EOL> elif parts [ <NUM_LIT:0> ] == '<STR_LIT>' : <EOL> return TwoLeggedOAuthRsaToken ( parts [ <NUM_LIT:1> ] , parts [ <NUM_LIT:2> ] , parts [ <NUM_LIT:3> ] ) <EOL> elif parts [ <NUM_LIT:0> ] == '<STR_LIT>' : <EOL> return TwoLeggedOAuthHmacToken ( parts [ <NUM_LIT:1> ] , parts [ <NUM_LIT:2> ] , parts [ <NUM_LIT:3> ] ) <EOL> elif parts [ <NUM_LIT:0> ] == '<STR_LIT>' : <EOL> auth_state = int ( parts [ <NUM_LIT:5> ] ) <EOL> return OAuthRsaToken ( parts [ <NUM_LIT:1> ] , parts [ <NUM_LIT:2> ] , parts [ <NUM_LIT:3> ] , parts [ <NUM_LIT:4> ] , auth_state , <EOL> parts [ <NUM_LIT:6> ] , parts [ <NUM_LIT:7> ] ) <EOL> elif parts [ <NUM_LIT:0> ] == '<STR_LIT>' : <EOL> auth_state = int ( parts [ <NUM_LIT:5> ] ) <EOL> return OAuthHmacToken ( parts [ <NUM_LIT:1> ] , parts [ <NUM_LIT:2> ] , parts [ <NUM_LIT:3> ] , parts [ <NUM_LIT:4> ] , auth_state , <EOL> parts [ <NUM_LIT:6> ] , parts [ <NUM_LIT:7> ] ) <EOL> elif parts [ <NUM_LIT:0> ] == '<STR_LIT>' : <EOL> return OAuth2Token ( parts [ <NUM_LIT:1> ] , parts [ <NUM_LIT:2> ] , parts [ <NUM_LIT:3> ] , parts [ <NUM_LIT:4> ] , parts [ <NUM_LIT:5> ] , <EOL> parts [ <NUM_LIT:6> ] , parts [ <NUM_LIT:7> ] , parts [ <NUM_LIT:8> ] ) <EOL> else : <EOL> raise UnsupportedTokenType ( <EOL> '<STR_LIT>' % parts [ <NUM_LIT:0> ] ) <EOL> TokenFromBlob = token_from_blob <EOL> def dump_tokens ( tokens ) : <EOL> return '<STR_LIT:U+002C>' . join ( [ token_to_blob ( t ) for t in tokens ] ) <EOL> def load_tokens ( blob ) : <EOL> return [ token_from_blob ( s ) for s in blob . split ( '<STR_LIT:U+002C>' ) ] <EOL> def find_scopes_for_services ( service_names = None ) : <EOL> """<STR_LIT>""" <EOL> result_scopes = [ ] <EOL> if service_names is None : <EOL> for service_name , scopes in AUTH_SCOPES . items ( ) : <EOL> result_scopes . extend ( scopes ) <EOL> else : <EOL> for service_name in service_names : <EOL> result_scopes . extend ( AUTH_SCOPES [ service_name ] ) <EOL> return result_scopes <EOL> FindScopesForServices = find_scopes_for_services <EOL> def ae_save ( token , token_key ) : <EOL> """<STR_LIT>""" <EOL> import gdata . alt . app_engine <EOL> key_name = '<STR_LIT>' . join ( ( '<STR_LIT>' , token_key ) ) <EOL> return gdata . alt . app_engine . set_token ( key_name , token_to_blob ( token ) ) <EOL> AeSave = ae_save <EOL> def ae_load ( token_key ) : <EOL> """<STR_LIT>""" <EOL> import gdata . alt . app_engine <EOL> key_name = '<STR_LIT>' . join ( ( '<STR_LIT>' , token_key ) ) <EOL> token_string = gdata . alt . app_engine . get_token ( key_name ) <EOL> if token_string is not None : <EOL> return token_from_blob ( token_string ) <EOL> else : <EOL> return None <EOL> AeLoad = ae_load <EOL> def ae_delete ( token_key ) : <EOL> """<STR_LIT>""" <EOL> import gdata . alt . app_engine <EOL> key_name = '<STR_LIT>' . join ( ( '<STR_LIT>' , token_key ) ) <EOL> gdata . alt . app_engine . delete_token ( key_name ) <EOL> AeDelete = ae_delete </s>
