tls: add optional server certificate verification

scapy/layers/tls/automaton_cli.py

@@ -38,6 +38,8 @@
 
 import socket
 import binascii
+import ipaddress
+import ssl
 import struct
 import time
 
@@ -80,12 +82,53 @@
 from scapy.packet import Raw
 from scapy.compat import bytes_encode
 
+if conf.crypto_valid:
+    from cryptography import x509
+    try:
+        from cryptography.x509.verification import PolicyBuilder, Store
+    except ImportError:
+        PolicyBuilder = Store = None
+
 # Typing imports
 from typing import (
     Optional,
 )
 
 
+def _load_trust_anchors(cafile):
+    if not conf.crypto_valid or PolicyBuilder is None:
+        return []
+    context = ssl.create_default_context(cafile=cafile)
+    return [
+        x509.load_der_x509_certificate(der)
+        for der in context.get_ca_certs(binary_form=True)
+    ]
+
+
+def _verify_server_certificate(certificates, trusted_certs, hostname):
+    if (not certificates or not trusted_certs or not conf.crypto_valid or
+            PolicyBuilder is None):
+        return False
+    try:
+        try:
+            subject = x509.IPAddress(ipaddress.ip_address(hostname))
+        except ValueError:
+            subject = x509.DNSName(hostname)
+        verifier = PolicyBuilder().store(
+            Store(trusted_certs)
+        ).build_server_verifier(subject)
+        verifier.verify(
+            x509.load_der_x509_certificate(certificates[0].der),
+            [
+                x509.load_der_x509_certificate(cert.der)
+                for cert in certificates[1:]
+            ],
+        )
+        return True
+    except Exception:
+        return False
+
+
 class TLSClientAutomaton(_TLSAutomaton):
     """
     A simple TLS test client automaton. Try to overload some states or
@@ -97,6 +140,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.
+    :param verify: whether to authenticate the server certificate. Defaults to True.
     :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 +162,7 @@ class TLSClientAutomaton(_TLSAutomaton):
     """
 
     def parse_args(self, server="127.0.0.1", dport=4433, server_name=None,
+                   cafile=None, verify=True,
                    mycert=None, mykey=None,
                    client_hello=None, version=None,
                    resumption_master_secret=None,
@@ -137,6 +184,11 @@ 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
+        self.verify_server = verify
+        self.server_trust_anchors = (
+            _load_trust_anchors(cafile) if verify else []
+        )
         self.local_ip = None
         self.local_port = None
         self.socket = None
@@ -402,7 +454,22 @@ def should_handle_ServerCertificate(self):
 
     @ATMT.state()
     def HANDLED_SERVERCERTIFICATE(self):
-        pass
+        if self.verify_server:
+            self.cur_session.server_cert_valid = _verify_server_certificate(
+                self.cur_session.server_certs,
+                self.server_trust_anchors,
+                self.expected_server_name,
+            )
+            if not self.cur_session.server_cert_valid:
+                raise self.INVALID_SERVER_CERTIFICATE()
+
+    @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()
 
     @ATMT.condition(HANDLED_SERVERHELLO, prio=2)
     def missing_ServerCertificate(self):
@@ -842,7 +909,14 @@ def sslv2_should_handle_ServerHello(self):
 
     @ATMT.state()
     def SSLv2_HANDLED_SERVERHELLO(self):
-        pass
+        if self.verify_server:
+            self.cur_session.server_cert_valid = _verify_server_certificate(
+                self.cur_session.server_certs,
+                self.server_trust_anchors,
+                self.expected_server_name,
+            )
+            if not self.cur_session.server_cert_valid:
+                raise self.SSLv2_CLOSE_NOTIFY()
 
     @ATMT.condition(SSLv2_RECEIVED_SERVERHELLO, prio=2)
     def sslv2_missing_ServerHello(self):
@@ -1341,7 +1415,14 @@ def tls13_should_handle_Certificate(self):
 
     @ATMT.state()
     def TLS13_HANDLED_CERTIFICATE(self):
-        pass
+        if self.verify_server:
+            self.cur_session.server_cert_valid = _verify_server_certificate(
+                self.cur_session.server_certs,
+                self.server_trust_anchors,
+                self.expected_server_name,
+            )
+            if not self.cur_session.server_cert_valid:
+                raise self.INVALID_SERVER_CERTIFICATE()
 
     @ATMT.condition(TLS13_HANDLED_CERTIFICATE, prio=1)
     def tls13_should_handle_CertificateVerify(self):

scapy/layers/tls/session.py

@@ -411,6 +411,7 @@ def __init__(self,
         # to be sent by the server through a Certificate message.
         # The server certificate should be self.server_certs[0].
         self.server_certs = []
+        self.server_cert_valid = None
 
         # The server private key, as a PrivKey instance, when acting as server.
         # XXX It would be nice to be able to provide both an RSA and an ECDSA

test/scapy/layers/tls/tlsclientserver.uts

@@ -271,16 +271,19 @@ def run_tls_test_client(send_data=None, cipher_suite_code=None, version=None,
     commands.append(b"quit")
     if version == "0002":
         t = TLSClientAutomaton(data=commands, version="sslv2", debug=4, mycert=mycert, mykey=mykey,
+                               verify=False,
                                session_ticket_file_in=session_ticket_file_in,
                                session_ticket_file_out=session_ticket_file_out)
     elif version == "0304":
         ch = TLS13ClientHello(ciphers=int(cipher_suite_code, 16))
         t = TLSClientAutomaton(client_hello=ch, data=commands, version="tls13", debug=4, mycert=mycert, mykey=mykey,
+                               verify=False,
                                session_ticket_file_in=session_ticket_file_in,
                                session_ticket_file_out=session_ticket_file_out)
     else:
         ch = TLSClientHello(version=int(version, 16), ciphers=int(cipher_suite_code, 16))
         t = TLSClientAutomaton(client_hello=ch, data=commands, debug=4, mycert=mycert, mykey=mykey,
+                               verify=False,
                                session_ticket_file_in=session_ticket_file_in,
                                session_ticket_file_out=session_ticket_file_out)
     print("Running client...")
@@ -445,6 +448,49 @@ with open(certfile, "wb") as fd:
 with open(keyfile, "wb") as fd:
     fd.write(rsa_key)
 
+= TLS client validates certificate trust and hostname
+
+from datetime import datetime, timedelta, timezone
+from cryptography import x509 as crypto_x509
+from cryptography.hazmat.primitives import hashes
+from cryptography.hazmat.primitives.asymmetric import rsa
+from cryptography.x509.oid import ExtendedKeyUsageOID, NameOID
+from scapy.layers.tls.cert import Cert
+from scapy.layers.tls.automaton_cli import _verify_server_certificate
+
+def make_test_cert(name, key, issuer, issuer_key, ca=False):
+    now = datetime.now(timezone.utc)
+    subject = crypto_x509.Name([crypto_x509.NameAttribute(NameOID.COMMON_NAME, name)])
+    cert = (crypto_x509.CertificateBuilder()
+            .subject_name(subject)
+            .issuer_name(issuer)
+            .public_key(key.public_key())
+            .serial_number(crypto_x509.random_serial_number())
+            .not_valid_before(now - timedelta(days=1))
+            .not_valid_after(now + timedelta(days=1))
+            .add_extension(crypto_x509.BasicConstraints(ca=ca, path_length=None), True)
+            .add_extension(crypto_x509.SubjectKeyIdentifier.from_public_key(key.public_key()), False)
+            .add_extension(crypto_x509.AuthorityKeyIdentifier.from_issuer_public_key(issuer_key.public_key()), False))
+    if ca:
+        usage = crypto_x509.KeyUsage(False, False, False, False, False, True, True, None, None)
+    else:
+        usage = crypto_x509.KeyUsage(True, False, True, False, False, False, False, None, None)
+        cert = (cert.add_extension(crypto_x509.SubjectAlternativeName([crypto_x509.DNSName(name)]), False)
+                    .add_extension(crypto_x509.ExtendedKeyUsage([ExtendedKeyUsageOID.SERVER_AUTH]), False))
+    cert = cert.add_extension(usage, True).sign(issuer_key, hashes.SHA256())
+    return cert, subject
+
+test_root_key = rsa.generate_private_key(public_exponent=65537, key_size=2048)
+test_root, test_root_name = make_test_cert("test root", test_root_key, crypto_x509.Name([
+    crypto_x509.NameAttribute(NameOID.COMMON_NAME, "test root")]), test_root_key, True)
+test_leaf_key = rsa.generate_private_key(public_exponent=65537, key_size=2048)
+test_leaf, _ = make_test_cert("example.test", test_leaf_key, test_root_name, test_root_key)
+test_leaf_cert = Cert(cryptography_obj=test_leaf)
+
+assert _verify_server_certificate([test_leaf_cert], [test_root], "example.test")
+assert not _verify_server_certificate([test_leaf_cert], [test_root], "wrong.example")
+assert not _verify_server_certificate([test_leaf_cert], [], "example.test")
+
 # Define server
 
 REQS = [
@@ -524,6 +570,7 @@ def test_tls_client_native(post_handshake_auth=False,
         server="127.0.0.1",
         dport=port,
         version="tls13",
+        verify=False,
         mycert=certfile,
         mykey=keyfile,
         # we select x25519 but the server enforces seco256r1, so a Hello Retry will be issued