<s> import copy <EOL> import hashlib <EOL> from future . backports . urllib . parse import urlsplit <EOL> from oic import oic <EOL> from oic import rndstr <EOL> from oic . utils . http_util import Redirect <EOL> from oic . exception import MissingAttribute <EOL> from oic . oauth2 import ErrorResponse <EOL> from oic . oauth2 import TokenError <EOL> from oic . oauth2 import ResponseError <EOL> from oic . oic import ProviderConfigurationResponse <EOL> from oic . oic import AuthorizationResponse <EOL> from oic . oic import RegistrationResponse <EOL> from oic . oic import AuthorizationRequest <EOL> from oic . oic . message import OpenIDSchema <EOL> from oic . utils . authn . client import CLIENT_AUTHN_METHOD <EOL> __author__ = '<STR_LIT>' <EOL> import logging <EOL> logger = logging . getLogger ( __name__ ) <EOL> class OIDCError ( Exception ) : <EOL> pass <EOL> class Client ( oic . Client ) : <EOL> def __init__ ( self , client_id = None , ca_certs = None , <EOL> client_prefs = None , client_authn_method = None , keyjar = None , <EOL> verify_ssl = True , behaviour = None , config = None , jwks_uri = '<STR_LIT>' , <EOL> kid = None ) : <EOL> oic . Client . __init__ ( self , client_id , ca_certs , client_prefs , <EOL> client_authn_method , keyjar , verify_ssl , <EOL> config = config ) <EOL> if behaviour : <EOL> self . behaviour = behaviour <EOL> self . userinfo_request_method = '<STR_LIT>' <EOL> self . allow_sign_alg_none = False <EOL> self . authz_req = { } <EOL> self . get_userinfo = True <EOL> self . oidc = True <EOL> self . jwks_uri = jwks_uri <EOL> self . kid = kid <EOL> def create_authn_request ( self , session , acr_value = None , ** kwargs ) : <EOL> session [ "<STR_LIT:state>" ] = rndstr ( <NUM_LIT:32> ) <EOL> request_args = { <EOL> "<STR_LIT>" : self . behaviour [ "<STR_LIT>" ] , <EOL> "<STR_LIT>" : self . behaviour [ "<STR_LIT>" ] , <EOL> "<STR_LIT:state>" : session [ "<STR_LIT:state>" ] , <EOL> "<STR_LIT>" : self . registration_response [ "<STR_LIT>" ] [ <NUM_LIT:0> ] <EOL> } <EOL> if self . oidc : <EOL> session [ "<STR_LIT>" ] = rndstr ( <NUM_LIT:32> ) <EOL> request_args [ '<STR_LIT>' ] = session [ '<STR_LIT>' ] <EOL> if acr_value is not None : <EOL> request_args [ "<STR_LIT>" ] = acr_value <EOL> request_args . update ( kwargs ) <EOL> cis = self . construct_AuthorizationRequest ( request_args = request_args ) <EOL> logger . debug ( "<STR_LIT>" % cis ) <EOL> url , body , ht_args , cis = self . uri_and_body ( AuthorizationRequest , cis , <EOL> method = "<STR_LIT:GET>" , <EOL> request_args = request_args ) <EOL> self . authz_req [ request_args [ '<STR_LIT:state>' ] ] = cis <EOL> logger . debug ( "<STR_LIT>" % body ) <EOL> logger . info ( "<STR_LIT>" % url ) <EOL> logger . debug ( "<STR_LIT>" % ht_args ) <EOL> resp = Redirect ( str ( url ) ) <EOL> if ht_args : <EOL> resp . headers . extend ( [ ( a , b ) for a , b in ht_args . items ( ) ] ) <EOL> logger . debug ( "<STR_LIT>" % resp . headers ) <EOL> return resp <EOL> def has_access_token ( self , ** kwargs ) : <EOL> try : <EOL> token = self . get_token ( ** kwargs ) <EOL> except TokenError : <EOL> pass <EOL> else : <EOL> if token . access_token : <EOL> return True <EOL> return False <EOL> def _err ( self , txt ) : <EOL> logger . error ( txt ) <EOL> raise OIDCError ( txt ) <EOL> def callback ( self , response , session , format = '<STR_LIT>' ) : <EOL> """<STR_LIT>""" <EOL> try : <EOL> authresp = self . parse_response ( AuthorizationResponse , response , <EOL> sformat = format , keyjar = self . keyjar ) <EOL> except ResponseError : <EOL> logger . error ( "<STR_LIT>" . format ( response ) ) <EOL> raise OIDCError ( "<STR_LIT>" ) <EOL> logger . info ( "<STR_LIT>" . format ( authresp ) ) <EOL> if isinstance ( authresp , ErrorResponse ) : <EOL> if authresp [ "<STR_LIT:error>" ] == "<STR_LIT>" : <EOL> return self . create_authn_request ( session ) <EOL> else : <EOL> raise OIDCError ( "<STR_LIT>" ) <EOL> if session [ "<STR_LIT:state>" ] != authresp [ "<STR_LIT:state>" ] : <EOL> self . _err ( "<STR_LIT>" ) <EOL> try : <EOL> _id_token = authresp [ '<STR_LIT>' ] <EOL> except KeyError : <EOL> _id_token = None <EOL> else : <EOL> if _id_token [ '<STR_LIT>' ] != session [ "<STR_LIT>" ] : <EOL> self . _err ( "<STR_LIT>" ) <EOL> if self . behaviour [ "<STR_LIT>" ] == "<STR_LIT:code>" : <EOL> try : <EOL> args = { <EOL> "<STR_LIT:code>" : authresp [ "<STR_LIT:code>" ] , <EOL> "<STR_LIT>" : self . registration_response [ <EOL> "<STR_LIT>" ] [ <NUM_LIT:0> ] , <EOL> "<STR_LIT>" : self . client_id , <EOL> "<STR_LIT>" : self . client_secret , <EOL> } <EOL> try : <EOL> args [ '<STR_LIT>' ] = response [ '<STR_LIT>' ] <EOL> except KeyError : <EOL> pass <EOL> atresp = self . do_access_token_request ( <EOL> state = authresp [ "<STR_LIT:state>" ] , <EOL> request_args = args , <EOL> authn_method = self . registration_response [ <EOL> "<STR_LIT>" ] ) <EOL> logger . info ( '<STR_LIT>' . format ( atresp ) ) <EOL> except Exception as err : <EOL> logger . error ( "<STR_LIT:%s>" % err ) <EOL> raise <EOL> if isinstance ( atresp , ErrorResponse ) : <EOL> self . _err ( '<STR_LIT>' . format ( atresp . to_dict ( ) ) ) <EOL> _token = atresp [ '<STR_LIT>' ] <EOL> try : <EOL> _id_token = atresp [ '<STR_LIT>' ] <EOL> except KeyError : <EOL> pass <EOL> else : <EOL> _token = authresp [ '<STR_LIT>' ] <EOL> if not self . oidc : <EOL> return { '<STR_LIT>' : _token } <EOL> if _id_token is None : <EOL> self . _err ( "<STR_LIT>" ) <EOL> if _id_token [ '<STR_LIT>' ] != self . provider_info [ '<STR_LIT>' ] : <EOL> self . _err ( "<STR_LIT>" ) <EOL> if _id_token [ '<STR_LIT>' ] != session [ '<STR_LIT>' ] : <EOL> self . _err ( "<STR_LIT>" ) <EOL> if not self . allow_sign_alg_none : <EOL> if _id_token . jws_header [ '<STR_LIT>' ] == '<STR_LIT:none>' : <EOL> self . _err ( '<STR_LIT>' ) <EOL> user_id = '<STR_LIT>' . format ( _id_token [ '<STR_LIT>' ] , _id_token [ '<STR_LIT>' ] ) <EOL> if self . get_userinfo : <EOL> if self . userinfo_request_method : <EOL> kwargs = { "<STR_LIT>" : self . userinfo_request_method } <EOL> else : <EOL> kwargs = { } <EOL> if self . has_access_token ( state = authresp [ "<STR_LIT:state>" ] ) : <EOL> inforesp = self . do_user_info_request ( state = authresp [ "<STR_LIT:state>" ] , <EOL> ** kwargs ) <EOL> if isinstance ( inforesp , ErrorResponse ) : <EOL> self . _err ( "<STR_LIT>" % inforesp [ "<STR_LIT:error>" ] ) <EOL> userinfo = inforesp . to_dict ( ) <EOL> if _id_token [ '<STR_LIT>' ] != userinfo [ '<STR_LIT>' ] : <EOL> self . _err ( "<STR_LIT>" ) <EOL> logger . debug ( "<STR_LIT>" % inforesp ) <EOL> try : <EOL> self . id_token [ user_id ] = _id_token <EOL> except TypeError : <EOL> self . id_token = { user_id : _id_token } <EOL> else : <EOL> userinfo = { } <EOL> for attr in OpenIDSchema . c_param : <EOL> try : <EOL> userinfo [ attr ] = _id_token [ attr ] <EOL> except KeyError : <EOL> pass <EOL> return { '<STR_LIT>' : user_id , '<STR_LIT>' : userinfo , <EOL> '<STR_LIT>' : _id_token , '<STR_LIT>' : _token } <EOL> else : <EOL> return { '<STR_LIT>' : user_id , '<STR_LIT>' : _id_token , <EOL> '<STR_LIT>' : _token } <EOL> class OIDCClients ( object ) : <EOL> def __init__ ( self , config , base_url , seed = '<STR_LIT>' , jwks_info = None , <EOL> verify_ssl = True ) : <EOL> """<STR_LIT>""" <EOL> self . client = { } <EOL> self . client_cls = Client <EOL> self . config = config <EOL> self . seed = seed or rndstr ( <NUM_LIT:16> ) <EOL> self . seed = self . seed . encode ( '<STR_LIT:utf8>' ) <EOL> self . path = { } <EOL> self . base_url = base_url <EOL> self . jwks_info = jwks_info <EOL> self . verify_ssl = verify_ssl <EOL> for key , val in config . CLIENTS . items ( ) : <EOL> if self . jwks_info : <EOL> _val = val . copy ( ) <EOL> _val . update ( self . jwks_info ) <EOL> else : <EOL> _val = val <EOL> if key == "<STR_LIT>" : <EOL> continue <EOL> else : <EOL> self . client [ key ] = self . create_client ( ** _val ) <EOL> def get_path ( self , redirect_uris , issuer ) : <EOL> for ruri in redirect_uris : <EOL> p = urlsplit ( ruri ) <EOL> self . path [ p . path [ <NUM_LIT:1> : ] ] = issuer <EOL> def create_client ( self , userid = "<STR_LIT>" , ** kwargs ) : <EOL> """<STR_LIT>""" <EOL> _key_set = set ( list ( kwargs . keys ( ) ) ) <EOL> try : <EOL> _verify_ssl = kwargs [ '<STR_LIT>' ] <EOL> except KeyError : <EOL> _verify_ssl = self . verify_ssl <EOL> else : <EOL> _key_set . discard ( '<STR_LIT>' ) <EOL> client = self . client_cls ( client_authn_method = CLIENT_AUTHN_METHOD , <EOL> behaviour = kwargs [ "<STR_LIT>" ] , <EOL> verify_ssl = _verify_ssl ) <EOL> try : <EOL> client . userinfo_request_method = kwargs [ "<STR_LIT>" ] <EOL> except KeyError : <EOL> pass <EOL> else : <EOL> _key_set . discard ( "<STR_LIT>" ) <EOL> _key_set . discard ( "<STR_LIT>" ) <EOL> for param in [ "<STR_LIT>" ] : <EOL> try : <EOL> setattr ( client , param , kwargs [ param ] ) <EOL> except KeyError : <EOL> pass <EOL> else : <EOL> _key_set . discard ( param ) <EOL> if _key_set == { "<STR_LIT>" } : <EOL> if not userid : <EOL> raise MissingAttribute ( "<STR_LIT>" ) <EOL> issuer = client . wf . discovery_query ( userid ) <EOL> pcr = client . provider_config ( issuer ) <EOL> client . register ( client . provider_info [ "<STR_LIT>" ] , <EOL> ** kwargs [ "<STR_LIT>" ] ) <EOL> self . get_path ( kwargs [ '<STR_LIT>' ] [ '<STR_LIT>' ] , issuer ) <EOL> elif _key_set == set ( [ "<STR_LIT>" , "<STR_LIT>" ] ) : <EOL> _ = client . provider_config ( kwargs [ "<STR_LIT>" ] ) <EOL> _ = client . register ( client . provider_info [ "<STR_LIT>" ] , <EOL> ** kwargs [ "<STR_LIT>" ] ) <EOL> self . get_path ( kwargs [ '<STR_LIT>' ] [ '<STR_LIT>' ] , <EOL> kwargs [ "<STR_LIT>" ] ) <EOL> elif _key_set == set ( [ "<STR_LIT>" , "<STR_LIT>" ] ) : <EOL> client . handle_provider_config ( <EOL> ProviderConfigurationResponse ( ** kwargs [ "<STR_LIT>" ] ) , <EOL> kwargs [ "<STR_LIT>" ] [ "<STR_LIT>" ] ) <EOL> _ = client . register ( client . provider_info [ "<STR_LIT>" ] , <EOL> ** kwargs [ "<STR_LIT>" ] ) <EOL> self . get_path ( kwargs [ '<STR_LIT>' ] [ '<STR_LIT>' ] , <EOL> kwargs [ "<STR_LIT>" ] [ "<STR_LIT>" ] ) <EOL> elif _key_set == set ( [ "<STR_LIT>" , "<STR_LIT>" ] ) : <EOL> client . handle_provider_config ( <EOL> ProviderConfigurationResponse ( ** kwargs [ "<STR_LIT>" ] ) , <EOL> kwargs [ "<STR_LIT>" ] [ "<STR_LIT>" ] ) <EOL> client . store_registration_info ( RegistrationResponse ( <EOL> ** kwargs [ "<STR_LIT>" ] ) ) <EOL> self . get_path ( kwargs [ '<STR_LIT>' ] [ '<STR_LIT>' ] , <EOL> kwargs [ "<STR_LIT>" ] [ "<STR_LIT>" ] ) <EOL> elif _key_set == set ( [ "<STR_LIT>" , "<STR_LIT>" ] ) : <EOL> _ = client . provider_config ( kwargs [ "<STR_LIT>" ] ) <EOL> client . store_registration_info ( RegistrationResponse ( <EOL> ** kwargs [ "<STR_LIT>" ] ) ) <EOL> self . get_path ( kwargs [ '<STR_LIT>' ] [ '<STR_LIT>' ] , <EOL> kwargs [ "<STR_LIT>" ] ) <EOL> else : <EOL> raise Exception ( "<STR_LIT>" ) <EOL> return client <EOL> def dynamic_client ( self , userid = '<STR_LIT>' , issuer = '<STR_LIT>' ) : <EOL> client = self . client_cls ( client_authn_method = CLIENT_AUTHN_METHOD , <EOL> verify_ssl = self . verify_ssl , ** self . jwks_info ) <EOL> if userid : <EOL> issuer = client . wf . discovery_query ( userid ) <EOL> if not issuer : <EOL> raise OIDCError ( '<STR_LIT>' ) <EOL> logger . info ( '<STR_LIT>' . format ( issuer ) ) <EOL> if issuer in self . client : <EOL> return self . client [ issuer ] <EOL> else : <EOL> _pcr = client . provider_config ( issuer ) <EOL> logger . info ( '<STR_LIT>' . format ( _pcr . to_dict ) ) <EOL> _cinfo = self . config . CLIENTS [ "<STR_LIT>" ] [ "<STR_LIT>" ] <EOL> reg_args = copy . copy ( _cinfo ) <EOL> h = hashlib . sha256 ( self . seed ) <EOL> h . update ( issuer . encode ( '<STR_LIT:utf8>' ) ) <EOL> base_urls = _cinfo [ "<STR_LIT>" ] <EOL> reg_args [ '<STR_LIT>' ] = [ <EOL> u . format ( base = self . base_url , iss = h . hexdigest ( ) ) <EOL> for u in base_urls ] <EOL> try : <EOL> reg_args [ '<STR_LIT>' ] = [ <EOL> u . format ( base = self . base_url , iss = h . hexdigest ( ) ) <EOL> for u in reg_args [ '<STR_LIT>' ] <EOL> ] <EOL> except KeyError : <EOL> pass <EOL> self . get_path ( reg_args [ '<STR_LIT>' ] , issuer ) <EOL> if client . jwks_uri : <EOL> reg_args [ '<STR_LIT>' ] = client . jwks_uri <EOL> rr = client . register ( _pcr [ "<STR_LIT>" ] , ** reg_args ) <EOL> logger . info ( '<STR_LIT>' . format ( rr . to_dict ( ) ) ) <EOL> try : <EOL> client . behaviour . update ( ** self . config . CLIENTS [ "<STR_LIT>" ] [ "<STR_LIT>" ] ) <EOL> except KeyError : <EOL> pass <EOL> self . client [ issuer ] = client <EOL> return client <EOL> def __getitem__ ( self , item ) : <EOL> """<STR_LIT>""" <EOL> try : <EOL> return self . client [ item ] <EOL> except KeyError : <EOL> return self . dynamic_client ( issuer = item ) <EOL> def keys ( self ) : <EOL> return list ( self . client . keys ( ) ) <EOL> def return_paths ( self ) : <EOL> return self . path . keys ( ) </s>
