Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.11/site-packages/jose/jwe.py: 28%

1import binascii

2import json

3import zlib

4from collections.abc import Mapping

5from struct import pack

7from . import jwk

8from .backends import get_random_bytes

9from .constants import ALGORITHMS, JWE_SIZE_LIMIT, ZIPS

10from .exceptions import JWEError, JWEParseError

11from .utils import base64url_decode, base64url_encode, ensure_binary

14def encrypt(plaintext, key, encryption=ALGORITHMS.A256GCM, algorithm=ALGORITHMS.DIR, zip=None, cty=None, kid=None):

15 """Encrypts plaintext and returns a JWE compact serialization string.

17 Args:

18 plaintext (bytes): A bytes object to encrypt

19 key (str or dict): The key(s) to use for encrypting the content. Can be

20 individual JWK or JWK set.

21 encryption (str, optional): The content encryption algorithm used to

22 perform authenticated encryption on the plaintext to produce the

23 ciphertext and the Authentication Tag. Defaults to A256GCM.

24 algorithm (str, optional): The cryptographic algorithm used

25 to encrypt or determine the value of the CEK. Defaults to dir.

26 zip (str, optional): The compression algorithm) applied to the

27 plaintext before encryption. Defaults to None.

28 cty (str, optional): The media type for the secured content.

29 See http://www.iana.org/assignments/media-types/media-types.xhtml

30 kid (str, optional): Key ID for the provided key

32 Returns:

33 bytes: The string representation of the header, encrypted key,

34 initialization vector, ciphertext, and authentication tag.

36 Raises:

37 JWEError: If there is an error signing the token.

39 Examples:

40 >>> from jose import jwe

41 >>> jwe.encrypt('Hello, World!', 'asecret128bitkey', algorithm='dir', encryption='A128GCM')

42 'eyJhbGciOiJkaXIiLCJlbmMiOiJBMTI4R0NNIn0..McILMB3dYsNJSuhcDzQshA.OfX9H_mcUpHDeRM4IA.CcnTWqaqxNsjT4eCaUABSg'

44 """

45 plaintext = ensure_binary(plaintext) # Make sure it's bytes

46 if algorithm not in ALGORITHMS.SUPPORTED:

47 raise JWEError("Algorithm %s not supported." % algorithm)

48 if encryption not in ALGORITHMS.SUPPORTED:

49 raise JWEError("Algorithm %s not supported." % encryption)

50 key = jwk.construct(key, algorithm)

51 encoded_header = _encoded_header(algorithm, encryption, zip, cty, kid)

53 plaintext = _compress(zip, plaintext)

54 enc_cek, iv, cipher_text, auth_tag = _encrypt_and_auth(key, algorithm, encryption, zip, plaintext, encoded_header)

56 jwe_string = _jwe_compact_serialize(encoded_header, enc_cek, iv, cipher_text, auth_tag)

57 return jwe_string

60def decrypt(jwe_str, key):

61 """Decrypts a JWE compact serialized string and returns the plaintext.

63 Args:

64 jwe_str (str): A JWE to be decrypt.

65 key (str or dict): A key to attempt to decrypt the payload with. Can be

66 individual JWK or JWK set.

68 Returns:

69 bytes: The plaintext bytes, assuming the authentication tag is valid.

71 Raises:

72 JWEError: If there is an exception verifying the token.

74 Examples:

75 >>> from jose import jwe

76 >>> jwe.decrypt(jwe_string, 'asecret128bitkey')

77 'Hello, World!'

78 """

80 # Limit the token size - if the data is compressed then decompressing the

81 # data could lead to large memory usage. This helps address This addresses

82 # CVE-2024-33664. Also see _decompress()

83 if len(jwe_str) > JWE_SIZE_LIMIT:

84 raise JWEError(f"JWE string {len(jwe_str)} bytes exceeds {JWE_SIZE_LIMIT} bytes")

86 header, encoded_header, encrypted_key, iv, cipher_text, auth_tag = _jwe_compact_deserialize(jwe_str)

88 # Verify that the implementation understands and can process all

89 # fields that it is required to support, whether required by this

90 # specification, by the algorithms being used, or by the "crit"

91 # Header Parameter value, and that the values of those parameters

92 # are also understood and supported.

94 try:

95 # Determine the Key Management Mode employed by the algorithm

96 # specified by the "alg" (algorithm) Header Parameter.

97 alg = header["alg"]

98 enc = header["enc"]

99 if alg not in ALGORITHMS.SUPPORTED:

100 raise JWEError("Algorithm %s not supported." % alg)

101 if enc not in ALGORITHMS.SUPPORTED:

102 raise JWEError("Algorithm %s not supported." % enc)

103

104 except KeyError:

105 raise JWEParseError("alg and enc headers are required!")

106

107 # Verify that the JWE uses a key known to the recipient.

108 key = jwk.construct(key, alg)

109

110 # When Direct Key Agreement or Key Agreement with Key Wrapping are

111 # employed, use the key agreement algorithm to compute the value

112 # of the agreed upon key. When Direct Key Agreement is employed,

113 # let the CEK be the agreed upon key. When Key Agreement with Key

114 # Wrapping is employed, the agreed upon key will be used to

115 # decrypt the JWE Encrypted Key.

116 #

117 # When Key Wrapping, Key Encryption, or Key Agreement with Key

118 # Wrapping are employed, decrypt the JWE Encrypted Key to produce

119 # the CEK. The CEK MUST have a length equal to that required for

120 # the content encryption algorithm. Note that when there are

121 # multiple recipients, each recipient will only be able to decrypt

122 # JWE Encrypted Key values that were encrypted to a key in that

123 # recipient's possession. It is therefore normal to only be able

124 # to decrypt one of the per-recipient JWE Encrypted Key values to

125 # obtain the CEK value. Also, see Section 11.5 for security

126 # considerations on mitigating timing attacks.

127 if alg == ALGORITHMS.DIR:

128 # When Direct Key Agreement or Direct Encryption are employed,

129 # verify that the JWE Encrypted Key value is an empty octet

130 # sequence.

131

132 # Record whether the CEK could be successfully determined for this

133 # recipient or not.

134 cek_valid = encrypted_key == b""

135

136 # When Direct Encryption is employed, let the CEK be the shared

137 # symmetric key.

138 cek_bytes = _get_key_bytes_from_key(key)

139 else:

140 try:

141 cek_bytes = key.unwrap_key(encrypted_key)

142

143 # Record whether the CEK could be successfully determined for this

144 # recipient or not.

145 cek_valid = True

146 except NotImplementedError:

147 raise JWEError(f"alg {alg} is not implemented")

148 except Exception:

149 # Record whether the CEK could be successfully determined for this

150 # recipient or not.

151 cek_valid = False

152

153 # To mitigate the attacks described in RFC 3218 [RFC3218], the

154 # recipient MUST NOT distinguish between format, padding, and length

155 # errors of encrypted keys. It is strongly recommended, in the event

156 # of receiving an improperly formatted key, that the recipient

157 # substitute a randomly generated CEK and proceed to the next step, to

158 # mitigate timing attacks.

159 cek_bytes = _get_random_cek_bytes_for_enc(enc)

160

161 # Compute the Encoded Protected Header value BASE64URL(UTF8(JWE

162 # Protected Header)). If the JWE Protected Header is not present

163 # (which can only happen when using the JWE JSON Serialization and

164 # no "protected" member is present), let this value be the empty

165 # string.

166 protected_header = encoded_header

167

168 # Let the Additional Authenticated Data encryption parameter be

169 # ASCII(Encoded Protected Header). However, if a JWE AAD value is

170 # present (which can only be the case when using the JWE JSON

171 # Serialization), instead let the Additional Authenticated Data

172 # encryption parameter be ASCII(Encoded Protected Header || '.' ||

173 # BASE64URL(JWE AAD)).

174 aad = protected_header

175

176 # Decrypt the JWE Ciphertext using the CEK, the JWE Initialization

177 # Vector, the Additional Authenticated Data value, and the JWE

178 # Authentication Tag (which is the Authentication Tag input to the

179 # calculation) using the specified content encryption algorithm,

180 # returning the decrypted plaintext and validating the JWE

181 # Authentication Tag in the manner specified for the algorithm,

182 # rejecting the input without emitting any decrypted output if the

183 # JWE Authentication Tag is incorrect.

184 try:

185 plain_text = _decrypt_and_auth(cek_bytes, enc, cipher_text, iv, aad, auth_tag)

186 except NotImplementedError:

187 raise JWEError(f"enc {enc} is not implemented")

188 except Exception as e:

189 raise JWEError(e)

190

191 # If a "zip" parameter was included, uncompress the decrypted

192 # plaintext using the specified compression algorithm.

193 if plain_text is not None:

194 plain_text = _decompress(header.get("zip"), plain_text)

195

196 return plain_text if cek_valid else None

197

198

199def get_unverified_header(jwe_str):

200 """Returns the decoded headers without verification of any kind.

201

202 Args:

203 jwe_str (str): A compact serialized JWE to decode the headers from.

204

205 Returns:

206 dict: The dict representation of the JWE headers.

207

208 Raises:

209 JWEError: If there is an exception decoding the JWE.

210 """

211 header = _jwe_compact_deserialize(jwe_str)[0]

212 return header

213

214

215def _decrypt_and_auth(cek_bytes, enc, cipher_text, iv, aad, auth_tag):

216 """

217 Decrypt and verify the data

218

219 Args:

220 cek_bytes (bytes): cek to derive encryption and possible auth key to

221 verify the auth tag

222 cipher_text (bytes): Encrypted data

223 iv (bytes): Initialization vector (iv) used to encrypt data

224 aad (bytes): Additional Authenticated Data used to verify the data

225 auth_tag (bytes): Authentication ntag to verify the data

226

227 Returns:

228 (bytes): Decrypted data

229 """

230 # Decrypt the JWE Ciphertext using the CEK, the JWE Initialization

231 # Vector, the Additional Authenticated Data value, and the JWE

232 # Authentication Tag (which is the Authentication Tag input to the

233 # calculation) using the specified content encryption algorithm,

234 # returning the decrypted plaintext

235 # and validating the JWE

236 # Authentication Tag in the manner specified for the algorithm,

237 if enc in ALGORITHMS.HMAC_AUTH_TAG:

238 encryption_key, mac_key, key_len = _get_encryption_key_mac_key_and_key_length_from_cek(cek_bytes, enc)

239 auth_tag_check = _auth_tag(cipher_text, iv, aad, mac_key, key_len)

240 elif enc in ALGORITHMS.GCM:

241 encryption_key = jwk.construct(cek_bytes, enc)

242 auth_tag_check = auth_tag # GCM check auth on decrypt

243 else:

244 raise NotImplementedError(f"enc {enc} is not implemented!")

245

246 plaintext = encryption_key.decrypt(cipher_text, iv, aad, auth_tag)

247 if auth_tag != auth_tag_check:

248 raise JWEError("Invalid JWE Auth Tag")

249

250 return plaintext

251

252

253def _get_encryption_key_mac_key_and_key_length_from_cek(cek_bytes, enc):

254 derived_key_len = len(cek_bytes) // 2

255 mac_key_bytes = cek_bytes[0:derived_key_len]

256 mac_key = _get_hmac_key(enc, mac_key_bytes)

257 encryption_key_bytes = cek_bytes[-derived_key_len:]

258 encryption_alg, _ = enc.split("-")

259 encryption_key = jwk.construct(encryption_key_bytes, encryption_alg)

260 return encryption_key, mac_key, derived_key_len

261

262

263def _jwe_compact_deserialize(jwe_bytes):

264 """

265 Deserialize and verify the header and segments are appropriate.

266

267 Args:

268 jwe_bytes (bytes): The compact serialized JWE

269 Returns:

270 (dict, bytes, bytes, bytes, bytes, bytes)

271 """

272

273 # Base64url decode the encoded representations of the JWE

274 # Protected Header, the JWE Encrypted Key, the JWE Initialization

275 # Vector, the JWE Ciphertext, the JWE Authentication Tag, and the

276 # JWE AAD, following the restriction that no line breaks,

277 # whitespace, or other additional characters have been used.

278 jwe_bytes = ensure_binary(jwe_bytes)

279 try:

280 header_segment, encrypted_key_segment, iv_segment, cipher_text_segment, auth_tag_segment = jwe_bytes.split(

281 b".", 4

282 )

283 header_data = base64url_decode(header_segment)

284 except ValueError:

285 raise JWEParseError("Not enough segments")

286 except (TypeError, binascii.Error):

287 raise JWEParseError("Invalid header")

288

289 # Verify that the octet sequence resulting from decoding the

290 # encoded JWE Protected Header is a UTF-8-encoded representation

291 # of a completely valid JSON object conforming to RFC 7159

292 # [RFC7159]; let the JWE Protected Header be this JSON object.

293 #

294 # If using the JWE Compact Serialization, let the JOSE Header be

295 # the JWE Protected Header. Otherwise, when using the JWE JSON

296 # Serialization, let the JOSE Header be the union of the members

297 # of the JWE Protected Header, the JWE Shared Unprotected Header

298 # and the corresponding JWE Per-Recipient Unprotected Header, all

299 # of which must be completely valid JSON objects. During this

300 # step, verify that the resulting JOSE Header does not contain

301 # duplicate Header Parameter names. When using the JWE JSON

302 # Serialization, this restriction includes that the same Header

303 # Parameter name also MUST NOT occur in distinct JSON object

304 # values that together comprise the JOSE Header.

305

306 try:

307 header = json.loads(header_data)

308 except ValueError as e:

309 raise JWEParseError(f"Invalid header string: {e}")

310

311 if not isinstance(header, Mapping):

312 raise JWEParseError("Invalid header string: must be a json object")

313

314 try:

315 encrypted_key = base64url_decode(encrypted_key_segment)

316 except (TypeError, binascii.Error):

317 raise JWEParseError("Invalid encrypted key")

318

319 try:

320 iv = base64url_decode(iv_segment)

321 except (TypeError, binascii.Error):

322 raise JWEParseError("Invalid IV")

323

324 try:

325 ciphertext = base64url_decode(cipher_text_segment)

326 except (TypeError, binascii.Error):

327 raise JWEParseError("Invalid cyphertext")

328

329 try:

330 auth_tag = base64url_decode(auth_tag_segment)

331 except (TypeError, binascii.Error):

332 raise JWEParseError("Invalid auth tag")

333

334 return header, header_segment, encrypted_key, iv, ciphertext, auth_tag

335

336

337def _encoded_header(alg, enc, zip, cty, kid):

338 """

339 Generate an appropriate JOSE header based on the values provided

340 Args:

341 alg (str): Key wrap/negotiation algorithm

342 enc (str): Encryption algorithm

343 zip (str): Compression method

344 cty (str): Content type of the encrypted data

345 kid (str): ID for the key used for the operation

346

347 Returns:

348 bytes: JSON object of header based on input

349 """

350 header = {"alg": alg, "enc": enc}

351 if zip:

352 header["zip"] = zip

353 if cty:

354 header["cty"] = cty

355 if kid:

356 header["kid"] = kid

357 json_header = json.dumps(

358 header,

359 separators=(",", ":"),

360 sort_keys=True,

361 ).encode("utf-8")

362 return base64url_encode(json_header)

363

364

365def _big_endian(int_val):

366 return pack("!Q", int_val)

367

368

369def _encrypt_and_auth(key, alg, enc, zip, plaintext, aad):

370 """

371 Generate a content encryption key (cek) and initialization

372 vector (iv) based on enc and alg, compress the plaintext based on zip,

373 encrypt the compressed plaintext using the cek and iv based on enc

374

375 Args:

376 key (Key): The key provided for encryption

377 alg (str): The algorithm use for key wrap/negotiation

378 enc (str): The encryption algorithm with which to encrypt the plaintext

379 zip (str): The compression algorithm with which to compress the plaintext

380 plaintext (bytes): The data to encrypt

381 aad (str): Additional authentication data utilized for generating an

382 auth tag

383

384 Returns:

385 (bytes, bytes, bytes, bytes): A tuple of the following data

386 (key wrapped cek, iv, cipher text, auth tag)

387 """

388 try:

389 cek_bytes, kw_cek = _get_cek(enc, alg, key)

390 except NotImplementedError:

391 raise JWEError(f"alg {alg} is not implemented")

392

393 if enc in ALGORITHMS.HMAC_AUTH_TAG:

394 encryption_key, mac_key, key_len = _get_encryption_key_mac_key_and_key_length_from_cek(cek_bytes, enc)

395 iv, ciphertext, tag = encryption_key.encrypt(plaintext, aad)

396 auth_tag = _auth_tag(ciphertext, iv, aad, mac_key, key_len)

397 elif enc in ALGORITHMS.GCM:

398 encryption_key = jwk.construct(cek_bytes, enc)

399 iv, ciphertext, auth_tag = encryption_key.encrypt(plaintext, aad)

400 else:

401 raise NotImplementedError(f"enc {enc} is not implemented!")

402

403 return kw_cek, iv, ciphertext, auth_tag

404

405

406def _get_hmac_key(enc, mac_key_bytes):

407 """

408 Get an HMACKey for the provided encryption algorithm and key bytes

409

410 Args:

411 enc (str): Encryption algorithm

412 mac_key_bytes (bytes): vytes for the HMAC key

413

414 Returns:

415 (HMACKey): The key to perform HMAC actions

416 """

417 _, hash_alg = enc.split("-")

418 mac_key = jwk.construct(mac_key_bytes, hash_alg)

419 return mac_key

420

421

422def _compress(zip, plaintext):

423 """

424 Compress the plaintext based on the algorithm supplied

425

426 Args:

427 zip (str): Compression Algorithm

428 plaintext (bytes): plaintext to compress

429

430 Returns:

431 (bytes): Compressed plaintext

432 """

433 if zip not in ZIPS.SUPPORTED:

434 raise NotImplementedError(f"ZIP {zip} is not supported!")

435 if zip is None:

436 compressed = plaintext

437 elif zip == ZIPS.DEF:

438 compressed = zlib.compress(plaintext)

439 else:

440 raise NotImplementedError(f"ZIP {zip} is not implemented!")

441 return compressed

442

443

444def _decompress(zip, compressed):

445 """

446 Decompress the plaintext based on the algorithm supplied

447

448 Args:

449 zip (str): Compression Algorithm

450 plaintext (bytes): plaintext to decompress

451

452 Returns:

453 (bytes): Compressed plaintext

454 """

455 if zip not in ZIPS.SUPPORTED:

456 raise NotImplementedError(f"ZIP {zip} is not supported!")

457 if zip is None:

458 decompressed = compressed

459 elif zip == ZIPS.DEF:

460 # If, during decompression, there is more data than expected, the

461 # decompression halts and raise an error. This addresses CVE-2024-33664

462 decompressor = zlib.decompressobj()

463 decompressed = decompressor.decompress(compressed, max_length=JWE_SIZE_LIMIT)

464 if decompressor.unconsumed_tail:

465 raise JWEError(f"Decompressed JWE string exceeds {JWE_SIZE_LIMIT} bytes")

466 else:

467 raise NotImplementedError(f"ZIP {zip} is not implemented!")

468 return decompressed

469

470

471def _get_cek(enc, alg, key):

472 """

473 Get the content encryption key

474

475 Args:

476 enc (str): Encryption algorithm

477 alg (str): kwy wrap/negotiation algorithm

478 key (Key): Key provided to encryption method

479

480 Return:

481 (bytes, bytes): Tuple of (cek bytes and wrapped cek)

482 """

483 if alg == ALGORITHMS.DIR:

484 cek, wrapped_cek = _get_direct_key_wrap_cek(key)

485 else:

486 cek, wrapped_cek = _get_key_wrap_cek(enc, key)

487

488 return cek, wrapped_cek

489

490

491def _get_direct_key_wrap_cek(key):

492 """

493 Get the cek and wrapped cek from the encryption key direct

494

495 Args:

496 key (Key): Key provided to encryption method

497

498 Return:

499 (Key, bytes): Tuple of (cek Key object and wrapped cek)

500 """

501 # Get the JWK data to determine how to derive the cek

502 jwk_data = key.to_dict()

503 if jwk_data["kty"] == "oct":

504 # Get the last half of an octal key as the cek

505 cek_bytes = _get_key_bytes_from_key(key)

506 wrapped_cek = b""

507 else:

508 raise NotImplementedError("JWK type {} not supported!".format(jwk_data["kty"]))

509 return cek_bytes, wrapped_cek

510

511

512def _get_key_bytes_from_key(key):

513 """

514 Get the raw key bytes from a Key object

515

516 Args:

517 key (Key): Key from which to extract the raw key bytes

518 Returns:

519 (bytes) key data

520 """

521 jwk_data = key.to_dict()

522 encoded_key = jwk_data["k"]

523 cek_bytes = base64url_decode(encoded_key)

524 return cek_bytes

525

526

527def _get_key_wrap_cek(enc, key):

528 """_get_rsa_key_wrap_cek

529 Get the content encryption key for RSA key wrap

530

531 Args:

532 enc (str): Encryption algorithm

533 key (Key): Key provided to encryption method

534

535 Returns:

536 (Key, bytes): Tuple of (cek Key object and wrapped cek)

537 """

538 cek_bytes = _get_random_cek_bytes_for_enc(enc)

539 wrapped_cek = key.wrap_key(cek_bytes)

540 return cek_bytes, wrapped_cek

541

542

543def _get_random_cek_bytes_for_enc(enc):

544 """

545 Get the random cek bytes based on the encryption algorithm

546

547 Args:

548 enc (str): Encryption algorithm

549

550 Returns:

551 (bytes) random bytes for cek key

552 """

553 if enc == ALGORITHMS.A128GCM:

554 num_bits = 128

555 elif enc == ALGORITHMS.A192GCM:

556 num_bits = 192

557 elif enc in (ALGORITHMS.A128CBC_HS256, ALGORITHMS.A256GCM):

558 num_bits = 256

559 elif enc == ALGORITHMS.A192CBC_HS384:

560 num_bits = 384

561 elif enc == ALGORITHMS.A256CBC_HS512:

562 num_bits = 512

563 else:

564 raise NotImplementedError(f"{enc} not supported")

565 cek_bytes = get_random_bytes(num_bits // 8)

566 return cek_bytes

567

568

569def _auth_tag(ciphertext, iv, aad, mac_key, tag_length):

570 """

571 Get ann auth tag from the provided data

572

573 Args:

574 ciphertext (bytes): Encrypted value

575 iv (bytes): Initialization vector

576 aad (bytes): Additional Authenticated Data

577 mac_key (bytes): Key to use in generating the MAC

578 tag_length (int): How log the tag should be

579

580 Returns:

581 (bytes) Auth tag

582 """

583 al = _big_endian(len(aad) * 8)

584 auth_tag_input = aad + iv + ciphertext + al

585 signature = mac_key.sign(auth_tag_input)

586 auth_tag = signature[0:tag_length]

587 return auth_tag

588

589

590def _jwe_compact_serialize(encoded_header, encrypted_cek, iv, cipher_text, auth_tag):

591 """

592 Generate a compact serialized JWE

593

594 Args:

595 encoded_header (bytes): Base64 URL Encoded JWE header JSON

596 encrypted_cek (bytes): Encrypted content encryption key (cek)

597 iv (bytes): Initialization vector (IV)

598 cipher_text (bytes): Cipher text

599 auth_tag (bytes): JWE Auth Tag

600

601 Returns:

602 (str): JWE compact serialized string

603 """

604 cipher_text = ensure_binary(cipher_text)

605 encoded_encrypted_cek = base64url_encode(encrypted_cek)

606 encoded_iv = base64url_encode(iv)

607 encoded_cipher_text = base64url_encode(cipher_text)

608 encoded_auth_tag = base64url_encode(auth_tag)

609 return (

610 encoded_header

611 + b"."

612 + encoded_encrypted_cek

613 + b"."

614 + encoded_iv

615 + b"."

616 + encoded_cipher_text

617 + b"."

618 + encoded_auth_tag

619 )