<s> import binascii <EOL> import json <EOL> import warnings <EOL> from collections import Mapping <EOL> from . algorithms import Algorithm , get_default_algorithms <EOL> from . compat import binary_type , string_types , text_type <EOL> from . exceptions import DecodeError , InvalidAlgorithmError , InvalidTokenError <EOL> from . utils import base64url_decode , base64url_encode , merge_dict <EOL> class PyJWS ( object ) : <EOL> header_typ = '<STR_LIT>' <EOL> def __init__ ( self , algorithms = None , options = None ) : <EOL> self . _algorithms = get_default_algorithms ( ) <EOL> self . _valid_algs = ( set ( algorithms ) if algorithms is not None <EOL> else set ( self . _algorithms ) ) <EOL> for key in list ( self . _algorithms . keys ( ) ) : <EOL> if key not in self . _valid_algs : <EOL> del self . _algorithms [ key ] <EOL> if not options : <EOL> options = { } <EOL> self . options = merge_dict ( self . _get_default_options ( ) , options ) <EOL> @ staticmethod <EOL> def _get_default_options ( ) : <EOL> return { <EOL> '<STR_LIT>' : True <EOL> } <EOL> def register_algorithm ( self , alg_id , alg_obj ) : <EOL> """<STR_LIT>""" <EOL> if alg_id in self . _algorithms : <EOL> raise ValueError ( '<STR_LIT>' ) <EOL> if not isinstance ( alg_obj , Algorithm ) : <EOL> raise TypeError ( '<STR_LIT>' ) <EOL> self . _algorithms [ alg_id ] = alg_obj <EOL> self . _valid_algs . add ( alg_id ) <EOL> def unregister_algorithm ( self , alg_id ) : <EOL> """<STR_LIT>""" <EOL> if alg_id not in self . _algorithms : <EOL> raise KeyError ( '<STR_LIT>' <EOL> '<STR_LIT>' ) <EOL> del self . _algorithms [ alg_id ] <EOL> self . _valid_algs . remove ( alg_id ) <EOL> def get_algorithms ( self ) : <EOL> """<STR_LIT>""" <EOL> return list ( self . _valid_algs ) <EOL> def encode ( self , payload , key , algorithm = '<STR_LIT>' , headers = None , <EOL> json_encoder = None ) : <EOL> segments = [ ] <EOL> if algorithm is None : <EOL> algorithm = '<STR_LIT:none>' <EOL> if algorithm not in self . _valid_algs : <EOL> pass <EOL> header = { '<STR_LIT>' : self . header_typ , '<STR_LIT>' : algorithm } <EOL> if headers : <EOL> self . _validate_headers ( headers ) <EOL> header . update ( headers ) <EOL> json_header = json . dumps ( <EOL> header , <EOL> separators = ( '<STR_LIT:U+002C>' , '<STR_LIT::>' ) , <EOL> cls = json_encoder <EOL> ) . encode ( '<STR_LIT:utf-8>' ) <EOL> segments . append ( base64url_encode ( json_header ) ) <EOL> segments . append ( base64url_encode ( payload ) ) <EOL> signing_input = b'<STR_LIT:.>' . join ( segments ) <EOL> try : <EOL> alg_obj = self . _algorithms [ algorithm ] <EOL> key = alg_obj . prepare_key ( key ) <EOL> signature = alg_obj . sign ( signing_input , key ) <EOL> except KeyError : <EOL> raise NotImplementedError ( '<STR_LIT>' ) <EOL> segments . append ( base64url_encode ( signature ) ) <EOL> return b'<STR_LIT:.>' . join ( segments ) <EOL> def decode ( self , jws , key = '<STR_LIT>' , verify = True , algorithms = None , options = None , <EOL> ** kwargs ) : <EOL> payload , signing_input , header , signature = self . _load ( jws ) <EOL> if verify : <EOL> merged_options = merge_dict ( self . options , options ) <EOL> if merged_options . get ( '<STR_LIT>' ) : <EOL> self . _verify_signature ( payload , signing_input , header , signature , <EOL> key , algorithms ) <EOL> else : <EOL> warnings . warn ( '<STR_LIT>' <EOL> '<STR_LIT>' , DeprecationWarning ) <EOL> return payload <EOL> def get_unverified_header ( self , jwt ) : <EOL> """<STR_LIT>""" <EOL> headers = self . _load ( jwt ) [ <NUM_LIT:2> ] <EOL> self . _validate_headers ( headers ) <EOL> return headers <EOL> def _load ( self , jwt ) : <EOL> if isinstance ( jwt , text_type ) : <EOL> jwt = jwt . encode ( '<STR_LIT:utf-8>' ) <EOL> if not issubclass ( type ( jwt ) , binary_type ) : <EOL> raise DecodeError ( "<STR_LIT>" . format ( <EOL> binary_type ) ) <EOL> try : <EOL> signing_input , crypto_segment = jwt . rsplit ( b'<STR_LIT:.>' , <NUM_LIT:1> ) <EOL> header_segment , payload_segment = signing_input . split ( b'<STR_LIT:.>' , <NUM_LIT:1> ) <EOL> except ValueError : <EOL> raise DecodeError ( '<STR_LIT>' ) <EOL> try : <EOL> header_data = base64url_decode ( header_segment ) <EOL> except ( TypeError , binascii . Error ) : <EOL> raise DecodeError ( '<STR_LIT>' ) <EOL> try : <EOL> header = json . loads ( header_data . decode ( '<STR_LIT:utf-8>' ) ) <EOL> except ValueError as e : <EOL> raise DecodeError ( '<STR_LIT>' % e ) <EOL> if not isinstance ( header , Mapping ) : <EOL> raise DecodeError ( '<STR_LIT>' ) <EOL> try : <EOL> payload = base64url_decode ( payload_segment ) <EOL> except ( TypeError , binascii . Error ) : <EOL> raise DecodeError ( '<STR_LIT>' ) <EOL> try : <EOL> signature = base64url_decode ( crypto_segment ) <EOL> except ( TypeError , binascii . Error ) : <EOL> raise DecodeError ( '<STR_LIT>' ) <EOL> return ( payload , signing_input , header , signature ) <EOL> def _verify_signature ( self , payload , signing_input , header , signature , <EOL> key = '<STR_LIT>' , algorithms = None ) : <EOL> alg = header . get ( '<STR_LIT>' ) <EOL> if algorithms is not None and alg not in algorithms : <EOL> raise InvalidAlgorithmError ( '<STR_LIT>' ) <EOL> try : <EOL> alg_obj = self . _algorithms [ alg ] <EOL> key = alg_obj . prepare_key ( key ) <EOL> if not alg_obj . verify ( signing_input , key , signature ) : <EOL> raise DecodeError ( '<STR_LIT>' ) <EOL> except KeyError : <EOL> raise InvalidAlgorithmError ( '<STR_LIT>' ) <EOL> def _validate_headers ( self , headers ) : <EOL> if '<STR_LIT>' in headers : <EOL> self . _validate_kid ( headers [ '<STR_LIT>' ] ) <EOL> def _validate_kid ( self , kid ) : <EOL> if not isinstance ( kid , string_types ) : <EOL> raise InvalidTokenError ( '<STR_LIT>' ) <EOL> _jws_global_obj = PyJWS ( ) <EOL> encode = _jws_global_obj . encode <EOL> decode = _jws_global_obj . decode <EOL> register_algorithm = _jws_global_obj . register_algorithm <EOL> unregister_algorithm = _jws_global_obj . unregister_algorithm <EOL> get_unverified_header = _jws_global_obj . get_unverified_header </s>
