diff --git a/scapy/layers/tls/automaton_cli.py b/scapy/layers/tls/automaton_cli.py index 5e442d3f198..4c919a16a6b 100644 --- a/scapy/layers/tls/automaton_cli.py +++ b/scapy/layers/tls/automaton_cli.py @@ -38,6 +38,8 @@ import socket import binascii +import ipaddress +import ssl import struct import time @@ -80,12 +82,53 @@ from scapy.packet import Raw from scapy.compat import bytes_encode +if conf.crypto_valid: + from cryptography import x509 + try: + from cryptography.x509.verification import PolicyBuilder, Store + except ImportError: + PolicyBuilder = Store = None + # Typing imports from typing import ( Optional, ) +def _load_trust_anchors(cafile): + if not conf.crypto_valid or PolicyBuilder is None: + return [] + context = ssl.create_default_context(cafile=cafile) + return [ + x509.load_der_x509_certificate(der) + for der in context.get_ca_certs(binary_form=True) + ] + + +def _verify_server_certificate(certificates, trusted_certs, hostname): + if (not certificates or not trusted_certs or not conf.crypto_valid or + PolicyBuilder is None): + return False + try: + try: + subject = x509.IPAddress(ipaddress.ip_address(hostname)) + except ValueError: + subject = x509.DNSName(hostname) + verifier = PolicyBuilder().store( + Store(trusted_certs) + ).build_server_verifier(subject) + verifier.verify( + x509.load_der_x509_certificate(certificates[0].der), + [ + x509.load_der_x509_certificate(cert.der) + for cert in certificates[1:] + ], + ) + return True + except Exception: + return False + + class TLSClientAutomaton(_TLSAutomaton): """ A simple TLS test client automaton. Try to overload some states or @@ -97,6 +140,9 @@ class TLSClientAutomaton(_TLSAutomaton): :param server: the server IP or hostname. defaults to 127.0.0.1 :param dport: the server port. defaults to 4433 :param server_name: the SNI to use. It does not need to be set + :param cafile: optional CA certificate bundle used to authenticate the server. + By default, the system trust store is used. + :param verify: whether to authenticate the server certificate. Defaults to True. :param mycert: :param mykey: may be provided as filenames. They will be used in the (or post) handshake, should the server ask for client authentication. @@ -116,6 +162,7 @@ class TLSClientAutomaton(_TLSAutomaton): """ def parse_args(self, server="127.0.0.1", dport=4433, server_name=None, + cafile=None, verify=True, mycert=None, mykey=None, client_hello=None, version=None, resumption_master_secret=None, @@ -137,6 +184,11 @@ def parse_args(self, server="127.0.0.1", dport=4433, server_name=None, self.remote_ip = tmp[0][4][0] self.remote_port = dport self.server_name = server_name + self.expected_server_name = server_name or server + self.verify_server = verify + self.server_trust_anchors = ( + _load_trust_anchors(cafile) if verify else [] + ) self.local_ip = None self.local_port = None self.socket = None @@ -402,7 +454,22 @@ def should_handle_ServerCertificate(self): @ATMT.state() def HANDLED_SERVERCERTIFICATE(self): - pass + if self.verify_server: + self.cur_session.server_cert_valid = _verify_server_certificate( + self.cur_session.server_certs, + self.server_trust_anchors, + self.expected_server_name, + ) + if not self.cur_session.server_cert_valid: + raise self.INVALID_SERVER_CERTIFICATE() + + @ATMT.state() + def INVALID_SERVER_CERTIFICATE(self): + self.vprint("Server certificate verification failed!") + self.add_record() + self.add_msg(TLSAlert(level=2, descr=46)) + self.flush_records() + raise self.FINAL() @ATMT.condition(HANDLED_SERVERHELLO, prio=2) def missing_ServerCertificate(self): @@ -842,7 +909,14 @@ def sslv2_should_handle_ServerHello(self): @ATMT.state() def SSLv2_HANDLED_SERVERHELLO(self): - pass + if self.verify_server: + self.cur_session.server_cert_valid = _verify_server_certificate( + self.cur_session.server_certs, + self.server_trust_anchors, + self.expected_server_name, + ) + if not self.cur_session.server_cert_valid: + raise self.SSLv2_CLOSE_NOTIFY() @ATMT.condition(SSLv2_RECEIVED_SERVERHELLO, prio=2) def sslv2_missing_ServerHello(self): @@ -1341,7 +1415,14 @@ def tls13_should_handle_Certificate(self): @ATMT.state() def TLS13_HANDLED_CERTIFICATE(self): - pass + if self.verify_server: + self.cur_session.server_cert_valid = _verify_server_certificate( + self.cur_session.server_certs, + self.server_trust_anchors, + self.expected_server_name, + ) + if not self.cur_session.server_cert_valid: + raise self.INVALID_SERVER_CERTIFICATE() @ATMT.condition(TLS13_HANDLED_CERTIFICATE, prio=1) def tls13_should_handle_CertificateVerify(self): diff --git a/scapy/layers/tls/session.py b/scapy/layers/tls/session.py index de9f342bb46..f38066df8ee 100644 --- a/scapy/layers/tls/session.py +++ b/scapy/layers/tls/session.py @@ -411,6 +411,7 @@ def __init__(self, # to be sent by the server through a Certificate message. # The server certificate should be self.server_certs[0]. self.server_certs = [] + self.server_cert_valid = None # The server private key, as a PrivKey instance, when acting as server. # XXX It would be nice to be able to provide both an RSA and an ECDSA diff --git a/test/scapy/layers/tls/tlsclientserver.uts b/test/scapy/layers/tls/tlsclientserver.uts index 7a552e3c779..c640d599832 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,49 @@ with open(certfile, "wb") as fd: with open(keyfile, "wb") as fd: fd.write(rsa_key) += TLS client validates certificate trust and hostname + +from datetime import datetime, timedelta, timezone +from cryptography import x509 as crypto_x509 +from cryptography.hazmat.primitives import hashes +from cryptography.hazmat.primitives.asymmetric import rsa +from cryptography.x509.oid import ExtendedKeyUsageOID, NameOID +from scapy.layers.tls.cert import Cert +from scapy.layers.tls.automaton_cli import _verify_server_certificate + +def make_test_cert(name, key, issuer, issuer_key, ca=False): + now = datetime.now(timezone.utc) + subject = crypto_x509.Name([crypto_x509.NameAttribute(NameOID.COMMON_NAME, name)]) + cert = (crypto_x509.CertificateBuilder() + .subject_name(subject) + .issuer_name(issuer) + .public_key(key.public_key()) + .serial_number(crypto_x509.random_serial_number()) + .not_valid_before(now - timedelta(days=1)) + .not_valid_after(now + timedelta(days=1)) + .add_extension(crypto_x509.BasicConstraints(ca=ca, path_length=None), True) + .add_extension(crypto_x509.SubjectKeyIdentifier.from_public_key(key.public_key()), False) + .add_extension(crypto_x509.AuthorityKeyIdentifier.from_issuer_public_key(issuer_key.public_key()), False)) + if ca: + usage = crypto_x509.KeyUsage(False, False, False, False, False, True, True, None, None) + else: + usage = crypto_x509.KeyUsage(True, False, True, False, False, False, False, None, None) + cert = (cert.add_extension(crypto_x509.SubjectAlternativeName([crypto_x509.DNSName(name)]), False) + .add_extension(crypto_x509.ExtendedKeyUsage([ExtendedKeyUsageOID.SERVER_AUTH]), False)) + cert = cert.add_extension(usage, True).sign(issuer_key, hashes.SHA256()) + return cert, subject + +test_root_key = rsa.generate_private_key(public_exponent=65537, key_size=2048) +test_root, test_root_name = make_test_cert("test root", test_root_key, crypto_x509.Name([ + crypto_x509.NameAttribute(NameOID.COMMON_NAME, "test root")]), test_root_key, True) +test_leaf_key = rsa.generate_private_key(public_exponent=65537, key_size=2048) +test_leaf, _ = make_test_cert("example.test", test_leaf_key, test_root_name, test_root_key) +test_leaf_cert = Cert(cryptography_obj=test_leaf) + +assert _verify_server_certificate([test_leaf_cert], [test_root], "example.test") +assert not _verify_server_certificate([test_leaf_cert], [test_root], "wrong.example") +assert not _verify_server_certificate([test_leaf_cert], [], "example.test") + # Define server REQS = [ @@ -524,6 +570,7 @@ def test_tls_client_native(post_handshake_auth=False, server="127.0.0.1", dport=port, version="tls13", + verify=False, mycert=certfile, mykey=keyfile, # we select x25519 but the server enforces seco256r1, so a Hello Retry will be issued