o T۷i7@sddlmZddlZddlZddlZddlmZddlmZm Z ddl m Z m Z m Z mZddlmZddlmZmZmZmZmZdd lmZmZdd lmZmZer`dd l mZmZdd lmZe Z!Gd ddZ"e"Z#e#j$Z$e#j%Z%e#j&Z&e#j'Z'e#j(Z(e#j)Z)e#j*Z*dS)) annotationsN)Sequence) TYPE_CHECKINGAny) Algorithmget_default_algorithms has_cryptorequires_cryptography)PyJWK) DecodeErrorInvalidAlgorithmErrorInvalidKeyErrorInvalidSignatureErrorInvalidTokenError)base64url_decodebase64url_encode)InsecureKeyLengthWarningRemovedInPyjwt3Warning)AllowedPrivateKeysAllowedPublicKeys) SigOptionsc@seZdZUdZ  dPdQd d ZedRd d ZdSddZdTddZdUddZ dVddZ e ddddfdWd+d,Z -   dXdYd5d6Z -   dXdZd8d9Zd[d:d;Zd\d=d>Z - d]d^dBdCZdDhZdEedF<ddGd_dIdJZd`dLdMZdadNdOZdS)bPyJWSJWTN algorithmsSequence[str] | NoneoptionsSigOptions | NonereturnNonecCstt|_|dur t|nt|j|_t|jD] }||jvr$|j|=q||_|dur8i|j||_dSdS)N)r _algorithmsset _valid_algslistkeys_get_default_optionsr)selfrrkeyr(A/home/ubuntu/vllm_env/lib/python3.10/site-packages/jwt/api_jws.py__init__$s  zPyJWS.__init__rcCs dddS)NTF)verify_signatureenforce_minimum_key_lengthr(r(r(r(r)r%7s zPyJWS._get_default_optionsalg_idstralg_objrcCs>||jvr tdt|tstd||j|<|j|dS)z Registers a new Algorithm for use when creating and verifying tokens. :param str alg_id: the ID of the Algorithm :param alg_obj: the Algorithm object :type alg_obj: Algorithm z Algorithm already has a handler.z!Object is not of type `Algorithm`N)r ValueError isinstancer TypeErrorr"add)r&r-r/r(r(r)register_algorithm;s   zPyJWS.register_algorithmcCs*||jvr td|j|=|j|dS)z Unregisters an Algorithm for use when creating and verifying tokens :param str alg_id: the ID of the Algorithm :raises KeyError: if algorithm is not registered. zJThe specified algorithm could not be removed because it is not registered.N)r KeyErrorr"remove)r&r-r(r(r)unregister_algorithmLs zPyJWS.unregister_algorithm list[str]cCs t|jS)zh Returns a list of supported values for the `alg` parameter. :rtype: list[str] )r#r")r&r(r(r)get_algorithms[s zPyJWS.get_algorithmsalg_namec CsNz|j|WSty&}zts|tvrtd|d|td|d}~ww)a/ For a given string name, return the matching Algorithm object. Example usage: >>> jws_obj = PyJWS() >>> jws_obj.get_algorithm_by_name("RS256") :param alg_name: The name of the algorithm to retrieve :type alg_name: str :rtype: Algorithm z Algorithm 'z9' could not be found. Do you have cryptography installed?Algorithm not supportedN)r r5r r NotImplementedError)r&r:er(r(r)get_algorithm_by_namecs    zPyJWS.get_algorithm_by_nameFTpayloadbytesr'(AllowedPrivateKeys | PyJWK | str | bytes algorithm str | Noneheadersdict[str, Any] | None json_encodertype[json.JSONEncoder] | Noneis_payload_detachedbool sort_headerscCsg}|turt|tr|j} nd} n|dur"t|tr|j} nd} n|} |r<|d} | r1|d} |d} | durr' prepare_keycheck_key_lengthrrwarningswarnrsigndecode)r&r?r'rBrDrFrHrJsegments algorithm_ headers_alg headers_b64header json_header msg_payload signing_inputr/key_length_msg signatureencoded_stringr(r(r)rexsj                 z PyJWS.encodejwt str | bytes'AllowedPublicKeys | PyJWK | str | bytesdetached_payload bytes | Nonekwargsdict[str, Any]c Ks|rtjdt|tdd|dur|j}ni|j|}|d}|r1|s1t|ts1td| |\} } } } | | | dddur^|durOtd |} d | d d d | g} |ri|| | | ||| | | d S)Nzypassing additional kwargs to decode_complete() is deprecated and will be removed in pyjwt version 3. Unsupported kwargs: rXrYr+z\It is required that you pass in a value for the "algorithms" argument when calling decode().rNTFzIt is required that you pass in a value for the "detached_payload" argument to decode a message having the b64 header set to false.rWrr)r?rrrw)rjrktupler$rrr1r r _loadrar_rgrsplit_verify_signature) r&rzr'rrr}rmerged_optionsr+r?rurrrwr(r(r)decode_completes>   zPyJWS.decode_completercKs>|rtjdt|tdd|j|||||d}|dS)Nzppassing additional kwargs to decode() is deprecated and will be removed in pyjwt version 3. Unsupported kwargs: rXrY)r}r?)rjrkrr$rr)r&rzr'rrr}rdecodedr(r(r)rms   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 rX)rra)r&rzrDr(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) Nr\z$Invalid token type. Token must be a rWrzNot enough segmentszInvalid header paddingzInvalid header string: z,Invalid header string: must be a json objectzInvalid payload paddingzInvalid crypto padding)r1r.rer@r rsplitr0rr2binasciiErrorrcloadsdict) r&rzrucrypto_segmentheader_segmentpayload_segmenterr header_datarrr=r?rwr(r(r)r&sL            z PyJWS._loadrurrrwc Cs|dur t|tr |jg}z|d}Wn tytddw|r*|dur.||vr.tdt|tr:|j}|j}nz||}WntyR} ztd| d} ~ ww| |}| |} | rr|j ddrjt | tj| tdd||||s}td dS) NrMzAlgorithm not specifiedz&The specified alg value is not allowedr;r,FrYzSignature verification failed)r1r r^r5r rr'r>r<rhrirr_rrjrkrverifyr) r&rurrrwr'rrMr/ prepared_keyr=rvr(r(r)rLs6       zPyJWS._verify_signaturerNzset[str]_supported_critrPrQcCs8d|vr ||d|sd|vr||dSdSdS)Nkidcrit) _validate_kid_validate_crit)r&rDrQr(r(r)raus  zPyJWS._validate_headersrcCst|ts tddS)Nz(Key ID header parameter must be a string)r1r.r)r&rr(r(r)r}s zPyJWS._validate_kidcCsv|d}t|trt|dkrtd|D]#}t|ts td||jvr,td|||vr8td|dqdS)Nrrz/Invalid 'crit' header: must be a non-empty listz-Invalid 'crit' header: values must be stringsz Unsupported critical extension: zCritical extension 'z' is missing from headers)r1r#lenrr.r)r&rDrextr(r(r)rs   zPyJWS._validate_crit)NN)rrrrrr)rr)r-r.r/rrr)r-r.rr)rr8)r:r.rr)r?r@r'rArBrCrDrErFrGrHrIrJrIrr.)ryNNN)rzr{r'r|rrrrr}r~rrrr)rzr{r'r|rrrrr}r~rrrr)rzr{rr)rzr{rr)ryN) rur@rrrrwr@r'r|rrrr)rDrrQrIrr)rrrr)rDrrr)__name__ __module__ __qualname__r`r* staticmethodr%r4r7r9r>r]rerrmrrrr__annotations__rarrr(r(r(r)r!sJ        ] 6  + ' r)+ __future__rrrcrjcollections.abcrtypingrrrrrr r api_jwkr exceptionsr r rrrutilsrrrrrrtypesrobjectr]r_jws_global_objrerrmr4r7r>rr(r(r(r)s4    q