<s> from __future__ import absolute_import <EOL> """<STR_LIT>""" <EOL> import logging <EOL> import time <EOL> import urlparse <EOL> from oauthlib . common import Request , urlencode , generate_nonce <EOL> from oauthlib . common import generate_timestamp <EOL> from . import parameters , signature , utils <EOL> logger = logging . getLogger ( __name__ ) <EOL> SIGNATURE_HMAC = u"<STR_LIT>" <EOL> SIGNATURE_RSA = u"<STR_LIT>" <EOL> SIGNATURE_PLAINTEXT = u"<STR_LIT>" <EOL> SIGNATURE_METHODS = ( SIGNATURE_HMAC , SIGNATURE_RSA , SIGNATURE_PLAINTEXT ) <EOL> SIGNATURE_TYPE_AUTH_HEADER = u'<STR_LIT>' <EOL> SIGNATURE_TYPE_QUERY = u'<STR_LIT>' <EOL> SIGNATURE_TYPE_BODY = u'<STR_LIT>' <EOL> CONTENT_TYPE_FORM_URLENCODED = u'<STR_LIT>' <EOL> class Client ( object ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , client_key , <EOL> client_secret = None , <EOL> resource_owner_key = None , <EOL> resource_owner_secret = None , <EOL> callback_uri = None , <EOL> signature_method = SIGNATURE_HMAC , <EOL> signature_type = SIGNATURE_TYPE_AUTH_HEADER , <EOL> rsa_key = None , verifier = None ) : <EOL> self . client_key = client_key <EOL> self . client_secret = client_secret <EOL> self . resource_owner_key = resource_owner_key <EOL> self . resource_owner_secret = resource_owner_secret <EOL> self . signature_method = signature_method <EOL> self . signature_type = signature_type <EOL> self . callback_uri = callback_uri <EOL> self . rsa_key = rsa_key <EOL> self . verifier = verifier <EOL> if self . signature_method == SIGNATURE_RSA and self . rsa_key is None : <EOL> raise ValueError ( '<STR_LIT>' ) <EOL> def get_oauth_signature ( self , request ) : <EOL> """<STR_LIT>""" <EOL> if self . signature_method == SIGNATURE_PLAINTEXT : <EOL> return signature . sign_plaintext ( self . client_secret , <EOL> self . resource_owner_secret ) <EOL> uri , headers , body = self . _render ( request ) <EOL> collected_params = signature . collect_parameters ( <EOL> uri_query = urlparse . urlparse ( uri ) . query , <EOL> body = body , <EOL> headers = headers ) <EOL> logger . debug ( "<STR_LIT>" . format ( collected_params ) ) <EOL> normalized_params = signature . normalize_parameters ( collected_params ) <EOL> normalized_uri = signature . normalize_base_string_uri ( request . uri ) <EOL> logger . debug ( "<STR_LIT>" . format ( normalized_params ) ) <EOL> logger . debug ( "<STR_LIT>" . format ( normalized_uri ) ) <EOL> base_string = signature . construct_base_string ( request . http_method , <EOL> normalized_uri , normalized_params ) <EOL> logger . debug ( "<STR_LIT>" . format ( base_string ) ) <EOL> if self . signature_method == SIGNATURE_HMAC : <EOL> sig = signature . sign_hmac_sha1 ( base_string , self . client_secret , <EOL> self . resource_owner_secret ) <EOL> elif self . signature_method == SIGNATURE_RSA : <EOL> sig = signature . sign_rsa_sha1 ( base_string , self . rsa_key ) <EOL> else : <EOL> sig = signature . sign_plaintext ( self . client_secret , <EOL> self . resource_owner_secret ) <EOL> logger . debug ( "<STR_LIT>" . format ( sig ) ) <EOL> return sig <EOL> def get_oauth_params ( self ) : <EOL> """<STR_LIT>""" <EOL> params = [ <EOL> ( u'<STR_LIT>' , generate_nonce ( ) ) , <EOL> ( u'<STR_LIT>' , generate_timestamp ( ) ) , <EOL> ( u'<STR_LIT>' , u'<STR_LIT:1.0>' ) , <EOL> ( u'<STR_LIT>' , self . signature_method ) , <EOL> ( u'<STR_LIT>' , self . client_key ) , <EOL> ] <EOL> if self . resource_owner_key : <EOL> params . append ( ( u'<STR_LIT>' , self . resource_owner_key ) ) <EOL> if self . callback_uri : <EOL> params . append ( ( u'<STR_LIT>' , self . callback_uri ) ) <EOL> if self . verifier : <EOL> params . append ( ( u'<STR_LIT>' , self . verifier ) ) <EOL> return params <EOL> def _render ( self , request , formencode = False ) : <EOL> """<STR_LIT>""" <EOL> uri , headers , body = request . uri , request . headers , request . body <EOL> if self . signature_type == SIGNATURE_TYPE_AUTH_HEADER : <EOL> headers = parameters . prepare_headers ( request . oauth_params , request . headers ) <EOL> elif self . signature_type == SIGNATURE_TYPE_BODY and request . decoded_body is not None : <EOL> body = parameters . prepare_form_encoded_body ( request . oauth_params , request . decoded_body ) <EOL> if formencode : <EOL> body = urlencode ( body ) <EOL> headers [ '<STR_LIT:Content-Type>' ] = u'<STR_LIT>' <EOL> elif self . signature_type == SIGNATURE_TYPE_QUERY : <EOL> uri = parameters . prepare_request_uri_query ( request . oauth_params , request . uri ) <EOL> else : <EOL> raise ValueError ( '<STR_LIT>' ) <EOL> return uri , headers , body <EOL> def sign ( self , uri , http_method = u'<STR_LIT:GET>' , body = None , headers = None ) : <EOL> """<STR_LIT>""" <EOL> request = Request ( uri , http_method , body , headers ) <EOL> content_type = request . headers . get ( '<STR_LIT:Content-Type>' , None ) <EOL> multipart = content_type and content_type . startswith ( '<STR_LIT>' ) <EOL> should_have_params = content_type == CONTENT_TYPE_FORM_URLENCODED <EOL> has_params = request . decoded_body is not None <EOL> if multipart and has_params : <EOL> raise ValueError ( "<STR_LIT>" ) <EOL> elif should_have_params and not has_params : <EOL> raise ValueError ( "<STR_LIT>" ) <EOL> elif not should_have_params and has_params : <EOL> raise ValueError ( "<STR_LIT>" ) <EOL> elif self . signature_type == SIGNATURE_TYPE_BODY and not ( <EOL> should_have_params and has_params and not multipart ) : <EOL> raise ValueError ( '<STR_LIT>' ) <EOL> request . oauth_params = self . get_oauth_params ( ) <EOL> request . oauth_params . append ( ( u'<STR_LIT>' , self . get_oauth_signature ( request ) ) ) <EOL> return self . _render ( request , formencode = True ) <EOL> class Server ( object ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self ) : <EOL> pass <EOL> @ property <EOL> def allowed_signature_methods ( self ) : <EOL> return SIGNATURE_METHODS <EOL> @ property <EOL> def safe_characters ( self ) : <EOL> return set ( utils . UNICODE_ASCII_CHARACTER_SET ) <EOL> @ property <EOL> def client_key_length ( self ) : <EOL> return <NUM_LIT:20> , <NUM_LIT:30> <EOL> @ property <EOL> def request_token_length ( self ) : <EOL> return <NUM_LIT:20> , <NUM_LIT:30> <EOL> @ property <EOL> def access_token_length ( self ) : <EOL> return <NUM_LIT:20> , <NUM_LIT:30> <EOL> @ property <EOL> def timestamp_lifetime ( self ) : <EOL> return <NUM_LIT> <EOL> @ property <EOL> def nonce_length ( self ) : <EOL> return <NUM_LIT:20> , <NUM_LIT:30> <EOL> @ property <EOL> def verifier_length ( self ) : <EOL> return <NUM_LIT:20> , <NUM_LIT:30> <EOL> @ property <EOL> def realms ( self ) : <EOL> return [ ] <EOL> @ property <EOL> def enforce_ssl ( self ) : <EOL> return True <EOL> def check_client_key ( self , client_key ) : <EOL> """<STR_LIT>""" <EOL> lower , upper = self . client_key_length <EOL> return ( set ( client_key ) <= self . safe_characters and <EOL> lower <= len ( client_key ) <= upper ) <EOL> def check_request_token ( self , request_token ) : <EOL> """<STR_LIT>""" <EOL> lower , upper = self . request_token_length <EOL> return ( set ( request_token ) <= self . safe_characters and <EOL> lower <= len ( request_token ) <= upper ) <EOL> def check_access_token ( self , request_token ) : <EOL> """<STR_LIT>""" <EOL> lower , upper = self . access_token_length <EOL> return ( set ( request_token ) <= self . safe_characters and <EOL> lower <= len ( request_token ) <= upper ) <EOL> def check_nonce ( self , nonce ) : <EOL> """<STR_LIT>""" <EOL> lower , upper = self . nonce_length <EOL> return ( set ( nonce ) <= self . safe_characters and <EOL> lower <= len ( nonce ) <= upper ) <EOL> def check_verifier ( self , verifier ) : <EOL> """<STR_LIT>""" <EOL> lower , upper = self . verifier_length <EOL> return ( set ( verifier ) <= self . safe_characters and <EOL> lower <= len ( verifier ) <= upper ) <EOL> def check_realm ( self , realm ) : <EOL> """<STR_LIT>""" <EOL> return realm in self . realms <EOL> def get_client_secret ( self , client_key ) : <EOL> """<STR_LIT>""" <EOL> raise NotImplementedError ( "<STR_LIT>" ) <EOL> @ property <EOL> def dummy_client ( self ) : <EOL> """<STR_LIT>""" <EOL> raise NotImplementedError ( "<STR_LIT>" ) <EOL> def get_request_token_secret ( self , client_key , request_token ) : <EOL> """<STR_LIT>""" <EOL> raise NotImplementedError ( "<STR_LIT>" ) <EOL> def get_access_token_secret ( self , client_key , access_token ) : <EOL> """<STR_LIT>""" <EOL> raise NotImplementedError ( "<STR_LIT>" ) <EOL> @ property <EOL> def dummy_request_token ( self ) : <EOL> """<STR_LIT>""" <EOL> raise NotImplementedError ( "<STR_LIT>" ) <EOL> @ property <EOL> def dummy_access_token ( self ) : <EOL> """<STR_LIT>""" <EOL> raise NotImplementedError ( "<STR_LIT>" ) <EOL> def get_rsa_key ( self , client_key ) : <EOL> """<STR_LIT>""" <EOL> raise NotImplementedError ( "<STR_LIT>" ) <EOL> def get_signature_type_and_params ( self , request ) : <EOL> """<STR_LIT>""" <EOL> header_params = signature . collect_parameters ( headers = request . headers , <EOL> exclude_oauth_signature = False ) <EOL> body_params = signature . collect_parameters ( body = request . body , <EOL> exclude_oauth_signature = False ) <EOL> query_params = signature . collect_parameters ( uri_query = request . uri_query , <EOL> exclude_oauth_signature = False ) <EOL> params = [ ] <EOL> params . extend ( header_params ) <EOL> params . extend ( body_params ) <EOL> params . extend ( query_params ) <EOL> signature_types_with_oauth_params = filter ( lambda s : s [ <NUM_LIT:2> ] , ( <EOL> ( SIGNATURE_TYPE_AUTH_HEADER , params , <EOL> utils . filter_oauth_params ( header_params ) ) , <EOL> ( SIGNATURE_TYPE_BODY , params , <EOL> utils . filter_oauth_params ( body_params ) ) , <EOL> ( SIGNATURE_TYPE_QUERY , params , <EOL> utils . filter_oauth_params ( query_params ) ) <EOL> ) ) <EOL> if len ( signature_types_with_oauth_params ) > <NUM_LIT:1> : <EOL> raise ValueError ( '<STR_LIT>' % '<STR_LIT:U+002CU+0020>' . join ( <EOL> [ s [ <NUM_LIT:0> ] for s in signature_types_with_oauth_params ] ) ) <EOL> try : <EOL> signature_type , params , oauth_params = signature_types_with_oauth_params [ <NUM_LIT:0> ] <EOL> except IndexError : <EOL> raise ValueError ( '<STR_LIT>' ) <EOL> return signature_type , params , oauth_params <EOL> def validate_client_key ( self , client_key ) : <EOL> """<STR_LIT>""" <EOL> raise NotImplementedError ( "<STR_LIT>" ) <EOL> def validate_request_token ( self , client_key , request_token ) : <EOL> """<STR_LIT>""" <EOL> raise NotImplementedError ( "<STR_LIT>" ) <EOL> def validate_access_token ( self , client_key , access_token ) : <EOL> """<STR_LIT>""" <EOL> raise NotImplementedError ( "<STR_LIT>" ) <EOL> def validate_timestamp_and_nonce ( self , client_key , timestamp , nonce , <EOL> request_token = None , access_token = None ) : <EOL> """<STR_LIT>""" <EOL> raise NotImplementedError ( "<STR_LIT>" ) <EOL> def validate_redirect_uri ( self , client_key , redirect_uri ) : <EOL> """<STR_LIT>""" <EOL> raise NotImplementedError ( "<STR_LIT>" ) <EOL> def validate_requested_realm ( self , client_key , realm ) : <EOL> """<STR_LIT>""" <EOL> raise NotImplementedError ( "<STR_LIT>" ) <EOL> def validate_realm ( self , client_key , access_token , uri = None , <EOL> required_realm = None ) : <EOL> """<STR_LIT>""" <EOL> raise NotImplementedError ( "<STR_LIT>" ) <EOL> def validate_verifier ( self , client_key , request_token , verifier ) : <EOL> """<STR_LIT>""" <EOL> raise NotImplementedError ( "<STR_LIT>" ) <EOL> def verify_request ( self , uri , http_method = u'<STR_LIT:GET>' , body = None , <EOL> headers = None , require_resource_owner = True , require_verifier = False , <EOL> require_realm = False , required_realm = None ) : <EOL> """<STR_LIT>""" <EOL> headers = headers or { } <EOL> if ( u"<STR_LIT:Content-Type>" in headers and <EOL> headers [ u"<STR_LIT:Content-Type>" ] == CONTENT_TYPE_FORM_URLENCODED ) : <EOL> request = Request ( uri , http_method , body , headers ) <EOL> else : <EOL> request = Request ( uri , http_method , u'<STR_LIT>' , headers ) <EOL> if self . enforce_ssl and not request . uri . lower ( ) . startswith ( "<STR_LIT>" ) : <EOL> raise ValueError ( "<STR_LIT>" ) <EOL> signature_type , params , oauth_params = self . get_signature_type_and_params ( request ) <EOL> if len ( dict ( oauth_params ) ) != len ( oauth_params ) : <EOL> raise ValueError ( "<STR_LIT>" ) <EOL> oauth_params = dict ( oauth_params ) <EOL> request_signature = oauth_params . get ( u'<STR_LIT>' ) <EOL> client_key = oauth_params . get ( u'<STR_LIT>' ) <EOL> resource_owner_key = oauth_params . get ( u'<STR_LIT>' ) <EOL> nonce = oauth_params . get ( u'<STR_LIT>' ) <EOL> timestamp = oauth_params . get ( u'<STR_LIT>' ) <EOL> callback_uri = oauth_params . get ( u'<STR_LIT>' ) <EOL> verifier = oauth_params . get ( u'<STR_LIT>' ) <EOL> signature_method = oauth_params . get ( u'<STR_LIT>' ) <EOL> realm = dict ( params ) . get ( u'<STR_LIT>' ) <EOL> if not all ( ( request_signature , client_key , nonce , <EOL> timestamp , signature_method ) ) : <EOL> raise ValueError ( "<STR_LIT>" ) <EOL> if not signature_method in self . allowed_signature_methods : <EOL> raise ValueError ( "<STR_LIT>" ) <EOL> if u'<STR_LIT>' in oauth_params and oauth_params [ u'<STR_LIT>' ] != u'<STR_LIT:1.0>' : <EOL> raise ValueError ( "<STR_LIT>" ) <EOL> if len ( timestamp ) != <NUM_LIT:10> : <EOL> raise ValueError ( "<STR_LIT>" ) <EOL> try : <EOL> ts = int ( timestamp ) <EOL> except ValueError : <EOL> raise ValueError ( "<STR_LIT>" ) <EOL> else : <EOL> if time . time ( ) - ts > self . timestamp_lifetime : <EOL> raise ValueError ( "<STR_LIT>" ) <EOL> if not self . check_client_key ( client_key ) : <EOL> raise ValueError ( "<STR_LIT>" ) <EOL> if not resource_owner_key and require_resource_owner : <EOL> raise ValueError ( "<STR_LIT>" ) <EOL> if ( require_resource_owner and not require_verifier and <EOL> not self . check_access_token ( resource_owner_key ) ) : <EOL> raise ValueError ( "<STR_LIT>" ) <EOL> if ( require_resource_owner and require_verifier and <EOL> not self . check_request_token ( resource_owner_key ) ) : <EOL> raise ValueError ( "<STR_LIT>" ) <EOL> if not self . check_nonce ( nonce ) : <EOL> raise ValueError ( "<STR_LIT>" ) <EOL> if realm and not self . check_realm ( realm ) : <EOL> raise ValueError ( "<STR_LIT>" % self . realms ) <EOL> if not verifier and require_verifier : <EOL> raise ValueError ( "<STR_LIT>" ) <EOL> if require_verifier and not self . check_verifier ( verifier ) : <EOL> raise ValueError ( "<STR_LIT>" ) <EOL> if require_verifier : <EOL> token = { "<STR_LIT>" : resource_owner_key } <EOL> else : <EOL> token = { "<STR_LIT>" : resource_owner_key } <EOL> if not self . validate_timestamp_and_nonce ( client_key , timestamp , <EOL> nonce , ** token ) : <EOL> return False <EOL> valid_client = self . validate_client_key ( client_key ) <EOL> if not valid_client : <EOL> client_key = self . dummy_client <EOL> valid_redirect = self . validate_redirect_uri ( client_key , callback_uri ) <EOL> if resource_owner_key : <EOL> if require_verifier : <EOL> valid_resource_owner = self . validate_request_token ( <EOL> client_key , resource_owner_key ) <EOL> else : <EOL> valid_resource_owner = self . validate_access_token ( <EOL> client_key , resource_owner_key ) <EOL> if not valid_resource_owner : <EOL> resource_owner_key = self . dummy_resource_owner <EOL> else : <EOL> valid_resource_owner = True <EOL> if require_realm and not resource_owner_key : <EOL> valid_realm = self . validate_requested_realm ( client_key , realm ) <EOL> elif require_verifier : <EOL> valid_realm = True <EOL> else : <EOL> valid_realm = self . validate_realm ( client_key , resource_owner_key , <EOL> uri = request . uri , required_realm = required_realm ) <EOL> if verifier : <EOL> valid_verifier = self . validate_verifier ( client_key , <EOL> resource_owner_key , verifier ) <EOL> else : <EOL> valid_verifier = True <EOL> request . params = filter ( lambda x : x [ <NUM_LIT:0> ] != "<STR_LIT>" , params ) <EOL> request . signature = request_signature <EOL> if signature_method == SIGNATURE_RSA : <EOL> rsa_key = self . get_rsa_key ( client_key ) <EOL> valid_signature = signature . verify_rsa_sha1 ( request , rsa_key ) <EOL> else : <EOL> client_secret = self . get_client_secret ( client_key ) <EOL> if require_verifier : <EOL> resource_owner_secret = self . get_request_token_secret ( <EOL> client_key , resource_owner_key ) <EOL> else : <EOL> resource_owner_secret = self . get_access_token_secret ( <EOL> client_key , resource_owner_key ) <EOL> if signature_method == SIGNATURE_HMAC : <EOL> valid_signature = signature . verify_hmac_sha1 ( request , <EOL> client_secret , resource_owner_secret ) <EOL> else : <EOL> valid_signature = signature . verify_plaintext ( request , <EOL> client_secret , resource_owner_secret ) <EOL> v = all ( ( valid_client , valid_resource_owner , valid_realm , <EOL> valid_redirect , valid_verifier , valid_signature ) ) <EOL> logger = logging . getLogger ( "<STR_LIT>" ) <EOL> if not v : <EOL> logger . info ( "<STR_LIT>" ) <EOL> logger . info ( "<STR_LIT>" % valid_client ) <EOL> logger . info ( "<STR_LIT>" % ( valid_resource_owner , require_resource_owner ) ) <EOL> logger . info ( "<STR_LIT>" % ( valid_realm , require_realm ) ) <EOL> logger . info ( "<STR_LIT>" % valid_redirect ) <EOL> logger . info ( "<STR_LIT>" % ( valid_verifier , require_verifier ) ) <EOL> logger . info ( "<STR_LIT>" % valid_signature ) <EOL> return v </s>
