Merge branch '119-tls-pqc-named-curve-checker'

Closes: #119

Author: c0r0n3r

cryptolyzer/tls/client.py

@@ -108,7 +108,7 @@ def key_share_entry_from_named_curve(named_curve):
             get_ecdh_ephemeral_key_forged(named_curve.value.named_group)
         )
 
-    if named_curve.value.named_group.value.group_type == NamedGroupType.DH_PARAM:
+    if named_curve.value.named_group.value.group_type == NamedGroupType.FINITE_FIELD:
         well_known_dh_param = NAMED_CURVE_TO_RFC7919_WELL_KNOWN[named_curve]
         return TlsKeyShareEntry(
             named_curve,
@@ -353,7 +353,7 @@ class TlsHandshakeClientHelloKeyExchangeDHE(  # pylint: disable=too-many-ancesto
         named_curve
         for named_curve in TlsNamedCurve
         if (named_curve.value.named_group is not None
-            and named_curve.value.named_group.value.group_type == NamedGroupType.DH_PARAM)
+            and named_curve.value.named_group.value.group_type == NamedGroupType.FINITE_FIELD)
     ]
 
     def __init__(
@@ -386,10 +386,9 @@ class TlsHandshakeClientHelloKeyExchangeECDHx(  # pylint: disable=too-many-ances
             TlsProtocolVersion(cipher_suite.value.initial_version) > TlsProtocolVersion(TlsVersion.TLS1_2))
     ]
     _NAMED_CURVES = [
-        named_curve
-        for named_curve in TlsNamedCurve
-        if (named_curve.value.named_group is not None
-            and named_curve.value.named_group.value.group_type == NamedGroupType.ELLIPTIC_CURVE)
+        curve
+        for curve, group in map(lambda curve: (curve, curve.value.named_group), TlsNamedCurve)
+        if (group is not None and group.value.group_type in (NamedGroupType.ELLIPTIC_CURVE, NamedGroupType.HYBRID_PQS))
     ]
 
     def __init__(

submodules/cryptoparser

@@ -90,11 +90,20 @@ def test_curves(self):
         result = self.get_result('www.cloudflare.com', 443, TlsProtocolVersion(TlsVersion.TLS1_3))
         self.assertEqual(
             result.curves,
-            [TlsNamedCurve.X25519, TlsNamedCurve.SECP256R1, TlsNamedCurve.SECP384R1, TlsNamedCurve.SECP521R1, ]
+            [
+                TlsNamedCurve.X25519_KYBER_512_DRAFT00,
+                TlsNamedCurve.X25519_KYBER_768_DRAFT00,
+                TlsNamedCurve.X25519,
+                TlsNamedCurve.SECP256R1,
+                TlsNamedCurve.SECP384R1,
+                TlsNamedCurve.SECP521R1,
+            ]
         )
         self.assertTrue(result.extension_supported)
         self.assertEqual(
             self.pop_log_lines(), [
+                'Server offers elliptic-curve X25519_KYBER_512_DRAFT00',
+                'Server offers elliptic-curve X25519_KYBER_768_DRAFT00',
                 'Server offers elliptic-curve CURVE25519',
                 'Server offers elliptic-curve PRIME256V1',
                 'Server offers elliptic-curve SECP384R1',
@@ -111,6 +120,8 @@ def test_tls_1_3(self):
         self.assertEqual(
             self.get_result('www.cloudflare.com', 443, TlsProtocolVersion(TlsVersion.TLS1_3)).curves,
             [
+                TlsNamedCurve.X25519_KYBER_512_DRAFT00,
+                TlsNamedCurve.X25519_KYBER_768_DRAFT00,
                 TlsNamedCurve.X25519,
                 TlsNamedCurve.SECP256R1,
                 TlsNamedCurve.SECP384R1,
@@ -119,13 +130,28 @@ def test_tls_1_3(self):
         )
         self.assertEqual(
             self.get_log_lines(), [
+                'Server offers elliptic-curve X25519_KYBER_512_DRAFT00',
+                'Server offers elliptic-curve X25519_KYBER_768_DRAFT00',
                 'Server offers elliptic-curve CURVE25519',
                 'Server offers elliptic-curve PRIME256V1',
                 'Server offers elliptic-curve SECP384R1',
                 'Server offers elliptic-curve SECP521R1',
             ]
         )
 
+    def test_pqc(self):
+        self.assertEqual(
+            self.get_result('pq.cloudflareresearch.com', 443, TlsProtocolVersion(TlsVersion.TLS1_3)).curves,
+            [
+                TlsNamedCurve.X25519_KYBER_512_DRAFT00,
+                TlsNamedCurve.X25519_KYBER_768_DRAFT00,
+                TlsNamedCurve.X25519,
+                TlsNamedCurve.SECP256R1,
+                TlsNamedCurve.SECP384R1,
+                TlsNamedCurve.SECP521R1,
+            ]
+        )
+
     def test_plain_text_response(self):
         threaded_server = L7ServerTlsTest(
             L7ServerTlsPlainTextResponse('localhost', 0, timeout=0.5),