Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.11/site-packages/jwt/api_jwt.py: 60%
Shortcuts on this page
r m x toggle line displays
j k next/prev highlighted chunk
0 (zero) top of page
1 (one) first highlighted chunk
1from __future__ import annotations
3import json
4import os
5import warnings
6from calendar import timegm
7from collections.abc import Container, Iterable, Sequence
8from datetime import datetime, timedelta, timezone
9from typing import TYPE_CHECKING, Any, Union, cast
11from .api_jws import PyJWS, _ALGORITHM_UNSET, _jws_global_obj
12from .exceptions import (
13 DecodeError,
14 ExpiredSignatureError,
15 ImmatureSignatureError,
16 InvalidAudienceError,
17 InvalidIssuedAtError,
18 InvalidIssuerError,
19 InvalidJTIError,
20 InvalidSubjectError,
21 MissingRequiredClaimError,
22)
23from .warnings import RemovedInPyjwt3Warning
25if TYPE_CHECKING or bool(os.getenv("SPHINX_BUILD", "")):
26 import sys
28 if sys.version_info >= (3, 10):
29 from typing import TypeAlias
30 else:
31 # Python 3.9 and lower
32 from typing_extensions import TypeAlias
34 from .algorithms import AllowedPrivateKeys, AllowedPublicKeys
35 from .api_jwk import PyJWK
36 from .types import FullOptions, Options, SigOptions
38 AllowedPrivateKeyTypes: TypeAlias = Union[AllowedPrivateKeys, PyJWK, str, bytes]
39 AllowedPublicKeyTypes: TypeAlias = Union[AllowedPublicKeys, PyJWK, str, bytes]
42class PyJWT:
43 def __init__(self, options: Options | None = None) -> None:
44 self.options: FullOptions
45 self.options = self._get_default_options()
46 if options is not None:
47 self.options = self._merge_options(options)
49 self._jws = PyJWS(options=self._get_sig_options())
51 @staticmethod
52 def _get_default_options() -> FullOptions:
53 return {
54 "verify_signature": True,
55 "verify_exp": True,
56 "verify_nbf": True,
57 "verify_iat": True,
58 "verify_aud": True,
59 "verify_iss": True,
60 "verify_sub": True,
61 "verify_jti": True,
62 "require": [],
63 "strict_aud": False,
64 "enforce_minimum_key_length": False,
65 }
67 def _get_sig_options(self) -> SigOptions:
68 return {
69 "verify_signature": self.options["verify_signature"],
70 "enforce_minimum_key_length": self.options.get(
71 "enforce_minimum_key_length", False
72 ),
73 }
75 def _merge_options(self, options: Options | None = None) -> FullOptions:
76 if options is None:
77 return self.options
79 merged_options = cast("Options", dict(options))
81 # (defensive) set defaults for verify_x to False if verify_signature is False
82 if not merged_options.get("verify_signature", True):
83 merged_options["verify_exp"] = merged_options.get("verify_exp", False)
84 merged_options["verify_nbf"] = merged_options.get("verify_nbf", False)
85 merged_options["verify_iat"] = merged_options.get("verify_iat", False)
86 merged_options["verify_aud"] = merged_options.get("verify_aud", False)
87 merged_options["verify_iss"] = merged_options.get("verify_iss", False)
88 merged_options["verify_sub"] = merged_options.get("verify_sub", False)
89 merged_options["verify_jti"] = merged_options.get("verify_jti", False)
90 return {**self.options, **merged_options}
92 def encode(
93 self,
94 payload: dict[str, Any],
95 key: AllowedPrivateKeyTypes,
96 algorithm: str | None = _ALGORITHM_UNSET, # type: ignore[assignment]
97 headers: dict[str, Any] | None = None,
98 json_encoder: type[json.JSONEncoder] | None = None,
99 sort_headers: bool = True,
100 ) -> str:
101 """Encode the ``payload`` as JSON Web Token.
103 :param payload: JWT claims, e.g. ``dict(iss=..., aud=..., sub=...)``
104 :type payload: dict[str, typing.Any]
105 :param key: a key suitable for the chosen algorithm:
107 * for **asymmetric algorithms**: PEM-formatted private key, a multiline string
108 * for **symmetric algorithms**: plain string, sufficiently long for security
110 :type key: str or bytes or PyJWK or :py:class:`jwt.algorithms.AllowedPrivateKeys`
111 :param algorithm: algorithm to sign the token with, e.g. ``"ES256"``.
112 If ``headers`` includes ``alg``, it will be preferred to this parameter.
113 If ``key`` is a :class:`PyJWK` object, by default the key algorithm will be used.
114 :type algorithm: str or None
115 :param headers: additional JWT header fields, e.g. ``dict(kid="my-key-id")``.
116 :type headers: dict[str, typing.Any] or None
117 :param json_encoder: custom JSON encoder for ``payload`` and ``headers``
118 :type json_encoder: json.JSONEncoder or None
120 :rtype: str
121 :returns: a JSON Web Token
123 :raises TypeError: if ``payload`` is not a ``dict``
124 """
125 # Check that we get a dict
126 if not isinstance(payload, dict):
127 raise TypeError(
128 "Expecting a dict object, as JWT only supports "
129 "JSON objects as payloads."
130 )
132 # Payload
133 payload = payload.copy()
134 for time_claim in ["exp", "iat", "nbf"]:
135 # Convert datetime to a intDate value in known time-format claims
136 if isinstance(payload.get(time_claim), datetime):
137 payload[time_claim] = timegm(payload[time_claim].utctimetuple())
139 # Issue #1039, iss being set to non-string
140 if "iss" in payload and not isinstance(payload["iss"], str):
141 raise TypeError("Issuer (iss) must be a string.")
143 json_payload = self._encode_payload(
144 payload,
145 headers=headers,
146 json_encoder=json_encoder,
147 )
149 return self._jws.encode(
150 json_payload,
151 key,
152 algorithm,
153 headers,
154 json_encoder,
155 sort_headers=sort_headers,
156 )
158 def _encode_payload(
159 self,
160 payload: dict[str, Any],
161 headers: dict[str, Any] | None = None,
162 json_encoder: type[json.JSONEncoder] | None = None,
163 ) -> bytes:
164 """
165 Encode a given payload to the bytes to be signed.
167 This method is intended to be overridden by subclasses that need to
168 encode the payload in a different way, e.g. compress the payload.
169 """
170 return json.dumps(
171 payload,
172 separators=(",", ":"),
173 cls=json_encoder,
174 ).encode("utf-8")
176 def decode_complete(
177 self,
178 jwt: str | bytes,
179 key: AllowedPublicKeyTypes = "",
180 algorithms: Sequence[str] | None = None,
181 options: Options | None = None,
182 # deprecated arg, remove in pyjwt3
183 verify: bool | None = None,
184 # could be used as passthrough to api_jws, consider removal in pyjwt3
185 detached_payload: bytes | None = None,
186 # passthrough arguments to _validate_claims
187 # consider putting in options
188 audience: str | Iterable[str] | None = None,
189 issuer: str | Container[str] | None = None,
190 subject: str | None = None,
191 leeway: float | timedelta = 0,
192 # kwargs
193 **kwargs: Any,
194 ) -> dict[str, Any]:
195 """Identical to ``jwt.decode`` except for return value which is a dictionary containing the token header (JOSE Header),
196 the token payload (JWT Payload), and token signature (JWT Signature) on the keys "header", "payload",
197 and "signature" respectively.
199 :param jwt: the token to be decoded
200 :type jwt: str or bytes
201 :param key: the key suitable for the allowed algorithm
202 :type key: str or bytes or PyJWK or :py:class:`jwt.algorithms.AllowedPublicKeys`
204 :param algorithms: allowed algorithms, e.g. ``["ES256"]``
206 .. warning::
208 Do **not** compute the ``algorithms`` parameter based on
209 the ``alg`` from the token itself, or on any other data
210 that an attacker may be able to influence, as that might
211 expose you to various vulnerabilities (see `RFC 8725 §2.1
212 <https://www.rfc-editor.org/rfc/rfc8725.html#section-2.1>`_). Instead,
213 either hard-code a fixed value for ``algorithms``, or
214 configure it in the same place you configure the
215 ``key``. Make sure not to mix symmetric and asymmetric
216 algorithms that interpret the ``key`` in different ways
217 (e.g. HS\\* and RS\\*).
218 :type algorithms: typing.Sequence[str] or None
220 :param jwt.types.Options options: extended decoding and validation options
221 Refer to :py:class:`jwt.types.Options` for more information.
223 :param audience: optional, the value for ``verify_aud`` check
224 :type audience: str or typing.Iterable[str] or None
225 :param issuer: optional, the value for ``verify_iss`` check
226 :type issuer: str or typing.Container[str] or None
227 :param leeway: a time margin in seconds for the expiration check
228 :type leeway: float or datetime.timedelta
229 :rtype: dict[str, typing.Any]
230 :returns: Decoded JWT with the JOSE Header on the key ``header``, the JWS
231 Payload on the key ``payload``, and the JWS Signature on the key ``signature``.
232 """
233 if kwargs:
234 warnings.warn(
235 "passing additional kwargs to decode_complete() is deprecated "
236 "and will be removed in pyjwt version 3. "
237 f"Unsupported kwargs: {tuple(kwargs.keys())}",
238 RemovedInPyjwt3Warning,
239 stacklevel=2,
240 )
242 if options is None:
243 verify_signature = True
244 else:
245 verify_signature = options.get("verify_signature", True)
247 # If the user has set the legacy `verify` argument, and it doesn't match
248 # what the relevant `options` entry for the argument is, inform the user
249 # that they're likely making a mistake.
250 if verify is not None and verify != verify_signature:
251 warnings.warn(
252 "The `verify` argument to `decode` does nothing in PyJWT 2.0 and newer. "
253 "The equivalent is setting `verify_signature` to False in the `options` dictionary. "
254 "This invocation has a mismatch between the kwarg and the option entry.",
255 category=DeprecationWarning,
256 stacklevel=2,
257 )
259 merged_options = self._merge_options(options)
261 sig_options: SigOptions = {
262 "verify_signature": verify_signature,
263 "enforce_minimum_key_length": merged_options.get(
264 "enforce_minimum_key_length", False
265 ),
266 }
267 decoded = self._jws.decode_complete(
268 jwt,
269 key=key,
270 algorithms=algorithms,
271 options=sig_options,
272 detached_payload=detached_payload,
273 )
275 payload = self._decode_payload(decoded)
277 self._validate_claims(
278 payload,
279 merged_options,
280 audience=audience,
281 issuer=issuer,
282 leeway=leeway,
283 subject=subject,
284 )
286 decoded["payload"] = payload
287 return decoded
289 def _decode_payload(self, decoded: dict[str, Any]) -> dict[str, Any]:
290 """
291 Decode the payload from a JWS dictionary (payload, signature, header).
293 This method is intended to be overridden by subclasses that need to
294 decode the payload in a different way, e.g. decompress compressed
295 payloads.
296 """
297 try:
298 payload: dict[str, Any] = json.loads(decoded["payload"])
299 except ValueError as e:
300 raise DecodeError(f"Invalid payload string: {e}") from e
301 if not isinstance(payload, dict):
302 raise DecodeError("Invalid payload string: must be a json object")
303 return payload
305 def decode(
306 self,
307 jwt: str | bytes,
308 key: AllowedPublicKeys | PyJWK | str | bytes = "",
309 algorithms: Sequence[str] | None = None,
310 options: Options | None = None,
311 # deprecated arg, remove in pyjwt3
312 verify: bool | None = None,
313 # could be used as passthrough to api_jws, consider removal in pyjwt3
314 detached_payload: bytes | None = None,
315 # passthrough arguments to _validate_claims
316 # consider putting in options
317 audience: str | Iterable[str] | None = None,
318 subject: str | None = None,
319 issuer: str | Container[str] | None = None,
320 leeway: float | timedelta = 0,
321 # kwargs
322 **kwargs: Any,
323 ) -> dict[str, Any]:
324 """Verify the ``jwt`` token signature and return the token claims.
326 :param jwt: the token to be decoded
327 :type jwt: str or bytes
328 :param key: the key suitable for the allowed algorithm
329 :type key: str or bytes or PyJWK or :py:class:`jwt.algorithms.AllowedPublicKeys`
331 :param algorithms: allowed algorithms, e.g. ``["ES256"]``
332 If ``key`` is a :class:`PyJWK` object, allowed algorithms will default to the key algorithm.
334 .. warning::
336 Do **not** compute the ``algorithms`` parameter based on
337 the ``alg`` from the token itself, or on any other data
338 that an attacker may be able to influence, as that might
339 expose you to various vulnerabilities (see `RFC 8725 §2.1
340 <https://www.rfc-editor.org/rfc/rfc8725.html#section-2.1>`_). Instead,
341 either hard-code a fixed value for ``algorithms``, or
342 configure it in the same place you configure the
343 ``key``. Make sure not to mix symmetric and asymmetric
344 algorithms that interpret the ``key`` in different ways
345 (e.g. HS\\* and RS\\*).
346 :type algorithms: typing.Sequence[str] or None
348 :param jwt.types.Options options: extended decoding and validation options
349 Refer to :py:class:`jwt.types.Options` for more information.
351 :param audience: optional, the value for ``verify_aud`` check
352 :type audience: str or typing.Iterable[str] or None
353 :param subject: optional, the value for ``verify_sub`` check
354 :type subject: str or None
355 :param issuer: optional, the value for ``verify_iss`` check
356 :type issuer: str or typing.Container[str] or None
357 :param leeway: a time margin in seconds for the expiration check
358 :type leeway: float or datetime.timedelta
359 :rtype: dict[str, typing.Any]
360 :returns: the JWT claims
361 """
362 if kwargs:
363 warnings.warn(
364 "passing additional kwargs to decode() is deprecated "
365 "and will be removed in pyjwt version 3. "
366 f"Unsupported kwargs: {tuple(kwargs.keys())}",
367 RemovedInPyjwt3Warning,
368 stacklevel=2,
369 )
370 decoded = self.decode_complete(
371 jwt,
372 key,
373 algorithms,
374 options,
375 verify=verify,
376 detached_payload=detached_payload,
377 audience=audience,
378 subject=subject,
379 issuer=issuer,
380 leeway=leeway,
381 )
382 return cast(dict[str, Any], decoded["payload"])
384 def _validate_claims(
385 self,
386 payload: dict[str, Any],
387 options: FullOptions,
388 audience: Iterable[str] | str | None = None,
389 issuer: Container[str] | str | None = None,
390 subject: str | None = None,
391 leeway: float | timedelta = 0,
392 ) -> None:
393 if isinstance(leeway, timedelta):
394 leeway = leeway.total_seconds()
396 if audience is not None and not isinstance(audience, (str, Iterable)):
397 raise TypeError("audience must be a string, iterable or None")
399 self._validate_required_claims(payload, options["require"])
401 now = datetime.now(tz=timezone.utc).timestamp()
403 if "iat" in payload and options["verify_iat"]:
404 self._validate_iat(payload, now, leeway)
406 if "nbf" in payload and options["verify_nbf"]:
407 self._validate_nbf(payload, now, leeway)
409 if "exp" in payload and options["verify_exp"]:
410 self._validate_exp(payload, now, leeway)
412 if options["verify_iss"]:
413 self._validate_iss(payload, issuer)
415 if options["verify_aud"]:
416 self._validate_aud(
417 payload, audience, strict=options.get("strict_aud", False)
418 )
420 if options["verify_sub"]:
421 self._validate_sub(payload, subject)
423 if options["verify_jti"]:
424 self._validate_jti(payload)
426 def _validate_required_claims(
427 self,
428 payload: dict[str, Any],
429 claims: Iterable[str],
430 ) -> None:
431 for claim in claims:
432 if payload.get(claim) is None:
433 raise MissingRequiredClaimError(claim)
435 def _validate_sub(
436 self, payload: dict[str, Any], subject: str | None = None
437 ) -> None:
438 """
439 Checks whether "sub" if in the payload is valid or not.
440 This is an Optional claim
442 :param payload(dict): The payload which needs to be validated
443 :param subject(str): The subject of the token
444 """
446 if "sub" not in payload:
447 return
449 if not isinstance(payload["sub"], str):
450 raise InvalidSubjectError("Subject must be a string")
452 if subject is not None:
453 if payload.get("sub") != subject:
454 raise InvalidSubjectError("Invalid subject")
456 def _validate_jti(self, payload: dict[str, Any]) -> None:
457 """
458 Checks whether "jti" if in the payload is valid or not
459 This is an Optional claim
461 :param payload(dict): The payload which needs to be validated
462 """
464 if "jti" not in payload:
465 return
467 if not isinstance(payload.get("jti"), str):
468 raise InvalidJTIError("JWT ID must be a string")
470 def _validate_iat(
471 self,
472 payload: dict[str, Any],
473 now: float,
474 leeway: float,
475 ) -> None:
476 try:
477 iat = int(payload["iat"])
478 except ValueError:
479 raise InvalidIssuedAtError(
480 "Issued At claim (iat) must be an integer."
481 ) from None
482 if iat > (now + leeway):
483 raise ImmatureSignatureError("The token is not yet valid (iat)")
485 def _validate_nbf(
486 self,
487 payload: dict[str, Any],
488 now: float,
489 leeway: float,
490 ) -> None:
491 try:
492 nbf = int(payload["nbf"])
493 except ValueError:
494 raise DecodeError("Not Before claim (nbf) must be an integer.") from None
496 if nbf > (now + leeway):
497 raise ImmatureSignatureError("The token is not yet valid (nbf)")
499 def _validate_exp(
500 self,
501 payload: dict[str, Any],
502 now: float,
503 leeway: float,
504 ) -> None:
505 try:
506 exp = int(payload["exp"])
507 except ValueError:
508 raise DecodeError(
509 "Expiration Time claim (exp) must be an integer."
510 ) from None
512 if exp <= (now - leeway):
513 raise ExpiredSignatureError("Signature has expired")
515 def _validate_aud(
516 self,
517 payload: dict[str, Any],
518 audience: str | Iterable[str] | None,
519 *,
520 strict: bool = False,
521 ) -> None:
522 if audience is None:
523 if "aud" not in payload or not payload["aud"]:
524 return
525 # Application did not specify an audience, but
526 # the token has the 'aud' claim
527 raise InvalidAudienceError("Invalid audience")
529 if "aud" not in payload or not payload["aud"]:
530 # Application specified an audience, but it could not be
531 # verified since the token does not contain a claim.
532 raise MissingRequiredClaimError("aud")
534 audience_claims = payload["aud"]
536 # In strict mode, we forbid list matching: the supplied audience
537 # must be a string, and it must exactly match the audience claim.
538 if strict:
539 # Only a single audience is allowed in strict mode.
540 if not isinstance(audience, str):
541 raise InvalidAudienceError("Invalid audience (strict)")
543 # Only a single audience claim is allowed in strict mode.
544 if not isinstance(audience_claims, str):
545 raise InvalidAudienceError("Invalid claim format in token (strict)")
547 if audience != audience_claims:
548 raise InvalidAudienceError("Audience doesn't match (strict)")
550 return
552 if isinstance(audience_claims, str):
553 audience_claims = [audience_claims]
554 if not isinstance(audience_claims, list):
555 raise InvalidAudienceError("Invalid claim format in token")
556 if any(not isinstance(c, str) for c in audience_claims):
557 raise InvalidAudienceError("Invalid claim format in token")
559 if isinstance(audience, str):
560 audience = [audience]
562 if all(aud not in audience_claims for aud in audience):
563 raise InvalidAudienceError("Audience doesn't match")
565 def _validate_iss(
566 self, payload: dict[str, Any], issuer: Container[str] | str | None
567 ) -> None:
568 if issuer is None:
569 return
571 if "iss" not in payload:
572 raise MissingRequiredClaimError("iss")
574 iss = payload["iss"]
575 if not isinstance(iss, str):
576 raise InvalidIssuerError("Payload Issuer (iss) must be a string")
578 if isinstance(issuer, str):
579 if iss != issuer:
580 raise InvalidIssuerError("Invalid issuer")
581 else:
582 try:
583 if iss not in issuer:
584 raise InvalidIssuerError("Invalid issuer")
585 except TypeError:
586 raise InvalidIssuerError(
587 'Issuer param must be "str" or "Container[str]"'
588 ) from None
591_jwt_global_obj = PyJWT()
592_jwt_global_obj._jws = _jws_global_obj
593encode = _jwt_global_obj.encode
594decode_complete = _jwt_global_obj.decode_complete
595decode = _jwt_global_obj.decode