Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.11/site-packages/jwcrypto/jwa.py: 55%

3import os

4import struct

5import warnings

6from abc import ABCMeta, abstractmethod

7from binascii import hexlify, unhexlify

9from cryptography.exceptions import InvalidSignature

10from cryptography.hazmat.backends import default_backend

11from cryptography.hazmat.primitives import constant_time, hashes, hmac

12from cryptography.hazmat.primitives.asymmetric import ec

13from cryptography.hazmat.primitives.asymmetric import padding

14from cryptography.hazmat.primitives.asymmetric import utils as ec_utils

15from cryptography.hazmat.primitives.ciphers import Cipher, algorithms, modes

16from cryptography.hazmat.primitives.kdf.concatkdf import ConcatKDFHash

17from cryptography.hazmat.primitives.kdf.pbkdf2 import PBKDF2HMAC

18from cryptography.hazmat.primitives.keywrap import aes_key_unwrap, aes_key_wrap

19from cryptography.hazmat.primitives.padding import PKCS7

21from jwcrypto.common import InvalidCEKeyLength

22from jwcrypto.common import InvalidJWAAlgorithm

23from jwcrypto.common import InvalidJWEData

24from jwcrypto.common import InvalidJWEKeyLength

25from jwcrypto.common import InvalidJWEKeyType

26from jwcrypto.common import InvalidJWEOperation

27from jwcrypto.common import base64url_decode, base64url_encode

28from jwcrypto.common import json_decode

29from jwcrypto.jwk import JWK

31# Implements:

32# - RFC 7518: JSON Web Algorithms (JWA)

33# - RFC 8037: CFRG Elliptic Curve Diffie-Hellman (ECDH) and Signatures

34# in JSON Object Signing and Encryption (JOSE)

35# - RFC 9864: Fully-Specified Algorithms for JSON Object Signing and

36# Encryption (JOSE) and CBOR Object Signing and Encryption

37# (COSE)

39default_max_pbkdf2_iterations = 16384

40"""The maximum number of iterations allowed for PBKDF2 key derivation.

42This is a security measure to prevent denial-of-service attacks by malicious

43actors providing a very high iteration count.

44"""

46default_warn_deprecated_algorithms = False

47"""When True, emit a DeprecationWarning when instantiating an algorithm

48whose status is 'deprecated'.

50Disabled by default so that consumers are not disrupted by warnings for

51algorithms that remain in wide use (e.g. EdDSA). Set to True to audit

52your code for deprecated algorithm usage.

53"""

55default_enforce_hmac_key_length = True

56"""Enforces that the HMAC key length is at least the size of the hash

57function's output, as recommended by RFC 7518.

59This can be disabled for compatibility with legacy or non-compliant systems

60that use shorter keys.

61"""

64class JWAAlgorithm(metaclass=ABCMeta):

66 @property

67 @abstractmethod

68 def name(self):

69 """The algorithm Name"""

71 @property

72 @abstractmethod

73 def description(self):

74 """A short description"""

76 @property

77 @abstractmethod

78 def keysize(self):

79 """The algorithm key size"""

81 @property

82 @abstractmethod

83 def algorithm_usage_location(self):

84 """One of 'alg', 'enc' or 'JWK'"""

86 @property

87 @abstractmethod

88 def algorithm_use(self):

89 """One of 'sig', 'kex', 'enc'"""

91 status = 'active'

92 """Algorithm status: 'active', 'deprecated', or 'prohibited'"""

94 deprecated_by = None

95 """Replacement algorithm(s) for deprecated algorithms"""

97 @property

98 def input_keysize(self):

99 """The input key size"""

100 try:

101 return self.wrap_key_size

102 except AttributeError:

103 return self.keysize

104

105

106def _bitsize(x):

107 return len(x) * 8

108

109

110def _inbytes(x):

111 return x // 8

112

113

114def _randombits(x):

115 if x % 8 != 0:

116 raise ValueError("length must be a multiple of 8")

117 return os.urandom(_inbytes(x))

118

119

120# Note: the number of bits should be a multiple of 16

121def _encode_int(n, bits):

122 e = '{:x}'.format(n)

123 ilen = ((bits + 7) // 8) * 2 # number of bytes rounded up times 2 bytes

124 return unhexlify(e.rjust(ilen, '0')[:ilen])

125

126

127def _decode_int(n):

128 return int(hexlify(n), 16)

129

130

131class _RawJWS:

132

133 def sign(self, key, payload):

134 raise NotImplementedError

135

136 def verify(self, key, payload, signature):

137 raise NotImplementedError

138

139

140class _RawHMAC(_RawJWS):

141

142 keysize = None

143

144 def __init__(self, hashfn):

145 self.backend = default_backend()

146 self.hashfn = hashfn

147

148 def _hmac_setup(self, key, payload):

149 # Ensure key size matches RFC 7518 requirements

150 if default_enforce_hmac_key_length and _bitsize(key) < self.keysize:

151 raise InvalidJWEKeyLength(self.keysize, _bitsize(key))

152 h = hmac.HMAC(key, self.hashfn, backend=self.backend)

153 h.update(payload)

154 return h

155

156 def sign(self, key, payload):

157 skey = base64url_decode(key.get_op_key('sign'))

158 h = self._hmac_setup(skey, payload)

159 return h.finalize()

160

161 def verify(self, key, payload, signature):

162 vkey = base64url_decode(key.get_op_key('verify'))

163 h = self._hmac_setup(vkey, payload)

164 h.verify(signature)

165

166

167class _RawRSA(_RawJWS):

168 def __init__(self, padfn, hashfn):

169 self.padfn = padfn

170 self.hashfn = hashfn

171

172 def sign(self, key, payload):

173 skey = key.get_op_key('sign')

174 return skey.sign(payload, self.padfn, self.hashfn)

175

176 def verify(self, key, payload, signature):

177 pkey = key.get_op_key('verify')

178 pkey.verify(signature, payload, self.padfn, self.hashfn)

179

180

181class _RawEC(_RawJWS):

182 def __init__(self, curve, hashfn):

183 self._curve = curve

184 self.hashfn = hashfn

185

186 @property

187 def curve(self):

188 return self._curve

189

190 def sign(self, key, payload):

191 skey = key.get_op_key('sign', self._curve)

192 size = skey.key_size

193 signature = skey.sign(payload, ec.ECDSA(self.hashfn))

194 r, s = ec_utils.decode_dss_signature(signature)

195 return _encode_int(r, size) + _encode_int(s, size)

196

197 def verify(self, key, payload, signature):

198 pkey = key.get_op_key('verify', self._curve)

199 r = signature[:len(signature) // 2]

200 s = signature[len(signature) // 2:]

201 enc_signature = ec_utils.encode_dss_signature(

202 int(hexlify(r), 16), int(hexlify(s), 16))

203 pkey.verify(enc_signature, payload, ec.ECDSA(self.hashfn))

204

205

206class _RawNone(_RawJWS):

207

208 def sign(self, key, payload):

209 return ''

210

211 def verify(self, key, payload, signature):

212 if key['kty'] != 'oct' or key.get_op_key() != '':

213 raise InvalidSignature('The "none" signature cannot be verified')

214

215

216class _HS256(_RawHMAC, JWAAlgorithm):

217

218 name = "HS256"

219 description = "HMAC using SHA-256"

220 keysize = 256

221 algorithm_usage_location = 'alg'

222 algorithm_use = 'sig'

223

224 def __init__(self):

225 super(_HS256, self).__init__(hashes.SHA256())

226

227

228class _HS384(_RawHMAC, JWAAlgorithm):

229

230 name = "HS384"

231 description = "HMAC using SHA-384"

232 keysize = 384

233 algorithm_usage_location = 'alg'

234 algorithm_use = 'sig'

235

236 def __init__(self):

237 super(_HS384, self).__init__(hashes.SHA384())

238

239

240class _HS512(_RawHMAC, JWAAlgorithm):

241

242 name = "HS512"

243 description = "HMAC using SHA-512"

244 keysize = 512

245 algorithm_usage_location = 'alg'

246 algorithm_use = 'sig'

247

248 def __init__(self):

249 super(_HS512, self).__init__(hashes.SHA512())

250

251

252class _RS256(_RawRSA, JWAAlgorithm):

253

254 name = "RS256"

255 description = "RSASSA-PKCS1-v1_5 using SHA-256"

256 keysize = 2048

257 algorithm_usage_location = 'alg'

258 algorithm_use = 'sig'

259

260 def __init__(self):

261 super(_RS256, self).__init__(padding.PKCS1v15(), hashes.SHA256())

262

263

264class _RS384(_RawRSA, JWAAlgorithm):

265

266 name = "RS384"

267 description = "RSASSA-PKCS1-v1_5 using SHA-384"

268 keysize = 2048

269 algorithm_usage_location = 'alg'

270 algorithm_use = 'sig'

271

272 def __init__(self):

273 super(_RS384, self).__init__(padding.PKCS1v15(), hashes.SHA384())

274

275

276class _RS512(_RawRSA, JWAAlgorithm):

277

278 name = "RS512"

279 description = "RSASSA-PKCS1-v1_5 using SHA-512"

280 keysize = 2048

281 algorithm_usage_location = 'alg'

282 algorithm_use = 'sig'

283

284 def __init__(self):

285 super(_RS512, self).__init__(padding.PKCS1v15(), hashes.SHA512())

286

287

288class _ES256(_RawEC, JWAAlgorithm):

289

290 name = "ES256"

291 description = "ECDSA using P-256 and SHA-256"

292 keysize = 256

293 algorithm_usage_location = 'alg'

294 algorithm_use = 'sig'

295

296 def __init__(self):

297 super(_ES256, self).__init__('P-256', hashes.SHA256())

298

299

300class _ES256K(_RawEC, JWAAlgorithm):

301

302 name = "ES256K"

303 description = "ECDSA using secp256k1 curve and SHA-256"

304 keysize = 256

305 algorithm_usage_location = 'alg'

306 algorithm_use = 'sig'

307

308 def __init__(self):

309 super(_ES256K, self).__init__('secp256k1', hashes.SHA256())

310

311

312class _ES384(_RawEC, JWAAlgorithm):

313

314 name = "ES384"

315 description = "ECDSA using P-384 and SHA-384"

316 keysize = 384

317 algorithm_usage_location = 'alg'

318 algorithm_use = 'sig'

319

320 def __init__(self):

321 super(_ES384, self).__init__('P-384', hashes.SHA384())

322

323

324class _ES512(_RawEC, JWAAlgorithm):

325

326 name = "ES512"

327 description = "ECDSA using P-521 and SHA-512"

328 keysize = 512

329 algorithm_usage_location = 'alg'

330 algorithm_use = 'sig'

331

332 def __init__(self):

333 super(_ES512, self).__init__('P-521', hashes.SHA512())

334

335

336class _PS256(_RawRSA, JWAAlgorithm):

337

338 name = "PS256"

339 description = "RSASSA-PSS using SHA-256 and MGF1 with SHA-256"

340 keysize = 2048

341 algorithm_usage_location = 'alg'

342 algorithm_use = 'sig'

343

344 def __init__(self):

345 padfn = padding.PSS(padding.MGF1(hashes.SHA256()),

346 hashes.SHA256.digest_size)

347 super(_PS256, self).__init__(padfn, hashes.SHA256())

348

349

350class _PS384(_RawRSA, JWAAlgorithm):

351

352 name = "PS384"

353 description = "RSASSA-PSS using SHA-384 and MGF1 with SHA-384"

354 keysize = 2048

355 algorithm_usage_location = 'alg'

356 algorithm_use = 'sig'

357

358 def __init__(self):

359 padfn = padding.PSS(padding.MGF1(hashes.SHA384()),

360 hashes.SHA384.digest_size)

361 super(_PS384, self).__init__(padfn, hashes.SHA384())

362

363

364class _PS512(_RawRSA, JWAAlgorithm):

365

366 name = "PS512"

367 description = "RSASSA-PSS using SHA-512 and MGF1 with SHA-512"

368 keysize = 2048

369 algorithm_usage_location = 'alg'

370 algorithm_use = 'sig'

371

372 def __init__(self):

373 padfn = padding.PSS(padding.MGF1(hashes.SHA512()),

374 hashes.SHA512.digest_size)

375 super(_PS512, self).__init__(padfn, hashes.SHA512())

376

377

378class _None(_RawNone, JWAAlgorithm):

379

380 name = "none"

381 description = "No digital signature or MAC performed"

382 keysize = 0

383 algorithm_usage_location = 'alg'

384 algorithm_use = 'sig'

385

386

387class _RawKeyMgmt:

388

389 def wrap(self, key, bitsize, cek, headers):

390 raise NotImplementedError

391

392 def unwrap(self, key, bitsize, ek, headers):

393 raise NotImplementedError

394

395

396class _RSA(_RawKeyMgmt):

397

398 def __init__(self, padfn):

399 self.padfn = padfn

400

401 def _check_key(self, key):

402 if not isinstance(key, JWK):

403 raise ValueError('key is not a JWK object')

404 if key['kty'] != 'RSA':

405 raise InvalidJWEKeyType('RSA', key['kty'])

406

407 # FIXME: get key size and ensure > 2048 bits

408 def wrap(self, key, bitsize, cek, headers):

409 self._check_key(key)

410 if not cek:

411 cek = _randombits(bitsize)

412 rk = key.get_op_key('wrapKey')

413 ek = rk.encrypt(cek, self.padfn)

414 return {'cek': cek, 'ek': ek}

415

416 def unwrap(self, key, bitsize, ek, headers):

417 self._check_key(key)

418 rk = key.get_op_key('unwrapKey')

419 cek = rk.decrypt(ek, self.padfn)

420 if _bitsize(cek) != bitsize:

421 raise InvalidJWEKeyLength(bitsize, _bitsize(cek))

422 return cek

423

424

425class _Rsa15(_RSA, JWAAlgorithm):

426

427 name = 'RSA1_5'

428 description = "RSAES-PKCS1-v1_5"

429 keysize = 2048

430 algorithm_usage_location = 'alg'

431 algorithm_use = 'kex'

432

433 def __init__(self):

434 super(_Rsa15, self).__init__(padding.PKCS1v15())

435

436 def unwrap(self, key, bitsize, ek, headers):

437 self._check_key(key)

438 # Address MMA attack by implementing RFC 3218 - 2.3.2. Random Filling

439 # provides a random cek that will cause the decryption engine to

440 # run to the end, but will fail decryption later.

441

442 # always generate a random cek so we spend roughly the

443 # same time as in the exception side of the branch

444 cek = _randombits(bitsize)

445 try:

446 cek = super(_Rsa15, self).unwrap(key, bitsize, ek, headers)

447 # always raise so we always run through the exception handling

448 # code in all cases

449 raise ValueError('Dummy')

450 except Exception: # pylint: disable=broad-except

451 return cek

452

453

454class _RsaOaep(_RSA, JWAAlgorithm):

455

456 name = 'RSA-OAEP'

457 description = "RSAES OAEP using default parameters"

458 keysize = 2048

459 algorithm_usage_location = 'alg'

460 algorithm_use = 'kex'

461

462 def __init__(self):

463 super(_RsaOaep, self).__init__(

464 padding.OAEP(padding.MGF1(hashes.SHA1()),

465 hashes.SHA1(), None))

466

467

468class _RsaOaep256(_RSA, JWAAlgorithm): # noqa: ignore=N801

469

470 name = 'RSA-OAEP-256'

471 description = "RSAES OAEP using SHA-256 and MGF1 with SHA-256"

472 keysize = 2048

473 algorithm_usage_location = 'alg'

474 algorithm_use = 'kex'

475

476 def __init__(self):

477 super(_RsaOaep256, self).__init__(

478 padding.OAEP(padding.MGF1(hashes.SHA256()),

479 hashes.SHA256(), None))

480

481

482class _AesKw(_RawKeyMgmt):

483

484 keysize = None

485

486 def __init__(self):

487 self.backend = default_backend()

488

489 def _get_key(self, key, op):

490 if not isinstance(key, JWK):

491 raise ValueError('key is not a JWK object')

492 if key['kty'] != 'oct':

493 raise InvalidJWEKeyType('oct', key['kty'])

494 rk = base64url_decode(key.get_op_key(op))

495 if _bitsize(rk) != self.keysize:

496 raise InvalidJWEKeyLength(self.keysize, _bitsize(rk))

497 return rk

498

499 def wrap(self, key, bitsize, cek, headers):

500 rk = self._get_key(key, 'encrypt')

501 if not cek:

502 cek = _randombits(bitsize)

503

504 ek = aes_key_wrap(rk, cek, default_backend())

505

506 return {'cek': cek, 'ek': ek}

507

508 def unwrap(self, key, bitsize, ek, headers):

509 rk = self._get_key(key, 'decrypt')

510

511 cek = aes_key_unwrap(rk, ek, default_backend())

512 if _bitsize(cek) != bitsize:

513 raise InvalidJWEKeyLength(bitsize, _bitsize(cek))

514 return cek

515

516

517class _A128KW(_AesKw, JWAAlgorithm):

518

519 name = 'A128KW'

520 description = "AES Key Wrap using 128-bit key"

521 keysize = 128

522 algorithm_usage_location = 'alg'

523 algorithm_use = 'kex'

524

525

526class _A192KW(_AesKw, JWAAlgorithm):

527

528 name = 'A192KW'

529 description = "AES Key Wrap using 192-bit key"

530 keysize = 192

531 algorithm_usage_location = 'alg'

532 algorithm_use = 'kex'

533

534

535class _A256KW(_AesKw, JWAAlgorithm):

536

537 name = 'A256KW'

538 description = "AES Key Wrap using 256-bit key"

539 keysize = 256

540 algorithm_usage_location = 'alg'

541 algorithm_use = 'kex'

542

543

544class _AesGcmKw(_RawKeyMgmt):

545

546 keysize = None

547

548 def __init__(self):

549 self.backend = default_backend()

550

551 def _get_key(self, key, op):

552 if not isinstance(key, JWK):

553 raise ValueError('key is not a JWK object')

554 if key['kty'] != 'oct':

555 raise InvalidJWEKeyType('oct', key['kty'])

556 rk = base64url_decode(key.get_op_key(op))

557 if _bitsize(rk) != self.keysize:

558 raise InvalidJWEKeyLength(self.keysize, _bitsize(rk))

559 return rk

560

561 def wrap(self, key, bitsize, cek, headers):

562 rk = self._get_key(key, 'encrypt')

563 if not cek:

564 cek = _randombits(bitsize)

565

566 iv = _randombits(96)

567 cipher = Cipher(algorithms.AES(rk), modes.GCM(iv),

568 backend=self.backend)

569 encryptor = cipher.encryptor()

570 ek = encryptor.update(cek) + encryptor.finalize()

571

572 tag = encryptor.tag

573 return {'cek': cek, 'ek': ek,

574 'header': {'iv': base64url_encode(iv),

575 'tag': base64url_encode(tag)}}

576

577 def unwrap(self, key, bitsize, ek, headers):

578 rk = self._get_key(key, 'decrypt')

579

580 if 'iv' not in headers:

581 raise ValueError('Invalid Header, missing "iv" parameter')

582 iv = base64url_decode(headers['iv'])

583 if 'tag' not in headers:

584 raise ValueError('Invalid Header, missing "tag" parameter')

585 tag = base64url_decode(headers['tag'])

586

587 cipher = Cipher(algorithms.AES(rk), modes.GCM(iv, tag),

588 backend=self.backend)

589 decryptor = cipher.decryptor()

590 cek = decryptor.update(ek) + decryptor.finalize()

591 if _bitsize(cek) != bitsize:

592 raise InvalidJWEKeyLength(bitsize, _bitsize(cek))

593 return cek

594

595

596class _A128GcmKw(_AesGcmKw, JWAAlgorithm):

597

598 name = 'A128GCMKW'

599 description = "Key wrapping with AES GCM using 128-bit key"

600 keysize = 128

601 algorithm_usage_location = 'alg'

602 algorithm_use = 'kex'

603

604

605class _A192GcmKw(_AesGcmKw, JWAAlgorithm):

606

607 name = 'A192GCMKW'

608 description = "Key wrapping with AES GCM using 192-bit key"

609 keysize = 192

610 algorithm_usage_location = 'alg'

611 algorithm_use = 'kex'

612

613

614class _A256GcmKw(_AesGcmKw, JWAAlgorithm):

615

616 name = 'A256GCMKW'

617 description = "Key wrapping with AES GCM using 256-bit key"

618 keysize = 256

619 algorithm_usage_location = 'alg'

620 algorithm_use = 'kex'

621

622

623class _Pbes2HsAesKw(_RawKeyMgmt):

624

625 name = None

626 keysize = None

627 hashsize = None

628

629 def __init__(self):

630 self.backend = default_backend()

631 self.aeskwmap = {128: _A128KW, 192: _A192KW, 256: _A256KW}

632

633 def _get_key(self, alg, key, p2s, p2c):

634 if p2c > default_max_pbkdf2_iterations:

635 raise ValueError('Invalid p2c value, too large')

636

637 if not isinstance(key, JWK):

638 # backwards compatibility for old interface

639 if isinstance(key, bytes):

640 plain = key

641 else:

642 plain = key.encode('utf8')

643 else:

644 plain = base64url_decode(key.get_op_key())

645

646 salt = bytes(self.name.encode('utf8')) + b'\x00' + p2s

647

648 if self.hashsize == 256:

649 hashalg = hashes.SHA256()

650 elif self.hashsize == 384:

651 hashalg = hashes.SHA384()

652 elif self.hashsize == 512:

653 hashalg = hashes.SHA512()

654 else:

655 raise ValueError('Unknown Hash Size')

656

657 kdf = PBKDF2HMAC(algorithm=hashalg, length=_inbytes(self.keysize),

658 salt=salt, iterations=p2c, backend=self.backend)

659 rk = kdf.derive(plain)

660 if _bitsize(rk) != self.keysize:

661 raise InvalidJWEKeyLength(self.keysize, len(rk))

662 return JWK(kty="oct", use="enc", k=base64url_encode(rk))

663

664 def wrap(self, key, bitsize, cek, headers):

665 ret_header = {}

666 if 'p2s' in headers:

667 p2s = base64url_decode(headers['p2s'])

668 if len(p2s) < 8:

669 raise ValueError('Invalid Salt, must be 8 or more octects')

670 else:

671 p2s = _randombits(128)

672 ret_header['p2s'] = base64url_encode(p2s)

673 if 'p2c' in headers:

674 p2c = headers['p2c']

675 else:

676 p2c = 8192

677 ret_header['p2c'] = p2c

678 kek = self._get_key(headers['alg'], key, p2s, p2c)

679

680 aeskw = self.aeskwmap[self.keysize]()

681 ret = aeskw.wrap(kek, bitsize, cek, headers)

682 if len(ret_header) > 0:

683 ret['header'] = ret_header

684 return ret

685

686 def unwrap(self, key, bitsize, ek, headers):

687 if 'p2s' not in headers:

688 raise ValueError('Invalid Header, missing "p2s" parameter')

689 if 'p2c' not in headers:

690 raise ValueError('Invalid Header, missing "p2c" parameter')

691 p2s = base64url_decode(headers['p2s'])

692 p2c = headers['p2c']

693 kek = self._get_key(headers['alg'], key, p2s, p2c)

694

695 aeskw = self.aeskwmap[self.keysize]()

696 return aeskw.unwrap(kek, bitsize, ek, headers)

697

698

699class _Pbes2Hs256A128Kw(_Pbes2HsAesKw, JWAAlgorithm):

700

701 name = 'PBES2-HS256+A128KW'

702 description = 'PBES2 with HMAC SHA-256 and "A128KW" wrapping'

703 keysize = 128

704 algorithm_usage_location = 'alg'

705 algorithm_use = 'kex'

706 hashsize = 256

707

708

709class _Pbes2Hs384A192Kw(_Pbes2HsAesKw, JWAAlgorithm):

710

711 name = 'PBES2-HS384+A192KW'

712 description = 'PBES2 with HMAC SHA-384 and "A192KW" wrapping'

713 keysize = 192

714 algorithm_usage_location = 'alg'

715 algorithm_use = 'kex'

716 hashsize = 384

717

718

719class _Pbes2Hs512A256Kw(_Pbes2HsAesKw, JWAAlgorithm):

720

721 name = 'PBES2-HS512+A256KW'

722 description = 'PBES2 with HMAC SHA-512 and "A256KW" wrapping'

723 keysize = 256

724 algorithm_usage_location = 'alg'

725 algorithm_use = 'kex'

726 hashsize = 512

727

728

729class _Direct(_RawKeyMgmt, JWAAlgorithm):

730

731 name = 'dir'

732 description = "Direct use of a shared symmetric key"

733 keysize = 128

734 algorithm_usage_location = 'alg'

735 algorithm_use = 'kex'

736

737 def _check_key(self, key):

738 if not isinstance(key, JWK):

739 raise ValueError('key is not a JWK object')

740 if key['kty'] != 'oct':

741 raise InvalidJWEKeyType('oct', key['kty'])

742

743 def wrap(self, key, bitsize, cek, headers):

744 self._check_key(key)

745 if cek:

746 return (cek, None)

747 k = base64url_decode(key.get_op_key('encrypt'))

748 if _bitsize(k) != bitsize:

749 raise InvalidCEKeyLength(bitsize, _bitsize(k))

750 return {'cek': k}

751

752 def unwrap(self, key, bitsize, ek, headers):

753 self._check_key(key)

754 if ek != b'':

755 raise ValueError('Invalid Encryption Key.')

756 cek = base64url_decode(key.get_op_key('decrypt'))

757 if _bitsize(cek) != bitsize:

758 raise InvalidJWEKeyLength(bitsize, _bitsize(cek))

759 return cek

760

761

762class _EcdhEs(_RawKeyMgmt, JWAAlgorithm):

763

764 name = 'ECDH-ES'

765 description = "ECDH-ES using Concat KDF"

766 algorithm_usage_location = 'alg'

767 algorithm_use = 'kex'

768 allow_ek = False

769 keysize = None

770

771 def __init__(self):

772 self.backend = default_backend()

773 self.aeskwmap = {128: _A128KW, 192: _A192KW, 256: _A256KW}

774

775 def _check_key(self, key):

776 if not isinstance(key, JWK):

777 raise ValueError('key is not a JWK object')

778 if key['kty'] not in ['EC', 'OKP']:

779 raise InvalidJWEKeyType('EC or OKP', key['kty'])

780 if key['kty'] == 'OKP':

781 if key['crv'] not in ['X25519', 'X448']:

782 raise InvalidJWEKeyType('X25519 or X448',

783 key['crv'])

784

785 def _derive(self, privkey, pubkey, alg, bitsize, headers):

786 # OtherInfo is defined in NIST SP 56A 5.8.1.2.1

787

788 # AlgorithmID

789 otherinfo = struct.pack('>I', len(alg))

790 otherinfo += bytes(alg.encode('utf8'))

791

792 # PartyUInfo

793 apu = base64url_decode(headers['apu']) if 'apu' in headers else b''

794 otherinfo += struct.pack('>I', len(apu))

795 otherinfo += apu

796

797 # PartyVInfo

798 apv = base64url_decode(headers['apv']) if 'apv' in headers else b''

799 otherinfo += struct.pack('>I', len(apv))

800 otherinfo += apv

801

802 # SuppPubInfo

803 otherinfo += struct.pack('>I', bitsize)

804

805 # no SuppPrivInfo

806

807 # Shared Key generation

808 if isinstance(privkey, ec.EllipticCurvePrivateKey):

809 shared_key = privkey.exchange(ec.ECDH(), pubkey)

810 else:

811 # X25519/X448

812 shared_key = privkey.exchange(pubkey)

813

814 ckdf = ConcatKDFHash(algorithm=hashes.SHA256(),

815 length=_inbytes(bitsize),

816 otherinfo=otherinfo,

817 backend=self.backend)

818 return ckdf.derive(shared_key)

819

820 def wrap(self, key, bitsize, cek, headers):

821 self._check_key(key)

822 dk_size = self.keysize

823 if self.keysize is None:

824 if cek is not None:

825 raise InvalidJWEOperation('ECDH-ES cannot use an existing CEK')

826 alg = headers['enc']

827 dk_size = bitsize

828 else:

829 alg = headers['alg']

830

831 epk = JWK.generate(kty=key['kty'], crv=key['crv'])

832 dk = self._derive(epk.get_op_key('unwrapKey'),

833 key.get_op_key('wrapKey'),

834 alg, dk_size, headers)

835

836 if self.keysize is None:

837 ret = {'cek': dk}

838 else:

839 aeskw = self.aeskwmap[self.keysize]()

840 kek = JWK(kty="oct", use="enc", k=base64url_encode(dk))

841 ret = aeskw.wrap(kek, bitsize, cek, headers)

842

843 ret['header'] = {'epk': json_decode(epk.export_public())}

844 return ret

845

846 def unwrap(self, key, bitsize, ek, headers):

847 if 'epk' not in headers:

848 raise ValueError('Invalid Header, missing "epk" parameter')

849 if len(ek) > 0 and not self.allow_ek:

850 raise InvalidJWEData('Non empty "ek" parameter')

851 self._check_key(key)

852 dk_size = self.keysize

853 if self.keysize is None:

854 alg = headers['enc']

855 dk_size = bitsize

856 else:

857 alg = headers['alg']

858

859 epk = JWK(**headers['epk'])

860 dk = self._derive(key.get_op_key('unwrapKey'),

861 epk.get_op_key('wrapKey'),

862 alg, dk_size, headers)

863 if self.keysize is None:

864 return dk

865 else:

866 aeskw = self.aeskwmap[self.keysize]()

867 kek = JWK(kty="oct", use="enc", k=base64url_encode(dk))

868 cek = aeskw.unwrap(kek, bitsize, ek, headers)

869 return cek

870

871

872class _EcdhEsAes128Kw(_EcdhEs):

873

874 name = 'ECDH-ES+A128KW'

875 description = 'ECDH-ES using Concat KDF and "A128KW" wrapping'

876 keysize = 128

877 algorithm_usage_location = 'alg'

878 algorithm_use = 'kex'

879 allow_ek = True

880

881

882class _EcdhEsAes192Kw(_EcdhEs):

883

884 name = 'ECDH-ES+A192KW'

885 description = 'ECDH-ES using Concat KDF and "A192KW" wrapping'

886 keysize = 192

887 algorithm_usage_location = 'alg'

888 algorithm_use = 'kex'

889 allow_ek = True

890

891

892class _EcdhEsAes256Kw(_EcdhEs):

893

894 name = 'ECDH-ES+A256KW'

895 description = 'ECDH-ES using Concat KDF and "A256KW" wrapping'

896 keysize = 256

897 algorithm_usage_location = 'alg'

898 algorithm_use = 'kex'

899 allow_ek = True

900

901

902class _EdDsa(_RawJWS, JWAAlgorithm):

903

904 name = 'EdDSA'

905 description = 'EdDSA using Ed25519 or Ed448 algorithms'

906 algorithm_usage_location = 'alg'

907 algorithm_use = 'sig'

908 keysize = None

909 status = 'deprecated'

910 deprecated_by = 'Ed25519 or Ed448'

911

912 def sign(self, key, payload):

913 if key['crv'] in ['Ed25519', 'Ed448']:

914 skey = key.get_op_key('sign')

915 return skey.sign(payload)

916 raise NotImplementedError

917

918 def verify(self, key, payload, signature):

919 if key['crv'] in ['Ed25519', 'Ed448']:

920 pkey = key.get_op_key('verify')

921 return pkey.verify(signature, payload)

922 raise NotImplementedError

923

924

925class _Ed25519(_RawJWS, JWAAlgorithm):

926

927 name = 'Ed25519'

928 description = 'EdDSA using Ed25519'

929 algorithm_usage_location = 'alg'

930 algorithm_use = 'sig'

931 keysize = None

932

933 def sign(self, key, payload):

934 if key['crv'] != 'Ed25519':

935 raise InvalidJWEKeyType('Ed25519', key['crv'])

936 skey = key.get_op_key('sign')

937 return skey.sign(payload)

938

939 def verify(self, key, payload, signature):

940 if key['crv'] != 'Ed25519':

941 raise InvalidJWEKeyType('Ed25519', key['crv'])

942 pkey = key.get_op_key('verify')

943 return pkey.verify(signature, payload)

944

945

946class _Ed448(_RawJWS, JWAAlgorithm):

947

948 name = 'Ed448'

949 description = 'EdDSA using Ed448'

950 algorithm_usage_location = 'alg'

951 algorithm_use = 'sig'

952 keysize = None

953

954 def sign(self, key, payload):

955 if key['crv'] != 'Ed448':

956 raise InvalidJWEKeyType('Ed448', key['crv'])

957 skey = key.get_op_key('sign')

958 return skey.sign(payload)

959

960 def verify(self, key, payload, signature):

961 if key['crv'] != 'Ed448':

962 raise InvalidJWEKeyType('Ed448', key['crv'])

963 pkey = key.get_op_key('verify')

964 return pkey.verify(signature, payload)

965

966

967class _RawMLDSA(_RawJWS):

968

969 _alg_name = None

970

971 def _check_key(self, key):

972 if key['kty'] != 'AKP':

973 raise InvalidJWEKeyType('AKP', key['kty'])

974 if key.get('alg') != self._alg_name:

975 raise InvalidJWEKeyType(self._alg_name, key.get('alg'))

976

977 def sign(self, key, payload):

978 self._check_key(key)

979 skey = key.get_op_key('sign')

980 return skey.sign(payload, context=b"")

981

982 def verify(self, key, payload, signature):

983 self._check_key(key)

984 pkey = key.get_op_key('verify')

985 pkey.verify(signature, payload, context=b"")

986

987

988class _MLDSA44(_RawMLDSA, JWAAlgorithm):

989

990 name = 'ML-DSA-44'

991 description = 'ML-DSA using ML-DSA-44 parameter set'

992 algorithm_usage_location = 'alg'

993 algorithm_use = 'sig'

994 keysize = None

995 _alg_name = 'ML-DSA-44'

996

997

998class _MLDSA65(_RawMLDSA, JWAAlgorithm):

999

1000 name = 'ML-DSA-65'

1001 description = 'ML-DSA using ML-DSA-65 parameter set'

1002 algorithm_usage_location = 'alg'

1003 algorithm_use = 'sig'

1004 keysize = None

1005 _alg_name = 'ML-DSA-65'

1006

1007

1008class _MLDSA87(_RawMLDSA, JWAAlgorithm):

1009

1010 name = 'ML-DSA-87'

1011 description = 'ML-DSA using ML-DSA-87 parameter set'

1012 algorithm_usage_location = 'alg'

1013 algorithm_use = 'sig'

1014 keysize = None

1015 _alg_name = 'ML-DSA-87'

1016

1017

1018class _RawJWE:

1019

1020 def encrypt(self, k, aad, m):

1021 raise NotImplementedError

1022

1023 def decrypt(self, k, aad, iv, e, t):

1024 raise NotImplementedError

1025

1026

1027class _AesCbcHmacSha2(_RawJWE):

1028

1029 keysize = None

1030

1031 def __init__(self, hashfn):

1032 self.backend = default_backend()

1033 self.hashfn = hashfn

1034 self.blocksize = algorithms.AES.block_size

1035 self.wrap_key_size = self.keysize * 2

1036

1037 def _mac(self, k, aad, iv, e):

1038 al = _encode_int(_bitsize(aad), 64)

1039 h = hmac.HMAC(k, self.hashfn, backend=self.backend)

1040 h.update(aad)

1041 h.update(iv)

1042 h.update(e)

1043 h.update(al)

1044 m = h.finalize()

1045 return m[:_inbytes(self.keysize)]

1046

1047 # RFC 7518 - 5.2.2

1048 def encrypt(self, k, aad, m):

1049 """ Encrypt according to the selected encryption and hashing

1050 functions.

1051

1052 :param k: Encryption key

1053 :param aad: Additional Authentication Data

1054 :param m: Plaintext

1055

1056 Returns a dictionary with the computed data.

1057 """

1058 if len(k) != _inbytes(self.wrap_key_size):

1059 raise ValueError("Invalid input key size")

1060

1061 hkey = k[:_inbytes(self.keysize)]

1062 ekey = k[_inbytes(self.keysize):]

1063

1064 # encrypt

1065 iv = _randombits(self.blocksize)

1066 cipher = Cipher(algorithms.AES(ekey), modes.CBC(iv),

1067 backend=self.backend)

1068 encryptor = cipher.encryptor()

1069 padder = PKCS7(self.blocksize).padder()

1070 padded_data = padder.update(m) + padder.finalize()

1071 e = encryptor.update(padded_data) + encryptor.finalize()

1072

1073 # mac

1074 t = self._mac(hkey, aad, iv, e)

1075

1076 return (iv, e, t)

1077

1078 def decrypt(self, k, aad, iv, e, t):

1079 """ Decrypt according to the selected encryption and hashing

1080 functions.

1081 :param k: Encryption key

1082 :param aad: Additional Authenticated Data

1083 :param iv: Initialization Vector

1084 :param e: Ciphertext

1085 :param t: Authentication Tag

1086

1087 Returns plaintext or raises an error

1088 """

1089 if len(k) != _inbytes(self.wrap_key_size):

1090 raise ValueError("Invalid input key size")

1091

1092 hkey = k[:_inbytes(self.keysize)]

1093 dkey = k[_inbytes(self.keysize):]

1094

1095 # verify mac

1096 if not constant_time.bytes_eq(t, self._mac(hkey, aad, iv, e)):

1097 raise InvalidSignature('Failed to verify MAC')

1098

1099 # decrypt

1100 cipher = Cipher(algorithms.AES(dkey), modes.CBC(iv),

1101 backend=self.backend)

1102 decryptor = cipher.decryptor()

1103 d = decryptor.update(e) + decryptor.finalize()

1104 unpadder = PKCS7(self.blocksize).unpadder()

1105 return unpadder.update(d) + unpadder.finalize()

1106

1107

1108class _A128CbcHs256(_AesCbcHmacSha2, JWAAlgorithm):

1109

1110 name = 'A128CBC-HS256'

1111 description = "AES_128_CBC_HMAC_SHA_256 authenticated"

1112 keysize = 128

1113 algorithm_usage_location = 'enc'

1114 algorithm_use = 'enc'

1115

1116 def __init__(self):

1117 super(_A128CbcHs256, self).__init__(hashes.SHA256())

1118

1119

1120class _A192CbcHs384(_AesCbcHmacSha2, JWAAlgorithm):

1121

1122 name = 'A192CBC-HS384'

1123 description = "AES_192_CBC_HMAC_SHA_384 authenticated"

1124 keysize = 192

1125 algorithm_usage_location = 'enc'

1126 algorithm_use = 'enc'

1127

1128 def __init__(self):

1129 super(_A192CbcHs384, self).__init__(hashes.SHA384())

1130

1131

1132class _A256CbcHs512(_AesCbcHmacSha2, JWAAlgorithm):

1133

1134 name = 'A256CBC-HS512'

1135 description = "AES_256_CBC_HMAC_SHA_512 authenticated"

1136 keysize = 256

1137 algorithm_usage_location = 'enc'

1138 algorithm_use = 'enc'

1139

1140 def __init__(self):

1141 super(_A256CbcHs512, self).__init__(hashes.SHA512())

1142

1143

1144class _AesGcm(_RawJWE):

1145

1146 keysize = None

1147

1148 def __init__(self):

1149 self.backend = default_backend()

1150 self.wrap_key_size = self.keysize

1151

1152 # RFC 7518 - 5.3

1153 def encrypt(self, k, aad, m):

1154 """ Encrypt according to the selected encryption and hashing

1155 functions.

1156

1157 :param k: Encryption key

1158 :param aad: Additional Authentication Data

1159 :param m: Plaintext

1160

1161 Returns a dictionary with the computed data.

1162 """

1163 iv = _randombits(96)

1164 cipher = Cipher(algorithms.AES(k), modes.GCM(iv),

1165 backend=self.backend)

1166 encryptor = cipher.encryptor()

1167 encryptor.authenticate_additional_data(aad)

1168 e = encryptor.update(m) + encryptor.finalize()

1169

1170 return (iv, e, encryptor.tag)

1171

1172 def decrypt(self, k, aad, iv, e, t):

1173 """ Decrypt according to the selected encryption and hashing

1174 functions.

1175 :param k: Encryption key

1176 :param aad: Additional Authenticated Data

1177 :param iv: Initialization Vector

1178 :param e: Ciphertext

1179 :param t: Authentication Tag

1180

1181 Returns plaintext or raises an error

1182 """

1183 cipher = Cipher(algorithms.AES(k), modes.GCM(iv, t),

1184 backend=self.backend)

1185 decryptor = cipher.decryptor()

1186 decryptor.authenticate_additional_data(aad)

1187 return decryptor.update(e) + decryptor.finalize()

1188

1189

1190class _A128Gcm(_AesGcm, JWAAlgorithm):

1191

1192 name = 'A128GCM'

1193 description = "AES GCM using 128-bit key"

1194 keysize = 128

1195 algorithm_usage_location = 'enc'

1196 algorithm_use = 'enc'

1197

1198

1199class _A192Gcm(_AesGcm, JWAAlgorithm):

1200

1201 name = 'A192GCM'

1202 description = "AES GCM using 192-bit key"

1203 keysize = 192

1204 algorithm_usage_location = 'enc'

1205 algorithm_use = 'enc'

1206

1207

1208class _A256Gcm(_AesGcm, JWAAlgorithm):

1209

1210 name = 'A256GCM'

1211 description = "AES GCM using 256-bit key"

1212 keysize = 256

1213 algorithm_usage_location = 'enc'

1214 algorithm_use = 'enc'

1215

1216

1217class _BP256R1(_RawEC, JWAAlgorithm):

1218

1219 name = "BP256R1"

1220 description = (

1221 "ECDSA using Brainpool256R1 curve and SHA-256"

1222 " (unregistered, custom-defined in breach"

1223 " of IETF rules by gematik GmbH)"

1224 )

1225 keysize = 256

1226 algorithm_usage_location = 'alg'

1227 algorithm_use = 'sig'

1228

1229 def __init__(self):

1230 super(_BP256R1, self).__init__('BP-256', hashes.SHA256())

1231

1232

1233class _BP384R1(_RawEC, JWAAlgorithm):

1234

1235 name = "BP384R1"

1236 description = (

1237 "ECDSA using Brainpool384R1 curve and SHA-384"

1238 " (unregistered, custom-defined in breach"

1239 " of IETF rules by gematik GmbH)"

1240 )

1241 keysize = 384

1242 algorithm_usage_location = 'alg'

1243 algorithm_use = 'sig'

1244

1245 def __init__(self):

1246 super(_BP384R1, self).__init__('BP-384', hashes.SHA384())

1247

1248

1249class _BP512R1(_RawEC, JWAAlgorithm):

1250

1251 name = "BP512R1"

1252 description = (

1253 "ECDSA using Brainpool512R1 curve and SHA-512"

1254 " (unregistered, custom-defined in breach"

1255 " of IETF rules by gematik GmbH)"

1256 )

1257 keysize = 512

1258 algorithm_usage_location = 'alg'

1259 algorithm_use = 'sig'

1260

1261 def __init__(self):

1262 super(_BP512R1, self).__init__('BP-512', hashes.SHA512())

1263

1264

1265class JWA:

1266 """JWA Signing Algorithms.

1267

1268 This class provides access to all JWA algorithms.

1269 """

1270

1271 algorithms_registry = {

1272 'HS256': _HS256,

1273 'HS384': _HS384,

1274 'HS512': _HS512,

1275 'RS256': _RS256,

1276 'RS384': _RS384,

1277 'RS512': _RS512,

1278 'ES256': _ES256,

1279 'ES256K': _ES256K,

1280 'ES384': _ES384,

1281 'ES512': _ES512,

1282 'PS256': _PS256,

1283 'PS384': _PS384,

1284 'PS512': _PS512,

1285 'none': _None,

1286 'RSA1_5': _Rsa15,

1287 'RSA-OAEP': _RsaOaep,

1288 'RSA-OAEP-256': _RsaOaep256,

1289 'A128KW': _A128KW,

1290 'A192KW': _A192KW,

1291 'A256KW': _A256KW,

1292 'dir': _Direct,

1293 'ECDH-ES': _EcdhEs,

1294 'ECDH-ES+A128KW': _EcdhEsAes128Kw,

1295 'ECDH-ES+A192KW': _EcdhEsAes192Kw,

1296 'ECDH-ES+A256KW': _EcdhEsAes256Kw,

1297 'EdDSA': _EdDsa,

1298 'A128GCMKW': _A128GcmKw,

1299 'A192GCMKW': _A192GcmKw,

1300 'A256GCMKW': _A256GcmKw,

1301 'PBES2-HS256+A128KW': _Pbes2Hs256A128Kw,

1302 'PBES2-HS384+A192KW': _Pbes2Hs384A192Kw,

1303 'PBES2-HS512+A256KW': _Pbes2Hs512A256Kw,

1304 'A128CBC-HS256': _A128CbcHs256,

1305 'A192CBC-HS384': _A192CbcHs384,

1306 'A256CBC-HS512': _A256CbcHs512,

1307 'A128GCM': _A128Gcm,

1308 'A192GCM': _A192Gcm,

1309 'A256GCM': _A256Gcm,

1310 'BP256R1': _BP256R1,

1311 'BP384R1': _BP384R1,

1312 'BP512R1': _BP512R1,

1313 'Ed25519': _Ed25519,

1314 'Ed448': _Ed448,

1315 'ML-DSA-44': _MLDSA44,

1316 'ML-DSA-65': _MLDSA65,

1317 'ML-DSA-87': _MLDSA87,

1318 }

1319

1320 @classmethod

1321 def instantiate_alg(cls, name, use=None):

1322 alg = cls.algorithms_registry[name]

1323 if use is not None and alg.algorithm_use != use:

1324 raise KeyError

1325 if alg.status == 'prohibited':

1326 raise InvalidJWAAlgorithm(

1327 '%s is prohibited and must not be used' % name)

1328 if alg.status == 'deprecated' and default_warn_deprecated_algorithms:

1329 msg = '%s is deprecated' % name

1330 if alg.deprecated_by:

1331 msg += '; use %s instead' % alg.deprecated_by

1332 warnings.warn(msg, DeprecationWarning, stacklevel=4)

1333 return alg()

1334

1335 @classmethod

1336 def signing_alg(cls, name):

1337 try:

1338 return cls.instantiate_alg(name, use='sig')

1339 except KeyError:

1340 raise InvalidJWAAlgorithm('%s is not a valid Signing algorithm'

1341 ' name' % name) from None

1342

1343 @classmethod

1344 def keymgmt_alg(cls, name):

1345 try:

1346 return cls.instantiate_alg(name, use='kex')

1347 except KeyError:

1348 raise InvalidJWAAlgorithm('%s is not a valid Key Management'

1349 ' algorithm name' % name) from None

1350

1351 @classmethod

1352 def encryption_alg(cls, name):

1353 try:

1354 return cls.instantiate_alg(name, use='enc')

1355 except KeyError:

1356 raise InvalidJWAAlgorithm('%s is not a valid Encryption'

1357 ' algorithm name' % name) from None