TLS 1.3: support EdDSA (#4463)

scapy/asn1/mib.py

@@ -260,6 +260,13 @@ def load_mib(filenames):
     "1.3.14.3.2.29": "sha1RSASign",
 }
 
+#       thawte      #
+
+thawte_oids = {
+    "1.3.101.112": "Ed25519",
+    "1.3.101.113": "Ed448",
+}
+
 #       pkcs9       #
 
 pkcs9_oids = {
@@ -669,6 +676,7 @@ def load_mib(filenames):
 x509_oids_sets = [
     pkcs1_oids,
     secsig_oids,
+    thawte_oids,
     pkcs9_oids,
     attributeType_oids,
     certificateExtension_oids,

scapy/layers/tls/automaton_cli.py

@@ -175,9 +175,11 @@ def parse_args(self, server="127.0.0.1", dport=4433, server_name=None,
 
         if supported_signature_algorithms is None:
             supported_signature_algorithms = [
+                "sha256+rsaepss",
                 "sha256+rsa",
+                "ed25519",
+                "ed448",
             ]
-            supported_signature_algorithms.insert(0, "sha256+rsaepss")
         self.supported_signature_algorithms = supported_signature_algorithms
 
         self.curve = None

scapy/layers/tls/automaton_srv.py

@@ -30,7 +30,7 @@
 from scapy.automaton import ATMT
 from scapy.error import warning
 from scapy.layers.tls.automaton import _TLSAutomaton
-from scapy.layers.tls.cert import PrivKeyRSA, PrivKeyECDSA
+from scapy.layers.tls.cert import PrivKeyRSA, PrivKeyECDSA, PrivKeyEdDSA
 from scapy.layers.tls.basefields import _tls_version
 from scapy.layers.tls.session import tlsSession
 from scapy.layers.tls.crypto.groups import _tls_named_groups
@@ -339,6 +339,8 @@ def HANDLED_CLIENTHELLO(self):
             kx = "RSA"
         elif isinstance(self.mykey, PrivKeyECDSA):
             kx = "ECDSA"
+        elif isinstance(self.mykey, PrivKeyEdDSA):
+            kx = ""
         if get_usable_ciphersuites(self.cur_pkt.ciphers, kx):
             raise self.PREPARE_SERVERFLIGHT1()
         raise self.NO_USABLE_CIPHERSUITE()
@@ -371,6 +373,8 @@ def should_add_ServerHello(self):
             kx = "RSA"
         elif isinstance(self.mykey, PrivKeyECDSA):
             kx = "ECDSA"
+        elif isinstance(self.mykey, PrivKeyEdDSA):
+            kx = ""
         usable_suites = get_usable_ciphersuites(self.cur_pkt.ciphers, kx)
         c = usable_suites[0]
         if self.preferred_ciphersuite in usable_suites:
@@ -646,6 +650,8 @@ def tls13_should_add_HelloRetryRequest(self):
             kx = "RSA"
         elif isinstance(self.mykey, PrivKeyECDSA):
             kx = "ECDSA"
+        elif isinstance(self.mykey, PrivKeyEdDSA):
+            kx = ""
         usable_suites = get_usable_ciphersuites(self.cur_pkt.ciphers, kx)
         c = usable_suites[0]
         ext = [TLS_Ext_SupportedVersion_SH(version="TLS 1.3"),
@@ -786,6 +792,8 @@ def tls13_should_add_ServerHello(self):
             kx = "RSA"
         elif isinstance(self.mykey, PrivKeyECDSA):
             kx = "ECDSA"
+        elif isinstance(self.mykey, PrivKeyEdDSA):
+            kx = ""
         usable_suites = get_usable_ciphersuites(self.cur_pkt.ciphers, kx)
         c = usable_suites[0]
         group = next(iter(self.cur_session.tls13_client_pubshares))

scapy/layers/tls/cert.py

@@ -36,11 +36,19 @@
 from scapy.utils import binrepr
 from scapy.asn1.asn1 import ASN1_BIT_STRING
 from scapy.asn1.mib import hash_by_oid
-from scapy.layers.x509 import (X509_SubjectPublicKeyInfo,
-                               RSAPublicKey, RSAPrivateKey,
-                               ECDSAPublicKey, ECDSAPrivateKey,
-                               RSAPrivateKey_OpenSSL, ECDSAPrivateKey_OpenSSL,
-                               X509_Cert, X509_CRL)
+from scapy.layers.x509 import (
+    ECDSAPrivateKey_OpenSSL,
+    ECDSAPrivateKey,
+    ECDSAPublicKey,
+    EdDSAPublicKey,
+    EdDSAPrivateKey,
+    RSAPrivateKey_OpenSSL,
+    RSAPrivateKey,
+    RSAPublicKey,
+    X509_Cert,
+    X509_CRL,
+    X509_SubjectPublicKeyInfo,
+)
 from scapy.layers.tls.crypto.pkcs1 import pkcs_os2ip, _get_hash, \
     _EncryptAndVerifyRSA, _DecryptAndSignRSA
 from scapy.compat import raw, bytes_encode
@@ -49,7 +57,7 @@
     from cryptography.exceptions import InvalidSignature
     from cryptography.hazmat.backends import default_backend
     from cryptography.hazmat.primitives import serialization
-    from cryptography.hazmat.primitives.asymmetric import rsa, ec
+    from cryptography.hazmat.primitives.asymmetric import rsa, ec, x25519
 
     # cryptography raised the minimum RSA key length to 1024 in 43.0+
     # https://github.com/pyca/cryptography/pull/10278
@@ -221,7 +229,8 @@ def __call__(cls, key_path=None):
         # Now for the usual calls, key_path may be the path to either:
         # _an X509_SubjectPublicKeyInfo, as processed by openssl;
         # _an RSAPublicKey;
-        # _an ECDSAPublicKey.
+        # _an ECDSAPublicKey;
+        # _an EdDSAPublicKey.
         obj = _PKIObjMaker.__call__(cls, key_path, _MAX_KEY_SIZE)
         try:
             spki = X509_SubjectPublicKeyInfo(obj.der)
@@ -231,10 +240,10 @@ def __call__(cls, key_path=None):
                 obj.import_from_asn1pkt(pubkey)
             elif isinstance(pubkey, ECDSAPublicKey):
                 obj.__class__ = PubKeyECDSA
-                try:
-                    obj.import_from_der(obj.der)
-                except ImportError:
-                    pass
+                obj.import_from_der(obj.der)
+            elif isinstance(pubkey, EdDSAPublicKey):
+                obj.__class__ = PubKeyEdDSA
+                obj.import_from_der(obj.der)
             else:
                 raise
             marker = b"PUBLIC KEY"
@@ -347,11 +356,12 @@ def fill_and_store(self, curve=None):
     @crypto_validator
     def import_from_der(self, pubkey):
         # No lib support for explicit curves nor compressed points.
-        self.pubkey = serialization.load_der_public_key(pubkey,
-                                                        backend=default_backend())  # noqa: E501
+        self.pubkey = serialization.load_der_public_key(
+            pubkey,
+            backend=default_backend(),
+        )
 
     def encrypt(self, msg, h="sha256", **kwargs):
-        # cryptography lib does not support ECDSA encryption
         raise Exception("No ECDSA encryption support")
 
     @crypto_validator
@@ -364,6 +374,37 @@ def verify(self, msg, sig, h="sha256", **kwargs):
             return False
 
 
+class PubKeyEdDSA(PubKey):
+    """
+    Wrapper for EdDSA keys based on the cryptography library.
+    Use the 'key' attribute to access original object.
+    """
+    @crypto_validator
+    def fill_and_store(self, curve=None):
+        curve = curve or x25519.X25519PrivateKey
+        private_key = curve.generate()
+        self.pubkey = private_key.public_key()
+
+    @crypto_validator
+    def import_from_der(self, pubkey):
+        self.pubkey = serialization.load_der_public_key(
+            pubkey,
+            backend=default_backend(),
+        )
+
+    def encrypt(self, msg, **kwargs):
+        raise Exception("No EdDSA encryption support")
+
+    @crypto_validator
+    def verify(self, msg, sig, **kwargs):
+        # 'sig' should be a DER-encoded signature, as per RFC 3279
+        try:
+            self.pubkey.verify(sig, msg)
+            return True
+        except InvalidSignature:
+            return False
+
+
 ################
 # Private Keys #
 ################
@@ -416,7 +457,12 @@ def __call__(cls, key_path=None):
                         obj.__class__ = PrivKeyECDSA
                         marker = b"EC PRIVATE KEY"
                     except Exception:
-                        raise Exception("Unable to import private key")
+                        try:
+                            privkey = EdDSAPrivateKey(obj.der)
+                            obj.__class__ = PrivKeyEdDSA
+                            marker = b"PRIVATE KEY"
+                        except Exception:
+                            raise Exception("Unable to import private key")
         try:
             obj.import_from_asn1pkt(privkey)
         except ImportError:
@@ -581,6 +627,37 @@ def sign(self, data, h="sha256", **kwargs):
         return self.key.sign(data, ec.ECDSA(_get_hash(h)))
 
 
+class PrivKeyEdDSA(PrivKey):
+    """
+    Wrapper for EdDSA keys
+    Use the 'key' attribute to access original object.
+    """
+    @crypto_validator
+    def fill_and_store(self, curve=None):
+        curve = curve or x25519.X25519PrivateKey
+        self.key = curve.generate()
+        self.pubkey = self.key.public_key()
+
+    @crypto_validator
+    def import_from_asn1pkt(self, privkey):
+        self.key = serialization.load_der_private_key(raw(privkey), None,
+                                                      backend=default_backend())  # noqa: E501
+        self.pubkey = self.key.public_key()
+
+    @crypto_validator
+    def verify(self, msg, sig, **kwargs):
+        # 'sig' should be a DER-encoded signature, as per RFC 3279
+        try:
+            self.pubkey.verify(sig, msg)
+            return True
+        except InvalidSignature:
+            return False
+
+    @crypto_validator
+    def sign(self, data, **kwargs):
+        return self.key.sign(data)
+
+
 ################
 # Certificates #
 ################

scapy/layers/tls/keyexchange_tls13.py

@@ -26,7 +26,7 @@
 )
 from scapy.packet import Packet
 from scapy.layers.tls.extensions import TLS_Ext_Unknown, _tls_ext
-from scapy.layers.tls.cert import PrivKeyECDSA, PrivKeyRSA
+from scapy.layers.tls.cert import PrivKeyECDSA, PrivKeyRSA, PrivKeyEdDSA
 from scapy.layers.tls.crypto.groups import (
     _tls_named_curves,
     _tls_named_ffdh_groups,
@@ -39,8 +39,8 @@
 if conf.crypto_valid:
     from cryptography.hazmat.primitives.asymmetric import ec
 if conf.crypto_valid_advanced:
-    from cryptography.hazmat.primitives.asymmetric import x25519
-    from cryptography.hazmat.primitives.asymmetric import x448
+    from cryptography.hazmat.primitives.asymmetric import ed25519
+    from cryptography.hazmat.primitives.asymmetric import ed448
 
 
 class KeyShareEntry(Packet):
@@ -329,15 +329,16 @@ def get_usable_tls13_sigalgs(li, key, location="certificateverify"):
     from scapy.layers.tls.keyexchange import _tls_hash_sig
     res = []
     if isinstance(key, PrivKeyRSA):
-        kxs = ["rsa"]
+        kx = "rsa"
     elif isinstance(key, PrivKeyECDSA):
-        kxs = []
-        if isinstance(key.key, x25519.X25519PrivateKey):
-            kxs.append("ed25519")
-        elif isinstance(key.key, x448.X448PrivateKey):
-            kxs.append("edx448")
+        kx = "ecdsa"
+    elif isinstance(key, PrivKeyEdDSA):
+        if isinstance(key.pubkey, ed25519.Ed25519PublicKey):
+            kx = "ed25519"
+        elif isinstance(key.pubkey, ed448.Ed448PublicKey):
+            kx = "ed448"
         else:
-            kxs = ["ecdsa"]
+            kx = "unknown"
     else:
         return res
     if location == "certificateverify":
@@ -353,6 +354,6 @@ def get_usable_tls13_sigalgs(li, key, location="certificateverify"):
                 _, sig = sigalg.split('+')
             else:
                 sig = sigalg
-            if any(kx in sig for kx in kxs):
+            if kx in sig:
                 res.append(c)
     return res

scapy/layers/x509.py

@@ -180,6 +180,35 @@ class ECDSASignature(ASN1_Packet):
         ASN1F_INTEGER("s", 0))
 
 
+####################################
+#      x25519/x448 packets         #
+####################################
+# based on RFC 8410
+
+class EdDSAPublicKey(ASN1_Packet):
+    ASN1_codec = ASN1_Codecs.BER
+    ASN1_root = ASN1F_BIT_STRING("ecPoint", "")
+
+
+class AlgorithmIdentifier(ASN1_Packet):
+    ASN1_codec = ASN1_Codecs.BER
+    ASN1_root = ASN1F_SEQUENCE(
+        ASN1F_OID("algorithm", None),
+    )
+
+
+class EdDSAPrivateKey(ASN1_Packet):
+    ASN1_codec = ASN1_Codecs.BER
+    ASN1_root = ASN1F_SEQUENCE(
+        ASN1F_enum_INTEGER("version", 1, {1: "ecPrivkeyVer1"}),
+        ASN1F_PACKET("privateKeyAlgorithm", AlgorithmIdentifier(), AlgorithmIdentifier),
+        ASN1F_STRING("privateKey", ""),
+        ASN1F_optional(
+            ASN1F_PACKET("publicKey", None,
+                         ECDSAPublicKey,
+                         explicit_tag=0xa1)))
+
+
 ######################
 #    X509 packets    #
 ######################
@@ -799,6 +828,10 @@ def __init__(self, **kargs):
                                      ECDSAPublicKey(),
                                      ECDSAPublicKey),
                         lambda pkt: "ecPublicKey" == pkt.signatureAlgorithm.algorithm.oidname),  # noqa: E501
+                       (ASN1F_PACKET("subjectPublicKey",
+                                     EdDSAPublicKey(),
+                                     EdDSAPublicKey),
+                        lambda pkt: pkt.signatureAlgorithm.algorithm.oidname in ["Ed25519", "Ed448"]),  # noqa: E501
                    ],
                    ASN1F_BIT_STRING("subjectPublicKey", ""))]
         ASN1F_SEQUENCE.__init__(self, *seq, **kargs)