Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.11/site-packages/jwcrypto/jwk.py: 26%
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
1# Copyright (C) 2015 JWCrypto Project Contributors - see LICENSE file
3import os
4from binascii import hexlify, unhexlify
5from collections import namedtuple
6from enum import Enum
8from cryptography import x509
9from cryptography.hazmat.backends import default_backend
10from cryptography.hazmat.primitives import hashes, serialization
11from cryptography.hazmat.primitives.asymmetric import ec
12from cryptography.hazmat.primitives.asymmetric import rsa
14from typing_extensions import deprecated
16from jwcrypto.common import JWException
17from jwcrypto.common import base64url_decode, base64url_encode
18from jwcrypto.common import json_decode, json_encode
21class UnimplementedOKPCurveKey:
22 @classmethod
23 def generate(cls):
24 raise NotImplementedError
26 @classmethod
27 def from_public_bytes(cls, *args):
28 raise NotImplementedError
30 @classmethod
31 def from_private_bytes(cls, *args):
32 raise NotImplementedError
35ImplementedOkpCurves = []
38# Handle the best we can older versions of python cryptography that
39# do not yet implement these interfaces properly
40try:
41 from cryptography.hazmat.primitives.asymmetric.ed25519 import (
42 Ed25519PublicKey, Ed25519PrivateKey
43 )
44 ImplementedOkpCurves.append('Ed25519')
45except ImportError:
46 Ed25519PublicKey = UnimplementedOKPCurveKey
47 Ed25519PrivateKey = UnimplementedOKPCurveKey
48try:
49 from cryptography.hazmat.primitives.asymmetric.ed448 import (
50 Ed448PublicKey, Ed448PrivateKey
51 )
52 ImplementedOkpCurves.append('Ed448')
53except ImportError:
54 Ed448PublicKey = UnimplementedOKPCurveKey
55 Ed448PrivateKey = UnimplementedOKPCurveKey
56try:
57 from cryptography.hazmat.primitives.asymmetric.x25519 import (
58 X25519PublicKey, X25519PrivateKey
59 )
60 priv_bytes = getattr(X25519PrivateKey, 'from_private_bytes', None)
61 if priv_bytes is None:
62 raise ImportError
63 ImplementedOkpCurves.append('X25519')
64except ImportError:
65 X25519PublicKey = UnimplementedOKPCurveKey
66 X25519PrivateKey = UnimplementedOKPCurveKey
67try:
68 from cryptography.hazmat.primitives.asymmetric.x448 import (
69 X448PublicKey, X448PrivateKey
70 )
71 ImplementedOkpCurves.append('X448')
72except ImportError:
73 X448PublicKey = UnimplementedOKPCurveKey
74 X448PrivateKey = UnimplementedOKPCurveKey
77_Ed25519_CURVE = namedtuple('Ed25519', 'pubkey privkey')
78_Ed448_CURVE = namedtuple('Ed448', 'pubkey privkey')
79_X25519_CURVE = namedtuple('X25519', 'pubkey privkey')
80_X448_CURVE = namedtuple('X448', 'pubkey privkey')
81_OKP_CURVES_TABLE = {
82 'Ed25519': _Ed25519_CURVE(Ed25519PublicKey, Ed25519PrivateKey),
83 'Ed448': _Ed448_CURVE(Ed448PublicKey, Ed448PrivateKey),
84 'X25519': _X25519_CURVE(X25519PublicKey, X25519PrivateKey),
85 'X448': _X448_CURVE(X448PublicKey, X448PrivateKey)
86}
89class UnimplementedAKPAlgorithm:
90 @classmethod
91 def generate(cls):
92 raise NotImplementedError
94 @classmethod
95 def from_public_bytes(cls, *args):
96 raise NotImplementedError
98 @classmethod
99 def from_seed_bytes(cls, *args):
100 raise NotImplementedError
103ImplementedAKPAlgorithms = []
105try:
106 from cryptography.hazmat.primitives.asymmetric.mldsa import (
107 MLDSA44PublicKey, MLDSA44PrivateKey
108 )
109 ImplementedAKPAlgorithms.append('ML-DSA-44')
110except ImportError:
111 MLDSA44PublicKey = UnimplementedAKPAlgorithm
112 MLDSA44PrivateKey = UnimplementedAKPAlgorithm
113try:
114 from cryptography.hazmat.primitives.asymmetric.mldsa import (
115 MLDSA65PublicKey, MLDSA65PrivateKey
116 )
117 ImplementedAKPAlgorithms.append('ML-DSA-65')
118except ImportError:
119 MLDSA65PublicKey = UnimplementedAKPAlgorithm
120 MLDSA65PrivateKey = UnimplementedAKPAlgorithm
121try:
122 from cryptography.hazmat.primitives.asymmetric.mldsa import (
123 MLDSA87PublicKey, MLDSA87PrivateKey
124 )
125 ImplementedAKPAlgorithms.append('ML-DSA-87')
126except ImportError:
127 MLDSA87PublicKey = UnimplementedAKPAlgorithm
128 MLDSA87PrivateKey = UnimplementedAKPAlgorithm
130_AKPAlg = namedtuple('AKPAlg', 'pubkey privkey')
131_AKP_ALG_TABLE = {
132 'ML-DSA-44': _AKPAlg(MLDSA44PublicKey, MLDSA44PrivateKey),
133 'ML-DSA-65': _AKPAlg(MLDSA65PublicKey, MLDSA65PrivateKey),
134 'ML-DSA-87': _AKPAlg(MLDSA87PublicKey, MLDSA87PrivateKey),
135}
138# RFC 7518 - 7.4 , RFC 8037 - 5
139JWKTypesRegistry = {'EC': 'Elliptic Curve',
140 'RSA': 'RSA',
141 'oct': 'Octet sequence',
142 'OKP': 'Octet Key Pair',
143 'AKP': 'Algorithm Key Pair'}
144"""Registry of valid Key Types"""
147# RFC 7518 - 7.5
148# It is part of the JWK Parameters Registry, but we want a more
149# specific map for internal usage
150class ParmType(Enum):
151 name = 'A string with a name'
152 b64 = 'Base64url Encoded'
153 b64u = 'Base64urlUint Encoded'
154 unsupported = 'Unsupported Parameter'
157JWKParameter = namedtuple('Parameter', 'description public required type')
158JWKValuesRegistry = {
159 'EC': {
160 'crv': JWKParameter('Curve', True, True, ParmType.name),
161 'x': JWKParameter('X Coordinate', True, True, ParmType.b64),
162 'y': JWKParameter('Y Coordinate', True, True, ParmType.b64),
163 'd': JWKParameter('ECC Private Key', False, False, ParmType.b64),
164 },
165 'RSA': {
166 'n': JWKParameter('Modulus', True, True, ParmType.b64),
167 'e': JWKParameter('Exponent', True, True, ParmType.b64u),
168 'd': JWKParameter('Private Exponent', False, False, ParmType.b64u),
169 'p': JWKParameter('First Prime Factor', False, False, ParmType.b64u),
170 'q': JWKParameter('Second Prime Factor', False, False, ParmType.b64u),
171 'dp': JWKParameter('First Factor CRT Exponent',
172 False, False, ParmType.b64u),
173 'dq': JWKParameter('Second Factor CRT Exponent',
174 False, False, ParmType.b64u),
175 'qi': JWKParameter('First CRT Coefficient',
176 False, False, ParmType.b64u),
177 'oth': JWKParameter('Other Primes Info',
178 False, False, ParmType.unsupported),
179 },
180 'oct': {
181 'k': JWKParameter('Key Value', False, True, ParmType.b64),
182 },
183 'OKP': {
184 'crv': JWKParameter('Curve', True, True, ParmType.name),
185 'x': JWKParameter('Public Key', True, True, ParmType.b64),
186 'd': JWKParameter('Private Key', False, False, ParmType.b64),
187 },
188 'AKP': {
189 'pub': JWKParameter('Public Key', True, True, ParmType.b64),
190 'priv': JWKParameter('Private Key Seed', False, False, ParmType.b64),
191 }
192}
193"""Registry of valid key values"""
195JWKParamsRegistry = {
196 'kty': JWKParameter('Key Type', True, None, None),
197 'use': JWKParameter('Public Key Use', True, None, None),
198 'key_ops': JWKParameter('Key Operations', True, None, None),
199 'alg': JWKParameter('Algorithm', True, None, None),
200 'kid': JWKParameter('Key ID', True, None, None),
201 'x5u': JWKParameter('X.509 URL', True, None, None),
202 'x5c': JWKParameter('X.509 Certificate Chain', True, None, None),
203 'x5t': JWKParameter('X.509 Certificate SHA-1 Thumbprint',
204 True, None, None),
205 'x5t#S256': JWKParameter('X.509 Certificate SHA-256 Thumbprint',
206 True, None, None)
207}
208"""Registry of valid key parameters"""
210# RFC 7518 - 7.6 , RFC 8037 - 5
211# RFC 8812 - 4.4
212JWKEllipticCurveRegistry = {'P-256': 'P-256 curve',
213 'P-384': 'P-384 curve',
214 'P-521': 'P-521 curve',
215 'secp256k1': 'SECG secp256k1 curve',
216 'Ed25519': 'Ed25519 signature algorithm key pairs',
217 'Ed448': 'Ed448 signature algorithm key pairs',
218 'X25519': 'X25519 function key pairs',
219 'X448': 'X448 function key pairs',
220 'BP-256': 'BrainpoolP256R1 curve'
221 ' (unregistered, custom-defined in breach'
222 ' of IETF rules by gematik GmbH)',
223 'BP-384': 'BrainpoolP384R1 curve'
224 ' (unregistered, custom-defined in breach'
225 ' of IETF rules by gematik GmbH)',
226 'BP-512': 'BrainpoolP512R1 curve'
227 ' (unregistered, custom-defined in breach'
228 ' of IETF rules by gematik GmbH)'
229 }
230"""Registry of allowed Elliptic Curves"""
232# RFC 7517 - 8.2
233JWKUseRegistry = {'sig': 'Digital Signature or MAC',
234 'enc': 'Encryption'}
235"""Registry of allowed uses"""
237# RFC 7517 - 8.3
238JWKOperationsRegistry = {'sign': 'Compute digital Signature or MAC',
239 'verify': 'Verify digital signature or MAC',
240 'encrypt': 'Encrypt content',
241 'decrypt': 'Decrypt content and validate'
242 ' decryption, if applicable',
243 'wrapKey': 'Encrypt key',
244 'unwrapKey': 'Decrypt key and validate'
245 ' decryption, if applicable',
246 'deriveKey': 'Derive key',
247 'deriveBits': 'Derive bits not to be used as a key'}
248"""Registry of allowed operations"""
250# Key Operations registry but augmented with use pair matching.
251JWKOpAndUsePairs = {'sign': 'sig',
252 'verify': 'sig',
253 'encrypt': 'enc',
254 'decrypt': 'enc',
255 'wrapKey': 'enc',
256 'unwrapKey': 'enc',
257 'deriveKey': 'enc',
258 'deriveBits': 'enc'}
260JWKpycaCurveMap = {'secp256r1': 'P-256',
261 'secp384r1': 'P-384',
262 'secp521r1': 'P-521',
263 'secp256k1': 'secp256k1',
264 'brainpoolP256r1': 'BP-256',
265 'brainpoolP384r1': 'BP-384',
266 'brainpoolP512r1': 'BP-512'}
268IANANamedInformationHashAlgorithmRegistry = {
269 'sha-256': hashes.SHA256(),
270 'sha-256-128': None,
271 'sha-256-120': None,
272 'sha-256-96': None,
273 'sha-256-64': None,
274 'sha-256-32': None,
275 'sha-384': hashes.SHA384(),
276 'sha-512': hashes.SHA512(),
277 'sha3-224': hashes.SHA3_224(),
278 'sha3-256': hashes.SHA3_256(),
279 'sha3-384': hashes.SHA3_384(),
280 'sha3-512': hashes.SHA3_512(),
281 'blake2s-256': hashes.BLAKE2s(32),
282 'blake2b-256': None, # pyca supports only 64 bytes for BLAKEb
283 'blake2b-512': hashes.BLAKE2b(64),
284}
287class InvalidJWKType(JWException):
288 """Invalid JWK Type Exception.
290 This exception is raised when an invalid parameter type is used.
291 """
293 def __init__(self, value=None):
294 super(InvalidJWKType, self).__init__()
295 self.value = value
297 def __str__(self):
298 return 'Unknown type "%s", valid types are: %s' % (
299 self.value, list(JWKTypesRegistry.keys()))
302class InvalidJWKUsage(JWException):
303 """Invalid JWK usage Exception.
305 This exception is raised when an invalid key usage is requested,
306 based on the key type and declared usage constraints.
307 """
309 def __init__(self, use, value):
310 super(InvalidJWKUsage, self).__init__()
311 self.value = value
312 self.use = use
314 def __str__(self):
315 if self.use in list(JWKUseRegistry.keys()):
316 usage = JWKUseRegistry[self.use]
317 else:
318 usage = 'Unknown(%s)' % self.use
319 if self.value in list(JWKUseRegistry.keys()):
320 valid = JWKUseRegistry[self.value]
321 else:
322 valid = 'Unknown(%s)' % self.value
323 return 'Invalid usage requested: "%s". Valid for: "%s"' % (usage,
324 valid)
327class InvalidJWKOperation(JWException):
328 """Invalid JWK Operation Exception.
330 This exception is raised when an invalid key operation is requested,
331 based on the key type and declared usage constraints.
332 """
334 def __init__(self, operation, values):
335 super(InvalidJWKOperation, self).__init__()
336 self.op = operation
337 self.values = values
339 def __str__(self):
340 if self.op in list(JWKOperationsRegistry.keys()):
341 op = JWKOperationsRegistry[self.op]
342 else:
343 op = 'Unknown(%s)' % self.op
344 valid = []
345 for v in self.values:
346 if v in list(JWKOperationsRegistry.keys()):
347 valid.append(JWKOperationsRegistry[v])
348 else:
349 valid.append('Unknown(%s)' % v)
350 return 'Invalid operation requested: "%s". Valid for: "%s"' % (op,
351 valid)
354class InvalidJWKValue(JWException):
355 """Invalid JWK Value Exception.
357 This exception is raised when an invalid/unknown value is used in the
358 context of an operation that requires specific values to be used based
359 on the key type or other constraints.
360 """
363class JWK(dict):
364 """JSON Web Key object
366 This object represents a Key.
367 It must be instantiated by using the standard defined key/value pairs
368 as arguments of the initialization function.
369 """
371 def __init__(self, **kwargs):
372 r"""Creates a new JWK object.
374 The function arguments must be valid parameters as defined in the
375 'IANA JSON Web Key Set Parameters registry' and specified in
376 the :data:`JWKParamsRegistry` variable. The 'kty' parameter must
377 always be provided and its value must be a valid one as defined
378 by the 'IANA JSON Web Key Types registry' and specified in the
379 :data:`JWKTypesRegistry` variable. The valid key parameters per
380 key type are defined in the :data:`JWKValuesRegistry` variable.
382 To generate a new random key call the class method generate() with
383 the appropriate 'kty' parameter, and other parameters as needed (key
384 size, public exponents, curve types, etc..)
386 Valid options per type, when generating new keys:
387 * oct: size(int)
388 * RSA: public_exponent(int), size(int)
389 * EC: crv(str) (one of P-256, P-384, P-521, secp256k1)
390 * OKP: crv(str) (one of Ed25519, Ed448, X25519, X448)
392 Deprecated:
393 Alternatively if the 'generate' parameter is provided with a
394 valid key type as value then a new key will be generated according
395 to the defaults or provided key strength options (type specific).
397 :param \**kwargs: parameters (optional).
399 :raises InvalidJWKType: if the key type is invalid
400 :raises InvalidJWKValue: if incorrect or inconsistent parameters
401 are provided.
402 """
403 super(JWK, self).__init__()
404 self._cache_pub_k = None
405 self._cache_pri_k = None
406 self.unsafe_skip_rsa_key_validation = False
408 if 'generate' in kwargs:
409 self.generate_key(**kwargs)
410 elif kwargs:
411 self.import_key(**kwargs)
413 @classmethod
414 def generate(cls, **kwargs):
415 obj = cls()
416 kty = None
417 try:
418 kty = kwargs['kty']
419 gen = getattr(obj, '_generate_%s' % kty)
420 except (KeyError, AttributeError) as e:
421 raise InvalidJWKType(kty) from e
422 gen(kwargs)
423 return obj
425 def generate_key(self, **params):
426 kty = None
427 try:
428 kty = params.pop('generate')
429 gen = getattr(self, '_generate_%s' % kty)
430 except (KeyError, AttributeError) as e:
431 raise InvalidJWKType(kty) from e
433 gen(params)
435 def _get_gen_size(self, params, default_size=None):
436 size = default_size
437 if 'size' in params:
438 size = params.pop('size')
439 elif 'alg' in params:
440 try:
441 from jwcrypto.jwa import JWA
442 alg = JWA.instantiate_alg(params['alg'])
443 except KeyError as e:
444 raise ValueError("Invalid 'alg' parameter") from e
445 size = alg.input_keysize
446 return size
448 def _generate_oct(self, params):
449 size = self._get_gen_size(params, 128)
450 key = os.urandom(size // 8)
451 params['kty'] = 'oct'
452 params['k'] = base64url_encode(key)
453 self.import_key(**params)
455 def _encode_int(self, i, bit_size=None):
456 extend = 0
457 if bit_size is not None:
458 extend = ((bit_size + 7) // 8) * 2
459 hexi = hex(i).rstrip("L").lstrip("0x")
460 hexl = len(hexi)
461 if extend > hexl:
462 extend -= hexl
463 else:
464 extend = hexl % 2
465 return base64url_encode(unhexlify(extend * '0' + hexi))
467 def _generate_RSA(self, params):
468 pubexp = 65537
469 size = self._get_gen_size(params, 2048)
470 if 'public_exponent' in params:
471 pubexp = params.pop('public_exponent')
472 key = rsa.generate_private_key(pubexp, size, default_backend())
473 self._import_pyca_pri_rsa(key, **params)
475 def _import_pyca_pri_rsa(self, key, **params):
476 pn = key.private_numbers()
477 params.update(
478 kty='RSA',
479 n=self._encode_int(pn.public_numbers.n),
480 e=self._encode_int(pn.public_numbers.e),
481 d=self._encode_int(pn.d),
482 p=self._encode_int(pn.p),
483 q=self._encode_int(pn.q),
484 dp=self._encode_int(pn.dmp1),
485 dq=self._encode_int(pn.dmq1),
486 qi=self._encode_int(pn.iqmp)
487 )
488 self.import_key(**params)
490 def _import_pyca_pub_rsa(self, key, **params):
491 pn = key.public_numbers()
492 params.update(
493 kty='RSA',
494 n=self._encode_int(pn.n),
495 e=self._encode_int(pn.e)
496 )
497 self.import_key(**params)
499 def _get_curve_by_name(self, name, ctype=None):
500 crv = self.get('crv')
502 if name is None:
503 cname = crv
504 elif name == 'P-256K':
505 # P-256K is an alias for 'secp256k1' to handle compatibility
506 # with some implementation using this old drafting name
507 cname = 'secp256k1'
508 else:
509 cname = name
511 # Check we are asking for the correct curve unless this is being
512 # requested for generation on a blank JWK object
513 if crv:
514 ccrv = crv
515 if ccrv == 'P-256K':
516 ccrv = 'secp256k1'
517 if ccrv != cname:
518 raise InvalidJWKValue('Curve requested is "%s", but '
519 'key curve is "%s"' % (name, crv))
520 kty = self.get('kty')
521 if kty is not None and ctype is not None and kty != ctype:
522 raise InvalidJWKType('Curve Requested is of type "%s", but '
523 'key curve is of type "%s"' % (ctype, kty))
525 # Return a curve object
526 if cname == 'P-256':
527 return ec.SECP256R1()
528 elif cname == 'P-384':
529 return ec.SECP384R1()
530 elif cname == 'P-521':
531 return ec.SECP521R1()
532 elif cname == 'secp256k1':
533 return ec.SECP256K1()
534 elif cname == 'BP-256':
535 return ec.BrainpoolP256R1()
536 elif cname == 'BP-384':
537 return ec.BrainpoolP384R1()
538 elif cname == 'BP-512':
539 return ec.BrainpoolP512R1()
540 elif cname in _OKP_CURVES_TABLE:
541 return _OKP_CURVES_TABLE[cname]
542 else:
543 raise InvalidJWKValue('Unknown Curve Name [%s]' % (name))
545 def _generate_EC(self, params):
546 curve = 'P-256'
547 if 'curve' in params:
548 curve = params.pop('curve')
549 # 'curve' is for backwards compat, if 'crv' is defined it takes
550 # precedence
551 if 'crv' in params:
552 curve = params.pop('crv')
553 curve_fn = self._get_curve_by_name(curve, 'EC')
554 key = ec.generate_private_key(curve_fn, default_backend())
555 self._import_pyca_pri_ec(key, **params)
557 def _import_pyca_pri_ec(self, key, **params):
558 pn = key.private_numbers()
559 key_size = pn.public_numbers.curve.key_size
560 params.update(
561 kty='EC',
562 crv=JWKpycaCurveMap[key.curve.name],
563 x=self._encode_int(pn.public_numbers.x, key_size),
564 y=self._encode_int(pn.public_numbers.y, key_size),
565 d=self._encode_int(pn.private_value, key_size)
566 )
567 self.import_key(**params)
569 def _import_pyca_pub_ec(self, key, **params):
570 pn = key.public_numbers()
571 key_size = pn.curve.key_size
572 params.update(
573 kty='EC',
574 crv=JWKpycaCurveMap[key.curve.name],
575 x=self._encode_int(pn.x, key_size),
576 y=self._encode_int(pn.y, key_size),
577 )
578 self.import_key(**params)
580 def _generate_OKP(self, params):
581 if 'crv' not in params:
582 raise InvalidJWKValue('Must specify "crv" for OKP key generation')
583 curve_fn = self._get_curve_by_name(params['crv'], 'OKP')
584 key = curve_fn.privkey.generate()
585 self._import_pyca_pri_okp(key, **params)
587 def _okp_curve_from_pyca_key(self, key):
588 for name, val in _OKP_CURVES_TABLE.items():
589 if isinstance(key, (val.pubkey, val.privkey)):
590 return name
591 raise InvalidJWKValue('Invalid OKP Key object %r' % key)
593 def _import_pyca_pri_okp(self, key, **params):
594 params.update(
595 kty='OKP',
596 crv=self._okp_curve_from_pyca_key(key),
597 d=base64url_encode(key.private_bytes(
598 serialization.Encoding.Raw,
599 serialization.PrivateFormat.Raw,
600 serialization.NoEncryption())),
601 x=base64url_encode(key.public_key().public_bytes(
602 serialization.Encoding.Raw,
603 serialization.PublicFormat.Raw))
604 )
605 self.import_key(**params)
607 def _import_pyca_pub_okp(self, key, **params):
608 params.update(
609 kty='OKP',
610 crv=self._okp_curve_from_pyca_key(key),
611 x=base64url_encode(key.public_bytes(
612 serialization.Encoding.Raw,
613 serialization.PublicFormat.Raw))
614 )
615 self.import_key(**params)
617 def _generate_AKP(self, params):
618 alg = params.pop('alg', None)
619 if alg not in _AKP_ALG_TABLE:
620 raise InvalidJWKValue(
621 'Must specify a valid "alg" for AKP key generation. '
622 'Valid values: %s' % list(_AKP_ALG_TABLE.keys()))
623 if alg not in ImplementedAKPAlgorithms:
624 raise NotImplementedError(
625 'Algorithm "%s" is not available' % alg)
626 privkey_class = _AKP_ALG_TABLE[alg].privkey
627 key = privkey_class.generate()
628 self._import_pyca_pri_akp(key, alg, **params)
630 def _akp_alg_from_pyca_key(self, key):
631 for name, val in _AKP_ALG_TABLE.items():
632 if isinstance(key, (val.pubkey, val.privkey)):
633 return name
634 raise InvalidJWKValue('Invalid AKP Key object %r' % key)
636 def _import_pyca_pri_akp(self, key, alg, **params):
637 params.update(
638 kty='AKP',
639 alg=alg,
640 pub=base64url_encode(key.public_key().public_bytes_raw()),
641 priv=base64url_encode(key.private_bytes_raw()),
642 )
643 self.import_key(**params)
645 def _import_pyca_pub_akp(self, key, alg, **params):
646 params.update(
647 kty='AKP',
648 alg=alg,
649 pub=base64url_encode(key.public_bytes_raw()),
650 )
651 self.import_key(**params)
653 def import_key(self, **kwargs):
654 newkey = {}
655 key_vals = 0
656 self._cache_pub_k = None
657 self._cache_pri_k = None
659 names = list(kwargs.keys())
661 for name in list(JWKParamsRegistry.keys()):
662 if name in kwargs:
663 newkey[name] = kwargs[name]
664 while name in names:
665 names.remove(name)
667 kty = newkey.get('kty')
668 if kty not in JWKTypesRegistry:
669 raise InvalidJWKType(kty)
671 for name in list(JWKValuesRegistry[kty].keys()):
672 if name in kwargs:
673 newkey[name] = kwargs[name]
674 key_vals += 1
675 while name in names:
676 names.remove(name)
678 for name, val in JWKValuesRegistry[kty].items():
679 if val.required and name not in newkey:
680 raise InvalidJWKValue('Missing required value %s' % name)
681 if val.type == ParmType.unsupported and name in newkey:
682 raise InvalidJWKValue('Unsupported parameter %s' % name)
683 if val.type == ParmType.b64 and name in newkey:
684 # Check that the value is base64url encoded
685 try:
686 base64url_decode(newkey[name])
687 except Exception as e: # pylint: disable=broad-except
688 raise InvalidJWKValue(
689 '"%s" is not base64url encoded' % name
690 ) from e
691 if val.type == ParmType.b64u and name in newkey:
692 # Check that the value is Base64urlUInt encoded
693 try:
694 self._decode_int(newkey[name])
695 except Exception as e: # pylint: disable=broad-except
696 raise InvalidJWKValue(
697 '"%s" is not Base64urlUInt encoded' % name
698 ) from e
700 # RFC 9964: 'alg' is required for AKP keys
701 if kty == 'AKP':
702 alg = newkey.get('alg')
703 if alg is None:
704 raise InvalidJWKValue(
705 '"alg" is required for AKP key type')
706 if alg not in _AKP_ALG_TABLE:
707 raise InvalidJWKValue(
708 'Invalid "alg" for AKP key type: "%s". '
709 'Valid values: %s' % (alg, list(_AKP_ALG_TABLE.keys())))
711 # Unknown key parameters are allowed
712 for name in names:
713 newkey[name] = kwargs[name]
715 if key_vals == 0:
716 raise InvalidJWKValue('No Key Values found')
718 # check key_ops
719 if 'key_ops' in newkey:
720 if len(newkey['key_ops']) > len(JWKOperationsRegistry):
721 raise InvalidJWKValue('Unknown or duplicate "key_ops" values')
723 if len(set(newkey['key_ops'])) != len(newkey['key_ops']):
724 raise InvalidJWKValue('Duplicate values in "key_ops"')
726 # check use/key_ops consistency
727 if 'use' in newkey:
728 for op in newkey['key_ops']:
729 if newkey['use'] != JWKOpAndUsePairs.get(op, 'bad'):
730 raise InvalidJWKValue('Incompatible "use" and'
731 ' "key_ops" values specified at'
732 ' the same time')
734 self.clear()
735 # must set 'kty' as first item
736 self.__setitem__('kty', newkey['kty'])
737 self.update(newkey)
739 @classmethod
740 def from_json(cls, key):
741 """Creates a RFC 7517 JWK from the standard JSON format.
743 :param key: The RFC 7517 representation of a JWK.
745 :return: A JWK object that holds the json key.
746 :rtype: JWK
747 """
748 obj = cls()
749 try:
750 jkey = json_decode(key)
751 except Exception as e: # pylint: disable=broad-except
752 raise InvalidJWKValue from e
753 obj.import_key(**jkey)
754 return obj
756 def export(self, private_key=True, as_dict=False):
757 """Exports the key in the standard JSON format.
758 Exports the key regardless of type, if private_key is False
759 and the key is_symmetric an exception is raised.
761 :param private_key(bool): Whether to export the private key.
762 Defaults to True.
764 :return: A portable representation of the key.
765 If as_dict is True then a dictionary is returned.
766 By default a json string
767 :rtype: `str` or `dict`
768 """
769 if private_key is True:
770 # Use _export_all for backwards compatibility, as this
771 # function allows to export symmetric keys too
772 return self._export_all(as_dict)
774 return self.export_public(as_dict)
776 def export_public(self, as_dict=False):
777 """Exports the public key in the standard JSON format.
778 It fails if one is not available like when this function
779 is called on a symmetric key.
781 :param as_dict(bool): If set to True export as python dict not JSON
783 :return: A portable representation of the public key only.
784 If as_dict is True then a dictionary is returned.
785 By default a json string
786 :rtype: `str` or `dict`
787 """
788 pub = self._public_params()
789 if as_dict is True:
790 return pub
791 return json_encode(pub)
793 def _public_params(self):
794 if not self.has_public:
795 raise InvalidJWKType("No public key available")
796 pub = {}
797 reg = JWKParamsRegistry
798 for name in reg:
799 if reg[name].public:
800 if name in self.keys():
801 pub[name] = self.get(name)
802 reg = JWKValuesRegistry[self.get('kty')]
803 for name in reg:
804 if reg[name].public:
805 pub[name] = self.get(name)
806 return pub
808 def _export_all(self, as_dict=False):
809 d = {}
810 d.update(self)
811 if as_dict is True:
812 return d
813 return json_encode(d)
815 def export_private(self, as_dict=False):
816 """Export the private key in the standard JSON format.
817 It fails for a JWK that has only a public key or is symmetric.
819 :param as_dict(bool): If set to True export as python dict not JSON
821 :return: A portable representation of a private key.
822 If as_dict is True then a dictionary is returned.
823 By default a json string
824 :rtype: `str` or `dict`
825 """
826 if self.has_private:
827 return self._export_all(as_dict)
828 raise InvalidJWKType("No private key available")
830 def export_symmetric(self, as_dict=False):
831 if self.is_symmetric:
832 return self._export_all(as_dict)
833 raise InvalidJWKType("Not a symmetric key")
835 def public(self):
836 pub = self._public_params()
837 return JWK(**pub)
839 @property
840 def has_public(self):
841 """Whether this JWK has an asymmetric Public key value."""
842 if self.is_symmetric:
843 return False
844 reg = JWKValuesRegistry[self.get('kty')]
845 for name in reg:
846 if reg[name].public and name in self.keys():
847 return True
848 return False
850 @property
851 def has_private(self):
852 """Whether this JWK has an asymmetric Private key value."""
853 if self.is_symmetric:
854 return False
855 reg = JWKValuesRegistry[self.get('kty')]
856 for name in reg:
857 if not reg[name].public and name in self.keys():
858 return True
859 return False
861 @property
862 def is_symmetric(self):
863 """Whether this JWK is a symmetric key."""
864 return self.get('kty') == 'oct'
866 @property
867 @deprecated('')
868 def key_type(self):
869 """The Key type"""
870 return self.get('kty')
872 @property
873 @deprecated('')
874 def key_id(self):
875 """The Key ID.
876 Provided by the kid parameter if present, otherwise returns None.
877 """
878 return self.get('kid')
880 @property
881 @deprecated('')
882 def key_curve(self):
883 """The Curve Name."""
884 if self.get('kty') not in ['EC', 'OKP']:
885 raise InvalidJWKType('Not an EC or OKP key')
886 return self.get('crv')
888 @deprecated('')
889 def get_curve(self, arg):
890 """Gets the Elliptic Curve associated with the key.
892 :param arg: an optional curve name
894 :raises InvalidJWKType: the key is not an EC or OKP key.
895 :raises InvalidJWKValue: if the curve name is invalid.
897 :return: An EllipticCurve object
898 :rtype: `EllipticCurve`
899 """
900 return self._get_curve_by_name(arg)
902 def _check_constraints(self, usage, operation):
903 use = self.get('use')
904 if use and use != usage:
905 raise InvalidJWKUsage(usage, use)
906 ops = self.get('key_ops')
907 if ops:
908 if not isinstance(ops, list):
909 ops = [ops]
910 if operation not in ops:
911 raise InvalidJWKOperation(operation, ops)
912 # TODO: check alg ?
914 def _decode_int(self, n):
915 return int(hexlify(base64url_decode(n)), 16)
917 def _rsa_pub_n(self):
918 e = self._decode_int(self.get('e'))
919 n = self._decode_int(self.get('n'))
920 return rsa.RSAPublicNumbers(e, n)
922 def _rsa_pri_n(self):
923 p = self._decode_int(self.get('p'))
924 q = self._decode_int(self.get('q'))
925 d = self._decode_int(self.get('d'))
926 dp = self._decode_int(self.get('dp'))
927 dq = self._decode_int(self.get('dq'))
928 qi = self._decode_int(self.get('qi'))
929 return rsa.RSAPrivateNumbers(p, q, d, dp, dq, qi, self._rsa_pub_n())
931 def _rsa_pub(self):
932 k = self._cache_pub_k
933 if k is None:
934 k = self._rsa_pub_n().public_key(default_backend())
935 self._cache_pub_k = k
936 return k
938 def _rsa_pri(self):
939 k = self._cache_pri_k
940 if k is None:
941 u = self.unsafe_skip_rsa_key_validation
942 k = self._rsa_pri_n().private_key(default_backend(),
943 unsafe_skip_rsa_key_validation=u)
944 self._cache_pri_k = k
945 return k
947 def _ec_pub_n(self, curve):
948 x = self._decode_int(self.get('x'))
949 y = self._decode_int(self.get('y'))
950 curve_fn = self._get_curve_by_name(curve, ctype='EC')
951 return ec.EllipticCurvePublicNumbers(x, y, curve_fn)
953 def _ec_pri_n(self, curve):
954 d = self._decode_int(self.get('d'))
955 return ec.EllipticCurvePrivateNumbers(d, self._ec_pub_n(curve))
957 def _ec_pub(self, curve):
958 k = self._cache_pub_k
959 if k is None:
960 k = self._ec_pub_n(curve).public_key(default_backend())
961 self._cache_pub_k = k
962 return k
964 def _ec_pri(self, curve):
965 k = self._cache_pri_k
966 if k is None:
967 k = self._ec_pri_n(curve).private_key(default_backend())
968 self._cache_pri_k = k
969 return k
971 def _okp_pub(self):
972 k = self._cache_pub_k
973 if k is None:
974 crv = self.get('crv')
975 try:
976 pubkey = _OKP_CURVES_TABLE[crv].pubkey
977 except KeyError as e:
978 raise InvalidJWKValue('Unknown curve "%s"' % crv) from e
980 x = base64url_decode(self.get('x'))
981 k = pubkey.from_public_bytes(x)
982 self._cache_pub_k = k
983 return k
985 def _okp_pri(self):
986 k = self._cache_pri_k
987 if k is None:
988 crv = self.get('crv')
989 try:
990 privkey = _OKP_CURVES_TABLE[crv].privkey
991 except KeyError as e:
992 raise InvalidJWKValue('Unknown curve "%s"' % crv) from e
994 d = base64url_decode(self.get('d'))
995 k = privkey.from_private_bytes(d)
996 self._cache_pri_k = k
997 return k
999 def _akp_pub(self):
1000 k = self._cache_pub_k
1001 if k is None:
1002 alg = self.get('alg')
1003 if alg not in _AKP_ALG_TABLE:
1004 raise InvalidJWKValue('Unknown algorithm "%s"' % alg)
1005 pubkey_class = _AKP_ALG_TABLE[alg].pubkey
1006 pub_bytes = base64url_decode(self.get('pub'))
1007 k = pubkey_class.from_public_bytes(pub_bytes)
1008 self._cache_pub_k = k
1009 return k
1011 def _akp_pri(self):
1012 k = self._cache_pri_k
1013 if k is None:
1014 alg = self.get('alg')
1015 if alg not in _AKP_ALG_TABLE:
1016 raise InvalidJWKValue('Unknown algorithm "%s"' % alg)
1017 privkey_class = _AKP_ALG_TABLE[alg].privkey
1018 seed = base64url_decode(self.get('priv'))
1019 k = privkey_class.from_seed_bytes(seed)
1020 self._cache_pri_k = k
1021 return k
1023 def _get_public_key(self, arg=None):
1024 ktype = self.get('kty')
1025 if ktype == 'oct':
1026 return self.get('k')
1027 elif ktype == 'RSA':
1028 return self._rsa_pub()
1029 elif ktype == 'EC':
1030 return self._ec_pub(arg)
1031 elif ktype == 'OKP':
1032 return self._okp_pub()
1033 elif ktype == 'AKP':
1034 return self._akp_pub()
1035 else:
1036 raise NotImplementedError
1038 def _get_private_key(self, arg=None):
1039 ktype = self.get('kty')
1040 if ktype == 'oct':
1041 return self.get('k')
1042 elif ktype == 'RSA':
1043 return self._rsa_pri()
1044 elif ktype == 'EC':
1045 return self._ec_pri(arg)
1046 elif ktype == 'OKP':
1047 return self._okp_pri()
1048 elif ktype == 'AKP':
1049 return self._akp_pri()
1050 else:
1051 raise NotImplementedError
1053 def get_op_key(self, operation=None, arg=None):
1054 """Get the key object associated to the requested operation.
1055 For example the public RSA key for the 'verify' operation or
1056 the private EC key for the 'decrypt' operation.
1058 :param operation: The requested operation.
1059 The valid set of operations is available in the
1060 :data:`JWKOperationsRegistry` registry.
1061 :param arg: An optional, context specific, argument.
1062 For example a curve name.
1064 :raises InvalidJWKOperation: if the operation is unknown or
1065 not permitted with this key.
1066 :raises InvalidJWKUsage: if the use constraints do not permit
1067 the operation.
1069 :return: A Python Cryptography key object for asymmetric keys
1070 or a baseurl64_encoded octet string for symmetric keys
1071 """
1072 validops = self.get('key_ops',
1073 list(JWKOperationsRegistry.keys()))
1074 if validops is not list:
1075 validops = [validops]
1076 if operation is None:
1077 if self.get('kty') == 'oct':
1078 return self.get('k')
1079 raise InvalidJWKOperation(operation, validops)
1080 elif operation == 'sign':
1081 self._check_constraints('sig', operation)
1082 return self._get_private_key(arg)
1083 elif operation == 'verify':
1084 self._check_constraints('sig', operation)
1085 return self._get_public_key(arg)
1086 elif operation == 'encrypt' or operation == 'wrapKey':
1087 self._check_constraints('enc', operation)
1088 return self._get_public_key(arg)
1089 elif operation == 'decrypt' or operation == 'unwrapKey':
1090 self._check_constraints('enc', operation)
1091 return self._get_private_key(arg)
1092 else:
1093 raise NotImplementedError
1095 def import_from_pyca(self, key):
1096 if isinstance(key, rsa.RSAPrivateKey):
1097 self._import_pyca_pri_rsa(key)
1098 elif isinstance(key, rsa.RSAPublicKey):
1099 self._import_pyca_pub_rsa(key)
1100 elif isinstance(key, ec.EllipticCurvePrivateKey):
1101 self._import_pyca_pri_ec(key)
1102 elif isinstance(key, ec.EllipticCurvePublicKey):
1103 self._import_pyca_pub_ec(key)
1104 elif isinstance(key, (Ed25519PrivateKey,
1105 Ed448PrivateKey,
1106 X25519PrivateKey)):
1107 self._import_pyca_pri_okp(key)
1108 elif isinstance(key, (Ed25519PublicKey,
1109 Ed448PublicKey,
1110 X25519PublicKey)):
1111 self._import_pyca_pub_okp(key)
1112 elif isinstance(key, (MLDSA44PrivateKey,
1113 MLDSA65PrivateKey,
1114 MLDSA87PrivateKey)):
1115 alg = self._akp_alg_from_pyca_key(key)
1116 self._import_pyca_pri_akp(key, alg)
1117 elif isinstance(key, (MLDSA44PublicKey,
1118 MLDSA65PublicKey,
1119 MLDSA87PublicKey)):
1120 alg = self._akp_alg_from_pyca_key(key)
1121 self._import_pyca_pub_akp(key, alg)
1122 else:
1123 raise InvalidJWKValue('Unknown key object %r' % key)
1124 self.__setitem__('kid', self.thumbprint())
1126 def import_from_pem(self, data, password=None, kid=None):
1127 """Imports a key from data loaded from a PEM file.
1128 The key may be encrypted with a password.
1129 Private keys (PKCS#8 format), public keys, and X509 certificate's
1130 public keys can be imported with this interface.
1132 :param data(bytes): The data contained in a PEM file.
1133 :param password(bytes): An optional password to unwrap the key.
1134 """
1136 try:
1137 u = self.unsafe_skip_rsa_key_validation
1138 key = serialization.load_pem_private_key(
1139 data, password=password, backend=default_backend(),
1140 unsafe_skip_rsa_key_validation=u)
1141 except ValueError as e:
1142 if password is not None:
1143 raise e
1144 try:
1145 key = serialization.load_pem_public_key(
1146 data, backend=default_backend())
1147 except ValueError:
1148 try:
1149 cert = x509.load_pem_x509_certificate(
1150 data, backend=default_backend())
1151 key = cert.public_key()
1152 except ValueError:
1153 # pylint: disable=raise-missing-from
1154 raise e
1156 self.import_from_pyca(key)
1157 if kid is not None:
1158 self.__setitem__('kid', kid)
1160 def export_to_pem(self, private_key=False, password=False):
1161 """Exports keys to a data buffer suitable to be stored as a PEM file.
1162 Either the public or the private key can be exported to a PEM file.
1163 For private keys the PKCS#8 format is used. If a password is provided
1164 the best encryption method available as determined by the cryptography
1165 module is used to wrap the key.
1167 :param private_key: Whether the private key should be exported.
1168 Defaults to `False` which means the public key is exported by default.
1169 :param password(bytes): A password for wrapping the private key.
1170 Defaults to False which will cause the operation to fail. To avoid
1171 encryption the user must explicitly pass None, otherwise the user
1172 needs to provide a password in a bytes buffer.
1174 :return: A serialized bytes buffer containing a PEM formatted key.
1175 :rtype: `bytes`
1176 """
1177 enc = serialization.Encoding.PEM
1178 if private_key:
1179 if not self.has_private:
1180 raise InvalidJWKType("No private key available")
1181 f = serialization.PrivateFormat.PKCS8
1182 if password is None:
1183 enc_alg = serialization.NoEncryption()
1184 elif isinstance(password, bytes):
1185 enc_alg = serialization.BestAvailableEncryption(password)
1186 elif password is False:
1187 raise ValueError("The password must be None or a bytes string")
1188 else:
1189 raise TypeError("The password string must be bytes")
1190 return self._get_private_key().private_bytes(
1191 encoding=enc, format=f, encryption_algorithm=enc_alg)
1192 else:
1193 if not self.has_public:
1194 raise InvalidJWKType("No public key available")
1195 f = serialization.PublicFormat.SubjectPublicKeyInfo
1196 return self._get_public_key().public_bytes(encoding=enc, format=f)
1198 @classmethod
1199 def from_pyca(cls, key):
1200 obj = cls()
1201 obj.import_from_pyca(key)
1202 return obj
1204 @classmethod
1205 def from_pem(cls, data, password=None):
1206 """Creates a key from PKCS#8 formatted data loaded from a PEM file.
1207 See the function `import_from_pem` for details.
1209 :param data(bytes): The data contained in a PEM file.
1210 :param password(bytes): An optional password to unwrap the key.
1212 :return: A JWK object.
1213 :rtype: JWK
1214 """
1215 obj = cls()
1216 obj.import_from_pem(data, password)
1217 return obj
1219 def thumbprint(self, hashalg=hashes.SHA256()):
1220 """Returns the key thumbprint as specified by RFC 7638.
1222 :param hashalg: A hash function (defaults to SHA256)
1224 :return: A base64url encoded digest of the key
1225 :rtype: `str`
1226 """
1228 t = {'kty': self.get('kty')}
1229 for name, val in JWKValuesRegistry[t['kty']].items():
1230 if val.required:
1231 t[name] = self.get(name)
1232 # RFC 9964 Section 6: AKP thumbprint includes 'alg'
1233 if t['kty'] == 'AKP':
1234 t['alg'] = self.get('alg')
1235 digest = hashes.Hash(hashalg, backend=default_backend())
1236 digest.update(bytes(json_encode(t).encode('utf8')))
1237 return base64url_encode(digest.finalize())
1239 def thumbprint_uri(self, hname='sha-256'):
1240 """Returns the key thumbprint URI as specified by RFC 9278.
1242 :param hname: A hash function name as specified in IANA's
1243 Named Information registry:
1244 https://www.iana.org/assignments/named-information/
1245 Values from `IANANamedInformationHashAlgorithmRegistry`
1247 :return: A JWK Thumbprint URI
1248 :rtype: `str`
1249 """
1251 try:
1252 h = IANANamedInformationHashAlgorithmRegistry[hname]
1253 except KeyError as e:
1254 raise InvalidJWKValue('Unknown hash "{}"'.format(hname)) from e
1255 if h is None:
1256 raise InvalidJWKValue('Unsupported hash "{}"'.format(hname))
1258 t = self.thumbprint(h)
1259 return "urn:ietf:params:oauth:jwk-thumbprint:{}:{}".format(hname, t)
1261 # Methods to constrain what this dict allows
1262 def __setitem__(self, item, value):
1263 kty = self.get('kty')
1265 if item == 'kty':
1266 if kty is None:
1267 if value not in JWKTypesRegistry:
1268 raise InvalidJWKType(value)
1269 super(JWK, self).__setitem__(item, value)
1270 return
1271 elif kty != value:
1272 raise ValueError('Cannot change key type')
1274 # Check if item is a key value and verify its format
1275 if item in list(JWKValuesRegistry[kty].keys()):
1276 # Invalidate cached keys if any
1277 self._cache_pub_k = None
1278 self._cache_pri_k = None
1279 if JWKValuesRegistry[kty][item].type == ParmType.b64:
1280 try:
1281 v = base64url_decode(value)
1282 # empty values are also invalid except for the
1283 # special case of 'oct' key where an empty value
1284 # is used to indicate a 'None' key
1285 if v == b'' and kty != 'oct' and item != 'k':
1286 raise ValueError
1287 except Exception as e: # pylint: disable=broad-except
1288 raise InvalidJWKValue(
1289 '"%s" is not base64url encoded' % item
1290 ) from e
1291 elif JWKValuesRegistry[kty][item].type == ParmType.b64u:
1292 try:
1293 self._decode_int(value)
1294 except Exception as e: # pylint: disable=broad-except
1295 raise InvalidJWKValue(
1296 '"%s" is not Base64urlUInt encoded' % item
1297 ) from e
1298 super(JWK, self).__setitem__(item, value)
1299 return
1301 # If not a key param check if it is a know parameter
1302 if item in list(JWKParamsRegistry.keys()):
1303 super(JWK, self).__setitem__(item, value)
1304 return
1306 # if neither a key param nor a known parameter, check if we are
1307 # trying to set a parameter for a different key type and refuse
1308 # in this case.
1309 for name in list(JWKTypesRegistry.keys()):
1310 if name == kty:
1311 continue
1312 if item in list(JWKValuesRegistry[name].keys()):
1313 raise KeyError("Cannot set '{}' on '{}' key type".format(
1314 item, kty))
1316 # ok if we've come this far it means we have an unknown parameter
1317 super(JWK, self).__setitem__(item, value)
1319 def update(self, *args, **kwargs):
1320 r"""
1321 :param \*args: arguments
1322 :param \**kwargs: keyword arguments
1323 """
1324 for k, v in dict(*args, **kwargs).items():
1325 self.__setitem__(k, v)
1327 def setdefault(self, key, default=None):
1328 if key not in self.keys():
1329 self.__setitem__(key, default)
1330 return self.get(key)
1332 def __delitem__(self, item):
1333 param = self.get(item)
1334 if param is None:
1335 raise KeyError(item)
1337 if item == 'kty':
1338 for name in list(JWKValuesRegistry[param].keys()):
1339 if self.get(name) is not None:
1340 raise KeyError("Cannot remove 'kty', values present")
1342 kty = self.get('kty')
1343 if kty is not None and item in list(JWKValuesRegistry[kty].keys()):
1344 # Invalidate cached keys if any
1345 self._cache_pub_k = None
1346 self._cache_pri_k = None
1348 super(JWK, self).__delitem__(item)
1350 def __eq__(self, other):
1351 if not isinstance(other, JWK):
1352 return NotImplemented
1354 return self.thumbprint() == other.thumbprint() and \
1355 self.get('kid') == other.get('kid')
1357 def __hash__(self):
1358 return hash((self.thumbprint(), self.get('kid')))
1360 def __getattr__(self, item):
1361 try:
1362 if item in JWKParamsRegistry.keys():
1363 if item in self.keys():
1364 return self.get(item)
1365 kty = self.get('kty')
1366 if kty is not None:
1367 if item in list(JWKValuesRegistry[kty].keys()):
1368 if item in self.keys():
1369 return self.get(item)
1370 raise KeyError
1371 except KeyError:
1372 raise AttributeError(item) from None
1374 def __setattr__(self, item, value):
1375 try:
1376 if item in JWKParamsRegistry.keys():
1377 self.__setitem__(item, value)
1378 for name in list(JWKTypesRegistry.keys()):
1379 if item in list(JWKValuesRegistry[name].keys()):
1380 self.__setitem__(item, value)
1381 super(JWK, self).__setattr__(item, value)
1382 except KeyError:
1383 raise AttributeError(item) from None
1385 @classmethod
1386 def from_password(cls, password):
1387 """Creates a symmetric JWK key from a user password.
1389 :param password: A password in utf8 format.
1391 :return: a JWK object
1392 :rtype: JWK
1393 """
1394 obj = cls()
1395 params = {'kty': 'oct'}
1396 try:
1397 params['k'] = base64url_encode(password.encode('utf8'))
1398 except Exception as e: # pylint: disable=broad-except
1399 raise InvalidJWKValue from e
1400 obj.import_key(**params)
1401 return obj
1403 # Prevent accidental disclosure of key material via repr()
1404 def __repr__(self):
1405 repr_dict = {}
1406 repr_dict['kid'] = self.get('kid', 'Missing Key ID')
1407 repr_dict['thumbprint'] = self.thumbprint()
1408 return json_encode(repr_dict)
1411class _JWKkeys(set):
1413 def add(self, elem):
1414 """Adds a JWK object to the set
1416 :param elem: the JWK object to add.
1418 :raises TypeError: if the object is not a JWK.
1419 """
1420 if not isinstance(elem, JWK):
1421 raise TypeError('Only JWK objects are valid elements')
1422 set.add(self, elem)
1425class JWKSet(dict):
1426 """A set of JWK objects.
1428 Inherits from the standard 'dict' builtin type.
1429 Creates a special key 'keys' that is of a type derived from 'set'
1430 The 'keys' attribute accepts only :class:`jwcrypto.jwk.JWK` elements.
1431 """
1433 def __init__(self, *args, **kwargs):
1434 super(JWKSet, self).__init__()
1435 super(JWKSet, self).__setitem__('keys', _JWKkeys())
1436 self.update(*args, **kwargs)
1438 def __iter__(self):
1439 return self['keys'].__iter__()
1441 def __contains__(self, key):
1442 return self['keys'].__contains__(key)
1444 def __setitem__(self, key, val):
1445 if key == 'keys' and not isinstance(val, _JWKkeys):
1446 self['keys'].add(val)
1447 else:
1448 super(JWKSet, self).__setitem__(key, val)
1450 def update(self, *args, **kwargs):
1451 r"""
1452 :param \*args: arguments
1453 :param \**kwargs: keyword arguments
1454 """
1455 for k, v in dict(*args, **kwargs).items():
1456 self.__setitem__(k, v)
1458 def setdefault(self, key, default=None):
1459 if key not in self.keys():
1460 self.__setitem__(key, default)
1461 return self.get(key)
1463 def add(self, elem):
1464 self['keys'].add(elem)
1466 def export(self, private_keys=True, as_dict=False):
1467 """Exports a RFC 7517 key set.
1468 Exports as json by default, or as dict if requested.
1470 :param private_key(bool): Whether to export private keys.
1471 Defaults to True.
1472 :param as_dict(bool): Whether to return a dict instead of
1473 a JSON object
1475 :return: A portable representation of the key set.
1476 If as_dict is True then a dictionary is returned.
1477 By default a json string
1478 :rtype: `str` or `dict`
1479 """
1480 exp_dict = {}
1481 for k, v in self.items():
1482 if k == 'keys':
1483 keys = []
1484 for jwk in v:
1485 keys.append(jwk.export(private_keys, as_dict=True))
1486 v = keys
1487 exp_dict[k] = v
1488 if as_dict is True:
1489 return exp_dict
1490 return json_encode(exp_dict)
1492 def import_keyset(self, keyset):
1493 """Imports a RFC 7517 key set using the standard JSON format.
1495 :param keyset: The RFC 7517 representation of a JOSE key set.
1496 """
1497 try:
1498 jwkset = json_decode(keyset)
1499 if 'keys' not in jwkset:
1500 raise ValueError("'keys' not in set")
1502 for k, v in jwkset.items():
1503 if k == 'keys':
1504 for jwk in v:
1505 try:
1506 self['keys'].add(JWK(**jwk))
1507 except InvalidJWKType:
1508 pass
1509 else:
1510 self[k] = v
1511 except Exception as e: # pylint: disable=broad-except
1512 raise InvalidJWKValue from e
1514 @classmethod
1515 def from_json(cls, keyset):
1516 """Creates a RFC 7517 key set from the standard JSON format.
1518 :param keyset: The RFC 7517 representation of a JOSE key set.
1520 :return: A JWKSet object.
1521 :rtype: JWKSet
1522 """
1523 obj = cls()
1524 obj.import_keyset(keyset)
1525 return obj
1527 def get_key(self, kid):
1528 """Gets a key from the set.
1529 :param kid: the 'kid' key identifier.
1531 :return: A JWK from the set
1532 :rtype: JWK
1533 """
1534 keys = self.get_keys(kid)
1535 if len(keys) > 1:
1536 raise InvalidJWKValue(
1537 'Duplicate keys found with requested kid: 1 expected')
1538 try:
1539 return tuple(keys)[0]
1540 except IndexError:
1541 return None
1543 def get_keys(self, kid):
1544 """Gets keys from the set with matching kid.
1545 :param kid: the 'kid' key identifier.
1547 :return: a List of keys
1548 :rtype: `list`
1549 """
1550 return {key for key in self['keys'] if key.get('kid') == kid}
1552 def __repr__(self):
1553 repr_dict = {}
1554 for k, v in self.items():
1555 if k == 'keys':
1556 keys = []
1557 for jwk in v:
1558 keys.append(repr(jwk))
1559 v = keys
1560 repr_dict[k] = v
1561 return json_encode(repr_dict)