Merge pull request #529 from tlsfuzzer/brainpool_in_tls13

add support for Brainpool curves in TLS 1.3 (RFC8734)

scripts/tls.py

@@ -109,7 +109,7 @@ def printUsage(s=None):
     [-c CERT] [-k KEY] [-t TACK] [-v VERIFIERDB] [-d DIR] [-l LABEL] [-L LENGTH]
     [--reqcert] [--param DHFILE] [--psk PSK] [--psk-ident IDENTITY]
     [--psk-sha384] [--ssl3] [--max-ver VER] [--tickets COUNT] [--cipherlist]
-    [--request-pha] [--require-pha] [--echo]
+    [--request-pha] [--require-pha] [--echo] [--groups GROUPS]
     HOST:PORT
 
   client
@@ -137,6 +137,8 @@ def printUsage(s=None):
   --require-pha - abort connection if client didn't provide certificate in
                   post-handshake authentication
   --echo - function as an echo server
+  --groups - specify what key exchange groups should be supported
+  GROUPS - comma-separated list of enabled key exchange groups
   CERT, KEY - the file with key and certificates that will be used by client or
         server. The server can accept multiple pairs of `-c` and `-k` options
         to configure different certificates (like RSA and ECDSA)
@@ -197,6 +199,7 @@ def handleArgs(argv, argString, flagsList=[]):
     request_pha = False
     require_pha = False
     echo = False
+    groups = None
 
     for opt, arg in opts:
         if opt == "-k":
@@ -270,6 +273,8 @@ def handleArgs(argv, argString, flagsList=[]):
             tickets = int(arg)
         elif opt == "--cipherlist":
             ciphers.append(arg)
+        elif opt == "--groups":
+            groups = arg.split(',')
         elif opt == "--request-pha":
             request_pha = True
         elif opt == "--require-pha":
@@ -344,6 +349,8 @@ def handleArgs(argv, argString, flagsList=[]):
         retList.append(require_pha)
     if "echo" in flagsList:
         retList.append(echo)
+    if "groups=" in flagsList:
+        retList.append(groups)
     return retList
 
 
@@ -547,12 +554,13 @@ def serverCmd(argv):
     (address, privateKey, cert_chain, virtual_hosts, tacks, verifierDB,
             directory, reqCert,
             expLabel, expLength, dhparam, psk, psk_ident, psk_hash, ssl3,
-            max_ver, tickets, cipherlist, request_pha, require_pha, echo) = \
+            max_ver, tickets, cipherlist, request_pha, require_pha, echo,
+            groups) = \
         handleArgs(argv, "kctbvdlL",
                    ["reqcert", "param=", "psk=",
                     "psk-ident=", "psk-sha384", "ssl3", "max-ver=",
                     "tickets=", "cipherlist=", "request-pha", "require-pha",
-                    "echo"])
+                    "echo", "groups="])
 
 
     if (cert_chain and not privateKey) or (not cert_chain and privateKey):
@@ -599,6 +607,17 @@ def serverCmd(argv):
     if cipherlist:
         settings.cipherNames = [item for cipher in cipherlist
                                 for item in cipher.split(',')]
+    if groups:
+        dh_groups = []
+        ecc_groups = []
+        for item in groups:
+            if "ffdh" in item:
+                dh_groups.append(item)
+            else:
+                ecc_groups.append(item)
+        settings.dhGroups = dh_groups
+        settings.eccCurves = ecc_groups
+        settings.keyShares = []
 
     class MySimpleEchoHandler(BaseRequestHandler):
         def handle(self):

tests/tlstest.py

@@ -44,7 +44,7 @@
     from xmlrpc import client as xmlrpclib
 import ssl
 from tlslite import *
-from tlslite.constants import KeyUpdateMessageType
+from tlslite.constants import KeyUpdateMessageType, SignatureScheme
 
 try:
     from tack.structures.Tack import Tack
@@ -340,6 +340,32 @@ def connect():
 
         test_no += 1
 
+    for curve, keySize, exp_sig_alg in (
+            ("brainpoolP256r1tls13", 256,
+             SignatureScheme.ecdsa_brainpoolP256r1tls13_sha256),
+            ("brainpoolP384r1tls13", 384,
+             SignatureScheme.ecdsa_brainpoolP384r1tls13_sha384),
+            ("brainpoolP512r1tls13", 512,
+             SignatureScheme.ecdsa_brainpoolP512r1tls13_sha512)):
+        print("Test {0} - Two good ECDSA certs - {1}, TLSv1.3".format(test_no, curve))
+        synchro.recv(1)
+        connection = connect()
+        settings = HandshakeSettings()
+        settings.minVersion = (3, 4)
+        settings.maxVersion = (3, 4)
+        settings.eccCurves = [curve]
+        settings.keyShares = []
+        connection.handshakeClientCert(settings=settings)
+        testConnClient(connection)
+        assert connection.serverSigAlg == exp_sig_alg, \
+            connection.serverSigAlg
+        assert isinstance(connection.session.serverCertChain, X509CertChain)
+        assert len(connection.session.serverCertChain.getEndEntityPublicKey()) \
+                == keySize
+        connection.close()
+
+        test_no += 1
+
     print("Test {0} - Two good ECDSA certs - secp256r1, TLSv1.2".format(test_no))
     synchro.recv(1)
     connection = connect()
@@ -431,7 +457,7 @@ def connect():
 
     test_no += 1
 
-    print("Test {0} - good X509 RSA and ECDSA, correct RSA and ECDSA sigalgs, RSA, TLSv1.3"
+    print("Test {0} - good X509 RSA and ECDSA, correct RSA and ECDSA sigalgs, ECDSA, TLSv1.3"
           .format(test_no))
     synchro.recv(1)
     connection = connect()
@@ -444,7 +470,7 @@ def connect():
     testConnClient(connection)
     assert isinstance(connection.session.serverCertChain, X509CertChain)
     assert connection.session.serverCertChain.getEndEntityPublicKey().key_type\
-            == "rsa"
+            == "ecdsa"
     assert connection.version == (3, 4)
     connection.close()
 
@@ -2233,6 +2259,29 @@ def connect():
 
         test_no += 1
 
+    for curve, certChain, key in (("brainpoolP256r1tls13", x509ecdsaBrainpoolP256r1Chain, x509ecdsaBrainpoolP256r1Key),
+                                  ("brainpoolP384r1tls13", x509ecdsaBrainpoolP384r1Chain, x509ecdsaBrainpoolP384r1Key),
+                                  ("brainpoolP512r1tls13", x509ecdsaBrainpoolP512r1Chain, x509ecdsaBrainpoolP512r1Key)):
+        print("Test {0} - Two good ECDSA certs - {1}, TLSv1.3".format(test_no, curve))
+        synchro.send(b'R')
+        connection = connect()
+        settings = HandshakeSettings()
+        settings.minVersion = (3, 4)
+        settings.maxVersion = (3, 4)
+        settings.eccCurves = [curve, "secp256r1"]
+        settings.keyShares = []
+        v_host = VirtualHost()
+        v_host.keys = [Keypair(x509ecdsaKey, x509ecdsaChain.x509List)]
+        settings.virtual_hosts = [v_host]
+        connection.handshakeServer(certChain=certChain,
+                                   privateKey=key, settings=settings)
+        assert connection.extendedMasterSecret
+        #XXX assert connection.session.serverCertChain == certChain
+        testConnServer(connection)
+        connection.close()
+
+        test_no += 1
+
     for curve, exp_chain in (("secp256r1", x509ecdsaChain),
                              ("secp384r1", x509ecdsaP384Chain)):
         print("Test {0} - Two good ECDSA certs - {1}, TLSv1.2"
@@ -2254,10 +2303,14 @@ def connect():
 
         test_no += 1
 
-    for tls_ver in ("TLSv1.2", "TLSv1,3"):
+    for tls_ver in ("TLSv1.2", "TLSv1.3"):
 
-        print("Test {0} - good X509 RSA and ECDSA, correct RSA and ECDSA sigalgs, RSA, {1}"
-              .format(test_no, tls_ver))
+        if tls_ver == "TLSv1.2":
+            expected = "RSA"
+        else:
+            expected = "ECDSA"
+        print("Test {0} - good X509 RSA and ECDSA, correct RSA and ECDSA sigalgs, {2}, {1}"
+              .format(test_no, tls_ver, expected))
         synchro.send(b'R')
         connection = connect()
         settings = HandshakeSettings()
@@ -2270,13 +2323,19 @@ def connect():
                                    privateKey=x509KeyRSANonCA,
                                    settings=settings)
         assert connection.extendedMasterSecret
-        assert connection.session.serverCertChain == x509ChainRSANonCA
+        if tls_ver == "TLSv1.2":
+            # because in TLS 1.2 we don't send the signature_algorithms_cert
+            # extension, but send sig_algs with PKCS#1v1.5 sigalgs, RSA can be picked
+            # in TLS 1.3 we filter out PKCS#v1.5 so RSA cert will be picked only
+            # as a fallback
+            assert connection.session.serverCertChain == x509ChainRSANonCA, connection.session.serverCertChain.getEndEntityPublicKey().key_type
+        else:
+            assert connection.session.serverCertChain == x509ChainECDSANonCA, connection.session.serverCertChain.getEndEntityPublicKey().key_type
         testConnServer(connection)
         connection.close()
 
         test_no += 1
 
-
         print("Test {0} - good X509 RSA and ECDSA, bad RSA and good ECDSA sigalgs, ECDSA, {1}"
               .format(test_no, tls_ver))
         synchro.send(b'R')

tlslite/constants.py

@@ -321,6 +321,14 @@ def getHash(scheme):
         return hName
 
 
+# set of TLS 1.3 specific schemes for Brainpool curves
+TLS_1_3_BRAINPOOL_SIG_SCHEMES = set([
+    SignatureScheme.ecdsa_brainpoolP256r1tls13_sha256,
+    SignatureScheme.ecdsa_brainpoolP384r1tls13_sha384,
+    SignatureScheme.ecdsa_brainpoolP512r1tls13_sha512,
+])
+
+
 class AlgorithmOID(TLSEnum):
     """
     Algorithm OIDs as defined in rfc5758(ecdsa),

tlslite/handshakesettings.py

@@ -32,7 +32,10 @@
 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"]
+SIGNATURE_SCHEMES = ["Ed25519", "Ed448",
+                     "ecdsa_brainpoolP512r1tls13_sha512",
+                     "ecdsa_brainpoolP384r1tls13_sha384",
+                     "ecdsa_brainpoolP256r1tls13_sha256"]
 RSA_SCHEMES = ["pss", "pkcs1"]
 CURVE_NAMES = []
 if ML_KEM_AVAILABLE:
@@ -41,9 +44,12 @@
 # while secp521r1 is the most secure, it's also much slower than the others
 # so place it as the last one
 CURVE_NAMES += ["x25519", "x448", "secp384r1", "secp256r1",
-                "secp521r1"]
-ALL_CURVE_NAMES = CURVE_NAMES + ["secp256k1", "brainpoolP512r1",
-                                 "brainpoolP384r1", "brainpoolP256r1"]
+                "secp521r1", "brainpoolP512r1",
+                "brainpoolP384r1", "brainpoolP256r1",
+                "brainpoolP256r1tls13",
+                "brainpoolP384r1tls13",
+                "brainpoolP512r1tls13"]
+ALL_CURVE_NAMES = CURVE_NAMES + ["secp256k1"]
 if ecdsaAllCurves:
     ALL_CURVE_NAMES += ["secp224r1", "secp192r1"]
 ALL_DH_GROUP_NAMES = ["ffdhe2048", "ffdhe3072", "ffdhe4096", "ffdhe6144",
@@ -62,7 +68,8 @@
                           "x25519", "x448", "ffdhe2048",
                           "ffdhe3072", "ffdhe4096", "ffdhe6144",
                           "ffdhe8192", "secp256r1mlkem768", "x25519mlkem768",
-                          "secp384r1mlkem1024"]
+                          "secp384r1mlkem1024", "brainpoolP256r1tls13",
+                          "brainpoolP384r1tls13", "brainpoolP512r1tls13"]
 KNOWN_VERSIONS = ((3, 0), (3, 1), (3, 2), (3, 3), (3, 4))
 TICKET_CIPHERS = ["chacha20-poly1305", "aes256gcm", "aes128gcm", "aes128ccm",
                   "aes128ccm_8", "aes256ccm", "aes256ccm_8"]
@@ -278,7 +285,10 @@ class HandshakeSettings(object):
     :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".
+        Currently supported are: "Ed25519", "Ed448",
+        "ecdsa_brainpoolP256r1tls13_sha256",
+        "ecdsa_brainpoolP384r1tls13_sha384",
+        "ecdsa_brainpoolP512r1tls13_sha512".
 
     :vartype eccCurves: list(str)
     :ivar eccCurves: List of named curves that are to be advertised as
@@ -534,7 +544,7 @@ def _sanityCheckECDHSettings(other):
                              .format(unknownDHGroup))
 
         # TLS 1.3 limits the allowed groups (RFC 8446,ch. 4.2.7.)
-        if other.maxVersion == (3, 4):
+        if (3, 3) not in other.versions and (3, 4) in other.versions:
             forbiddenGroup = HandshakeSettings._not_matching(other.eccCurves, TLS13_PERMITTED_GROUPS)
             if forbiddenGroup:
                 raise ValueError("The following enabled groups are forbidden in TLS 1.3: {0}"