From 05d40913d1bae6c8950a01f1e7787d95f722c7c4 Mon Sep 17 00:00:00 2001 From: Clinton Thomas <1033162+KernelClint@users.noreply.github.com> Date: Wed, 9 Sep 2026 13:25:38 -0400 Subject: [PATCH 1/5] tls: add optional server certificate verification AI-Assisted: yes (GPT-5.6-Cyber) --- scapy/layers/tls/automaton_cli.py | 75 +++++++++- scapy/layers/tls/cert.py | 168 +++++++++++++++++++++- scapy/layers/tls/session.py | 1 + test/scapy/layers/tls/cert.uts | 28 +++- test/scapy/layers/tls/tlsclientserver.uts | 118 +++++++++++++++ 5 files changed, 379 insertions(+), 11 deletions(-) diff --git a/scapy/layers/tls/automaton_cli.py b/scapy/layers/tls/automaton_cli.py index ff0e9125ff9..0fc5a8ec1b9 100644 --- a/scapy/layers/tls/automaton_cli.py +++ b/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): @@ -1347,7 +1409,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): diff --git a/scapy/layers/tls/cert.py b/scapy/layers/tls/cert.py index 5b1dc565bd8..8b1a0ef149c 100644 --- a/scapy/layers/tls/cert.py +++ b/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): """ diff --git a/scapy/layers/tls/session.py b/scapy/layers/tls/session.py index 26813e49298..45f7cf23cf0 100644 --- a/scapy/layers/tls/session.py +++ b/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 diff --git a/test/scapy/layers/tls/cert.uts b/test/scapy/layers/tls/cert.uts index b8dc78bf394..bb85a706f5b 100644 --- a/test/scapy/layers/tls/cert.uts +++ b/test/scapy/layers/tls/cert.uts @@ -652,17 +652,39 @@ assert repr_str == '/OU=Domain Control Validated/CN=*.tools.ietf.org [Not Self S = CertTree class : verify -CertTree([c1, c2]).verify(c0) +# c0 expired in 2016, so the validity check needs a date it was still current on. +in2016 = (2016, 1, 1, 0, 0, 0, 4, 1, 0) + +CertTree([c1, c2]).verify(c0, now=in2016) CertTree([c2]).verify(c1) try: - CertTree([c1]).verify(c0) + CertTree([c1]).verify(c0, now=in2016) + assert False +except ValueError: + pass + +try: + CertTree([c2]).verify(c0, now=in2016) + assert False +except ValueError: + pass + += CertTree class : verify rejects an expired certificate + +# The chain is intact; only the date is wrong. +try: + CertTree([c1, c2]).verify(c0) assert False except ValueError: pass += CertTree class : verify checks the hostname when given one + +CertTree([c1, c2]).verify(c0, hostname="www.tools.ietf.org", now=in2016) + try: - CertTree([c2]).verify(c0) + CertTree([c1, c2]).verify(c0, hostname="www.example.com", now=in2016) assert False except ValueError: pass diff --git a/test/scapy/layers/tls/tlsclientserver.uts b/test/scapy/layers/tls/tlsclientserver.uts index 64736edcfe5..9b48b0b194d 100644 --- a/test/scapy/layers/tls/tlsclientserver.uts +++ b/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...") @@ -430,6 +433,7 @@ def run_tls13_client_auth_exchange(client_key): server="127.0.0.1", dport=port, version="tls13", + verify=False, mycert=client_cert, mykey=client_key, data=[marker, b"stop_server", b"quit"], @@ -446,6 +450,7 @@ def run_tls13_client_auth_exchange(client_key): server="127.0.0.1", dport=port, version="tls13", + verify=False, mycert=client_cert, mykey=scapy_path("/test/scapy/layers/tls/pki/cli_key.pem"), data=[b"stop_server", b"quit"], @@ -615,6 +620,118 @@ 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, CertTree +from scapy.layers.tls.automaton_cli import _verify_server_certificate +from unittest.mock import patch + +def make_test_cert(name, key, issuer, issuer_key, ca=False, age=timedelta()): + now = datetime.now(timezone.utc) - age + 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_root_cert = Cert(cryptography_obj=test_root) +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_cert], "example.test") +assert not _verify_server_certificate([test_leaf_cert], [test_root_cert], "wrong.example") + +# Explicit anchors take precedence over the system store. +with patch("scapy.layers.tls.cert.ssl.create_default_context", + side_effect=AssertionError("system store consulted")): + assert _verify_server_certificate( + [test_leaf_cert], [test_root_cert], "example.test") + +# An empty system store must not fall back to a peer-provided self-signed root. +class _EmptySystemStore: + def get_ca_certs(self, binary_form=False): + assert binary_form + return [] + +with patch("scapy.layers.tls.cert.ssl.create_default_context", + return_value=_EmptySystemStore()): + try: + CertTree([test_root_cert], load_system_store=True) + assert False, "accepted a peer root when the system store was empty" + except ValueError: + pass + +# No trust anchors at all must fail closed, not fall back to trusting the +# self-signed certificates the peer happened to send. +assert not _verify_server_certificate([test_leaf_cert], [], "example.test") + +# A certificate signed by someone we do not trust. +other_root_key = rsa.generate_private_key(public_exponent=65537, key_size=2048) +other_root, other_root_name = make_test_cert("other root", other_root_key, crypto_x509.Name([ + crypto_x509.NameAttribute(NameOID.COMMON_NAME, "other root")]), other_root_key, True) +assert not _verify_server_certificate( + [test_leaf_cert], [Cert(cryptography_obj=other_root)], "example.test") + +# An expired certificate, and an expired issuer for a leaf that is still current. +expired_leaf, _ = make_test_cert("example.test", test_leaf_key, test_root_name, + test_root_key, age=timedelta(days=10)) +assert not _verify_server_certificate( + [Cert(cryptography_obj=expired_leaf)], [test_root_cert], "example.test") + +old_root_key = rsa.generate_private_key(public_exponent=65537, key_size=2048) +old_root, old_root_name = make_test_cert("old root", old_root_key, crypto_x509.Name([ + crypto_x509.NameAttribute(NameOID.COMMON_NAME, "old root")]), old_root_key, True, + age=timedelta(days=10)) +fresh_leaf, _ = make_test_cert("example.test", test_leaf_key, old_root_name, old_root_key) +assert not _verify_server_certificate( + [Cert(cryptography_obj=fresh_leaf)], [Cert(cryptography_obj=old_root)], "example.test") + += CertTree.verify checks the name and the validity period + +from scapy.layers.tls.cert import CertTree + +wildcard_leaf, _ = make_test_cert("*.wild.test", test_leaf_key, test_root_name, test_root_key) +wildcard_cert = Cert(cryptography_obj=wildcard_leaf) +tree = CertTree([wildcard_cert, test_root_cert], [test_root_cert]) + +tree.verify(wildcard_cert) +tree.verify(wildcard_cert, hostname="host.wild.test") + +# A wildcard covers one label, and only the leftmost one. +for rejected in ["wild.test", "a.b.wild.test", "host.other.test"]: + try: + tree.verify(wildcard_cert, hostname=rejected) + assert False, "accepted %s" % rejected + except ValueError: + pass + +# subjectAltName wins outright: the Common Name is not consulted beside it. +assert wildcard_cert.subjectAltName == [("DNS", "*.wild.test")] +assert not wildcard_cert.matchesHostname("test root") + # Define server REQS = [ @@ -694,6 +811,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 From 76871688a5207db9a1eee80f017d4be529897c2e Mon Sep 17 00:00:00 2001 From: Clinton Thomas <1033162+KernelClint@users.noreply.github.com> Date: Wed, 23 Sep 2026 17:02:42 -0400 Subject: [PATCH 2/5] tls: make server verification opt-in, and say why it failed Verification defaulted to on, which changed behaviour for every existing caller: this client is routinely pointed at servers whose certificates are not meant to verify. It is now off unless asked for, which is what the existing tests wanted -- the six verify=False opt-outs they needed are gone. Passing cafile turns it on, since supplying a CA and getting no checking would be worse than either. A refusal logged only "verification failed". It now logs which check failed, alongside the other TLS refusals. AI-Assisted: yes (gpt-5.6-sol) --- scapy/layers/tls/automaton_cli.py | 27 +++++++++++++++-------- test/scapy/layers/tls/tlsclientserver.uts | 6 ----- 2 files changed, 18 insertions(+), 15 deletions(-) diff --git a/scapy/layers/tls/automaton_cli.py b/scapy/layers/tls/automaton_cli.py index 0fc5a8ec1b9..b79ef186484 100644 --- a/scapy/layers/tls/automaton_cli.py +++ b/scapy/layers/tls/automaton_cli.py @@ -44,7 +44,7 @@ from scapy.config import conf from scapy.utils import randstring, repr_hex from scapy.automaton import ATMT, select_objects -from scapy.error import warning +from scapy.error import log_runtime, 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 @@ -94,12 +94,17 @@ def _verify_server_certificate(certificates, trusted_certs, hostname): :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 + :return: True if the server is authenticated. The reason for a refusal is + logged, because a bare "verification failed" is not actionable. """ # 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): + if not certificates: + log_runtime.info("TLS: the server sent no certificate") + return False + if trusted_certs is not None and not trusted_certs: + log_runtime.info("TLS: no certificate authority was supplied to trust") return False try: CertTree( @@ -109,7 +114,9 @@ def _verify_server_certificate(certificates, trusted_certs, hostname): ).verify( certificates[0], hostname=hostname ) - except Exception: + except Exception as error: + log_runtime.info("TLS: server certificate rejected for %s [%s]", + hostname, error or error.__class__.__name__) return False return True @@ -127,7 +134,9 @@ class TLSClientAutomaton(_TLSAutomaton): :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 verify: whether to authenticate the server certificate against a trust + store and the requested hostname. Off by default, because this client is + commonly pointed at servers whose certificates are not meant to verify. :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. @@ -147,7 +156,7 @@ class TLSClientAutomaton(_TLSAutomaton): """ def parse_args(self, server="127.0.0.1", dport=4433, server_name=None, - cafile=None, verify=True, + cafile=None, verify=False, mycert=None, mykey=None, client_hello=None, version=None, resumption_master_secret=None, @@ -170,10 +179,10 @@ def parse_args(self, server="127.0.0.1", dport=4433, server_name=None, 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.verify_server = verify or bool(cafile) + if self.verify_server and cafile: self.server_trust_anchors = CertList(cafile) - elif verify: + elif self.verify_server: self.server_trust_anchors = None else: self.server_trust_anchors = [] diff --git a/test/scapy/layers/tls/tlsclientserver.uts b/test/scapy/layers/tls/tlsclientserver.uts index 9b48b0b194d..9805e71ddb7 100644 --- a/test/scapy/layers/tls/tlsclientserver.uts +++ b/test/scapy/layers/tls/tlsclientserver.uts @@ -271,19 +271,16 @@ 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...") @@ -433,7 +430,6 @@ def run_tls13_client_auth_exchange(client_key): server="127.0.0.1", dport=port, version="tls13", - verify=False, mycert=client_cert, mykey=client_key, data=[marker, b"stop_server", b"quit"], @@ -450,7 +446,6 @@ def run_tls13_client_auth_exchange(client_key): server="127.0.0.1", dport=port, version="tls13", - verify=False, mycert=client_cert, mykey=scapy_path("/test/scapy/layers/tls/pki/cli_key.pem"), data=[b"stop_server", b"quit"], @@ -811,7 +806,6 @@ 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 From 245725756cb951b090362dfed1abe5ac8c53380f Mon Sep 17 00:00:00 2001 From: Clinton Thomas <1033162+KernelClint@users.noreply.github.com> Date: Thu, 24 Sep 2026 12:11:52 -0400 Subject: [PATCH 3/5] tls: let a caller accept an expired chain CertTree.verify() and CMS_Engine.verify() gain allow_expired, defaulting to False so the expiry check stays on everywhere it already was. A CMS signature is usually checked long after it was made, and the PKINIT tests are a real captured MIT Kerberos exchange whose client certificate was valid 2025-09-20 to 2026-09-20. The signature it carries is genuine; only the clock has moved on, so those two tests pass allow_expired=True. AI-Assisted: yes --- scapy/layers/tls/cert.py | 15 ++++++++++++--- test/scapy/layers/kerberos.uts | 9 ++++++--- 2 files changed, 18 insertions(+), 6 deletions(-) diff --git a/scapy/layers/tls/cert.py b/scapy/layers/tls/cert.py index 8b1a0ef149c..88d0c5ef996 100644 --- a/scapy/layers/tls/cert.py +++ b/scapy/layers/tls/cert.py @@ -1791,7 +1791,7 @@ def _rec_getchain(chain, curtree): else: return None - def verify(self, cert, hostname=None, now=None): + def verify(self, cert, hostname=None, now=None, allow_expired=False): """ Verify that a certificate is properly signed, current, and the right one. @@ -1803,6 +1803,10 @@ def verify(self, cert, hostname=None, now=None): 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 + :param allow_expired: (optional) accept a chain that is outside its + validity period. Useful when checking a signature made while the + certificate was still valid, or against a stored capture, where + the dates say nothing about whether the signature was genuine. """ # Check that we can find a chain to this certificate chain = self.getchain(cert) @@ -1814,7 +1818,7 @@ def verify(self, cert, hostname=None, now=None): for c in chain: if not isinstance(c, Cert): continue - if not c.isValidAt(now): + if not allow_expired and not c.isValidAt(now): raise ValueError( "Certificate %s is outside its validity period " "(%s to %s) !" % ( @@ -2053,6 +2057,7 @@ def verify( eContentType: Optional[ASN1_OID] = None, eContent: Optional[bytes] = None, no_verify_cert: bool = False, + allow_expired: bool = False, ): """ Verify a CMS message against the list of trusted certificates, @@ -2062,6 +2067,10 @@ def verify( :param eContentType: if provided, verifies that the content type is valid :param eContent: in PKCS 7.1, provide the content to verify :param no_verify_cert: do not check the remote certificate (unsafe) + :param allow_expired: accept a signer certificate that is outside its + validity period. A CMS signature is often checked long after it was + made, so an expired signer does not by itself mean the signature is + not genuine. """ if contentInfo.contentType.oidname != "id-signedData": raise ValueError("ContentInfo isn't signed !") @@ -2091,7 +2100,7 @@ def verify( # Verify certificate signature if not no_verify_cert: - certTree.verify(cert) + certTree.verify(cert, allow_expired=allow_expired) # Verify the message hash if signerInfo.signedAttrs: diff --git a/test/scapy/layers/kerberos.uts b/test/scapy/layers/kerberos.uts index da71abb8fe3..4e776d697ad 100644 --- a/test/scapy/layers/kerberos.uts +++ b/test/scapy/layers/kerberos.uts @@ -277,8 +277,11 @@ ca = Cert(bytes.fromhex('3082036930820251a00302010202106b671318bb858b8e437e4229b # Build CMS engine to verify the authpack cms = CMS_Engine(CertList([ca])) -# Verify signature -authpack = cms.verify(signedauthpack, ASN1_OID('id-pkinit-authData')) +# Verify signature. +# allow_expired: this is a real captured MIT Kerberos exchange, and its client +# certificate was valid 2025-09-20 to 2026-09-20. The signature it carries is +# genuine; only the clock has moved on. +authpack = cms.verify(signedauthpack, ASN1_OID('id-pkinit-authData'), allow_expired=True) assert isinstance(authpack, KRB_AuthPack) = PKINIT - Resign AuthPack and re-verify signature @@ -295,7 +298,7 @@ signed = cms.sign( key, ) -authpack = cms.verify(signed, ASN1_OID('id-pkinit-authData')) +authpack = cms.verify(signed, ASN1_OID('id-pkinit-authData'), allow_expired=True) assert isinstance(authpack, KRB_AuthPack) = PKINIT - Parse AS-REP with CMS structures (MIT Kerberos) From bfec945b3790f2d223c25fcf51861180f0c086a1 Mon Sep 17 00:00:00 2001 From: gpotter2 <10530980+gpotter2@users.noreply.github.com> Date: Fri, 25 Sep 2026 16:43:44 +0200 Subject: [PATCH 4/5] Apply feedback & code cleanup AI-Assisted: no --- scapy/layers/tls/automaton.py | 3 + scapy/layers/tls/automaton_cli.py | 115 +++----- scapy/layers/tls/automaton_srv.py | 45 +-- scapy/layers/tls/cert.py | 107 +++---- scapy/layers/tls/handshake.py | 6 +- scapy/layers/tls/handshake_sslv2.py | 4 +- scapy/layers/tls/session.py | 4 +- test/scapy/layers/tls/cert.uts | 1 - test/scapy/layers/tls/tlsclientserver.uts | 336 +++------------------- 9 files changed, 154 insertions(+), 467 deletions(-) diff --git a/scapy/layers/tls/automaton.py b/scapy/layers/tls/automaton.py index 9395388de91..647ad5280b4 100644 --- a/scapy/layers/tls/automaton.py +++ b/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 diff --git a/scapy/layers/tls/automaton_cli.py b/scapy/layers/tls/automaton_cli.py index b79ef186484..f2eb6f246f8 100644 --- a/scapy/layers/tls/automaton_cli.py +++ b/scapy/layers/tls/automaton_cli.py @@ -87,40 +87,6 @@ ) -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. The reason for a refusal is - logged, because a bare "verification failed" is not actionable. - """ - # 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: - log_runtime.info("TLS: the server sent no certificate") - return False - if trusted_certs is not None and not trusted_certs: - log_runtime.info("TLS: no certificate authority was supplied to trust") - return False - try: - CertTree( - list(certificates), - trusted_certs, - load_system_store=trusted_certs is None, - ).verify( - certificates[0], hostname=hostname - ) - except Exception as error: - log_runtime.info("TLS: server certificate rejected for %s [%s]", - hostname, error or error.__class__.__name__) - return False - return True - - class TLSClientAutomaton(_TLSAutomaton): """ A simple TLS test client automaton. Try to overload some states or @@ -133,10 +99,8 @@ class TLSClientAutomaton(_TLSAutomaton): :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 against a trust - store and the requested hostname. Off by default, because this client is - commonly pointed at servers whose certificates are not meant to verify. + 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. @@ -156,7 +120,7 @@ class TLSClientAutomaton(_TLSAutomaton): """ def parse_args(self, server="127.0.0.1", dport=4433, server_name=None, - cafile=None, verify=False, + cafile=None, verify_server=False, mycert=None, mykey=None, client_hello=None, version=None, resumption_master_secret=None, @@ -179,13 +143,14 @@ def parse_args(self, server="127.0.0.1", dport=4433, server_name=None, self.remote_port = dport self.server_name = server_name self.expected_server_name = server_name or server - self.verify_server = verify or bool(cafile) + print(verify_server) + self.verify_server = verify_server or bool(cafile) if self.verify_server and cafile: - self.server_trust_anchors = CertList(cafile) + self.server_trust_store = CertList(cafile) elif self.verify_server: - self.server_trust_anchors = None + self.server_trust_store = CertList.load_system_store() else: - self.server_trust_anchors = [] + self.server_trust_store = None self.local_ip = None self.local_port = None self.socket = None @@ -449,24 +414,31 @@ def should_handle_ServerCertificate(self): self.HANDLED_SERVERCERTIFICATE) raise self.HANDLED_SERVERCERTIFICATE() - @ATMT.state() - def HANDLED_SERVERCERTIFICATE(self): + def _verify_server_cert(self): 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() + 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): + 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() + raise self.FINAL(reason=self.INVALID_SERVER_CERTIFICATE) @ATMT.condition(HANDLED_SERVERHELLO, prio=2) def missing_ServerCertificate(self): @@ -493,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): @@ -862,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 # @@ -906,14 +878,10 @@ def sslv2_should_handle_ServerHello(self): @ATMT.state() def SSLv2_HANDLED_SERVERHELLO(self): - 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() + 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): @@ -993,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() @@ -1163,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 # @@ -1418,14 +1386,7 @@ def tls13_should_handle_Certificate(self): @ATMT.state() def TLS13_HANDLED_CERTIFICATE(self): - 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() + self._verify_server_cert() @ATMT.condition(TLS13_HANDLED_CERTIFICATE, prio=1) def tls13_should_handle_CertificateVerify(self): @@ -1448,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): @@ -1544,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): @@ -1555,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.") diff --git a/scapy/layers/tls/automaton_srv.py b/scapy/layers/tls/automaton_srv.py index 769d6c4b45b..fc4bec84c8a 100644 --- a/scapy/layers/tls/automaton_srv.py +++ b/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.") diff --git a/scapy/layers/tls/cert.py b/scapy/layers/tls/cert.py index 88d0c5ef996..66db47c775c 100644 --- a/scapy/layers/tls/cert.py +++ b/scapy/layers/tls/cert.py @@ -949,53 +949,20 @@ 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("*."): + if pattern[:2] == "*.": + return pattern[2:] == hostname.split(".", 1)[-1] + else: 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): @@ -1040,11 +1007,7 @@ def import_from_asn1pkt(self, cert): self.notAfter_str_simple = time.strftime("%x", self.notAfter) 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 = [] + self.subjectAltName = None if tbsCert.extensions: for extn in tbsCert.extensions: @@ -1059,9 +1022,7 @@ def import_from_asn1pkt(self, cert): 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.subjectAltName = extn.extnValue.subjectAltName self.signatureValue = bytes(cert.signatureValue) self.signatureLen = len(self.signatureValue) @@ -1195,17 +1156,27 @@ def matchesHostname(self, hostname): if not hostname: return False hostname = plain_str(hostname) - if self.subjectAltName: - for kind, name in self.subjectAltName: - if kind == "IP": - if name == hostname: + + if self.subjectAltName is not None: + # Check subjectAltName + for generalName in self.subjectAltName: + name = generalName.generalName + if isinstance(name, X509_DNSName): + if _match_dns_name(plain_str(name.dNSName.val), hostname): return True - elif _match_dns_name(name, hostname): - return True + elif isinstance(name, X509_IPAddress): + raw = name.iPAddress.val + if len(raw) == 4: + if inet_ntop(socket.AF_INET, raw) == hostname: + return True + elif len(raw) == 16: + if inet_ntop(socket.AF_INET6, raw) == 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 + else: + # Check Common Name + if "commonName" in self.subject: + return _match_dns_name(self.subject["commonName"], hostname) return False def isRevoked(self, crl_list): @@ -1245,15 +1216,7 @@ def pem(self): @property def der(self): - # 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 + return bytes(self.x509Cert) @property def pubKey(self): @@ -1616,6 +1579,16 @@ def __init__( super(CertList, self).__init__(certList) + @classmethod + def load_system_store(cls): + """ + Return a CertList containing the default trusted system store. + """ + context = ssl.create_default_context() + return cls([ + Cert(der) for der in context.get_ca_certs(binary_form=True) + ]) + def findCertBySid(self, sid): """ Find a certificate in the list by SubjectIDentifier. @@ -1685,7 +1658,6 @@ 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 @@ -1698,19 +1670,10 @@ 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. diff --git a/scapy/layers/tls/handshake.py b/scapy/layers/tls/handshake.py index abde901a246..847ae10d894 100644 --- a/scapy/layers/tls/handshake.py +++ b/scapy/layers/tls/handshake.py @@ -39,7 +39,7 @@ from scapy.packet import Packet, Raw, Padding from scapy.utils import randstring, repr_hex from scapy.layers.x509 import OCSP_Response -from scapy.layers.tls.cert import Cert +from scapy.layers.tls.cert import Cert, CertList from scapy.layers.tls.basefields import (_tls_version, _TLSVersionField, _TLSClientVersionField) from scapy.layers.tls.extensions import (_ExtensionsLenField, _ExtensionsField, @@ -969,9 +969,9 @@ def post_dissection_tls_session_update(self, msg_str): self.tls_session_update(msg_str) connection_end = self.tls_session.connection_end if connection_end == "client": - self.tls_session.server_certs = [x[1] for x in self.certs] + self.tls_session.server_certs = CertList([x[1] for x in self.certs]) else: - self.tls_session.client_certs = [x[1] for x in self.certs] + self.tls_session.client_certs = CertList([x[1] for x in self.certs]) class _ASN1CertAndExt(_GenericTLSSessionInheritance): diff --git a/scapy/layers/tls/handshake_sslv2.py b/scapy/layers/tls/handshake_sslv2.py index 1004280ec54..48dfb6ad732 100644 --- a/scapy/layers/tls/handshake_sslv2.py +++ b/scapy/layers/tls/handshake_sslv2.py @@ -15,7 +15,7 @@ ShortEnumField, StrLenField, XStrField, XStrLenField from scapy.packet import Padding -from scapy.layers.tls.cert import Cert +from scapy.layers.tls.cert import Cert, CertList from scapy.layers.tls.basefields import _tls_version, _TLSVersionField from scapy.layers.tls.handshake import _CipherSuitesField from scapy.layers.tls.keyexchange import _TLSSignatureField, _TLSSignature @@ -196,7 +196,7 @@ def tls_session_update(self, msg_str): s.sslv2_connection_id = self.connection_id s.tls_version = self.version if self.cert is not None: - s.server_certs = [self.cert] + s.server_certs = CertList([self.cert]) ############################################################################### diff --git a/scapy/layers/tls/session.py b/scapy/layers/tls/session.py index 45f7cf23cf0..16baea88909 100644 --- a/scapy/layers/tls/session.py +++ b/scapy/layers/tls/session.py @@ -22,6 +22,7 @@ from scapy.sessions import TCPSession from scapy.utils import repr_hex, strxor from scapy.layers.inet import TCP +from scapy.layers.tls.cert import CertList from scapy.layers.tls.crypto.compression import Comp_NULL from scapy.layers.tls.crypto.hkdf import TLS13_HKDF from scapy.layers.tls.crypto.prf import PRF @@ -417,8 +418,7 @@ def __init__(self, # Either we act as server and it has to be provided, or it is expected # 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 + self.server_certs = CertList([]) # 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 diff --git a/test/scapy/layers/tls/cert.uts b/test/scapy/layers/tls/cert.uts index bb85a706f5b..c5f368a15ea 100644 --- a/test/scapy/layers/tls/cert.uts +++ b/test/scapy/layers/tls/cert.uts @@ -689,7 +689,6 @@ try: except ValueError: pass - = Test GeneralizedTime data = b"MHAwXAIBADANBgkqhkiG9w0BAQ0FADAAMCIYDzIwMTExMDA2MDgzOTU2WhgPMjA0NjEwMDYwODM5NTZaMAAwHDANBgkqhkiG9w0BAQEFAAMLADAIAgEAAgMBAAGjAjAAMA0GCSqGSIb3DQEBDQUAAwEA" diff --git a/test/scapy/layers/tls/tlsclientserver.uts b/test/scapy/layers/tls/tlsclientserver.uts index 9805e71ddb7..4e97cd99730 100644 --- a/test/scapy/layers/tls/tlsclientserver.uts +++ b/test/scapy/layers/tls/tlsclientserver.uts @@ -257,7 +257,8 @@ send_data = cipher_suite_code = version = None def run_tls_test_client(send_data=None, cipher_suite_code=None, version=None, client_auth=False, key_update=False, stop_server=True, - session_ticket_file_out=None, session_ticket_file_in=None): + session_ticket_file_out=None, session_ticket_file_in=None, + verify_server=False): print("Loading client...") mycert = scapy_path("/test/scapy/layers/tls/pki/cli_cert.pem") if client_auth else None mykey = scapy_path("/test/scapy/layers/tls/pki/cli_key.pem") if client_auth else None @@ -272,22 +273,26 @@ def run_tls_test_client(send_data=None, cipher_suite_code=None, version=None, if version == "0002": t = TLSClientAutomaton(data=commands, version="sslv2", debug=4, mycert=mycert, mykey=mykey, session_ticket_file_in=session_ticket_file_in, - session_ticket_file_out=session_ticket_file_out) + session_ticket_file_out=session_ticket_file_out, + verify_server=verify_server) 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, session_ticket_file_in=session_ticket_file_in, - session_ticket_file_out=session_ticket_file_out) + session_ticket_file_out=session_ticket_file_out, + verify_server=verify_server) 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, session_ticket_file_in=session_ticket_file_in, - session_ticket_file_out=session_ticket_file_out) + session_ticket_file_out=session_ticket_file_out, + verify_server=verify_server) print("Running client...") t.run() + return t.final_reason def test_tls_client(suite, version, curve=None, cookie=False, client_auth=False, - key_update=False, sess_in_out=False, sigalgo="rsa"): + key_update=False, sess_in_out=False, sigalgo="rsa", verify_server=False): msg = ("TestC_%s_data" % suite).encode() # Run server q_ = Queue() @@ -305,25 +310,38 @@ def test_tls_client(suite, version, curve=None, cookie=False, client_auth=False, try: wait_tls_test_server_online() except Exception as ex: - atmtsrv.stop() + atmtsrv.forcestop() raise ex print("Thread synchronised") # Run client if sess_in_out: file_sess = scapy_path("/test/session") run_tls_test_client(msg, suite, version, client_auth, key_update, session_ticket_file_out=file_sess, - stop_server=False) - run_tls_test_client(msg, suite, version, client_auth, key_update, session_ticket_file_in=file_sess, - stop_server=True) + stop_server=False, verify_server=verify_server) + + final_reason = run_tls_test_client(msg, suite, version, client_auth, key_update, session_ticket_file_in=file_sess, + stop_server=True, verify_server=verify_server) else: - run_tls_test_client(msg, suite, version, client_auth, key_update) - # Wait for server - print("Client running, waiting...") - ret = q_.get(timeout=5) - if not ret: - raise RuntimeError("Test timed out") - atmtsrv.stop() - print(ret) + final_reason = run_tls_test_client(msg, suite, version, client_auth, key_update, verify_server=verify_server) + print("Client finished: ", final_reason.atmt_origfunc) + # If client crashed, kill server + if final_reason.atmt_origfunc not in [TLSClientAutomaton.CLOSE_NOTIFY.atmt_origfunc, TLSClientAutomaton.SSLv2_CLOSE_NOTIFY.atmt_origfunc]: + atmtsrv.stop() + atmtsrv.destroy() + else: + final_reason = None + # Wait for server + print("Server still running, waiting...") + ret = q_.get(timeout=1) + if not ret: + raise RuntimeError("Test timed out") + atmtsrv.stop() + atmtsrv.destroy() + print(ret) + # re-raise if client failed. + if final_reason: + raise ValueError(final_reason.atmt_origfunc) + # final assertion that server didn't fail. assert ret[0] = Testing TLS server and client with SSLv2 and SSL_CK_DES_192_EDE3_CBC_WITH_MD5 @@ -402,63 +420,14 @@ test_tls_client("1305", "0304", key_update=True) test_tls_client("1305", "0304", client_auth=True, sess_in_out=True) -= Reject a CertificateVerify made with a key unrelated to the client certificate += Testing TLS server and client with TLS 1.3 and TLS_AES_128_CCM_8_SHA256 and server verification ~ crypto_advanced -def run_tls13_client_auth_exchange(client_key): - marker = b"certificate proof accepted" - server_cert = scapy_path("/test/scapy/layers/tls/pki/srv_cert.pem") - server_key = scapy_path("/test/scapy/layers/tls/pki/srv_key.pem") - client_cert = scapy_path("/test/scapy/layers/tls/pki/cli_cert.pem") - probe = socket.socket() - probe.bind(("127.0.0.1", 0)) - port = probe.getsockname()[1] - probe.close() - server = TLSServerAutomaton( - server="127.0.0.1", - sport=port, - mycert=server_cert, - mykey=server_key, - client_auth=True, - debug=0, - ) - server_thread = threading.Thread(target=server.run, daemon=True) - with captured_output() as (out, _): - server_thread.start() - time.sleep(0.1) - client = TLSClientAutomaton( - server="127.0.0.1", - dport=port, - version="tls13", - mycert=client_cert, - mykey=client_key, - data=[marker, b"stop_server", b"quit"], - debug=0, - ) - client_thread = threading.Thread(target=client.run, daemon=True) - client_thread.start() - client_thread.join(timeout=2) - time.sleep(0.1) - if client_thread.is_alive(): - client.stop(wait=False) - if server_thread.is_alive(): - stopper = TLSClientAutomaton( - server="127.0.0.1", - dport=port, - version="tls13", - mycert=client_cert, - mykey=scapy_path("/test/scapy/layers/tls/pki/cli_key.pem"), - data=[b"stop_server", b"quit"], - debug=0, - ) - stopper.run() - server_thread.join(timeout=1) - return "> Received: %r" % marker in out.s - -server_key = scapy_path("/test/scapy/layers/tls/pki/srv_key.pem") -client_key = scapy_path("/test/scapy/layers/tls/pki/cli_key.pem") -assert not run_tls13_client_auth_exchange(server_key) -assert run_tls13_client_auth_exchange(client_key) +try: + test_tls_client("1305", "0304", verify_server=True) + assert False, "Client should have rejected the server" +except ValueError as ex: + assert "INVALID_SERVER_CERTIFICATE" in str(ex) = Clear session file @@ -473,118 +442,6 @@ except: + TLS client automaton tests against builtin ssl using Post Handshake Authentication ~ client post_handshake_auth -= TLS 1.3 client rejects an invalid server CertificateVerify -~ crypto_advanced - -import contextlib, io, ipaddress, socket -from datetime import datetime, timedelta, timezone -from cryptography import x509 as crypto_x509 -from cryptography.hazmat.primitives import hashes, serialization -from cryptography.x509.oid import ExtendedKeyUsageOID, NameOID -from scapy.layers.tls.cert import PrivKey - -server_key = scapy_path("/test/scapy/layers/tls/pki/srv_key.pem") -with open(server_key, "rb") as cert_key_file: - cert_key = serialization.load_pem_private_key(cert_key_file.read(), None) - -with open(scapy_path("/test/scapy/layers/tls/pki/ca_key.pem"), "rb") as ca_key_file: - ca_key = serialization.load_pem_private_key(ca_key_file.read(), None) - -ca_name = crypto_x509.Name([ - crypto_x509.NameAttribute(NameOID.COMMON_NAME, "CertificateVerify test CA") -]) -cert_name = crypto_x509.Name([ - crypto_x509.NameAttribute(NameOID.COMMON_NAME, "127.0.0.1") -]) -now = datetime.now(timezone.utc) -ca_cert = (crypto_x509.CertificateBuilder() - .subject_name(ca_name) - .issuer_name(ca_name) - .public_key(ca_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=True, path_length=None), True) - .add_extension(crypto_x509.SubjectKeyIdentifier.from_public_key(ca_key.public_key()), False) - .add_extension(crypto_x509.KeyUsage(False, False, False, False, False, True, True, None, None), True) - .sign(ca_key, hashes.SHA256())) -ca_certfile = get_temp_file() -with open(ca_certfile, "wb") as ca_cert_file: - ca_cert_file.write(ca_cert.public_bytes(serialization.Encoding.PEM)) - -trusted_server_cert = (crypto_x509.CertificateBuilder() - .subject_name(cert_name) - .issuer_name(ca_name) - .public_key(cert_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=False, path_length=None), True) - .add_extension(crypto_x509.SubjectKeyIdentifier.from_public_key(cert_key.public_key()), False) - .add_extension(crypto_x509.AuthorityKeyIdentifier.from_issuer_public_key(ca_key.public_key()), False) - .add_extension(crypto_x509.KeyUsage(True, False, True, False, False, False, False, None, None), True) - .add_extension(crypto_x509.SubjectAlternativeName([ - crypto_x509.IPAddress(ipaddress.ip_address("127.0.0.1")) - ]), False) - .add_extension(crypto_x509.ExtendedKeyUsage([ - ExtendedKeyUsageOID.SERVER_AUTH - ]), False) - .sign(ca_key, hashes.SHA256())) -trusted_server_certfile = get_temp_file() -with open(trusted_server_certfile, "wb") as cert_file: - cert_file.write(trusted_server_cert.public_bytes(serialization.Encoding.PEM)) - -def run_server_certificate_verify_case(signing_key): - marker = b"server proof accepted" - probe = socket.socket() - probe.bind(("127.0.0.1", 0)) - port = probe.getsockname()[1] - probe.close() - server = TLSServerAutomaton( - server="127.0.0.1", - sport=port, - mycert=trusted_server_certfile, - mykey=signing_key, - is_echo_server=True, - max_client_idle_time=0.3, - debug=0, - ) - server_thread = threading.Thread(target=server.run, daemon=True) - output = io.StringIO() - with contextlib.redirect_stdout(output): - server_thread.start() - for _ in range(100): - if server.serversocket is not None: - break - time.sleep(0.01) - client = TLSClientAutomaton( - server="127.0.0.1", - dport=port, - version="tls13", - cafile=ca_certfile, - data=[marker, b"stop_server", b"quit"], - debug=0, - ) - client.run() - server_thread.join(timeout=0.5) - if server_thread.is_alive(): - server.mykey = PrivKey(server_key) - stopper = TLSClientAutomaton( - server="127.0.0.1", - dport=port, - version="tls13", - cafile=ca_certfile, - data=[b"stop_server", b"quit"], - debug=0, - ) - stopper.run() - server_thread.join(timeout=1) - return ("> Received: %r" % marker) in output.getvalue() - -mismatched_key = scapy_path("/test/scapy/layers/tls/pki/cli_key.pem") -assert not run_server_certificate_verify_case(mismatched_key) -assert run_server_certificate_verify_case(server_key) - = Load native server util functions # Imports @@ -615,117 +472,6 @@ 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, CertTree -from scapy.layers.tls.automaton_cli import _verify_server_certificate -from unittest.mock import patch - -def make_test_cert(name, key, issuer, issuer_key, ca=False, age=timedelta()): - now = datetime.now(timezone.utc) - age - 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_root_cert = Cert(cryptography_obj=test_root) -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_cert], "example.test") -assert not _verify_server_certificate([test_leaf_cert], [test_root_cert], "wrong.example") - -# Explicit anchors take precedence over the system store. -with patch("scapy.layers.tls.cert.ssl.create_default_context", - side_effect=AssertionError("system store consulted")): - assert _verify_server_certificate( - [test_leaf_cert], [test_root_cert], "example.test") - -# An empty system store must not fall back to a peer-provided self-signed root. -class _EmptySystemStore: - def get_ca_certs(self, binary_form=False): - assert binary_form - return [] - -with patch("scapy.layers.tls.cert.ssl.create_default_context", - return_value=_EmptySystemStore()): - try: - CertTree([test_root_cert], load_system_store=True) - assert False, "accepted a peer root when the system store was empty" - except ValueError: - pass - -# No trust anchors at all must fail closed, not fall back to trusting the -# self-signed certificates the peer happened to send. -assert not _verify_server_certificate([test_leaf_cert], [], "example.test") - -# A certificate signed by someone we do not trust. -other_root_key = rsa.generate_private_key(public_exponent=65537, key_size=2048) -other_root, other_root_name = make_test_cert("other root", other_root_key, crypto_x509.Name([ - crypto_x509.NameAttribute(NameOID.COMMON_NAME, "other root")]), other_root_key, True) -assert not _verify_server_certificate( - [test_leaf_cert], [Cert(cryptography_obj=other_root)], "example.test") - -# An expired certificate, and an expired issuer for a leaf that is still current. -expired_leaf, _ = make_test_cert("example.test", test_leaf_key, test_root_name, - test_root_key, age=timedelta(days=10)) -assert not _verify_server_certificate( - [Cert(cryptography_obj=expired_leaf)], [test_root_cert], "example.test") - -old_root_key = rsa.generate_private_key(public_exponent=65537, key_size=2048) -old_root, old_root_name = make_test_cert("old root", old_root_key, crypto_x509.Name([ - crypto_x509.NameAttribute(NameOID.COMMON_NAME, "old root")]), old_root_key, True, - age=timedelta(days=10)) -fresh_leaf, _ = make_test_cert("example.test", test_leaf_key, old_root_name, old_root_key) -assert not _verify_server_certificate( - [Cert(cryptography_obj=fresh_leaf)], [Cert(cryptography_obj=old_root)], "example.test") - -= CertTree.verify checks the name and the validity period - -from scapy.layers.tls.cert import CertTree - -wildcard_leaf, _ = make_test_cert("*.wild.test", test_leaf_key, test_root_name, test_root_key) -wildcard_cert = Cert(cryptography_obj=wildcard_leaf) -tree = CertTree([wildcard_cert, test_root_cert], [test_root_cert]) - -tree.verify(wildcard_cert) -tree.verify(wildcard_cert, hostname="host.wild.test") - -# A wildcard covers one label, and only the leftmost one. -for rejected in ["wild.test", "a.b.wild.test", "host.other.test"]: - try: - tree.verify(wildcard_cert, hostname=rejected) - assert False, "accepted %s" % rejected - except ValueError: - pass - -# subjectAltName wins outright: the Common Name is not consulted beside it. -assert wildcard_cert.subjectAltName == [("DNS", "*.wild.test")] -assert not wildcard_cert.matchesHostname("test root") # Define server From e591fc1bcb67a6c177bfe571ab872ae48434a81f Mon Sep 17 00:00:00 2001 From: gpotter2 <10530980+gpotter2@users.noreply.github.com> Date: Sat, 26 Sep 2026 17:44:51 +0200 Subject: [PATCH 5/5] Re-enable TLS 1.3 tests on Windows AI-Assisted: no --- scapy/layers/tls/automaton_cli.py | 2 +- test/configs/windows.utsc | 1 - test/scapy/layers/tls/tlsclientserver.uts | 4 ---- 3 files changed, 1 insertion(+), 6 deletions(-) diff --git a/scapy/layers/tls/automaton_cli.py b/scapy/layers/tls/automaton_cli.py index f2eb6f246f8..65cc186175c 100644 --- a/scapy/layers/tls/automaton_cli.py +++ b/scapy/layers/tls/automaton_cli.py @@ -44,7 +44,7 @@ from scapy.config import conf from scapy.utils import randstring, repr_hex from scapy.automaton import ATMT, select_objects -from scapy.error import log_runtime, warning +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 diff --git a/test/configs/windows.utsc b/test/configs/windows.utsc index a38f065e8ca..22973c4c77e 100644 --- a/test/configs/windows.utsc +++ b/test/configs/windows.utsc @@ -30,7 +30,6 @@ "broken_windows", "ipv6", "linux", - "native_tls13", "mock_read_routes_bsd", "open_ssl_client", "osx", diff --git a/test/scapy/layers/tls/tlsclientserver.uts b/test/scapy/layers/tls/tlsclientserver.uts index 4e97cd99730..ea03e103849 100644 --- a/test/scapy/layers/tls/tlsclientserver.uts +++ b/test/scapy/layers/tls/tlsclientserver.uts @@ -572,15 +572,11 @@ def test_tls_client_native(post_handshake_auth=False, assert not server.is_alive() -# XXX: Ugh, Appveyor uses an ancient Windows 10 build that doesn't support TLS 1.3 natively. - = Testing TLS client against ssl.SSLContext server with TLS 1.3 and a post-handshake authentication -~ native_tls13 test_tls_client_native(post_handshake_auth=True) = Testing TLS client against ssl.SSLContext server with TLS 1.3 and a Hello-Retry request -~ native_tls13 test_tls_client_native(with_hello_retry=True)