o if[+@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)rr0rr NotImplementedError)r!r5er#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)typrGrI),:) separatorscls sort_keys.rutf-8)get header_typ_validate_headersupdatejsondumpsencodeappendrjoinr9 prepare_keysigndecode)r!r;r"r>r@rArCrEsegments algorithm_ headers_alg headers_b64header json_header msg_payload signing_inputr* signatureencoded_stringr#r#r$rXgsL              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().rHTFzIt is required that you pass in a value for the "detached_payload" argument to decode a message having the b64 header set to false.rOrr)r;rbrf) warningswarntuplerrrr _loadrRrZrsplit_verify_signature) r!rir"rrrlkwargsmerged_optionsr'r;rerbrfr#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: )rlr;)rorprqrrrw)r!rir"rrrlrudecodedr#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 )rrrT)r!rir@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) NrQz$Invalid token type. Token must be a rOrzNot enough segmentszInvalid header paddingzInvalid header string: z,Invalid header string: must be a json objectzInvalid payload paddingzInvalid crypto padding)r,r)rXr<r rssplitr+rr-binasciiErrorrVloadsdict) r!rirecrypto_segmentheader_segmentpayload_segmenterr header_datarbr8r;rfr#r#r$rrsL            z PyJWS._loadrerbrfc Csz|d}Wn tytdw|r|dur ||vr tdz||}Wnty8}ztd|d}~ww||} ||| |sItddS)NrGzAlgorithm not specifiedz&The specified alg value is not allowedr6zSignature verification failed)r0r r9r7r[verifyr ) r!rerbrfr"rrGr*r8 prepared_keyr#r#r$rts"    zPyJWS._verify_signaturecCsd|vr ||ddSdS)Nkid) _validate_kid)r!r@r#r#r$rT8szPyJWS._validate_headersrcCst|ts tddS)Nz(Key ID header parameter must be a string)r,r)r )r!rr#r#r$r<s zPyJWS._validate_kid)NN)rrrrrr)rr&)r(r)r*rrr)r(r)rr)rr3)r5r)rr)r:NNFT)r;r<r"r=r>r?r@rrArBrCrDrErDrr))rhNNN) rirjr"rkrrrrrlrmrrn) rirjr"rkrrrrrlrmrr)rirjrrn)rirjrr{)rhN) rer<rbrnrfr<r"rkrrrr)r@rnrr)rrrr)__name__ __module__ __qualname__rSr% staticmethodr r/r2r4r9rXrwr]rzrrrtrTrr#r#r#r$rsD      H 0  + r)! __future__rr}rVrotypingrrrrrrr exceptionsr r r r utilsrrrrrr_jws_global_objrXrwr]r/r2r9rzr#r#r#r$s,  (