o T۷iU@sUddlmZddlZddlZddlZddlmZddlmZm Z m Z ddl m Z m Z m Z ddlmZmZmZmZddlmZmZmZdd lmZmZmZmZmZmZmZmZm Z dd lm!Z!esie"e#d d rddl$Z$e$j%d kryddlm&Z&nddl'm&Z&ddl(m)Z)m*Z*ddl+m,Z,ddl-m.Z.m/Z/m0Z0ee)e,e1e2fZ3de4d<ee*e,e1e2fZ5de4d<GdddZ6e6Z7ee7_8e7j9Z9e7j:Z:e7j;Z;dS)) annotationsN)timegm) ContainerIterableSequence)datetime timedeltatimezone) TYPE_CHECKINGAnyUnioncast)PyJWS_ALGORITHM_UNSET_jws_global_obj) DecodeErrorExpiredSignatureErrorImmatureSignatureErrorInvalidAudienceErrorInvalidIssuedAtErrorInvalidIssuerErrorInvalidJTIErrorInvalidSubjectErrorMissingRequiredClaimError)RemovedInPyjwt3Warning SPHINX_BUILD) ) TypeAlias)AllowedPrivateKeysAllowedPublicKeys)PyJWK) FullOptionsOptions SigOptionsr AllowedPrivateKeyTypesAllowedPublicKeyTypesc@seZdZdZd[ddZed\d d Zd]d d ZdZd^ddZedddfd_ddZ  d`dad!d"Z #        $dbdcd7d8Z ddd:d;Z #        $dbded=d>Z    $dfdgdAdBZdhdEdFZ dZdidGdHZdjdIdJZdkdMdNZdkdOdPZdkdQdRZdSdTdldVdWZdmdXdYZdS)nPyJWTNoptionsOptions | NonereturnNonecCs6|||_|dur|||_t|d|_dS)N)r*)_get_default_optionsr*_merge_optionsr_get_sig_options_jwsselfr*r4A/home/ubuntu/vllm_env/lib/python3.10/site-packages/jwt/api_jwt.py__init__+s   zPyJWT.__init__r$c Csddddddddgddd S)NTF) verify_signature verify_exp verify_nbf verify_iat verify_aud verify_iss verify_sub verify_jtirequire strict_audenforce_minimum_key_lengthr4r4r4r4r5r.3szPyJWT._get_default_optionsr&cCs|jd|jdddS)Nr7rAF)r7rAr*get)r3r4r4r5r0Cs zPyJWT._get_sig_optionscCs|dur|jS|ddsE|dd|d<|dd|d<|dd|d<|dd|d<|dd|d<|d d|d <|d d|d <i|j|S) Nr7Tr8Fr9r:r;r<r=r>rBr2r4r4r5r/Ks zPyJWT._merge_optionsTpayloaddict[str, Any]keyr' algorithm str | Noneheadersdict[str, Any] | None json_encodertype[json.JSONEncoder] | None sort_headersboolstrc Cst|ts td|}dD]}t||tr#t||||<qd|vr3t|dts3td|j |||d}|j j ||||||dS)aEncode the ``payload`` as JSON Web Token. :param payload: JWT claims, e.g. ``dict(iss=..., aud=..., sub=...)`` :type payload: dict[str, typing.Any] :param key: a key suitable for the chosen algorithm: * for **asymmetric algorithms**: PEM-formatted private key, a multiline string * for **symmetric algorithms**: plain string, sufficiently long for security :type key: str or bytes or PyJWK or :py:class:`jwt.algorithms.AllowedPrivateKeys` :param algorithm: algorithm to sign the token with, e.g. ``"ES256"``. If ``headers`` includes ``alg``, it will be preferred to this parameter. If ``key`` is a :class:`PyJWK` object, by default the key algorithm will be used. :type algorithm: str or None :param headers: additional JWT header fields, e.g. ``dict(kid="my-key-id")``. :type headers: dict[str, typing.Any] or None :param json_encoder: custom JSON encoder for ``payload`` and ``headers`` :type json_encoder: json.JSONEncoder or None :rtype: str :returns: a JSON Web Token :raises TypeError: if ``payload`` is not a ``dict`` zGExpecting a dict object, as JWT only supports JSON objects as payloads.)expiatnbfisszIssuer (iss) must be a string.)rIrK)rM) isinstancedict TypeErrorcopyrCrr utctimetuplerO_encode_payloadr1encode) r3rDrFrGrIrKrM time_claim json_payloadr4r4r5rZZs0 "z PyJWT.encodebytescCstj|d|ddS)z Encode a given payload to the bytes to be signed. This method is intended to be overridden by subclasses that need to encode the payload in a different way, e.g. compress the payload. ),:) separatorsclszutf-8)jsondumpsrZ)r3rDrIrKr4r4r5rYs zPyJWT._encode_payloadrrjwt str | bytesr( algorithmsSequence[str] | Noneverify bool | Nonedetached_payload bytes | Noneaudiencestr | Iterable[str] | Noneissuerstr | Container[str] | Nonesubjectleewayfloat | timedeltakwargsr c Ks| rtjdt| tdd|durd} n|dd} |dur.|| kr.tjdtdd||} d| i}|jj |||||d }| |}|j || ||| | d ||d <|S) u$Identical to ``jwt.decode`` except for return value which is a dictionary containing the token header (JOSE Header), the token payload (JWT Payload), and token signature (JWT Signature) on the keys "header", "payload", and "signature" respectively. :param jwt: the token to be decoded :type jwt: str or bytes :param key: the key suitable for the allowed algorithm :type key: str or bytes or PyJWK or :py:class:`jwt.algorithms.AllowedPublicKeys` :param algorithms: allowed algorithms, e.g. ``["ES256"]`` .. warning:: Do **not** compute the ``algorithms`` parameter based on the ``alg`` from the token itself, or on any other data that an attacker may be able to influence, as that might expose you to various vulnerabilities (see `RFC 8725 §2.1 `_). Instead, either hard-code a fixed value for ``algorithms``, or configure it in the same place you configure the ``key``. Make sure not to mix symmetric and asymmetric algorithms that interpret the ``key`` in different ways (e.g. HS\* and RS\*). :type algorithms: typing.Sequence[str] or None :param jwt.types.Options options: extended decoding and validation options Refer to :py:class:`jwt.types.Options` for more information. :param audience: optional, the value for ``verify_aud`` check :type audience: str or typing.Iterable[str] or None :param issuer: optional, the value for ``verify_iss`` check :type issuer: str or typing.Container[str] or None :param leeway: a time margin in seconds for the expiration check :type leeway: float or datetime.timedelta :rtype: dict[str, typing.Any] :returns: Decoded JWT with the JOSE Header on the key ``header``, the JWS Payload on the key ``payload``, and the JWS Signature on the key ``signature``. zypassing additional kwargs to decode_complete() is deprecated and will be removed in pyjwt version 3. Unsupported kwargs:  stacklevelNTr7zThe `verify` argument to `decode` does nothing in PyJWT 2.0 and newer. The equivalent is setting `verify_signature` to False in the `options` dictionary. This invocation has a mismatch between the kwarg and the option entry.)categoryrv)rFrfr*rj)rlrnrqrprD) warningswarntuplekeysrrCDeprecationWarningr/r1decode_complete_decode_payload_validate_claims)r3rdrFrfr*rhrjrlrnrprqrsr7merged_options sig_optionsdecodedrDr4r4r5r}sL9     zPyJWT.decode_completerc CsRz t|d}Wnty}ztd||d}~wwt|ts'td|S)a Decode the payload from a JWS dictionary (payload, signature, header). This method is intended to be overridden by subclasses that need to decode the payload in a different way, e.g. decompress compressed payloads. rDzInvalid payload string: Nz-Invalid payload string: must be a json object)rbloads ValueErrorrrTrU)r3rrDer4r4r5r~s zPyJWT._decode_payload'AllowedPublicKeys | PyJWK | str | bytesc KsV| rtjdt| tdd|j||||||||| | d } tttt f| dS)uVerify the ``jwt`` token signature and return the token claims. :param jwt: the token to be decoded :type jwt: str or bytes :param key: the key suitable for the allowed algorithm :type key: str or bytes or PyJWK or :py:class:`jwt.algorithms.AllowedPublicKeys` :param algorithms: allowed algorithms, e.g. ``["ES256"]`` If ``key`` is a :class:`PyJWK` object, allowed algorithms will default to the key algorithm. .. warning:: Do **not** compute the ``algorithms`` parameter based on the ``alg`` from the token itself, or on any other data that an attacker may be able to influence, as that might expose you to various vulnerabilities (see `RFC 8725 §2.1 `_). Instead, either hard-code a fixed value for ``algorithms``, or configure it in the same place you configure the ``key``. Make sure not to mix symmetric and asymmetric algorithms that interpret the ``key`` in different ways (e.g. HS\* and RS\*). :type algorithms: typing.Sequence[str] or None :param jwt.types.Options options: extended decoding and validation options Refer to :py:class:`jwt.types.Options` for more information. :param audience: optional, the value for ``verify_aud`` check :type audience: str or typing.Iterable[str] or None :param subject: optional, the value for ``verify_sub`` check :type subject: str or None :param issuer: optional, the value for ``verify_iss`` check :type issuer: str or typing.Container[str] or None :param leeway: a time margin in seconds for the expiration check :type leeway: float or datetime.timedelta :rtype: dict[str, typing.Any] :returns: the JWT claims zppassing additional kwargs to decode() is deprecated and will be removed in pyjwt version 3. Unsupported kwargs: rtru)rhrjrlrprnrqrD) rxryrzr{rr}r rUrOr ) r3rdrFrfr*rhrjrlrprnrqrsrr4r4r5decode,s*9  z PyJWT.decodeIterable[str] | str | NoneContainer[str] | str | NonecCst|tr |}|durt|ttfstd|||dtjt j d }d|vr8|dr8| |||d|vrG|drG| |||d|vrV|d rV|||||d r`||||d rp|j|||d d d|drz||||dr||dSdS)Nz+audience must be a string, iterable or Noner?)tzrQr:rRr9rPr8r<r;r@Fstrictr=r>)rTr total_secondsrOrrV_validate_required_claimsrnowr utc timestamp _validate_iat _validate_nbf _validate_exp _validate_iss _validate_audrC _validate_sub _validate_jti)r3rDr*rlrnrprqrr4r4r5r{s.   zPyJWT._validate_claimsclaims Iterable[str]cCs$|D] }||durt|qdSN)rCr)r3rDrclaimr4r4r5rs zPyJWT._validate_required_claimscCsHd|vrdSt|dtstd|dur |d|kr"tddSdS)z Checks whether "sub" if in the payload is valid or not. This is an Optional claim :param payload(dict): The payload which needs to be validated :param subject(str): The subject of the token subNzSubject must be a stringzInvalid subject)rTrOrrC)r3rDrpr4r4r5rs zPyJWT._validate_subcCs(d|vrdSt|dtstddS)z Checks whether "jti" if in the payload is valid or not This is an Optional claim :param payload(dict): The payload which needs to be validated jtiNzJWT ID must be a string)rTrCrOr)r3rDr4r4r5rs zPyJWT._validate_jtirfloatcCBzt|d}Wn tytddw|||krtddS)NrQz)Issued At claim (iat) must be an integer.z The token is not yet valid (iat))intrrr)r3rDrrqrQr4r4r5rs  zPyJWT._validate_iatcCr)NrRz*Not Before claim (nbf) must be an integer.z The token is not yet valid (nbf))rrrr)r3rDrrqrRr4r4r5rs   zPyJWT._validate_nbfcCsBzt|d}Wn tytddw|||krtddS)NrPz/Expiration Time claim (exp) must be an integer.zSignature has expired)rrrr)r3rDrrqrPr4r4r5rs  zPyJWT._validate_expFrrcs|durd|vs |dsdStdd|vs|dstd|d|r@t|ts-tdtts6td|kr>tddSttrHgttsQtdtddDr^tdt|trf|g}tfd d|Drutd dS) NaudzInvalid audiencezInvalid audience (strict)z&Invalid claim format in token (strict)zAudience doesn't match (strict)zInvalid claim format in tokencss|] }t|t VqdSr)rTrO).0cr4r4r5 'sz&PyJWT._validate_aud..c3s|]}|vVqdSrr4)rraudience_claimsr4r5r-szAudience doesn't match)rrrTrOlistanyall)r3rDrlrr4rr5rs4     zPyJWT._validate_audcCs|durdSd|vrtd|d}t|tstdt|tr*||kr(tddSz ||vr3tdWdStyAtddw)NrSz%Payload Issuer (iss) must be a stringzInvalid issuerz.Issuer param must be "str" or "Container[str]")rrTrOrrV)r3rDrnrSr4r4r5r0s,   zPyJWT._validate_issr)r*r+r,r-)r,r$)r,r&)r*r+r,r$)rDrErFr'rGrHrIrJrKrLrMrNr,rO)NN)rDrErIrJrKrLr,r]) rNNNNNNNr)rdrerFr(rfrgr*r+rhrirjrkrlrmrnrorprHrqrrrsr r,rE)rrEr,rE)rdrerFrrfrgr*r+rhrirjrkrlrmrprHrnrorqrrrsr r,rE)NNNr)rDrEr*r$rlrrnrrprHrqrrr,r-)rDrErrr,r-)rDrErprHr,r-)rDrEr,r-)rDrErrrqrr,r-)rDrErlrmrrNr,r-)rDrErnrr,r-)__name__ __module__ __qualname__r6 staticmethodr.r0r/rrZrYr}r~rrrrrrrrrrr4r4r4r5r)*sd     E  n S *    2r))< __future__rrbosrxcalendarrcollections.abcrrrrrr typingr r r r api_jwsrrr exceptionsrrrrrrrrrrrNgetenvsys version_infor typing_extensionsrfr!r"api_jwkr#typesr$r%r&rOr]r'__annotations__r(r)_jwt_global_objr1rZr}rr4r4r4r5s> ,    $