<s> """<STR_LIT>""" <EOL> import cgi <EOL> import urllib <EOL> import time <EOL> import random <EOL> import urlparse <EOL> import hmac <EOL> import base64 <EOL> import logging <EOL> import hashlib <EOL> VERSION = '<STR_LIT:1.0>' <EOL> HTTP_METHOD = '<STR_LIT:GET>' <EOL> SIGNATURE_METHOD = '<STR_LIT>' <EOL> class OAuthError ( RuntimeError ) : <EOL> def __init__ ( self , message = '<STR_LIT>' ) : <EOL> self . message = message <EOL> def build_authenticate_header ( realm = '<STR_LIT>' ) : <EOL> return { '<STR_LIT>' : '<STR_LIT>' % realm } <EOL> def escape ( s ) : <EOL> return urllib . quote ( s , safe = '<STR_LIT>' ) <EOL> def generate_timestamp ( ) : <EOL> return int ( time . time ( ) ) <EOL> def generate_nonce ( length = <NUM_LIT:8> ) : <EOL> return '<STR_LIT>' . join ( str ( random . randint ( <NUM_LIT:0> , <NUM_LIT:9> ) ) for i in range ( length ) ) <EOL> class OAuthConsumer ( object ) : <EOL> key = None <EOL> secret = None <EOL> def __init__ ( self , key , secret ) : <EOL> self . key = key <EOL> self . secret = secret <EOL> class OAuthToken ( object ) : <EOL> key = None <EOL> secret = None <EOL> '''<STR_LIT>''' <EOL> def __init__ ( self , key , secret ) : <EOL> self . key = key <EOL> self . secret = secret <EOL> def to_string ( self ) : <EOL> return urllib . urlencode ( { '<STR_LIT>' : self . key , '<STR_LIT>' : self . secret } ) <EOL> @ staticmethod <EOL> def from_string ( s ) : <EOL> params = cgi . parse_qs ( s , keep_blank_values = False ) <EOL> key = params [ '<STR_LIT>' ] [ <NUM_LIT:0> ] <EOL> secret = params [ '<STR_LIT>' ] [ <NUM_LIT:0> ] <EOL> return OAuthToken ( key , secret ) <EOL> def __str__ ( self ) : <EOL> return self . to_string ( ) <EOL> class OAuthRequest ( object ) : <EOL> '''<STR_LIT>''' <EOL> parameters = None <EOL> http_method = HTTP_METHOD <EOL> http_url = None <EOL> version = VERSION <EOL> OAUTH_PARAMS = [ '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' ] <EOL> def __init__ ( self , http_method = HTTP_METHOD , http_url = None , parameters = None ) : <EOL> self . http_method = http_method <EOL> self . http_url = http_url <EOL> self . parameters = parameters or { } <EOL> def set_parameter ( self , parameter , value ) : <EOL> self . parameters [ parameter ] = value <EOL> def get_parameter ( self , parameter ) : <EOL> try : <EOL> return self . parameters [ parameter ] <EOL> except : <EOL> raise OAuthError ( '<STR_LIT>' % parameter ) <EOL> def _get_timestamp_nonce ( self ) : <EOL> return self . get_parameter ( '<STR_LIT>' ) , self . get_parameter ( '<STR_LIT>' ) <EOL> def get_nonoauth_parameters ( self ) : <EOL> parameters = { } <EOL> for k , v in self . parameters . iteritems ( ) : <EOL> if k . find ( '<STR_LIT>' ) < <NUM_LIT:0> : <EOL> parameters [ k ] = v <EOL> return parameters <EOL> def to_header ( self , realm = '<STR_LIT>' ) : <EOL> auth_header = '<STR_LIT>' % realm <EOL> if self . parameters : <EOL> for k , v in self . parameters . iteritems ( ) : <EOL> if k in self . OAUTH_PARAMS : <EOL> auth_header += '<STR_LIT>' % ( k , escape ( str ( v ) ) ) <EOL> return { '<STR_LIT>' : auth_header } <EOL> def to_postdata ( self ) : <EOL> return '<STR_LIT:&>' . join ( '<STR_LIT>' % ( escape ( str ( k ) ) , escape ( str ( v ) ) ) for k , v in self . parameters . iteritems ( ) ) <EOL> def to_url ( self ) : <EOL> return '<STR_LIT>' % ( self . get_normalized_http_url ( ) , self . to_postdata ( ) ) <EOL> def get_normalized_parameters ( self ) : <EOL> params = self . parameters <EOL> try : <EOL> del params [ '<STR_LIT>' ] <EOL> except : <EOL> pass <EOL> key_values = params . items ( ) <EOL> key_values . sort ( ) <EOL> return '<STR_LIT:&>' . join ( '<STR_LIT>' % ( escape ( str ( k ) ) , escape ( str ( v ) ) ) for k , v in key_values ) <EOL> def get_normalized_http_method ( self ) : <EOL> return self . http_method . upper ( ) <EOL> def get_normalized_http_url ( self ) : <EOL> parts = urlparse . urlparse ( self . http_url ) <EOL> url_string = '<STR_LIT>' % ( parts [ <NUM_LIT:0> ] , parts [ <NUM_LIT:1> ] , parts [ <NUM_LIT:2> ] ) <EOL> return url_string <EOL> def sign_request ( self , signature_method , consumer , token ) : <EOL> self . set_parameter ( '<STR_LIT>' , signature_method . get_name ( ) ) <EOL> self . set_parameter ( '<STR_LIT>' , self . build_signature ( signature_method , consumer , token ) ) <EOL> def build_signature ( self , signature_method , consumer , token ) : <EOL> return signature_method . build_signature ( self , consumer , token ) <EOL> @ staticmethod <EOL> def from_request ( http_method , http_url , headers = None , parameters = None , query_string = None ) : <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 . index ( '<STR_LIT>' ) > - <NUM_LIT:1> : <EOL> try : <EOL> header_params = OAuthRequest . _split_header ( auth_header ) <EOL> parameters . update ( header_params ) <EOL> except : <EOL> raise OAuthError ( '<STR_LIT>' ) <EOL> if query_string : <EOL> query_params = OAuthRequest . _split_url_string ( query_string ) <EOL> parameters . update ( query_params ) <EOL> param_str = urlparse . urlparse ( http_url ) [ <NUM_LIT:4> ] <EOL> url_params = OAuthRequest . _split_url_string ( param_str ) <EOL> parameters . update ( url_params ) <EOL> if parameters : <EOL> return OAuthRequest ( http_method , http_url , parameters ) <EOL> return None <EOL> @ staticmethod <EOL> def from_consumer_and_token ( oauth_consumer , token = None , http_method = HTTP_METHOD , http_url = None , parameters = None ) : <EOL> if not parameters : <EOL> parameters = { } <EOL> defaults = { <EOL> '<STR_LIT>' : oauth_consumer . key , <EOL> '<STR_LIT>' : generate_timestamp ( ) , <EOL> '<STR_LIT>' : generate_nonce ( ) , <EOL> '<STR_LIT>' : OAuthRequest . version , <EOL> } <EOL> defaults . update ( parameters ) <EOL> parameters = defaults <EOL> if token : <EOL> parameters [ '<STR_LIT>' ] = token . key <EOL> return OAuthRequest ( http_method , http_url , parameters ) <EOL> @ staticmethod <EOL> def from_token_and_callback ( token , callback = None , 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>' ] = escape ( callback ) <EOL> return OAuthRequest ( http_method , http_url , parameters ) <EOL> @ staticmethod <EOL> def _split_header ( header ) : <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 . unquote ( param_parts [ <NUM_LIT:1> ] . strip ( '<STR_LIT>' ) ) <EOL> return params <EOL> @ staticmethod <EOL> def _split_url_string ( param_str ) : <EOL> parameters = cgi . parse_qs ( param_str , keep_blank_values = False ) <EOL> for k , v in parameters . iteritems ( ) : <EOL> parameters [ k ] = urllib . unquote ( v [ <NUM_LIT:0> ] ) <EOL> return parameters <EOL> class OAuthServer ( object ) : <EOL> timestamp_threshold = <NUM_LIT> <EOL> version = VERSION <EOL> signature_methods = None <EOL> data_store = None <EOL> def __init__ ( self , data_store = None , signature_methods = None ) : <EOL> self . data_store = data_store <EOL> self . signature_methods = signature_methods or { } <EOL> def set_data_store ( self , oauth_data_store ) : <EOL> self . data_store = data_store <EOL> def get_data_store ( self ) : <EOL> return self . data_store <EOL> def add_signature_method ( self , signature_method ) : <EOL> self . signature_methods [ signature_method . get_name ( ) ] = signature_method <EOL> return self . signature_methods <EOL> def fetch_request_token ( self , oauth_request ) : <EOL> try : <EOL> token = self . _get_token ( oauth_request , '<STR_LIT>' ) <EOL> except OAuthError : <EOL> version = self . _get_version ( oauth_request ) <EOL> consumer = self . _get_consumer ( oauth_request ) <EOL> self . _check_signature ( oauth_request , consumer , None ) <EOL> token = self . data_store . fetch_request_token ( consumer ) <EOL> return token <EOL> def fetch_access_token ( self , oauth_request ) : <EOL> version = self . _get_version ( oauth_request ) <EOL> consumer = self . _get_consumer ( oauth_request ) <EOL> token = self . _get_token ( oauth_request , '<STR_LIT>' ) <EOL> self . _check_signature ( oauth_request , consumer , token ) <EOL> new_token = self . data_store . fetch_access_token ( consumer , token ) <EOL> return new_token <EOL> def verify_request ( self , oauth_request ) : <EOL> version = self . _get_version ( oauth_request ) <EOL> consumer = self . _get_consumer ( oauth_request ) <EOL> token = self . _get_token ( oauth_request , consumer , '<STR_LIT>' ) <EOL> self . _check_signature ( oauth_request , consumer , token ) <EOL> parameters = oauth_request . get_nonoauth_parameters ( ) <EOL> return consumer , token , parameters <EOL> def authorize_token ( self , token , user ) : <EOL> return self . data_store . authorize_request_token ( token , user ) <EOL> def get_callback ( self , oauth_request ) : <EOL> return oauth_request . get_parameter ( '<STR_LIT>' ) <EOL> def build_authenticate_header ( self , realm = '<STR_LIT>' ) : <EOL> return { '<STR_LIT>' : '<STR_LIT>' % realm } <EOL> def _get_version ( self , oauth_request ) : <EOL> try : <EOL> version = oauth_request . get_parameter ( '<STR_LIT>' ) <EOL> except : <EOL> version = VERSION <EOL> if version and version != self . version : <EOL> raise OAuthError ( '<STR_LIT>' % str ( version ) ) <EOL> return version <EOL> def _get_signature_method ( self , oauth_request ) : <EOL> try : <EOL> signature_method = oauth_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 ( self . signature_methods . keys ( ) ) <EOL> raise OAuthError ( '<STR_LIT>' % ( signature_method , signature_method_names ) ) <EOL> return signature_method <EOL> def _get_consumer ( self , oauth_request ) : <EOL> consumer_key = oauth_request . get_parameter ( '<STR_LIT>' ) <EOL> if not consumer_key : <EOL> raise OAuthError ( '<STR_LIT>' ) <EOL> consumer = self . data_store . lookup_consumer ( consumer_key ) <EOL> if not consumer : <EOL> raise OAuthError ( '<STR_LIT>' ) <EOL> return consumer <EOL> def _get_token ( self , oauth_request , consumer , token_type = '<STR_LIT>' ) : <EOL> token_field = oauth_request . get_parameter ( '<STR_LIT>' ) <EOL> token = self . data_store . lookup_token ( consumer , token_type , token_field ) <EOL> if not token : <EOL> raise OAuthError ( '<STR_LIT>' % ( token_type , token_field ) ) <EOL> return token <EOL> def _check_signature ( self , oauth_request , consumer , token ) : <EOL> timestamp , nonce = oauth_request . _get_timestamp_nonce ( ) <EOL> self . _check_timestamp ( timestamp ) <EOL> self . _check_nonce ( consumer , token , nonce ) <EOL> signature_method = self . _get_signature_method ( oauth_request ) <EOL> try : <EOL> signature = oauth_request . get_parameter ( '<STR_LIT>' ) <EOL> except : <EOL> raise OAuthError ( '<STR_LIT>' ) <EOL> valid_sig = signature_method . check_signature ( oauth_request , consumer , token , signature ) <EOL> if not valid_sig : <EOL> key , base = signature_method . build_signature_base_string ( oauth_request , consumer , token ) <EOL> raise OAuthError ( '<STR_LIT>' % base ) <EOL> built = signature_method . build_signature ( oauth_request , consumer , token ) <EOL> def _check_timestamp ( self , timestamp ) : <EOL> timestamp = int ( timestamp ) <EOL> now = int ( time . time ( ) ) <EOL> lapsed = now - timestamp <EOL> if lapsed > self . timestamp_threshold : <EOL> raise OAuthError ( '<STR_LIT>' % ( timestamp , now , self . timestamp_threshold ) ) <EOL> def _check_nonce ( self , consumer , token , nonce ) : <EOL> nonce = self . data_store . lookup_nonce ( consumer , token , nonce ) <EOL> if nonce : <EOL> raise OAuthError ( '<STR_LIT>' % str ( nonce ) ) <EOL> class OAuthClient ( object ) : <EOL> consumer = None <EOL> token = None <EOL> def __init__ ( self , oauth_consumer , oauth_token ) : <EOL> self . consumer = oauth_consumer <EOL> self . token = oauth_token <EOL> def get_consumer ( self ) : <EOL> return self . consumer <EOL> def get_token ( self ) : <EOL> return self . token <EOL> def fetch_request_token ( self , oauth_request ) : <EOL> raise NotImplementedError <EOL> def fetch_access_token ( self , oauth_request ) : <EOL> raise NotImplementedError <EOL> def access_resource ( self , oauth_request ) : <EOL> raise NotImplementedError <EOL> class OAuthDataStore ( object ) : <EOL> def lookup_consumer ( self , key ) : <EOL> raise NotImplementedError <EOL> def lookup_token ( self , oauth_consumer , token_type , token_token ) : <EOL> raise NotImplementedError <EOL> def lookup_nonce ( self , oauth_consumer , oauth_token , nonce , timestamp ) : <EOL> raise NotImplementedError <EOL> def fetch_request_token ( self , oauth_consumer ) : <EOL> raise NotImplementedError <EOL> def fetch_access_token ( self , oauth_consumer , oauth_token ) : <EOL> raise NotImplementedError <EOL> def authorize_request_token ( self , oauth_token , user ) : <EOL> raise NotImplementedError <EOL> class OAuthSignatureMethod ( object ) : <EOL> def get_name ( self ) : <EOL> raise NotImplementedError <EOL> def build_signature_base_string ( self , oauth_request , oauth_consumer , oauth_token ) : <EOL> raise NotImplementedError <EOL> def build_signature ( self , oauth_request , oauth_consumer , oauth_token ) : <EOL> raise NotImplementedError <EOL> def check_signature ( self , oauth_request , consumer , token , signature ) : <EOL> built = self . build_signature ( oauth_request , consumer , token ) <EOL> logging . info ( "<STR_LIT>" % built ) <EOL> logging . info ( "<STR_LIT>" % signature ) <EOL> return built == signature <EOL> class OAuthSignatureMethod_HMAC_SHA1 ( OAuthSignatureMethod ) : <EOL> def get_name ( self ) : <EOL> return '<STR_LIT>' <EOL> def build_signature_base_string ( self , oauth_request , consumer , token ) : <EOL> sig = ( <EOL> escape ( oauth_request . get_normalized_http_method ( ) ) , <EOL> escape ( oauth_request . get_normalized_http_url ( ) ) , <EOL> escape ( oauth_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 build_signature ( self , oauth_request , consumer , token ) : <EOL> key , raw = self . build_signature_base_string ( oauth_request , consumer , token ) <EOL> hashed = hmac . new ( key , raw , hashlib . sha1 ) <EOL> return base64 . b64encode ( hashed . digest ( ) ) <EOL> class OAuthSignatureMethod_PLAINTEXT ( OAuthSignatureMethod ) : <EOL> def get_name ( self ) : <EOL> return '<STR_LIT>' <EOL> def build_signature_base_string ( self , oauth_request , consumer , token ) : <EOL> sig = escape ( consumer . secret ) + '<STR_LIT:&>' <EOL> if token : <EOL> sig = sig + escape ( token . secret ) <EOL> return sig <EOL> def build_signature ( self , oauth_request , consumer , token ) : <EOL> return self . build_signature_base_string ( oauth_request , consumer , token ) </s>
