diff --git a/scapy/layers/tls/automaton_cli.py b/scapy/layers/tls/automaton_cli.py index 5e442d3f198..7beb060850e 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): @@ -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): diff --git a/scapy/layers/tls/cert.py b/scapy/layers/tls/cert.py index b4abb8097f0..a5273efe1c7 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, @@ -133,6 +137,8 @@ X509_AttributeValue, X509_Cert, X509_CRL, + X509_DNSName, + X509_IPAddress, X509_SubjectPublicKeyInfo, ) from scapy.layers.tls.crypto.hash import _get_hash @@ -940,6 +946,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. @@ -983,6 +1038,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": @@ -995,6 +1055,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) @@ -1094,6 +1158,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 @@ -1131,7 +1242,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): @@ -1563,6 +1682,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 @@ -1575,10 +1695,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. @@ -1659,13 +1788,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 3b00023dab2..0c11edd83f7 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 1a0610f5b47..7fd464a5e9e 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 74db7be504a..9a0a08b9edc 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...") @@ -445,6 +448,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 = [ @@ -524,6 +639,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