Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.11/site-packages/jwt/api_jws.py: 61%
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 binascii
4import json
5import warnings
6from collections.abc import Sequence
7from typing import TYPE_CHECKING, Any
9from .algorithms import (
10 Algorithm,
11 get_default_algorithms,
12 has_crypto,
13 requires_cryptography,
14)
15from .api_jwk import PyJWK
16from .exceptions import (
17 DecodeError,
18 InvalidAlgorithmError,
19 InvalidKeyError,
20 InvalidSignatureError,
21 InvalidTokenError,
22)
23from .utils import base64url_decode, base64url_encode
24from .warnings import InsecureKeyLengthWarning, RemovedInPyjwt3Warning
26if TYPE_CHECKING:
27 from .algorithms import AllowedPrivateKeys, AllowedPublicKeys
28 from .types import SigOptions
30_ALGORITHM_UNSET = object()
31_BASE64URL_ALPHABET = (
32 b"ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz0123456789-_"
33)
36class PyJWS:
37 header_typ = "JWT"
39 def __init__(
40 self,
41 algorithms: Sequence[str] | None = None,
42 options: SigOptions | None = None,
43 ) -> None:
44 self._algorithms = get_default_algorithms()
45 self._valid_algs = (
46 set(algorithms) if algorithms is not None else set(self._algorithms)
47 )
49 # Remove algorithms that aren't on the whitelist
50 for key in list(self._algorithms.keys()):
51 if key not in self._valid_algs:
52 del self._algorithms[key]
54 self.options: SigOptions = self._get_default_options()
55 if options is not None:
56 self.options = {**self.options, **options}
58 @staticmethod
59 def _get_default_options() -> SigOptions:
60 return {"verify_signature": True, "enforce_minimum_key_length": False}
62 def register_algorithm(self, alg_id: str, alg_obj: Algorithm) -> None:
63 """
64 Registers a new Algorithm for use when creating and verifying tokens.
66 :param str alg_id: the ID of the Algorithm
67 :param alg_obj: the Algorithm object
68 :type alg_obj: Algorithm
69 """
70 if alg_id in self._algorithms:
71 raise ValueError("Algorithm already has a handler.")
73 if not isinstance(alg_obj, Algorithm):
74 raise TypeError("Object is not of type `Algorithm`")
76 self._algorithms[alg_id] = alg_obj
77 self._valid_algs.add(alg_id)
79 def unregister_algorithm(self, alg_id: str) -> None:
80 """
81 Unregisters an Algorithm for use when creating and verifying tokens
82 :param str alg_id: the ID of the Algorithm
83 :raises KeyError: if algorithm is not registered.
84 """
85 if alg_id not in self._algorithms:
86 raise KeyError(
87 "The specified algorithm could not be removed"
88 " because it is not registered."
89 )
91 del self._algorithms[alg_id]
92 self._valid_algs.remove(alg_id)
94 def get_algorithms(self) -> list[str]:
95 """
96 Returns a list of supported values for the `alg` parameter.
98 :rtype: list[str]
99 """
100 return list(self._valid_algs)
102 def get_algorithm_by_name(self, alg_name: str) -> Algorithm:
103 """
104 For a given string name, return the matching Algorithm object.
106 Example usage:
107 >>> jws_obj = PyJWS()
108 >>> jws_obj.get_algorithm_by_name("RS256")
110 :param alg_name: The name of the algorithm to retrieve
111 :type alg_name: str
112 :rtype: Algorithm
113 """
114 try:
115 return self._algorithms[alg_name]
116 except KeyError as e:
117 if not has_crypto and alg_name in requires_cryptography:
118 raise NotImplementedError(
119 f"Algorithm '{alg_name}' could not be found. Do you have cryptography installed?"
120 ) from e
121 raise NotImplementedError("Algorithm not supported") from e
123 def encode(
124 self,
125 payload: bytes,
126 key: AllowedPrivateKeys | PyJWK | str | bytes,
127 algorithm: str | None = _ALGORITHM_UNSET, # type: ignore[assignment]
128 headers: dict[str, Any] | None = None,
129 json_encoder: type[json.JSONEncoder] | None = None,
130 is_payload_detached: bool = False,
131 sort_headers: bool = True,
132 ) -> str:
133 segments: list[bytes] = []
135 # declare a new var to narrow the type for type checkers
136 if algorithm is _ALGORITHM_UNSET:
137 if isinstance(key, PyJWK):
138 algorithm_ = key.algorithm_name
139 else:
140 algorithm_ = "HS256"
141 elif algorithm is None:
142 if isinstance(key, PyJWK):
143 algorithm_ = key.algorithm_name
144 else:
145 algorithm_ = "none"
146 else:
147 algorithm_ = algorithm
149 # Prefer headers values if present to function parameters.
150 if headers:
151 headers_alg = headers.get("alg")
152 if headers_alg:
153 algorithm_ = headers["alg"]
155 headers_b64 = headers.get("b64")
156 if headers_b64 is False:
157 is_payload_detached = True
159 # Header
160 header: dict[str, Any] = {"typ": self.header_typ, "alg": algorithm_}
162 if headers:
163 self._validate_headers(headers, encoding=True)
164 header.update(headers)
166 if not header["typ"]:
167 del header["typ"]
169 if is_payload_detached:
170 header["b64"] = False
171 # RFC 7797 §3: producers MUST list "b64" in "crit" whenever
172 # "b64" appears in the protected header, so b64-unaware
173 # verifiers don't silently treat an unencoded payload as
174 # base64-encoded.
175 existing_crit = header.get("crit", [])
176 if not isinstance(existing_crit, list):
177 raise InvalidTokenError("Invalid 'crit' header: must be a list")
178 if "b64" not in existing_crit:
179 header["crit"] = [*existing_crit, "b64"]
180 elif "b64" in header:
181 # True is the standard value for b64, so no need for it
182 del header["b64"]
184 json_header = json.dumps(
185 header, separators=(",", ":"), cls=json_encoder, sort_keys=sort_headers
186 ).encode()
188 segments.append(base64url_encode(json_header))
190 if is_payload_detached:
191 msg_payload = payload
192 else:
193 msg_payload = base64url_encode(payload)
194 segments.append(msg_payload)
196 # Segments
197 signing_input = b".".join(segments)
199 alg_obj = self.get_algorithm_by_name(algorithm_)
200 if isinstance(key, PyJWK):
201 key = key.key
202 key = alg_obj.prepare_key(key)
204 key_length_msg = alg_obj.check_key_length(key)
205 if key_length_msg:
206 if self.options.get("enforce_minimum_key_length", False):
207 raise InvalidKeyError(key_length_msg)
208 else:
209 warnings.warn(key_length_msg, InsecureKeyLengthWarning, stacklevel=2)
211 signature = alg_obj.sign(signing_input, key)
213 segments.append(base64url_encode(signature))
215 # Don't put the payload content inside the encoded token when detached
216 if is_payload_detached:
217 segments[1] = b""
218 encoded_string = b".".join(segments)
220 return encoded_string.decode("utf-8")
222 def decode_complete(
223 self,
224 jwt: str | bytes,
225 key: AllowedPublicKeys | PyJWK | str | bytes = "",
226 algorithms: Sequence[str] | None = None,
227 options: SigOptions | None = None,
228 detached_payload: bytes | None = None,
229 **kwargs: dict[str, Any],
230 ) -> dict[str, Any]:
231 if kwargs:
232 warnings.warn(
233 "passing additional kwargs to decode_complete() is deprecated "
234 "and will be removed in pyjwt version 3. "
235 f"Unsupported kwargs: {tuple(kwargs.keys())}",
236 RemovedInPyjwt3Warning,
237 stacklevel=2,
238 )
239 merged_options: SigOptions
240 if options is None:
241 merged_options = self.options
242 else:
243 merged_options = {**self.options, **options}
245 verify_signature = merged_options["verify_signature"]
247 if verify_signature and not algorithms and not isinstance(key, PyJWK):
248 raise DecodeError(
249 'It is required that you pass in a value for the "algorithms" argument when calling decode().'
250 )
252 payload, signing_input, header, signature = self._load(jwt)
254 self._validate_headers(header)
256 if detached_payload is not None and header.get("b64", True) is not False:
257 raise DecodeError(
258 'It is only valid to pass "detached_payload" when the protected header has "b64" set to false.'
259 )
261 if header.get("b64", True) is False:
262 # RFC 7797 §3: when "b64" is present in the protected header,
263 # it MUST also appear in "crit". A token that sets b64=false
264 # without declaring it critical is malformed.
265 crit = header.get("crit") or []
266 if not isinstance(crit, list) or "b64" not in crit:
267 raise InvalidTokenError(
268 "The 'b64' header parameter requires 'b64' to be listed in 'crit'."
269 )
270 if detached_payload is None:
271 raise DecodeError(
272 'It is required that you pass in a value for the "detached_payload" argument to decode a message having the b64 header set to false.'
273 )
274 payload = detached_payload
275 signing_input = b".".join([signing_input.rsplit(b".", 1)[0], payload])
277 if verify_signature:
278 self._verify_signature(
279 signing_input,
280 header,
281 signature,
282 key,
283 algorithms,
284 options=merged_options,
285 )
287 return {
288 "payload": payload,
289 "header": header,
290 "signature": signature,
291 }
293 def decode(
294 self,
295 jwt: str | bytes,
296 key: AllowedPublicKeys | PyJWK | str | bytes = "",
297 algorithms: Sequence[str] | None = None,
298 options: SigOptions | None = None,
299 detached_payload: bytes | None = None,
300 **kwargs: dict[str, Any],
301 ) -> Any:
302 if kwargs:
303 warnings.warn(
304 "passing additional kwargs to decode() is deprecated "
305 "and will be removed in pyjwt version 3. "
306 f"Unsupported kwargs: {tuple(kwargs.keys())}",
307 RemovedInPyjwt3Warning,
308 stacklevel=2,
309 )
310 decoded = self.decode_complete(
311 jwt, key, algorithms, options, detached_payload=detached_payload
312 )
313 return decoded["payload"]
315 def get_unverified_header(self, jwt: str | bytes) -> dict[str, Any]:
316 """Returns back the JWT header parameters as a `dict`
318 Note: The signature is not verified so the header parameters
319 should not be fully trusted until signature verification is complete
320 """
321 headers = self._load(jwt)[2]
322 self._validate_headers(headers)
324 return headers
326 @staticmethod
327 def _decode_base64url_segment(segment: bytes, name: str) -> bytes:
328 # Accept trailing '=' used by some issuers (e.g. AWS ALB). Still
329 # reject non-alphabet junk such as '!!!!' (GHSA-hxm8-2xgr-2p9m).
330 padding = 0
331 stripped = segment
332 while stripped.endswith(b"="):
333 stripped = stripped[:-1]
334 padding += 1
335 if padding > 2:
336 raise DecodeError(f"Invalid {name} padding")
338 if padding and len(segment) % 4 != 0:
339 raise DecodeError(f"Invalid {name} padding")
341 if len(stripped) % 4 == 1 or any(
342 character not in _BASE64URL_ALPHABET for character in stripped
343 ):
344 raise DecodeError(f"Invalid {name} padding")
346 try:
347 decoded = base64url_decode(stripped)
348 except (TypeError, binascii.Error) as err:
349 raise DecodeError(f"Invalid {name} padding") from err
351 if base64url_encode(decoded) != stripped:
352 raise DecodeError(f"Invalid {name} padding")
354 return decoded
356 def _load(self, jwt: str | bytes) -> tuple[bytes, bytes, dict[str, Any], bytes]:
357 if isinstance(jwt, str):
358 jwt = jwt.encode("utf-8")
360 if not isinstance(jwt, bytes):
361 raise DecodeError(f"Invalid token type. Token must be a {bytes}")
363 try:
364 signing_input, crypto_segment = jwt.rsplit(b".", 1)
365 header_segment, payload_segment = signing_input.split(b".", 1)
366 except ValueError as err:
367 raise DecodeError("Not enough segments") from err
369 header_data = self._decode_base64url_segment(header_segment, "header")
371 try:
372 header: dict[str, Any] = json.loads(header_data)
373 except (ValueError, RecursionError) as e:
374 raise DecodeError(f"Invalid header string: {e}") from e
376 if not isinstance(header, dict):
377 raise DecodeError("Invalid header string: must be a json object")
379 if header.get("b64", True) is False:
380 # Detached payload form (RFC 7515 Appendix F): the compact-form
381 # payload segment must be empty; the caller supplies the actual
382 # payload via the `detached_payload` argument in decode_complete.
383 # Skipping the base64 decode here removes an unauthenticated work
384 # amplifier — otherwise an attacker can inflate the unused
385 # segment to force CPU + memory cost before the signature is
386 # even checked.
387 if payload_segment:
388 raise DecodeError("Payload segment must be empty when 'b64' is false.")
389 payload = b""
390 else:
391 payload = self._decode_base64url_segment(payload_segment, "payload")
393 signature = self._decode_base64url_segment(crypto_segment, "crypto")
395 return (payload, signing_input, header, signature)
397 def _verify_signature(
398 self,
399 signing_input: bytes,
400 header: dict[str, Any],
401 signature: bytes,
402 key: AllowedPublicKeys | PyJWK | str | bytes = "",
403 algorithms: Sequence[str] | None = None,
404 options: SigOptions | None = None,
405 ) -> None:
406 effective_options = options if options is not None else self.options
408 if algorithms is None and isinstance(key, PyJWK):
409 algorithms = [key.algorithm_name]
410 try:
411 alg = header["alg"]
412 except KeyError:
413 raise InvalidAlgorithmError("Algorithm not specified") from None
415 if not alg or (algorithms is not None and alg not in algorithms):
416 raise InvalidAlgorithmError("The specified alg value is not allowed")
418 if isinstance(key, PyJWK):
419 # The PyJWK has a fixed algorithm bound at construction time.
420 # Verification must use that algorithm, not whatever the token
421 # header advertises, otherwise the caller's allow-list check
422 # above degenerates into a string compare with no behavioural
423 # effect on which algorithm actually verifies the signature.
424 if alg != key.algorithm_name:
425 raise InvalidAlgorithmError(
426 f"Token algorithm {alg!r} does not match the key's "
427 f"algorithm {key.algorithm_name!r}"
428 )
429 alg_obj = key.Algorithm
430 prepared_key = alg_obj.prepare_key(key.key)
431 else:
432 try:
433 alg_obj = self.get_algorithm_by_name(alg)
434 except NotImplementedError as e:
435 raise InvalidAlgorithmError("Algorithm not supported") from e
436 prepared_key = alg_obj.prepare_key(key)
438 key_length_msg = alg_obj.check_key_length(prepared_key)
439 if key_length_msg:
440 if effective_options.get("enforce_minimum_key_length", False):
441 raise InvalidKeyError(key_length_msg)
442 else:
443 warnings.warn(key_length_msg, InsecureKeyLengthWarning, stacklevel=4)
445 if not alg_obj.verify(signing_input, prepared_key, signature):
446 raise InvalidSignatureError("Signature verification failed")
448 # Extensions that PyJWT actually understands and supports
449 _supported_crit: set[str] = {"b64"}
451 def _validate_headers(
452 self, headers: dict[str, Any], *, encoding: bool = False
453 ) -> None:
454 if "kid" in headers:
455 self._validate_kid(headers["kid"])
456 if not encoding and "crit" in headers:
457 self._validate_crit(headers)
459 def _validate_kid(self, kid: Any) -> None:
460 if not isinstance(kid, str):
461 raise InvalidTokenError("Key ID header parameter must be a string")
463 def _validate_crit(self, headers: dict[str, Any]) -> None:
464 crit = headers["crit"]
465 if not isinstance(crit, list) or len(crit) == 0:
466 raise InvalidTokenError("Invalid 'crit' header: must be a non-empty list")
467 for ext in crit:
468 if not isinstance(ext, str):
469 raise InvalidTokenError("Invalid 'crit' header: values must be strings")
470 if ext not in self._supported_crit:
471 raise InvalidTokenError(f"Unsupported critical extension: {ext}")
472 if ext not in headers:
473 raise InvalidTokenError(
474 f"Critical extension '{ext}' is missing from headers"
475 )
478_jws_global_obj = PyJWS()
479encode = _jws_global_obj.encode
480decode_complete = _jws_global_obj.decode_complete
481decode = _jws_global_obj.decode
482register_algorithm = _jws_global_obj.register_algorithm
483unregister_algorithm = _jws_global_obj.unregister_algorithm
484get_algorithm_by_name = _jws_global_obj.get_algorithm_by_name
485get_unverified_header = _jws_global_obj.get_unverified_header