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# SPDX-License-Identifier: GPL-2.0-only
2# This file is part of Scapy
3# See https://scapy.net/ for more information
4# Copyright (C) 2008 Arnaud Ebalard <arnaud.ebalard@eads.net>
5# <arno@natisbad.org>
6# 2015, 2016, 2017 Maxence Tury <maxence.tury@ssi.gouv.fr>
7# 2022-2025 Gabriel Potter
9r"""
10High-level methods for PKI objects (X.509 certificates, CRLs, CSR, Keys, CMS).
11Supported keys include RSA, ECDSA and EdDSA.
13The classes below are wrappers for the ASN.1 objects defined in x509.py.
15Example 1: Certificate & Private key
16____________________________________
18For instance, here is what you could do in order to modify the subject public
19key info of a 'cert' and then resign it with whatever 'key'::
21 >>> from scapy.layers.tls.cert import *
22 >>> cert = Cert("cert.der")
23 >>> k = PrivKeyRSA() # generate a private key
24 >>> cert.setSubjectPublicKeyFromPrivateKey(k)
25 >>> cert.resignWith(k)
26 >>> cert.export("newcert.pem")
27 >>> k.export("mykey.pem")
29One could also edit arguments like the serial number, as such::
31 >>> from scapy.layers.tls.cert import *
32 >>> c = Cert("mycert.pem")
33 >>> c.tbsCertificate.serialNumber = 0x4B1D
34 >>> k = PrivKey("mykey.pem") # import an existing private key
35 >>> c.resignWith(k)
36 >>> c.export("newcert.pem")
38To export the public key of a private key::
40 >>> k = PrivKey("mykey.pem")
41 >>> k.pubkey.export("mypubkey.pem")
43Example 2: CertList and CertTree
44________________________________
46Load a .pem file that contains multiple certificates::
48 >>> l = CertList("ca_chain.pem")
49 >>> l.show()
50 0000 [X.509 Cert Subject:/C=FR/OU=Scapy Test PKI/CN=Scapy Test CA...]
51 0001 [X.509 Cert Subject:/C=FR/OU=Scapy Test PKI/CN=Scapy Test Client...]
53Use 'CertTree' to organize the certificates in a tree::
55 >>> tree = CertTree("ca_chain.pem") # or tree = CertTree(l)
56 >>> tree.show()
57 /C=Ulaanbaatar/OU=Scapy Test PKI/CN=Scapy Test CA [Self Signed]
58 /C=FR/OU=Scapy Test PKI/CN=Scapy Test Client [Not Self Signed]
60Example 3: Certificate Signing Request (CSR)
61____________________________________________
63Scapy's :py:class:`~scapy.layers.tls.cert.CSR` class supports both PKCS#10 and CMC
64formats.
66Load and display a CSR::
68 >>> csr = CSR("cert.req")
69 >>> csr
70 [CSR Format: CMC, Subject:/O=TestOrg/CN=TestCN, Verified: True]
71 >>> csr.certReq.show()
72 ###[ PKCS10_CertificationRequest ]###
73 \certificationRequestInfo\
74 |###[ PKCS10_CertificationRequestInfo ]###
75 | version = 0x0 <ASN1_INTEGER[0]
76 | \subject \
77 | |###[ X509_RDN ]###
78 | | \rdn \
79 | | |###[ X509_AttributeTypeAndValue ]###
80 | | | type = <ASN1_OID['organizationName']>
81 | | | value = <ASN1_UTF8_STRING[b'TestOrg']>
82 [...]
84Get its public key and verify its signature::
86 >>> csr.pubkey
87 <scapy.layers.tls.cert.PubKeyRSA at 0x7f3481149310>
88 >>> csr.verifySelf()
89 True
91No need for obnoxious openssl tweaking anymore. :)
92"""
94import base64
95import calendar
96import enum
97import os
98import socket
99import ssl
100import time
101import warnings
103from scapy.config import conf, crypto_validator
104from scapy.compat import Self, plain_str
105from scapy.error import log_runtime, warning
106from scapy.utils import binrepr
107from scapy.pton_ntop import inet_ntop
108from scapy.asn1.asn1 import (
109 ASN1_BIT_STRING,
110 ASN1_NULL,
111 ASN1_OID,
112 ASN1_STRING,
113)
114from scapy.asn1.mib import hash_by_oid
115from scapy.layers.tls.crypto.hash import _tls_hash_algs
116from scapy.packet import Packet
117from scapy.layers.x509 import (
118 CMS_CertificateChoices,
119 CMS_ContentInfo,
120 CMS_EncapsulatedContentInfo,
121 CMS_IssuerAndSerialNumber,
122 CMS_RevocationInfoChoice,
123 CMS_SignedAttrsForSignature,
124 CMS_SignedData,
125 CMS_SignerInfo,
126 CMS_SubjectKeyIdentifier,
127 ECDSAPrivateKey,
128 ECDSAPrivateKey_OpenSSL,
129 ECDSAPublicKey,
130 EdDSAPrivateKey,
131 EdDSAPublicKey,
132 MLDSAPrivateKey,
133 MLDSAPublicKey,
134 PKCS10_CertificationRequest,
135 RSAPrivateKey,
136 RSAPrivateKey_OpenSSL,
137 RSAPublicKey,
138 X509_AlgorithmIdentifier,
139 X509_Attribute,
140 X509_AttributeValue,
141 X509_CRL,
142 X509_Cert,
143 X509_DNSName,
144 X509_IPAddress,
145 X509_OneAsymmetricKey,
146 X509_SubjectPublicKeyInfo,
147)
148from scapy.layers.tls.crypto.hash import _get_hash
149from scapy.layers.tls.crypto.pkcs1 import (
150 _DecryptAndSignRSA,
151 _EncryptAndVerifyRSA,
152 pkcs_os2ip,
153)
154from scapy.compat import bytes_encode
156# Typing imports
157from typing import (
158 List,
159 Optional,
160 Union,
161)
163if conf.crypto_valid:
164 from cryptography.exceptions import InvalidSignature
165 from cryptography.hazmat.backends import default_backend
166 from cryptography.hazmat.primitives import hashes
167 from cryptography.hazmat.primitives import serialization
168 from cryptography.hazmat.primitives.asymmetric import rsa, ec, x25519, x448
170 try:
171 # cryptography >= 47.0
172 from cryptography.hazmat.primitives.asymmetric import mldsa
173 except ImportError:
174 pass
176 # cryptography raised the minimum RSA key length to 1024 in 43.0+
177 # https://github.com/pyca/cryptography/pull/10278
178 # but we need still 512 for EXPORT40 ciphers (yes EXPORT is terrible)
179 # https://datatracker.ietf.org/doc/html/rfc2246#autoid-66
180 # The following detects the change and hacks around it using the backend
182 try:
183 rsa.generate_private_key(public_exponent=65537, key_size=512)
184 _RSA_512_SUPPORTED = True
185 except ValueError:
186 # cryptography > 43.0
187 _RSA_512_SUPPORTED = False
188 from cryptography.hazmat.primitives.asymmetric.rsa import rust_openssl
191# Maximum allowed size in bytes for a certificate file, to avoid
192# loading huge file when importing a cert
193_MAX_KEY_SIZE = 50 * 1024
194_MAX_CERT_SIZE = 50 * 1024
195_MAX_CRL_SIZE = 10 * 1024 * 1024 # some are that big
196_MAX_CSR_SIZE = 50 * 1024
199#####################################################################
200# Some helpers
201#####################################################################
204@conf.commands.register
205def der2pem(der_string, obj="UNKNOWN"):
206 """Convert DER octet string to PEM format (with optional header)"""
207 # Encode a byte string in PEM format. Header advertises <obj> type.
208 pem_string = "-----BEGIN %s-----\n" % obj
209 base64_string = base64.b64encode(der_string).decode()
210 chunks = [base64_string[i : i + 64] for i in range(0, len(base64_string), 64)]
211 pem_string += "\n".join(chunks)
212 pem_string += "\n-----END %s-----\n" % obj
213 return pem_string
216@conf.commands.register
217def pem2der(pem_string):
218 """Convert PEM string to DER format"""
219 # Encode all lines between the first '-----\n' and the 2nd-to-last '-----'.
220 pem_string = pem_string.replace(b"\r", b"")
221 first_idx = pem_string.find(b"-----\n") + 6
222 if pem_string.find(b"-----BEGIN", first_idx) != -1:
223 raise Exception("pem2der() expects only one PEM-encoded object")
224 last_idx = pem_string.rfind(b"-----", 0, pem_string.rfind(b"-----"))
225 base64_string = pem_string[first_idx:last_idx]
226 base64_string.replace(b"\n", b"")
227 der_string = base64.b64decode(base64_string)
228 return der_string
231def split_pem(s):
232 """
233 Split PEM objects. Useful to process concatenated certificates.
234 """
235 pem_strings = []
236 while s != b"":
237 start_idx = s.find(b"-----BEGIN")
238 if start_idx == -1:
239 break
240 end_idx = s.find(b"-----END")
241 if end_idx == -1:
242 raise Exception("Invalid PEM object (missing END tag)")
243 end_idx = s.find(b"\n", end_idx) + 1
244 if end_idx == 0:
245 # There is no final \n
246 end_idx = len(s)
247 pem_strings.append(s[start_idx:end_idx])
248 s = s[end_idx:]
249 return pem_strings
252class _PKIObj(object):
253 def __init__(self, frmt, der):
254 self.frmt = frmt
255 self._der = der
258class _PKIObjMaker(type):
259 def __call__(cls, obj_path, obj_max_size, pem_marker=None):
260 # This enables transparent DER and PEM-encoded data imports.
261 # Note that when importing a PEM file with multiple objects (like ECDSA
262 # private keys output by openssl), it will concatenate every object in
263 # order to create a 'der' attribute. When converting a 'multi' DER file
264 # into a PEM file, though, the PEM attribute will not be valid,
265 # because we do not try to identify the class of each object.
266 error_msg = "Unable to import data"
268 if obj_path is None:
269 raise Exception(error_msg)
270 # Make sure that only a str object can store a path to read a certificate
271 is_file = isinstance(obj_path, str) and os.path.isfile(obj_path)
272 obj_path = bytes_encode(obj_path)
274 if is_file:
275 _size = os.path.getsize(obj_path)
276 if _size > obj_max_size:
277 raise Exception(error_msg)
278 try:
279 with open(obj_path, "rb") as f:
280 _raw = f.read()
281 except Exception:
282 raise Exception(error_msg)
283 else:
284 _raw = obj_path
286 try:
287 if b"-----BEGIN" in _raw:
288 frmt = "PEM"
289 pem = _raw
290 der_list = split_pem(pem)
291 der = b"".join(map(pem2der, der_list))
292 else:
293 frmt = "DER"
294 der = _raw
295 except Exception:
296 raise Exception(error_msg)
298 p = _PKIObj(frmt, der)
299 return p
302#####################################################################
303# PKI objects wrappers
304#####################################################################
306###############
307# Public Keys #
308###############
311class _PubKeyFactory(_PKIObjMaker):
312 """
313 Metaclass for PubKey creation.
314 It casts the appropriate class on the fly, then fills in
315 the appropriate attributes with import_from_asn1pkt() submethod.
316 """
318 def __call__(cls, key_path=None, cryptography_obj=None):
319 # This allows to import cryptography objects directly
320 if cryptography_obj is not None:
321 obj = type.__call__(cls)
322 obj.__class__ = cls
323 obj.frmt = "original"
324 obj.marker = "PUBLIC KEY"
325 obj.pubkey = cryptography_obj
326 return obj
328 if key_path is None:
329 obj = type.__call__(cls)
330 if cls is PubKey:
331 cls = PubKeyRSA
332 obj.__class__ = cls
333 obj.frmt = "original"
334 obj.fill_and_store()
335 return obj
337 # This deals with the rare RSA 'kx export' call.
338 if isinstance(key_path, tuple):
339 obj = type.__call__(cls)
340 obj.__class__ = PubKeyRSA
341 obj.frmt = "tuple"
342 obj.import_from_tuple(key_path)
343 return obj
345 # Now for the usual calls, key_path may be the path to either:
346 # _an X509_SubjectPublicKeyInfo, as processed by openssl;
347 # _an RSAPublicKey;
348 # _an ECDSAPublicKey;
349 # _an EdDSAPublicKey;
350 # _an MLDSAPublicKey.
351 obj = _PKIObjMaker.__call__(cls, key_path, _MAX_KEY_SIZE)
352 try:
353 spki = X509_SubjectPublicKeyInfo(obj._der)
354 pubkey = spki.subjectPublicKey
355 if isinstance(pubkey, RSAPublicKey):
356 obj.__class__ = PubKeyRSA
357 obj.import_from_asn1pkt(pubkey)
358 elif isinstance(pubkey, ECDSAPublicKey):
359 obj.__class__ = PubKeyECDSA
360 obj.import_from_der(obj._der)
361 elif isinstance(pubkey, EdDSAPublicKey):
362 obj.__class__ = PubKeyEdDSA
363 obj.import_from_der(obj._der)
364 elif isinstance(pubkey, MLDSAPublicKey):
365 obj.__class__ = PubKeyMLDSA
366 obj.import_from_der(obj._der)
367 else:
368 raise
369 obj.marker = "PUBLIC KEY"
370 except Exception:
371 try:
372 pubkey = RSAPublicKey(obj._der)
373 obj.__class__ = PubKeyRSA
374 obj.import_from_asn1pkt(pubkey)
375 obj.marker = "RSA PUBLIC KEY"
376 except Exception:
377 # We cannot import an ECDSA public key without curve knowledge
378 if conf.debug_dissector:
379 raise
380 raise Exception("Unable to import public key")
381 return obj
384class PubKey(metaclass=_PubKeyFactory):
385 """
386 Parent class for PubKeyRSA, PubKeyECDSA, PubKeyEdDSA and PubKeyMLDSA.
387 Provides common verifyCert() and export() methods.
388 """
390 def verifyCert(self, cert):
391 """Verifies either a Cert or an X509_Cert."""
392 h = _get_cert_sig_hashname(cert)
393 tbsCert = cert.tbsCertificate
394 sigVal = bytes(cert.signatureValue)
395 return self.verify(bytes(tbsCert), sigVal, h=h, t="pkcs")
397 def verifyCsr(self, csr):
398 """Verifies a CSR."""
399 h = _get_csr_sig_hashname(csr)
400 certReqInfo = csr.certReq.certificationRequestInfo
401 sigVal = bytes(csr.certReq.signature)
402 return self.verify(bytes(certReqInfo), sigVal, h=h, t="pkcs")
404 @property
405 def pem(self):
406 return der2pem(self.der, self.marker)
408 @property
409 def der(self):
410 return self.pubkey.public_bytes(
411 encoding=serialization.Encoding.DER,
412 format=serialization.PublicFormat.SubjectPublicKeyInfo,
413 )
415 def public_numbers(self, *args, **kwargs):
416 return self.pubkey.public_numbers(*args, **kwargs)
418 @property
419 def key_size(self):
420 return self.pubkey.key_size
422 def export(self, filename, fmt=None):
423 """
424 Export public key in 'fmt' format (DER or PEM) to file 'filename'
425 """
426 if fmt is None:
427 if filename.endswith(".pem"):
428 fmt = "PEM"
429 else:
430 fmt = "DER"
431 with open(filename, "wb") as f:
432 if fmt == "DER":
433 return f.write(self.der)
434 elif fmt == "PEM":
435 return f.write(self.pem.encode())
437 @crypto_validator
438 def verify(self, msg, sig, h="sha256", **kwargs):
439 """
440 Verify signed data.
441 """
442 raise NotImplementedError
445class PubKeyRSA(PubKey, _EncryptAndVerifyRSA):
446 """
447 Wrapper for RSA keys based on _EncryptAndVerifyRSA from crypto/pkcs1.py
448 Use the 'key' attribute to access original object.
449 """
451 @crypto_validator
452 def fill_and_store(self, modulus=None, modulusLen=None, pubExp=None):
453 pubExp = pubExp or 65537
454 if not modulus:
455 real_modulusLen = modulusLen or 2048
456 if real_modulusLen < 1024 and not _RSA_512_SUPPORTED:
457 # cryptography > 43.0 compatibility
458 private_key = rust_openssl.rsa.generate_private_key(
459 public_exponent=pubExp,
460 key_size=real_modulusLen,
461 )
462 else:
463 private_key = rsa.generate_private_key(
464 public_exponent=pubExp,
465 key_size=real_modulusLen,
466 backend=default_backend(),
467 )
468 self.pubkey = private_key.public_key()
469 else:
470 real_modulusLen = len(binrepr(modulus))
471 if modulusLen and real_modulusLen != modulusLen:
472 warning("modulus and modulusLen do not match!")
473 pubNum = rsa.RSAPublicNumbers(n=modulus, e=pubExp)
474 self.pubkey = pubNum.public_key(default_backend())
476 self.marker = "PUBLIC KEY"
478 # Lines below are only useful for the legacy part of pkcs1.py
479 pubNum = self.pubkey.public_numbers()
480 self._modulusLen = real_modulusLen
481 self._modulus = pubNum.n
482 self._pubExp = pubNum.e
484 @crypto_validator
485 def import_from_tuple(self, tup):
486 # this is rarely used
487 e, m, mLen = tup
488 if isinstance(m, bytes):
489 m = pkcs_os2ip(m)
490 if isinstance(e, bytes):
491 e = pkcs_os2ip(e)
492 self.fill_and_store(modulus=m, pubExp=e)
494 def import_from_asn1pkt(self, pubkey):
495 modulus = pubkey.modulus.val
496 pubExp = pubkey.publicExponent.val
497 self.fill_and_store(modulus=modulus, pubExp=pubExp)
499 def encrypt(self, msg, t="pkcs", h="sha256", mgf=None, L=None):
500 # no ECDSA encryption support, hence no ECDSA specific keywords here
501 return _EncryptAndVerifyRSA.encrypt(self, msg, t=t, h=h, mgf=mgf, L=L)
503 def verify(self, msg, sig, t="pkcs", h="sha256", mgf=None, L=None):
504 return _EncryptAndVerifyRSA.verify(self, msg, sig, t=t, h=h, mgf=mgf, L=L)
507class PubKeyECDSA(PubKey):
508 """
509 Wrapper for ECDSA keys based on the cryptography library.
510 Use the 'key' attribute to access original object.
511 """
513 @crypto_validator
514 def fill_and_store(self, curve=None):
515 curve = curve or ec.SECP256R1
516 private_key = ec.generate_private_key(curve(), default_backend())
517 self.pubkey = private_key.public_key()
519 @crypto_validator
520 def import_from_der(self, pubkey):
521 # No lib support for explicit curves nor compressed points.
522 self.pubkey = serialization.load_der_public_key(
523 pubkey,
524 backend=default_backend(),
525 )
527 def encrypt(self, msg, h="sha256", **kwargs):
528 raise Exception("No ECDSA encryption support")
530 @crypto_validator
531 def verify(self, msg, sig, h="sha256", **kwargs):
532 # 'sig' should be a DER-encoded signature, as per RFC 3279
533 try:
534 self.pubkey.verify(sig, msg, ec.ECDSA(_get_hash(h)))
535 return True
536 except InvalidSignature:
537 return False
540class PubKeyEdDSA(PubKey):
541 """
542 Wrapper for EdDSA keys based on the cryptography library.
543 Use the 'key' attribute to access original object.
544 """
546 @crypto_validator
547 def fill_and_store(self, curve=None):
548 curve = curve or x25519.X25519PrivateKey
549 private_key = curve.generate()
550 self.pubkey = private_key.public_key()
552 @crypto_validator
553 def import_from_der(self, pubkey):
554 self.pubkey = serialization.load_der_public_key(
555 pubkey,
556 backend=default_backend(),
557 )
559 def encrypt(self, msg, **kwargs):
560 raise Exception("No EdDSA encryption support")
562 @crypto_validator
563 def verify(self, msg, sig, **kwargs):
564 # 'sig' should be a DER-encoded signature, as per RFC 3279
565 try:
566 self.pubkey.verify(sig, msg)
567 return True
568 except InvalidSignature:
569 return False
572class PubKeyMLDSA(PubKey):
573 """
574 Wrapper for MLDSA keys based on the cryptography library.
575 Use the 'key' attribute to access original object.
576 """
578 @crypto_validator
579 def fill_and_store(self, curve=None):
580 curve = curve or mldsa.MLDSA87PrivateKey
581 private_key = curve.generate()
582 self.pubkey = private_key.public_key()
584 @crypto_validator
585 def import_from_der(self, pubkey):
586 self.pubkey = serialization.load_der_public_key(
587 pubkey,
588 backend=default_backend(),
589 )
591 def encrypt(self, msg, **kwargs):
592 raise Exception("No MLDSA encryption support")
594 @crypto_validator
595 def verify(self, msg, sig, **kwargs):
596 # 'sig' should be a DER-encoded signature, as per RFC 3279
597 try:
598 self.pubkey.verify(sig, msg)
599 return True
600 except InvalidSignature:
601 return False
604################
605# Private Keys #
606################
609class _PrivKeyFactory(_PKIObjMaker):
610 """
611 Metaclass for PrivKey creation.
612 It casts the appropriate class on the fly, then fills in
613 the appropriate attributes with import_from_asn1pkt() submethod.
614 """
616 def __call__(cls, key_path=None, cryptography_obj=None):
617 """
618 key_path may be the path to either:
619 _an RSAPrivateKey_OpenSSL (as generated by openssl);
620 _an ECDSAPrivateKey_OpenSSL (as generated by openssl);
621 _an RSAPrivateKey;
622 _an ECDSAPrivateKey.
623 """
624 # This allows to import cryptography objects directly
625 if cryptography_obj is not None:
626 # We (stupidly) need to go through the whole import process because RSA
627 # does more than just importing the cryptography objects...
628 obj = _PKIObj(
629 "DER",
630 cryptography_obj.private_bytes(
631 encoding=serialization.Encoding.DER,
632 format=serialization.PrivateFormat.PKCS8,
633 encryption_algorithm=serialization.NoEncryption(),
634 ),
635 )
636 elif key_path is None:
637 obj = type.__call__(cls)
638 if cls is PrivKey:
639 cls = PrivKeyECDSA
640 obj.__class__ = cls
641 obj.frmt = "original"
642 obj.fill_and_store()
643 return obj
644 else:
645 # Load from file
646 obj = _PKIObjMaker.__call__(cls, key_path, _MAX_KEY_SIZE)
648 try:
649 # The modern format "OneAsymmetricKey" supports specifying the algorithm
650 privkey = X509_OneAsymmetricKey(obj._der)
651 if isinstance(privkey.privateKey, EdDSAPrivateKey):
652 obj.__class__ = PrivKeyEdDSA
653 elif isinstance(privkey.privateKey, MLDSAPrivateKey):
654 obj.__class__ = PrivKeyMLDSA
655 else:
656 raise
657 obj.marker = "PRIVATE KEY"
658 except Exception:
659 # If it fails, we have a chain of legacy fallbacks.
660 try:
661 privkey = RSAPrivateKey_OpenSSL(obj._der)
662 privkey = privkey.privateKey
663 obj.__class__ = PrivKeyRSA
664 obj.marker = "PRIVATE KEY"
665 except Exception:
666 try:
667 privkey = ECDSAPrivateKey_OpenSSL(obj._der)
668 privkey = privkey.privateKey
669 obj.__class__ = PrivKeyECDSA
670 obj.marker = "EC PRIVATE KEY"
671 except Exception:
672 try:
673 privkey = RSAPrivateKey(obj._der)
674 obj.__class__ = PrivKeyRSA
675 obj.marker = "RSA PRIVATE KEY"
676 except Exception:
677 try:
678 privkey = ECDSAPrivateKey(obj._der)
679 obj.__class__ = PrivKeyECDSA
680 obj.marker = "EC PRIVATE KEY"
681 except Exception:
682 raise Exception("Unable to import private key")
683 try:
684 obj.import_from_asn1pkt(privkey)
685 except ImportError:
686 pass
687 return obj
690class _Raw_ASN1_BIT_STRING(ASN1_BIT_STRING):
691 """A ASN1_BIT_STRING that ignores BER encoding"""
693 def __bytes__(self):
694 return self.val_readable
696 __str__ = __bytes__
699class PrivKey(metaclass=_PrivKeyFactory):
700 """
701 Parent class for PrivKeyRSA, PrivKeyECDSA and PrivKeyEdDSA.
702 Provides common signTBSCert(), resignCert(), verifyCert()
703 and export() methods.
704 """
706 def signTBSCert(self, tbsCert, h="sha256"):
707 """
708 Note that this will always copy the signature field from the
709 tbsCertificate into the signatureAlgorithm field of the result,
710 regardless of the coherence between its contents (which might
711 indicate ecdsa-with-SHA512) and the result (e.g. RSA signing MD2).
713 There is a small inheritance trick for the computation of sigVal
714 below: in order to use a sign() method which would apply
715 to both PrivKeyRSA and PrivKeyECDSA, the sign() methods of the
716 subclasses accept any argument, be it from the RSA or ECDSA world,
717 and then they keep the ones they're interested in.
718 Here, t will be passed eventually to pkcs1._DecryptAndSignRSA.sign().
719 """
720 sigAlg = tbsCert.signature
721 h = h or hash_by_oid[sigAlg.algorithm.val]
722 sigVal = self.sign(bytes(tbsCert), h=h, t="pkcs")
723 c = X509_Cert()
724 c.tbsCertificate = tbsCert
725 c.signatureAlgorithm = sigAlg
726 c.signatureValue = _Raw_ASN1_BIT_STRING(sigVal, readable=True)
727 return c
729 def resignCert(self, cert):
730 """Rewrite the signature of either a Cert or an X509_Cert."""
731 return self.signTBSCert(cert.tbsCertificate, h=None)
733 def verifyCert(self, cert):
734 """Verifies either a Cert or an X509_Cert."""
735 return self.pubkey.verifyCert(cert)
737 def verifyCsr(self, cert):
738 """Verifies either a CSR."""
739 return self.pubkey.verifyCsr(cert)
741 @property
742 def pem(self):
743 return der2pem(self.der, self.marker)
745 @property
746 def der(self):
747 return self.key.private_bytes(
748 encoding=serialization.Encoding.DER,
749 format=serialization.PrivateFormat.PKCS8,
750 encryption_algorithm=serialization.NoEncryption(),
751 )
753 def export(self, filename, fmt=None):
754 """
755 Export private key in 'fmt' format (DER or PEM) to file 'filename'
756 """
757 if fmt is None:
758 if filename.endswith(".pem"):
759 fmt = "PEM"
760 else:
761 fmt = "DER"
762 with open(filename, "wb") as f:
763 if fmt == "DER":
764 return f.write(self.der)
765 elif fmt == "PEM":
766 return f.write(self.pem.encode())
768 @crypto_validator
769 def sign(self, data, h="sha256", **kwargs):
770 """
771 Sign data.
772 """
773 raise NotImplementedError
775 @crypto_validator
776 def verify(self, msg, sig, h="sha256", **kwargs):
777 """
778 Verify signed data.
779 """
780 raise NotImplementedError
783class PrivKeyRSA(PrivKey, _DecryptAndSignRSA):
784 """
785 Wrapper for RSA keys based on _DecryptAndSignRSA from crypto/pkcs1.py
786 Use the 'key' attribute to access original object.
787 """
789 @crypto_validator
790 def fill_and_store(
791 self,
792 modulus=None,
793 modulusLen=None,
794 pubExp=None,
795 prime1=None,
796 prime2=None,
797 coefficient=None,
798 exponent1=None,
799 exponent2=None,
800 privExp=None,
801 ):
802 pubExp = pubExp or 65537
803 if None in [
804 modulus,
805 prime1,
806 prime2,
807 coefficient,
808 privExp,
809 exponent1,
810 exponent2,
811 ]:
812 # note that the library requires every parameter
813 # in order to call RSAPrivateNumbers(...)
814 # if one of these is missing, we generate a whole new key
815 real_modulusLen = modulusLen or 2048
816 if real_modulusLen < 1024 and not _RSA_512_SUPPORTED:
817 # cryptography > 43.0 compatibility
818 self.key = rust_openssl.rsa.generate_private_key(
819 public_exponent=pubExp,
820 key_size=real_modulusLen,
821 )
822 else:
823 self.key = rsa.generate_private_key(
824 public_exponent=pubExp,
825 key_size=real_modulusLen,
826 backend=default_backend(),
827 )
828 pubkey = self.key.public_key()
829 else:
830 real_modulusLen = len(binrepr(modulus))
831 if modulusLen and real_modulusLen != modulusLen:
832 warning("modulus and modulusLen do not match!")
833 pubNum = rsa.RSAPublicNumbers(n=modulus, e=pubExp)
834 privNum = rsa.RSAPrivateNumbers(
835 p=prime1,
836 q=prime2,
837 dmp1=exponent1,
838 dmq1=exponent2,
839 iqmp=coefficient,
840 d=privExp,
841 public_numbers=pubNum,
842 )
843 self.key = privNum.private_key(default_backend())
844 pubkey = self.key.public_key()
846 self.marker = "PRIVATE KEY"
848 # Lines below are only useful for the legacy part of pkcs1.py
849 pubNum = pubkey.public_numbers()
850 self._modulusLen = real_modulusLen
851 self._modulus = pubNum.n
852 self._pubExp = pubNum.e
854 self.pubkey = PubKeyRSA((pubNum.e, pubNum.n, real_modulusLen))
856 def import_from_asn1pkt(self, privkey):
857 modulus = privkey.modulus.val
858 pubExp = privkey.publicExponent.val
859 privExp = privkey.privateExponent.val
860 prime1 = privkey.prime1.val
861 prime2 = privkey.prime2.val
862 exponent1 = privkey.exponent1.val
863 exponent2 = privkey.exponent2.val
864 coefficient = privkey.coefficient.val
865 self.fill_and_store(
866 modulus=modulus,
867 pubExp=pubExp,
868 privExp=privExp,
869 prime1=prime1,
870 prime2=prime2,
871 exponent1=exponent1,
872 exponent2=exponent2,
873 coefficient=coefficient,
874 )
876 def verify(self, msg, sig, t="pkcs", h="sha256", mgf=None, L=None):
877 return self.pubkey.verify(
878 msg=msg,
879 sig=sig,
880 t=t,
881 h=h,
882 mgf=mgf,
883 L=L,
884 )
886 def sign(self, data, t="pkcs", h="sha256", mgf=None, L=None):
887 return _DecryptAndSignRSA.sign(self, data, t=t, h=h, mgf=mgf, L=L)
890class PrivKeyECDSA(PrivKey):
891 """
892 Wrapper for ECDSA keys based on SigningKey from ecdsa library.
893 Use the 'key' attribute to access original object.
894 """
896 @crypto_validator
897 def fill_and_store(self, curve=None):
898 curve = curve or ec.SECP256R1
899 self.key = ec.generate_private_key(curve(), default_backend())
900 self.pubkey = PubKeyECDSA(cryptography_obj=self.key.public_key())
901 self.marker = "EC PRIVATE KEY"
903 @crypto_validator
904 def import_from_asn1pkt(self, privkey):
905 self.key = serialization.load_der_private_key(
906 bytes(privkey), None, backend=default_backend()
907 )
908 self.pubkey = PubKeyECDSA(cryptography_obj=self.key.public_key())
909 self.marker = "EC PRIVATE KEY"
911 @crypto_validator
912 def verify(self, msg, sig, h="sha256", **kwargs):
913 return self.pubkey.verify(msg=msg, sig=sig, h=h, **kwargs)
915 @crypto_validator
916 def sign(self, data, h="sha256", **kwargs):
917 return self.key.sign(data, ec.ECDSA(_get_hash(h)))
920class PrivKeyEdDSA(PrivKey):
921 """
922 Wrapper for EdDSA keys
923 Use the 'key' attribute to access original object.
924 """
926 @crypto_validator
927 def fill_and_store(self, curve=None):
928 curve = curve or x25519.X25519PrivateKey
929 self.key = curve.generate()
930 self.pubkey = PubKeyEdDSA(cryptography_obj=self.key.public_key())
931 self.marker = "PRIVATE KEY"
933 @crypto_validator
934 def import_from_asn1pkt(self, privkey):
935 self.key = serialization.load_der_private_key(
936 bytes(privkey), None, backend=default_backend()
937 )
938 self.pubkey = PubKeyEdDSA(cryptography_obj=self.key.public_key())
939 self.marker = "PRIVATE KEY"
941 @crypto_validator
942 def verify(self, msg, sig, **kwargs):
943 return self.pubkey.verify(msg=msg, sig=sig, **kwargs)
945 @crypto_validator
946 def sign(self, data, **kwargs):
947 return self.key.sign(data)
950class PrivKeyMLDSA(PrivKey):
951 """
952 Wrapper for MLDSA keys
953 Use the 'key' attribute to access original object.
954 """
956 @crypto_validator
957 def fill_and_store(self, curve=None):
958 curve = curve or mldsa.MLDSA87PrivateKey
959 self.key = curve.generate()
960 self.pubkey = PubKeyMLDSA(cryptography_obj=self.key.public_key())
961 self.marker = "PRIVATE KEY"
963 @crypto_validator
964 def import_from_asn1pkt(self, privkey):
965 self.key = serialization.load_der_private_key(
966 bytes(privkey), None, backend=default_backend()
967 )
968 self.pubkey = PubKeyMLDSA(cryptography_obj=self.key.public_key())
969 self.marker = "PRIVATE KEY"
971 @crypto_validator
972 def verify(self, msg, sig, **kwargs):
973 return self.pubkey.verify(msg=msg, sig=sig, **kwargs)
975 @crypto_validator
976 def sign(self, data, **kwargs):
977 return self.key.sign(data)
980################
981# Certificates #
982################
985class _CertMaker(_PKIObjMaker):
986 """
987 Metaclass for Cert creation. It is not necessary as it was for the keys,
988 but we reuse the model instead of creating redundant constructors.
989 """
991 def __call__(cls, cert_path=None, cryptography_obj=None):
992 # This allows to import cryptography objects directly
993 if cryptography_obj is not None:
994 obj = _PKIObj(
995 "DER",
996 cryptography_obj.public_bytes(
997 encoding=serialization.Encoding.DER,
998 ),
999 )
1000 else:
1001 # Load from file
1002 obj = _PKIObjMaker.__call__(cls, cert_path, _MAX_CERT_SIZE, "CERTIFICATE")
1003 obj.__class__ = Cert
1004 obj.marker = "CERTIFICATE"
1005 try:
1006 cert = X509_Cert(obj._der)
1007 except Exception:
1008 if conf.debug_dissector:
1009 raise
1010 raise Exception("Unable to import certificate")
1011 cert.remove_payload()
1012 obj.import_from_asn1pkt(cert)
1013 return obj
1016def _get_cert_sig_hashname(cert):
1017 """
1018 Return the hash associated with the signature algorithm of a certificate.
1019 """
1020 tbsCert = cert.tbsCertificate
1021 sigAlg = tbsCert.signature
1022 return hash_by_oid[sigAlg.algorithm.val]
1025def _get_csr_sig_hashname(csr):
1026 """
1027 Return the hash associated with the signature algorithm of a CSR.
1028 """
1029 certReq = csr.certReq
1030 sigAlg = certReq.signatureAlgorithm
1031 return hash_by_oid[sigAlg.algorithm.val]
1034def _match_dns_name(pattern, hostname):
1035 """
1036 Whether a certificate DNS name matches a hostname, RFC 6125 sect 6.4.3.
1038 :param pattern: a dNSName from the certificate
1039 :param hostname: the name the client asked for
1040 :return: True if they match
1041 """
1042 pattern = pattern.lower().rstrip(".")
1043 hostname = hostname.lower().rstrip(".")
1044 if pattern[:2] == "*.":
1045 return pattern[2:] == hostname.split(".", 1)[-1]
1046 else:
1047 return pattern == hostname
1050class Cert(metaclass=_CertMaker):
1051 """
1052 Wrapper for the X509_Cert from layers/x509.py.
1053 Use the 'x509Cert' attribute to access original object.
1054 """
1056 def import_from_asn1pkt(self, cert):
1057 error_msg = "Unable to import certificate"
1059 self.x509Cert = cert
1061 tbsCert = cert.tbsCertificate
1063 if tbsCert.version:
1064 self.version = tbsCert.version.val + 1
1065 else:
1066 self.version = 1
1067 self.serial = tbsCert.serialNumber.val
1068 self.sigAlg = tbsCert.signature.algorithm.oidname
1069 self.issuer = tbsCert.get_issuer()
1070 self.issuer_str = tbsCert.get_issuer_str()
1071 self.issuer_hash = hash(self.issuer_str)
1072 self.subject = tbsCert.get_subject()
1073 self.subject_str = tbsCert.get_subject_str()
1074 self.subject_hash = hash(self.subject_str)
1075 self.authorityKeyID = None
1077 self.notBefore_str = tbsCert.validity.not_before.pretty_time
1078 try:
1079 self.notBefore = tbsCert.validity.not_before.datetime.timetuple()
1080 except ValueError:
1081 raise Exception(error_msg)
1082 self.notBefore_str_simple = time.strftime("%x", self.notBefore)
1084 self.notAfter_str = tbsCert.validity.not_after.pretty_time
1085 try:
1086 self.notAfter = tbsCert.validity.not_after.datetime.timetuple()
1087 except ValueError:
1088 raise Exception(error_msg)
1089 self.notAfter_str_simple = time.strftime("%x", self.notAfter)
1091 self.pubkey = PubKey(bytes(tbsCert.subjectPublicKeyInfo))
1092 self.subjectAltName = None
1094 if tbsCert.extensions:
1095 for extn in tbsCert.extensions:
1096 if extn.extnID.oidname == "basicConstraints":
1097 self.cA = False
1098 if extn.extnValue.cA:
1099 self.cA = not (extn.extnValue.cA.val == 0)
1100 elif extn.extnID.oidname == "keyUsage":
1101 self.keyUsage = extn.extnValue.get_keyUsage()
1102 elif extn.extnID.oidname == "extKeyUsage":
1103 self.extKeyUsage = extn.extnValue.get_extendedKeyUsage()
1104 elif extn.extnID.oidname == "authorityKeyIdentifier":
1105 self.authorityKeyID = extn.extnValue.keyIdentifier.val
1106 elif extn.extnID.oidname == "subjectAltName":
1107 self.subjectAltName = extn.extnValue.subjectAltName
1109 self.signatureValue = bytes(cert.signatureValue)
1110 self.signatureLen = len(self.signatureValue)
1112 def isIssuer(self, other):
1113 """
1114 True if 'other' issued 'self', i.e.:
1115 - self.issuer == other.subject
1116 - self is signed by other
1117 """
1118 if self.issuer_hash != other.subject_hash:
1119 return False
1120 return other.pubkey.verifyCert(self)
1122 def isIssuerCert(self, other):
1123 return self.isIssuer(other)
1125 def isSelfSigned(self):
1126 """
1127 Return True if the certificate is self-signed:
1128 - issuer and subject are the same
1129 - the signature of the certificate is valid.
1130 """
1131 if self.issuer_hash == self.subject_hash:
1132 return self.isIssuer(self)
1133 return False
1135 def encrypt(self, msg, t="pkcs", h="sha256", mgf=None, L=None):
1136 # no ECDSA *encryption* support, hence only RSA specific keywords here
1137 return self.pubkey.encrypt(msg, t=t, h=h, mgf=mgf, L=L)
1139 def verify(self, msg, sig, t="pkcs", h="sha256", mgf=None, L=None):
1140 return self.pubkey.verify(msg, sig, t=t, h=h, mgf=mgf, L=L)
1142 def getCertSignatureHash(self):
1143 """
1144 Return the hash cryptography object used by the 'signatureAlgorithm'
1145 """
1146 return _get_hash(_get_cert_sig_hashname(self))
1148 def setSubjectPublicKeyFromPrivateKey(self, key):
1149 """
1150 Replace the subjectPublicKeyInfo of this certificate with the one from
1151 the provided key.
1152 """
1153 if isinstance(key, (PubKey, PrivKey)):
1154 if isinstance(key, PrivKey):
1155 pubkey = key.pubkey
1156 else:
1157 pubkey = key
1158 self.tbsCertificate.subjectPublicKeyInfo = X509_SubjectPublicKeyInfo(
1159 pubkey.der
1160 )
1161 else:
1162 raise ValueError("Unknown type 'key', should be PubKey or PrivKey")
1164 def resignWith(self, key):
1165 """
1166 Resign a certificate with a specific key
1167 """
1168 self.import_from_asn1pkt(key.resignCert(self))
1170 def remainingDays(self, now=None):
1171 """
1172 Based on the value of notAfter field, returns the number of
1173 days the certificate will still be valid. The date used for the
1174 comparison is the current and local date, as returned by
1175 time.localtime(), except if 'now' argument is provided another
1176 one. 'now' argument can be given as either a time tuple or a string
1177 representing the date. Accepted format for the string version
1178 are:
1180 - '%b %d %H:%M:%S %Y %Z' e.g. 'Jan 30 07:38:59 2008 GMT'
1181 - '%m/%d/%y' e.g. '01/30/08' (less precise)
1183 If the certificate is no more valid at the date considered, then
1184 a negative value is returned representing the number of days
1185 since it has expired.
1187 The number of days is returned as a float to deal with the unlikely
1188 case of certificates that are still just valid.
1189 """
1190 if now is None:
1191 now = time.localtime()
1192 elif isinstance(now, str):
1193 try:
1194 if "/" in now:
1195 now = time.strptime(now, "%m/%d/%y")
1196 else:
1197 now = time.strptime(now, "%b %d %H:%M:%S %Y %Z")
1198 except Exception:
1199 warning("Bad time string provided, will use localtime() instead.")
1200 now = time.localtime()
1202 now = time.mktime(now)
1203 nft = time.mktime(self.notAfter)
1204 diff = (nft - now) / (24.0 * 3600)
1205 return diff
1207 def isValidAt(self, now=None):
1208 """
1209 Whether the current time falls inside the certificate's validity period.
1211 The comparison is made in UTC, which is how notBefore and notAfter are
1212 stored. (:func:`remainingDays` compares in local time and so is off by
1213 the local UTC offset.)
1215 :param now: (optional) a UTC time tuple to compare against, defaulting
1216 to the current time
1217 :return: True if the certificate is neither expired nor not yet valid
1218 """
1219 if now is None:
1220 now = time.gmtime()
1221 now = calendar.timegm(now)
1222 return (
1223 calendar.timegm(self.notBefore) <= now <=
1224 calendar.timegm(self.notAfter)
1225 )
1227 def matchesHostname(self, hostname):
1228 """
1229 Whether this certificate was issued to the given host.
1231 Names come from the subjectAltName extension. RFC 6125 sect 6.4.4 says
1232 the Common Name is only consulted when there is no subjectAltName at
1233 all, and that is what happens here.
1235 :param hostname: the DNS name or IP address the client asked for
1236 :return: True if the certificate names that host
1237 """
1238 if not hostname:
1239 return False
1240 hostname = plain_str(hostname)
1242 if self.subjectAltName is not None:
1243 # Check subjectAltName
1244 for generalName in self.subjectAltName:
1245 name = generalName.generalName
1246 if isinstance(name, X509_DNSName):
1247 if _match_dns_name(plain_str(name.dNSName.val), hostname):
1248 return True
1249 elif isinstance(name, X509_IPAddress):
1250 raw = name.iPAddress.val
1251 if len(raw) == 4:
1252 if inet_ntop(socket.AF_INET, raw) == hostname:
1253 return True
1254 elif len(raw) == 16:
1255 if inet_ntop(socket.AF_INET6, raw) == hostname:
1256 return True
1257 return False
1258 else:
1259 # Check Common Name
1260 if "commonName" in self.subject:
1261 return _match_dns_name(self.subject["commonName"], hostname)
1262 return False
1264 def isRevoked(self, crl_list):
1265 """
1266 Given a list of trusted CRL (their signature has already been
1267 verified with trusted anchors), this function returns True if
1268 the certificate is marked as revoked by one of those CRL.
1270 Note that if the Certificate was on hold in a previous CRL and
1271 is now valid again in a new CRL and bot are in the list, it
1272 will be considered revoked: this is because _all_ CRLs are
1273 checked (not only the freshest) and revocation status is not
1274 handled.
1276 Also note that the check on the issuer is performed on the
1277 Authority Key Identifier if available in _both_ the CRL and the
1278 Cert. Otherwise, the issuers are simply compared.
1279 """
1280 for c in crl_list:
1281 if (
1282 self.authorityKeyID is not None
1283 and c.authorityKeyID is not None
1284 and self.authorityKeyID == c.authorityKeyID
1285 ):
1286 return self.serial in (x[0] for x in c.revoked_cert_serials)
1287 elif self.issuer == c.issuer:
1288 return self.serial in (x[0] for x in c.revoked_cert_serials)
1289 return False
1291 @property
1292 def tbsCertificate(self):
1293 return self.x509Cert.tbsCertificate
1295 @property
1296 def pem(self):
1297 return der2pem(self.der, self.marker)
1299 @property
1300 def der(self):
1301 return bytes(self.x509Cert)
1303 @property
1304 def pubKey(self):
1305 warnings.warn(
1306 "Cert.pubKey is deprecated and will be removed in a future version. "
1307 "Use Cert.pubkey",
1308 DeprecationWarning,
1309 )
1310 return self.pubkey
1312 @property
1313 def extensions(self):
1314 return self.tbsCertificate.extensions
1316 def __eq__(self, other):
1317 return self.der == other.der
1319 def __hash__(self):
1320 return hash(self.der)
1322 def export(self, filename, fmt=None):
1323 """
1324 Export certificate in 'fmt' format (DER or PEM) to file 'filename'
1325 """
1326 if fmt is None:
1327 if filename.endswith(".pem"):
1328 fmt = "PEM"
1329 else:
1330 fmt = "DER"
1331 with open(filename, "wb") as f:
1332 if fmt == "DER":
1333 return f.write(self.der)
1334 elif fmt == "PEM":
1335 return f.write(self.pem.encode())
1337 def show(self):
1338 print("Serial: %s" % self.serial)
1339 print("Issuer: " + self.issuer_str)
1340 print("Subject: " + self.subject_str)
1341 print("Validity: %s to %s" % (self.notBefore_str, self.notAfter_str))
1343 def __repr__(self):
1344 return "[X.509 Cert. Subject:%s, Issuer:%s]" % (
1345 self.subject_str,
1346 self.issuer_str,
1347 )
1350################################
1351# Certificate Revocation Lists #
1352################################
1355class _CRLMaker(_PKIObjMaker):
1356 """
1357 Metaclass for CRL creation. It is not necessary as it was for the keys,
1358 but we reuse the model instead of creating redundant constructors.
1359 """
1361 def __call__(cls, cert_path):
1362 obj = _PKIObjMaker.__call__(cls, cert_path, _MAX_CRL_SIZE, "X509 CRL")
1363 obj.__class__ = CRL
1364 try:
1365 crl = X509_CRL(obj._der)
1366 except Exception:
1367 raise Exception("Unable to import CRL")
1368 obj.import_from_asn1pkt(crl)
1369 return obj
1372class CRL(metaclass=_CRLMaker):
1373 """
1374 Wrapper for the X509_CRL from layers/x509.py.
1375 Use the 'x509CRL' attribute to access original object.
1376 """
1378 def import_from_asn1pkt(self, crl):
1379 error_msg = "Unable to import CRL"
1381 self.x509CRL = crl
1383 tbsCertList = crl.tbsCertList
1384 self.tbsCertList = bytes(tbsCertList)
1386 if tbsCertList.version:
1387 self.version = tbsCertList.version.val + 1
1388 else:
1389 self.version = 1
1390 self.sigAlg = tbsCertList.signature.algorithm.oidname
1391 self.issuer = tbsCertList.get_issuer()
1392 self.issuer_str = tbsCertList.get_issuer_str()
1393 self.issuer_hash = hash(self.issuer_str)
1395 self.lastUpdate_str = tbsCertList.this_update.pretty_time
1396 lastUpdate = tbsCertList.this_update.val
1397 if lastUpdate[-1] == "Z":
1398 lastUpdate = lastUpdate[:-1]
1399 try:
1400 self.lastUpdate = time.strptime(lastUpdate, "%y%m%d%H%M%S")
1401 except Exception:
1402 raise Exception(error_msg)
1403 self.lastUpdate_str_simple = time.strftime("%x", self.lastUpdate)
1405 self.nextUpdate = None
1406 self.nextUpdate_str_simple = None
1407 if tbsCertList.next_update:
1408 self.nextUpdate_str = tbsCertList.next_update.pretty_time
1409 nextUpdate = tbsCertList.next_update.val
1410 if nextUpdate[-1] == "Z":
1411 nextUpdate = nextUpdate[:-1]
1412 try:
1413 self.nextUpdate = time.strptime(nextUpdate, "%y%m%d%H%M%S")
1414 except Exception:
1415 raise Exception(error_msg)
1416 self.nextUpdate_str_simple = time.strftime("%x", self.nextUpdate)
1418 if tbsCertList.crlExtensions:
1419 for extension in tbsCertList.crlExtensions:
1420 if extension.extnID.oidname == "cRLNumber":
1421 self.number = extension.extnValue.cRLNumber.val
1423 revoked = []
1424 if tbsCertList.revokedCertificates:
1425 for cert in tbsCertList.revokedCertificates:
1426 serial = cert.serialNumber.val
1427 date = cert.revocationDate.val
1428 if date[-1] == "Z":
1429 date = date[:-1]
1430 try:
1431 time.strptime(date, "%y%m%d%H%M%S")
1432 except Exception:
1433 raise Exception(error_msg)
1434 revoked.append((serial, date))
1435 self.revoked_cert_serials = revoked
1437 self.signatureValue = bytes(crl.signatureValue)
1438 self.signatureLen = len(self.signatureValue)
1440 def isIssuer(self, other):
1441 # This is exactly the same thing as in Cert method.
1442 if self.issuer_hash != other.subject_hash:
1443 return False
1444 return other.pubkey.verifyCert(self)
1446 def verify(self, anchors):
1447 # Return True iff the CRL is signed by one of the provided anchors.
1448 return any(self.isIssuer(a) for a in anchors)
1450 def show(self):
1451 print("Version: %d" % self.version)
1452 print("sigAlg: " + self.sigAlg)
1453 print("Issuer: " + self.issuer_str)
1454 print("lastUpdate: %s" % self.lastUpdate_str)
1455 print("nextUpdate: %s" % self.nextUpdate_str)
1458###############################
1459# Certificate Signing Request #
1460###############################
1463class _CSRMaker(_PKIObjMaker):
1464 """
1465 Metaclass for CSR creation. It is not necessary as it was for the keys,
1466 but we reuse the model instead of creating redundant constructors.
1467 """
1469 def __call__(cls, cert_path):
1470 obj = _PKIObjMaker.__call__(cls, cert_path, _MAX_CSR_SIZE)
1471 obj.__class__ = CSR
1472 try:
1473 # PKCS#10 format
1474 csr = PKCS10_CertificationRequest(obj._der)
1475 obj.marker = "NEW CERTIFICATE REQUEST"
1476 obj.fmt = CSR.FORMAT.PKCS10
1477 except Exception:
1478 try:
1479 # CMC format
1480 csr = CMS_ContentInfo(obj._der)
1481 obj.marker = "NEW CERTIFICATE REQUEST"
1482 obj.fmt = CSR.FORMAT.CMC
1483 except Exception:
1484 raise Exception("Unable to import CSR")
1486 obj.import_from_asn1pkt(csr)
1487 return obj
1490class CSR(metaclass=_CSRMaker):
1491 """
1492 Wrapper for the CSR formats.
1493 This can handle both PKCS#10 and CMC formats.
1494 """
1496 class FORMAT(enum.Enum):
1497 """
1498 The format used by the CSR.
1499 """
1501 PKCS10 = "PKCS#10"
1502 CMC = "CMC"
1504 def import_from_asn1pkt(self, csr):
1505 self.csr = csr
1506 certReqInfo = self.certReq.certificationRequestInfo
1508 # Subject
1509 self.subject = certReqInfo.get_subject()
1510 self.subject_str = certReqInfo.get_subject_str()
1511 self.subject_hash = hash(self.subject_str)
1513 # pubkey
1514 self.pubkey = PubKey(bytes(certReqInfo.subjectPublicKeyInfo))
1516 # Get the "subjectKeyIdentifier" from the "extensionRequest" attribute
1517 try:
1518 extReq = next(
1519 x.values[0].value
1520 for x in certReqInfo.attributes
1521 if x.type.val == "1.2.840.113549.1.9.14" # extKeyUsage
1522 )
1523 self.sid = next(
1524 x.extnValue.keyIdentifier
1525 for x in extReq.extensions
1526 if x.extnID.val == "2.5.29.14" # subjectKeyIdentifier
1527 )
1528 except StopIteration:
1529 self.sid = None
1531 @property
1532 def certReq(self):
1533 csr = self.csr
1535 if self.fmt == CSR.FORMAT.PKCS10:
1536 return csr
1537 elif self.fmt == CSR.FORMAT.CMC:
1538 if (
1539 csr.contentType.oidname != "id-signedData"
1540 or csr.content.encapContentInfo.eContentType.oidname != "id-cct-PKIData"
1541 ):
1542 raise ValueError("Invalid CMC wrapping !")
1543 req = csr.content.encapContentInfo.eContent.reqSequence[0]
1544 return req.request.certificationRequest
1545 else:
1546 raise ValueError("Invalid CSR format !")
1548 @property
1549 def pem(self):
1550 return der2pem(self.der, self.marker)
1552 @property
1553 def der(self):
1554 return bytes(self.csr)
1556 def __eq__(self, other):
1557 return self.der == other.der
1559 def __hash__(self):
1560 return hash(self.der)
1562 def isIssuer(self, other):
1563 return other.sid == self.sid
1565 def isSelfSigned(self):
1566 return True
1568 def verify(self, msg, sig, t="pkcs", h="sha256", mgf=None, L=None):
1569 return self.pubkey.verify(msg, sig, t=t, h=h, mgf=mgf, L=L)
1571 def export(self, filename, fmt=None):
1572 """
1573 Export certificate in 'fmt' format (DER or PEM) to file 'filename'
1574 """
1575 if fmt is None:
1576 if filename.endswith(".pem"):
1577 fmt = "PEM"
1578 else:
1579 fmt = "DER"
1580 with open(filename, "wb") as f:
1581 if fmt == "DER":
1582 return f.write(self.der)
1583 elif fmt == "PEM":
1584 return f.write(self.pem.encode())
1586 def show(self):
1587 certReqInfo = self.certReq.certificationRequestInfo
1589 print("Subject: " + self.subject_str)
1590 print("Attributes:")
1591 for attr in certReqInfo.attributes:
1592 print(" - %s" % attr.type.oidname)
1594 def verifySelf(self) -> bool:
1595 """
1596 Verify the signatures of the CSR
1597 """
1598 if self.fmt == self.FORMAT.CMC:
1599 try:
1600 cms_engine = CMS_Engine([self])
1601 cms_engine.verify(self.csr)
1602 return self.pubkey.verifyCsr(self)
1603 except ValueError:
1604 return False
1605 elif self.fmt == self.FORMAT.PKCS10:
1606 return self.pubkey.verifyCsr(self)
1607 else:
1608 return False
1610 def __repr__(self):
1611 return "[CSR Format: %s, Subject:%s, Verified: %s]" % (
1612 self.fmt.value,
1613 self.subject_str,
1614 self.verifySelf(),
1615 )
1618####################
1619# Certificate list #
1620####################
1623class CertList(list):
1624 """
1625 An object that can store a list of Cert objects, load them and export them
1626 into DER/PEM format.
1627 """
1629 def __init__(
1630 self,
1631 certList: Union[Self, List[Cert], List[CSR], Cert, str],
1632 ):
1633 """
1634 Construct a list of certificates/CRLs to be used as list of ROOT certificates.
1635 """
1636 # Parse the certificate list / CA
1637 if isinstance(certList, str):
1638 # It's a path. First get the _PKIObj
1639 obj = _PKIObjMaker.__call__(
1640 CertList, certList, _MAX_CERT_SIZE, "CERTIFICATE"
1641 )
1643 # Then parse the der until there's nothing left
1644 certList = []
1645 payload = obj._der
1646 while payload:
1647 cert = X509_Cert(payload)
1648 if conf.raw_layer in cert.payload:
1649 payload = cert.payload.load
1650 else:
1651 payload = None
1652 cert.remove_payload()
1653 certList.append(Cert(cert))
1655 self.frmt = obj.frmt
1656 elif isinstance(certList, Cert):
1657 certList = [certList]
1658 self.frmt = "PEM"
1659 else:
1660 self.frmt = "PEM"
1662 super(CertList, self).__init__(certList)
1664 @classmethod
1665 def load_system_store(cls):
1666 """
1667 Return a CertList containing the default trusted system store.
1668 """
1669 context = ssl.create_default_context()
1670 certs = []
1671 for der in context.get_ca_certs(binary_form=True):
1672 try:
1673 certs.append(Cert(der))
1674 except Exception as ex:
1675 log_runtime.error("Failed loading cert.")
1676 log_runtime.error(der2pem(der))
1677 raise ex
1678 return cls(certs)
1680 def findCertBySid(self, sid):
1681 """
1682 Find a certificate in the list by SubjectIDentifier.
1683 """
1684 for cert in self:
1685 if isinstance(cert, Cert) and isinstance(sid, CMS_IssuerAndSerialNumber):
1686 if cert.issuer == sid.get_issuer():
1687 return cert
1688 elif isinstance(cert, CSR) and isinstance(sid, CMS_SubjectKeyIdentifier):
1689 if cert.sid == sid.sid:
1690 return cert
1691 raise KeyError("Certificate not found !")
1693 def export(self, filename, fmt=None):
1694 """
1695 Export a list of certificates 'fmt' format (DER or PEM) to file 'filename'
1696 """
1697 if fmt is None:
1698 if filename.endswith(".pem"):
1699 fmt = "PEM"
1700 else:
1701 fmt = "DER"
1702 with open(filename, "wb") as f:
1703 if fmt == "DER":
1704 return f.write(self.der)
1705 elif fmt == "PEM":
1706 return f.write(self.pem.encode())
1708 @property
1709 def der(self):
1710 return b"".join(x.der for x in self)
1712 @property
1713 def pem(self):
1714 return "".join(x.pem for x in self)
1716 def __repr__(self):
1717 return "<CertList %s certificates>" % (len(self),)
1719 def show(self):
1720 for i, c in enumerate(self):
1721 print(conf.color_theme.id(i, fmt="%04i"), end=" ")
1722 print(repr(c))
1725######################
1726# Certificate chains #
1727######################
1730class CertTree(CertList):
1731 """
1732 An extension to CertList that additionally has a list of ROOT CAs
1733 that are trusted.
1735 Example::
1737 >>> tree = CertTree("ca_chain.pem")
1738 >>> tree.show()
1739 /CN=DOMAIN-DC1-CA/dc=DOMAIN [Self Signed]
1740 /CN=Administrator/dc=DOMAIN [Not Self Signed]
1741 """
1743 __slots__ = ["frmt", "rootCAs"]
1745 def __init__(
1746 self,
1747 certList: Union[List[Cert], CertList, str],
1748 rootCAs: Union[List[Cert], CertList, Cert, str, None] = None,
1749 ):
1750 """
1751 Construct a chain of certificates that follows issuer/subject matching and
1752 respects signature validity.
1754 Note that we do not check AKID/{SKID/issuer/serial} matching,
1755 nor the presence of keyCertSign in keyUsage extension (if present).
1757 :param certList: a list of Cert/CRL objects (or path to PEM/DER file containing
1758 multiple certs/CRL) to try to chain.
1759 :param rootCAs: (optional) a list of certificates to trust. If not provided,
1760 trusts any self-signed certificates from the certList.
1761 """
1762 # Parse the certificate list
1763 certList = CertList(certList)
1765 # Find the ROOT CAs if store isn't specified
1766 if not rootCAs:
1767 # Build cert store.
1768 self.rootCAs = CertList([x for x in certList if x.isSelfSigned()])
1769 # And remove those certs from the list
1770 for cert in self.rootCAs:
1771 certList.remove(cert)
1772 else:
1773 # Store cert store.
1774 self.rootCAs = CertList(rootCAs)
1775 # And remove those certs from the list if present (remove dups)
1776 for cert in self.rootCAs:
1777 if cert in certList:
1778 certList.remove(cert)
1780 # Append our root CAs to the certList
1781 certList.extend(self.rootCAs)
1783 # Super instantiate
1784 super(CertTree, self).__init__(certList)
1786 @property
1787 def tree(self):
1788 """
1789 Get a tree-like object of the certificate list
1790 """
1791 # We store the tree object as a dictionary that contains children.
1792 tree = [(x, []) for x in self.rootCAs]
1794 # We'll empty this list eventually
1795 certList = list(self)
1797 # We make a list of certificates we have to search children for, and iterate
1798 # through it until it's empty.
1799 todo = list(tree)
1801 # Iterate
1802 while todo:
1803 cert, children = todo.pop()
1804 for c in certList:
1805 # Check if this certificate matches the one we're looking at
1806 if c.isIssuer(cert) and c != cert:
1807 item = (c, [])
1808 children.append(item)
1809 certList.remove(c)
1810 todo.append(item)
1812 return tree
1814 def getchain(self, cert):
1815 """
1816 Return a chain of certificate that points from a ROOT CA to a certificate.
1817 """
1819 def _rec_getchain(chain, curtree):
1820 # See if an element of the current tree signs the cert, if so add it to
1821 # the chain, else recurse.
1822 for c, subtree in curtree:
1823 curchain = chain + [c]
1824 # If 'cert' is issued by c
1825 if cert.isIssuer(c) or c == cert:
1826 # Final node of the chain !
1827 # (add the final cert if not self signed)
1828 if c != cert:
1829 curchain += [cert]
1830 return curchain
1831 else:
1832 # Not the final node of the chain ! Recurse.
1833 curchain = _rec_getchain(curchain, subtree)
1834 if curchain:
1835 return curchain
1836 return None
1838 chain = _rec_getchain([], self.tree)
1839 if chain is not None:
1840 # We add the first certificate to the ROOT in all cases
1841 return CertTree(chain, [chain[0]])
1842 else:
1843 return None
1845 def verify(self, cert, hostname=None, now=None, allow_expired=False):
1846 """
1847 Verify that a certificate is properly signed, current, and the right one.
1849 Raises ValueError when the certificate fails any of the checks.
1851 :param cert: the certificate to verify
1852 :param hostname: (optional) the DNS name or IP address the peer was
1853 expected to be. Without it the identity of the peer is not checked,
1854 so any certificate the store can chain is accepted.
1855 :param now: (optional) a UTC time tuple to check validity against,
1856 defaulting to the current time
1857 :param allow_expired: (optional) accept a chain that is outside its
1858 validity period. Useful when checking a signature made while the
1859 certificate was still valid, or against a stored capture, where
1860 the dates say nothing about whether the signature was genuine.
1861 """
1862 # Check that we can find a chain to this certificate
1863 chain = self.getchain(cert)
1864 if not chain:
1865 raise ValueError("Certificate verification failed !")
1866 # Nothing in the chain may have expired or be in the future: an issuer
1867 # that is out of date does not vouch for anything below it. A chain can
1868 # also hold a CSR, which has no validity period to check.
1869 for c in chain:
1870 if not isinstance(c, Cert):
1871 continue
1872 if not allow_expired and not c.isValidAt(now):
1873 raise ValueError(
1874 "Certificate %s is outside its validity period "
1875 "(%s to %s) !" % (
1876 c.subject_str, c.notBefore_str, c.notAfter_str
1877 )
1878 )
1879 if hostname is not None and not cert.matchesHostname(hostname):
1880 raise ValueError(
1881 "Certificate %s was not issued to %s !" % (
1882 cert.subject_str, plain_str(hostname)
1883 )
1884 )
1886 def show(self, ret: bool = False):
1887 """
1888 Return the CertTree as a string certificate tree
1889 """
1891 def _rec_show(c, children, lvl=0):
1892 s = ""
1893 # Process the current CA
1894 if c:
1895 if not c.isSelfSigned():
1896 s += "%s [Not Self Signed]\n" % c.subject_str
1897 else:
1898 s += "%s [Self Signed]\n" % c.subject_str
1899 s = lvl * " " + s
1900 lvl += 1
1901 # Process all sub-CAs at a lower level
1902 for child, subchildren in children:
1903 s += _rec_show(child, subchildren, lvl=lvl)
1904 return s
1906 showed = _rec_show(None, self.tree)
1907 if ret:
1908 return showed
1909 else:
1910 print(showed)
1912 def __repr__(self):
1913 return "<CertTree %s certificates (%s ROOT CA)>" % (
1914 len(self),
1915 len(self.rootCAs),
1916 )
1919#######
1920# CMS #
1921#######
1923# RFC3852
1926class CMS_Engine:
1927 """
1928 A utility class to perform CMS/PKCS7 operations, as specified by RFC3852.
1930 :param store: a ROOT CA certificate list to trust.
1931 :param crls: a list of CRLs to include. This is currently not checked.
1932 """
1934 def __init__(
1935 self,
1936 store: CertList,
1937 crls: List[X509_CRL] = [],
1938 ):
1939 self.store = store
1940 self.crls = crls
1942 def _get_algorithms(self, key: PrivKey, h="sha256") -> ASN1_OID:
1943 """
1944 Get the algorithms matching a private key
1945 """
1946 if isinstance(key, PrivKeyRSA):
1947 # RFC3370 sect 3.2
1948 return (
1949 ASN1_OID("rsaEncryption"),
1950 _get_hash(h),
1951 h,
1952 )
1953 elif isinstance(key, PrivKeyECDSA):
1954 # RFC5753 sect 2.1.1
1955 if h == "sha1":
1956 return (
1957 ASN1_OID("ecdsa-with-SHA1"),
1958 hashes.SHA1(),
1959 "sha1",
1960 )
1961 elif h == "sha224":
1962 return (
1963 ASN1_OID("ecdsa-with-SHA224"),
1964 hashes.SHA224(),
1965 "sha224",
1966 )
1967 elif h == "sha256":
1968 return (
1969 ASN1_OID("ecdsa-with-SHA256"),
1970 hashes.SHA256(),
1971 "sha256",
1972 )
1973 elif h == "sha384":
1974 return (
1975 ASN1_OID("ecdsa-with-SHA384"),
1976 hashes.SHA384(),
1977 "sha384",
1978 )
1979 elif h == "sha512":
1980 return (
1981 ASN1_OID("ecdsa-with-SHA512"),
1982 hashes.SHA512(),
1983 "sha512",
1984 )
1985 else:
1986 raise ValueError("Unknown hash for private key !")
1987 elif isinstance(key, PrivKeyEdDSA):
1988 # RFC8419 sect 2.3
1989 if isinstance(key.key, x25519.X25519PrivateKey):
1990 return (
1991 ASN1_OID("Ed25519"),
1992 hashes.SHA512(),
1993 "sha512",
1994 )
1995 elif isinstance(key.key, x448.X448PrivateKey):
1996 return (
1997 ASN1_OID("Ed448"),
1998 hashes.SHAKE256(64),
1999 "shake256",
2000 )
2001 else:
2002 raise ValueError("Unknown private key type !")
2004 def sign(
2005 self,
2006 message: Union[bytes, Packet],
2007 eContentType: ASN1_OID,
2008 cert: Cert,
2009 key: PrivKey,
2010 dhash: Optional[str] = "sha256",
2011 ):
2012 """
2013 Sign a message using CMS.
2015 :param message: the inner content to sign.
2016 :param eContentType: the OID of the inner content.
2017 :param cert: the certificate whose key to use use for signing.
2018 :param key: the private key to use for signing.
2019 :param dhash: the hash to use for message digest (ECDSA only).
2021 We currently only support X.509 certificates !
2022 """
2023 sigalg, cdhash, dhash = self._get_algorithms(key, h=dhash)
2025 # RFC3852 5.4. Message Digest Calculation Process
2026 hash = hashes.Hash(cdhash)
2027 hash.update(bytes(message))
2028 hashed_message = hash.finalize()
2030 # RFC3852 5.5. Signature Generation Process
2031 signerInfo = CMS_SignerInfo(
2032 version=1,
2033 sid=CMS_IssuerAndSerialNumber(
2034 issuer=cert.tbsCertificate.issuer,
2035 serialNumber=cert.tbsCertificate.serialNumber,
2036 ),
2037 digestAlgorithm=X509_AlgorithmIdentifier(
2038 algorithm=ASN1_OID(dhash),
2039 parameters=ASN1_NULL(0),
2040 ),
2041 signedAttrs=[
2042 X509_Attribute(
2043 type=ASN1_OID("contentType"),
2044 values=[
2045 X509_AttributeValue(value=eContentType),
2046 ],
2047 ),
2048 X509_Attribute(
2049 type=ASN1_OID("messageDigest"),
2050 # "A message-digest attribute MUST have a single attribute value"
2051 values=[
2052 X509_AttributeValue(value=ASN1_STRING(hashed_message)),
2053 ],
2054 ),
2055 ],
2056 signatureAlgorithm=X509_AlgorithmIdentifier(
2057 algorithm=sigalg,
2058 parameters=ASN1_NULL(0),
2059 ),
2060 )
2061 signerInfo.signature = ASN1_STRING(
2062 key.sign(
2063 bytes(
2064 CMS_SignedAttrsForSignature(
2065 signedAttrs=signerInfo.signedAttrs,
2066 )
2067 ),
2068 h=dhash,
2069 )
2070 )
2072 # Build a chain of X509_Cert to ship (but skip the ROOT certificate)
2073 certTree = CertTree(cert, self.store)
2074 certificates = [x.x509Cert for x in certTree if not x.isSelfSigned()]
2076 # Build final structure
2077 return CMS_ContentInfo(
2078 contentType=ASN1_OID("id-signedData"),
2079 content=CMS_SignedData(
2080 version=3 if certificates else 1,
2081 digestAlgorithms=X509_AlgorithmIdentifier(
2082 algorithm=ASN1_OID(dhash),
2083 parameters=ASN1_NULL(0),
2084 ),
2085 encapContentInfo=CMS_EncapsulatedContentInfo(
2086 eContentType=eContentType,
2087 eContent=message,
2088 ),
2089 certificates=(
2090 [CMS_CertificateChoices(certificate=cert) for cert in certificates]
2091 if certificates
2092 else None
2093 ),
2094 crls=(
2095 [CMS_RevocationInfoChoice(crl=crl) for crl in self.crls]
2096 if self.crls
2097 else None
2098 ),
2099 signerInfos=[
2100 signerInfo,
2101 ],
2102 ),
2103 )
2105 def verify(
2106 self,
2107 contentInfo: CMS_ContentInfo,
2108 eContentType: Optional[ASN1_OID] = None,
2109 eContent: Optional[bytes] = None,
2110 no_verify_cert: bool = False,
2111 allow_expired: bool = False,
2112 ):
2113 """
2114 Verify a CMS message against the list of trusted certificates,
2115 and return the unpacked message if the verification succeeds.
2117 :param contentInfo: the ContentInfo whose signature to verify
2118 :param eContentType: if provided, verifies that the content type is valid
2119 :param eContent: in PKCS 7.1, provide the content to verify
2120 :param no_verify_cert: do not check the remote certificate (unsafe)
2121 :param allow_expired: accept a signer certificate that is outside its
2122 validity period. A CMS signature is often checked long after it was
2123 made, so an expired signer does not by itself mean the signature is
2124 not genuine.
2125 """
2126 if contentInfo.contentType.oidname != "id-signedData":
2127 raise ValueError("ContentInfo isn't signed !")
2129 signeddata = contentInfo.content
2131 # Build the certificate chain
2132 certificates = []
2133 if signeddata.certificates:
2134 certificates = [Cert(x.certificate) for x in signeddata.certificates]
2135 certTree = CertTree(certificates, self.store)
2137 # Check there's at least one signature
2138 if not signeddata.signerInfos:
2139 raise ValueError("ContentInfo contained no signature !")
2141 # Check all signatures
2142 for signerInfo in signeddata.signerInfos:
2143 # RFC 5652 sect 5.4: the digest is the one named by
2144 # digestAlgorithm, not by signatureAlgorithm.
2145 sigh = signerInfo.digestAlgorithm.algorithm.oidname
2146 if sigh not in _tls_hash_algs:
2147 sigh = hash_by_oid[signerInfo.signatureAlgorithm.algorithm.val]
2149 # Find certificate in the chain that did this
2150 cert: Cert = certTree.findCertBySid(signerInfo.sid)
2152 # Verify certificate signature
2153 if not no_verify_cert:
2154 certTree.verify(cert, allow_expired=allow_expired)
2156 # Verify the message hash
2157 if signerInfo.signedAttrs:
2158 # Verify the contentType
2159 try:
2160 contentType = next(
2161 x.values[0].value
2162 for x in signerInfo.signedAttrs
2163 if x.type.oidname == "contentType"
2164 )
2166 if contentType != signeddata.encapContentInfo.eContentType:
2167 raise ValueError(
2168 "Inconsistent 'contentType' was detected in packet !"
2169 )
2171 if eContentType is not None and eContentType != contentType:
2172 raise ValueError(
2173 "Expected '%s' but got '%s' contentType !"
2174 % (
2175 eContentType,
2176 contentType,
2177 )
2178 )
2179 except StopIteration:
2180 raise ValueError("Missing contentType in signedAttrs !")
2182 # Verify the messageDigest value
2183 try:
2184 # "A message-digest attribute MUST have a single attribute value"
2185 messageDigest = next(
2186 x.values[0].value
2187 for x in signerInfo.signedAttrs
2188 if x.type.oidname == "messageDigest"
2189 )
2191 if signeddata.encapContentInfo.eContent is not None:
2192 eContent = bytes(signeddata.encapContentInfo.eContent)
2193 elif eContent is None:
2194 raise ValueError("No eContent was provided !")
2196 # Re-calculate hash
2197 h = signerInfo.digestAlgorithm.algorithm.oidname
2198 hash = hashes.Hash(_get_hash(h))
2199 hash.update(eContent)
2200 hashed_message = hash.finalize()
2202 if hashed_message != messageDigest:
2203 raise ValueError("Invalid messageDigest value !")
2204 except StopIteration:
2205 raise ValueError("Missing messageDigest in signedAttrs !")
2207 # Verify the signature
2208 if not cert.verify(
2209 msg=bytes(
2210 CMS_SignedAttrsForSignature(
2211 signedAttrs=signerInfo.signedAttrs,
2212 )
2213 ),
2214 sig=signerInfo.signature.val,
2215 h=sigh,
2216 ):
2217 raise ValueError("Invalid signature !")
2218 else:
2219 if not cert.verify(
2220 msg=bytes(signeddata.encapContentInfo),
2221 sig=signerInfo.signature.val,
2222 h=sigh,
2223 ):
2224 raise ValueError("Invalid signature !")
2226 # Return the content
2227 return signeddata.encapContentInfo.eContent