[tls] add support for ED25519 certificates

src/aioquic/tls.py

@@ -29,6 +29,7 @@
 from cryptography.hazmat.primitives.asymmetric import (
     dsa,
     ec,
+    ed25519,
     padding,
     rsa,
     x448,
@@ -1112,7 +1113,12 @@ def negotiate(
 
 def signature_algorithm_params(
     signature_algorithm: int,
-) -> Union[Tuple[ec.ECDSA], Tuple[padding.AsymmetricPadding, hashes.HashAlgorithm]]:
+) -> Union[
+    Tuple[()], Tuple[ec.ECDSA], Tuple[padding.AsymmetricPadding, hashes.HashAlgorithm]
+]:
+    if signature_algorithm == SignatureAlgorithm.ED25519:
+        return tuple()
+
     padding_cls, algorithm_cls = SIGNATURE_ALGORITHMS[signature_algorithm]
     algorithm = algorithm_cls()
     if padding_cls is None:
@@ -1228,6 +1234,7 @@ def __init__(
             SignatureAlgorithm.ECDSA_SECP256R1_SHA256,
             SignatureAlgorithm.RSA_PKCS1_SHA256,
             SignatureAlgorithm.RSA_PKCS1_SHA1,
+            SignatureAlgorithm.ED25519,
         ]
         self._supported_groups = [Group.SECP256R1]
         if default_backend().x25519_supported():
@@ -1664,6 +1671,8 @@ def _server_handle_hello(
             self.certificate_private_key, ec.EllipticCurvePrivateKey
         ) and isinstance(self.certificate_private_key.curve, ec.SECP256R1):
             signature_algorithms = [SignatureAlgorithm.ECDSA_SECP256R1_SHA256]
+        elif isinstance(self.certificate_private_key, ed25519.Ed25519PrivateKey):
+            signature_algorithms = [SignatureAlgorithm.ED25519]
 
         # negotiate parameters
         cipher_suite = negotiate(

tests/test_asyncio.py

@@ -20,6 +20,7 @@
     SERVER_KEYFILE,
     SKIP_TESTS,
     generate_ec_certificate,
+    generate_ed25519_certificate,
     run,
 )
 
@@ -152,6 +153,28 @@ def test_connect_and_serve_ec_certificate(self):
 
         self.assertEqual(response, b"gnip")
 
+    def test_connect_and_serve_ed25519_certificate(self):
+        certificate, private_key = generate_ed25519_certificate(common_name="localhost")
+
+        run(
+            self.run_server(
+                configuration=QuicConfiguration(
+                    certificate=certificate,
+                    private_key=private_key,
+                    is_client=False,
+                )
+            )
+        )
+
+        response = run(
+            self.run_client(
+                cadata=certificate.public_bytes(serialization.Encoding.PEM),
+                cafile=None,
+            )
+        )
+
+        self.assertEqual(response, b"gnip")
+
     def test_connect_and_serve_large(self):
         """
         Transfer enough data to require raising MAX_DATA and MAX_STREAM_DATA.

tests/test_tls.py

@@ -45,6 +45,7 @@
     SERVER_CERTFILE,
     SERVER_KEYFILE,
     generate_ec_certificate,
+    generate_ed25519_certificate,
     load,
 )
 
@@ -325,7 +326,7 @@ def _handshake(self, client, server):
         server.handle_message(server_input, server_buf)
         self.assertEqual(server.state, State.SERVER_EXPECT_FINISHED)
         client_input = merge_buffers(server_buf)
-        self.assertGreaterEqual(len(client_input), 600)
+        self.assertGreaterEqual(len(client_input), 588)
         self.assertLessEqual(len(client_input), 2316)
 
         reset_buffers(server_buf)
@@ -383,6 +384,24 @@ def test_handshake_ecdsa_secp256r1(self):
         self.assertEqual(client.alpn_negotiated, None)
         self.assertEqual(server.alpn_negotiated, None)
 
+    def test_handshake_ed25519(self):
+        server = self.create_server()
+        (
+            server.certificate,
+            server.certificate_private_key,
+        ) = generate_ed25519_certificate(common_name="example.com")
+
+        client = self.create_client(
+            cadata=server.certificate.public_bytes(serialization.Encoding.PEM),
+            cafile=None,
+        )
+
+        self._handshake(client, server)
+
+        # check ALPN matches
+        self.assertEqual(client.alpn_negotiated, None)
+        self.assertEqual(server.alpn_negotiated, None)
+
     def test_handshake_with_alpn(self):
         client = self.create_client(alpn_protocols=["hq-20"])
         server = self.create_server(alpn_protocols=["hq-20", "h3-20"])

tests/utils.py

@@ -7,12 +7,10 @@
 from cryptography import x509
 from cryptography.hazmat.backends import default_backend
 from cryptography.hazmat.primitives import hashes
-from cryptography.hazmat.primitives.asymmetric import ec
+from cryptography.hazmat.primitives.asymmetric import ec, ed25519
 
 
-def generate_ec_certificate(common_name, curve=ec.SECP256R1, alternative_names=[]):
-    key = ec.generate_private_key(backend=default_backend(), curve=curve)
-
+def generate_certificate(*, alternative_names, common_name, hash_algorithm, key):
     subject = issuer = x509.Name(
         [x509.NameAttribute(x509.NameOID.COMMON_NAME, common_name)]
     )
@@ -33,10 +31,30 @@ def generate_ec_certificate(common_name, curve=ec.SECP256R1, alternative_names=[
             ),
             critical=False,
         )
-    cert = builder.sign(key, hashes.SHA256(), default_backend())
+    cert = builder.sign(key, hash_algorithm, default_backend())
     return cert, key
 
 
+def generate_ec_certificate(common_name, curve=ec.SECP256R1, alternative_names=[]):
+    key = ec.generate_private_key(backend=default_backend(), curve=curve)
+    return generate_certificate(
+        alternative_names=alternative_names,
+        common_name=common_name,
+        hash_algorithm=hashes.SHA256(),
+        key=key,
+    )
+
+
+def generate_ed25519_certificate(common_name, alternative_names=[]):
+    key = ed25519.Ed25519PrivateKey.generate()
+    return generate_certificate(
+        alternative_names=alternative_names,
+        common_name=common_name,
+        hash_algorithm=None,
+        key=key,
+    )
+
+
 def load(name):
     path = os.path.join(os.path.dirname(__file__), name)
     with open(path, "rb") as fp: