|
| 1 | +#!/usr/bin/env python3 |
| 2 | +"""Real independent fixture generation and strict TLS handshakes; no server library.""" |
| 3 | +import os |
| 4 | +from pathlib import Path |
| 5 | +import shlex |
| 6 | +import shutil |
| 7 | +import ssl |
| 8 | +import subprocess |
| 9 | +import tempfile |
| 10 | +import unittest |
| 11 | + |
| 12 | +SOURCE = Path(__file__).resolve().parent.parent |
| 13 | + |
| 14 | + |
| 15 | +class TLSFixtures(unittest.TestCase): |
| 16 | + @classmethod |
| 17 | + def setUpClass(cls): |
| 18 | + cls.temporary = tempfile.TemporaryDirectory(prefix='v3-tls-fixtures-') |
| 19 | + cls.addClassCleanup(cls.temporary.cleanup) |
| 20 | + cls.root = Path(cls.temporary.name).resolve() |
| 21 | + cls.fixtures = cls.root / 'fixtures' |
| 22 | + shutil.copytree(SOURCE / 'test/data/tls_credentials', cls.fixtures) |
| 23 | + executable = cls.root / ('client.exe' if os.name == 'nt' else 'client') |
| 24 | + cls.executable = executable |
| 25 | + command = (shlex.split(os.environ.get('CXX', 'c++')) + ['-std=c++20'] |
| 26 | + + shlex.split(os.environ.get('V3_TLS_CFLAGS', '')) |
| 27 | + + [str(SOURCE / 'test/installed/tls_client.cpp')] |
| 28 | + + shlex.split(os.environ.get('V3_TLS_LIBS', '-lssl -lcrypto')) |
| 29 | + + (['-lws2_32'] if os.name == 'nt' else []) + ['-o', str(executable)]) |
| 30 | + subprocess.run(command, check=True, capture_output=True, text=True, timeout=60) |
| 31 | + subprocess.run([str(executable), 'fixtures', str(cls.fixtures)], check=True, |
| 32 | + capture_output=True, text=True, timeout=10) |
| 33 | + |
| 34 | + def handshake(self, identity, host='a.example', trust='root', client=None): |
| 35 | + """Exchange real TLS records in memory, including certificate/hostname verification.""" |
| 36 | + chain = self.root / (identity + '-chain.pem') |
| 37 | + chain.write_bytes((self.fixtures / (identity + '.pem')).read_bytes() |
| 38 | + + ((self.fixtures / 'intermediate.pem').read_bytes() |
| 39 | + if identity in ('a', 'b') else b'')) |
| 40 | + server = ssl.SSLContext(ssl.PROTOCOL_TLS_SERVER) |
| 41 | + server.load_cert_chain(str(chain), str(self.fixtures / (identity + '-key.pem'))) |
| 42 | + if client: |
| 43 | + server.load_verify_locations(str(self.fixtures / 'client-root.pem')) |
| 44 | + server.verify_mode = ssl.CERT_REQUIRED |
| 45 | + context = ssl.create_default_context(cafile=str(self.fixtures / (trust + '.pem'))) |
| 46 | + # Python 3.14 enables this by default; exercise the same policy on older Python. |
| 47 | + context.verify_flags |= ssl.VERIFY_X509_STRICT |
| 48 | + self.assertTrue(context.check_hostname) |
| 49 | + self.assertEqual(ssl.CERT_REQUIRED, context.verify_mode) |
| 50 | + if client: |
| 51 | + client_chain = self.root / (client + '-chain.pem') |
| 52 | + client_chain.write_bytes((self.fixtures / (client + '.pem')).read_bytes() |
| 53 | + + ((self.fixtures / 'client-intermediate.pem').read_bytes() |
| 54 | + if client != 'client-other' else b'')) |
| 55 | + context.load_cert_chain(str(client_chain), str(self.fixtures / (client + '-key.pem'))) |
| 56 | + inbound, outbound = [ssl.MemoryBIO(), ssl.MemoryBIO()], [ssl.MemoryBIO(), ssl.MemoryBIO()] |
| 57 | + peers = [context.wrap_bio(inbound[0], outbound[0], server_hostname=host), |
| 58 | + server.wrap_bio(inbound[1], outbound[1], server_side=True)] |
| 59 | + done = [False, False] |
| 60 | + for _ in range(20): |
| 61 | + for index, peer in enumerate(peers): |
| 62 | + if not done[index]: |
| 63 | + try: |
| 64 | + peer.do_handshake() |
| 65 | + done[index] = True |
| 66 | + except ssl.SSLWantReadError: |
| 67 | + pass |
| 68 | + inbound[1 - index].write(outbound[index].read()) |
| 69 | + if all(done): |
| 70 | + peers[0].write(b'fixture-ok') |
| 71 | + inbound[1].write(outbound[0].read()) |
| 72 | + self.assertEqual(b'fixture-ok', peers[1].read()) |
| 73 | + return |
| 74 | + self.fail('TLS handshake did not complete within 20 record exchanges') |
| 75 | + |
| 76 | + def test_static_sni_chains_pass_strict_handshake(self): |
| 77 | + for name in ('a', 'b'): |
| 78 | + with self.subTest(identity=name): |
| 79 | + self.handshake(name, name + '.example') |
| 80 | + |
| 81 | + def test_rotation_chain_passes_strict_handshake(self): |
| 82 | + self.handshake('a-replacement') |
| 83 | + |
| 84 | + def test_wrong_hostname_is_rejected(self): |
| 85 | + for name in ('a', 'b', 'a-replacement'): |
| 86 | + with self.subTest(identity=name): |
| 87 | + with self.assertRaises(ssl.SSLCertVerificationError) as error: |
| 88 | + self.handshake(name, 'wrong.example') |
| 89 | + self.assertEqual(62, error.exception.verify_code, str(error.exception)) |
| 90 | + |
| 91 | + def test_untrusted_chain_is_rejected(self): |
| 92 | + for name in ('a', 'b', 'a-replacement'): |
| 93 | + with self.subTest(identity=name): |
| 94 | + with self.assertRaises(ssl.SSLCertVerificationError) as error: |
| 95 | + self.handshake(name, name + '.example' if name == 'b' else 'a.example', 'client-root') |
| 96 | + self.assertIn(error.exception.verify_code, (20, 21), str(error.exception)) |
| 97 | + |
| 98 | + def test_server_fixture_regeneration_preserves_leaf_material(self): |
| 99 | + regenerated = self.root / 'regenerated' |
| 100 | + shutil.copytree(SOURCE / 'test/data/tls_credentials', regenerated) |
| 101 | + result = subprocess.run([str(self.executable), 'regenerate-server', str(regenerated)], |
| 102 | + capture_output=True, text=True, timeout=10) |
| 103 | + self.assertEqual(0, result.returncode, result.stdout + result.stderr) |
| 104 | + for path in (SOURCE / 'test/data/tls_credentials').iterdir(): |
| 105 | + if path.name not in ('root.pem', 'intermediate.pem', 'a.pem', 'b.pem'): |
| 106 | + self.assertEqual(path.read_bytes(), (regenerated / path.name).read_bytes(), path.name) |
| 107 | + for name in ('root', 'intermediate', 'a', 'b'): |
| 108 | + flags = ['-noout', '-dates', '-serial', '-subject', '-issuer'] |
| 109 | + if name in ('a', 'b'): |
| 110 | + flags.append('-pubkey') |
| 111 | + def metadata(directory): |
| 112 | + return subprocess.run([os.environ.get('OPENSSL', 'openssl'), 'x509', |
| 113 | + '-in', str(directory / (name + '.pem'))] + flags, |
| 114 | + check=True, capture_output=True, timeout=5).stdout |
| 115 | + self.assertEqual(metadata(SOURCE / 'test/data/tls_credentials'), metadata(regenerated)) |
| 116 | + self.fixtures = regenerated |
| 117 | + self.handshake('a') |
| 118 | + self.handshake('b', 'b.example') |
| 119 | + |
| 120 | + def test_valid_mtls_client_is_accepted(self): |
| 121 | + self.handshake('a', client='client-valid') |
| 122 | + |
| 123 | + def test_negative_mtls_clients_are_rejected(self): |
| 124 | + for client in ('client-other', 'client-expired', 'client-wrong-purpose'): |
| 125 | + with self.subTest(client=client): |
| 126 | + with self.assertRaises(ssl.SSLCertVerificationError) as error: |
| 127 | + self.handshake('a', client=client) |
| 128 | + self.assertEqual({'client-other': 20, 'client-expired': 10, |
| 129 | + 'client-wrong-purpose': 26}[client], error.exception.verify_code) |
| 130 | + |
| 131 | + |
| 132 | +if __name__ == '__main__': |
| 133 | + unittest.main() |
0 commit comments