<s> import hashlib <EOL> import logging <EOL> import os <EOL> import sys <EOL> import traceback <EOL> from future . backports . urllib . parse import unquote <EOL> from future . backports . urllib . parse import urljoin <EOL> from future . backports . urllib . parse import urlparse <EOL> from future . backports . urllib . parse import splitquery <EOL> from future . backports . urllib . parse import parse_qs <EOL> from oic import rndstr <EOL> from oic . exception import FailedAuthentication <EOL> from oic . exception import InvalidRequest <EOL> from oic . exception import MissingParameter <EOL> from oic . exception import ParameterError <EOL> from oic . exception import RedirectURIError <EOL> from oic . exception import UnknownClient <EOL> from oic . exception import URIError <EOL> from oic . oauth2 import Server <EOL> from oic . oauth2 . message import AccessTokenRequest <EOL> from oic . oauth2 . message import AccessTokenResponse <EOL> from oic . oauth2 . message import AuthorizationErrorResponse <EOL> from oic . oauth2 . message import AuthorizationRequest <EOL> from oic . oauth2 . message import AuthorizationResponse <EOL> from oic . oauth2 . message import ErrorResponse <EOL> from oic . oauth2 . message import MissingRequiredAttribute <EOL> from oic . oauth2 . message import MissingRequiredValue <EOL> from oic . oauth2 . message import NoneResponse <EOL> from oic . oauth2 . message import TokenErrorResponse <EOL> from oic . oauth2 . message import add_non_standard <EOL> from oic . oauth2 . message import by_schema <EOL> from oic . utils . authn . user import NoSuchAuthentication <EOL> from oic . utils . authn . user import TamperAllert <EOL> from oic . utils . authn . user import ToOld <EOL> from oic . utils . http_util import BadRequest <EOL> from oic . utils . http_util import CookieDealer <EOL> from oic . utils . http_util import SeeOther <EOL> from oic . utils . http_util import Response <EOL> from oic . utils . http_util import make_cookie <EOL> from oic . utils . sdb import AccessCodeUsed <EOL> from oic . utils . sdb import AuthnEvent <EOL> __author__ = '<STR_LIT>' <EOL> logger = logging . getLogger ( __name__ ) <EOL> LOG_INFO = logger . info <EOL> LOG_DEBUG = logger . debug <EOL> class Endpoint ( object ) : <EOL> """<STR_LIT>""" <EOL> etype = "<STR_LIT>" <EOL> url = "<STR_LIT>" <EOL> def __init__ ( self , func = None ) : <EOL> self . func = func <EOL> def __call__ ( self , * args , ** kwargs ) : <EOL> return self . func ( * args , ** kwargs ) <EOL> class AuthorizationEndpoint ( Endpoint ) : <EOL> etype = "<STR_LIT>" <EOL> url = "<STR_LIT>" <EOL> class TokenEndpoint ( Endpoint ) : <EOL> etype = "<STR_LIT>" <EOL> url = "<STR_LIT>" <EOL> def endpoint_ava ( endp , baseurl ) : <EOL> key = '<STR_LIT>' . format ( endp . etype ) <EOL> val = urljoin ( baseurl , endp . url ) <EOL> return { key : val } <EOL> def code_response ( ** kwargs ) : <EOL> aresp = AuthorizationResponse ( ) <EOL> _areq = kwargs [ "<STR_LIT>" ] <EOL> try : <EOL> aresp [ "<STR_LIT:state>" ] = _areq [ "<STR_LIT:state>" ] <EOL> except KeyError : <EOL> pass <EOL> aresp [ "<STR_LIT:code>" ] = kwargs [ "<STR_LIT>" ] <EOL> if kwargs [ '<STR_LIT>' ] : <EOL> aresp [ '<STR_LIT>' ] = kwargs [ '<STR_LIT>' ] <EOL> aresp [ '<STR_LIT>' ] = _areq [ '<STR_LIT>' ] <EOL> add_non_standard ( _areq , aresp ) <EOL> return aresp <EOL> def token_response ( ** kwargs ) : <EOL> _areq = kwargs [ "<STR_LIT>" ] <EOL> _scode = kwargs [ "<STR_LIT>" ] <EOL> _sdb = kwargs [ "<STR_LIT>" ] <EOL> _dic = _sdb . upgrade_to_token ( _scode , issue_refresh = False ) <EOL> aresp = AccessTokenResponse ( ** by_schema ( AccessTokenResponse , ** _dic ) ) <EOL> try : <EOL> aresp [ "<STR_LIT:state>" ] = _areq [ "<STR_LIT:state>" ] <EOL> except KeyError : <EOL> pass <EOL> add_non_standard ( _areq , aresp ) <EOL> return aresp <EOL> def none_response ( ** kwargs ) : <EOL> _areq = kwargs [ "<STR_LIT>" ] <EOL> aresp = NoneResponse ( ) <EOL> if "<STR_LIT:state>" in _areq : <EOL> aresp [ "<STR_LIT:state>" ] = _areq [ "<STR_LIT:state>" ] <EOL> return aresp <EOL> def location_url ( response_type , redirect_uri , query ) : <EOL> if response_type in [ [ "<STR_LIT:code>" ] , [ "<STR_LIT>" ] , [ "<STR_LIT:none>" ] ] : <EOL> return "<STR_LIT>" % ( redirect_uri , query ) <EOL> else : <EOL> return "<STR_LIT>" % ( redirect_uri , query ) <EOL> def max_age ( areq ) : <EOL> try : <EOL> return areq [ "<STR_LIT>" ] [ "<STR_LIT>" ] <EOL> except KeyError : <EOL> try : <EOL> return areq [ "<STR_LIT>" ] <EOL> except KeyError : <EOL> return <NUM_LIT:0> <EOL> def re_authenticate ( areq , authn ) : <EOL> if "<STR_LIT>" in areq and "<STR_LIT>" in areq [ "<STR_LIT>" ] : <EOL> if authn . done ( areq ) : <EOL> return True <EOL> return False <EOL> class Provider ( object ) : <EOL> endp = [ AuthorizationEndpoint , TokenEndpoint ] <EOL> def __init__ ( self , name , sdb , cdb , authn_broker , authz , client_authn , <EOL> symkey = "<STR_LIT>" , urlmap = None , iv = <NUM_LIT:0> , default_scope = "<STR_LIT>" , <EOL> ca_bundle = None , verify_ssl = True , default_acr = "<STR_LIT>" , <EOL> baseurl = '<STR_LIT>' ) : <EOL> self . name = name <EOL> self . sdb = sdb <EOL> self . cdb = cdb <EOL> self . server = Server ( ca_certs = ca_bundle , verify_ssl = verify_ssl ) <EOL> self . authn_broker = authn_broker <EOL> if authn_broker is None : <EOL> self . cookie_func = CookieDealer ( srv = self ) . create_cookie <EOL> else : <EOL> self . cookie_func = self . authn_broker [ <NUM_LIT:0> ] [ <NUM_LIT:0> ] . create_cookie <EOL> for item in self . authn_broker : <EOL> item . srv = self <EOL> self . authz = authz <EOL> self . client_authn = client_authn <EOL> self . symkey = symkey <EOL> self . seed = rndstr ( ) . encode ( "<STR_LIT:utf-8>" ) <EOL> self . iv = iv or os . urandom ( <NUM_LIT:16> ) <EOL> self . cookie_name = "<STR_LIT>" <EOL> self . default_scope = default_scope <EOL> self . sso_ttl = <NUM_LIT:0> <EOL> self . default_acr = default_acr <EOL> if urlmap is None : <EOL> self . urlmap = { } <EOL> else : <EOL> self . urlmap = urlmap <EOL> self . response_type_map = { <EOL> "<STR_LIT:code>" : code_response , <EOL> "<STR_LIT>" : token_response , <EOL> "<STR_LIT:none>" : none_response , <EOL> } <EOL> self . session_cookie_name = "<STR_LIT>" <EOL> self . baseurl = baseurl <EOL> self . keyjar = None <EOL> self . trace = None <EOL> self . events = None <EOL> @ staticmethod <EOL> def input ( query = "<STR_LIT>" , post = None ) : <EOL> if query : <EOL> return query <EOL> elif post : <EOL> return post <EOL> else : <EOL> raise MissingParameter ( "<STR_LIT>" ) <EOL> @ staticmethod <EOL> def _error_response ( error , descr = None ) : <EOL> logger . error ( "<STR_LIT:%s>" % error ) <EOL> response = ErrorResponse ( error = error , error_description = descr ) <EOL> return Response ( response . to_json ( ) , content = "<STR_LIT:application/json>" , <EOL> status = "<STR_LIT>" ) <EOL> @ staticmethod <EOL> def _error ( error , descr = None ) : <EOL> response = ErrorResponse ( error = error , error_description = descr ) <EOL> return Response ( response . to_json ( ) , content = "<STR_LIT:application/json>" , <EOL> status = "<STR_LIT>" ) <EOL> @ staticmethod <EOL> def _authz_error ( error , descr = None ) : <EOL> response = AuthorizationErrorResponse ( error = error ) <EOL> if descr : <EOL> response [ "<STR_LIT>" ] = descr <EOL> return Response ( response . to_json ( ) , content = "<STR_LIT:application/json>" , <EOL> status = "<STR_LIT>" ) <EOL> @ staticmethod <EOL> def _redirect_authz_error ( error , redirect_uri , descr = None , state = "<STR_LIT>" , <EOL> return_type = None ) : <EOL> err = AuthorizationErrorResponse ( error = error ) <EOL> if descr : <EOL> err [ "<STR_LIT>" ] = descr <EOL> if state : <EOL> err [ "<STR_LIT:state>" ] = state <EOL> if return_type is None or return_type == [ "<STR_LIT:code>" ] : <EOL> location = err . request ( redirect_uri ) <EOL> else : <EOL> location = err . request ( redirect_uri , True ) <EOL> return SeeOther ( location ) <EOL> def _verify_redirect_uri ( self , areq ) : <EOL> """<STR_LIT>""" <EOL> try : <EOL> _redirect_uri = unquote ( areq [ "<STR_LIT>" ] ) <EOL> part = urlparse ( _redirect_uri ) <EOL> if part . fragment : <EOL> raise URIError ( "<STR_LIT>" ) <EOL> ( _base , _query ) = splitquery ( _redirect_uri ) <EOL> if _query : <EOL> _query = parse_qs ( _query ) <EOL> match = False <EOL> for regbase , rquery in self . cdb [ str ( areq [ "<STR_LIT>" ] ) ] [ <EOL> "<STR_LIT>" ] : <EOL> if _base == regbase or _redirect_uri . startswith ( regbase ) : <EOL> if rquery : <EOL> for key , vals in rquery . items ( ) : <EOL> assert key in _query <EOL> for val in vals : <EOL> assert val in _query [ key ] <EOL> if _query : <EOL> if rquery is None : <EOL> raise ValueError <EOL> for key , vals in _query . items ( ) : <EOL> assert key in rquery <EOL> for val in vals : <EOL> assert val in rquery [ key ] <EOL> match = True <EOL> break <EOL> if not match : <EOL> raise RedirectURIError ( "<STR_LIT>" ) <EOL> return None <EOL> except Exception as err : <EOL> logger . error ( "<STR_LIT>" % areq [ "<STR_LIT>" ] ) <EOL> try : <EOL> _cinfo = self . cdb [ str ( areq [ "<STR_LIT>" ] ) ] <EOL> except KeyError : <EOL> try : <EOL> cid = areq [ "<STR_LIT>" ] <EOL> except KeyError : <EOL> logger . error ( '<STR_LIT>' ) <EOL> raise UnknownClient ( '<STR_LIT>' ) <EOL> else : <EOL> logger . info ( "<STR_LIT>" % cid ) <EOL> raise UnknownClient ( areq [ "<STR_LIT>" ] ) <EOL> else : <EOL> logger . info ( "<STR_LIT>" % _cinfo ) <EOL> raise RedirectURIError ( <EOL> "<STR_LIT>" % areq [ "<STR_LIT>" ] ) <EOL> def get_redirect_uri ( self , areq ) : <EOL> """<STR_LIT>""" <EOL> if '<STR_LIT>' in areq : <EOL> self . _verify_redirect_uri ( areq ) <EOL> uri = areq [ "<STR_LIT>" ] <EOL> else : <EOL> raise ParameterError ( <EOL> "<STR_LIT>" ) <EOL> return uri <EOL> def pick_auth ( self , areq , comparision_type = "<STR_LIT>" ) : <EOL> """<STR_LIT>""" <EOL> if comparision_type == "<STR_LIT>" : <EOL> return self . authn_broker [ <NUM_LIT:0> ] <EOL> try : <EOL> if len ( self . authn_broker ) == <NUM_LIT:1> : <EOL> return self . authn_broker [ <NUM_LIT:0> ] <EOL> elif "<STR_LIT>" in areq : <EOL> if not comparision_type : <EOL> comparision_type = "<STR_LIT>" <EOL> if not isinstance ( areq [ "<STR_LIT>" ] , list ) : <EOL> areq [ "<STR_LIT>" ] = [ areq [ "<STR_LIT>" ] ] <EOL> for acr in areq [ "<STR_LIT>" ] : <EOL> res = self . authn_broker . pick ( acr , comparision_type ) <EOL> logger . debug ( "<STR_LIT>" % ( <EOL> str ( acr ) , str ( res ) ) ) <EOL> if res : <EOL> return res [ <NUM_LIT:0> ] <EOL> else : <EOL> try : <EOL> acrs = areq [ "<STR_LIT>" ] [ "<STR_LIT>" ] [ "<STR_LIT>" ] [ "<STR_LIT>" ] <EOL> except KeyError : <EOL> return self . authn_broker [ <NUM_LIT:0> ] <EOL> else : <EOL> for acr in acrs : <EOL> res = self . authn_broker . pick ( acr , comparision_type ) <EOL> logger . debug ( "<STR_LIT>" % ( <EOL> str ( acr ) , str ( res ) ) ) <EOL> if res : <EOL> return res [ <NUM_LIT:0> ] <EOL> except KeyError as exc : <EOL> logger . debug ( <EOL> "<STR_LIT>" % str ( <EOL> exc ) ) <EOL> return None , None <EOL> def filter_request ( self , req ) : <EOL> return req <EOL> def auth_init ( self , request , request_class = AuthorizationRequest ) : <EOL> """<STR_LIT>""" <EOL> logger . debug ( "<STR_LIT>" % request ) <EOL> try : <EOL> areq = self . server . parse_authorization_request ( <EOL> request = request_class , query = request ) <EOL> except ( MissingRequiredValue , MissingRequiredAttribute ) as err : <EOL> logger . debug ( "<STR_LIT:%s>" % err ) <EOL> areq = request_class ( ) . deserialize ( request , "<STR_LIT>" ) <EOL> try : <EOL> redirect_uri = self . get_redirect_uri ( areq ) <EOL> except ( RedirectURIError , ParameterError ) as err : <EOL> return self . _error ( "<STR_LIT>" , "<STR_LIT:%s>" % err ) <EOL> try : <EOL> _rtype = areq [ "<STR_LIT>" ] <EOL> except : <EOL> _rtype = [ "<STR_LIT:code>" ] <EOL> return self . _redirect_authz_error ( "<STR_LIT>" , redirect_uri , <EOL> "<STR_LIT:%s>" % err , areq [ "<STR_LIT:state>" ] , <EOL> _rtype ) <EOL> except KeyError : <EOL> areq = request_class ( ) . deserialize ( request , "<STR_LIT>" ) <EOL> try : <EOL> self . get_redirect_uri ( areq ) <EOL> except ( RedirectURIError , ParameterError ) as err : <EOL> return self . _error ( "<STR_LIT>" , "<STR_LIT:%s>" % err ) <EOL> except Exception as err : <EOL> message = traceback . format_exception ( * sys . exc_info ( ) ) <EOL> logger . error ( message ) <EOL> logger . debug ( "<STR_LIT>" % ( err , err . __class__ . __name__ ) ) <EOL> return BadRequest ( "<STR_LIT:%s>" % err ) <EOL> if not areq : <EOL> logger . debug ( "<STR_LIT>" ) <EOL> return self . _error ( "<STR_LIT>" , "<STR_LIT>" ) <EOL> areq = self . filter_request ( areq ) <EOL> if self . events : <EOL> self . events . store ( '<STR_LIT>' , areq ) <EOL> if self . trace : <EOL> self . trace . info ( '<STR_LIT:{}>' . format ( areq . to_dict ( ) ) ) <EOL> if '<STR_LIT>' in areq : <EOL> if areq [ '<STR_LIT>' ] . jwe_header : <EOL> self . trace . info ( '<STR_LIT>' . format ( <EOL> areq [ '<STR_LIT>' ] . jwe_header ) ) <EOL> if areq [ '<STR_LIT>' ] . jws_header : <EOL> self . trace . info ( '<STR_LIT>' . format ( <EOL> areq [ '<STR_LIT>' ] . jws_header ) ) <EOL> try : <EOL> _cinfo = self . cdb [ areq [ '<STR_LIT>' ] ] <EOL> except KeyError : <EOL> return self . _error ( '<STR_LIT>' , '<STR_LIT>' ) <EOL> else : <EOL> try : <EOL> rtypes = _cinfo [ '<STR_LIT>' ] <EOL> except KeyError : <EOL> rtypes = [ '<STR_LIT:code>' ] <EOL> if '<STR_LIT:U+0020>' . join ( areq [ "<STR_LIT>" ] ) not in rtypes : <EOL> return self . _error ( "<STR_LIT>" , <EOL> "<STR_LIT>" ) <EOL> logger . debug ( "<STR_LIT>" % ( areq . to_dict ( ) , ) ) <EOL> try : <EOL> redirect_uri = self . get_redirect_uri ( areq ) <EOL> except ( RedirectURIError , ParameterError , UnknownClient ) as err : <EOL> return self . _error ( "<STR_LIT>" , "<STR_LIT:%s>" % err ) <EOL> try : <EOL> keyjar = self . keyjar <EOL> except AttributeError : <EOL> keyjar = "<STR_LIT>" <EOL> try : <EOL> areq . verify ( keyjar = keyjar , opponent_id = areq [ "<STR_LIT>" ] ) <EOL> except ( MissingRequiredAttribute , ValueError ) as err : <EOL> return self . _redirect_authz_error ( "<STR_LIT>" , redirect_uri , <EOL> "<STR_LIT:%s>" % err ) <EOL> return { "<STR_LIT>" : areq , "<STR_LIT>" : redirect_uri } <EOL> @ staticmethod <EOL> def _acr_claims ( areq ) : <EOL> try : <EOL> acrdef = areq [ "<STR_LIT>" ] [ "<STR_LIT>" ] [ "<STR_LIT>" ] <EOL> except KeyError : <EOL> return None <EOL> else : <EOL> if isinstance ( acrdef , dict ) : <EOL> try : <EOL> return [ acrdef [ "<STR_LIT:value>" ] ] <EOL> except KeyError : <EOL> try : <EOL> return acrdef [ "<STR_LIT>" ] <EOL> except KeyError : <EOL> pass <EOL> return None <EOL> def do_auth ( self , areq , redirect_uri , cinfo , request , cookie , ** kwargs ) : <EOL> """<STR_LIT>""" <EOL> acrs = self . _acr_claims ( areq ) <EOL> if acrs : <EOL> tup = ( None , None ) <EOL> for acr in acrs : <EOL> res = self . authn_broker . pick ( acr , "<STR_LIT>" ) <EOL> logger . debug ( "<STR_LIT>" % ( <EOL> str ( acr ) , str ( res ) ) ) <EOL> if res : <EOL> tup = res [ <NUM_LIT:0> ] <EOL> break <EOL> authn , authn_class_ref = tup <EOL> else : <EOL> authn , authn_class_ref = self . pick_auth ( areq ) <EOL> if not authn : <EOL> authn , authn_class_ref = self . pick_auth ( areq , "<STR_LIT>" ) <EOL> if not authn : <EOL> authn , authn_class_ref = self . pick_auth ( areq , "<STR_LIT>" ) <EOL> if authn is None : <EOL> return self . _redirect_authz_error ( "<STR_LIT>" , redirect_uri , <EOL> return_type = areq [ "<STR_LIT>" ] ) <EOL> try : <EOL> try : <EOL> _auth_info = kwargs [ "<STR_LIT>" ] <EOL> except KeyError : <EOL> _auth_info = "<STR_LIT>" <EOL> if "<STR_LIT>" in areq and areq [ "<STR_LIT>" ] == "<STR_LIT:true>" : <EOL> _max_age = <NUM_LIT:0> <EOL> else : <EOL> _max_age = max_age ( areq ) <EOL> identity , _ts = authn . authenticated_as ( <EOL> cookie , authorization = _auth_info , max_age = _max_age ) <EOL> except ( NoSuchAuthentication , TamperAllert ) : <EOL> identity = None <EOL> _ts = <NUM_LIT:0> <EOL> except ToOld : <EOL> logger . info ( "<STR_LIT>" ) <EOL> identity = None <EOL> _ts = <NUM_LIT:0> <EOL> else : <EOL> logger . info ( "<STR_LIT>" ) <EOL> authn_args = { "<STR_LIT>" : request , <EOL> "<STR_LIT>" : authn_class_ref } <EOL> if "<STR_LIT>" in kwargs : <EOL> authn_args [ "<STR_LIT>" ] = kwargs [ "<STR_LIT>" ] , <EOL> for attr in [ "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT>" ] : <EOL> try : <EOL> authn_args [ attr ] = cinfo [ attr ] <EOL> except KeyError : <EOL> pass <EOL> for attr in [ "<STR_LIT>" , "<STR_LIT>" ] : <EOL> try : <EOL> authn_args [ attr ] = areq [ attr ] <EOL> except KeyError : <EOL> pass <EOL> if identity is None : <EOL> if "<STR_LIT>" in areq and "<STR_LIT:none>" in areq [ "<STR_LIT>" ] : <EOL> return self . _redirect_authz_error ( <EOL> "<STR_LIT>" , redirect_uri , <EOL> return_type = areq [ "<STR_LIT>" ] ) <EOL> else : <EOL> return authn ( ** authn_args ) <EOL> else : <EOL> if re_authenticate ( areq , authn ) : <EOL> return authn ( ** authn_args ) <EOL> else : <EOL> user = identity [ "<STR_LIT>" ] <EOL> if "<STR_LIT>" in kwargs : <EOL> sids_for_sub = self . sdb . get_sids_by_sub ( kwargs [ "<STR_LIT>" ] ) <EOL> if sids_for_sub and user != self . sdb . get_authentication_event ( <EOL> sids_for_sub [ - <NUM_LIT:1> ] ) . uid : <EOL> logger . debug ( "<STR_LIT>" ) <EOL> if "<STR_LIT>" in areq and "<STR_LIT:none>" in areq [ "<STR_LIT>" ] : <EOL> return self . _redirect_authz_error ( "<STR_LIT>" , <EOL> redirect_uri ) <EOL> else : <EOL> return authn ( ** authn_args ) <EOL> authn_event = AuthnEvent ( identity [ "<STR_LIT>" ] , identity . get ( '<STR_LIT>' , '<STR_LIT>' ) , <EOL> authn_info = authn_class_ref , <EOL> time_stamp = _ts ) <EOL> return { "<STR_LIT>" : authn_event , "<STR_LIT>" : identity , "<STR_LIT:user>" : user } <EOL> def setup_session ( self , areq , authn_event , cinfo ) : <EOL> sid = self . sdb . create_authz_session ( authn_event , areq ) <EOL> self . sdb . do_sub ( sid , '<STR_LIT>' ) <EOL> return sid <EOL> def authorization_endpoint ( self , request = "<STR_LIT>" , cookie = "<STR_LIT>" , ** kwargs ) : <EOL> """<STR_LIT>""" <EOL> info = self . auth_init ( request ) <EOL> if isinstance ( info , Response ) : <EOL> return info <EOL> _cid = info [ "<STR_LIT>" ] [ "<STR_LIT>" ] <EOL> cinfo = self . cdb [ _cid ] <EOL> authnres = self . do_auth ( info [ "<STR_LIT>" ] , info [ "<STR_LIT>" ] , <EOL> cinfo , request , cookie , ** kwargs ) <EOL> if isinstance ( authnres , Response ) : <EOL> return authnres <EOL> logger . debug ( "<STR_LIT>" ) <EOL> logger . debug ( "<STR_LIT>" % info [ "<STR_LIT>" ] . keys ( ) ) <EOL> sid = self . setup_session ( info [ "<STR_LIT>" ] , authnres [ "<STR_LIT>" ] , <EOL> cinfo ) <EOL> return self . authz_part2 ( authnres [ "<STR_LIT:user>" ] , info [ "<STR_LIT>" ] , sid , <EOL> cookie = cookie ) <EOL> def aresp_check ( self , aresp , areq ) : <EOL> return "<STR_LIT>" <EOL> def create_authn_response ( self , areq , sid ) : <EOL> rtype = areq [ "<STR_LIT>" ] [ <NUM_LIT:0> ] <EOL> _func = self . response_type_map [ rtype ] <EOL> aresp = _func ( areq = areq , scode = self . sdb [ sid ] [ "<STR_LIT:code>" ] , sdb = self . sdb , <EOL> myself = self . baseurl ) <EOL> if rtype == "<STR_LIT:code>" : <EOL> fragment_enc = False <EOL> else : <EOL> fragment_enc = True <EOL> return aresp , fragment_enc <EOL> def response_mode ( self , areq , fragment_enc , ** kwargs ) : <EOL> resp_mode = areq [ "<STR_LIT>" ] <EOL> if resp_mode == '<STR_LIT>' and not fragment_enc : <EOL> raise InvalidRequest ( "<STR_LIT>" ) <EOL> elif resp_mode == '<STR_LIT>' and fragment_enc : <EOL> return InvalidRequest ( "<STR_LIT>" ) <EOL> return None <EOL> def authz_part2 ( self , user , areq , sid , ** kwargs ) : <EOL> """<STR_LIT>""" <EOL> result = self . _complete_authz ( user , areq , sid , ** kwargs ) <EOL> if isinstance ( result , Response ) : <EOL> return result <EOL> else : <EOL> aresp , headers , redirect_uri , fragment_enc = result <EOL> aresp [ '<STR_LIT>' ] = self . baseurl <EOL> aresp [ '<STR_LIT>' ] = areq [ '<STR_LIT>' ] <EOL> location = aresp . request ( redirect_uri , fragment_enc ) <EOL> logger . debug ( "<STR_LIT>" % ( location , type ( location ) ) ) <EOL> return SeeOther ( str ( location ) , headers = headers ) <EOL> def _complete_authz ( self , user , areq , sid , ** kwargs ) : <EOL> _log_debug = logger . debug <EOL> _log_debug ( "<STR_LIT>" ) <EOL> try : <EOL> permission = self . authz ( user , client_id = areq [ '<STR_LIT>' ] ) <EOL> self . sdb . update ( sid , "<STR_LIT>" , permission ) <EOL> except Exception : <EOL> raise <EOL> _log_debug ( "<STR_LIT>" % areq [ "<STR_LIT>" ] ) <EOL> if self . sdb . is_revoked ( sid ) : <EOL> return self . _error ( error = "<STR_LIT>" , <EOL> descr = "<STR_LIT>" ) <EOL> info = self . create_authn_response ( areq , sid ) <EOL> if isinstance ( info , Response ) : <EOL> return info <EOL> else : <EOL> aresp , fragment_enc = info <EOL> try : <EOL> redirect_uri = self . get_redirect_uri ( areq ) <EOL> except ( RedirectURIError , ParameterError ) as err : <EOL> return BadRequest ( "<STR_LIT:%s>" % err ) <EOL> info = self . aresp_check ( aresp , areq ) <EOL> if isinstance ( info , Response ) : <EOL> return info <EOL> headers = [ ] <EOL> try : <EOL> _kaka = kwargs [ "<STR_LIT>" ] <EOL> except KeyError : <EOL> pass <EOL> else : <EOL> if _kaka and self . cookie_name not in _kaka : <EOL> headers . append ( <EOL> self . cookie_func ( user , typ = "<STR_LIT>" , ttl = self . sso_ttl ) ) <EOL> if "<STR_LIT>" in areq : <EOL> try : <EOL> resp = self . response_mode ( areq , fragment_enc , aresp = aresp , <EOL> redirect_uri = redirect_uri , <EOL> headers = headers ) <EOL> except InvalidRequest as err : <EOL> return self . _error ( "<STR_LIT>" , err ) <EOL> else : <EOL> if resp is not None : <EOL> return resp <EOL> return aresp , headers , redirect_uri , fragment_enc <EOL> def token_scope_check ( self , areq , info ) : <EOL> """<STR_LIT>""" <EOL> return None <EOL> def token_endpoint ( self , authn = "<STR_LIT>" , ** kwargs ) : <EOL> """<STR_LIT>""" <EOL> _sdb = self . sdb <EOL> LOG_DEBUG ( "<STR_LIT>" ) <EOL> body = kwargs [ "<STR_LIT>" ] <EOL> LOG_DEBUG ( "<STR_LIT>" % body ) <EOL> areq = AccessTokenRequest ( ) . deserialize ( body , "<STR_LIT>" ) <EOL> try : <EOL> client = self . client_authn ( self , areq , authn ) <EOL> except FailedAuthentication as err : <EOL> err = TokenErrorResponse ( error = "<STR_LIT>" , <EOL> error_description = "<STR_LIT:%s>" % err ) <EOL> return Response ( err . to_json ( ) , content = "<STR_LIT:application/json>" , <EOL> status = "<STR_LIT>" ) <EOL> LOG_DEBUG ( "<STR_LIT>" % areq ) <EOL> try : <EOL> assert areq [ "<STR_LIT>" ] == "<STR_LIT>" <EOL> except AssertionError : <EOL> err = TokenErrorResponse ( error = "<STR_LIT>" , <EOL> error_description = "<STR_LIT>" ) <EOL> return Response ( err . to_json ( ) , content = "<STR_LIT:application/json>" , <EOL> status = "<STR_LIT>" ) <EOL> _info = _sdb [ areq [ "<STR_LIT:code>" ] ] <EOL> resp = self . token_scope_check ( areq , _info ) <EOL> if resp : <EOL> return resp <EOL> if "<STR_LIT>" in _info : <EOL> assert areq [ "<STR_LIT>" ] == _info [ "<STR_LIT>" ] <EOL> try : <EOL> _tinfo = _sdb . upgrade_to_token ( areq [ "<STR_LIT:code>" ] , issue_refresh = True ) <EOL> except AccessCodeUsed : <EOL> err = TokenErrorResponse ( error = "<STR_LIT>" , <EOL> error_description = "<STR_LIT>" ) <EOL> return Response ( err . to_json ( ) , content = "<STR_LIT:application/json>" , <EOL> status = "<STR_LIT>" ) <EOL> LOG_DEBUG ( "<STR_LIT>" % _tinfo ) <EOL> atr = AccessTokenResponse ( ** by_schema ( AccessTokenResponse , ** _tinfo ) ) <EOL> LOG_DEBUG ( "<STR_LIT>" % atr ) <EOL> return Response ( atr . to_json ( ) , content = "<STR_LIT:application/json>" ) <EOL> def verify_endpoint ( self , request = "<STR_LIT>" , cookie = None , ** kwargs ) : <EOL> _req = parse_qs ( request ) <EOL> try : <EOL> areq = parse_qs ( _req [ "<STR_LIT>" ] [ <NUM_LIT:0> ] ) <EOL> except KeyError : <EOL> return BadRequest ( ) <EOL> authn , acr = self . pick_auth ( areq = areq ) <EOL> kwargs [ "<STR_LIT>" ] = cookie <EOL> return authn . verify ( _req , ** kwargs ) <EOL> def write_session_cookie ( self , value ) : <EOL> return make_cookie ( self . session_cookie_name , value , self . seed , path = "<STR_LIT:/>" ) <EOL> def delete_session_cookie ( self ) : <EOL> return make_cookie ( self . session_cookie_name , "<STR_LIT>" , b"<STR_LIT>" , path = "<STR_LIT:/>" , <EOL> expire = - <NUM_LIT:1> ) <EOL> def _compute_session_state ( self , state , salt , client_id , redirect_uri ) : <EOL> parsed_uri = urlparse ( redirect_uri ) <EOL> rp_origin_url = "<STR_LIT>" . format ( uri = parsed_uri ) <EOL> session_str = client_id + "<STR_LIT:U+0020>" + rp_origin_url + "<STR_LIT:U+0020>" + state + "<STR_LIT:U+0020>" + salt <EOL> return hashlib . sha256 ( <EOL> session_str . encode ( "<STR_LIT:utf-8>" ) ) . hexdigest ( ) + "<STR_LIT:.>" + salt </s>
