tls: add optional server certificate verification

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,
@@ -86,6 +87,33 @@
 )
 
 
+def _verify_server_certificate(certificates, trusted_certs, hostname):
+    """
+    Whether the server's certificate chains to a trusted CA and names the host.
+
+    :param certificates: the chain the server sent, leaf first
+    :param trusted_certs: the CAs to trust, or None to use the system store
+    :param hostname: the name the client asked for
+    :return: True if the server is authenticated
+    """
+    # None means "use the system store". An empty list is not the same thing and
+    # must fail closed: CertTree reads no roots as "trust any self-signed
+    # certificate in the list", and that list is the one the peer just sent.
+    if not certificates or (trusted_certs is not None and not trusted_certs):
+        return False
+    try:
+        CertTree(
+            list(certificates),
+            trusted_certs,
+            load_system_store=trusted_certs is None,
+        ).verify(
+            certificates[0], hostname=hostname
+        )
+    except Exception:
+        return False
+    return True
+
+
 class TLSClientAutomaton(_TLSAutomaton):
     """
     A simple TLS test client automaton. Try to overload some states or
@@ -97,6 +125,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 +147,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 +169,14 @@ 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
+        if verify and cafile:
+            self.server_trust_anchors = CertList(cafile)
+        elif verify:
+            self.server_trust_anchors = None
+        else:
+            self.server_trust_anchors = []
         self.local_ip = None
         self.local_port = None
         self.socket = None
@@ -402,7 +442,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 +897,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 +1403,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/cert.py

@@ -92,15 +92,19 @@
 """
 
 import base64
+import calendar
 import enum
 import os
+import socket
+import ssl
 import time
 import warnings
 
 from scapy.config import conf, crypto_validator
-from scapy.compat import Self
+from scapy.compat import Self, plain_str
 from scapy.error import warning
 from scapy.utils import binrepr
+from scapy.pton_ntop import inet_ntop
 from scapy.asn1.asn1 import (
     ASN1_BIT_STRING,
     ASN1_NULL,
@@ -134,6 +138,8 @@
     X509_AttributeValue,
     X509_Cert,
     X509_CRL,
+    X509_DNSName,
+    X509_IPAddress,
     X509_SubjectPublicKeyInfo,
 )
 from scapy.layers.tls.crypto.hash import _get_hash
@@ -943,6 +949,55 @@ def _get_csr_sig_hashname(csr):
     return hash_by_oid[sigAlg.algorithm.val]
 
 
+def _parse_subject_alt_name(extnValue):
+    """
+    Collect the DNS names and IP addresses from a subjectAltName extension.
+
+    :param extnValue: the X509_ExtSubjectAltName packet
+    :return: a list of ("DNS", name) and ("IP", address) pairs
+    """
+    names = []
+    for generalName in extnValue.subjectAltName or []:
+        name = generalName.generalName
+        if isinstance(name, X509_DNSName):
+            names.append(("DNS", plain_str(name.dNSName.val)))
+        elif isinstance(name, X509_IPAddress):
+            raw = name.iPAddress.val
+            if len(raw) == 4:
+                names.append(("IP", inet_ntop(socket.AF_INET, raw)))
+            elif len(raw) == 16:
+                names.append(("IP", inet_ntop(socket.AF_INET6, raw)))
+    return names
+
+
+def _match_dns_name(pattern, hostname):
+    """
+    Whether a certificate DNS name matches a hostname, RFC 6125 sect 6.4.3.
+
+    A wildcard is only honoured as the whole leftmost label, and only when the
+    name has at least two more labels after it, so ``*.example.com`` matches
+    ``a.example.com`` but not ``example.com`` or ``a.b.example.com``, and ``*.com``
+    matches nothing.
+
+    :param pattern: a dNSName from the certificate
+    :param hostname: the name the client asked for
+    :return: True if they match
+    """
+    pattern = pattern.lower().rstrip(".")
+    hostname = hostname.lower().rstrip(".")
+    if not pattern or not hostname:
+        return False
+    if not pattern.startswith("*."):
+        return pattern == hostname
+    suffix = pattern[1:]
+    if suffix.count(".") < 2:
+        # A wildcard directly under a public suffix would match too much.
+        return False
+    if not hostname.endswith(suffix):
+        return False
+    return "." not in hostname[:-len(suffix)]
+
+
 class Cert(metaclass=_CertMaker):
     """
     Wrapper for the X509_Cert from layers/x509.py.
@@ -986,6 +1041,11 @@ def import_from_asn1pkt(self, cert):
 
         self.pubkey = PubKey(bytes(tbsCert.subjectPublicKeyInfo))
 
+        # The names this certificate is issued to, as ("DNS", name) or
+        # ("IP", address) pairs. Other GeneralName kinds are not used to
+        # identify a server, so they are not recorded here.
+        self.subjectAltName = []
+
         if tbsCert.extensions:
             for extn in tbsCert.extensions:
                 if extn.extnID.oidname == "basicConstraints":
@@ -998,6 +1058,10 @@ def import_from_asn1pkt(self, cert):
                     self.extKeyUsage = extn.extnValue.get_extendedKeyUsage()
                 elif extn.extnID.oidname == "authorityKeyIdentifier":
                     self.authorityKeyID = extn.extnValue.keyIdentifier.val
+                elif extn.extnID.oidname == "subjectAltName":
+                    self.subjectAltName = _parse_subject_alt_name(
+                        extn.extnValue
+                    )
 
         self.signatureValue = bytes(cert.signatureValue)
         self.signatureLen = len(self.signatureValue)
@@ -1097,6 +1161,53 @@ def remainingDays(self, now=None):
         diff = (nft - now) / (24.0 * 3600)
         return diff
 
+    def isValidAt(self, now=None):
+        """
+        Whether the current time falls inside the certificate's validity period.
+
+        The comparison is made in UTC, which is how notBefore and notAfter are
+        stored. (:func:`remainingDays` compares in local time and so is off by
+        the local UTC offset.)
+
+        :param now: (optional) a UTC time tuple to compare against, defaulting
+            to the current time
+        :return: True if the certificate is neither expired nor not yet valid
+        """
+        if now is None:
+            now = time.gmtime()
+        now = calendar.timegm(now)
+        return (
+            calendar.timegm(self.notBefore) <= now <=
+            calendar.timegm(self.notAfter)
+        )
+
+    def matchesHostname(self, hostname):
+        """
+        Whether this certificate was issued to the given host.
+
+        Names come from the subjectAltName extension. RFC 6125 sect 6.4.4 says
+        the Common Name is only consulted when there is no subjectAltName at
+        all, and that is what happens here.
+
+        :param hostname: the DNS name or IP address the client asked for
+        :return: True if the certificate names that host
+        """
+        if not hostname:
+            return False
+        hostname = plain_str(hostname)
+        if self.subjectAltName:
+            for kind, name in self.subjectAltName:
+                if kind == "IP":
+                    if name == hostname:
+                        return True
+                elif _match_dns_name(name, hostname):
+                    return True
+            return False
+        for attr in self.subject_str.split("/"):
+            if attr.startswith("CN=") and _match_dns_name(attr[3:], hostname):
+                return True
+        return False
+
     def isRevoked(self, crl_list):
         """
         Given a list of trusted CRL (their signature has already been
@@ -1134,7 +1245,15 @@ def pem(self):
 
     @property
     def der(self):
-        return bytes(self.x509Cert)
+        # Cached because __eq__ and __hash__ both read it, and re-encoding the
+        # whole certificate for every comparison is slow enough to matter:
+        # chaining against a system trust store is thousands of comparisons.
+        # A Cert is built once from immutable parsed ASN.1 and never edited.
+        try:
+            return self._der_cache
+        except AttributeError:
+            self._der_cache = bytes(self.x509Cert)
+            return self._der_cache
 
     @property
     def pubKey(self):
@@ -1566,6 +1685,7 @@ def __init__(
         self,
         certList: Union[List[Cert], CertList, str],
         rootCAs: Union[List[Cert], CertList, Cert, str, None] = None,
+        load_system_store: bool = False,
     ):
         """
         Construct a chain of certificates that follows issuer/subject matching and
@@ -1578,10 +1698,19 @@ def __init__(
             multiple certs/CRL) to try to chain.
         :param rootCAs: (optional) a list of certificates to trust. If not provided,
             trusts any self-signed certificates from the certList.
+        :param load_system_store: use the system trust store when rootCAs is empty.
         """
         # Parse the certificate list
         certList = CertList(certList)
 
+        if not rootCAs and load_system_store:
+            context = ssl.create_default_context()
+            rootCAs = [
+                Cert(der) for der in context.get_ca_certs(binary_form=True)
+            ]
+            if not rootCAs:
+                raise ValueError("The system trust store contains no certificates")
+
         # Find the ROOT CAs if store isn't specified
         if not rootCAs:
             # Build cert store.
@@ -1662,13 +1791,42 @@ def _rec_getchain(chain, curtree):
         else:
             return None
 
-    def verify(self, cert):
+    def verify(self, cert, hostname=None, now=None):
         """
-        Verify that a certificate is properly signed.
+        Verify that a certificate is properly signed, current, and the right one.
+
+        Raises ValueError when the certificate fails any of the checks.
+
+        :param cert: the certificate to verify
+        :param hostname: (optional) the DNS name or IP address the peer was
+            expected to be. Without it the identity of the peer is not checked,
+            so any certificate the store can chain is accepted.
+        :param now: (optional) a UTC time tuple to check validity against,
+            defaulting to the current time
         """
         # Check that we can find a chain to this certificate
-        if not self.getchain(cert):
+        chain = self.getchain(cert)
+        if not chain:
             raise ValueError("Certificate verification failed !")
+        # Nothing in the chain may have expired or be in the future: an issuer
+        # that is out of date does not vouch for anything below it. A chain can
+        # also hold a CSR, which has no validity period to check.
+        for c in chain:
+            if not isinstance(c, Cert):
+                continue
+            if not c.isValidAt(now):
+                raise ValueError(
+                    "Certificate %s is outside its validity period "
+                    "(%s to %s) !" % (
+                        c.subject_str, c.notBefore_str, c.notAfter_str
+                    )
+                )
+        if hostname is not None and not cert.matchesHostname(hostname):
+            raise ValueError(
+                "Certificate %s was not issued to %s !" % (
+                    cert.subject_str, plain_str(hostname)
+                )
+            )
 
     def show(self, ret: bool = False):
         """

scapy/layers/tls/session.py

@@ -418,6 +418,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