Merge pull request #333 from simo5/tls13clicert

Add support for CertificateRequest (TLS 1.3)

Author: tomato42

tests/tlstest.py

@@ -416,6 +416,32 @@ def connect():
     settings.maxVersion = (3, 3)
     connection.handshakeClientCert(x509Chain, x509Key, settings=settings)
     testConnClient(connection)
+    assert isinstance(connection.session.serverCertChain, X509CertChain)
+    connection.close()
+
+    test_no += 1
+
+    print("Test {0} - good mutual X509, TLSv1.3 no certs".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 isinstance(connection.session.serverCertChain, X509CertChain)
+    connection.close()
+
+    test_no += 1
+
+    print("Test {0} - good mutual X509, TLSv1.3".format(test_no))
+    synchro.recv(1)
+    connection = connect()
+    settings = HandshakeSettings()
+    settings.minVersion = (3,4)
+    settings.maxVersion = (3,4)
+    connection.handshakeClientCert(x509Chain, x509Key, settings=settings)
+    testConnClient(connection)
     assert(isinstance(connection.session.serverCertChain, X509CertChain))
     connection.close()
 
@@ -857,6 +883,37 @@ def connect():
 
     test_no += 1
 
+    print("Test {0} - resumption in TLSv1.3 with mutual X509".format(test_no))
+    synchro.recv(1)
+    connection = connect()
+    settings = HandshakeSettings()
+    settings.minVersion = (3,4)
+    # force HRR
+    settings.keyShares = []
+    connection.handshakeClientCert(x509Chain, x509Key, serverName=address[0],
+                                   settings=settings)
+    testConnClient(connection)
+    assert isinstance(connection.session.serverCertChain, X509CertChain)
+    assert connection.session.serverName == address[0]
+    assert not connection.resumed
+    assert connection.session.tickets
+    connection.close()
+    session = connection.session
+
+    # resume
+    synchro.recv(1)
+    settings = HandshakeSettings()
+    settings.minVersion = (3,4)
+    settings.keyShares = []
+    connection = connect()
+    connection.handshakeClientCert(x509Chain, x509Key, serverName=address[0], session=session,
+                                   settings=settings)
+    testConnClient(connection)
+    assert connection.resumed
+    connection.close()
+
+    test_no += 1
+
     print("Test {0} - Heartbeat extension response callback in TLSv1.2".format(test_no))
     heartbeat_payload = os.urandom(50)
     def heartbeat_response_check(message):
@@ -1312,6 +1369,32 @@ def connect():
 
     test_no += 1
 
+    print("Test {0} - good mutual X.509, TLSv1.3 no certs".format(test_no))
+    synchro.send(b'R')
+    connection = connect()
+    settings = HandshakeSettings()
+    settings.minVersion = (3,4)
+    settings.maxVersion = (3,4)
+    connection.handshakeServer(certChain=x509Chain, privateKey=x509Key, reqCert=True, settings=settings)
+    testConnServer(connection)
+    assert not connection.session.clientCertChain
+    connection.close()
+
+    test_no += 1
+
+    print("Test {0} - good mutual X.509, TLSv1.3".format(test_no))
+    synchro.send(b'R')
+    connection = connect()
+    settings = HandshakeSettings()
+    settings.minVersion = (3,4)
+    settings.maxVersion = (3,4)
+    connection.handshakeServer(certChain=x509Chain, privateKey=x509Key, reqCert=True, settings=settings)
+    testConnServer(connection)
+    assert isinstance(connection.session.clientCertChain, X509CertChain)
+    connection.close()
+
+    test_no += 1
+
     print("Test {0} - good mutual X.509, TLSv1.1".format(test_no))
     synchro.send(b'R')
     connection = connect()
@@ -1660,6 +1743,7 @@ def server_bind(self):
     synchro.send(b'R')
     connection = connect()
     settings = HandshakeSettings()
+    settings.minVersion = (3,4)
     settings.ticketKeys = [getRandomBytes(32)]
     connection.handshakeServer(certChain=x509Chain, privateKey=x509Key,
                                settings=settings)
@@ -1676,6 +1760,28 @@ def server_bind(self):
 
     test_no += 1
 
+    print("Test {0} - resumption in TLSv1.3 with mutual X509".format(test_no))
+    synchro.send(b'R')
+    connection = connect()
+    settings = HandshakeSettings()
+    settings.minVersion = (3,4)
+    settings.ticketKeys = [getRandomBytes(32)]
+    connection.handshakeServer(certChain=x509Chain, privateKey=x509Key,
+                               reqCert=True, settings=settings)
+    testConnServer(connection)
+    connection.close()
+
+    # resume
+    synchro.send(b'R')
+    connection = connect()
+    connection.handshakeServer(certChain=x509Chain, privateKey=x509Key,
+                               reqCert=True, settings=settings)
+    testConnServer(connection)
+    assert connection.session.clientCertChain
+    connection.close()
+
+    test_no += 1
+
     print("Test {0} - Heartbeat extension response callback in TLSv1.2".format(test_no))
     heartbeat_payload = os.urandom(50)
     def heartbeat_response_check(message):

tlslite/constants.py

@@ -203,6 +203,7 @@ class SignatureScheme(TLSEnum):
     """
 
     rsa_pkcs1_sha1 = (2, 1)
+    rsa_pkcs1_sha224 = (3, 1)
     rsa_pkcs1_sha256 = (4, 1)
     rsa_pkcs1_sha384 = (5, 1)
     rsa_pkcs1_sha512 = (6, 1)

tlslite/keyexchange.py

@@ -17,7 +17,7 @@
         getPointByteSize
 from .utils.rsakey import RSAKey
 from .utils.cryptomath import bytesToNumber, getRandomBytes, powMod, \
-        numBits, numberToByteArray, divceil, numBytes
+        numBits, numberToByteArray, divceil, numBytes, secureHash
 from .utils.lists import getFirstMatching
 from .utils import tlshashlib as hashlib
 from .utils.x25519 import x25519, x448, X25519_G, X448_G, X25519_ORDER_SIZE, \
@@ -200,7 +200,8 @@ def verifyServerKeyExchange(serverKeyExchange, publicKey, clientRandom,
 
     @staticmethod
     def calcVerifyBytes(version, handshakeHashes, signatureAlg,
-                        premasterSecret, clientRandom, serverRandom):
+                        premasterSecret, clientRandom, serverRandom,
+                        prf_name = None, peer_tag=b'client'):
         """Calculate signed bytes for Certificate Verify"""
         if version == (3, 0):
             masterSecret = calcMasterSecret(version,
@@ -222,6 +223,17 @@ def calcVerifyBytes(version, handshakeHashes, signatureAlg,
             verifyBytes = handshakeHashes.digest(hashName)
             if padding == 'pkcs1':
                 verifyBytes = RSAKey.addPKCS1Prefix(verifyBytes, hashName)
+        elif version == (3, 4):
+            scheme = SignatureScheme.toRepr(signatureAlg)
+            hash_name = SignatureScheme.getHash(scheme)
+            verifyBytes = bytearray(b'\x20' * 64 +
+                                    b'TLS 1.3, ' + peer_tag +
+                                    b' CertificateVerify' +
+                                    b'\x00') + \
+                          handshakeHashes.digest(prf_name)
+            verifyBytes = secureHash(verifyBytes, hash_name)
+        else:
+            raise ValueError("Unsupported TLS version {}".format(version))
         return verifyBytes
 
     @staticmethod