Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.11/site-packages/jwt/api_jws.py: 22%

1from __future__ import annotations 

2 

3import binascii 

4import json 

5import warnings 

6from collections.abc import Sequence 

7from typing import TYPE_CHECKING, Any 

8 

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 

25 

26if TYPE_CHECKING: 

27 from .algorithms import AllowedPrivateKeys, AllowedPublicKeys 

28 from .types import SigOptions 

29 

30_ALGORITHM_UNSET = object() 

31_BASE64URL_ALPHABET = ( 

32 b"ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz0123456789-_" 

33) 

34 

35 

36class PyJWS: 

37 header_typ = "JWT" 

38 

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 ) 

48 

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] 

53 

54 self.options: SigOptions = self._get_default_options() 

55 if options is not None: 

56 self.options = {**self.options, **options} 

57 

58 @staticmethod 

59 def _get_default_options() -> SigOptions: 

60 return {"verify_signature": True, "enforce_minimum_key_length": False} 

61 

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. 

65 

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.") 

72 

73 if not isinstance(alg_obj, Algorithm): 

74 raise TypeError("Object is not of type `Algorithm`") 

75 

76 self._algorithms[alg_id] = alg_obj 

77 self._valid_algs.add(alg_id) 

78 

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 ) 

90 

91 del self._algorithms[alg_id] 

92 self._valid_algs.remove(alg_id) 

93 

94 def get_algorithms(self) -> list[str]: 

95 """ 

96 Returns a list of supported values for the `alg` parameter. 

97 

98 :rtype: list[str] 

99 """ 

100 return list(self._valid_algs) 

101 

102 def get_algorithm_by_name(self, alg_name: str) -> Algorithm: 

103 """ 

104 For a given string name, return the matching Algorithm object. 

105 

106 Example usage: 

107 >>> jws_obj = PyJWS() 

108 >>> jws_obj.get_algorithm_by_name("RS256") 

109 

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 

122 

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] = [] 

134 

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 

148 

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"] 

154 

155 headers_b64 = headers.get("b64") 

156 if headers_b64 is False: 

157 is_payload_detached = True 

158 

159 # Header 

160 header: dict[str, Any] = {"typ": self.header_typ, "alg": algorithm_} 

161 

162 if headers: 

163 self._validate_headers(headers, encoding=True) 

164 header.update(headers) 

165 

166 if not header["typ"]: 

167 del header["typ"] 

168 

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"] 

183 

184 json_header = json.dumps( 

185 header, separators=(",", ":"), cls=json_encoder, sort_keys=sort_headers 

186 ).encode() 

187 

188 segments.append(base64url_encode(json_header)) 

189 

190 if is_payload_detached: 

191 msg_payload = payload 

192 else: 

193 msg_payload = base64url_encode(payload) 

194 segments.append(msg_payload) 

195 

196 # Segments 

197 signing_input = b".".join(segments) 

198 

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) 

203 

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) 

210 

211 signature = alg_obj.sign(signing_input, key) 

212 

213 segments.append(base64url_encode(signature)) 

214 

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) 

219 

220 return encoded_string.decode("utf-8") 

221 

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} 

244 

245 verify_signature = merged_options["verify_signature"] 

246 

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 ) 

251 

252 payload, signing_input, header, signature = self._load(jwt) 

253 

254 self._validate_headers(header) 

255 

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 ) 

260 

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]) 

276 

277 if verify_signature: 

278 self._verify_signature( 

279 signing_input, 

280 header, 

281 signature, 

282 key, 

283 algorithms, 

284 options=merged_options, 

285 ) 

286 

287 return { 

288 "payload": payload, 

289 "header": header, 

290 "signature": signature, 

291 } 

292 

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"] 

314 

315 def get_unverified_header(self, jwt: str | bytes) -> dict[str, Any]: 

316 """Returns back the JWT header parameters as a `dict` 

317 

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) 

323 

324 return headers 

325 

326 @staticmethod 

327 def _decode_base64url_segment(segment: bytes, name: str) -> bytes: 

328 if len(segment) % 4 == 1 or any( 

329 character not in _BASE64URL_ALPHABET for character in segment 

330 ): 

331 raise DecodeError(f"Invalid {name} padding") 

332 

333 try: 

334 decoded = base64url_decode(segment) 

335 except (TypeError, binascii.Error) as err: 

336 raise DecodeError(f"Invalid {name} padding") from err 

337 

338 if base64url_encode(decoded) != segment: 

339 raise DecodeError(f"Invalid {name} padding") 

340 

341 return decoded 

342 

343 def _load(self, jwt: str | bytes) -> tuple[bytes, bytes, dict[str, Any], bytes]: 

344 if isinstance(jwt, str): 

345 jwt = jwt.encode("utf-8") 

346 

347 if not isinstance(jwt, bytes): 

348 raise DecodeError(f"Invalid token type. Token must be a {bytes}") 

349 

350 try: 

351 signing_input, crypto_segment = jwt.rsplit(b".", 1) 

352 header_segment, payload_segment = signing_input.split(b".", 1) 

353 except ValueError as err: 

354 raise DecodeError("Not enough segments") from err 

355 

356 header_data = self._decode_base64url_segment(header_segment, "header") 

357 

358 try: 

359 header: dict[str, Any] = json.loads(header_data) 

360 except (ValueError, RecursionError) as e: 

361 raise DecodeError(f"Invalid header string: {e}") from e 

362 

363 if not isinstance(header, dict): 

364 raise DecodeError("Invalid header string: must be a json object") 

365 

366 if header.get("b64", True) is False: 

367 # Detached payload form (RFC 7515 Appendix F): the compact-form 

368 # payload segment must be empty; the caller supplies the actual 

369 # payload via the `detached_payload` argument in decode_complete. 

370 # Skipping the base64 decode here removes an unauthenticated work 

371 # amplifier — otherwise an attacker can inflate the unused 

372 # segment to force CPU + memory cost before the signature is 

373 # even checked. 

374 if payload_segment: 

375 raise DecodeError("Payload segment must be empty when 'b64' is false.") 

376 payload = b"" 

377 else: 

378 payload = self._decode_base64url_segment(payload_segment, "payload") 

379 

380 signature = self._decode_base64url_segment(crypto_segment, "crypto") 

381 

382 return (payload, signing_input, header, signature) 

383 

384 def _verify_signature( 

385 self, 

386 signing_input: bytes, 

387 header: dict[str, Any], 

388 signature: bytes, 

389 key: AllowedPublicKeys | PyJWK | str | bytes = "", 

390 algorithms: Sequence[str] | None = None, 

391 options: SigOptions | None = None, 

392 ) -> None: 

393 effective_options = options if options is not None else self.options 

394 

395 if algorithms is None and isinstance(key, PyJWK): 

396 algorithms = [key.algorithm_name] 

397 try: 

398 alg = header["alg"] 

399 except KeyError: 

400 raise InvalidAlgorithmError("Algorithm not specified") from None 

401 

402 if not alg or (algorithms is not None and alg not in algorithms): 

403 raise InvalidAlgorithmError("The specified alg value is not allowed") 

404 

405 if isinstance(key, PyJWK): 

406 # The PyJWK has a fixed algorithm bound at construction time. 

407 # Verification must use that algorithm, not whatever the token 

408 # header advertises, otherwise the caller's allow-list check 

409 # above degenerates into a string compare with no behavioural 

410 # effect on which algorithm actually verifies the signature. 

411 if alg != key.algorithm_name: 

412 raise InvalidAlgorithmError( 

413 f"Token algorithm {alg!r} does not match the key's " 

414 f"algorithm {key.algorithm_name!r}" 

415 ) 

416 alg_obj = key.Algorithm 

417 prepared_key = alg_obj.prepare_key(key.key) 

418 else: 

419 try: 

420 alg_obj = self.get_algorithm_by_name(alg) 

421 except NotImplementedError as e: 

422 raise InvalidAlgorithmError("Algorithm not supported") from e 

423 prepared_key = alg_obj.prepare_key(key) 

424 

425 key_length_msg = alg_obj.check_key_length(prepared_key) 

426 if key_length_msg: 

427 if effective_options.get("enforce_minimum_key_length", False): 

428 raise InvalidKeyError(key_length_msg) 

429 else: 

430 warnings.warn(key_length_msg, InsecureKeyLengthWarning, stacklevel=4) 

431 

432 if not alg_obj.verify(signing_input, prepared_key, signature): 

433 raise InvalidSignatureError("Signature verification failed") 

434 

435 # Extensions that PyJWT actually understands and supports 

436 _supported_crit: set[str] = {"b64"} 

437 

438 def _validate_headers( 

439 self, headers: dict[str, Any], *, encoding: bool = False 

440 ) -> None: 

441 if "kid" in headers: 

442 self._validate_kid(headers["kid"]) 

443 if not encoding and "crit" in headers: 

444 self._validate_crit(headers) 

445 

446 def _validate_kid(self, kid: Any) -> None: 

447 if not isinstance(kid, str): 

448 raise InvalidTokenError("Key ID header parameter must be a string") 

449 

450 def _validate_crit(self, headers: dict[str, Any]) -> None: 

451 crit = headers["crit"] 

452 if not isinstance(crit, list) or len(crit) == 0: 

453 raise InvalidTokenError("Invalid 'crit' header: must be a non-empty list") 

454 for ext in crit: 

455 if not isinstance(ext, str): 

456 raise InvalidTokenError("Invalid 'crit' header: values must be strings") 

457 if ext not in self._supported_crit: 

458 raise InvalidTokenError(f"Unsupported critical extension: {ext}") 

459 if ext not in headers: 

460 raise InvalidTokenError( 

461 f"Critical extension '{ext}' is missing from headers" 

462 ) 

463 

464 

465_jws_global_obj = PyJWS() 

466encode = _jws_global_obj.encode 

467decode_complete = _jws_global_obj.decode_complete 

468decode = _jws_global_obj.decode 

469register_algorithm = _jws_global_obj.register_algorithm 

470unregister_algorithm = _jws_global_obj.unregister_algorithm 

471get_algorithm_by_name = _jws_global_obj.get_algorithm_by_name 

472get_unverified_header = _jws_global_obj.get_unverified_header