tls: add optional server certificate verification

scapy/layers/tls/automaton.py

@@ -93,6 +93,9 @@ def parse_args(self, mycert=None, mykey=None, **kargs):
         else:
             self.mykey = None
 
+        # Output of the automaton (final status)
+        self.final_reason = None
+
     def get_next_msg(self, socket_timeout=2, retry=2):
         """
         The purpose of the function is to make next message(s) available in

scapy/layers/tls/automaton_cli.py

@@ -47,6 +47,7 @@
 from scapy.error import warning
 from scapy.layers.tls.automaton import _TLSAutomaton
 from scapy.layers.tls.basefields import _tls_version, _tls_version_options
+from scapy.layers.tls.cert import CertList, CertTree
 from scapy.layers.tls.session import tlsSession
 from scapy.layers.tls.extensions import (
     ServerName,
@@ -97,6 +98,9 @@ class TLSClientAutomaton(_TLSAutomaton):
     :param server: the server IP or hostname. defaults to 127.0.0.1
     :param dport: the server port. defaults to 4433
     :param server_name: the SNI to use. It does not need to be set
+    :param cafile: optional CA certificate bundle used to authenticate the server.
+        By default, the system trust store is used (if verify_server is set to True).
+    :param verify_server: whether to verify the server certificate. False by default.
     :param mycert:
     :param mykey: may be provided as filenames. They will be used in the (or post)
         handshake, should the server ask for client authentication.
@@ -116,6 +120,7 @@ class TLSClientAutomaton(_TLSAutomaton):
     """
 
     def parse_args(self, server="127.0.0.1", dport=4433, server_name=None,
+                   cafile=None, verify_server=False,
                    mycert=None, mykey=None,
                    client_hello=None, version=None,
                    resumption_master_secret=None,
@@ -137,6 +142,15 @@ def parse_args(self, server="127.0.0.1", dport=4433, server_name=None,
         self.remote_ip = tmp[0][4][0]
         self.remote_port = dport
         self.server_name = server_name
+        self.expected_server_name = server_name or server
+        print(verify_server)
+        self.verify_server = verify_server or bool(cafile)
+        if self.verify_server and cafile:
+            self.server_trust_store = CertList(cafile)
+        elif self.verify_server:
+            self.server_trust_store = CertList.load_system_store()
+        else:
+            self.server_trust_store = None
         self.local_ip = None
         self.local_port = None
         self.socket = None
@@ -400,9 +414,31 @@ def should_handle_ServerCertificate(self):
                                  self.HANDLED_SERVERCERTIFICATE)
         raise self.HANDLED_SERVERCERTIFICATE()
 
+    def _verify_server_cert(self):
+        if self.verify_server:
+            raise self.INVALID_SERVER_CERTIFICATE()
+            try:
+                CertTree(
+                    self.cur_session.server_certs,
+                    self.server_trust_store,
+                ).verify(
+                    self.cur_session.server_certs[0],
+                    hostname=self.server_name or self.remote_ip,
+                )
+            except ValueError:
+                raise self.INVALID_SERVER_CERTIFICATE()
+
     @ATMT.state()
     def HANDLED_SERVERCERTIFICATE(self):
-        pass
+        self._verify_server_cert()
+
+    @ATMT.state()
+    def INVALID_SERVER_CERTIFICATE(self):
+        self.vprint("Server certificate verification failed!")
+        self.add_record()
+        self.add_msg(TLSAlert(level=2, descr=46))
+        self.flush_records()
+        raise self.FINAL(reason=self.INVALID_SERVER_CERTIFICATE)
 
     @ATMT.condition(HANDLED_SERVERHELLO, prio=2)
     def missing_ServerCertificate(self):
@@ -429,9 +465,9 @@ def should_handle_ServerKeyExchange_from_ServerCertificate(self):
         self.raise_on_packet(TLSServerKeyExchange,
                              self.HANDLED_SERVERKEYEXCHANGE)
 
-    @ATMT.state(final=True)
+    @ATMT.state()
     def MISSING_SERVERKEYEXCHANGE(self):
-        pass
+        raise self.FINAL(reason=self.MISSING_SERVERKEYEXCHANGE)
 
     @ATMT.condition(HANDLED_SERVERCERTIFICATE, prio=2)
     def missing_ServerKeyExchange(self):
@@ -798,7 +834,7 @@ def close_session(self):
             self.flush_records()
         except Exception:
             self.vprint("Could not send termination Alert, maybe the server stopped?")  # noqa: E501
-        raise self.FINAL()
+        raise self.FINAL(reason=self.CLOSE_NOTIFY)
 
     #                          SSLv2 handshake                                #
 
@@ -842,7 +878,10 @@ def sslv2_should_handle_ServerHello(self):
 
     @ATMT.state()
     def SSLv2_HANDLED_SERVERHELLO(self):
-        pass
+        try:
+            self._verify_server_cert()
+        except self.INVALID_SERVER_CERTIFICATE:
+            raise self.SSLv2_CLOSE_NOTIFY()
 
     @ATMT.condition(SSLv2_RECEIVED_SERVERHELLO, prio=2)
     def sslv2_missing_ServerHello(self):
@@ -922,7 +961,7 @@ def sslv2_should_add_ClientFinished_from_NoServerVerify(self):
     def sslv2_missing_ServerVerify(self):
         raise self.SSLv2_MISSING_SERVERVERIFY()
 
-    @ATMT.state(final=True)
+    @ATMT.state()
     def SSLv2_MISSING_SERVERVERIFY(self):
         self.vprint("Missing SSLv2 ServerVerify message!")
         raise self.SSLv2_CLOSE_NOTIFY()
@@ -1092,7 +1131,7 @@ def sslv2_close_session(self):
         except Exception:
             self.vprint("Could not send our goodbye. The server probably stopped.")  # noqa: E501
         self.socket.close()
-        raise self.FINAL()
+        raise self.FINAL(reason=self.SSLv2_CLOSE_NOTIFY)
 
     #                         TLS 1.3 handshake                               #
 
@@ -1347,7 +1386,7 @@ def tls13_should_handle_Certificate(self):
 
     @ATMT.state()
     def TLS13_HANDLED_CERTIFICATE(self):
-        pass
+        self._verify_server_cert()
 
     @ATMT.condition(TLS13_HANDLED_CERTIFICATE, prio=1)
     def tls13_should_handle_CertificateVerify(self):
@@ -1370,7 +1409,7 @@ def TLS13_INVALID_CERTIFICATE_VERIFY(self):
         self.add_record()
         self.add_msg(TLSAlert(level=2, descr=51))
         self.flush_records()
-        raise self.FINAL()
+        raise self.FINAL(reason=self.TLS13_INVALID_CERTIFICATE_VERIFY)
 
     @ATMT.condition(TLS13_HANDLED_CERTIFICATE_VERIFY, prio=1)
     def tls13_should_handle_finished(self):
@@ -1466,7 +1505,7 @@ def TLS13_SENT_CLIENTFLIGHT2(self):
 
     @ATMT.state()
     def SOCKET_CLOSED(self):
-        raise self.FINAL()
+        raise self.FINAL(reason=self.SOCKET_CLOSED)
 
     @ATMT.state(stop=True)
     def STOP(self):
@@ -1477,10 +1516,12 @@ def STOP(self):
             raise self.CLOSE_NOTIFY()
 
     @ATMT.state(final=True)
-    def FINAL(self):
+    def FINAL(self, reason=None):
         # We might call shutdown, but it may happen that the server
         # did not wait for us to shutdown after answering our data query.
         # self.socket.shutdown(1)
         self.vprint("Closing client socket...")
         self.socket.close()
+        if reason is not None:
+            self.final_reason = reason
         self.vprint("Ending TLS client automaton.")

scapy/layers/tls/automaton_srv.py

@@ -30,7 +30,7 @@
 from scapy.automaton import ATMT
 from scapy.error import warning
 from scapy.layers.tls.automaton import _TLSAutomaton
-from scapy.layers.tls.cert import PrivKeyRSA, PrivKeyECDSA, PrivKeyEdDSA
+from scapy.layers.tls.cert import PrivKeyRSA, PrivKeyECDSA, PrivKeyEdDSA, CertList
 from scapy.layers.tls.basefields import _tls_version
 from scapy.layers.tls.session import tlsSession
 from scapy.layers.tls.crypto.groups import _tls_named_groups
@@ -86,7 +86,8 @@
 class TLSServerAutomaton(_TLSAutomaton):
     """
     A simple TLS test server automaton. Try to overload some states or
-    conditions and see what happens on the other side.
+    conditions and see what happens on the other side. This server
+    can only serve a SINGLE CLIENT at a time.
 
     Because of socket and automaton limitations, for now, the best way to
     interrupt the server is by sending him 'stop_server'. Interruptions with
@@ -96,15 +97,17 @@ class TLSServerAutomaton(_TLSAutomaton):
     message in a SSLv2 version, he will close the client session with a
     similar message, and start waiting for new client connections.
 
-    _'mycert' and 'mykey' may be provided as filenames. They are needed for any
-    server authenticated handshake.
-    _'preferred_ciphersuite' allows the automaton to choose a cipher suite when
-    offered in the ClientHello. If absent, another one will be chosen.
-    _'client_auth' means the client has to provide a certificate.
-    _'is_echo_server' means that everything received will be sent back.
-    _'max_client_idle_time' is the maximum silence duration from the client.
-    Once this limit has been reached, the client (if still here) is dropped,
-    and we wait for a new connection.
+    :param mycert: the Cert or path to certificate
+    :param mykey: the PrivKey or path to private key
+    :param preferred_ciphersuite: a ciphersuite to prefer in the client hello
+    :param client_auth: whether the client has to provide a certificate
+
+    Server behavior:
+
+    :param is_echo_server: means that everything received will be sent back
+    :param max_client_idle_time: is the maximum silence duration from the client.
+        Once this limit has been reached, the client (if still here) is dropped,
+        and we wait for a new connection.
     """
 
     def parse_args(self, server="127.0.0.1", sport=4433,
@@ -134,6 +137,7 @@ def parse_args(self, server="127.0.0.1", sport=4433,
         except Exception:
             tmp = socket.getaddrinfo(socket.getfqdn(server), sport)
 
+        self.closed = False
         self.serversocket = None
         self.ip_family = tmp[0][0]
         self.local_ip = tmp[0][4][0]
@@ -232,6 +236,12 @@ def http_sessioninfo(self):
         answer = (header + body) % len(body)
         return answer
 
+    def stop(self):
+        # We shutdown the server socket so that WAITING_CLIENT exits.
+        if not self.closed:
+            self.serversocket.shutdown(socket.SHUT_RDWR)
+        return super(TLSServerAutomaton, self).stop()
+
     @ATMT.state(initial=True)
     def INITIAL(self):
         self.vprint("Starting TLS server automaton.")
@@ -257,7 +267,7 @@ def BIND(self):
             self.vprint(m)
             self.vprint("Maybe some server is already listening there?")
             self.vprint()
-            raise self.FINAL()
+            raise self.FINAL(reason=self.BIND)
         raise self.WAITING_CLIENT()
 
     @ATMT.state()
@@ -291,7 +301,7 @@ def INIT_TLS_SESSION(self):
         every server_key with both server_rsa_key and server_ecdsa_key.
         """
         self.cur_session = tlsSession(connection_end="server")
-        self.cur_session.server_certs = [self.mycert]
+        self.cur_session.server_certs = CertList([self.mycert])
         self.cur_session.server_key = self.mykey
         if isinstance(self.mykey, PrivKeyRSA):
             self.cur_session.server_rsa_key = self.mykey
@@ -1209,7 +1219,7 @@ def close_session_final(self):
         # We might call shutdown, but unit tests with s_client fail with this
         # self.socket.shutdown(1)
         self.socket.close()
-        raise self.FINAL()
+        raise self.FINAL(reason=self.CLOSE_NOTIFY_FINAL)
 
     #                          SSLv2 handshake                                #
 
@@ -1511,10 +1521,13 @@ def sslv2_close_session_final(self):
         except Exception:
             self.vprint("Could not send our goodbye. The client probably left.")  # noqa: E501
         self.socket.close()
-        raise self.FINAL()
+        raise self.FINAL(reason=self.SSLv2_CLOSE_NOTIFY_FINAL)
 
     @ATMT.state(stop=True, final=True)
-    def FINAL(self):
+    def FINAL(self, reason=None):
         self.vprint("Closing server socket...")
         self.serversocket.close()
+        self.closed = True
+        if reason is not None:
+            self.final_reason = reason
         self.vprint("Ending TLS server automaton.")