TLSChangeCipherSpec middlebox in TLS1.3 & tests

Author: gpotter2

scapy/automaton.py

@@ -698,7 +698,10 @@ class CommandMessage(AutomatonException):
     # Services
     def debug(self, lvl, msg):
         if self.debug_level >= lvl:
-            log_interactive.debug(msg)
+            if conf.interactive:
+                log_interactive.debug(msg)
+            else:
+                print(msg)
 
     def send(self, pkt):
         if self.state.state in self.interception_points:

scapy/layers/tls/automaton_cli.py

@@ -879,7 +879,7 @@ def tls13_should_add_ClientHello(self):
                 c = 0x1301
             else:
                 c = self.ciphersuite
-            p = TLS13ClientHello(ciphers=self.ciphersuite, ext=ext)
+            p = TLS13ClientHello(ciphers=c, ext=ext)
         self.add_msg(p)
         raise self.TLS13_ADDED_CLIENTHELLO()
 
@@ -1097,7 +1097,7 @@ def tls13_should_add_ClientCertificateVerify(self):
     def TLS13_ADDED_CERTIFICATEVERIFY(self):
         return self.tls13_should_add_ClientFinished()
 
-    @ATMT.condition(TLS13_PREPARE_CLIENTFLIGHT2)
+    @ATMT.condition(TLS13_PREPARE_CLIENTFLIGHT2, prio=2)
     def tls13_should_add_ClientFinished(self):
         self.add_msg(TLSFinished())
         raise self.TLS13_ADDED_CLIENTFINISHED()

scapy/layers/tls/automaton_srv.py

@@ -641,6 +641,27 @@ def tls13_ADDED_SERVERFINISHED(self):
     @ATMT.condition(tls13_ADDED_SERVERFINISHED)
     def tls13_should_send_ServerFlight1(self):
         self.flush_records()
+        raise self.tls13_HANDLED_SERVERFLIGHT1()
+
+    @ATMT.state()
+    def tls13_HANDLED_SERVERFLIGHT1(self):
+        pass
+
+    @ATMT.condition(tls13_HANDLED_SERVERFLIGHT1, prio=1)
+    def tls13_should_handle_ChangeCipherSpec(self):
+        self.raise_on_packet(TLSChangeCipherSpec,
+                             self.tls13_HANDLED_CHANGECIPHERSPEC)
+
+    @ATMT.state()
+    def tls13_HANDLED_CHANGECIPHERSPEC(self):
+        pass
+
+    @ATMT.condition(tls13_HANDLED_SERVERFLIGHT1, prio=2)
+    def tls13_missing_ChangeCipherSpec(self):
+        raise self.tls13_WAITING_CLIENTFLIGHT2()
+
+    @ATMT.condition(tls13_HANDLED_CHANGECIPHERSPEC)
+    def tls13_should_wait_ClientFlight2(self):
         raise self.tls13_WAITING_CLIENTFLIGHT2()
 
     @ATMT.state()
@@ -650,19 +671,19 @@ def tls13_WAITING_CLIENTFLIGHT2(self):
 
     @ATMT.state()
     def tls13_RECEIVED_CLIENTFLIGHT2(self):
-        print("tls13_RECEIVED_CLIENTFLIGHT2")
         pass
 
     @ATMT.condition(tls13_RECEIVED_CLIENTFLIGHT2, prio=1)
     def tls13_should_handle_ClientFlight2(self):
-        print("tls13_should_handle_ClientFinished")
+        self.raise_on_packet(TLS13Certificate,
+                             self.TLS13_HANDLED_CLIENTCERTIFICATE)
+
+    @ATMT.condition(tls13_RECEIVED_CLIENTFLIGHT2, prio=2)
+    def tls13_no_ClientCertificate(self):
         if self.client_auth:
-            print("raise_on_packet(TLS13Certificate")
-            self.raise_on_packet(TLS13Certificate,
-                                 self.TLS13_HANDLED_CLIENTCERTIFICATE)
-        else:
-            self.raise_on_packet(TLSFinished,
-                                 self.TLS13_HANDLED_CLIENTFINISHED)
+            raise self.TLS13_MISSING_CLIENTCERTIFICATE()
+        self.raise_on_packet(TLSFinished,
+                             self.TLS13_HANDLED_CLIENTFINISHED)
 
     # RFC8446, section 4.4.2.4 :
     # "If the client does not send any certificates (i.e., it sends an empty
@@ -674,9 +695,9 @@ def tls13_should_handle_ClientFlight2(self):
     def TLS13_HANDLED_CLIENTCERTIFICATE(self):
         if self.client_auth:
             self.vprint("Received client certificate chain...")
-            self.vprint(self.cur_pkt.show())
             if isinstance(self.cur_pkt, TLS13Certificate):
                 if self.cur_pkt.certslen == 0:
+                    self.vprint("but it's empty !")
                     raise self.TLS13_MISSING_CLIENTCERTIFICATE()
 
     @ATMT.condition(TLS13_HANDLED_CLIENTCERTIFICATE)
@@ -699,11 +720,15 @@ def tls13_should_handle_ClientFinished(self):
         self.raise_on_packet(TLSFinished,
                              self.TLS13_HANDLED_CLIENTFINISHED)
 
-    # TODO : change alert code
     @ATMT.state()
     def TLS13_MISSING_CLIENTCERTIFICATE(self):
         self.vprint("Missing ClientCertificate!")
-        raise self.CLOSE_NOTIFY()
+        self.add_record()
+        self.add_msg(TLSAlert(level=2, descr=0x74))
+        self.flush_records()
+        self.vprint("Sending TLSAlert 116")
+        self.socket.close()
+        raise self.WAITING_CLIENT()
 
     @ATMT.state()
     def TLS13_HANDLED_CLIENTFINISHED(self):

scapy/layers/tls/handshake.py

@@ -55,6 +55,7 @@
     from cryptography.hazmat.backends import default_backend
     from cryptography.hazmat.primitives import hashes
 
+
 ###############################################################################
 #   Generic TLS Handshake message                                             #
 ###############################################################################
@@ -295,6 +296,10 @@ def tls_session_update(self, msg_str):
 
         s = self.tls_session
         s.advertised_tls_version = self.version
+        # This ClientHello could be a 1.3 one. Let's store the sid
+        # in all cases
+        if self.sidlen and self.sidlen > 0:
+            s.sid = self.sid
         self.random_bytes = msg_str[10:38]
         s.client_random = (struct.pack('!I', self.gmt_unix_time) +
                            self.random_bytes)
@@ -306,7 +311,6 @@ def tls_session_update(self, msg_str):
             for e in self.ext:
                 if isinstance(e, TLS_Ext_SupportedVersion_CH):
                     s.advertised_tls_version = e.versions[0]
-
                 if isinstance(e, TLS_Ext_SignatureAlgorithms):
                     s.advertised_sig_algs = e.sig_algs
 
@@ -1060,6 +1064,7 @@ class TLS13CertificateRequest(_TLSHandshake):
 #   ServerHelloDone                                                           #
 ###############################################################################
 
+
 class TLSServerHelloDone(_TLSHandshake):
     name = "TLS Handshake - Server Hello Done"
     fields_desc = [ByteEnumField("msgtype", 14, _tls_handshake_type),

scapy/layers/tls/record.py

@@ -140,6 +140,8 @@ def getfield(self, pkt, s):
             if (((pkt.tls_session.tls_version or 0x0303) > 0x0200) and
                     hasattr(pkt, "type") and pkt.type == 23):
                 return ret, [TLSApplicationData(data=b"")]
+            elif hasattr(pkt, "type") and pkt.type == 20:
+                return ret, [TLSChangeCipherSpec()]
             else:
                 return ret, [Raw(load=b"")]
 

scapy/layers/tls/record_tls13.py

@@ -61,16 +61,30 @@ def pre_dissect(self, s):
         return s
 
 
+class TLSInnerChangeCipherSpec(_GenericTLSSessionInheritance):
+    __slots__ = ["type"]
+    name = "TLS Inner Plaintext (CCS)"
+    fields_desc = [_TLSMsgListField("msg", [], length_from=lambda x: 1)]
+
+    def __init__(self, _pkt=None, *args, **kwargs):
+        self.type = 0x14
+        super(TLSInnerChangeCipherSpec, self).__init__(_pkt, *args, **kwargs)
+
+
 class _TLSInnerPlaintextField(PacketField):
     def __init__(self, name, default, *args, **kargs):
         super(_TLSInnerPlaintextField, self).__init__(name,
                                                       default,
                                                       TLSInnerPlaintext)
 
     def m2i(self, pkt, m):
+        if pkt.type == 0x14:
+            return TLSInnerChangeCipherSpec(m, tls_session=pkt.tls_session)
         return self.cls(m, tls_session=pkt.tls_session)
 
     def getfield(self, pkt, s):
+        if pkt.type == 0x14:
+            return super(_TLSInnerPlaintextField, self).getfield(pkt, s)
         tag_len = pkt.tls_session.rcs.mac_len
         frag_len = pkt.len - tag_len
         if frag_len < 1:

test/run_tests_py2.bat

@@ -9,4 +9,3 @@ if [%1]==[] (
 ) else (
   python "%MYDIR%\scapy\tools\UTscapy.py" %*
 )
-PAUSE

test/run_tests_py3.bat

@@ -9,4 +9,3 @@ if [%1]==[] (
 ) else (
   python3 "%MYDIR%\scapy\tools\UTscapy.py" %*
 )
-PAUSE

test/tls/tests_tls_netaccess.uts

@@ -7,6 +7,7 @@
 ############
 ############
 + TLS server automaton tests
+~ server
 
 ### DISCLAIMER: Those tests are slow ###
 
@@ -60,15 +61,17 @@ def check_output_for_data(out, err, expected_data):
     else:
         return (False, None)
 
+def get_file(filename):
+    return os.getenv("SCAPY_ROOT_DIR")+filename if not os.path.exists(filename) else filename
+
+
 def run_tls_test_server(expected_data, q, curve=None, cookie=False, client_auth=False):
     correct = False
     print("Server started !")
     with captured_output() as (out, err):
         # Prepare automaton
-        filename = "/test/tls/pki/srv_cert.pem"
-        mycert = os.getenv("SCAPY_ROOT_DIR")+filename if not os.path.exists(filename) else filename
-        filename = "/test/tls/pki/srv_key.pem"
-        mykey = os.getenv("SCAPY_ROOT_DIR")+filename if not os.path.exists(filename) else filename
+        mycert = get_file("/test/tls/pki/srv_cert.pem")
+        mykey = get_file("/test/tls/pki/srv_key.pem")
         print(os.environ["SCAPY_ROOT_DIR"])
         print(mykey)
         print(mycert)
@@ -100,18 +103,27 @@ def test_tls_server(suite="", version="", tls13=False, client_auth=False):
     q_.get()
     time.sleep(1)
     # Run client
-    filename = "/test/tls/pki/ca_cert.pem"
-    filename = os.getenv("SCAPY_ROOT_DIR")+filename if not os.path.exists(filename) else filename
-    CA_f = os.path.abspath(filename)
+    CA_f = get_file("/test/tls/pki/ca_cert.pem")
+    mycert = get_file("/test/tls/pki/cli_cert.pem")
+    mykey = get_file("/test/tls/pki/cli_key.pem")
+    args = [
+        "openssl", "s_client",
+        "-connect", "127.0.0.1:4433", "-debug",
+        "-ciphersuites" if tls13 else "-cipher", suite,
+        version,
+        "-CAfile", CA_f
+    ]
+    if client_auth:
+        args.extend(["-cert", mycert, "-key", mykey])
     p = subprocess.Popen(
-        ["openssl", "s_client", "-connect", "127.0.0.1:4433", "-debug", "-ciphersuites" if tls13 else "-cipher", suite, version, "-CAfile", CA_f],
+        args,
         stdin=subprocess.PIPE, stdout=subprocess.PIPE, stderr=subprocess.STDOUT
     )
     msg += b"\nstop_server\n"
     out = p.communicate(input=msg)[0]
     print(out.decode())
     if p.returncode != 0:
-        raise RuntimeError("OpenSSL returned with error code")
+        raise RuntimeError("OpenSSL returned with error code %s" % p.returncode)
     else:
         p = re.compile(br'verify return:(\d+)')
         _failed = False
@@ -165,6 +177,7 @@ test_tls_server("TLS_AES_256_GCM_SHA384", "-tls1_3", tls13=True)
 test_tls_server("TLS_AES_256_GCM_SHA384", "-tls1_3", tls13=True, client_auth=True)
 
 + TLS client automaton tests
+~ client
 
 = Load client utils functions
 
@@ -177,16 +190,18 @@ from scapy.modules.six.moves.queue import Queue
 
 send_data = cipher_suite_code = version = None
 
-def run_tls_test_client(send_data=None, cipher_suite_code=None, version=None):
+def run_tls_test_client(send_data=None, cipher_suite_code=None, version=None, client_auth=False):
     print("Loading client...")
+    mycert = get_file("/test/tls/pki/cli_cert.pem") if client_auth else None
+    mykey = get_file("/test/tls/pki/cli_key.pem") if client_auth else None
     if version == "0002":
-        t = TLSClientAutomaton(data=[send_data, b"stop_server", b"quit"], version="sslv2", debug=5)
+        t = TLSClientAutomaton(data=[send_data, b"stop_server", b"quit"], version="sslv2", debug=5, mycert=mycert, mykey=mykey)
     elif version == "0304":
         ch = TLS13ClientHello(ciphers=int(cipher_suite_code, 16))
-        t = TLSClientAutomaton(client_hello=ch, data=[send_data, b"stop_server", b"quit"], version="tls13", debug=5)
+        t = TLSClientAutomaton(client_hello=ch, data=[send_data, b"stop_server", b"quit"], version="tls13", debug=5, mycert=mycert, mykey=mykey)
     else:
         ch = TLSClientHello(version=int(version, 16), ciphers=int(cipher_suite_code, 16))
-        t = TLSClientAutomaton(client_hello=ch, data=[send_data, b"stop_server", b"quit"], debug=5)
+        t = TLSClientAutomaton(client_hello=ch, data=[send_data, b"stop_server", b"quit"], debug=5, mycert=mycert, mykey=mykey)
     print("Running client...")
     t.run()
 
@@ -204,7 +219,7 @@ def test_tls_client(suite, version, curve=None, cookie=False, client_auth=False)
     time.sleep(1)
     print("Thread synchronised")
     # Run client
-    run_tls_test_client(msg, suite, version)
+    run_tls_test_client(msg, suite, version, client_auth)
     # Wait for server
     print("Client running, waiting...")
     th_.join(5)