o i'e[+@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)NZnonealgb64FT)typrErG),:)Z separatorsclsZ 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) signatureZencoded_stringr"r"r#rTgsL              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().rFTFzIt is required that you pass in a value for the "detached_payload" argument to decode a message having the b64 header set to false.rKrr)r:r[r]) warningswarntuplerrrr _loadrNrVrsplit_verify_signature) r r_r!rrrbkwargsZmerged_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: )rbr:)rerfrgrrrl)r r_r!rrrbrkZdecodedr"r"r#rYs   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 )rhrP)r r_r?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) NrMz$Invalid token type. Token must be a rKrzNot enough segmentszInvalid header paddingzInvalid header string: z,Invalid header string: must be a json objectzInvalid payload paddingzInvalid crypto padding)r+r(rTr;r risplitr*rr,binasciiErrorrRloadsdict) r r_r\Zcrypto_segmentZheader_segmentZpayload_segmenterrZ header_datar[r7r:r]r"r"r#rhsL            z PyJWS._loadr\r[r]c Csz|d}Wn tytdw|r|dur ||vr tdz||}Wnty8}ztd|d}~ww||} ||| |sItddS)NrEzAlgorithm not specifiedz&The specified alg value is not allowedr5zSignature verification failed)r/r r8r6rWZverifyr ) r r\r[r]r!rrEr)r7Z prepared_keyr"r"r#rjs"    zPyJWS._verify_signaturecCsd|vr ||ddSdS)Nkid) _validate_kid)r r?r"r"r#rP8szPyJWS._validate_headersrvcCst|ts tddS)Nz(Key ID header parameter must be a string)r+r(r )r rvr"r"r#rw<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()r^NNN) r_r`r!rarrrrrbrcrrd) r_r`r!rarrrrrbrcrr)r_r`rrd)r_r`rro)r^N) r\r;r[rdr]r;r!rarrrr)r?rdrr)rvrrr)__name__ __module__ __qualname__rOr$ staticmethodrr.r1r3r8rTrlrYrnrhrjrPrwr"r"r"r#rsD      H 0  + r)!Z __future__rrqrRretypingrrrrrrr exceptionsr r r r ZutilsrrrrrrZ_jws_global_objrTrlrYr.r1r8rnr"r"r"r#s,  (