""" test3_mutual_tls_fix.py -- Test 3, with the NEGATIVE CONTROL and the reverse-direction case added. Why this exists --------------- Original Test 3 proved: an identity-checking endpoint rejects Device B's genuine, CA-signed certificate. That shows the identity check *fires*. It does NOT, on its own, show the check is *necessary*. To prove necessity you need the control -- an endpoint that verifies CA-validity ONLY (no identity binding) and is shown to ACCEPT Device B's cert. That acceptance is the empirical content behind the claim "CA-validity alone is just Test 2's shared-secret flaw wearing a TLS costume." Without running it, the contrast between the weak fix and the real fix is asserted, not demonstrated -- which is the same critique the README correctly levels at Test 2. What this runs -------------- STRICT endpoint (mutual TLS: CA verify + per-device identity via SAN): [1] Device A's own client cert -> GRANTED [2] no client cert at all -> REJECTED (handshake) [3] Device B's genuine CA-signed client cert -> DENIED (identity mismatch) NAIVE endpoint (mutual TLS: CA verify ONLY -- the "TLS costume" weak fix): [4] Device B's genuine CA-signed client cert -> GRANTED <-- THE CONTROL Reverse direction (client checks the SERVER's identity): [5] rogue server presenting Device B's cert, client expecting 'device-a' -> REJECTED (server identity) [4] is what makes [3] mean something: it demonstrates that without the identity check, any fleet cert opens any device -- fleet-wide compromise from one leaked-but-CA-valid cert, structurally identical to Test 2. [5] closes the one-directional limitation the original README disclosed: both ends now bind identity, which is what "mutual" in CIP Security means. Identity is bound via SubjectAlternativeName (DNSName), NOT CommonName. CN-as-identity is deprecated because CN carries no authorization semantics; the original Test 3 used CN, which models the right idea using the wrong field. Server-side client identity is checked against the peer cert's SAN; client-side server identity is checked by the TLS stack itself via check_hostname against server_hostname (the standard, correct mechanism). Note on TLS version: both ends are pinned to TLS 1.2 so that a missing client certificate under CERT_REQUIRED fails deterministically *during the handshake*. Under TLS 1.3 that failure can surface post-handshake instead, which makes case [2] nondeterministic in a demo. Realism cost is negligible; determinism of the demonstration is the point. Run: python test3_mutual_tls_fix_v2.py """ import datetime import socket import ssl import tempfile import threading import time from pathlib import Path from cryptography import x509 from cryptography.hazmat.primitives import hashes, serialization from cryptography.hazmat.primitives.asymmetric import rsa from cryptography.x509.oid import NameOID WORK = Path(tempfile.mkdtemp(prefix="cip_test3v2_")) HOST = "127.0.0.1" PORT_STRICT, PORT_NAIVE, PORT_ROGUE = 60443, 60444, 60445 def _key(): return rsa.generate_private_key(public_exponent=65537, key_size=2048) def make_ca(): key = _key() name = x509.Name([x509.NameAttribute(NameOID.COMMON_NAME, "TestFleet Root CA")]) cert = ( x509.CertificateBuilder() .subject_name(name).issuer_name(name) .public_key(key.public_key()) .serial_number(x509.random_serial_number()) .not_valid_before(datetime.datetime.now(datetime.timezone.utc)) .not_valid_after(datetime.datetime.now(datetime.timezone.utc) + datetime.timedelta(days=1)) .add_extension(x509.BasicConstraints(ca=True, path_length=None), critical=True) .sign(key, hashes.SHA256()) ) return key, cert def make_cert(ca_key, ca_cert, name): """Identity lives in the SAN (DNSName == device name), not just the CN.""" key = _key() subj = x509.Name([x509.NameAttribute(NameOID.COMMON_NAME, name)]) cert = ( x509.CertificateBuilder() .subject_name(subj).issuer_name(ca_cert.subject) .public_key(key.public_key()) .serial_number(x509.random_serial_number()) .not_valid_before(datetime.datetime.now(datetime.timezone.utc)) .not_valid_after(datetime.datetime.now(datetime.timezone.utc) + datetime.timedelta(days=1)) .add_extension(x509.SubjectAlternativeName([x509.DNSName(name)]), critical=False) .sign(ca_key, hashes.SHA256()) ) return key, cert def write_pem(key_path, key, cert_path, cert): key_path.write_bytes( key.private_bytes( serialization.Encoding.PEM, serialization.PrivateFormat.TraditionalOpenSSL, serialization.NoEncryption(), ) ) cert_path.write_bytes(cert.public_bytes(serialization.Encoding.PEM)) def san_dnsnames(der_bytes): cert = x509.load_der_x509_certificate(der_bytes) try: san = cert.extensions.get_extension_for_class(x509.SubjectAlternativeName).value return san.get_values_for_type(x509.DNSName) except x509.ExtensionNotFound: return [] class Endpoint(threading.Thread): """A TLS server. If expected_client_identity is None it performs NO identity check (the naive/weak fix); otherwise it requires the peer cert's SAN to contain that identity (the real fix).""" def __init__(self, port, server_cert, server_key, ca_cert, require_client_cert=True, expected_client_identity=None): super().__init__(daemon=True) self.port = port self.expected = expected_client_identity ctx = ssl.SSLContext(ssl.PROTOCOL_TLS_SERVER) ctx.maximum_version = ssl.TLSVersion.TLSv1_2 # deterministic client-auth failure ctx.load_cert_chain(str(server_cert), str(server_key)) if require_client_cert: ctx.load_verify_locations(str(ca_cert)) ctx.verify_mode = ssl.CERT_REQUIRED else: ctx.verify_mode = ssl.CERT_NONE self.ctx = ctx self.sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM) self.sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1) self.sock.bind((HOST, port)) self.sock.listen(5) self.running = True def run(self): while self.running: try: raw, _ = self.sock.accept() except OSError: break try: tls = self.ctx.wrap_socket(raw, server_side=True) except ssl.SSLError: try: raw.close() except OSError: pass continue try: if self.expected is None: tls.sendall(b"ACCESS GRANTED (cert is CA-valid; NO identity check performed)") else: ids = san_dnsnames(tls.getpeercert(binary_form=True)) if self.expected in ids: tls.sendall(f"ACCESS GRANTED (client identity {ids} matches)".encode()) else: tls.sendall( f"ACCESS DENIED (cert is CA-valid, but identity {ids} " f"!= expected '{self.expected}')".encode() ) tls.close() except (OSError, ssl.SSLError): pass def stop(self): self.running = False try: self.sock.close() except OSError: pass def client(port, ca_cert, expected_server_name, client_cert=None, client_key=None): ctx = ssl.SSLContext(ssl.PROTOCOL_TLS_CLIENT) ctx.maximum_version = ssl.TLSVersion.TLSv1_2 ctx.load_verify_locations(str(ca_cert)) ctx.check_hostname = True # client binds the SERVER's identity ctx.verify_mode = ssl.CERT_REQUIRED if client_cert: ctx.load_cert_chain(str(client_cert), str(client_key)) try: with socket.create_connection((HOST, port), timeout=3) as raw: with ctx.wrap_socket(raw, server_hostname=expected_server_name) as tls: return tls.recv(1024).decode() except ssl.SSLCertVerificationError as e: return f"REJECTED (server identity mismatch): {e.verify_message or e}" except ssl.SSLError as e: return f"REJECTED (handshake): {e.__class__.__name__}: {e}" def main(): print("[*] Building CA + two individually-unique device certs (identity in SAN).\n") ca_key, ca_cert = make_ca() a_key, a_cert = make_cert(ca_key, ca_cert, "device-a") b_key, b_cert = make_cert(ca_key, ca_cert, "device-b") ca_path = WORK / "ca.pem"; ca_path.write_bytes(ca_cert.public_bytes(serialization.Encoding.PEM)) a_key_p, a_crt_p = WORK / "a.key", WORK / "a.pem"; write_pem(a_key_p, a_key, a_crt_p, a_cert) b_key_p, b_crt_p = WORK / "b.key", WORK / "b.pem"; write_pem(b_key_p, b_key, b_crt_p, b_cert) strict = Endpoint(PORT_STRICT, a_crt_p, a_key_p, ca_path, require_client_cert=True, expected_client_identity="device-a") naive = Endpoint(PORT_NAIVE, a_crt_p, a_key_p, ca_path, require_client_cert=True, expected_client_identity=None) rogue = Endpoint(PORT_ROGUE, b_crt_p, b_key_p, ca_path, require_client_cert=False, expected_client_identity=None) for e in (strict, naive, rogue): e.start() time.sleep(0.5) try: print("STRICT endpoint -- mutual TLS, CA verify + per-device identity (SAN):") r1 = client(PORT_STRICT, ca_path, "device-a", a_crt_p, a_key_p) print(" [1] Device A's own cert ->", r1) r2 = client(PORT_STRICT, ca_path, "device-a", None, None) print(" [2] no client cert ->", r2) r3 = client(PORT_STRICT, ca_path, "device-a", b_crt_p, b_key_p) print(" [3] Device B's CA-valid cert ->", r3) print("\nNAIVE endpoint -- mutual TLS, CA verify ONLY (the 'TLS costume'):") r4 = client(PORT_NAIVE, ca_path, "device-a", b_crt_p, b_key_p) print(" [4] Device B's CA-valid cert ->", r4, " <-- THE CONTROL") print("\nREVERSE direction -- client binds the server's identity:") r5 = client(PORT_ROGUE, ca_path, "device-a", a_crt_p, a_key_p) print(" [5] rogue server = device-b ->", r5) print("\n" + "=" * 72) strict_ok = ("GRANTED" in r1) and ("REJECTED" in r2) and ("DENIED" in r3) control_shows_hole = "GRANTED" in r4 # naive accepts a foreign fleet cert reverse_ok = "REJECTED" in r5 if strict_ok and control_shows_hole and reverse_ok: print("[!] The identity check REJECTS Device B (strict, case 3) -- but the") print(" same Device B cert is ACCEPTED when only CA-validity is checked") print(" (naive, case 4). Case 4 is the control: it demonstrates that") print(" 'validly CA-signed' alone == fleet-wide access == Test 2 in TLS") print(" clothing. Case 3 only means something because case 4 shows what") print(" happens without it. Case 5 shows the binding holds in BOTH") print(" directions, which is what makes CIP Security's mutual auth mutual.") else: print("[?] Unexpected result set -- inspect above.") print(f" strict_ok={strict_ok} control_shows_hole={control_shows_hole} reverse_ok={reverse_ok}") finally: for e in (strict, naive, rogue): e.stop() if __name__ == "__main__": main()