<s> import json <EOL> import logging <EOL> from requests import Response <EOL> from cStringIO import StringIO <EOL> try : <EOL> from werkzeug . exceptions import Unauthorized <EOL> except ImportError : <EOL> Unauthorized = Exception <EOL> from . import utils <EOL> class Provider ( object ) : <EOL> """<STR_LIT>""" <EOL> def _handle_exception ( self , exc ) : <EOL> """<STR_LIT>""" <EOL> logger = logging . getLogger ( __name__ ) <EOL> logger . exception ( exc ) <EOL> def _make_response ( self , body = '<STR_LIT>' , headers = None , status_code = <NUM_LIT:200> ) : <EOL> """<STR_LIT>""" <EOL> res = Response ( ) <EOL> res . status_code = status_code <EOL> if headers is not None : <EOL> res . headers . update ( headers ) <EOL> res . raw = StringIO ( body ) <EOL> return res <EOL> def _make_redirect_error_response ( self , redirect_uri , err ) : <EOL> """<STR_LIT>""" <EOL> params = { <EOL> '<STR_LIT:error>' : err , <EOL> '<STR_LIT>' : None , <EOL> '<STR_LIT>' : None , <EOL> '<STR_LIT>' : None <EOL> } <EOL> redirect = utils . build_url ( redirect_uri , params ) <EOL> return self . _make_response ( headers = { '<STR_LIT>' : redirect } , <EOL> status_code = <NUM_LIT> ) <EOL> def _make_json_response ( self , data , headers = None , status_code = <NUM_LIT:200> ) : <EOL> """<STR_LIT>""" <EOL> response_headers = { } <EOL> if headers is not None : <EOL> response_headers . update ( headers ) <EOL> response_headers [ '<STR_LIT:Content-Type>' ] = '<STR_LIT>' <EOL> response_headers [ '<STR_LIT>' ] = '<STR_LIT>' <EOL> response_headers [ '<STR_LIT>' ] = '<STR_LIT>' <EOL> return self . _make_response ( json . dumps ( data ) , <EOL> response_headers , <EOL> status_code ) <EOL> def _make_json_error_response ( self , err ) : <EOL> """<STR_LIT>""" <EOL> return self . _make_json_response ( { '<STR_LIT:error>' : err } , status_code = <NUM_LIT> ) <EOL> def _invalid_redirect_uri_response ( self ) : <EOL> """<STR_LIT>""" <EOL> return self . _make_json_error_response ( '<STR_LIT>' ) <EOL> class AuthorizationProvider ( Provider ) : <EOL> """<STR_LIT>""" <EOL> @ property <EOL> def token_length ( self ) : <EOL> """<STR_LIT>""" <EOL> return <NUM_LIT> <EOL> @ property <EOL> def token_type ( self ) : <EOL> """<STR_LIT>""" <EOL> return '<STR_LIT>' <EOL> @ property <EOL> def token_expires_in ( self ) : <EOL> """<STR_LIT>""" <EOL> return <NUM_LIT> <EOL> def generate_authorization_code ( self ) : <EOL> """<STR_LIT>""" <EOL> return utils . random_ascii_string ( self . token_length ) <EOL> def generate_access_token ( self ) : <EOL> """<STR_LIT>""" <EOL> return utils . random_ascii_string ( self . token_length ) <EOL> def generate_refresh_token ( self ) : <EOL> """<STR_LIT>""" <EOL> return utils . random_ascii_string ( self . token_length ) <EOL> def get_authorization_code ( self , <EOL> response_type , <EOL> client_id , <EOL> redirect_uri , <EOL> ** params ) : <EOL> """<STR_LIT>""" <EOL> if response_type != '<STR_LIT:code>' : <EOL> err = '<STR_LIT>' <EOL> return self . _make_redirect_error_response ( redirect_uri , err ) <EOL> is_valid_redirect_uri = self . validate_redirect_uri ( client_id , <EOL> redirect_uri ) <EOL> if not is_valid_redirect_uri : <EOL> return self . _invalid_redirect_uri_response ( ) <EOL> is_valid_client_id = self . validate_client_id ( client_id ) <EOL> is_valid_access = self . validate_access ( ) <EOL> scope = params . get ( '<STR_LIT>' , '<STR_LIT>' ) <EOL> is_valid_scope = self . validate_scope ( client_id , scope ) <EOL> if not is_valid_client_id : <EOL> err = '<STR_LIT>' <EOL> return self . _make_redirect_error_response ( redirect_uri , err ) <EOL> if not is_valid_access : <EOL> err = '<STR_LIT>' <EOL> return self . _make_redirect_error_response ( redirect_uri , err ) <EOL> if not is_valid_scope : <EOL> err = '<STR_LIT>' <EOL> return self . _make_redirect_error_response ( redirect_uri , err ) <EOL> code = self . generate_authorization_code ( ) <EOL> self . persist_authorization_code ( client_id = client_id , <EOL> code = code , <EOL> scope = scope ) <EOL> params . update ( { <EOL> '<STR_LIT:code>' : code , <EOL> '<STR_LIT>' : None , <EOL> '<STR_LIT>' : None , <EOL> '<STR_LIT>' : None <EOL> } ) <EOL> redirect = utils . build_url ( redirect_uri , params ) <EOL> return self . _make_response ( headers = { '<STR_LIT>' : redirect } , <EOL> status_code = <NUM_LIT> ) <EOL> def refresh_token ( self , <EOL> grant_type , <EOL> client_id , <EOL> client_secret , <EOL> refresh_token , <EOL> ** params ) : <EOL> """<STR_LIT>""" <EOL> if grant_type != '<STR_LIT>' : <EOL> return self . _make_json_error_response ( '<STR_LIT>' ) <EOL> is_valid_client_id = self . validate_client_id ( client_id ) <EOL> is_valid_client_secret = self . validate_client_secret ( client_id , <EOL> client_secret ) <EOL> scope = params . get ( '<STR_LIT>' , '<STR_LIT>' ) <EOL> is_valid_scope = self . validate_scope ( client_id , scope ) <EOL> data = self . from_refresh_token ( client_id , refresh_token , scope ) <EOL> is_valid_refresh_token = data is not None <EOL> if not ( is_valid_client_id and is_valid_client_secret ) : <EOL> return self . _make_json_error_response ( '<STR_LIT>' ) <EOL> if not is_valid_scope : <EOL> return self . _make_json_error_response ( '<STR_LIT>' ) <EOL> if not is_valid_refresh_token : <EOL> return self . _make_json_error_response ( '<STR_LIT>' ) <EOL> self . discard_refresh_token ( client_id , refresh_token ) <EOL> access_token = self . generate_access_token ( ) <EOL> token_type = self . token_type <EOL> expires_in = self . token_expires_in <EOL> refresh_token = self . generate_refresh_token ( ) <EOL> self . persist_token_information ( client_id = client_id , <EOL> scope = scope , <EOL> access_token = access_token , <EOL> token_type = token_type , <EOL> expires_in = expires_in , <EOL> refresh_token = refresh_token , <EOL> data = data ) <EOL> return self . _make_json_response ( { <EOL> '<STR_LIT>' : access_token , <EOL> '<STR_LIT>' : token_type , <EOL> '<STR_LIT>' : expires_in , <EOL> '<STR_LIT>' : refresh_token <EOL> } ) <EOL> def get_token ( self , <EOL> grant_type , <EOL> client_id , <EOL> client_secret , <EOL> redirect_uri , <EOL> code , <EOL> ** params ) : <EOL> """<STR_LIT>""" <EOL> if grant_type != '<STR_LIT>' : <EOL> return self . _make_json_error_response ( '<STR_LIT>' ) <EOL> is_valid_client_id = self . validate_client_id ( client_id ) <EOL> is_valid_client_secret = self . validate_client_secret ( client_id , <EOL> client_secret ) <EOL> is_valid_redirect_uri = self . validate_redirect_uri ( client_id , <EOL> redirect_uri ) <EOL> scope = params . get ( '<STR_LIT>' , '<STR_LIT>' ) <EOL> is_valid_scope = self . validate_scope ( client_id , scope ) <EOL> data = self . from_authorization_code ( client_id , code , scope ) <EOL> is_valid_grant = data is not None <EOL> if not ( is_valid_client_id and is_valid_client_secret ) : <EOL> return self . _make_json_error_response ( '<STR_LIT>' ) <EOL> if not is_valid_grant or not is_valid_redirect_uri : <EOL> return self . _make_json_error_response ( '<STR_LIT>' ) <EOL> if not is_valid_scope : <EOL> return self . _make_json_error_response ( '<STR_LIT>' ) <EOL> self . discard_authorization_code ( client_id , code ) <EOL> access_token = self . generate_access_token ( ) <EOL> token_type = self . token_type <EOL> expires_in = self . token_expires_in <EOL> refresh_token = self . generate_refresh_token ( ) <EOL> self . persist_token_information ( client_id = client_id , <EOL> scope = scope , <EOL> access_token = access_token , <EOL> token_type = token_type , <EOL> expires_in = expires_in , <EOL> refresh_token = refresh_token , <EOL> data = data ) <EOL> return self . _make_json_response ( { <EOL> '<STR_LIT>' : access_token , <EOL> '<STR_LIT>' : token_type , <EOL> '<STR_LIT>' : expires_in , <EOL> '<STR_LIT>' : refresh_token <EOL> } ) <EOL> def get_authorization_code_from_uri ( self , uri ) : <EOL> """<STR_LIT>""" <EOL> params = utils . url_query_params ( uri ) <EOL> try : <EOL> if '<STR_LIT>' not in params : <EOL> raise TypeError ( '<STR_LIT>' ) <EOL> if '<STR_LIT>' not in params : <EOL> raise TypeError ( '<STR_LIT>' ) <EOL> if '<STR_LIT>' not in params : <EOL> raise TypeError ( '<STR_LIT>' ) <EOL> return self . get_authorization_code ( ** params ) <EOL> except TypeError as exc : <EOL> self . _handle_exception ( exc ) <EOL> err = '<STR_LIT>' <EOL> if '<STR_LIT>' in params : <EOL> u = params [ '<STR_LIT>' ] <EOL> return self . _make_redirect_error_response ( u , err ) <EOL> else : <EOL> return self . _invalid_redirect_uri_response ( ) <EOL> except StandardError as exc : <EOL> self . _handle_exception ( exc ) <EOL> err = '<STR_LIT>' <EOL> u = params [ '<STR_LIT>' ] <EOL> return self . _make_redirect_error_response ( u , err ) <EOL> def get_token_from_post_data ( self , data ) : <EOL> """<STR_LIT>""" <EOL> try : <EOL> for x in [ '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' ] : <EOL> if not data . get ( x ) : <EOL> raise TypeError ( "<STR_LIT>" . format ( x ) ) <EOL> if '<STR_LIT>' in data : <EOL> return self . refresh_token ( ** data ) <EOL> for x in [ '<STR_LIT>' , '<STR_LIT:code>' ] : <EOL> if not data . get ( x ) : <EOL> raise TypeError ( "<STR_LIT>" . format ( x ) ) <EOL> return self . get_token ( ** data ) <EOL> except TypeError as exc : <EOL> self . _handle_exception ( exc ) <EOL> return self . _make_json_error_response ( '<STR_LIT>' ) <EOL> except StandardError as exc : <EOL> self . _handle_exception ( exc ) <EOL> return self . _make_json_error_response ( '<STR_LIT>' ) <EOL> def validate_client_id ( self , client_id ) : <EOL> raise NotImplementedError ( '<STR_LIT>' '<STR_LIT>' ) <EOL> def validate_client_secret ( self , client_id , client_secret ) : <EOL> raise NotImplementedError ( '<STR_LIT>' '<STR_LIT>' ) <EOL> def validate_redirect_uri ( self , client_id , redirect_uri ) : <EOL> raise NotImplementedError ( '<STR_LIT>' '<STR_LIT>' ) <EOL> def validate_scope ( self , client_id , scope ) : <EOL> raise NotImplementedError ( '<STR_LIT>' '<STR_LIT>' ) <EOL> def validate_access ( self ) : <EOL> raise NotImplementedError ( '<STR_LIT>' '<STR_LIT>' ) <EOL> def from_authorization_code ( self , client_id , code , scope ) : <EOL> raise NotImplementedError ( '<STR_LIT>' '<STR_LIT>' ) <EOL> def from_refresh_token ( self , client_id , refresh_token , scope ) : <EOL> raise NotImplementedError ( '<STR_LIT>' '<STR_LIT>' ) <EOL> def persist_authorization_code ( self , client_id , code , scope ) : <EOL> raise NotImplementedError ( '<STR_LIT>' '<STR_LIT>' ) <EOL> def persist_token_information ( self , client_id , scope , access_token , <EOL> token_type , expires_in , refresh_token , <EOL> data ) : <EOL> raise NotImplementedError ( '<STR_LIT>' '<STR_LIT>' ) <EOL> def discard_authorization_code ( self , client_id , code ) : <EOL> raise NotImplementedError ( '<STR_LIT>' '<STR_LIT>' ) <EOL> def discard_refresh_token ( self , client_id , refresh_token ) : <EOL> raise NotImplementedError ( '<STR_LIT>' '<STR_LIT>' ) <EOL> class OAuthError ( Unauthorized ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , reason , * args , ** kwargs ) : <EOL> self . reason = reason <EOL> super ( OAuthError , self ) . __init__ ( * args , ** kwargs ) <EOL> class ResourceAuthorization ( object ) : <EOL> """<STR_LIT>""" <EOL> is_oauth = False <EOL> is_valid = None <EOL> token = None <EOL> client_id = None <EOL> expires_in = None <EOL> error = None <EOL> def raise_error_if_invalid ( self ) : <EOL> if not self . is_valid : <EOL> raise OAuthError ( self . error , '<STR_LIT>' ) <EOL> class ResourceProvider ( Provider ) : <EOL> """<STR_LIT>""" <EOL> @ property <EOL> def authorization_class ( self ) : <EOL> return ResourceAuthorization <EOL> def get_authorization ( self ) : <EOL> """<STR_LIT>""" <EOL> auth = self . authorization_class ( ) <EOL> header = self . get_authorization_header ( ) <EOL> if not header or not header . split : <EOL> return auth <EOL> header = header . split ( ) <EOL> if len ( header ) > <NUM_LIT:1> and header [ <NUM_LIT:0> ] == '<STR_LIT>' : <EOL> auth . is_oauth = True <EOL> access_token = header [ <NUM_LIT:1> ] <EOL> self . validate_access_token ( access_token , auth ) <EOL> if not auth . is_valid : <EOL> auth . error = '<STR_LIT>' <EOL> return auth <EOL> def get_authorization_header ( self ) : <EOL> raise NotImplementedError ( '<STR_LIT>' '<STR_LIT>' ) <EOL> def validate_access_token ( self , access_token , authorization ) : <EOL> raise NotImplementedError ( '<STR_LIT>' '<STR_LIT>' ) </s>
