o Ti[+@sddlmZddlZddlZddlZddlmZmZddlm Z m Z m Z m Z ddl mZmZmZmZddlmZmZddlmZerJdd lmZmZGd d d ZeZejZejZejZejZejZejZej Z dS) ) annotationsN) TYPE_CHECKINGAny) Algorithmget_default_algorithms has_cryptorequires_cryptography) DecodeErrorInvalidAlgorithmErrorInvalidSignatureErrorInvalidTokenError)base64url_decodebase64url_encode)RemovedInPyjwt3Warning)AllowedPrivateKeysAllowedPublicKeysc@seZdZdZ  dHdId d ZedJd d ZdKddZdLddZdMddZ dNddZ     dOdPd+d,Z -   dQdRd4d5Z -   dQdSd7d8Z dTd9d:ZdUd||jvr tdt|tstd||j|<|j|dS)zW Registers a new Algorithm for use when creating and verifying tokens. z Algorithm already has a handler.z!Object is not of type `Algorithm`N)r ValueError isinstancer TypeErrorradd)r r'r)r"r"r#register_algorithm5s   zPyJWS.register_algorithmcCs*||jvr td|j|=|j|dS)z Unregisters an Algorithm for use when creating and verifying tokens Throws KeyError if algorithm is not registered. zJThe specified algorithm could not be removed because it is not registered.N)rKeyErrorrremove)r r'r"r"r#unregister_algorithmBs zPyJWS.unregister_algorithm list[str]cCs t|jS)zM Returns a list of supported values for the 'alg' parameter. )rr)r r"r"r#get_algorithmsPs zPyJWS.get_algorithmsalg_namec CsNz|j|WSty&}zts|tvrtd|d|td|d}~ww)z For a given string name, return the matching Algorithm object. Example usage: >>> jws_obj.get_algorithm_by_name("RS256") z Algorithm 'z9' could not be found. Do you have cryptography installed?Algorithm not supportedN)rr/rr NotImplementedError)r r4er"r"r#get_algorithm_by_nameVs    zPyJWS.get_algorithm_by_nameHS256FTpayloadbytesr! AllowedPrivateKeys | str | bytes algorithm str | Noneheaders json_encodertype[json.JSONEncoder] | Noneis_payload_detachedbool sort_headerscCs*g}|dur|nd} |r"|d} | r|d} |d} | dur"d}|j| d} |r4||| || ds;| d=|rBd| d<nd| vrI| d=tj| d||d } |t| |r`|}nt|}||d |}| | }| |}| ||}|t||rd |d <d |}| d S)Nnonealgb64FT)typrFrH),:) separatorscls sort_keys.rutf-8)get header_typ_validate_headersupdatejsondumpsencodeappendrjoinr8 prepare_keysigndecode)r r:r!r=r?r@rBrDsegmentsZ algorithm_Z headers_algZ headers_b64headerZ json_headerZ msg_payload signing_inputr) signatureencoded_stringr"r"r#rWgsL              z PyJWS.encodejwt str | bytesAllowedPublicKeys | str | bytesdetached_payload bytes | Nonedict[str, Any]c Ks|rtdt|t|duri}i|j|}|d}|r(|s(td||\} } } } | dddurP|durAtd|} d | dd d | g} |r[| | | | ||| | | d S) Nzypassing additional kwargs to decode_complete() is deprecated and will be removed in pyjwt version 3. Unsupported kwargs: r&z\It is required that you pass in a value for the "algorithms" argument when calling decode().rGTFzIt is required that you pass in a value for the "detached_payload" argument to decode a message having the b64 header set to false.rNrr)r:r^r`) warningswarntuplerrrr _loadrQrYrsplit_verify_signature) r rcr!rrrfkwargsZmerged_optionsr&r:r_r^r`r"r"r#decode_completes:  zPyJWS.decode_completercKs:|rtdt|t|j|||||d}|dS)Nzppassing additional kwargs to decode() is deprecated and will be removed in pyjwt version 3. Unsupported kwargs: )rfr:)rirjrkrrrp)r rcr!rrrfrodecodedr"r"r#r\s   z PyJWS.decodecCs||d}|||S)zReturns back the JWT header parameters as a dict() Note: The signature is not verified so the header parameters should not be fully trusted until signature verification is complete )rlrS)r rcr?r"r"r#get_unverified_headers zPyJWS.get_unverified_header*tuple[bytes, bytes, dict[str, Any], bytes]c Cslt|tr |d}t|tstdtz|dd\}}|dd\}}Wnty9}ztd|d}~wwzt|}Wnt t j fyT}ztd|d}~wwzt |}Wntyp} ztd| | d} ~ wwt|tsztdzt|} Wnt t j fy}ztd |d}~wwzt|} Wnt t j fy}ztd |d}~ww| ||| fS) NrPz$Invalid token type. Token must be a rNrzNot enough segmentszInvalid header paddingzInvalid header string: z,Invalid header string: must be a json objectzInvalid payload paddingzInvalid crypto padding)r+r(rWr;r rmsplitr*rr,binasciiErrorrUloadsdict) r rcr_Zcrypto_segmentZheader_segmentZpayload_segmenterrZ header_datar^r7r:r`r"r"r#rlsL            z PyJWS._loadr_r^r`c Csz|d}Wn tytdw|r|dur ||vr tdz||}Wnty8}ztd|d}~ww||} ||| |sItddS)NrFzAlgorithm not specifiedz&The specified alg value is not allowedr5zSignature verification failed)r/r r8r6rZverifyr ) r r_r^r`r!rrFr)r7Z prepared_keyr"r"r#rns"    zPyJWS._verify_signaturecCsd|vr ||ddSdS)Nkid) _validate_kid)r r?r"r"r#rS8szPyJWS._validate_headersr|cCst|ts tddS)Nz(Key ID header parameter must be a string)r+r(r )r r|r"r"r#r}<s zPyJWS._validate_kid)NN)rrrrrr)rr%)r'r(r)rrr)r'r(rr)rr2)r4r(rr)r9NNFT)r:r;r!r<r=r>r?rr@rArBrCrDrCrr()rbNNN) rcrdr!rerrrrrfrgrrh) rcrdr!rerrrrrfrgrr)rcrdrrh)rcrdrrt)rbN) r_r;r^rhr`r;r!rerrrr)r?rhrr)r|rrr)__name__ __module__ __qualname__rRr$ staticmethodrr.r1r3r8rWrpr\rsrlrnrSr}r"r"r"r#rsD      H 0  + r)! __future__rrvrUritypingrrrrrrr exceptionsr r r r utilsrrrrrrZ_jws_global_objrWrpr\r.r1r8rsr"r"r"r#s,  (