send CCS only when client asks, don't allow CCS post handshake

Author: tomato42

tlslite/tlsconnection.py

@@ -734,7 +734,9 @@ def _clientSendClientHello(self, settings, session, srpUsername,
                     create(bytearray(b'')))
                 self._client_keypair = certParams
 
+            # fake session_id for middlebox compatibility mode
             session_id = getRandomBytes(32)
+
             extensions.append(SupportedVersionsExtension().
                               create(settings.versions))
 
@@ -964,6 +966,13 @@ def _clientGetServerHello(self, settings, session, clientHello):
                                               "update to Client Hello"):
                     yield result
 
+            if clientHello.session_id != hello_retry.session_id:
+                for result in self._sendError(
+                        AlertDescription.illegal_parameter,
+                        "Received HRR session_id does not match the one in "
+                        "ClientHello"):
+                    yield result
+
             ext = clientHello.getExtension(ExtensionType.pre_shared_key)
             if ext:
                 # move the extension to end (in case extension like cookie was
@@ -979,8 +988,12 @@ def _clientGetServerHello(self, settings, session, clientHello):
                                                 if session else None)
 
             # resend the client hello with performed changes
-            ccs = ChangeCipherSpec().create()
-            for result in self._sendMsgs([ccs, clientHello]):
+            msgs = []
+            if clientHello.session_id:
+                ccs = ChangeCipherSpec().create()
+                msgs.append(ccs)
+            msgs.append(clientHello)
+            for result in self._sendMsgs(msgs):
                 yield result
             self._ccs_sent = True
 
@@ -1023,6 +1036,13 @@ def _clientGetServerHello(self, settings, session, clientHello):
                     "Too new version: {0} (max: {1})"
                     .format(real_version, settings.maxVersion)):
                 yield result
+        if real_version > (3, 3) and \
+                serverHello.session_id != clientHello.session_id:
+            for result in self._sendError(
+                    AlertDescription.illegal_parameter,
+                    "Received ServerHello session_id does not match the one "
+                    "in ClientHello"):
+                yield result
         cipherSuites = CipherSuite.filterForVersion(clientHello.cipher_suites,
                                                     minVersion=real_version,
                                                     maxVersion=real_version)
@@ -1460,7 +1480,7 @@ def _clientTLS13Handshake(self, settings, session, clientHello,
         cl_finished = Finished(self.version, prf_size)
         cl_finished.create(cl_verify_data)
 
-        if not self._ccs_sent:
+        if not self._ccs_sent and clientHello.session_id:
             ccs = ChangeCipherSpec().create()
             msgs = [ccs, cl_finished]
         else:
@@ -1469,6 +1489,8 @@ def _clientTLS13Handshake(self, settings, session, clientHello,
         for result in self._sendMsgs(msgs):
             yield result
 
+        # CCS messages are not allowed in post handshake authentication
+        self._middlebox_compat_mode = False
 
         # fully switch to application data
         self._changeWriteState()
@@ -2637,7 +2659,7 @@ def _serverTLS13Handshake(self, settings, clientHello, cipherSuite,
 
         msgs = []
         msgs.append(serverHello)
-        if not self._ccs_sent:
+        if not self._ccs_sent and clientHello.session_id:
             ccs = ChangeCipherSpec().create()
             msgs.append(ccs)
         for result in self._sendMsgs(msgs):
@@ -2888,6 +2910,9 @@ def _serverTLS13Handshake(self, settings, clientHello, cipherSuite,
                     "Finished value is not valid"):
                 yield result
 
+        # disallow CCS messages after handshake
+        self._middlebox_compat_mode = False
+
         resumption_master_secret = derive_secret(secret,
                                                  bytearray(b'res master'),
                                                  self._handshake_hash,
@@ -3647,8 +3672,11 @@ def _serverGetClientHello(self, settings, private_key, cert_chain,
                 hrr.create((3, 3), TLS_1_3_HRR, clientHello.session_id,
                            cipherSuite, extensions=hrr_ext)
 
-                ccs = ChangeCipherSpec().create()
-                for result in self._sendMsgs([hrr, ccs]):
+                msgs = [hrr]
+                if clientHello.session_id:
+                    ccs = ChangeCipherSpec().create()
+                    msgs.append(ccs)
+                for result in self._sendMsgs(msgs):
                     yield result
                 self._ccs_sent = True
 

tlslite/tlsrecordlayer.py

@@ -202,6 +202,11 @@ def __init__(self, sock):
         # doesn't provide a certificate
         self.client_cert_required = False
 
+        # boolean to control if ChangeCipherSpec in TLS1.3 is allowed or not
+        # we start with True, as peers MUST always processe the messages
+        # before the handshake is done, only then we disable it (for PHA)
+        self._middlebox_compat_mode = True
+
     @property
     def _send_record_limit(self):
         """Maximum size of payload that can be sent."""
@@ -1006,6 +1011,7 @@ def _getMsg(self, expectedType, secondaryType=None, constructorType=None):
                 # continue
                 if self.version > (3, 3) and \
                         ContentType.handshake in expectedType and \
+                        self._middlebox_compat_mode and \
                         recordHeader.type == ContentType.change_cipher_spec:
                     ccs = ChangeCipherSpec().parse(p)
                     if ccs.type != 1:

unit_tests/test_tlslite_tlsconnection.py

@@ -14,14 +14,17 @@
 
 from tlslite.recordlayer import RecordLayer
 from tlslite.messages import ServerHello, ClientHello, Alert, RecordHeader3
-from tlslite.constants import CipherSuite, AlertDescription, ContentType
+from tlslite.constants import CipherSuite, AlertDescription, ContentType, \
+    GroupName, TLS_1_3_HRR
 from tlslite.tlsconnection import TLSConnection
 from tlslite.errors import TLSLocalAlert, TLSRemoteAlert
 from tlslite.x509 import X509
 from tlslite.x509certchain import X509CertChain
 from tlslite.utils.keyfactory import parsePEMKey
 from tlslite.handshakesettings import HandshakeSettings
 from tlslite.session import Session
+from tlslite.extensions import SrvSupportedVersionsExtension, \
+    ServerKeyShareExtension, KeyShareEntry, HRRKeyShareExtension
 
 from unit_tests.mocksock import MockSocket
 
@@ -205,6 +208,86 @@ def test_server_with_client_not_using_required_EMS(self):
         self.assertEqual(err.exception.description,
                          AlertDescription.insufficient_security)
 
+    def test_client_with_server_responing_with_wrong_session_id_in_TLS1_3(self):
+        # socket to generate the faux response
+        gen_sock = MockSocket(bytearray(0))
+
+        gen_record_layer = RecordLayer(gen_sock)
+        gen_record_layer.version = (3, 3)
+
+        srv_ext = []
+        srv_ext.append(SrvSupportedVersionsExtension().create((3, 4)))
+        srv_ext.append(ServerKeyShareExtension().create(
+            KeyShareEntry().create(
+            GroupName.secp256r1, bytearray(b'\x03' + b'\x01' * 32))))
+
+        server_hello = ServerHello().create(
+                version=(3, 3),
+                random=bytearray(32),
+                session_id=bytearray(b"test"),
+                cipher_suite=CipherSuite.TLS_AES_128_GCM_SHA256,
+                certificate_type=None,
+                tackExt=None,
+                next_protos_advertised=None,
+                extensions=srv_ext)
+
+        for res in gen_record_layer.sendRecord(server_hello):
+            if res in (0, 1):
+                self.assertTrue(False, "Blocking socket")
+            else:
+                break
+
+        # test proper
+        sock = MockSocket(gen_sock.sent[0])
+
+        conn = TLSConnection(sock)
+
+        with self.assertRaises(TLSLocalAlert) as err:
+            conn.handshakeClientCert()
+
+        self.assertEqual(err.exception.description,
+                         AlertDescription.illegal_parameter)
+
+    def test_client_with_server_responing_with_wrong_session_id_in_TLS1_3_HRR(self):
+        # socket to generate the faux response
+        gen_sock = MockSocket(bytearray(0))
+
+        gen_record_layer = RecordLayer(gen_sock)
+        gen_record_layer.version = (3, 3)
+
+        srv_ext = []
+        srv_ext.append(SrvSupportedVersionsExtension().create((3, 4)))
+        srv_ext.append(HRRKeyShareExtension().create(
+            GroupName.secp521r1))
+
+        server_hello = ServerHello().create(
+                version=(3, 3),
+                random=TLS_1_3_HRR,
+                session_id=bytearray(b"test"),
+                cipher_suite=CipherSuite.TLS_AES_128_GCM_SHA256,
+                certificate_type=None,
+                tackExt=None,
+                next_protos_advertised=None,
+                extensions=srv_ext)
+
+        for res in gen_record_layer.sendRecord(server_hello):
+            if res in (0, 1):
+                self.assertTrue(False, "Blocking socket")
+            else:
+                break
+
+        # test proper
+        sock = MockSocket(gen_sock.sent[0])
+
+        conn = TLSConnection(sock)
+
+        with self.assertRaises(TLSLocalAlert) as err:
+            conn.handshakeClientCert()
+
+        self.assertEqual(err.exception.description,
+                         AlertDescription.illegal_parameter)
+
+
     def prepare_mock_socket_with_handshake_failure(self):
         alertObj = Alert().create(AlertDescription.handshake_failure)
         alert = alertObj.write()