<s> """<STR_LIT>""" <EOL> from __future__ import absolute_import , unicode_literals <EOL> import time <EOL> from oauthlib . common import Request , generate_token <EOL> from . . import signature , utils , errors <EOL> from . . import CONTENT_TYPE_FORM_URLENCODED <EOL> from . . import SIGNATURE_HMAC , SIGNATURE_RSA <EOL> from . . import SIGNATURE_TYPE_AUTH_HEADER <EOL> from . . import SIGNATURE_TYPE_QUERY <EOL> from . . import SIGNATURE_TYPE_BODY <EOL> class BaseEndpoint ( object ) : <EOL> def __init__ ( self , request_validator , token_generator = None ) : <EOL> self . request_validator = request_validator <EOL> self . token_generator = token_generator or generate_token <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 , with_realm = True ) <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 = list ( 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> found_types = [ s [ <NUM_LIT:0> ] for s in signature_types_with_oauth_params ] <EOL> raise errors . InvalidRequestError ( <EOL> description = ( '<STR_LIT>' <EOL> '<STR_LIT>' , <EOL> '<STR_LIT:U+002CU+0020>' . join ( found_types ) ) ) <EOL> try : <EOL> signature_type , params , oauth_params = signature_types_with_oauth_params [ <EOL> <NUM_LIT:0> ] <EOL> except IndexError : <EOL> raise errors . InvalidRequestError ( <EOL> description = '<STR_LIT>' ) <EOL> return signature_type , params , oauth_params <EOL> def _create_request ( self , uri , http_method , body , headers ) : <EOL> headers = headers or { } <EOL> if ( "<STR_LIT:Content-Type>" in headers and <EOL> CONTENT_TYPE_FORM_URLENCODED in headers [ "<STR_LIT:Content-Type>" ] ) : <EOL> request = Request ( uri , http_method , body , headers ) <EOL> else : <EOL> request = Request ( uri , http_method , '<STR_LIT>' , headers ) <EOL> signature_type , params , oauth_params = ( <EOL> self . _get_signature_type_and_params ( request ) ) <EOL> if len ( dict ( oauth_params ) ) != len ( oauth_params ) : <EOL> raise errors . InvalidRequestError ( <EOL> description = '<STR_LIT>' ) <EOL> oauth_params = dict ( oauth_params ) <EOL> request . signature = oauth_params . get ( '<STR_LIT>' ) <EOL> request . client_key = oauth_params . get ( '<STR_LIT>' ) <EOL> request . resource_owner_key = oauth_params . get ( '<STR_LIT>' ) <EOL> request . nonce = oauth_params . get ( '<STR_LIT>' ) <EOL> request . timestamp = oauth_params . get ( '<STR_LIT>' ) <EOL> request . redirect_uri = oauth_params . get ( '<STR_LIT>' ) <EOL> request . verifier = oauth_params . get ( '<STR_LIT>' ) <EOL> request . signature_method = oauth_params . get ( '<STR_LIT>' ) <EOL> request . realm = dict ( params ) . get ( '<STR_LIT>' ) <EOL> request . oauth_params = oauth_params <EOL> request . params = [ ( k , v ) for k , v in params if k != "<STR_LIT>" ] <EOL> if '<STR_LIT>' in request . headers . get ( '<STR_LIT>' , '<STR_LIT>' ) : <EOL> request . params = [ ( k , v ) <EOL> for k , v in request . params if k != "<STR_LIT>" ] <EOL> return request <EOL> def _check_transport_security ( self , request ) : <EOL> if ( self . request_validator . enforce_ssl and <EOL> not request . uri . lower ( ) . startswith ( "<STR_LIT>" ) ) : <EOL> raise errors . InsecureTransportError ( ) <EOL> def _check_mandatory_parameters ( self , request ) : <EOL> if not all ( ( request . signature , request . client_key , <EOL> request . nonce , request . timestamp , <EOL> request . signature_method ) ) : <EOL> raise errors . InvalidRequestError ( <EOL> description = '<STR_LIT>' ) <EOL> if ( not request . signature_method in <EOL> self . request_validator . allowed_signature_methods ) : <EOL> raise errors . InvalidSignatureMethodError ( <EOL> description = "<STR_LIT>" % ( <EOL> request . signature_method , <EOL> self . request_validator . allowed_signature_methods ) ) <EOL> if ( '<STR_LIT>' in request . oauth_params and <EOL> request . oauth_params [ '<STR_LIT>' ] != '<STR_LIT:1.0>' ) : <EOL> raise errors . InvalidRequestError ( <EOL> description = '<STR_LIT>' ) <EOL> if len ( request . timestamp ) != <NUM_LIT:10> : <EOL> raise errors . InvalidRequestError ( <EOL> description = '<STR_LIT>' ) <EOL> try : <EOL> ts = int ( request . timestamp ) <EOL> except ValueError : <EOL> raise errors . InvalidRequestError ( <EOL> description = '<STR_LIT>' ) <EOL> else : <EOL> if abs ( time . time ( ) - ts ) > self . request_validator . timestamp_lifetime : <EOL> raise errors . InvalidRequestError ( <EOL> description = ( '<STR_LIT>' <EOL> '<STR_LIT>' % ( <EOL> self . request_validator . timestamp_lifetime ) ) ) <EOL> if not self . request_validator . check_client_key ( request . client_key ) : <EOL> raise errors . InvalidRequestError ( <EOL> description = '<STR_LIT>' ) <EOL> if not self . request_validator . check_nonce ( request . nonce ) : <EOL> raise errors . InvalidRequestError ( <EOL> description = '<STR_LIT>' ) <EOL> def _check_signature ( self , request , is_token_request = False ) : <EOL> if request . signature_method == SIGNATURE_RSA : <EOL> rsa_key = self . request_validator . get_rsa_key ( <EOL> request . client_key , request ) <EOL> valid_signature = signature . verify_rsa_sha1 ( request , rsa_key ) <EOL> else : <EOL> client_secret = self . request_validator . get_client_secret ( <EOL> request . client_key , request ) <EOL> resource_owner_secret = None <EOL> if request . resource_owner_key : <EOL> if is_token_request : <EOL> resource_owner_secret = self . request_validator . get_request_token_secret ( <EOL> request . client_key , request . resource_owner_key , request ) <EOL> else : <EOL> resource_owner_secret = self . request_validator . get_access_token_secret ( <EOL> request . client_key , request . resource_owner_key , request ) <EOL> if request . 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> return valid_signature </s>
