<s> from functools import cmp_to_key <EOL> import json <EOL> import logging <EOL> import os <EOL> import traceback <EOL> import sys <EOL> import six <EOL> import socket <EOL> from future . backports . urllib . parse import parse_qs <EOL> from future . backports . urllib . parse import splitquery <EOL> from jwkest import jws , b64e <EOL> from oic import rndstr <EOL> from oic . exception import FailedAuthentication <EOL> from oic . exception import ModificationForbidden <EOL> from oic . exception import RestrictionError <EOL> from oic . exception import UnSupported <EOL> from oic . exception import UnknownAssertionType <EOL> from oic . extension . message import ClientInfoResponse <EOL> from oic . extension . message import ClientRegistrationError <EOL> from oic . extension . message import ServerMetadata <EOL> from oic . extension . client import CC_METHOD <EOL> from oic . extension . message import ClientUpdateRequest <EOL> from oic . extension . message import RegistrationRequest <EOL> from oic . extension . message import InvalidRedirectUri <EOL> from oic . extension . message import MissingPage <EOL> from oic . extension . message import TokenRevocationRequest <EOL> from oic . extension . message import TokenIntrospectionRequest <EOL> from oic . extension . message import TokenIntrospectionResponse <EOL> from oic . oauth2 import provider <EOL> from oic . oauth2 import AccessTokenRequest <EOL> from oic . oauth2 import TokenErrorResponse <EOL> from oic . oauth2 import AccessTokenResponse <EOL> from oic . oauth2 import by_schema <EOL> from oic . oauth2 . provider import Endpoint <EOL> from oic . oauth2 . exception import VerificationError <EOL> from oic . oauth2 . exception import CapabilitiesMisMatch <EOL> from oic . oauth2 . message import ASConfigurationResponse <EOL> from oic . oic import PREFERENCE2PROVIDER <EOL> from oic . oic . provider import RegistrationEndpoint <EOL> from oic . oic . provider import STR <EOL> from oic . oic . provider import secret <EOL> from oic . utils import restrict <EOL> from oic . utils import sort_sign_alg <EOL> from oic . utils . authn . client import AuthnFailure <EOL> from oic . utils . authn . client import UnknownAuthnMethod <EOL> from oic . utils . authn . client import get_client_id <EOL> from oic . utils . http_util import Unauthorized <EOL> from oic . utils . http_util import NoContent <EOL> from oic . utils . http_util import Response <EOL> from oic . utils . http_util import BadRequest <EOL> from oic . utils . http_util import Forbidden <EOL> from oic . utils . keyio import KeyBundle <EOL> from oic . utils . keyio import KeyJar <EOL> from oic . utils . keyio import key_export <EOL> from oic . utils . sdb import AccessCodeUsed <EOL> from oic . utils . time_util import utc_time_sans_frac <EOL> from oic . utils . token_handler import TokenHandler , NotAllowed <EOL> __author__ = '<STR_LIT>' <EOL> logger = logging . getLogger ( __name__ ) <EOL> CAPABILITIES = { <EOL> "<STR_LIT>" : [ "<STR_LIT:code>" , "<STR_LIT>" ] , <EOL> "<STR_LIT>" : [ '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' ] , <EOL> "<STR_LIT>" : [ <EOL> "<STR_LIT>" , "<STR_LIT>" , <EOL> "<STR_LIT>" ] , <EOL> } <EOL> AUTH_METHODS_SUPPORTED = [ "<STR_LIT>" , "<STR_LIT>" , <EOL> "<STR_LIT>" , "<STR_LIT>" ] <EOL> class ClientInfoEndpoint ( Endpoint ) : <EOL> etype = "<STR_LIT>" <EOL> url = '<STR_LIT>' <EOL> class RevocationEndpoint ( Endpoint ) : <EOL> etype = "<STR_LIT>" <EOL> url = '<STR_LIT>' <EOL> class IntrospectionEndpoint ( Endpoint ) : <EOL> etype = "<STR_LIT>" <EOL> url = '<STR_LIT>' <EOL> class Provider ( provider . Provider ) : <EOL> """<STR_LIT>""" <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 , seed = b"<STR_LIT>" , client_authn_methods = None , <EOL> authn_at_registration = "<STR_LIT>" , client_info_url = "<STR_LIT>" , <EOL> secret_lifetime = <NUM_LIT> , jwks_uri = '<STR_LIT>' , keyjar = None , <EOL> capabilities = None , verify_ssl = True , baseurl = '<STR_LIT>' , hostname = '<STR_LIT>' , <EOL> config = None , behavior = None , lifetime_policy = None , ** kwargs ) : <EOL> if not name . endswith ( "<STR_LIT:/>" ) : <EOL> name += "<STR_LIT:/>" <EOL> provider . Provider . __init__ ( self , name , sdb , cdb , authn_broker , authz , <EOL> client_authn , symkey , urlmap , iv , <EOL> default_scope , ca_bundle ) <EOL> self . endp . extend ( [ RegistrationEndpoint , ClientInfoEndpoint , <EOL> RevocationEndpoint , IntrospectionEndpoint ] ) <EOL> self . client_authn_methods = client_authn_methods <EOL> if authn_at_registration : <EOL> if authn_at_registration not in client_authn_methods : <EOL> raise UnknownAuthnMethod ( authn_at_registration ) <EOL> self . authn_at_registration = authn_at_registration <EOL> self . seed = seed <EOL> self . client_info_url = client_info_url <EOL> self . secret_lifetime = secret_lifetime <EOL> self . jwks_uri = jwks_uri <EOL> self . verify_ssl = verify_ssl <EOL> try : <EOL> self . scopes = kwargs [ '<STR_LIT>' ] <EOL> except KeyError : <EOL> self . scopes = [ '<STR_LIT>' ] <EOL> self . keyjar = keyjar <EOL> if self . keyjar is None : <EOL> self . keyjar = KeyJar ( verify_ssl = self . verify_ssl ) <EOL> if capabilities : <EOL> self . capabilities = self . provider_features ( <EOL> provider_config = capabilities ) <EOL> else : <EOL> self . capabilities = self . provider_features ( ) <EOL> self . baseurl = baseurl <EOL> self . hostname = hostname or socket . gethostname ( ) <EOL> self . kid = { "<STR_LIT>" : { } , "<STR_LIT>" : { } } <EOL> self . config = config or { } <EOL> self . behavior = behavior or { } <EOL> self . token_policy = { '<STR_LIT>' : { } , '<STR_LIT>' : { } } <EOL> if lifetime_policy is None : <EOL> self . lifetime_policy = { <EOL> '<STR_LIT>' : { <EOL> '<STR_LIT:code>' : <NUM_LIT> , <EOL> '<STR_LIT>' : <NUM_LIT> , <EOL> '<STR_LIT>' : <NUM_LIT> , <EOL> '<STR_LIT>' : <NUM_LIT> , <EOL> '<STR_LIT>' : <NUM_LIT> , <EOL> '<STR_LIT:password>' : <NUM_LIT> <EOL> } , <EOL> '<STR_LIT>' : { <EOL> '<STR_LIT:code>' : <NUM_LIT> , <EOL> '<STR_LIT>' : <NUM_LIT> , <EOL> '<STR_LIT>' : <NUM_LIT> , <EOL> '<STR_LIT>' : <NUM_LIT> , <EOL> '<STR_LIT>' : <NUM_LIT> , <EOL> '<STR_LIT:password>' : <NUM_LIT> <EOL> } <EOL> } <EOL> else : <EOL> self . lifetime_policy = lifetime_policy <EOL> self . token_handler = TokenHandler ( self . baseurl , self . token_policy , <EOL> keyjar = self . keyjar ) <EOL> @ staticmethod <EOL> def _uris_to_tuples ( uris ) : <EOL> tup = [ ] <EOL> for uri in uris : <EOL> base , query = splitquery ( uri ) <EOL> if query : <EOL> tup . append ( ( base , query ) ) <EOL> else : <EOL> tup . append ( ( base , "<STR_LIT>" ) ) <EOL> return tup <EOL> @ staticmethod <EOL> def _tuples_to_uris ( items ) : <EOL> _uri = [ ] <EOL> for url , query in items : <EOL> if query : <EOL> _uri . append ( "<STR_LIT>" % ( url , query ) ) <EOL> else : <EOL> _uri . append ( url ) <EOL> return _uri <EOL> def load_keys ( self , request , client_id , client_secret ) : <EOL> try : <EOL> self . keyjar . load_keys ( request , client_id ) <EOL> try : <EOL> logger . debug ( "<STR_LIT>" % ( <EOL> client_id , <EOL> "<STR_LIT:U+002C>" . join ( [ "<STR_LIT:%s>" % x for x in self . keyjar [ client_id ] ] ) ) ) <EOL> except KeyError : <EOL> pass <EOL> except Exception as err : <EOL> logger . error ( "<STR_LIT>" % request . to_dict ( ) ) <EOL> logger . error ( "<STR_LIT:%s>" , err ) <EOL> err = ClientRegistrationError ( <EOL> 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> _kc = KeyBundle ( [ { "<STR_LIT>" : "<STR_LIT>" , "<STR_LIT:key>" : client_secret , <EOL> "<STR_LIT>" : "<STR_LIT>" } , <EOL> { "<STR_LIT>" : "<STR_LIT>" , "<STR_LIT:key>" : client_secret , <EOL> "<STR_LIT>" : "<STR_LIT>" } ] ) <EOL> try : <EOL> self . keyjar [ client_id ] . append ( _kc ) <EOL> except KeyError : <EOL> self . keyjar [ client_id ] = [ _kc ] <EOL> @ staticmethod <EOL> def verify_correct ( cinfo , restrictions ) : <EOL> for fname , arg in restrictions . items ( ) : <EOL> func = restrict . factory ( fname ) <EOL> res = func ( arg , cinfo ) <EOL> if res : <EOL> raise RestrictionError ( res ) <EOL> def set_token_policy ( self , cid , cinfo ) : <EOL> for ttyp in [ '<STR_LIT>' , '<STR_LIT>' ] : <EOL> pol = { } <EOL> for rgtyp in [ '<STR_LIT>' , '<STR_LIT>' ] : <EOL> try : <EOL> rtyp = cinfo [ rgtyp ] <EOL> except KeyError : <EOL> pass <EOL> else : <EOL> for typ in rtyp : <EOL> try : <EOL> pol [ typ ] = self . lifetime_policy [ ttyp ] [ typ ] <EOL> except KeyError : <EOL> pass <EOL> self . token_policy [ ttyp ] [ cid ] = pol <EOL> def create_new_client ( self , request , restrictions ) : <EOL> """<STR_LIT>""" <EOL> _cinfo = request . to_dict ( ) <EOL> self . match_client_request ( _cinfo ) <EOL> _id = rndstr ( <NUM_LIT:12> ) <EOL> while _id in self . cdb : <EOL> _id = rndstr ( <NUM_LIT:12> ) <EOL> _cinfo [ "<STR_LIT>" ] = _id <EOL> _cinfo [ "<STR_LIT>" ] = secret ( self . seed , _id ) <EOL> _cinfo [ "<STR_LIT>" ] = utc_time_sans_frac ( ) <EOL> _cinfo [ "<STR_LIT>" ] = utc_time_sans_frac ( ) + self . secret_lifetime <EOL> if ClientInfoEndpoint in self . endp : <EOL> _cinfo [ "<STR_LIT>" ] = rndstr ( <NUM_LIT:32> ) <EOL> _cinfo [ "<STR_LIT>" ] = "<STR_LIT>" % ( <EOL> self . name , self . client_info_url , ClientInfoEndpoint . etype , <EOL> _id ) <EOL> if "<STR_LIT>" in request : <EOL> _cinfo [ "<STR_LIT>" ] = self . _uris_to_tuples ( <EOL> request [ "<STR_LIT>" ] ) <EOL> self . load_keys ( request , _id , _cinfo [ "<STR_LIT>" ] ) <EOL> try : <EOL> _behav = self . behavior [ '<STR_LIT>' ] <EOL> except KeyError : <EOL> pass <EOL> else : <EOL> self . verify_correct ( _cinfo , _behav ) <EOL> self . set_token_policy ( _id , _cinfo ) <EOL> self . cdb [ _id ] = _cinfo <EOL> return _id <EOL> def match_client_request ( self , request ) : <EOL> for _pref , _prov in PREFERENCE2PROVIDER . items ( ) : <EOL> if _pref in request : <EOL> if _pref == "<STR_LIT>" : <EOL> for val in request [ _pref ] : <EOL> match = False <EOL> p = set ( val . split ( "<STR_LIT:U+0020>" ) ) <EOL> for cv in self . capabilities [ _prov ] : <EOL> if p == set ( cv . split ( '<STR_LIT:U+0020>' ) ) : <EOL> match = True <EOL> break <EOL> if not match : <EOL> raise CapabilitiesMisMatch ( <EOL> '<STR_LIT>' . format ( _pref ) ) <EOL> else : <EOL> if isinstance ( request [ _pref ] , six . string_types ) : <EOL> if request [ _pref ] not in self . capabilities [ _prov ] : <EOL> raise CapabilitiesMisMatch ( <EOL> '<STR_LIT>' . format ( _pref ) ) <EOL> else : <EOL> if not set ( request [ _pref ] ) . issubset ( <EOL> set ( self . capabilities [ _prov ] ) ) : <EOL> raise CapabilitiesMisMatch ( <EOL> '<STR_LIT>' . format ( _pref ) ) <EOL> def client_info ( self , client_id ) : <EOL> _cinfo = self . cdb [ client_id ] . copy ( ) <EOL> try : <EOL> _cinfo [ "<STR_LIT>" ] = self . _tuples_to_uris ( <EOL> _cinfo [ "<STR_LIT>" ] ) <EOL> except KeyError : <EOL> pass <EOL> msg = ClientInfoResponse ( ** _cinfo ) <EOL> return Response ( msg . to_json ( ) , content = "<STR_LIT:application/json>" ) <EOL> def client_info_update ( self , client_id , request ) : <EOL> _cinfo = self . cdb [ client_id ] . copy ( ) <EOL> try : <EOL> _cinfo [ "<STR_LIT>" ] = self . _tuples_to_uris ( <EOL> _cinfo [ "<STR_LIT>" ] ) <EOL> except KeyError : <EOL> pass <EOL> for key , value in request . items ( ) : <EOL> if key in [ "<STR_LIT>" , "<STR_LIT>" ] : <EOL> try : <EOL> assert value == _cinfo [ key ] <EOL> except AssertionError : <EOL> raise ModificationForbidden ( "<STR_LIT>" ) <EOL> else : <EOL> _cinfo [ key ] = value <EOL> for key in list ( _cinfo . keys ( ) ) : <EOL> if key in [ "<STR_LIT>" , "<STR_LIT>" , <EOL> "<STR_LIT>" , "<STR_LIT>" ] : <EOL> continue <EOL> if key not in request : <EOL> del _cinfo [ key ] <EOL> if "<STR_LIT>" in request : <EOL> _cinfo [ "<STR_LIT>" ] = self . _uris_to_tuples ( <EOL> request [ "<STR_LIT>" ] ) <EOL> self . cdb [ client_id ] = _cinfo <EOL> def verify_client ( self , environ , areq , authn_method , client_id = "<STR_LIT>" ) : <EOL> """<STR_LIT>""" <EOL> if not client_id : <EOL> client_id = get_client_id ( self . cdb , areq , <EOL> environ [ "<STR_LIT>" ] ) <EOL> try : <EOL> method = self . client_authn_methods [ authn_method ] <EOL> except KeyError : <EOL> raise UnSupported ( ) <EOL> return method ( self ) . verify ( environ , client_id = client_id ) <EOL> def consume_software_statement ( self , software_statement ) : <EOL> return { } <EOL> def registration_endpoint ( self , ** kwargs ) : <EOL> """<STR_LIT>""" <EOL> _request = RegistrationRequest ( ) . deserialize ( kwargs [ '<STR_LIT>' ] , "<STR_LIT>" ) <EOL> try : <EOL> _request . verify ( keyjar = self . keyjar ) <EOL> except InvalidRedirectUri as err : <EOL> msg = ClientRegistrationError ( error = "<STR_LIT>" , <EOL> error_description = "<STR_LIT:%s>" % err ) <EOL> return BadRequest ( msg . to_json ( ) , content = "<STR_LIT:application/json>" ) <EOL> except ( MissingPage , VerificationError ) as err : <EOL> msg = ClientRegistrationError ( error = "<STR_LIT>" , <EOL> error_description = "<STR_LIT:%s>" % err ) <EOL> return BadRequest ( msg . to_json ( ) , content = "<STR_LIT:application/json>" ) <EOL> if self . authn_at_registration : <EOL> try : <EOL> self . verify_client ( kwargs [ '<STR_LIT>' ] , _request , <EOL> self . authn_at_registration ) <EOL> except ( AuthnFailure , UnknownAssertionType ) : <EOL> return Unauthorized ( ) <EOL> client_restrictions = { } <EOL> if '<STR_LIT>' in _request : <EOL> for ss in _request [ '<STR_LIT>' ] : <EOL> client_restrictions . update ( self . consume_software_statement ( ss ) ) <EOL> del _request [ '<STR_LIT>' ] <EOL> del _request [ '<STR_LIT>' ] <EOL> try : <EOL> client_id = self . create_new_client ( _request , client_restrictions ) <EOL> except CapabilitiesMisMatch as err : <EOL> msg = ClientRegistrationError ( error = "<STR_LIT>" , <EOL> error_description = "<STR_LIT:%s>" % err ) <EOL> return BadRequest ( msg . to_json ( ) , content = "<STR_LIT:application/json>" ) <EOL> except RestrictionError as err : <EOL> msg = ClientRegistrationError ( error = "<STR_LIT>" , <EOL> error_description = "<STR_LIT:%s>" % err ) <EOL> return BadRequest ( msg . to_json ( ) , content = "<STR_LIT:application/json>" ) <EOL> return self . client_info ( client_id ) <EOL> def client_info_endpoint ( self , method = "<STR_LIT:GET>" , ** kwargs ) : <EOL> """<STR_LIT>""" <EOL> _query = parse_qs ( kwargs [ '<STR_LIT>' ] ) <EOL> try : <EOL> _id = _query [ "<STR_LIT>" ] [ <NUM_LIT:0> ] <EOL> except KeyError : <EOL> return BadRequest ( "<STR_LIT>" ) <EOL> try : <EOL> assert _id in self . cdb <EOL> except AssertionError : <EOL> return Unauthorized ( ) <EOL> try : <EOL> self . verify_client ( kwargs [ '<STR_LIT>' ] , kwargs [ '<STR_LIT>' ] , <EOL> "<STR_LIT>" , client_id = _id ) <EOL> except ( AuthnFailure , UnknownAssertionType ) : <EOL> return Unauthorized ( ) <EOL> if method == "<STR_LIT:GET>" : <EOL> return self . client_info ( _id ) <EOL> elif method == "<STR_LIT>" : <EOL> try : <EOL> _request = ClientUpdateRequest ( ) . from_json ( kwargs [ '<STR_LIT>' ] ) <EOL> except ValueError as err : <EOL> return BadRequest ( str ( err ) ) <EOL> try : <EOL> _request . verify ( ) <EOL> except InvalidRedirectUri as err : <EOL> msg = ClientRegistrationError ( error = "<STR_LIT>" , <EOL> error_description = "<STR_LIT:%s>" % err ) <EOL> return BadRequest ( msg . to_json ( ) , content = "<STR_LIT:application/json>" ) <EOL> except ( MissingPage , VerificationError ) as err : <EOL> msg = ClientRegistrationError ( error = "<STR_LIT>" , <EOL> error_description = "<STR_LIT:%s>" % err ) <EOL> return BadRequest ( msg . to_json ( ) , content = "<STR_LIT:application/json>" ) <EOL> try : <EOL> self . client_info_update ( _id , _request ) <EOL> return self . client_info ( _id ) <EOL> except ModificationForbidden : <EOL> return Forbidden ( ) <EOL> elif method == "<STR_LIT>" : <EOL> try : <EOL> del self . cdb [ _id ] <EOL> except KeyError : <EOL> return Unauthorized ( ) <EOL> else : <EOL> return NoContent ( ) <EOL> def provider_features ( self , pcr_class = ServerMetadata , provider_config = None ) : <EOL> """<STR_LIT>""" <EOL> _provider_info = pcr_class ( ** CAPABILITIES ) <EOL> _provider_info [ "<STR_LIT>" ] = self . scopes <EOL> sign_algs = list ( jws . SIGNER_ALGS . keys ( ) ) <EOL> sign_algs . remove ( '<STR_LIT:none>' ) <EOL> sign_algs = sorted ( sign_algs , key = cmp_to_key ( sort_sign_alg ) ) <EOL> _pat1 = "<STR_LIT>" <EOL> _pat2 = "<STR_LIT>" <EOL> for typ in [ "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT>" ] : <EOL> _provider_info [ _pat1 . format ( typ ) ] = sign_algs <EOL> _provider_info [ _pat2 . format ( typ ) ] = AUTH_METHODS_SUPPORTED <EOL> if provider_config : <EOL> _provider_info . update ( provider_config ) <EOL> return _provider_info <EOL> def verify_capabilities ( self , capabilities ) : <EOL> """<STR_LIT>""" <EOL> _pinfo = self . provider_features ( ) <EOL> for key , val in capabilities . items ( ) : <EOL> if isinstance ( val , six . string_types ) : <EOL> try : <EOL> if val in _pinfo [ key ] : <EOL> continue <EOL> else : <EOL> return False <EOL> except KeyError : <EOL> return False <EOL> return True <EOL> def create_providerinfo ( self , pcr_class = ASConfigurationResponse , <EOL> setup = None ) : <EOL> """<STR_LIT>""" <EOL> _provider_info = self . capabilities <EOL> if self . jwks_uri and self . keyjar : <EOL> _provider_info [ "<STR_LIT>" ] = self . jwks_uri <EOL> for endp in self . endp : <EOL> _provider_info [ '<STR_LIT>' . format ( endp . etype ) ] = os . path . join ( <EOL> self . baseurl , endp . url ) <EOL> if setup and isinstance ( setup , dict ) : <EOL> for key in pcr_class . c_param . keys ( ) : <EOL> if key in setup : <EOL> _provider_info [ key ] = setup [ key ] <EOL> _provider_info [ "<STR_LIT>" ] = self . baseurl <EOL> _provider_info [ "<STR_LIT:version>" ] = "<STR_LIT>" <EOL> return _provider_info <EOL> def providerinfo_endpoint ( self , ** kwargs ) : <EOL> _log_info = logger . info <EOL> _log_info ( "<STR_LIT>" ) <EOL> try : <EOL> _response = self . create_providerinfo ( ) <EOL> _log_info ( "<STR_LIT>" % ( _response . to_dict ( ) , ) ) <EOL> headers = [ ( "<STR_LIT>" , "<STR_LIT>" ) , ( "<STR_LIT>" , "<STR_LIT:bar>" ) ] <EOL> if '<STR_LIT>' in kwargs : <EOL> ( key , timestamp ) = kwargs [ '<STR_LIT>' ] <EOL> if key . startswith ( STR ) and key . endswith ( STR ) : <EOL> cookie = self . cookie_func ( key , self . cookie_name , "<STR_LIT>" , <EOL> self . sso_ttl ) <EOL> headers . append ( cookie ) <EOL> resp = Response ( _response . to_json ( ) , content = "<STR_LIT:application/json>" , <EOL> headers = headers ) <EOL> except Exception as err : <EOL> message = traceback . format_exception ( * sys . exc_info ( ) ) <EOL> logger . error ( message ) <EOL> resp = Response ( message , content = "<STR_LIT>" ) <EOL> return resp <EOL> @ staticmethod <EOL> def verify_code_challenge ( code_verifier , code_challenge , <EOL> code_challenge_method = '<STR_LIT>' ) : <EOL> """<STR_LIT>""" <EOL> _h = CC_METHOD [ code_challenge_method ] ( <EOL> code_verifier . encode ( ) ) . hexdigest ( ) <EOL> _cc = b64e ( _h . encode ( ) ) <EOL> if _cc . decode ( ) != code_challenge : <EOL> logger . error ( '<STR_LIT>' ) <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> return True <EOL> def do_access_token_response ( self , access_token , atinfo , state , <EOL> refresh_token = None ) : <EOL> _tinfo = { '<STR_LIT>' : access_token , '<STR_LIT>' : atinfo [ '<STR_LIT>' ] , <EOL> '<STR_LIT>' : '<STR_LIT>' , '<STR_LIT:state>' : state } <EOL> try : <EOL> _tinfo [ '<STR_LIT>' ] = atinfo [ '<STR_LIT>' ] <EOL> except KeyError : <EOL> pass <EOL> if refresh_token : <EOL> _tinfo [ '<STR_LIT>' ] = refresh_token <EOL> return AccessTokenResponse ( ** by_schema ( AccessTokenResponse , ** _tinfo ) ) <EOL> def code_grant_type ( self , areq ) : <EOL> try : <EOL> _info = self . sdb [ areq [ "<STR_LIT:code>" ] ] <EOL> except KeyError : <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> authzreq = json . loads ( _info [ '<STR_LIT>' ] ) <EOL> if '<STR_LIT>' in areq : <EOL> try : <EOL> _method = authzreq [ '<STR_LIT>' ] <EOL> except KeyError : <EOL> _method = '<STR_LIT>' <EOL> resp = self . verify_code_challenge ( areq [ '<STR_LIT>' ] , <EOL> authzreq [ '<STR_LIT>' ] , <EOL> _method ) <EOL> if resp : <EOL> return resp <EOL> if '<STR_LIT:state>' in areq : <EOL> if self . sdb [ areq [ '<STR_LIT:code>' ] ] [ '<STR_LIT:state>' ] != areq [ '<STR_LIT:state>' ] : <EOL> err = TokenErrorResponse ( error = "<STR_LIT>" ) <EOL> return Unauthorized ( err . to_json ( ) , content = "<STR_LIT:application/json>" ) <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> issue_refresh = False <EOL> if '<STR_LIT>' in authzreq and '<STR_LIT>' in authzreq [ '<STR_LIT>' ] : <EOL> if authzreq [ '<STR_LIT>' ] == '<STR_LIT:code>' : <EOL> issue_refresh = True <EOL> try : <EOL> _tinfo = self . sdb . upgrade_to_token ( areq [ "<STR_LIT:code>" ] , <EOL> issue_refresh = issue_refresh ) <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> logger . debug ( "<STR_LIT>" % _tinfo ) <EOL> atr = AccessTokenResponse ( ** by_schema ( AccessTokenResponse , ** _tinfo ) ) <EOL> logger . debug ( "<STR_LIT>" % atr ) <EOL> return Response ( atr . to_json ( ) , content = "<STR_LIT:application/json>" ) <EOL> def client_credentials_grant_type ( self , areq ) : <EOL> _at = self . token_handler . get_access_token ( areq [ '<STR_LIT>' ] , <EOL> scope = areq [ '<STR_LIT>' ] , <EOL> grant_type = '<STR_LIT>' ) <EOL> _info = self . token_handler . token_factory . get_info ( _at ) <EOL> try : <EOL> _rt = self . token_handler . get_refresh_token ( <EOL> self . baseurl , _info [ '<STR_LIT>' ] , '<STR_LIT>' ) <EOL> except NotAllowed : <EOL> atr = self . do_access_token_response ( _at , _info , areq [ '<STR_LIT:state>' ] ) <EOL> else : <EOL> atr = self . do_access_token_response ( _at , _info , areq [ '<STR_LIT:state>' ] , _rt ) <EOL> return Response ( atr . to_json ( ) , content = "<STR_LIT:application/json>" ) <EOL> def password_grant_type ( self , areq ) : <EOL> _at = self . token_handler . get_access_token ( areq [ '<STR_LIT>' ] , <EOL> scope = areq [ '<STR_LIT>' ] , <EOL> grant_type = '<STR_LIT:password>' ) <EOL> _info = self . token_handler . token_factory . get_info ( _at ) <EOL> try : <EOL> _rt = self . token_handler . get_refresh_token ( <EOL> self . baseurl , _info [ '<STR_LIT>' ] , '<STR_LIT:password>' ) <EOL> except NotAllowed : <EOL> atr = self . do_access_token_response ( _at , _info , areq [ '<STR_LIT:state>' ] ) <EOL> else : <EOL> atr = self . do_access_token_response ( _at , _info , areq [ '<STR_LIT:state>' ] , _rt ) <EOL> return Response ( atr . to_json ( ) , content = "<STR_LIT:application/json>" ) <EOL> def refresh_token_grant_type ( self , areq ) : <EOL> at = self . token_handler . refresh_access_token ( <EOL> self . baseurl , areq [ '<STR_LIT>' ] , '<STR_LIT>' ) <EOL> atr = AccessTokenResponse ( ** by_schema ( AccessTokenResponse , ** at ) ) <EOL> return Response ( atr . to_json ( ) , content = "<STR_LIT:application/json>" ) <EOL> def token_endpoint ( self , authn = "<STR_LIT>" , ** kwargs ) : <EOL> """<STR_LIT>""" <EOL> _sdb = self . sdb <EOL> logger . debug ( "<STR_LIT>" ) <EOL> body = kwargs [ "<STR_LIT>" ] <EOL> logger . debug ( "<STR_LIT>" % body ) <EOL> areq = AccessTokenRequest ( ) . deserialize ( body , "<STR_LIT>" ) <EOL> try : <EOL> client_id = 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> logger . debug ( "<STR_LIT>" % areq ) <EOL> _grant_type = areq [ "<STR_LIT>" ] <EOL> if _grant_type == "<STR_LIT>" : <EOL> return self . code_grant_type ( areq ) <EOL> elif _grant_type == '<STR_LIT>' : <EOL> return self . client_credentials_grant_type ( areq ) <EOL> elif _grant_type == '<STR_LIT:password>' : <EOL> return self . password_grant_type ( areq ) <EOL> elif _grant_type == '<STR_LIT>' : <EOL> return self . refresh_token_grant_type ( areq ) <EOL> else : <EOL> raise UnSupported ( '<STR_LIT>' . format ( _grant_type ) ) <EOL> def key_setup ( self , local_path , vault = "<STR_LIT>" , sig = None , enc = None ) : <EOL> """<STR_LIT>""" <EOL> self . jwks_uri = key_export ( self . baseurl , local_path , vault , self . keyjar , <EOL> fqdn = self . hostname , sig = sig , enc = enc ) <EOL> @ staticmethod <EOL> def token_access ( endpoint , client_id , token_info ) : <EOL> allow = False <EOL> if endpoint == '<STR_LIT>' : <EOL> if '<STR_LIT>' in token_info and client_id == token_info [ '<STR_LIT>' ] : <EOL> allow = True <EOL> elif len ( token_info [ '<STR_LIT>' ] ) == <NUM_LIT:1> and token_info [ '<STR_LIT>' ] == [ <EOL> client_id ] : <EOL> allow = True <EOL> else : <EOL> if '<STR_LIT>' in token_info and client_id == token_info [ '<STR_LIT>' ] : <EOL> allow = True <EOL> elif '<STR_LIT>' in token_info : <EOL> if client_id in token_info [ '<STR_LIT>' ] : <EOL> allow = True <EOL> return allow <EOL> def get_token_info ( self , authn , req , endpoint ) : <EOL> """<STR_LIT>""" <EOL> try : <EOL> client_id = self . client_authn ( self , req , 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> logger . debug ( '<STR_LIT>' . format ( endpoint , client_id , <EOL> req . to_dict ( ) ) ) <EOL> try : <EOL> token_type = req [ '<STR_LIT>' ] <EOL> except KeyError : <EOL> try : <EOL> _info = self . sdb . token_factory [ '<STR_LIT>' ] . info ( <EOL> req [ '<STR_LIT>' ] ) <EOL> except KeyError : <EOL> try : <EOL> _info = self . sdb . token_factory [ '<STR_LIT>' ] . get_info ( <EOL> req [ '<STR_LIT>' ] ) <EOL> except KeyError : <EOL> raise <EOL> else : <EOL> token_type = '<STR_LIT>' <EOL> else : <EOL> token_type = '<STR_LIT>' <EOL> else : <EOL> try : <EOL> _info = self . sdb . token_factory [ token_type ] . get_info ( <EOL> req [ '<STR_LIT>' ] ) <EOL> except KeyError : <EOL> raise <EOL> if not self . token_access ( endpoint , client_id , _info ) : <EOL> return BadRequest ( ) <EOL> return client_id , token_type , _info <EOL> def revocation_endpoint ( self , authn = '<STR_LIT>' , request = None , ** kwargs ) : <EOL> """<STR_LIT>""" <EOL> trr = TokenRevocationRequest ( ) . deserialize ( request , "<STR_LIT>" ) <EOL> resp = self . get_token_info ( authn , trr , '<STR_LIT>' ) <EOL> if isinstance ( resp , Response ) : <EOL> return resp <EOL> else : <EOL> client_id , token_type , _info = resp <EOL> logger . info ( '<STR_LIT>' . format ( client_id , trr . to_dict ( ) ) ) <EOL> try : <EOL> self . sdb . token_factory [ token_type ] . invalidate ( trr [ '<STR_LIT>' ] ) <EOL> except KeyError : <EOL> return BadRequest ( ) <EOL> else : <EOL> return Response ( '<STR_LIT:OK>' ) <EOL> def introspection_endpoint ( self , authn = '<STR_LIT>' , request = None , ** kwargs ) : <EOL> """<STR_LIT>""" <EOL> tir = TokenIntrospectionRequest ( ) . deserialize ( request , "<STR_LIT>" ) <EOL> resp = self . get_token_info ( authn , tir , '<STR_LIT>' ) <EOL> if isinstance ( resp , Response ) : <EOL> return resp <EOL> else : <EOL> client_id , token_type , _info = resp <EOL> logger . info ( '<STR_LIT>' . format ( client_id , <EOL> tir . to_dict ( ) ) ) <EOL> ir = TokenIntrospectionResponse ( <EOL> active = self . sdb . token_factory [ token_type ] . is_valid ( _info ) , <EOL> ** _info . to_dict ( ) ) <EOL> ir . weed ( ) <EOL> return Response ( ir . to_json ( ) , content = "<STR_LIT:application/json>" ) </s>
