Add support for EdDSA on server side messages

Author: tomato42

.gitignore

@@ -1,3 +1,4 @@
+*venv*
 *.pyc
 .project
 .pydevproject

tests/serverEd25519Cert.pem

@@ -0,0 +1,9 @@
+-----BEGIN CERTIFICATE-----
+MIIBPDCB76ADAgECAhQkqENccCvOQyI4iKFuuOKwl860bTAFBgMrZXAwFDESMBAG
+A1UEAwwJbG9jYWxob3N0MB4XDTIxMDcyNjE0MjcwN1oXDTIxMDgyNTE0MjcwN1ow
+FDESMBAGA1UEAwwJbG9jYWxob3N0MCowBQYDK2VwAyEA1KMGmAZealfgakBuCx/E
+n69fo072qm90eM40ulGex0ajUzBRMB0GA1UdDgQWBBTHKWv5l/SxnkkYJhh5r3Pv
+ESAh1DAfBgNVHSMEGDAWgBTHKWv5l/SxnkkYJhh5r3PvESAh1DAPBgNVHRMBAf8E
+BTADAQH/MAUGAytlcANBAF/vSBfOHAdRl29sWDTkuqy1dCuSf7j7jKE/Be8Fk7xs
+WteXJmIa0HlRAZjxNfWbsSGLnTYbsGTbxKx3QU9H9g0=
+-----END CERTIFICATE-----

tests/serverEd25519Key.pem

@@ -0,0 +1,3 @@
+-----BEGIN PRIVATE KEY-----
+MC4CAQAwBQYDK2VwBCIEIAjtEwCECqbot5RZxSmiNDWcPp+Xc9Y9WJcUhti3JgSP
+-----END PRIVATE KEY-----

tests/serverEd448Cert.pem

@@ -0,0 +1,11 @@
+-----BEGIN CERTIFICATE-----
+MIIBiDCCAQigAwIBAgIUZoaDDgE5Cy2GuAMtk4lnsmrPF04wBQYDK2VxMBQxEjAQ
+BgNVBAMMCWxvY2FsaG9zdDAeFw0yMTA3MjYxODAzMzhaFw0yMTA4MjUxODAzMzha
+MBQxEjAQBgNVBAMMCWxvY2FsaG9zdDBDMAUGAytlcQM6AKxTNGJ39O4kUx7BopPK
+prb1Jkoo0csq0Cmpa+VhpDlbR9/gVsb3pchexzjxXyRkNv71naHmOkQvAKNTMFEw
+HQYDVR0OBBYEFBb153yRh5IZOfBxoakGVuviFKujMB8GA1UdIwQYMBaAFBb153yR
+h5IZOfBxoakGVuviFKujMA8GA1UdEwEB/wQFMAMBAf8wBQYDK2VxA3MAiXEqTPRb
+u+56ebfiGjdE++H+YvHVxxxycqKAIAikfsLFfw2LUGQVBMhl+nzS4zRDOKa34uGz
+DwEApFuOWurH/y8zqM5NFyXfwbHRlhG4xwUet52CbrtC7Dy1HYnvWdEjbKDSJXpJ
+MmNSiO0oBtQ62CsA
+-----END CERTIFICATE-----

tests/serverEd448Key.pem

@@ -0,0 +1,4 @@
+-----BEGIN PRIVATE KEY-----
+MEcCAQAwBQYDK2VxBDsEOWC42wrEHt4sse84L8oi/2LfqtYvT+Xwd5USLJuAUi6h
+Ht8RBuFGD/DoZIfwfBgBfemM56jAnbQIug==
+-----END PRIVATE KEY-----

tests/tlstest.py

@@ -538,6 +538,40 @@ def connect():
 
     test_no += 1
 
+    print("Test {0} - good X.509 Ed25519, TLSv1.3".format(test_no))
+    synchro.recv(1)
+    connection = connect()
+    settings = HandshakeSettings()
+    settings.minVersion = (3, 4)
+    settings.maxVersion = (3, 4)
+    connection.handshakeClientCert(settings=settings)
+    testConnClient(connection)
+    assert connection.session.cipherSuite in\
+            constants.CipherSuite.tls13Suites
+    assert isinstance(connection.session.serverCertChain, X509CertChain)
+    assert connection.session.serverCertChain.getEndEntityPublicKey().key_type \
+            == "Ed25519"
+    connection.close()
+
+    test_no += 1
+
+    print("Test {0} - good X.509 Ed448, TLSv1.3".format(test_no))
+    synchro.recv(1)
+    connection = connect()
+    settings = HandshakeSettings()
+    settings.minVersion = (3, 4)
+    settings.maxVersion = (3, 4)
+    connection.handshakeClientCert(settings=settings)
+    testConnClient(connection)
+    assert connection.session.cipherSuite in\
+            constants.CipherSuite.tls13Suites
+    assert isinstance(connection.session.serverCertChain, X509CertChain)
+    assert connection.session.serverCertChain.getEndEntityPublicKey().key_type \
+            == "Ed448"
+    connection.close()
+
+    test_no += 1
+
     print("Test {0} - good RSA and ECDSA, TLSv1.3, rsa"
           .format(test_no))
     synchro.recv(1)
@@ -877,6 +911,42 @@ def connect():
     assert isinstance(connection.session.serverCertChain, X509CertChain)
     connection.close()
 
+    test_no += 1
+
+    print("Test {0} - good X.509 Ed25519, TLSv1.2".format(test_no))
+    synchro.recv(1)
+    connection = connect()
+    settings = HandshakeSettings()
+    settings.minVersion = (3, 3)
+    settings.maxVersion = (3, 3)
+    connection.handshakeClientCert(settings=settings)
+    testConnClient(connection)
+    assert connection.session.cipherSuite in\
+            constants.CipherSuite.ecdheEcdsaSuites
+    assert isinstance(connection.session.serverCertChain, X509CertChain)
+    assert connection.session.serverCertChain.getEndEntityPublicKey().key_type \
+            == "Ed25519"
+    connection.close()
+
+    test_no += 1
+
+    print("Test {0} - good X.509 Ed448, TLSv1.2".format(test_no))
+    synchro.recv(1)
+    connection = connect()
+    settings = HandshakeSettings()
+    settings.minVersion = (3, 3)
+    settings.maxVersion = (3, 3)
+    connection.handshakeClientCert(settings=settings)
+    testConnClient(connection)
+    assert connection.session.cipherSuite in\
+            constants.CipherSuite.ecdheEcdsaSuites
+    assert isinstance(connection.session.serverCertChain, X509CertChain)
+    assert connection.session.serverCertChain.getEndEntityPublicKey().key_type \
+            == "Ed448"
+    connection.close()
+
+    test_no += 1
+
     print("Test {0} - good mutual X.509, TLSv1.3 no certs".format(test_no))
     synchro.recv(1)
     connection = connect()
@@ -1739,6 +1809,23 @@ def connect():
     with open(os.path.join(dir, "serverDSAKey.pem")) as f:
         x509KeyDSA = parsePEMKey(f.read(), private=True,
                                     implementations=["python"])
+
+    with open(os.path.join(dir, "serverEd25519Cert.pem")) as f:
+        x509CertEd25519 = X509().parse(f.read())
+    x509Ed25519Chain = X509CertChain([x509CertEd25519])
+    assert x509CertEd25519.certAlg == "Ed25519"
+    with open(os.path.join(dir, "serverEd25519Key.pem")) as f:
+        x509Ed25519Key = parsePEMKey(f.read(), private=True,
+                                     implementations=["python"])
+
+    with open(os.path.join(dir, "serverEd448Cert.pem")) as f:
+        x509CertEd448 = X509().parse(f.read())
+    x509Ed448Chain = X509CertChain([x509CertEd448])
+    assert x509CertEd448.certAlg == "Ed448"
+    with open(os.path.join(dir, "serverEd448Key.pem")) as f:
+        x509Ed448Key = parsePEMKey(f.read(), private=True,
+                                   implementations=["python"])
+
     test_no = 0
 
     print("Test {0} - Anonymous server handshake".format(test_no))
@@ -2080,6 +2167,34 @@ def connect():
 
     test_no += 1
 
+    print("Test {0} - good X.509 Ed25519, TLSv1.3".format(test_no))
+    synchro.send(b'R')
+    connection = connect()
+    settings = HandshakeSettings()
+    settings.minVersion = (3, 4)
+    settings.maxVersion = (3, 4)
+    connection.handshakeServer(certChain=x509Ed25519Chain,
+                               privateKey=x509Ed25519Key, settings=settings)
+    assert connection.extendedMasterSecret
+    testConnServer(connection)
+    connection.close()
+
+    test_no += 1
+
+    print("Test {0} - good X.509 Ed448, TLSv1.3".format(test_no))
+    synchro.send(b'R')
+    connection = connect()
+    settings = HandshakeSettings()
+    settings.minVersion = (3, 4)
+    settings.maxVersion = (3, 4)
+    connection.handshakeServer(certChain=x509Ed448Chain,
+                               privateKey=x509Ed448Key, settings=settings)
+    assert connection.extendedMasterSecret
+    testConnServer(connection)
+    connection.close()
+
+    test_no += 1
+
     for prot in ["TLSv1.3", "TLSv1.2"]:
         for c_type, exp_chain in (("rsa", x509Chain),
                                   ("ecdsa", x509ecdsaChain)):
@@ -2347,6 +2462,32 @@ def connect():
 
     test_no += 1
 
+    print("Test {0} - good X.509 Ed25519, TLSv1.2".format(test_no))
+    synchro.send(b'R')
+    connection = connect()
+    settings = HandshakeSettings()
+    settings.minVersion = (3, 3)
+    settings.maxVersion = (3, 3)
+    connection.handshakeServer(certChain=x509Ed25519Chain,
+                               privateKey=x509Ed25519Key, settings=settings)
+    testConnServer(connection)
+    connection.close()
+
+    test_no += 1
+
+    print("Test {0} - good X.509 Ed448, TLSv1.2".format(test_no))
+    synchro.send(b'R')
+    connection = connect()
+    settings = HandshakeSettings()
+    settings.minVersion = (3, 3)
+    settings.maxVersion = (3, 3)
+    connection.handshakeServer(certChain=x509Ed448Chain,
+                               privateKey=x509Ed448Key, settings=settings)
+    testConnServer(connection)
+    connection.close()
+
+    test_no += 1
+
     print("Test {0} - good mutual X.509, TLSv1.3 no certs".format(test_no))
     synchro.send(b'R')
     connection = connect()

tlslite/constants.py

@@ -260,9 +260,8 @@ def getKeyType(scheme):
 
         E.g. for "rsa_pkcs1_sha1" it returns "rsa"
         """
-        # they need to be threated as ECDSA algorithms, see RFC 8422
         if scheme in ("ed25519", "ed448"):
-            return "ecdsa"
+            return "eddsa"
         try:
             getattr(SignatureScheme, scheme)
         except AttributeError:
@@ -363,6 +362,10 @@ class AlgorithmOID(TLSEnum):
             SignatureScheme.dsa_sha384
     oid[bytes(a2b_hex('0609608648016503040304'))] = \
             SignatureScheme.dsa_sha512
+    oid[bytes(a2b_hex('06032b6570'))] = \
+            SignatureScheme.ed25519
+    oid[bytes(a2b_hex('06032b6571'))] = \
+            SignatureScheme.ed448
 
 
 class GroupName(TLSEnum):
@@ -1265,7 +1268,7 @@ def filter_for_certificate(suites, cert_chain):
                 # rsa-pss
                 includeSuites.symmetric_difference_update(
                     CipherSuite.certSuites)
-            if cert_chain.x509List[0].certAlg == "ecdsa":
+            if cert_chain.x509List[0].certAlg in ("ecdsa", "Ed25519", "Ed448"):
                 includeSuites.update(CipherSuite.ecdheEcdsaSuites)
             if cert_chain.x509List[0].certAlg == "dsa":
                 includeSuites.update(CipherSuite.dheDsaSuites)

tlslite/handshakesettings.py

@@ -31,6 +31,7 @@
 DSA_SIGNATURE_HASHES = ["sha512", "sha384", "sha256", "sha224", "sha1"]
 ECDSA_SIGNATURE_HASHES = ["sha512", "sha384", "sha256", "sha224", "sha1"]
 ALL_RSA_SIGNATURE_HASHES = RSA_SIGNATURE_HASHES + ["md5"]
+SIGNATURE_SCHEMES = ["Ed25519", "Ed448"]
 RSA_SCHEMES = ["pss", "pkcs1"]
 # while secp521r1 is the most secure, it's also much slower than the others
 # so place it as the last one
@@ -255,8 +256,15 @@ class HandshakeSettings(object):
         The allowed hashes are: "sha1", "sha224", "sha256",
         "sha384" and "sha512".
 
+    "vartype more_sig_schemes: list(str)
+    :ivar more_sig_schemes: List of additional signatures schemes (ones
+        that don't use RSA-PKCS#1 v1.5, RSA-PSS, DSA, or ECDSA) to advertise
+        as supported.
+        Currently supported are: "Ed25519", and "Ed448".
+
     :vartype eccCurves: list(str)
-    :ivar eccCurves: List of named curves that are to be supported
+    :ivar eccCurves: List of named curves that are to be advertised as
+        supported in supported_groups extension.
 
     :vartype useEncryptThenMAC: bool
     :ivar useEncryptThenMAC: whether to support the encrypt then MAC extension
@@ -367,6 +375,7 @@ def _init_misc_extensions(self):
         self.sendFallbackSCSV = False
         self.useEncryptThenMAC = True
         self.ecdsaSigHashes = list(ECDSA_SIGNATURE_HASHES)
+        self.more_sig_schemes = list(SIGNATURE_SCHEMES)
         self.usePaddingExtension = True
         self.useExtendedMasterSecret = True
         self.requireExtendedMasterSecret = False
@@ -467,6 +476,12 @@ def _sanityCheckECDHSettings(other):
             raise ValueError("Unknown ECDSA signature hash: '{0}'".\
                              format(unknownSigHash))
 
+        unknownSigHash = not_matching(other.more_sig_schemes,
+                                      SIGNATURE_SCHEMES)
+        if unknownSigHash:
+            raise ValueError("Unkonwn more_sig_schemes specified: '{0}'"
+                             .format(unknownSigHash))
+
         unknownDHGroup = not_matching(other.dhGroups, ALL_DH_GROUP_NAMES)
         if unknownDHGroup:
             raise ValueError("Unknown FFDHE group name: '{0}'"
@@ -529,7 +544,8 @@ def _sanityCheckPrimitivesNames(other):
                              .format(unknownSigHash))
 
         if not other.rsaSigHashes and not other.ecdsaSigHashes and \
-                not other.dsaSigHashes and other.maxVersion >= (3, 3):
+                not other.dsaSigHashes and not other.more_sig_schemes and \
+                other.maxVersion >= (3, 3):
             raise ValueError("TLS 1.2 requires signature algorithms to be set")
 
     @staticmethod
@@ -688,6 +704,7 @@ def _copy_key_settings(self, other):
         other.rsaSchemes = self.rsaSchemes
         other.dsaSigHashes = self.dsaSigHashes
         other.ecdsaSigHashes = self.ecdsaSigHashes
+        other.more_sig_schemes = self.more_sig_schemes
         other.virtual_hosts = self.virtual_hosts
         # DH key params
         other.eccCurves = self.eccCurves

tlslite/keyexchange.py

@@ -123,6 +123,34 @@ def _tls12_sign_dsa_SKE(self, serverKeyExchange, sigHash=None):
                                       hashBytes):
             raise TLSInternalError("Server Key Exchange signature invalid")
 
+    def _tls12_sign_eddsa_ske(self, server_key_exchange, sig_hash):
+        """Sign a TLSv1.2 SKE message."""
+        server_key_exchange.hashAlg, server_key_exchange.signAlg = \
+                getattr(SignatureScheme, sig_hash)
+        pad_type = None
+        hash_name = None
+        salt_len = None
+
+        hash_bytes = server_key_exchange.hash(self.clientHello.random,
+                                              self.serverHello.random)
+
+        server_key_exchange.signature = \
+            self.privateKey.hashAndSign(hash_bytes,
+                                        pad_type,
+                                        hash_name,
+                                        salt_len)
+
+        if not server_key_exchange.signature:
+            raise TLSInternalError("Empty signature")
+
+        if not self.privateKey.hashAndVerify(
+                server_key_exchange.signature,
+                hash_bytes,
+                pad_type,
+                hash_name,
+                salt_len):
+            raise TLSInternalError("Server Key Exchange signature invalid")
+
     def _tls12_signSKE(self, serverKeyExchange, sigHash=None):
         """Sign a TLSv1.2 SKE message."""
         try:
@@ -189,6 +217,8 @@ def signServerKeyExchange(self, serverKeyExchange, sigHash=None):
                 self._tls12_sign_ecdsa_SKE(serverKeyExchange, sigHash)
             elif self.privateKey.key_type == "dsa":
                 self._tls12_sign_dsa_SKE(serverKeyExchange, sigHash)
+            elif self.privateKey.key_type in ("Ed25519", "Ed448"):
+                self._tls12_sign_eddsa_ske(serverKeyExchange, sigHash)
             else:
                 self._tls12_signSKE(serverKeyExchange, sigHash)
 
@@ -210,6 +240,21 @@ def _tls12_verify_ecdsa_SKE(serverKeyExchange, publicKey, clientRandom,
             raise TLSDecryptionFailed("Server Key Exchange signature "
                                       "invalid")
 
+    @staticmethod
+    def _tls12_verify_eddsa_ske(server_key_exchange, public_key, client_random,
+                                server_random, valid_sig_algs):
+        """Verify SeverKeyExchange messages with EdDSA signatures."""
+        del valid_sig_algs
+        sig_bytes = server_key_exchange.signature
+        if not sig_bytes:
+            raise TLSIllegalParameterException("Empty signature")
+
+        hash_bytes = server_key_exchange.hash(client_random, server_random)
+
+        if not public_key.hashAndVerify(sig_bytes,
+                                        hash_bytes):
+            raise TLSDecryptionFailed("Server Key Exchange signature invalid")
+
     @staticmethod
     def _tls12_verify_dsa_SKE(serverKeyExchange, publicKey, clientRandom,
                               serverRandom, validSigAlgs):
@@ -229,6 +274,13 @@ def _tls12_verify_SKE(serverKeyExchange, publicKey, clientRandom,
             raise TLSIllegalParameterException("Server selected "
                                                "invalid signature "
                                                "algorithm")
+        if (serverKeyExchange.hashAlg, serverKeyExchange.signAlg) in (
+                SignatureScheme.ed25519, SignatureScheme.ed448):
+            return KeyExchange._tls12_verify_eddsa_ske(serverKeyExchange,
+                                                       publicKey,
+                                                       clientRandom,
+                                                       serverRandom,
+                                                       validSigAlgs)
         if serverKeyExchange.signAlg == SignatureAlgorithm.ecdsa:
             return KeyExchange._tls12_verify_ecdsa_SKE(serverKeyExchange,
                                                        publicKey,
@@ -344,7 +396,8 @@ def calcVerifyBytes(version, handshakeHashes, signatureAlg,
                                     b' CertificateVerify' +
                                     b'\x00') + \
                           handshakeHashes.digest(prf_name)
-            verifyBytes = secureHash(verifyBytes, hash_name)
+            if hash_name != "intrinsic":
+                verifyBytes = secureHash(verifyBytes, hash_name)
         else:
             raise ValueError("Unsupported TLS version {0}".format(version))
         return verifyBytes