Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.11/site-packages/jwcrypto/jws.py: 25%
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
3from jwcrypto.common import JWException, JWKeyNotFound
4from jwcrypto.common import JWSEHeaderParameter, JWSEHeaderRegistry
5from jwcrypto.common import base64url_decode, base64url_encode
6from jwcrypto.common import json_decode, json_encode
7from jwcrypto.jwa import JWA
8from jwcrypto.jwk import JWK, JWKSet
10JWSHeaderRegistry = {
11 'alg': JWSEHeaderParameter('Algorithm', False, True, None),
12 'jku': JWSEHeaderParameter('JWK Set URL', False, False, None),
13 'jwk': JWSEHeaderParameter('JSON Web Key', False, False, None),
14 'kid': JWSEHeaderParameter('Key ID', False, True, None),
15 'x5u': JWSEHeaderParameter('X.509 URL', False, False, None),
16 'x5c': JWSEHeaderParameter('X.509 Certificate Chain', False, False, None),
17 'x5t': JWSEHeaderParameter(
18 'X.509 Certificate SHA-1 Thumbprint', False, False, None),
19 'x5t#S256': JWSEHeaderParameter(
20 'X.509 Certificate SHA-256 Thumbprint', False, False, None),
21 'typ': JWSEHeaderParameter('Type', False, True, None),
22 'cty': JWSEHeaderParameter('Content Type', False, True, None),
23 'crit': JWSEHeaderParameter('Critical', True, True, None),
24 'b64': JWSEHeaderParameter('Base64url-Encode Payload', True, True, None)
25}
26"""Registry of valid header parameters"""
28default_allowed_algs = [
29 'HS256', 'HS384', 'HS512',
30 'RS256', 'RS384', 'RS512',
31 'ES256', 'ES384', 'ES512',
32 'PS256', 'PS384', 'PS512',
33 'EdDSA', 'ES256K', 'Ed25519',
34 'Ed448',
35 'ML-DSA-44', 'ML-DSA-65', 'ML-DSA-87',
36]
37"""Default allowed algorithms"""
40class InvalidJWSSignature(JWException):
41 """Invalid JWS Signature.
43 This exception is raised when a signature cannot be validated.
44 """
46 def __init__(self, message=None, exception=None):
47 msg = None
48 if message:
49 msg = str(message)
50 else:
51 msg = 'Unknown Signature Verification Failure'
52 if exception:
53 msg += ' {%s}' % str(exception)
54 super(InvalidJWSSignature, self).__init__(msg)
57class InvalidJWSObject(JWException):
58 """Invalid JWS Object.
60 This exception is raised when the JWS Object is invalid and/or
61 improperly formatted.
62 """
64 def __init__(self, message=None, exception=None):
65 msg = 'Invalid JWS Object'
66 if message:
67 msg += ' [%s]' % message
68 if exception:
69 msg += ' {%s}' % str(exception)
70 super(InvalidJWSObject, self).__init__(msg)
73class InvalidJWSOperation(JWException):
74 """Invalid JWS Object.
76 This exception is raised when a requested operation cannot
77 be execute due to unsatisfied conditions.
78 """
80 def __init__(self, message=None, exception=None):
81 msg = None
82 if message:
83 msg = message
84 else:
85 msg = 'Unknown Operation Failure'
86 if exception:
87 msg += ' {%s}' % str(exception)
88 super(InvalidJWSOperation, self).__init__(msg)
91class JWSCore:
92 """The inner JWS Core object.
94 This object SHOULD NOT be used directly, the JWS object should be
95 used instead as JWS perform necessary checks on the validity of
96 the object and requested operations.
98 """
100 def __init__(self, alg, key, header, payload, algs=None):
101 """Core JWS token handling.
103 :param alg: The algorithm used to produce the signature.
104 See RFC 7518
105 :param key: A (:class:`jwcrypto.jwk.JWK`) verification or
106 a (:class:`jwcrypto.jwk.JWKSet`) that contains a key indexed by the
107 'kid' header. A JWKSet is allowed only for verification operations.
108 :param header: A JSON string representing the protected header.
109 :param payload(bytes): An arbitrary value
110 :param algs: An optional list of allowed algorithms
112 :raises ValueError: if the key is not a (:class:`jwcrypto.jwk.JWK`)
113 :raises InvalidJWAAlgorithm: if the algorithm is not valid, is
114 unknown or otherwise not yet implemented.
115 :raises InvalidJWSOperation: if the algorithm is not allowed.
116 """
117 self.alg = alg
118 self.engine = self._jwa(alg, algs)
119 self.key = key
121 if header is not None:
122 if isinstance(header, dict):
123 header = json_encode(header)
124 # Make sure this is always a deep copy of the dict
125 self.header = json_decode(header)
127 self.protected = base64url_encode(header.encode('utf-8'))
128 else:
129 self.header = {}
130 self.protected = ''
131 self.payload = self._payload(payload)
133 def _jwa(self, name, allowed):
134 if allowed is None:
135 allowed = default_allowed_algs
136 if name not in allowed:
137 raise InvalidJWSOperation('Algorithm not allowed')
138 return JWA.signing_alg(name)
140 def _payload(self, payload):
141 if self.header.get('b64', True):
142 return base64url_encode(payload).encode('utf-8')
143 else:
144 if isinstance(payload, bytes):
145 return payload
146 else:
147 return payload.encode('utf-8')
149 def sign(self):
150 """Generates a signature"""
151 if not isinstance(self.key, JWK):
152 raise ValueError('key is not a JWK object')
153 sigin = b'.'.join([self.protected.encode('utf-8'),
154 self.payload])
155 signature = self.engine.sign(self.key, sigin)
156 return {'protected': self.protected,
157 'payload': self.payload,
158 'signature': base64url_encode(signature)}
160 def verify(self, signature):
161 """Verifies a signature
163 :raises InvalidJWSSignature: if the verification fails.
165 :return: Returns True or an Exception
166 :rtype: `bool`
167 """
168 try:
169 sigin = b'.'.join([self.protected.encode('utf-8'),
170 self.payload])
171 self.engine.verify(self.key, sigin, signature)
172 except Exception as e: # pylint: disable=broad-except
173 raise InvalidJWSSignature('Verification failed') from e
174 return True
177class JWS:
178 """JSON Web Signature object
180 This object represent a JWS token.
181 """
183 def __init__(self, payload=None, header_registry=None):
184 """Creates a JWS object.
186 :param payload(bytes): An arbitrary value (optional).
187 :param header_registry: Optional additions to the header registry
188 """
189 self.objects = {}
190 self.objects['payload'] = payload
191 self.verifylog = None
192 self._allowed_algs = None
193 self.header_registry = JWSEHeaderRegistry(JWSHeaderRegistry)
194 if header_registry:
195 self.header_registry.update(header_registry)
197 @property
198 def allowed_algs(self):
199 """Allowed algorithms.
201 The list of allowed algorithms.
202 Can be changed by setting a list of algorithm names.
203 """
205 if self._allowed_algs:
206 return self._allowed_algs
207 else:
208 return default_allowed_algs
210 @allowed_algs.setter
211 def allowed_algs(self, algs):
212 if not isinstance(algs, list):
213 raise TypeError('Allowed Algs must be a list')
214 self._allowed_algs = algs
216 @property
217 def is_valid(self):
218 return self.objects.get('valid', False)
220 # TODO: allow caller to specify list of headers it understands
221 # FIXME: Merge and check to be changed to two separate functions
222 def _merge_check_headers(self, protected, *headers):
223 header = None
224 crit = []
225 if protected is not None:
226 if 'crit' in protected:
227 crit = protected['crit']
228 # Check immediately if we support these critical headers
229 for k in crit:
230 if k not in self.header_registry:
231 raise InvalidJWSObject(
232 'Unknown critical header: "%s"' % k)
233 else:
234 if not self.header_registry[k].supported:
235 raise InvalidJWSObject(
236 'Unsupported critical header: "%s"' % k)
237 header = protected
238 if 'b64' in header:
239 if not isinstance(header['b64'], bool):
240 raise InvalidJWSObject('b64 header must be a boolean')
242 for hn in headers:
243 if hn is None:
244 continue
245 if header is None:
246 header = {}
247 for h in list(hn.keys()):
248 if h in self.header_registry:
249 if self.header_registry[h].mustprotect:
250 raise InvalidJWSObject('"%s" must be protected' % h)
251 if h in header:
252 raise InvalidJWSObject('Duplicate header: "%s"' % h)
253 header.update(hn)
255 for k in crit:
256 if k not in header:
257 raise InvalidJWSObject('Missing critical header "%s"' % k)
259 return header
261 def _verify(self, alg, key, payload, signature, protected, header=None):
262 p = {}
263 # verify it is a valid JSON object and decode
264 if protected is not None:
265 p = json_decode(protected)
266 if not isinstance(p, dict):
267 raise InvalidJWSSignature('Invalid Protected header')
268 # merge headers, and verify there are no duplicates
269 if header:
270 if not isinstance(header, dict):
271 raise InvalidJWSSignature('Invalid Unprotected header')
273 # Merge and check (critical) headers
274 chk_hdrs = self._merge_check_headers(p, header)
275 for hdr in chk_hdrs:
276 if hdr in self.header_registry:
277 if not self.header_registry.check_header(hdr, self):
278 raise InvalidJWSSignature('Failed header check')
280 # check 'alg' is present
281 if alg is None and 'alg' not in p:
282 raise InvalidJWSSignature('No "alg" in headers')
283 if alg:
284 if 'alg' in p and alg != p['alg']:
285 raise InvalidJWSSignature(
286 '"alg" mismatch, requested'
287 f''' "{alg}", found "{p['alg']}"'''
288 )
289 resulting_alg = alg
290 else:
291 resulting_alg = p['alg']
293 # the following will verify the "alg" is supported and the signature
294 # verifies
295 if isinstance(key, JWK):
296 signer = JWSCore(resulting_alg, key, protected,
297 payload, self._allowed_algs)
298 signer.verify(signature)
299 self.verifylog.append("Success")
300 elif isinstance(key, JWKSet):
301 keys = key
302 if 'kid' in chk_hdrs:
303 kid_keys = key.get_keys(chk_hdrs['kid'])
304 if not kid_keys:
305 raise JWKeyNotFound('Key ID {} not in key set'.format(
306 chk_hdrs['kid']))
307 keys = kid_keys
309 for k in keys:
310 try:
311 signer2 = JWSCore(
312 resulting_alg, k, protected,
313 payload, self._allowed_algs
314 )
315 signer2.verify(signature)
316 self.verifylog.append("Success")
317 break
318 except Exception as e: # pylint: disable=broad-except
319 keyid = k.get('kid', k.thumbprint())
320 self.verifylog.append('Key [{}] failed: [{}]'.format(
321 keyid, repr(e)))
322 if "Success" not in self.verifylog:
323 raise JWKeyNotFound('No working key found in key set')
324 else:
325 raise ValueError("Unrecognized key type")
327 # Helper to deal with detached payloads in verification
328 def _get_obj_payload(self, obj, dp):
329 op = obj.get('payload')
330 if dp is not None:
331 if op is None or len(op) == 0:
332 return dp
333 else:
334 raise InvalidJWSOperation('Object Payload present but'
335 ' Detached Payload provided')
336 return op
338 def verify(self, key, alg=None, detached_payload=None):
339 """Verifies a JWS token.
341 :param key: A (:class:`jwcrypto.jwk.JWK`) verification or
342 a (:class:`jwcrypto.jwk.JWKSet`) that contains a key indexed by the
343 'kid' header.
344 :param alg: The signing algorithm (optional). Usually the algorithm
345 is known as it is provided with the JOSE Headers of the token.
346 :param detached_payload: A detached payload to verify the signature
347 against. Only valid for tokens that are not carrying a payload.
349 :raises InvalidJWSSignature: if the verification fails.
350 :raises InvalidJWSOperation: if a detached_payload is provided but
351 an object payload exists
352 :raises JWKeyNotFound: if key is a JWKSet and the key is not found.
353 """
355 self.verifylog = []
356 self.objects['valid'] = False
357 obj = self.objects
358 missingkey = False
359 if 'signature' in obj:
360 payload = self._get_obj_payload(obj, detached_payload)
361 try:
362 self._verify(alg, key,
363 payload,
364 obj['signature'],
365 obj.get('protected', None),
366 obj.get('header', None))
367 obj['valid'] = True
368 except Exception as e: # pylint: disable=broad-except
369 if isinstance(e, JWKeyNotFound):
370 missingkey = True
371 self.verifylog.append('Failed: [%s]' % repr(e))
373 elif 'signatures' in obj:
374 payload = self._get_obj_payload(obj, detached_payload)
375 for o in obj['signatures']:
376 try:
377 self._verify(alg, key,
378 payload,
379 o['signature'],
380 o.get('protected', None),
381 o.get('header', None))
382 # Ok if at least one verifies
383 obj['valid'] = True
384 except Exception as e: # pylint: disable=broad-except
385 if isinstance(e, JWKeyNotFound):
386 missingkey = True
387 self.verifylog.append('Failed: [%s]' % repr(e))
388 else:
389 raise InvalidJWSSignature('No signatures available')
391 if not self.is_valid:
392 if missingkey:
393 raise JWKeyNotFound('No working key found in key set')
394 raise InvalidJWSSignature('Verification failed for all '
395 'signatures' + repr(self.verifylog))
397 def _deserialize_signature(self, s):
398 o = {'signature': base64url_decode(str(s['signature']))}
399 if 'protected' in s:
400 p = base64url_decode(str(s['protected']))
401 o['protected'] = p.decode('utf-8')
402 if 'header' in s:
403 o['header'] = s['header']
404 return o
406 def _deserialize_b64(self, o, protected):
407 if protected is None:
408 b64n = None
409 else:
410 p = json_decode(protected)
411 b64n = p.get('b64')
412 if b64n is not None:
413 if not isinstance(b64n, bool):
414 raise InvalidJWSObject('b64 header must be boolean')
415 b64 = o.get('b64')
416 if b64 == b64n:
417 return
418 elif b64 is None:
419 o['b64'] = b64n
420 else:
421 raise InvalidJWSObject('conflicting b64 values')
423 def deserialize(self, raw_jws, key=None, alg=None):
424 """Deserialize a JWS token.
426 NOTE: Destroys any current status and tries to import the raw
427 JWS provided.
429 If a key is provided a verification step will be attempted after
430 the object is successfully deserialized.
432 :param raw_jws: a 'raw' JWS token (JSON Encoded or Compact
433 notation) string.
434 :param key: A (:class:`jwcrypto.jwk.JWK`) verification or
435 a (:class:`jwcrypto.jwk.JWKSet`) that contains a key indexed by the
436 'kid' header (optional).
437 :param alg: The signing algorithm (optional). Usually the algorithm
438 is known as it is provided with the JOSE Headers of the token.
440 :raises InvalidJWSObject: if the raw object is an invalid JWS token.
441 :raises InvalidJWSSignature: if the verification fails.
442 :raises JWKeyNotFound: if key is a JWKSet and the key is not found.
443 """
444 self.objects = {}
445 o = {}
446 try:
447 try:
448 djws = json_decode(raw_jws)
449 if 'signatures' in djws:
450 o['signatures'] = []
451 for s in djws['signatures']:
452 os = self._deserialize_signature(s)
453 o['signatures'].append(os)
454 self._deserialize_b64(o, os.get('protected'))
455 else:
456 o = self._deserialize_signature(djws)
457 self._deserialize_b64(o, o.get('protected'))
459 if 'payload' in djws:
460 if o.get('b64', True):
461 o['payload'] = base64url_decode(str(djws['payload']))
462 else:
463 o['payload'] = djws['payload']
465 except ValueError:
466 # process up to 3 dots, if we have 3+ dots we get more than 3
467 # chunks and that means we have a malformed JWS
468 data = raw_jws.split('.', 3)
469 if len(data) != 3:
470 raise InvalidJWSObject('Unrecognized'
471 ' representation') from None
472 p = base64url_decode(str(data[0]))
473 if len(p) > 0:
474 o['protected'] = p.decode('utf-8')
475 self._deserialize_b64(o, o['protected'])
476 o['payload'] = base64url_decode(str(data[1]))
477 o['signature'] = base64url_decode(str(data[2]))
479 self.objects = o
481 except Exception as e: # pylint: disable=broad-except
482 raise InvalidJWSObject('Invalid format') from e
484 if key:
485 self.verify(key, alg)
487 def add_signature(self, key, alg=None, protected=None, header=None):
488 """Adds a new signature to the object.
490 :param key: A (:class:`jwcrypto.jwk.JWK`) key of appropriate for
491 the "alg" provided.
492 :param alg: An optional algorithm name. If already provided as an
493 element of the protected or unprotected header it can be safely
494 omitted.
495 :param protected: The Protected Header (optional)
496 :param header: The Unprotected Header (optional)
498 :raises InvalidJWSObject: if invalid headers are provided.
499 :raises ValueError: if the key is not a (:class:`jwcrypto.jwk.JWK`)
500 :raises ValueError: if the algorithm is missing or is not provided
501 by one of the headers.
502 :raises InvalidJWAAlgorithm: if the algorithm is not valid, is
503 unknown or otherwise not yet implemented.
504 """
506 b64 = True
508 if protected:
509 if isinstance(protected, dict):
510 protected = json_encode(protected)
511 # Make sure p is always a deep copy of the dict
512 p = json_decode(protected)
513 else:
514 p = dict()
516 # If b64 is present we must enforce criticality
517 if 'b64' in list(p.keys()):
518 crit = p.get('crit', [])
519 if 'b64' not in crit:
520 raise InvalidJWSObject('b64 header must always be critical')
521 b64 = p['b64']
523 if 'b64' in self.objects:
524 if b64 != self.objects['b64']:
525 raise InvalidJWSObject('Mixed b64 headers on signatures')
527 h = None
528 if header:
529 if isinstance(header, dict):
530 header = json_encode(header)
531 # Make sure h is always a deep copy of the dict
532 h = json_decode(header)
534 p = self._merge_check_headers(p, h)
536 if 'alg' in p:
537 if alg is None:
538 alg = p['alg']
539 elif alg != p['alg']:
540 raise ValueError('"alg" value mismatch, specified "alg" '
541 'does not match JOSE header value')
543 if alg is None:
544 raise ValueError('"alg" not specified')
546 c = JWSCore(
547 alg, key, protected, self.objects.get('payload'),
548 self.allowed_algs
549 )
550 sig = c.sign()
552 o = {
553 'signature': base64url_decode(sig['signature']),
554 'valid': True,
555 }
556 if protected:
557 o['protected'] = protected
558 if header:
559 o['header'] = h
561 if 'signatures' in self.objects:
562 self.objects['signatures'].append(o)
563 elif 'signature' in self.objects:
564 self.objects['signatures'] = []
565 n = {'signature': self.objects.pop('signature')}
566 if 'protected' in self.objects:
567 n['protected'] = self.objects.pop('protected')
568 if 'header' in self.objects:
569 n['header'] = self.objects.pop('header')
570 if 'valid' in self.objects:
571 n['valid'] = self.objects.pop('valid')
572 self.objects['signatures'].append(n)
573 self.objects['signatures'].append(o)
574 else:
575 self.objects.update(o)
576 self.objects['b64'] = b64
578 def serialize(self, compact=False):
579 """Serializes the object into a JWS token.
581 :param compact(boolean): if True generates the compact
582 representation, otherwise generates a standard JSON format.
584 :raises InvalidJWSOperation: if the object cannot serialized
585 with the compact representation and `compact` is True.
586 :raises InvalidJWSSignature: if no signature has been added
587 to the object, or no valid signature can be found.
589 :return: A json formatted string or a compact representation string
590 :rtype: `str`
591 """
592 if compact:
593 if 'signatures' in self.objects:
594 raise InvalidJWSOperation("Can't use compact encoding with "
595 "multiple signatures")
596 if 'signature' not in self.objects:
597 raise InvalidJWSSignature("No available signature")
598 if not self.objects.get('valid', False):
599 raise InvalidJWSSignature("No valid signature found")
600 if 'protected' in self.objects:
601 p = json_decode(self.objects['protected'])
602 if 'alg' not in p:
603 raise InvalidJWSOperation("Compact encoding must carry "
604 "'alg' in protected header")
605 protected = base64url_encode(self.objects['protected'])
606 else:
607 raise InvalidJWSOperation("Can't use compact encoding "
608 "without protected header")
609 if self.objects.get('payload'):
610 if self.objects.get('b64', True):
611 payload = base64url_encode(self.objects['payload'])
612 else:
613 if isinstance(self.objects['payload'], bytes):
614 payload = self.objects['payload'].decode('utf-8')
615 else:
616 payload = self.objects['payload']
617 if '.' in payload:
618 raise InvalidJWSOperation(
619 "Can't use compact encoding with unencoded "
620 "payload that uses the . character")
621 else:
622 payload = ''
623 return '.'.join([protected, payload,
624 base64url_encode(self.objects['signature'])])
625 else:
626 obj = self.objects
627 sig = {}
628 payload = self.objects.get('payload', '')
629 if self.objects.get('b64', True):
630 sig['payload'] = base64url_encode(payload)
631 else:
632 sig['payload'] = payload
633 if 'signature' in obj:
634 if not obj.get('valid', False):
635 raise InvalidJWSSignature("No valid signature found")
636 sig['signature'] = base64url_encode(obj['signature'])
637 if 'protected' in obj:
638 sig['protected'] = base64url_encode(obj['protected'])
639 if 'header' in obj:
640 sig['header'] = obj['header']
641 elif 'signatures' in obj:
642 sig['signatures'] = []
643 for o in obj['signatures']:
644 if not o.get('valid', False):
645 continue
646 s = {'signature': base64url_encode(o['signature'])}
647 if 'protected' in o:
648 s['protected'] = base64url_encode(o['protected'])
649 if 'header' in o:
650 s['header'] = o['header']
651 sig['signatures'].append(s)
652 if len(sig['signatures']) == 0:
653 raise InvalidJWSSignature("No valid signature found")
654 else:
655 raise InvalidJWSSignature("No available signature")
656 return json_encode(sig)
658 @property
659 def payload(self):
660 if not self.is_valid:
661 raise InvalidJWSOperation("Payload not verified")
662 return self.objects.get('payload')
664 def detach_payload(self):
665 self.objects.pop('payload', None)
667 @property
668 def jose_header(self):
669 obj = self.objects
670 if 'signature' in obj:
671 if 'protected' in obj:
672 p = json_decode(obj['protected'])
673 else:
674 p = None
675 return self._merge_check_headers(p, obj.get('header', {}))
676 elif 'signatures' in self.objects:
677 jhl = []
678 for o in obj['signatures']:
679 jh = {}
680 if 'protected' in o:
681 p = json_decode(o['protected'])
682 else:
683 p = None
684 jh = self._merge_check_headers(p, o.get('header', {}))
685 jhl.append(jh)
686 return jhl
687 else:
688 raise InvalidJWSOperation("JOSE Header(s) not available")
690 @classmethod
691 def from_jose_token(cls, token):
692 """Creates a JWS object from a serialized JWS token.
694 :param token: A string with the json or compat representation
695 of the token.
697 :raises InvalidJWSObject: if the raw object is an invalid JWS token.
699 :return: A JWS token
700 :rtype: JWS
701 """
703 obj = cls()
704 obj.deserialize(token)
705 return obj
707 def __eq__(self, other):
708 if not isinstance(other, JWS):
709 return False
710 try:
711 return self.serialize() == other.serialize()
712 except Exception: # pylint: disable=broad-except
713 return self.objects == other.objects
715 def __str__(self):
716 try:
717 return self.serialize()
718 except Exception: # pylint: disable=broad-except
719 return self.__repr__()
721 def __repr__(self):
722 try:
723 return f'JWS.from_json_token("{self.serialize()}")'
724 except Exception: # pylint: disable=broad-except
725 payload = self.objects['payload'].decode('utf-8')
726 return f'JWS(payload={payload})'