<s> import logging <EOL> import traceback <EOL> import sys <EOL> import requests <EOL> from oic . utils . http_util import Response <EOL> from oic . utils . authn . client import CLIENT_AUTHN_METHOD <EOL> from oic import oic , rndstr <EOL> from oic . oauth2 import PBase <EOL> from oic . oauth2 . message import ErrorResponse <EOL> from oic . oic . message import AuthorizationResponse <EOL> from oic . oic . message import AuthorizationRequest <EOL> from oic . oic . message import AccessTokenResponse <EOL> from oic . utils . webfinger import WebFinger <EOL> __author__ = '<STR_LIT>' <EOL> logger = logging . getLogger ( __name__ ) <EOL> def token_secret_key ( sid ) : <EOL> return "<STR_LIT>" % sid <EOL> SERVICE_NAME = "<STR_LIT>" <EOL> FLOW_TYPE = "<STR_LIT:code>" <EOL> CLIENT_CONFIG = { } <EOL> class OpenIDConnect ( object ) : <EOL> def __init__ ( self , attribute_map = None , authenticating_authority = None , <EOL> name = "<STR_LIT>" , registration_info = None , ** kwargs ) : <EOL> self . attribute_map = attribute_map <EOL> self . authenticating_authority = authenticating_authority <EOL> self . name = name <EOL> self . client_id = "<STR_LIT>" <EOL> self . client_secret = "<STR_LIT>" <EOL> for param in [ "<STR_LIT>" , "<STR_LIT>" ] : <EOL> try : <EOL> setattr ( self , param , kwargs [ param ] ) <EOL> del kwargs [ param ] <EOL> except KeyError : <EOL> pass <EOL> self . extra = kwargs <EOL> try : <EOL> self . srv_discovery_url = kwargs [ "<STR_LIT>" ] <EOL> except KeyError : <EOL> self . srv_discovery_url = None <EOL> self . flow_type = FLOW_TYPE <EOL> self . access_token_response = AccessTokenResponse <EOL> self . client_cls = oic . Client <EOL> self . authn_method = None <EOL> self . registration_info = registration_info <EOL> def dynamic ( self , server_env , callback , logout_callback , session , key ) : <EOL> try : <EOL> client = server_env [ "<STR_LIT>" ] [ key ] <EOL> except KeyError : <EOL> client = self . client_cls ( client_authn_method = CLIENT_AUTHN_METHOD ) <EOL> client . redirect_uris = [ callback ] <EOL> client . post_logout_redirect_uris = [ logout_callback ] <EOL> _me = self . registration_info . copy ( ) <EOL> _me [ "<STR_LIT>" ] = [ callback ] <EOL> provider_conf = client . provider_config ( self . srv_discovery_url ) <EOL> logger . debug ( "<STR_LIT>" , provider_conf ) <EOL> session [ '<STR_LIT>' ] = provider_conf [ "<STR_LIT>" ] <EOL> logger . debug ( "<STR_LIT>" ) <EOL> reg_info = client . register ( provider_conf [ "<STR_LIT>" ] , <EOL> ** _me ) <EOL> logger . debug ( "<STR_LIT>" , reg_info ) <EOL> for prop in [ "<STR_LIT>" , "<STR_LIT>" ] : <EOL> try : <EOL> setattr ( client , prop , reg_info [ prop ] ) <EOL> except KeyError : <EOL> pass <EOL> try : <EOL> server_env [ "<STR_LIT>" ] [ key ] = client <EOL> except KeyError : <EOL> server_env [ "<STR_LIT>" ] = { key : client } <EOL> return client <EOL> def static ( self , server_env , callback , logout_callback , key ) : <EOL> try : <EOL> client = server_env [ "<STR_LIT>" ] [ key ] <EOL> logger . debug ( "<STR_LIT>" , server_env [ "<STR_LIT>" ] ) <EOL> except KeyError : <EOL> client = self . client_cls ( client_authn_method = CLIENT_AUTHN_METHOD ) <EOL> client . redirect_uris = [ callback ] <EOL> client . post_logout_redirect_uris = [ logout_callback ] <EOL> for typ in [ "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT>" ] : <EOL> endpoint = "<STR_LIT>" % typ <EOL> setattr ( client , endpoint , self . extra [ endpoint ] ) <EOL> client . client_id = self . client_id <EOL> client . client_secret = self . client_secret <EOL> if "<STR_LIT>" in self . extra : <EOL> client . keyjar . add ( self . extra [ "<STR_LIT>" ] [ <NUM_LIT:0> ] , self . extra [ "<STR_LIT>" ] [ <NUM_LIT:1> ] ) <EOL> try : <EOL> server_env [ "<STR_LIT>" ] [ key ] = client <EOL> except KeyError : <EOL> server_env [ "<STR_LIT>" ] = { key : client } <EOL> return client <EOL> def begin ( self , environ , server_env , start_response , session , key ) : <EOL> """<STR_LIT>""" <EOL> try : <EOL> logger . debug ( "<STR_LIT>" , self . flow_type ) <EOL> logger . debug ( "<STR_LIT>" , server_env ) <EOL> client = session [ '<STR_LIT>' ] <EOL> if client is not None and self . srv_discovery_url : <EOL> data = { "<STR_LIT>" : client . client_id } <EOL> resp = requests . get ( self . srv_discovery_url + "<STR_LIT>" , <EOL> params = data , verify = False ) <EOL> if not resp . ok and resp . status_code == <NUM_LIT> : <EOL> client = None <EOL> server_env [ "<STR_LIT>" ] . pop ( key , None ) <EOL> _state = "<STR_LIT>" <EOL> if client is None : <EOL> callback = server_env [ "<STR_LIT>" ] + key <EOL> logout_callback = server_env [ "<STR_LIT>" ] <EOL> if self . srv_discovery_url : <EOL> client = self . dynamic ( server_env , callback , logout_callback , <EOL> session , key ) <EOL> else : <EOL> client = self . static ( server_env , callback , logout_callback , <EOL> key ) <EOL> _state = session [ '<STR_LIT:state>' ] <EOL> session [ '<STR_LIT>' ] = client <EOL> acr_value = session . get_acr_value ( client . authorization_endpoint ) <EOL> try : <EOL> acr_values = client . provider_info [ "<STR_LIT>" ] <EOL> session [ '<STR_LIT>' ] = acr_values <EOL> except KeyError : <EOL> acr_values = None <EOL> if acr_value is None and acr_values is not None and len ( acr_values ) > <NUM_LIT:1> : <EOL> resp_headers = [ ( "<STR_LIT>" , str ( "<STR_LIT>" ) ) ] <EOL> start_response ( "<STR_LIT>" , resp_headers ) <EOL> return [ ] <EOL> elif acr_values is not None and len ( acr_values ) == <NUM_LIT:1> : <EOL> acr_value = acr_values [ <NUM_LIT:0> ] <EOL> return self . create_authnrequest ( environ , server_env , start_response , <EOL> session , acr_value , _state ) <EOL> except Exception : <EOL> message = traceback . format_exception ( * sys . exc_info ( ) ) <EOL> logger . error ( message ) <EOL> return self . result ( <EOL> environ , start_response , server_env , <EOL> ( False , "<STR_LIT>" ) ) <EOL> def create_authnrequest ( self , environ , server_env , start_response , session , <EOL> acr_value , state ) : <EOL> try : <EOL> client = session [ '<STR_LIT>' ] <EOL> session . set_acr_value ( client . authorization_endpoint , acr_value ) <EOL> request_args = { <EOL> "<STR_LIT>" : self . flow_type , <EOL> "<STR_LIT>" : server_env [ "<STR_LIT>" ] , <EOL> "<STR_LIT:state>" : state , <EOL> } <EOL> if acr_value is not None : <EOL> request_args [ "<STR_LIT>" ] = acr_value <EOL> if self . flow_type == "<STR_LIT>" : <EOL> request_args [ "<STR_LIT>" ] = rndstr ( <NUM_LIT:16> ) <EOL> session [ '<STR_LIT>' ] = request_args [ "<STR_LIT>" ] <EOL> else : <EOL> use_nonce = getattr ( self , "<STR_LIT>" , None ) <EOL> if use_nonce : <EOL> request_args [ "<STR_LIT>" ] = rndstr ( <NUM_LIT:16> ) <EOL> session [ '<STR_LIT>' ] = request_args [ "<STR_LIT>" ] <EOL> logger . info ( "<STR_LIT>" , list ( client . __dict__ . items ( ) ) ) <EOL> logger . info ( "<STR_LIT>" , request_args ) <EOL> except Exception : <EOL> message = traceback . format_exception ( * sys . exc_info ( ) ) <EOL> logger . error ( message ) <EOL> return self . result ( <EOL> environ , start_response , server_env , <EOL> ( False , "<STR_LIT>" ) ) <EOL> try : <EOL> cis = client . construct_AuthorizationRequest ( <EOL> request_args = request_args ) <EOL> logger . debug ( "<STR_LIT>" , cis ) <EOL> url , body , ht_args , cis = client . uri_and_body ( <EOL> AuthorizationRequest , cis , method = "<STR_LIT:GET>" , <EOL> request_args = request_args ) <EOL> logger . debug ( "<STR_LIT>" , body ) <EOL> except Exception : <EOL> message = traceback . format_exception ( * sys . exc_info ( ) ) <EOL> logger . error ( message ) <EOL> return self . result ( environ , start_response , server_env , ( <EOL> False , "<STR_LIT>" ) ) <EOL> logger . info ( "<STR_LIT>" , url ) <EOL> logger . debug ( "<STR_LIT>" , ht_args ) <EOL> session [ '<STR_LIT>' ] = client <EOL> resp_headers = [ ( "<STR_LIT>" , 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> start_response ( "<STR_LIT>" , resp_headers ) <EOL> return [ ] <EOL> def get_accesstoken ( self , client , authresp ) : <EOL> if self . srv_discovery_url : <EOL> issuer = list ( client . provider_info . keys ( ) ) [ <NUM_LIT:0> ] <EOL> key = client . keyjar . get_verify_key ( owner = issuer ) <EOL> kwargs = { "<STR_LIT:key>" : key } <EOL> logger . debug ( "<STR_LIT>" , key ) <EOL> else : <EOL> kwargs = { "<STR_LIT>" : client . keyjar } <EOL> if self . authn_method : <EOL> kwargs [ "<STR_LIT>" ] = self . authn_method <EOL> return client . do_access_token_request ( <EOL> state = authresp [ "<STR_LIT:state>" ] , response_cls = self . access_token_response , <EOL> ** kwargs ) <EOL> def verify_token ( self , client , access_token ) : <EOL> return { } <EOL> def get_userinfo ( self , client , authresp , access_token , ** kwargs ) : <EOL> return client . do_user_info_request ( state = authresp [ "<STR_LIT:state>" ] , <EOL> schema = "<STR_LIT>" , <EOL> access_token = access_token , <EOL> ** kwargs ) <EOL> def phaseN ( self , environ , query , server_env , session ) : <EOL> """<STR_LIT>""" <EOL> client = session [ '<STR_LIT>' ] <EOL> logger . debug ( "<STR_LIT>" , query ) <EOL> logger . debug ( "<STR_LIT>" , client . keyjar ) <EOL> authresp = client . parse_response ( AuthorizationResponse , query , <EOL> sformat = "<STR_LIT>" , keyjar = client . keyjar ) <EOL> if isinstance ( authresp , ErrorResponse ) : <EOL> return False , "<STR_LIT>" <EOL> try : <EOL> client . id_token = authresp [ "<STR_LIT>" ] <EOL> except : <EOL> pass <EOL> logger . debug ( "<STR_LIT>" , environ ) <EOL> if self . flow_type == "<STR_LIT:code>" : <EOL> try : <EOL> tokenresp = self . get_accesstoken ( client , authresp ) <EOL> except Exception as err : <EOL> logger . error ( "<STR_LIT:%s>" , err ) <EOL> raise <EOL> if isinstance ( tokenresp , ErrorResponse ) : <EOL> return ( False , "<STR_LIT>" % tokenresp [ "<STR_LIT:error>" ] ) <EOL> access_token = tokenresp [ "<STR_LIT>" ] <EOL> else : <EOL> access_token = authresp [ "<STR_LIT>" ] <EOL> userinfo = self . verify_token ( client , access_token ) <EOL> inforesp = self . get_userinfo ( client , authresp , access_token ) <EOL> if isinstance ( inforesp , ErrorResponse ) : <EOL> return False , "<STR_LIT>" % inforesp [ "<STR_LIT:error>" ] , session <EOL> tot_info = userinfo . update ( inforesp . to_dict ( ) ) <EOL> logger . debug ( "<STR_LIT>" , inforesp ) <EOL> return True , userinfo , access_token , client <EOL> def callback ( self , environ , server_env , start_response , query , session ) : <EOL> """<STR_LIT>""" <EOL> _service = self . __class__ . __name__ <EOL> logger . debug ( "<STR_LIT>" , _service , environ ) <EOL> logger . debug ( "<STR_LIT>" , _service , query ) <EOL> try : <EOL> result = self . phaseN ( environ , query , server_env , session ) <EOL> session [ '<STR_LIT>' ] = True <EOL> logger . debug ( "<STR_LIT>" , _service , result ) <EOL> except Exception : <EOL> message = traceback . format_exception ( * sys . exc_info ( ) ) <EOL> logger . error ( message ) <EOL> result = ( False , "<STR_LIT>" ) <EOL> return self . result ( environ , start_response , server_env , result ) <EOL> def result ( self , environ , start_response , server_env , result ) : <EOL> resp = Response ( mako_template = "<STR_LIT>" , <EOL> template_lookup = server_env [ "<STR_LIT>" ] , <EOL> headers = [ ] ) <EOL> argv = { <EOL> "<STR_LIT:result>" : result <EOL> } <EOL> return resp ( environ , start_response , ** argv ) <EOL> def find_srv_discovery_url ( self , resource ) : <EOL> """<STR_LIT>""" <EOL> wf = WebFinger ( httpd = PBase ( ca_certs = self . extra [ "<STR_LIT>" ] ) ) <EOL> return wf . discovery_query ( resource ) </s>
