From d9cb53bdb30f55d32336d406191bf82433569381 Mon Sep 17 00:00:00 2001 From: Clinton Thomas <1033162+KernelClint@users.noreply.github.com> Date: Tue, 1 Sep 2026 14:15:32 -0400 Subject: [PATCH 1/2] fwdmachine: verify upstream TLS certificates AI-Assisted: yes (GPT-5.6-Cyber) --- scapy/fwdmachine.py | 17 ++++-- test/scapy/layers/tls/tlsclientserver.uts | 74 +++++++++++++++++++++++ 2 files changed, 86 insertions(+), 5 deletions(-) diff --git a/scapy/fwdmachine.py b/scapy/fwdmachine.py index 1fb97495d56..cbdb78b26e6 100644 --- a/scapy/fwdmachine.py +++ b/scapy/fwdmachine.py @@ -114,6 +114,7 @@ def __init__( remote_af: Optional[socket.AddressFamily] = None, bind_address: str = None, tls: bool = False, + verify_upstream: bool = True, crtfile: Optional[str] = None, keyfile: Optional[str] = None, keyfilepwd: Optional[str] = None, @@ -128,6 +129,7 @@ def __init__( self.remote_af = remote_af if remote_af is not None else af self.proto = proto self.tls = tls + self.verify_upstream = verify_upstream self.crtfile = crtfile self.keyfile = keyfile self.keyfilepwd = keyfilepwd @@ -373,10 +375,13 @@ def handler(self, sock, addr, dest): # Wrap both server and peer sockets in SSL if self.tls: # Build client SSL context - clisslcontext = ssl.SSLContext(ssl.PROTOCOL_TLS) - clisslcontext.load_default_certs() - clisslcontext.check_hostname = False - clisslcontext.verify_mode = ssl.CERT_NONE + if self.verify_upstream: + clisslcontext = ssl.create_default_context() + else: + clisslcontext = ssl.SSLContext(ssl.PROTOCOL_TLS) + clisslcontext.load_default_certs() + clisslcontext.check_hostname = False + clisslcontext.verify_mode = ssl.CERT_NONE # This acts as follows: # - start the server-side TLS handshake @@ -393,7 +398,9 @@ def cb_sni(sock, server_name, _): ss = _clisock[0] ctx.tls_sni_name = server_name # the requested SNI # Use that SNI to wrap the client socket - ss = clisslcontext.wrap_socket(ss, server_hostname=server_name) + ss = clisslcontext.wrap_socket( + ss, server_hostname=server_name or dest[0] + ) # Get certificate chain cas = ss._sslobj.get_unverified_chain() if self.crtfile is None: diff --git a/test/scapy/layers/tls/tlsclientserver.uts b/test/scapy/layers/tls/tlsclientserver.uts index 7a552e3c779..de930a4e648 100644 --- a/test/scapy/layers/tls/tlsclientserver.uts +++ b/test/scapy/layers/tls/tlsclientserver.uts @@ -577,3 +577,77 @@ def _test_connection(): assert b"" in pkt[HTTPResponse].load retry_test(_test_connection) + +############ +############ ++ ForwardMachine upstream TLS authentication +~ crypto + += The upstream context verifies certificates by default, and can be turned off + +from unittest.mock import patch +from scapy.fwdmachine import ForwardMachine + +class _Machine(ForwardMachine): + def _getpeersock(self, dest, ctx, server_hostname=None): + return object() + +def _contexts_used(verify): + machine = object.__new__(_Machine) + machine.tls = True + machine.verify_upstream = verify + with patch( + "scapy.fwdmachine.ssl.create_default_context", + side_effect=RuntimeError("stop"), + ) as verifying, patch( + "scapy.fwdmachine.ssl.SSLContext", side_effect=RuntimeError("stop") + ) as unverifying: + try: + machine.handler(None, ("127.0.0.1", 1), ("upstream.test", 443)) + except RuntimeError: + pass + return verifying.call_count, unverifying.call_count + +assert _contexts_used(True) == (1, 0) +assert _contexts_used(False) == (0, 1) + += A client that sends no SNI is authenticated against the destination host + +from unittest.mock import patch +from scapy.config import conf +from scapy.fwdmachine import ForwardMachine + +class _Machine(ForwardMachine): + def _getpeersock(self, dest, ctx, server_hostname=None): + return object() + +class _ClientContext: + def __init__(self): + self.hostnames = [] + def wrap_socket(self, sock, server_hostname=None): + self.hostnames.append(server_hostname) + raise RuntimeError("stop before the handshake") + +class _Sock: + def close(self): + pass + +def _hostname_for(server_name): + client = _ClientContext() + class _ServerContext: + def __init__(self, *args, **kwargs): + self.sni_callback = None + def wrap_socket(self, sock, server_side=False): + self.sni_callback(None, server_name, None) + raise RuntimeError("unreachable") + machine = object.__new__(_Machine) + machine.tls = True + machine.verify_upstream = True + machine.ct = conf.color_theme + with patch("scapy.fwdmachine.ssl.create_default_context", return_value=client), \ + patch("scapy.fwdmachine.ssl.SSLContext", _ServerContext): + machine.handler(_Sock(), ("127.0.0.1", 1), ("upstream.test", 443)) + return client.hostnames + +assert _hostname_for("sni.test") == ["sni.test"] +assert _hostname_for(None) == ["upstream.test"] From aaa35846adbdcc1ed34581b2674b58467b28b2b7 Mon Sep 17 00:00:00 2001 From: Clinton Thomas <1033162+KernelClint@users.noreply.github.com> Date: Thu, 3 Sep 2026 13:54:18 -0400 Subject: [PATCH 2/2] fwdmachine: rename the flag to no_check_certificate AI-Assisted: yes (GPT-5.6-Cyber) --- scapy/fwdmachine.py | 15 ++++++++------- test/scapy/layers/tls/tlsclientserver.uts | 10 +++++----- 2 files changed, 13 insertions(+), 12 deletions(-) diff --git a/scapy/fwdmachine.py b/scapy/fwdmachine.py index cbdb78b26e6..86b601aa186 100644 --- a/scapy/fwdmachine.py +++ b/scapy/fwdmachine.py @@ -81,6 +81,8 @@ class ForwardMachine: but "2.2.2.2" by default in TPROXY (if you are using the provided 'vethrelay.sh' script). :param tls: enable TLS (in both the server and client) + :param no_check_certificate: with TLS, do not check the certificate of the + upstream server. :param crtfile: (optional) if provided, uses a certificate instead of self signed ones. :param keyfile: (optional) path to the key file @@ -114,7 +116,7 @@ def __init__( remote_af: Optional[socket.AddressFamily] = None, bind_address: str = None, tls: bool = False, - verify_upstream: bool = True, + no_check_certificate: bool = False, crtfile: Optional[str] = None, keyfile: Optional[str] = None, keyfilepwd: Optional[str] = None, @@ -129,7 +131,7 @@ def __init__( self.remote_af = remote_af if remote_af is not None else af self.proto = proto self.tls = tls - self.verify_upstream = verify_upstream + self.no_check_certificate = no_check_certificate self.crtfile = crtfile self.keyfile = keyfile self.keyfilepwd = keyfilepwd @@ -375,13 +377,12 @@ def handler(self, sock, addr, dest): # Wrap both server and peer sockets in SSL if self.tls: # Build client SSL context - if self.verify_upstream: - clisslcontext = ssl.create_default_context() - else: - clisslcontext = ssl.SSLContext(ssl.PROTOCOL_TLS) - clisslcontext.load_default_certs() + if self.no_check_certificate: + clisslcontext = ssl.SSLContext(ssl.PROTOCOL_TLS_CLIENT) clisslcontext.check_hostname = False clisslcontext.verify_mode = ssl.CERT_NONE + else: + clisslcontext = ssl.create_default_context() # This acts as follows: # - start the server-side TLS handshake diff --git a/test/scapy/layers/tls/tlsclientserver.uts b/test/scapy/layers/tls/tlsclientserver.uts index de930a4e648..74db7be504a 100644 --- a/test/scapy/layers/tls/tlsclientserver.uts +++ b/test/scapy/layers/tls/tlsclientserver.uts @@ -592,10 +592,10 @@ class _Machine(ForwardMachine): def _getpeersock(self, dest, ctx, server_hostname=None): return object() -def _contexts_used(verify): +def _contexts_used(no_check_certificate): machine = object.__new__(_Machine) machine.tls = True - machine.verify_upstream = verify + machine.no_check_certificate = no_check_certificate with patch( "scapy.fwdmachine.ssl.create_default_context", side_effect=RuntimeError("stop"), @@ -608,8 +608,8 @@ def _contexts_used(verify): pass return verifying.call_count, unverifying.call_count -assert _contexts_used(True) == (1, 0) -assert _contexts_used(False) == (0, 1) +assert _contexts_used(False) == (1, 0) +assert _contexts_used(True) == (0, 1) = A client that sends no SNI is authenticated against the destination host @@ -642,7 +642,7 @@ def _hostname_for(server_name): raise RuntimeError("unreachable") machine = object.__new__(_Machine) machine.tls = True - machine.verify_upstream = True + machine.no_check_certificate = False machine.ct = conf.color_theme with patch("scapy.fwdmachine.ssl.create_default_context", return_value=client), \ patch("scapy.fwdmachine.ssl.SSLContext", _ServerContext):