<s> import datetime <EOL> from calendar import timegm <EOL> from jwt import InvalidTokenError , decode as jwt_decode <EOL> from openid . consumer . consumer import Consumer , SUCCESS , CANCEL , FAILURE <EOL> from openid . consumer . discover import DiscoveryFailure <EOL> from openid . extensions import sreg , ax , pape <EOL> from social . utils import url_add_parameters <EOL> from social . backends . base import BaseAuth <EOL> from social . backends . oauth import BaseOAuth2 <EOL> from social . exceptions import AuthException , AuthFailed , AuthCanceled , AuthUnknownError , AuthMissingParameter , AuthTokenError <EOL> OLD_AX_ATTRS = [ <EOL> ( '<STR_LIT>' , '<STR_LIT>' ) , <EOL> ( '<STR_LIT>' , '<STR_LIT>' ) , <EOL> ( '<STR_LIT>' , '<STR_LIT>' ) <EOL> ] <EOL> AX_SCHEMA_ATTRS = [ <EOL> ( '<STR_LIT>' , '<STR_LIT:email>' ) , <EOL> ( '<STR_LIT>' , '<STR_LIT>' ) , <EOL> ( '<STR_LIT>' , '<STR_LIT>' ) , <EOL> ( '<STR_LIT>' , '<STR_LIT>' ) , <EOL> ( '<STR_LIT>' , '<STR_LIT>' ) , <EOL> ] <EOL> SREG_ATTR = [ <EOL> ( '<STR_LIT:email>' , '<STR_LIT:email>' ) , <EOL> ( '<STR_LIT>' , '<STR_LIT>' ) , <EOL> ( '<STR_LIT>' , '<STR_LIT>' ) <EOL> ] <EOL> OPENID_ID_FIELD = '<STR_LIT>' <EOL> SESSION_NAME = '<STR_LIT>' <EOL> class OpenIdAuth ( BaseAuth ) : <EOL> """<STR_LIT>""" <EOL> name = '<STR_LIT>' <EOL> URL = None <EOL> USERNAME_KEY = '<STR_LIT:username>' <EOL> def get_user_id ( self , details , response ) : <EOL> """<STR_LIT>""" <EOL> return response . identity_url <EOL> def get_ax_attributes ( self ) : <EOL> attrs = self . setting ( '<STR_LIT>' , [ ] ) <EOL> if attrs and self . setting ( '<STR_LIT>' , True ) : <EOL> return attrs <EOL> return attrs + AX_SCHEMA_ATTRS + OLD_AX_ATTRS <EOL> def get_sreg_attributes ( self ) : <EOL> return self . setting ( '<STR_LIT>' ) or SREG_ATTR <EOL> def values_from_response ( self , response , sreg_names = None , ax_names = None ) : <EOL> """<STR_LIT>""" <EOL> values = { } <EOL> if sreg_names : <EOL> resp = sreg . SRegResponse . fromSuccessResponse ( response ) <EOL> if resp : <EOL> values . update ( ( alias , resp . get ( name ) or '<STR_LIT>' ) <EOL> for name , alias in sreg_names ) <EOL> if ax_names : <EOL> resp = ax . FetchResponse . fromSuccessResponse ( response ) <EOL> if resp : <EOL> for src , alias in ax_names : <EOL> name = alias . replace ( '<STR_LIT>' , '<STR_LIT>' ) <EOL> values [ name ] = resp . getSingle ( src , '<STR_LIT>' ) or values . get ( name ) <EOL> return values <EOL> def get_user_details ( self , response ) : <EOL> """<STR_LIT>""" <EOL> values = { '<STR_LIT:username>' : '<STR_LIT>' , '<STR_LIT:email>' : '<STR_LIT>' , '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , '<STR_LIT>' : '<STR_LIT>' } <EOL> values . update ( self . values_from_response ( <EOL> response , self . get_sreg_attributes ( ) , self . get_ax_attributes ( ) <EOL> ) ) <EOL> fullname = values . get ( '<STR_LIT>' ) or '<STR_LIT>' <EOL> first_name = values . get ( '<STR_LIT>' ) or '<STR_LIT>' <EOL> last_name = values . get ( '<STR_LIT>' ) or '<STR_LIT>' <EOL> email = values . get ( '<STR_LIT:email>' ) or '<STR_LIT>' <EOL> if not fullname and first_name and last_name : <EOL> fullname = first_name + '<STR_LIT:U+0020>' + last_name <EOL> elif fullname : <EOL> try : <EOL> first_name , last_name = fullname . rsplit ( '<STR_LIT:U+0020>' , <NUM_LIT:1> ) <EOL> except ValueError : <EOL> last_name = fullname <EOL> username_key = self . setting ( '<STR_LIT>' ) or self . USERNAME_KEY <EOL> values . update ( { '<STR_LIT>' : fullname , '<STR_LIT>' : first_name , <EOL> '<STR_LIT>' : last_name , <EOL> '<STR_LIT:username>' : values . get ( username_key ) or <EOL> ( first_name . title ( ) + last_name . title ( ) ) , <EOL> '<STR_LIT:email>' : email } ) <EOL> return values <EOL> def extra_data ( self , user , uid , response , details = None , * args , ** kwargs ) : <EOL> """<STR_LIT>""" <EOL> sreg_names = self . setting ( '<STR_LIT>' ) <EOL> ax_names = self . setting ( '<STR_LIT>' ) <EOL> values = self . values_from_response ( response , sreg_names , ax_names ) <EOL> from_details = super ( OpenIdAuth , self ) . extra_data ( <EOL> user , uid , { } , details , * args , ** kwargs <EOL> ) <EOL> values . update ( from_details ) <EOL> return values <EOL> def auth_url ( self ) : <EOL> """<STR_LIT>""" <EOL> openid_request = self . setup_request ( self . auth_extra_arguments ( ) ) <EOL> return_to = self . strategy . absolute_uri ( self . redirect_uri ) <EOL> return openid_request . redirectURL ( self . trust_root ( ) , return_to ) <EOL> def auth_html ( self ) : <EOL> """<STR_LIT>""" <EOL> openid_request = self . setup_request ( self . auth_extra_arguments ( ) ) <EOL> return_to = self . strategy . absolute_uri ( self . redirect_uri ) <EOL> form_tag = { '<STR_LIT:id>' : '<STR_LIT>' } <EOL> return openid_request . htmlMarkup ( self . trust_root ( ) , return_to , <EOL> form_tag_attrs = form_tag ) <EOL> def trust_root ( self ) : <EOL> """<STR_LIT>""" <EOL> return self . setting ( '<STR_LIT>' ) or self . strategy . absolute_uri ( '<STR_LIT:/>' ) <EOL> def continue_pipeline ( self , * args , ** kwargs ) : <EOL> """<STR_LIT>""" <EOL> response = self . consumer ( ) . complete ( dict ( self . data . items ( ) ) , <EOL> self . strategy . absolute_uri ( <EOL> self . redirect_uri <EOL> ) ) <EOL> kwargs . update ( { '<STR_LIT>' : response , '<STR_LIT>' : self } ) <EOL> return self . strategy . authenticate ( * args , ** kwargs ) <EOL> def auth_complete ( self , * args , ** kwargs ) : <EOL> """<STR_LIT>""" <EOL> response = self . consumer ( ) . complete ( dict ( self . data . items ( ) ) , <EOL> self . strategy . absolute_uri ( <EOL> self . redirect_uri <EOL> ) ) <EOL> self . process_error ( response ) <EOL> kwargs . update ( { '<STR_LIT>' : response , '<STR_LIT>' : self } ) <EOL> return self . strategy . authenticate ( * args , ** kwargs ) <EOL> def process_error ( self , data ) : <EOL> if not data : <EOL> raise AuthException ( self , '<STR_LIT>' ) <EOL> elif data . status == FAILURE : <EOL> raise AuthFailed ( self , data . message ) <EOL> elif data . status == CANCEL : <EOL> raise AuthCanceled ( self ) <EOL> elif data . status != SUCCESS : <EOL> raise AuthUnknownError ( self , data . status ) <EOL> def setup_request ( self , params = None ) : <EOL> """<STR_LIT>""" <EOL> request = self . openid_request ( params ) <EOL> if request . endpoint . supportsType ( ax . AXMessage . ns_uri ) : <EOL> fetch_request = ax . FetchRequest ( ) <EOL> for attr , alias in self . get_ax_attributes ( ) : <EOL> fetch_request . add ( ax . AttrInfo ( attr , alias = alias , <EOL> required = True ) ) <EOL> else : <EOL> fetch_request = sreg . SRegRequest ( <EOL> optional = list ( dict ( self . get_sreg_attributes ( ) ) . keys ( ) ) <EOL> ) <EOL> request . addExtension ( fetch_request ) <EOL> preferred_policies = self . setting ( <EOL> '<STR_LIT>' <EOL> ) <EOL> preferred_level_types = self . setting ( <EOL> '<STR_LIT>' <EOL> ) <EOL> max_age = self . setting ( '<STR_LIT>' ) <EOL> if max_age is not None : <EOL> try : <EOL> max_age = int ( max_age ) <EOL> except ( ValueError , TypeError ) : <EOL> max_age = None <EOL> if max_age is not None or preferred_policies or preferred_level_types : <EOL> pape_request = pape . Request ( <EOL> max_auth_age = max_age , <EOL> preferred_auth_policies = preferred_policies , <EOL> preferred_auth_level_types = preferred_level_types <EOL> ) <EOL> request . addExtension ( pape_request ) <EOL> return request <EOL> def consumer ( self ) : <EOL> """<STR_LIT>""" <EOL> if not hasattr ( self , '<STR_LIT>' ) : <EOL> self . _consumer = self . create_consumer ( self . strategy . openid_store ( ) ) <EOL> return self . _consumer <EOL> def create_consumer ( self , store = None ) : <EOL> return Consumer ( self . strategy . openid_session_dict ( SESSION_NAME ) , store ) <EOL> def uses_redirect ( self ) : <EOL> """<STR_LIT>""" <EOL> return self . openid_request ( ) . shouldSendRedirect ( ) <EOL> def openid_request ( self , params = None ) : <EOL> """<STR_LIT>""" <EOL> try : <EOL> return self . consumer ( ) . begin ( url_add_parameters ( self . openid_url ( ) , <EOL> params ) ) <EOL> except DiscoveryFailure as err : <EOL> raise AuthException ( self , '<STR_LIT>' . format ( <EOL> err <EOL> ) ) <EOL> def openid_url ( self ) : <EOL> """<STR_LIT>""" <EOL> if self . URL : <EOL> return self . URL <EOL> elif OPENID_ID_FIELD in self . data : <EOL> return self . data [ OPENID_ID_FIELD ] <EOL> else : <EOL> raise AuthMissingParameter ( self , OPENID_ID_FIELD ) <EOL> class OpenIdConnectAssociation ( object ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , handle , secret = '<STR_LIT>' , issued = <NUM_LIT:0> , lifetime = <NUM_LIT:0> , assoc_type = '<STR_LIT>' ) : <EOL> self . handle = handle <EOL> self . secret = secret . encode ( ) <EOL> self . issued = issued <EOL> self . lifetime = lifetime <EOL> self . assoc_type = assoc_type <EOL> class OpenIdConnectAuth ( BaseOAuth2 ) : <EOL> """<STR_LIT>""" <EOL> ID_TOKEN_ISSUER = None <EOL> ID_TOKEN_MAX_AGE = <NUM_LIT> <EOL> DEFAULT_SCOPE = [ '<STR_LIT>' ] <EOL> EXTRA_DATA = [ '<STR_LIT>' , '<STR_LIT>' , ( '<STR_LIT>' , '<STR_LIT:id>' ) ] <EOL> id_token = None <EOL> def auth_params ( self , state = None ) : <EOL> """<STR_LIT>""" <EOL> params = super ( OpenIdConnectAuth , self ) . auth_params ( state ) <EOL> params [ '<STR_LIT>' ] = self . get_and_store_nonce ( <EOL> self . AUTHORIZATION_URL , state <EOL> ) <EOL> return params <EOL> def auth_complete_params ( self , state = None ) : <EOL> params = super ( OpenIdConnectAuth , self ) . auth_complete_params ( state ) <EOL> params [ '<STR_LIT>' ] = self . get_and_store_nonce ( <EOL> self . ACCESS_TOKEN_URL , state <EOL> ) <EOL> return params <EOL> def get_and_store_nonce ( self , url , state ) : <EOL> nonce = self . strategy . random_string ( <NUM_LIT:64> ) <EOL> association = OpenIdConnectAssociation ( nonce , assoc_type = state ) <EOL> self . strategy . storage . association . store ( url , association ) <EOL> return nonce <EOL> def get_nonce ( self , nonce ) : <EOL> try : <EOL> return self . strategy . storage . association . get ( <EOL> server_url = self . ACCESS_TOKEN_URL , <EOL> handle = nonce <EOL> ) [ <NUM_LIT:0> ] <EOL> except IndexError : <EOL> pass <EOL> def remove_nonce ( self , nonce_id ) : <EOL> self . strategy . storage . association . remove ( [ nonce_id ] ) <EOL> def validate_and_return_id_token ( self , id_token ) : <EOL> """<STR_LIT>""" <EOL> client_id , _client_secret = self . get_key_and_secret ( ) <EOL> decode_kwargs = { <EOL> '<STR_LIT>' : [ '<STR_LIT>' ] , <EOL> '<STR_LIT>' : client_id , <EOL> '<STR_LIT>' : self . ID_TOKEN_ISSUER , <EOL> '<STR_LIT:key>' : self . setting ( '<STR_LIT>' ) , <EOL> '<STR_LIT>' : { <EOL> '<STR_LIT>' : True , <EOL> '<STR_LIT>' : True , <EOL> '<STR_LIT>' : True , <EOL> '<STR_LIT>' : True , <EOL> '<STR_LIT>' : True , <EOL> '<STR_LIT>' : True , <EOL> '<STR_LIT>' : True , <EOL> } , <EOL> } <EOL> decode_kwargs . update ( self . setting ( '<STR_LIT>' , { } ) ) <EOL> try : <EOL> id_token = jwt_decode ( id_token , ** decode_kwargs ) <EOL> except InvalidTokenError as err : <EOL> raise AuthTokenError ( self , err ) <EOL> iat_leeway = self . setting ( '<STR_LIT>' , self . ID_TOKEN_MAX_AGE ) <EOL> utc_timestamp = timegm ( datetime . datetime . utcnow ( ) . utctimetuple ( ) ) <EOL> if id_token [ '<STR_LIT>' ] < ( utc_timestamp - iat_leeway ) : <EOL> raise AuthTokenError ( self , '<STR_LIT>' ) <EOL> nonce = id_token . get ( '<STR_LIT>' ) <EOL> if not nonce : <EOL> raise AuthTokenError ( self , '<STR_LIT>' ) <EOL> nonce_obj = self . get_nonce ( nonce ) <EOL> if nonce_obj : <EOL> self . remove_nonce ( nonce_obj . id ) <EOL> else : <EOL> raise AuthTokenError ( self , '<STR_LIT>' ) <EOL> return id_token <EOL> def request_access_token ( self , * args , ** kwargs ) : <EOL> """<STR_LIT>""" <EOL> response = self . get_json ( * args , ** kwargs ) <EOL> self . id_token = self . validate_and_return_id_token ( response [ '<STR_LIT>' ] ) <EOL> return response </s>
