<s> """<STR_LIT>""" <EOL> import base64 <EOL> import urllib . request , urllib . parse , urllib . error <EOL> import time <EOL> import random <EOL> import urllib . parse <EOL> import hmac <EOL> import binascii <EOL> import httplib2 <EOL> try : <EOL> from urllib . parse import parse_qs <EOL> parse_qs <EOL> except ImportError : <EOL> from cgi import parse_qs <EOL> try : <EOL> from hashlib import sha1 <EOL> sha = sha1 <EOL> except ImportError : <EOL> import sha <EOL> from . import _version <EOL> __version__ = _version . __version__ <EOL> OAUTH_VERSION = '<STR_LIT:1.0>' <EOL> HTTP_METHOD = '<STR_LIT:GET>' <EOL> SIGNATURE_METHOD = '<STR_LIT>' <EOL> class Error ( RuntimeError ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , message = '<STR_LIT>' ) : <EOL> self . _message = message <EOL> @ property <EOL> def message ( self ) : <EOL> """<STR_LIT>""" <EOL> return self . _message <EOL> def __str__ ( self ) : <EOL> return self . _message <EOL> class MissingSignature ( Error ) : <EOL> pass <EOL> def build_authenticate_header ( realm = '<STR_LIT>' ) : <EOL> """<STR_LIT>""" <EOL> return { '<STR_LIT>' : '<STR_LIT>' % realm } <EOL> def build_xoauth_string ( url , consumer , token = None ) : <EOL> """<STR_LIT>""" <EOL> request = Request . from_consumer_and_token ( consumer , token , <EOL> "<STR_LIT:GET>" , url ) <EOL> signing_method = SignatureMethod_HMAC_SHA1 ( ) <EOL> request . sign_request ( signing_method , consumer , token ) <EOL> params = [ ] <EOL> for k , v in sorted ( request . items ( ) ) : <EOL> if v is not None : <EOL> params . append ( '<STR_LIT>' % ( k , escape ( v ) ) ) <EOL> return "<STR_LIT>" % ( "<STR_LIT:GET>" , url , '<STR_LIT:U+002C>' . join ( params ) ) <EOL> def to_unicode ( s ) : <EOL> """<STR_LIT>""" <EOL> if not isinstance ( s , str ) : <EOL> if not isinstance ( s , str ) : <EOL> raise TypeError ( '<STR_LIT>' % ( type ( s ) , s ) ) <EOL> try : <EOL> s = s . decode ( '<STR_LIT:utf-8>' ) <EOL> except UnicodeDecodeError as le : <EOL> raise TypeError ( '<STR_LIT>' % ( s , le , ) ) <EOL> return s <EOL> def to_utf8 ( s ) : <EOL> return to_unicode ( s ) . encode ( '<STR_LIT:utf-8>' ) <EOL> def to_unicode_if_string ( s ) : <EOL> if isinstance ( s , str ) : <EOL> return to_unicode ( s ) <EOL> else : <EOL> return s <EOL> def to_utf8_if_string ( s ) : <EOL> if isinstance ( s , str ) : <EOL> return to_utf8 ( s ) <EOL> else : <EOL> return s <EOL> def to_unicode_optional_iterator ( x ) : <EOL> """<STR_LIT>""" <EOL> if isinstance ( x , str ) : <EOL> return to_unicode ( x ) <EOL> try : <EOL> l = list ( x ) <EOL> except TypeError as e : <EOL> assert '<STR_LIT>' in str ( e ) <EOL> return x <EOL> else : <EOL> return [ to_unicode ( e ) for e in l ] <EOL> def to_utf8_optional_iterator ( x ) : <EOL> """<STR_LIT>""" <EOL> if isinstance ( x , str ) : <EOL> return to_utf8 ( x ) <EOL> try : <EOL> l = list ( x ) <EOL> except TypeError as e : <EOL> assert '<STR_LIT>' in str ( e ) <EOL> return x <EOL> else : <EOL> return [ to_utf8_if_string ( e ) for e in l ] <EOL> def escape ( s ) : <EOL> """<STR_LIT>""" <EOL> return urllib . parse . quote ( s . encode ( '<STR_LIT:utf-8>' ) , safe = '<STR_LIT>' ) <EOL> def generate_timestamp ( ) : <EOL> """<STR_LIT>""" <EOL> return int ( time . time ( ) ) <EOL> def generate_nonce ( length = <NUM_LIT:8> ) : <EOL> """<STR_LIT>""" <EOL> return '<STR_LIT>' . join ( [ str ( random . randint ( <NUM_LIT:0> , <NUM_LIT:9> ) ) for i in range ( length ) ] ) <EOL> def generate_verifier ( length = <NUM_LIT:8> ) : <EOL> """<STR_LIT>""" <EOL> return '<STR_LIT>' . join ( [ str ( random . randint ( <NUM_LIT:0> , <NUM_LIT:9> ) ) for i in range ( length ) ] ) <EOL> class Consumer ( object ) : <EOL> """<STR_LIT>""" <EOL> key = None <EOL> secret = None <EOL> def __init__ ( self , key , secret ) : <EOL> self . key = key <EOL> self . secret = secret <EOL> if self . key is None or self . secret is None : <EOL> raise ValueError ( "<STR_LIT>" ) <EOL> def __str__ ( self ) : <EOL> data = { '<STR_LIT>' : self . key , <EOL> '<STR_LIT>' : self . secret } <EOL> return urllib . parse . urlencode ( data ) <EOL> class Token ( object ) : <EOL> """<STR_LIT>""" <EOL> key = None <EOL> secret = None <EOL> callback = None <EOL> callback_confirmed = None <EOL> verifier = None <EOL> def __init__ ( self , key , secret ) : <EOL> self . key = key <EOL> self . secret = secret <EOL> if self . key is None or self . secret is None : <EOL> raise ValueError ( "<STR_LIT>" ) <EOL> def set_callback ( self , callback ) : <EOL> self . callback = callback <EOL> self . callback_confirmed = '<STR_LIT:true>' <EOL> def set_verifier ( self , verifier = None ) : <EOL> if verifier is not None : <EOL> self . verifier = verifier <EOL> else : <EOL> self . verifier = generate_verifier ( ) <EOL> def get_callback_url ( self ) : <EOL> if self . callback and self . verifier : <EOL> parts = urllib . parse . urlparse ( self . callback ) <EOL> scheme , netloc , path , params , query , fragment = parts [ : <NUM_LIT:6> ] <EOL> if query : <EOL> query = '<STR_LIT>' % ( query , self . verifier ) <EOL> else : <EOL> query = '<STR_LIT>' % self . verifier <EOL> return urllib . parse . urlunparse ( ( scheme , netloc , path , params , <EOL> query , fragment ) ) <EOL> return self . callback <EOL> def to_string ( self ) : <EOL> """<STR_LIT>""" <EOL> data = { <EOL> '<STR_LIT>' : self . key , <EOL> '<STR_LIT>' : self . secret , <EOL> } <EOL> if self . callback_confirmed is not None : <EOL> data [ '<STR_LIT>' ] = self . callback_confirmed <EOL> return urllib . parse . urlencode ( data ) <EOL> @ staticmethod <EOL> def from_string ( s ) : <EOL> """<STR_LIT>""" <EOL> if not len ( s ) : <EOL> raise ValueError ( "<STR_LIT>" ) <EOL> params = parse_qs ( s , keep_blank_values = False ) <EOL> if not len ( params ) : <EOL> raise ValueError ( "<STR_LIT>" ) <EOL> try : <EOL> key = params [ '<STR_LIT>' ] [ <NUM_LIT:0> ] <EOL> except Exception : <EOL> raise ValueError ( "<STR_LIT>" ) <EOL> try : <EOL> secret = params [ '<STR_LIT>' ] [ <NUM_LIT:0> ] <EOL> except Exception : <EOL> raise ValueError ( "<STR_LIT>" <EOL> "<STR_LIT>" ) <EOL> token = Token ( key , secret ) <EOL> try : <EOL> token . callback_confirmed = params [ '<STR_LIT>' ] [ <NUM_LIT:0> ] <EOL> except KeyError : <EOL> pass <EOL> return token <EOL> def __str__ ( self ) : <EOL> return self . to_string ( ) <EOL> def setter ( attr ) : <EOL> name = attr . __name__ <EOL> def getter ( self ) : <EOL> try : <EOL> return self . __dict__ [ name ] <EOL> except KeyError : <EOL> raise AttributeError ( name ) <EOL> def deleter ( self ) : <EOL> del self . __dict__ [ name ] <EOL> return property ( getter , attr , deleter ) <EOL> class Request ( dict ) : <EOL> """<STR_LIT>""" <EOL> version = OAUTH_VERSION <EOL> def __init__ ( self , method = HTTP_METHOD , url = None , parameters = None , <EOL> body = '<STR_LIT>' , is_form_encoded = False ) : <EOL> if url is not None : <EOL> self . url = to_unicode ( url ) <EOL> self . method = method <EOL> if parameters is not None : <EOL> for k , v in list ( parameters . items ( ) ) : <EOL> k = to_unicode ( k ) <EOL> v = to_unicode_optional_iterator ( v ) <EOL> self [ k ] = v <EOL> self . body = body <EOL> self . is_form_encoded = is_form_encoded <EOL> @ setter <EOL> def url ( self , value ) : <EOL> self . __dict__ [ '<STR_LIT:url>' ] = value <EOL> if value is not None : <EOL> scheme , netloc , path , params , query , fragment = urllib . parse . urlparse ( value ) <EOL> if scheme == '<STR_LIT:http>' and netloc [ - <NUM_LIT:3> : ] == '<STR_LIT>' : <EOL> netloc = netloc [ : - <NUM_LIT:3> ] <EOL> elif scheme == '<STR_LIT>' and netloc [ - <NUM_LIT:4> : ] == '<STR_LIT>' : <EOL> netloc = netloc [ : - <NUM_LIT:4> ] <EOL> if scheme not in ( '<STR_LIT:http>' , '<STR_LIT>' ) : <EOL> raise ValueError ( "<STR_LIT>" % ( value , scheme ) ) <EOL> self . normalized_url = urllib . parse . urlunparse ( ( scheme , netloc , path , None , None , None ) ) <EOL> else : <EOL> self . normalized_url = None <EOL> self . __dict__ [ '<STR_LIT:url>' ] = None <EOL> @ setter <EOL> def method ( self , value ) : <EOL> self . __dict__ [ '<STR_LIT>' ] = value . upper ( ) <EOL> def _get_timestamp_nonce ( self ) : <EOL> return self [ '<STR_LIT>' ] , self [ '<STR_LIT>' ] <EOL> def get_nonoauth_parameters ( self ) : <EOL> """<STR_LIT>""" <EOL> return dict ( [ ( k , v ) for k , v in list ( self . items ( ) ) <EOL> if not k . startswith ( '<STR_LIT>' ) ] ) <EOL> def to_header ( self , realm = '<STR_LIT>' ) : <EOL> """<STR_LIT>""" <EOL> oauth_params = ( ( k , v ) for k , v in list ( self . items ( ) ) <EOL> if k . startswith ( '<STR_LIT>' ) ) <EOL> stringy_params = ( ( k , escape ( str ( v ) ) ) for k , v in oauth_params ) <EOL> header_params = ( '<STR_LIT>' % ( k , v ) for k , v in stringy_params ) <EOL> params_header = '<STR_LIT:U+002CU+0020>' . join ( header_params ) <EOL> auth_header = '<STR_LIT>' % realm <EOL> if params_header : <EOL> auth_header = "<STR_LIT>" % ( auth_header , params_header ) <EOL> return { '<STR_LIT>' : auth_header } <EOL> def to_postdata ( self ) : <EOL> """<STR_LIT>""" <EOL> d = { } <EOL> for k , v in list ( self . items ( ) ) : <EOL> d [ k . encode ( '<STR_LIT:utf-8>' ) ] = to_utf8_optional_iterator ( v ) <EOL> return urllib . parse . urlencode ( d , True ) . replace ( '<STR_LIT:+>' , '<STR_LIT>' ) <EOL> def to_url ( self ) : <EOL> """<STR_LIT>""" <EOL> base_url = urllib . parse . urlparse ( self . url ) <EOL> try : <EOL> query = base_url . query <EOL> except AttributeError : <EOL> query = base_url [ <NUM_LIT:4> ] <EOL> query = parse_qs ( query ) <EOL> for k , v in list ( self . items ( ) ) : <EOL> query . setdefault ( k , [ ] ) . append ( v ) <EOL> try : <EOL> scheme = base_url . scheme <EOL> netloc = base_url . netloc <EOL> path = base_url . path <EOL> params = base_url . params <EOL> fragment = base_url . fragment <EOL> except AttributeError : <EOL> scheme = base_url [ <NUM_LIT:0> ] <EOL> netloc = base_url [ <NUM_LIT:1> ] <EOL> path = base_url [ <NUM_LIT:2> ] <EOL> params = base_url [ <NUM_LIT:3> ] <EOL> fragment = base_url [ <NUM_LIT:5> ] <EOL> url = ( scheme , netloc , path , params , <EOL> urllib . parse . urlencode ( query , True ) , fragment ) <EOL> return urllib . parse . urlunparse ( url ) <EOL> def get_parameter ( self , parameter ) : <EOL> ret = self . get ( parameter ) <EOL> if ret is None : <EOL> raise Error ( '<STR_LIT>' % parameter ) <EOL> return ret <EOL> def get_normalized_parameters ( self ) : <EOL> """<STR_LIT>""" <EOL> items = [ ] <EOL> for key , value in list ( self . items ( ) ) : <EOL> if key == '<STR_LIT>' : <EOL> continue <EOL> if isinstance ( value , str ) : <EOL> items . append ( ( to_utf8_if_string ( key ) , to_utf8 ( value ) ) ) <EOL> else : <EOL> try : <EOL> value = list ( value ) <EOL> except TypeError as e : <EOL> assert '<STR_LIT>' in str ( e ) <EOL> items . append ( ( to_utf8_if_string ( key ) , to_utf8_if_string ( value ) ) ) <EOL> else : <EOL> items . extend ( ( to_utf8_if_string ( key ) , to_utf8_if_string ( item ) ) for item in value ) <EOL> query = urllib . parse . urlparse ( self . url ) [ <NUM_LIT:4> ] <EOL> url_items = list ( self . _split_url_string ( query ) . items ( ) ) <EOL> url_items = [ ( to_utf8 ( k ) , to_utf8 ( v ) ) for k , v in url_items if k != '<STR_LIT>' ] <EOL> items . extend ( url_items ) <EOL> items . sort ( ) <EOL> encoded_str = urllib . parse . urlencode ( items ) <EOL> return encoded_str . replace ( '<STR_LIT:+>' , '<STR_LIT>' ) . replace ( '<STR_LIT>' , '<STR_LIT>' ) <EOL> def sign_request ( self , signature_method , consumer , token ) : <EOL> """<STR_LIT>""" <EOL> if not self . is_form_encoded : <EOL> self [ '<STR_LIT>' ] = base64 . b64encode ( sha ( self . body ) . digest ( ) ) <EOL> if '<STR_LIT>' not in self : <EOL> self [ '<STR_LIT>' ] = consumer . key <EOL> if token and '<STR_LIT>' not in self : <EOL> self [ '<STR_LIT>' ] = token . key <EOL> self [ '<STR_LIT>' ] = signature_method . name <EOL> self [ '<STR_LIT>' ] = signature_method . sign ( self , consumer , token ) <EOL> @ classmethod <EOL> def make_timestamp ( cls ) : <EOL> """<STR_LIT>""" <EOL> return str ( int ( time . time ( ) ) ) <EOL> @ classmethod <EOL> def make_nonce ( cls ) : <EOL> """<STR_LIT>""" <EOL> return str ( random . randint ( <NUM_LIT:0> , <NUM_LIT> ) ) <EOL> @ classmethod <EOL> def from_request ( cls , http_method , http_url , headers = None , parameters = None , <EOL> query_string = None ) : <EOL> """<STR_LIT>""" <EOL> if parameters is None : <EOL> parameters = { } <EOL> if headers and '<STR_LIT>' in headers : <EOL> auth_header = headers [ '<STR_LIT>' ] <EOL> if auth_header [ : <NUM_LIT:6> ] == '<STR_LIT>' : <EOL> auth_header = auth_header [ <NUM_LIT:6> : ] <EOL> try : <EOL> header_params = cls . _split_header ( auth_header ) <EOL> parameters . update ( header_params ) <EOL> except : <EOL> raise Error ( '<STR_LIT>' <EOL> '<STR_LIT>' ) <EOL> if query_string : <EOL> query_params = cls . _split_url_string ( query_string ) <EOL> parameters . update ( query_params ) <EOL> param_str = urllib . parse . urlparse ( http_url ) [ <NUM_LIT:4> ] <EOL> url_params = cls . _split_url_string ( param_str ) <EOL> parameters . update ( url_params ) <EOL> if parameters : <EOL> return cls ( http_method , http_url , parameters ) <EOL> return None <EOL> @ classmethod <EOL> def from_consumer_and_token ( cls , consumer , token = None , <EOL> http_method = HTTP_METHOD , http_url = None , parameters = None , <EOL> body = '<STR_LIT>' , is_form_encoded = False ) : <EOL> if not parameters : <EOL> parameters = { } <EOL> defaults = { <EOL> '<STR_LIT>' : consumer . key , <EOL> '<STR_LIT>' : cls . make_timestamp ( ) , <EOL> '<STR_LIT>' : cls . make_nonce ( ) , <EOL> '<STR_LIT>' : cls . version , <EOL> } <EOL> defaults . update ( parameters ) <EOL> parameters = defaults <EOL> if token : <EOL> parameters [ '<STR_LIT>' ] = token . key <EOL> if token . verifier : <EOL> parameters [ '<STR_LIT>' ] = token . verifier <EOL> return Request ( http_method , http_url , parameters , body = body , <EOL> is_form_encoded = is_form_encoded ) <EOL> @ classmethod <EOL> def from_token_and_callback ( cls , token , callback = None , <EOL> http_method = HTTP_METHOD , http_url = None , parameters = None ) : <EOL> if not parameters : <EOL> parameters = { } <EOL> parameters [ '<STR_LIT>' ] = token . key <EOL> if callback : <EOL> parameters [ '<STR_LIT>' ] = callback <EOL> return cls ( http_method , http_url , parameters ) <EOL> @ staticmethod <EOL> def _split_header ( header ) : <EOL> """<STR_LIT>""" <EOL> params = { } <EOL> parts = header . split ( '<STR_LIT:U+002C>' ) <EOL> for param in parts : <EOL> if param . find ( '<STR_LIT>' ) > - <NUM_LIT:1> : <EOL> continue <EOL> param = param . strip ( ) <EOL> param_parts = param . split ( '<STR_LIT:=>' , <NUM_LIT:1> ) <EOL> params [ param_parts [ <NUM_LIT:0> ] ] = urllib . parse . unquote ( param_parts [ <NUM_LIT:1> ] . strip ( '<STR_LIT>' ) ) <EOL> return params <EOL> @ staticmethod <EOL> def _split_url_string ( param_str ) : <EOL> """<STR_LIT>""" <EOL> parameters = parse_qs ( param_str . encode ( '<STR_LIT:utf-8>' ) , keep_blank_values = True ) <EOL> for k , v in list ( parameters . items ( ) ) : <EOL> parameters [ k ] = urllib . parse . unquote ( v [ <NUM_LIT:0> ] ) <EOL> return parameters <EOL> class Client ( httplib2 . Http ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , consumer , token = None , cache = None , timeout = None , <EOL> proxy_info = None ) : <EOL> if consumer is not None and not isinstance ( consumer , Consumer ) : <EOL> raise ValueError ( "<STR_LIT>" ) <EOL> if token is not None and not isinstance ( token , Token ) : <EOL> raise ValueError ( "<STR_LIT>" ) <EOL> self . consumer = consumer <EOL> self . token = token <EOL> self . method = SignatureMethod_HMAC_SHA1 ( ) <EOL> httplib2 . Http . __init__ ( self , cache = cache , timeout = timeout , proxy_info = proxy_info ) <EOL> def set_signature_method ( self , method ) : <EOL> if not isinstance ( method , SignatureMethod ) : <EOL> raise ValueError ( "<STR_LIT>" ) <EOL> self . method = method <EOL> def request ( self , uri , method = "<STR_LIT:GET>" , body = '<STR_LIT>' , headers = None , <EOL> redirections = httplib2 . DEFAULT_MAX_REDIRECTS , connection_type = None ) : <EOL> DEFAULT_POST_CONTENT_TYPE = '<STR_LIT>' <EOL> if not isinstance ( headers , dict ) : <EOL> headers = { } <EOL> if method == "<STR_LIT:POST>" : <EOL> headers [ '<STR_LIT:Content-Type>' ] = headers . get ( '<STR_LIT:Content-Type>' , <EOL> DEFAULT_POST_CONTENT_TYPE ) <EOL> is_form_encoded = headers . get ( '<STR_LIT:Content-Type>' ) == '<STR_LIT>' <EOL> if is_form_encoded and body : <EOL> parameters = parse_qs ( body ) <EOL> else : <EOL> parameters = None <EOL> req = Request . from_consumer_and_token ( self . consumer , <EOL> token = self . token , http_method = method , http_url = uri , <EOL> parameters = parameters , body = body , is_form_encoded = is_form_encoded ) <EOL> req . sign_request ( self . method , self . consumer , self . token ) <EOL> schema , rest = urllib . parse . splittype ( uri ) <EOL> if rest . startswith ( '<STR_LIT>' ) : <EOL> hierpart = '<STR_LIT>' <EOL> else : <EOL> hierpart = '<STR_LIT>' <EOL> host , rest = urllib . parse . splithost ( rest ) <EOL> realm = schema + '<STR_LIT::>' + hierpart + host <EOL> if is_form_encoded : <EOL> body = req . to_postdata ( ) <EOL> elif method == "<STR_LIT:GET>" : <EOL> uri = req . to_url ( ) <EOL> else : <EOL> headers . update ( req . to_header ( realm = realm ) ) <EOL> return httplib2 . Http . request ( self , uri , method = method , body = body , <EOL> headers = headers , redirections = redirections , <EOL> connection_type = connection_type ) <EOL> class Server ( object ) : <EOL> """<STR_LIT>""" <EOL> timestamp_threshold = <NUM_LIT> <EOL> version = OAUTH_VERSION <EOL> signature_methods = None <EOL> def __init__ ( self , signature_methods = None ) : <EOL> self . signature_methods = signature_methods or { } <EOL> def add_signature_method ( self , signature_method ) : <EOL> self . signature_methods [ signature_method . name ] = signature_method <EOL> return self . signature_methods <EOL> def verify_request ( self , request , consumer , token ) : <EOL> """<STR_LIT>""" <EOL> self . _check_version ( request ) <EOL> self . _check_signature ( request , consumer , token ) <EOL> parameters = request . get_nonoauth_parameters ( ) <EOL> return parameters <EOL> def build_authenticate_header ( self , realm = '<STR_LIT>' ) : <EOL> """<STR_LIT>""" <EOL> return { '<STR_LIT>' : '<STR_LIT>' % realm } <EOL> def _check_version ( self , request ) : <EOL> """<STR_LIT>""" <EOL> version = self . _get_version ( request ) <EOL> if version and version != self . version : <EOL> raise Error ( '<STR_LIT>' % str ( version ) ) <EOL> def _get_version ( self , request ) : <EOL> """<STR_LIT>""" <EOL> try : <EOL> version = request . get_parameter ( '<STR_LIT>' ) <EOL> except : <EOL> version = OAUTH_VERSION <EOL> return version <EOL> def _get_signature_method ( self , request ) : <EOL> """<STR_LIT>""" <EOL> try : <EOL> signature_method = request . get_parameter ( '<STR_LIT>' ) <EOL> except : <EOL> signature_method = SIGNATURE_METHOD <EOL> try : <EOL> signature_method = self . signature_methods [ signature_method ] <EOL> except : <EOL> signature_method_names = '<STR_LIT:U+002CU+0020>' . join ( list ( self . signature_methods . keys ( ) ) ) <EOL> raise Error ( '<STR_LIT>' % ( signature_method , signature_method_names ) ) <EOL> return signature_method <EOL> def _get_verifier ( self , request ) : <EOL> return request . get_parameter ( '<STR_LIT>' ) <EOL> def _check_signature ( self , request , consumer , token ) : <EOL> timestamp , nonce = request . _get_timestamp_nonce ( ) <EOL> self . _check_timestamp ( timestamp ) <EOL> signature_method = self . _get_signature_method ( request ) <EOL> try : <EOL> signature = request . get_parameter ( '<STR_LIT>' ) <EOL> except : <EOL> raise MissingSignature ( '<STR_LIT>' ) <EOL> valid = signature_method . check ( request , consumer , token , signature ) <EOL> if not valid : <EOL> key , base = signature_method . signing_base ( request , consumer , token ) <EOL> raise Error ( '<STR_LIT>' <EOL> '<STR_LIT>' % base ) <EOL> def _check_timestamp ( self , timestamp ) : <EOL> """<STR_LIT>""" <EOL> timestamp = int ( timestamp ) <EOL> now = int ( time . time ( ) ) <EOL> lapsed = now - timestamp <EOL> if lapsed > self . timestamp_threshold : <EOL> raise Error ( '<STR_LIT>' <EOL> '<STR_LIT>' % ( timestamp , now , <EOL> self . timestamp_threshold ) ) <EOL> class SignatureMethod ( object ) : <EOL> """<STR_LIT>""" <EOL> def signing_base ( self , request , consumer , token ) : <EOL> """<STR_LIT>""" <EOL> raise NotImplementedError <EOL> def sign ( self , request , consumer , token ) : <EOL> """<STR_LIT>""" <EOL> raise NotImplementedError <EOL> def check ( self , request , consumer , token , signature ) : <EOL> """<STR_LIT>""" <EOL> built = self . sign ( request , consumer , token ) <EOL> return built == signature <EOL> class SignatureMethod_HMAC_SHA1 ( SignatureMethod ) : <EOL> name = '<STR_LIT>' <EOL> def signing_base ( self , request , consumer , token ) : <EOL> if not hasattr ( request , '<STR_LIT>' ) or request . normalized_url is None : <EOL> raise ValueError ( "<STR_LIT>" ) <EOL> sig = ( <EOL> escape ( request . method ) , <EOL> escape ( request . normalized_url ) , <EOL> escape ( request . get_normalized_parameters ( ) ) , <EOL> ) <EOL> key = '<STR_LIT>' % escape ( consumer . secret ) <EOL> if token : <EOL> key += escape ( token . secret ) <EOL> raw = '<STR_LIT:&>' . join ( sig ) <EOL> return key , raw <EOL> def sign ( self , request , consumer , token ) : <EOL> """<STR_LIT>""" <EOL> key , raw = self . signing_base ( request , consumer , token ) <EOL> hashed = hmac . new ( key , raw , sha ) <EOL> return binascii . b2a_base64 ( hashed . digest ( ) ) [ : - <NUM_LIT:1> ] <EOL> class SignatureMethod_PLAINTEXT ( SignatureMethod ) : <EOL> name = '<STR_LIT>' <EOL> def signing_base ( self , request , consumer , token ) : <EOL> """<STR_LIT>""" <EOL> sig = '<STR_LIT>' % escape ( consumer . secret ) <EOL> if token : <EOL> sig = sig + escape ( token . secret ) <EOL> return sig , sig <EOL> def sign ( self , request , consumer , token ) : <EOL> key , raw = self . signing_base ( request , consumer , token ) <EOL> return raw </s>
