For reading and verification only. You may not copy, use, modify or redistribute this code. Full notice on the main page.

src/all_mcp/tls_policy.py

SHA-256: 9278201c4df320f7666978ddb7741d44dfde15d3177dbcc329dd6472675ec7c0
985 lines · 52510 bytes · View on GitHub

1	"""The TLS policy of the ALL 2.0 terminator and the proof of what it can honestly claim (plan sections 2, 3 and 7; WP16).
2
3	Three things live here, and nothing else in the device decides them:
4
5	* The server context. TLS 1.3 at least, ALPN ``http/1.1`` (the two words come from ``policy.terminator``), the certificate the generation names. The terminator
6	refuses to start in exactly one case: the Python has no TLS 1.3 (``TlsUnavailable``, closed word ``tls13_required``). That is a reported ``degraded`` state
7	(e2e_status, the tray, the heartbeat), never a silent fallback. The OpenSSL version does not stop anything: it is reported per operating system and it gates
8	ONE claim, the hybrid key exchange (OpenSSL 3.5 brings X25519MLKEM768; an older one serves X25519 and says nothing about hybrid).
9	* A parser of the TLS hellos that cross the wire in the clear (``parse_client_hello``, ``observe_server``): version, cipher suite and the group the server
10	selected, read from the ServerHello bytes themselves, so a node observer, a pcap and the blindness gate (WP23) all read the same thing, and nothing depends on
11	what Python's ``ssl`` module is willing to tell. It follows a hello across records and across TCP segments, a HelloRetryRequest and an alert; malformed bytes are
12	``TlsParseError`` with a closed word, never an ``IndexError``.
13	* The capability report (``capability_report``): real loopback handshakes against the real server context (a client that offers hybrid, one that does not,
14	one that speaks only TLS 1.2), and from what they showed, exactly which claims this operating system can prove. It is the JSON that feeds ``e2e-proof.json`` and the
15	TLS sentences of the site (WP27): a sentence that is not in ``allowed_claims`` is not written.
16
17	``OPENSSL_CONF`` (the installer's OpenSSL config file) restricts the key-exchange groups and orders the TLS 1.3 suites. It was evaluated against the relay process and is
18	OFF by default: it is a process-wide setting, and a restricted list refuses a peer that offers only a group left out (a test shows one). It can be
19	turned on for the supervised e2e child alone (``runtime.env`` word ``ALL_E2E_OPENSSL_CONF=on``, ``apply_openssl_conf``), never for the relay process, and a runtime that
20	ignores it is detected (``openssl_conf_effective``) instead of trusted.
21
22	python -m all_mcp.tls_policy report [--out FILE] [--no-probe] # the per-OS report of this computer, JSON on stdout
23	"""
24	from __future__ import annotations
25
26	import argparse
27	import contextlib
28	import datetime
29	import json
30	import logging
31	import os
32	import platform
33	import queue
34	import ssl
35	import struct
36	import subprocess
37	import sys
38	import tempfile
39	import threading
40	from dataclasses import dataclass, field
41	from pathlib import Path
42	from typing import Any, Callable, Mapping, Sequence
43
44	from . import e2e_policy as policy
45	from .windows_console import hide_child_consoles, no_window_flags
46
47	log = logging.getLogger("all_mcp.tls_policy")
48
49	REPORT_VERSION = 1
50	TLS_13, TLS_12 = 0x0304, 0x0303
51	# Registry values (IANA TLS parameters), not policy: what a hello carries on the wire.
52	GROUP_X25519, GROUP_SECP256R1, GROUP_SECP384R1, GROUP_SECP521R1 = 0x001D, 0x0017, 0x0018, 0x0019
53	GROUP_X25519_MLKEM768, GROUP_SECP256R1_MLKEM768, GROUP_X25519_KYBER768_DRAFT = 0x11EC, 0x11EB, 0x6399
54	HYBRID_GROUPS = frozenset({GROUP_X25519_MLKEM768, GROUP_SECP256R1_MLKEM768, GROUP_X25519_KYBER768_DRAFT})
55	GROUP_NAMES = {
56	GROUP_X25519: "X25519", GROUP_SECP256R1: "secp256r1", GROUP_SECP384R1: "secp384r1", GROUP_SECP521R1: "secp521r1", 0x001E: "X448",
57	0x0100: "ffdhe2048", 0x0101: "ffdhe3072", 0x0102: "ffdhe4096", GROUP_X25519_MLKEM768: "X25519MLKEM768", GROUP_SECP256R1_MLKEM768: "SecP256r1MLKEM768",
58	GROUP_X25519_KYBER768_DRAFT: "X25519Kyber768Draft00",
59	}
60	CIPHER_NAMES = {
61	0x1301: "TLS_AES_128_GCM_SHA256", 0x1302: "TLS_AES_256_GCM_SHA384", 0x1303: "TLS_CHACHA20_POLY1305_SHA256", 0x1304: "TLS_AES_128_CCM_SHA256",
62	0x1305: "TLS_AES_128_CCM_8_SHA256",
63	}
64	ALERT_NAMES = {
65	0: "close_notify", 10: "unexpected_message", 20: "bad_record_mac", 40: "handshake_failure", 42: "bad_certificate", 47: "illegal_parameter",
66	70: "protocol_version", 71: "insufficient_security", 80: "internal_error", 86: "inappropriate_fallback", 90: "user_canceled", 109: "missing_extension",
67	112: "unrecognized_name", 120: "no_application_protocol",
68	}
69	VERSION_NAMES = {0x0304: "TLSv1.3", 0x0303: "TLSv1.2", 0x0302: "TLSv1.1", 0x0301: "TLSv1.0", 0x0300: "SSLv3"}
70	HELLO_RETRY_RANDOM = bytes.fromhex("cf21ad74e59a6111be1d8c021e65b891c2a211167abb8c5e079e09e2c8a8339c") # RFC 8446 section 4.1.3: SHA-256("HelloRetryRequest")
71	EXT_SERVER_NAME, EXT_SUPPORTED_GROUPS, EXT_ALPN, EXT_SUPPORTED_VERSIONS, EXT_KEY_SHARE, EXT_ECH = 0, 10, 16, 43, 51, 0xFE0D
72	RECORD_HANDSHAKE, RECORD_ALERT, RECORD_CCS, RECORD_APPLICATION = 22, 21, 20, 23
73	RECORD_HEADER_BYTES = 5
74	RECORD_MAX_BYTES = 16384 + 2048 # a handshake record is at most 2^14 (the field allows more only for ciphertext); the slack is for sloppy senders
75	HELLO_MAX_BYTES = 1 << 17 # the largest hello that is read: a ClientHello offering X25519MLKEM768 is about 1.5 KB, 128 KB is already abuse
76	# The OpenSSL that brings X25519MLKEM768 in its default groups; the one number of this file that is a fact about a library, not a choice of ours.
77	HYBRID_MIN_OPENSSL = (3, 5)
78	PROBE_NAME = "probe.all.invalid"
79	PROBE_SECONDS = 20.0
80	PROBE_ROUNDS = 16
81	# The claims this module can prove, in the order the report lists them, and the one sentence of evidence each needs. A claim not in this table is never written.
82	CLAIMS = {
83	"tls_1_3_minimum": "the terminator's context refuses below TLS 1.3 and a loopback handshake negotiated TLS 1.3",
84	"tls_1_2_refused": "a client that speaks only TLS 1.2 was refused with an alert",
85	"alpn_http_1_1": "a client that offered http/1.1 was given http/1.1",
86	"forward_secrecy": "the ServerHello selected an ephemeral key-exchange group (TLS 1.3 has no static one)",
87	"classical_fallback": "a client that offers no hybrid group still completed the handshake (X25519 or another classical group)",
88	"hybrid_x25519mlkem768": "OpenSSL 3.5 or later AND a client that offered X25519MLKEM768 was answered with X25519MLKEM768",
89	}
90	# What no loopback report can prove, with the reason (the plan's section 10): they wait for the gate's real captures or are never true.
91	NEVER_CLAIMED = {
92	"post_quantum_with_chatgpt": "needs the node counters of a real ChatGPT that negotiated 0x11EC (G1); a loopback client is not ChatGPT",
93	"aes_256_only": "the client chooses the suite: ChatGPT may pick AES-128-GCM or ChaCha20",
94	"tls_1_3_always": "only what the gate proves per system; the other end decides what it offers",
95	"ech": "no Encrypted Client Hello: the node reads the SNI by design",
96	}
97	STATES = ("ready", "degraded")
98
99	# The switch of the OPENSSL_CONF restriction: a word of runtime.env that only the supervised e2e child honours. Absent means off.
100	OPENSSL_CONF_SWITCH = "ALL_E2E_OPENSSL_CONF"
101	OPENSSL_CONF_FILE = "openssl-all.cnf"
102	OPENSSL_CONF_DEFAULT_ON = False
103	ON_WORDS = frozenset({"on", "1", "true", "yes"})
104	# The groups the restricted list offers: hybrid first where the library knows it ("?" skips a name an older OpenSSL does not know), then the classical groups every
105	# TLS 1.3 peer of ours speaks. The TLS 1.3 suites are the three every OpenSSL has, ChaCha20 last so that a machine with AES hardware keeps its choice.
106	OPENSSL_CONF_GROUPS = "?X25519MLKEM768:X25519:secp256r1:secp384r1"
107	OPENSSL_CONF_SUITES = "TLS_AES_256_GCM_SHA384:TLS_AES_128_GCM_SHA256:TLS_CHACHA20_POLY1305_SHA256"
108
109
110	class TlsParseError(ValueError):
111	"""The bytes are not a TLS hello. ``word`` is a closed word (``bad_record``, ``bad_hello``, ``bad_extension``, ``too_large``, ``not_handshake``)."""
112
113	def __init__(self, word: str):
114	super().__init__(word)
115	self.word = word
116
117
118	class TlsUnavailable(RuntimeError):
119	"""The terminator cannot serve: this Python has no TLS 1.3. ``word`` is what the status file and the heartbeat carry."""
120
121	def __init__(self, word: str = "tls13_required"):
122	super().__init__(word)
123	self.word = word
124
125
126	# ------------------------------------------------------------------------------------------------------------------------ names
127
128	def is_grease(value: int) -> bool:
129	return (value & 0x0F0F) == 0x0A0A and (value >> 8) == (value & 0xFF)
130
131
132	def group_name(group: int \| None) -> str \| None:
133	if group is None:
134	return None
135	return "grease" if is_grease(group) else GROUP_NAMES.get(group, f"0x{group:04x}")
136
137
138	def cipher_name(cipher: int \| None) -> str \| None:
139	if cipher is None:
140	return None
141	return "grease" if is_grease(cipher) else CIPHER_NAMES.get(cipher, f"0x{cipher:04x}")
142
143
144	def version_name(version: int \| None) -> str \| None:
145	return None if version is None else VERSION_NAMES.get(version, f"0x{version:04x}")
146
147
148	# ------------------------------------------------------------------------------------------------------------------------ the hello parser
149
150	@dataclass(frozen=True)
151	class ClientHelloInfo:
152	"""What a ClientHello says in the clear. ``key_share_groups`` are the groups it actually sent a key for (the server can choose only among those without a retry)."""
153	versions: tuple[int, ...]
154	cipher_suites: tuple[int, ...]
155	groups: tuple[int, ...]
156	key_share_groups: tuple[int, ...]
157	server_name: str \| None
158	alpn: tuple[str, ...]
159	extensions: tuple[int, ...]
160	records: int
161
162	@property
163	def offers_tls13(self) -> bool:
164	return TLS_13 in self.versions
165
166	@property
167	def offers_hybrid(self) -> bool:
168	return any(group in HYBRID_GROUPS for group in self.groups)
169
170	@property
171	def sends_hybrid_key_share(self) -> bool:
172	return any(group in HYBRID_GROUPS for group in self.key_share_groups)
173
174	def as_dict(self) -> dict[str, Any]:
175	return {"versions": [version_name(v) for v in self.versions if not is_grease(v)], "offers_hybrid": self.offers_hybrid,
176	"sends_hybrid_key_share": self.sends_hybrid_key_share, "groups": [group_name(g) for g in self.groups if not is_grease(g)],
177	"key_share_groups": [group_name(g) for g in self.key_share_groups if not is_grease(g)],
178	"cipher_suites": [cipher_name(c) for c in self.cipher_suites if not is_grease(c)], "server_name": self.server_name, "alpn": list(self.alpn),
179	"ech": EXT_ECH in self.extensions, "records": self.records}
180
181
182	@dataclass(frozen=True)
183	class ServerHelloInfo:
184	"""The ServerHello (or the HelloRetryRequest): the negotiated version, the suite and the group the server selected. In TLS 1.3 the version is the
185	``supported_versions`` extension, the legacy field of the hello says 1.2 for every handshake; ``group`` is None for TLS 1.2 (its curve is in a later message)."""
186	version: int
187	cipher: int
188	group: int \| None
189	hello_retry_request: bool
190	extensions: tuple[int, ...]
191	records: int
192
193	@property
194	def version_name(self) -> str \| None:
195	return version_name(self.version)
196
197	@property
198	def cipher_name(self) -> str \| None:
199	return cipher_name(self.cipher)
200
201	@property
202	def group_name(self) -> str \| None:
203	return group_name(self.group)
204
205	@property
206	def hybrid(self) -> bool:
207	return self.group in HYBRID_GROUPS
208
209	def as_dict(self) -> dict[str, Any]:
210	return {"version": self.version_name, "cipher": self.cipher_name, "group": self.group_name, "group_id": self.group,
211	"hybrid": self.hybrid, "hello_retry_request": self.hello_retry_request}
212
213
214	@dataclass(frozen=True)
215	class TlsAlert:
216	level: int
217	description: int
218
219	@property
220	def name(self) -> str:
221	return ALERT_NAMES.get(self.description, f"alert_{self.description}")
222
223
224	@dataclass(frozen=True)
225	class ServerObservation:
226	"""What the server side of a connection said before it went encrypted: the hellos (a retry request, then the real one), an alert if it refused, and whether
227	the stream held enough to be sure (``complete`` is False while a record or a message is cut: read more bytes and ask again)."""
228	hellos: tuple[ServerHelloInfo, ...] = ()
229	alert: TlsAlert \| None = None
230	complete: bool = False
231
232	@property
233	def negotiated(self) -> ServerHelloInfo \| None:
234	"""The hello that settled the handshake: the last one that is not a retry request (or the retry request when nothing followed it)."""
235	real = [hello for hello in self.hellos if not hello.hello_retry_request]
236	return real[-1] if real else (self.hellos[-1] if self.hellos else None)
237
238
239	def _read_records(stream: bytes) -> list[tuple[int, bytes]]:
240	"""The whole TLS records at the start of `stream` as (type, body); a record that is still arriving is left out. Stops at the first encrypted record: everything
241	after the hello of a TLS 1.3 handshake is ciphertext."""
242	records: list[tuple[int, bytes]] = []
243	offset = 0
244	while offset + RECORD_HEADER_BYTES <= len(stream):
245	kind, major, length = stream[offset], stream[offset + 1], struct.unpack_from(">H", stream, offset + 3)[0]
246	if kind not in (RECORD_HANDSHAKE, RECORD_ALERT, RECORD_CCS, RECORD_APPLICATION) or major != 3 or length > RECORD_MAX_BYTES or length == 0:
247	raise TlsParseError("bad_record")
248	if offset + RECORD_HEADER_BYTES + length > len(stream):
249	break
250	records.append((kind, stream[offset + RECORD_HEADER_BYTES:offset + RECORD_HEADER_BYTES + length]))
251	offset += RECORD_HEADER_BYTES + length
252	if kind == RECORD_APPLICATION:
253	break
254	return records
255
256
257	def _handshake_messages(stream: bytes, *, server: bool = False) -> tuple[list[tuple[int, bytes, int]], TlsAlert \| None, bool]:
258	"""The handshake messages (type, body, records they span) of the clear part of `stream`, rebuilt across records: one message may be cut into several records and
259	one record may hold several messages. Returns them with the alert (if a record was an alert) and whether the last message is whole.
260
261	The clear part ends at the first change_cipher_spec: after it a TLS 1.2 handshake is encrypted (its Finished is still a "handshake" record), and a TLS 1.3 client
262	or server sends it only for middleboxes, followed by ciphertext. The one exception is a server that answered with a HelloRetryRequest: its real ServerHello comes
263	after that record, in the clear, once the client has sent its second hello."""
264	messages: list[tuple[int, bytes, int]] = []
265	pending = bytearray()
266	spans = 0
267	alert: TlsAlert \| None = None
268	for kind, body in _read_records(stream):
269	if kind == RECORD_ALERT:
270	if len(body) >= 2 and alert is None:
271	alert = TlsAlert(body[0], body[1])
272	continue
273	if kind == RECORD_CCS:
274	if server and not pending and messages and messages[-1][0] == 2 and messages[-1][1][2:34] == HELLO_RETRY_RANDOM:
275	continue
276	break
277	if kind != RECORD_HANDSHAKE:
278	break
279	pending += body
280	spans += 1
281	while len(pending) >= 4:
282	if server and pending[0] != 2: # what follows a ServerHello in TLS 1.2 (a certificate chain, say) is not read: it may be large and it is not ours
283	messages.append((pending[0], b"", spans))
284	return messages, alert, True
285	size = int.from_bytes(pending[1:4], "big")
286	if size > HELLO_MAX_BYTES:
287	raise TlsParseError("too_large")
288	if len(pending) < 4 + size:
289	break
290	messages.append((pending[0], bytes(pending[4:4 + size]), spans))
291	del pending[:4 + size]
292	spans = 1 if pending else 0
293	return messages, alert, not pending
294
295
296	def _take(data: bytes, offset: int, size: int) -> tuple[bytes, int]:
297	if size < 0 or offset + size > len(data):
298	raise TlsParseError("bad_hello")
299	return data[offset:offset + size], offset + size
300
301
302	def _extensions(block: bytes) -> list[tuple[int, bytes]]:
303	out: list[tuple[int, bytes]] = []
304	offset = 0
305	while offset < len(block):
306	header, offset = _take(block, offset, 4)
307	kind, size = struct.unpack(">HH", header)
308	data, offset = _take(block, offset, size)
309	out.append((kind, data))
310	return out
311
312
313	def _u16_list(data: bytes, *, width: int = 2) -> tuple[int, ...]:
314	"""A length-prefixed list of 16-bit values (the prefix is `width` bytes: 1 for supported_versions, 2 for supported_groups)."""
315	if len(data) < width or int.from_bytes(data[:width], "big") != len(data) - width or (len(data) - width) % 2:
316	raise TlsParseError("bad_extension")
317	body = data[width:]
318	return struct.unpack(">" + "H" * (len(body) // 2), body)
319
320
321	def _hello_prefix(body: bytes) -> tuple[int, bytes, int]:
322	"""(legacy version, random, offset after the session id) of a Client or Server hello."""
323	head, offset = _take(body, 0, 34)
324	version, random = struct.unpack(">H", head[:2])[0], head[2:34]
325	size = _take(body, offset, 1)[0][0]
326	_, offset = _take(body, offset + 1, size)
327	return version, random, offset
328
329
330	def _client_hello(body: bytes, records: int) -> ClientHelloInfo:
331	_, _, offset = _hello_prefix(body)
332	size, offset = _take(body, offset, 2)
333	suites_raw, offset = _take(body, offset, struct.unpack(">H", size)[0])
334	if len(suites_raw) % 2:
335	raise TlsParseError("bad_hello")
336	skip, offset = _take(body, offset, 1)
337	_, offset = _take(body, offset, skip[0])
338	versions: tuple[int, ...] = ()
339	groups: tuple[int, ...] = ()
340	shares: list[int] = []
341	name: str \| None = None
342	alpn: list[str] = []
343	kinds: list[int] = []
344	if offset < len(body):
345	size, offset = _take(body, offset, 2)
346	block, offset = _take(body, offset, struct.unpack(">H", size)[0])
347	for kind, data in _extensions(block):
348	kinds.append(kind)
349	if kind == EXT_SUPPORTED_VERSIONS:
350	versions = _u16_list(data, width=1)
351	elif kind == EXT_SUPPORTED_GROUPS:
352	groups = _u16_list(data)
353	elif kind == EXT_KEY_SHARE:
354	if len(data) < 2 or struct.unpack_from(">H", data, 0)[0] != len(data) - 2:
355	raise TlsParseError("bad_extension")
356	position = 2
357	while position < len(data):
358	entry, position = _take(data, position, 4)
359	group, key_size = struct.unpack(">HH", entry)
360	_, position = _take(data, position, key_size)
361	shares.append(group)
362	elif kind == EXT_SERVER_NAME:
363	if len(data) >= 5 and data[2] == 0:
364	raw, _ = _take(data, 5, struct.unpack_from(">H", data, 3)[0])
365	name = raw.decode("ascii", errors="replace")
366	elif kind == EXT_ALPN:
367	if len(data) < 2 or struct.unpack_from(">H", data, 0)[0] != len(data) - 2:
368	raise TlsParseError("bad_extension")
369	position = 2
370	while position < len(data):
371	item, position = _take(data, position + 1, data[position])
372	alpn.append(item.decode("ascii", errors="replace"))
373	return ClientHelloInfo(versions=versions, cipher_suites=struct.unpack(">" + "H" * (len(suites_raw) // 2), suites_raw), groups=groups,
374	key_share_groups=tuple(shares), server_name=name, alpn=tuple(alpn), extensions=tuple(kinds), records=records)
375
376
377	def _server_hello(body: bytes, records: int) -> ServerHelloInfo:
378	legacy, random, offset = _hello_prefix(body)
379	head, offset = _take(body, offset, 3)
380	cipher = struct.unpack(">H", head[:2])[0]
381	version, group = legacy, None
382	kinds: list[int] = []
383	retry = random == HELLO_RETRY_RANDOM
384	if offset < len(body):
385	size, offset = _take(body, offset, 2)
386	block, offset = _take(body, offset, struct.unpack(">H", size)[0])
387	for kind, data in _extensions(block):
388	kinds.append(kind)
389	if kind == EXT_SUPPORTED_VERSIONS:
390	if len(data) != 2:
391	raise TlsParseError("bad_extension")
392	version = struct.unpack(">H", data)[0]
393	elif kind == EXT_KEY_SHARE:
394	# ServerHello: a KeyShareEntry (group, length, key); HelloRetryRequest: only the group it wants
395	if len(data) < 2 or (not retry and len(data) < 4):
396	raise TlsParseError("bad_extension")
397	group = struct.unpack_from(">H", data, 0)[0]
398	if not retry and struct.unpack_from(">H", data, 2)[0] != len(data) - 4:
399	raise TlsParseError("bad_extension")
400	return ServerHelloInfo(version=version, cipher=cipher, group=group, hello_retry_request=retry, extensions=tuple(kinds), records=records)
401
402
403	def parse_client_hello(stream: bytes) -> ClientHelloInfo \| None:
404	"""The ClientHello that starts `stream` (the client's bytes of one TCP connection, reassembled), or None while it is not whole yet. A hello may be cut across any
405	number of records and segments; anything that is not a ClientHello is ``TlsParseError``."""
406	messages, _alert, _clean = _handshake_messages(stream)
407	if not messages:
408	if len(stream) >= RECORD_HEADER_BYTES and stream[0] != RECORD_HANDSHAKE:
409	raise TlsParseError("not_handshake")
410	return None
411	kind, body, records = messages[0]
412	if kind != 1:
413	raise TlsParseError("not_handshake")
414	return _client_hello(body, records)
415
416
417	def observe_server(stream: bytes) -> ServerObservation:
418	"""What the server's bytes of one connection say in the clear: its ServerHello (and a HelloRetryRequest before it), or the alert it answered with."""
419	messages, alert, clean = _handshake_messages(stream, server=True)
420	hellos = tuple(_server_hello(body, records) for kind, body, records in messages if kind == 2)
421	if messages and messages[0][0] != 2:
422	raise TlsParseError("not_handshake")
423	if not messages and alert is None and len(stream) >= RECORD_HEADER_BYTES and stream[0] not in (RECORD_HANDSHAKE, RECORD_ALERT):
424	raise TlsParseError("not_handshake")
425	return ServerObservation(hellos=hellos, alert=alert, complete=bool(hellos or alert) and clean)
426
427
428	def parse_server_hello(stream: bytes) -> ServerHelloInfo \| None:
429	"""The ServerHello that settled the handshake (see ``ServerObservation.negotiated``), or None while it is not whole or when the server answered with an alert."""
430	return observe_server(stream).negotiated
431
432
433	def summarize(hellos: Sequence[ServerHelloInfo]) -> dict[str, dict[str, int]]:
434	"""Counts of version, suite and group over many ServerHellos (a capture, a node's counters): the three tables ``e2e-proof.json`` carries."""
435	out: dict[str, dict[str, int]] = {"versions": {}, "ciphers": {}, "groups": {}}
436	for hello in hellos:
437	if hello.hello_retry_request:
438	continue
439	for table, word in (("versions", hello.version_name), ("ciphers", hello.cipher_name), ("groups", hello.group_name or "none")):
440	out[table][word or "unknown"] = out[table].get(word or "unknown", 0) + 1
441	return out
442
443
444	# ------------------------------------------------------------------------------------------------------------------------ the runtime and the server context
445
446	@dataclass(frozen=True)
447	class Runtime:
448	"""What this Python's TLS library is: reported per operating system, tying the hybrid claim to the OpenSSL that carries it."""
449	openssl: str
450	openssl_info: tuple[int, ...]
451	flavor: str
452	has_tls13: bool
453	python: str
454	system: str
455	release: str
456	machine: str
457
458	@classmethod
459	def current(cls) -> "Runtime":
460	text = getattr(ssl, "OPENSSL_VERSION", "")
461	flavor = "libressl" if "LibreSSL" in text else "openssl" if text.startswith("OpenSSL") else "other"
462	has13 = bool(getattr(ssl, "HAS_TLSv1_3", False)) and hasattr(ssl.TLSVersion, "TLSv1_3")
463	return cls(openssl=text, openssl_info=tuple(getattr(ssl, "OPENSSL_VERSION_INFO", ())[:3]), flavor=flavor, has_tls13=has13,
464	python=platform.python_version(), system=platform.system(), release=platform.release(), machine=platform.machine())
465
466	@property
467	def hybrid_capable(self) -> bool:
468	"""OpenSSL 3.5 or later: the only library family that ships X25519MLKEM768 in its defaults. This is the gate of the hybrid claim, nothing more."""
469	return self.flavor == "openssl" and tuple(self.openssl_info[:2]) >= HYBRID_MIN_OPENSSL
470
471	def as_dict(self) -> dict[str, Any]:
472	return {"os": {"system": self.system, "release": self.release, "machine": self.machine}, "python": self.python,
473	"openssl": {"version": self.openssl, "version_info": list(self.openssl_info), "flavor": self.flavor, "has_tls13": self.has_tls13,
474	"hybrid_capable": self.hybrid_capable}}
475
476
477	def tls13_available(runtime: Runtime \| None = None) -> bool:
478	return (runtime or Runtime.current()).has_tls13
479
480
481	def require_tls13(runtime: Runtime \| None = None) -> None:
482	"""The one condition the terminator refuses to start on. The caller reports it (status ``degraded``, reason ``tls13_required``); this is never swallowed."""
483	if not tls13_available(runtime):
484	log.error("this Python's TLS library has no TLS 1.3: the encrypted path cannot serve (%s)", (runtime or Runtime.current()).openssl)
485	raise TlsUnavailable("tls13_required")
486
487
488	def start_decision(runtime: Runtime \| None = None) -> dict[str, str]:
489	"""May the terminator start here, as the status file says it: ``ready`` or ``degraded`` with the closed word that says why."""
490	if tls13_available(runtime):
491	return {"state": "ready", "reason": ""}
492	return {"state": "degraded", "reason": "tls13_required"}
493
494
495	_VERSIONS = {"TLSv1.3": "TLSv1_3", "TLSv1.2": "TLSv1_2"}
496
497
498	def minimum_version() -> "ssl.TLSVersion":
499	word = policy.value("terminator.tls_min_version")
500	if word not in _VERSIONS:
501	raise TlsUnavailable("tls13_required")
502	return getattr(ssl.TLSVersion, _VERSIONS[word])
503
504
505	def server_context(chain_path: Path \| str, key_path: Path \| str, *, runtime: Runtime \| None = None) -> ssl.SSLContext:
506	"""The terminator's context for one certificate generation: TLS 1.3 at least, ALPN of the policy, nothing narrower. Raises ``TlsUnavailable`` when the Python
507	has no TLS 1.3; every other property of the library (version, hybrid) is reported, not enforced."""
508	require_tls13(runtime)
509	context = ssl.SSLContext(ssl.PROTOCOL_TLS_SERVER)
510	context.minimum_version = minimum_version()
511	context.set_alpn_protocols([policy.value("terminator.tls_alpn")])
512	context.load_cert_chain(str(chain_path), str(key_path))
513	return context
514
515
516	# ------------------------------------------------------------------------------------------------------------------------ OPENSSL_CONF
517
518	def render_openssl_conf() -> str:
519	"""The text of the installer's OpenSSL config file: groups and TLS 1.3 suites only. No ``MinProtocol`` (an outbound TLS 1.2 peer, a corporate proxy, must keep working),
520	and the hybrid group behind ``?`` so that an OpenSSL that does not know it ignores the name instead of refusing the whole list."""
521	return (
522	"# ALL 2.0: key-exchange groups and TLS 1.3 suites for the e2e child process. OFF unless runtime.env says ALL_E2E_OPENSSL_CONF=on; see src/all_mcp/tls_policy.py.\n"
523	"# Generated by tls_policy.render_openssl_conf(): a test compares this file with it.\n"
524	"openssl_conf = all_openssl_init\n\n"
525	"[all_openssl_init]\n"
526	"ssl_conf = all_ssl\n\n"
527	"[all_ssl]\n"
528	"system_default = all_tls_defaults\n\n"
529	"[all_tls_defaults]\n"
530	f"Groups = {OPENSSL_CONF_GROUPS}\n"
531	f"Ciphersuites = {OPENSSL_CONF_SUITES}\n"
532	)
533
534
535	def openssl_conf_wanted(env: Mapping[str, str]) -> bool:
536	word = str(env.get(OPENSSL_CONF_SWITCH, "")).strip().lower()
537	return word in ON_WORDS if word else OPENSSL_CONF_DEFAULT_ON
538
539
540	def apply_openssl_conf(env: Mapping[str, str], directory: Path \| str \| None = None, *, switch: str \| None = None) -> dict[str, str]:
541	"""The environment of the e2e child: a copy of `env`, with ``OPENSSL_CONF`` pointing at the restricted configuration only when it was turned on (`switch`, a
542	word of runtime.env, or ``ALL_E2E_OPENSSL_CONF`` in `env`). Off (the default) returns the environment unchanged, so the child behaves as it did without this
543	module. A file that cannot be written leaves the environment as it was: a policy nice-to-have never stops the child."""
544	out = dict(env)
545	wanted = openssl_conf_wanted({OPENSSL_CONF_SWITCH: switch} if switch is not None else out)
546	if not wanted or directory is None:
547	return out
548	try:
549	target = Path(directory) / OPENSSL_CONF_FILE
550	target.parent.mkdir(parents=True, exist_ok=True)
551	text = render_openssl_conf()
552	if not target.exists() or target.read_text(encoding="utf-8") != text:
553	target.write_text(text, encoding="utf-8")
554	out["OPENSSL_CONF"] = str(target)
555	except OSError as exc:
556	log.warning("the OpenSSL restriction could not be written (%s): the child runs without it", type(exc).__name__)
557	return out
558
559
560	# ------------------------------------------------------------------------------------------------------------------------ loopback handshakes
561
562	@dataclass(frozen=True)
563	class ClientProfile:
564	"""One kind of client for a probe. ``groups`` and ``ciphersuites`` are OpenSSL lists applied to a child process (Python cannot restrict a client's groups
565	in-process); a profile without them runs in-process. ``max_version`` is a policy word (``TLSv1.2``)."""
566	name: str
567	groups: str \| None = None
568	ciphersuites: str \| None = None
569	max_version: str \| None = None
570	alpn: tuple[str, ...] = ("http/1.1",)
571	server_name: str = PROBE_NAME
572
573	@property
574	def in_process(self) -> bool:
575	return self.groups is None and self.ciphersuites is None
576
577
578	HYBRID_CLIENT = ClientProfile("hybrid_client")
579	CLASSIC_CLIENT = ClientProfile("classic_client", groups="X25519")
580	TLS12_CLIENT = ClientProfile("tls12_client", max_version="TLSv1.2")
581
582
583	@dataclass
584	class Handshake:
585	"""The outcome of one loopback handshake: what each side sent in the clear, parsed, and what the two ends reported themselves."""
586	profile: str
587	completed: bool = False
588	error: str = ""
589	client_hello: ClientHelloInfo \| None = None
590	server: ServerObservation = field(default_factory=ServerObservation)
591	python_version: str \| None = None # the server's own view, to cross-check the parser against
592	python_cipher: str \| None = None
593	alpn: str \| None = None
594	client_error: str = ""
595	client_bytes: bytes = b"" # what each side wrote on the wire: a gate turns them into a capture, a test into a hello to cut and mutate
596	server_bytes: bytes = b""
597
598	@property
599	def negotiated(self) -> ServerHelloInfo \| None:
600	return self.server.negotiated
601
602	def as_dict(self) -> dict[str, Any]:
603	hello = self.negotiated
604	out: dict[str, Any] = {"completed": self.completed, "error": self.error, "alpn": self.alpn,
605	"client": self.client_hello.as_dict() if self.client_hello else None,
606	"server": hello.as_dict() if hello and self.completed else None,
607	"alert": self.server.alert.name if self.server.alert else None}
608	if hello and self.completed:
609	out["consistent"] = (hello.version_name == self.python_version and hello.cipher_name == self.python_cipher)
610	return out
611
612
613	class _InProcessClient:
614	def __init__(self, profile: ClientProfile):
615	context = ssl.SSLContext(ssl.PROTOCOL_TLS_CLIENT)
616	context.check_hostname = False
617	context.verify_mode = ssl.CERT_NONE # a loopback probe of our own server with a throw-away certificate: nothing is trusted, nothing is sent
618	if profile.max_version:
619	context.maximum_version = getattr(ssl.TLSVersion, _VERSIONS[profile.max_version])
620	if profile.alpn:
621	context.set_alpn_protocols(list(profile.alpn))
622	self._in, self._out = ssl.MemoryBIO(), ssl.MemoryBIO()
623	self._ssl = context.wrap_bio(self._in, self._out, server_hostname=profile.server_name)
624	self.error = ""
625
626	def step(self, data: bytes) -> tuple[bytes, bool]:
627	if data:
628	self._in.write(data)
629	done = False
630	try:
631	self._ssl.do_handshake()
632	done = True
633	except ssl.SSLWantReadError:
634	pass
635	except ssl.SSLError as exc:
636	self.error = (getattr(exc, "reason", None) or type(exc).__name__).lower()
637	done = True
638	return self._out.read(), done
639
640	def close(self) -> None:
641	return None
642
643
644	# The child of a probe: a client that reads the server's bytes from stdin and writes its own to stdout, one frame each (4-byte length, a tag, the bytes), so a probe
645	# needs no port and works inside any sandbox. Tags: D = bytes, a reply is expected; F = the last bytes (no reply), R = the result as JSON.
646	_CHILD = r"""
647	import json, ssl, struct, sys
648	cfg = json.loads(sys.argv[1])
649	def put(tag, payload):
650	sys.stdout.buffer.write(struct.pack(">I", len(payload) + 1) + tag + payload); sys.stdout.buffer.flush()
651	def get():
652	head = sys.stdin.buffer.read(4)
653	if len(head) < 4: raise SystemExit(0)
654	body = sys.stdin.buffer.read(struct.unpack(">I", head)[0])
655	return body[1:]
656	ctx = ssl.SSLContext(ssl.PROTOCOL_TLS_CLIENT)
657	ctx.check_hostname = False
658	ctx.verify_mode = ssl.CERT_NONE
659	if cfg["max_version"]:
660	ctx.maximum_version = getattr(ssl.TLSVersion, cfg["max_version"].replace("TLSv1.", "TLSv1_"))
661	if cfg["alpn"]:
662	ctx.set_alpn_protocols(cfg["alpn"])
663	i, o = ssl.MemoryBIO(), ssl.MemoryBIO()
664	s = ctx.wrap_bio(i, o, server_hostname=cfg["server_name"])
665	data = b""
666	while True:
667	if data:
668	i.write(data)
669	done, err = False, ""
670	try:
671	s.do_handshake(); done = True
672	except ssl.SSLWantReadError:
673	pass
674	except ssl.SSLError as exc:
675	done, err = True, (getattr(exc, "reason", None) or type(exc).__name__).lower()
676	out = o.read()
677	if done:
678	put(b"F", out)
679	put(b"R", json.dumps({"error": err}).encode())
680	break
681	put(b"D", out)
682	data = get()
683	"""
684
685
686	class _ChildClient:
687	"""A client in a child process whose OpenSSL is configured by a conf file made for the profile (groups, suites): the only way to make a client that does not
688	offer a group the library knows. The child is the same interpreter, so the library under test is the one the product ships."""
689
690	def __init__(self, profile: ClientProfile, workdir: Path):
691	lines = ["openssl_conf = probe_init", "", "[probe_init]", "ssl_conf = probe_ssl", "", "[probe_ssl]", "system_default = probe_defaults", "", "[probe_defaults]"]
692	if profile.groups:
693	lines.append(f"Groups = {profile.groups}")
694	if profile.ciphersuites:
695	lines.append(f"Ciphersuites = {profile.ciphersuites}")
696	conf = workdir / f"{profile.name}.cnf"
697	conf.write_text("\n".join(lines) + "\n", encoding="utf-8")
698	env = {key: value for key, value in os.environ.items() if key != "OPENSSL_CONF"}
699	env["OPENSSL_CONF"] = str(conf)
700	config = json.dumps({"max_version": profile.max_version, "alpn": list(profile.alpn), "server_name": profile.server_name})
701	self._proc = subprocess.Popen([sys.executable, "-I", "-c", _CHILD, config], stdin=subprocess.PIPE, stdout=subprocess.PIPE, stderr=subprocess.DEVNULL,
702	env=env, creationflags=no_window_flags())
703	self._frames: "queue.Queue[tuple[bytes, bytes] \| None]" = queue.Queue()
704	self._reader = threading.Thread(target=self._read, daemon=True)
705	self._reader.start()
706	self.error = ""
707	self._first = True
708
709	def _read(self) -> None:
710	try:
711	stdout = self._proc.stdout
712	while stdout is not None:
713	head = stdout.read(4)
714	if len(head) < 4:
715	break
716	body = stdout.read(struct.unpack(">I", head)[0])
717	self._frames.put((body[:1], body[1:]))
718	except (OSError, ValueError):
719	pass
720	self._frames.put(None)
721
722	def step(self, data: bytes) -> tuple[bytes, bool]:
723	if not self._first:
724	try:
725	assert self._proc.stdin is not None
726	self._proc.stdin.write(struct.pack(">I", len(data) + 1) + b"D" + data)
727	self._proc.stdin.flush()
728	except (OSError, ValueError):
729	return b"", True
730	self._first = False
731	try:
732	frame = self._frames.get(timeout=PROBE_SECONDS)
733	except queue.Empty:
734	self.error = "probe_timeout"
735	return b"", True
736	if frame is None:
737	self.error = "probe_child_failed"
738	return b"", True
739	tag, payload = frame
740	if tag == b"F":
741	tail = self._frames.get(timeout=PROBE_SECONDS)
742	if tail is not None and tail[0] == b"R":
743	self.error = str(json.loads(tail[1].decode("utf-8")).get("error", ""))
744	return payload, True
745	return payload, False
746
747	def close(self) -> None:
748	with contextlib.suppress(Exception):
749	if self._proc.stdin:
750	self._proc.stdin.close()
751	with contextlib.suppress(Exception):
752	self._proc.wait(timeout=5)
753	if self._proc.poll() is None:
754	self._proc.kill()
755	with contextlib.suppress(Exception):
756	self._proc.wait(timeout=5)
757	if self._proc.stdout:
758	with contextlib.suppress(Exception):
759	self._proc.stdout.close()
760
761
762	def handshake(context: ssl.SSLContext, profile: ClientProfile = HYBRID_CLIENT, *, workdir: Path \| None = None) -> Handshake:
763	"""A real TLS handshake between `context` (a server context) and a client of `profile`, pumped through memory (no socket), with every byte each side wrote kept
764	and parsed: the ClientHello, and the ServerHello or the alert the server answered with. Never raises for a refused handshake: that is the answer."""
765	result = Handshake(profile=profile.name)
766	server_in, server_out = ssl.MemoryBIO(), ssl.MemoryBIO()
767	server = context.wrap_bio(server_in, server_out, server_side=True)
768	scratch = None
769	if profile.in_process:
770	client: Any = _InProcessClient(profile)
771	else:
772	if workdir is None:
773	scratch = tempfile.TemporaryDirectory(prefix="all-tls-probe-")
774	workdir = Path(scratch.name)
775	client = _ChildClient(profile, workdir)
776	sent_by_client = bytearray()
777	sent_by_server = bytearray()
778	server_done = False
779	try:
780	data = b""
781	for _ in range(PROBE_ROUNDS):
782	to_server, client_done = client.step(data)
783	sent_by_client += to_server
784	if to_server:
785	server_in.write(to_server)
786	if not server_done:
787	try:
788	server.do_handshake()
789	server_done = True
790	except ssl.SSLWantReadError:
791	pass
792	except ssl.SSLError as exc:
793	result.error = (getattr(exc, "reason", None) or type(exc).__name__).lower()
794	server_done = True
795	data = server_out.read()
796	sent_by_server += data
797	if server_done or (client_done and not to_server):
798	break
799	result.client_error = getattr(client, "error", "")
800	result.completed = server_done and not result.error and not result.client_error
801	if result.completed:
802	result.python_version = server.version()
803	cipher = server.cipher()
804	result.python_cipher = cipher[0] if cipher else None
805	result.alpn = server.selected_alpn_protocol()
806	finally:
807	client.close()
808	if scratch is not None:
809	scratch.cleanup()
810	result.client_bytes, result.server_bytes = bytes(sent_by_client), bytes(sent_by_server)
811	with contextlib.suppress(TlsParseError):
812	result.client_hello = parse_client_hello(bytes(sent_by_client))
813	with contextlib.suppress(TlsParseError):
814	result.server = observe_server(bytes(sent_by_server))
815	return result
816
817
818	def probe_certificate(directory: Path) -> tuple[Path, Path]:
819	"""A throw-away ECDSA P-256 certificate for ``probe.all.invalid`` (chain and key, 0600), good for a day: the loopback probe's own, never a device key."""
820	from cryptography import x509
821	from cryptography.hazmat.primitives import hashes, serialization
822	from cryptography.hazmat.primitives.asymmetric import ec
823	from cryptography.x509.oid import NameOID
824
825	key = ec.generate_private_key(ec.SECP256R1())
826	name = x509.Name([x509.NameAttribute(NameOID.COMMON_NAME, PROBE_NAME)])
827	now = datetime.datetime.now(datetime.timezone.utc)
828	cert = (x509.CertificateBuilder().subject_name(name).issuer_name(name).public_key(key.public_key()).serial_number(x509.random_serial_number())
829	.not_valid_before(now - datetime.timedelta(minutes=5)).not_valid_after(now + datetime.timedelta(days=1))
830	.add_extension(x509.SubjectAlternativeName([x509.DNSName(PROBE_NAME)]), critical=False).sign(key, hashes.SHA256()))
831	chain, private = directory / "probe-chain.pem", directory / "probe-key.pem"
832	private.write_bytes(key.private_bytes(serialization.Encoding.PEM, serialization.PrivateFormat.PKCS8, serialization.NoEncryption()))
833	chain.write_bytes(cert.public_bytes(serialization.Encoding.PEM))
834	with contextlib.suppress(OSError):
835	os.chmod(private, 0o600)
836	return chain, private
837
838
839	# ------------------------------------------------------------------------------------------------------------------------ the report
840
841	def _claim(provable: bool, evidence: str = "", reason: str = "") -> dict[str, Any]:
842	return {"provable": provable, "evidence": evidence if provable else "", "reason": "" if provable else reason}
843
844
845	def evaluate_claims(runtime: Runtime, probes: Mapping[str, Handshake \| None], minimum_ok: bool) -> dict[str, dict[str, Any]]:
846	"""Which claims the probes prove, each with its evidence or the closed word that says why not. The OpenSSL version gates the hybrid claim and no other."""
847	default, classic, old = probes.get("hybrid_client"), probes.get("classic_client"), probes.get("tls12_client")
848	claims: dict[str, dict[str, Any]] = {}
849	hello = default.negotiated if default and default.completed else None
850	if not runtime.has_tls13:
851	for name in CLAIMS:
852	claims[name] = _claim(False, reason="tls13_required")
853	return claims
854	claims["tls_1_3_minimum"] = (_claim(True, CLAIMS["tls_1_3_minimum"]) if minimum_ok and hello is not None and hello.version == TLS_13
855	else _claim(False, reason="probe_failed" if hello is None else "not_tls13"))
856	if old is None:
857	claims["tls_1_2_refused"] = _claim(False, reason="not_probed")
858	elif not old.completed and (old.server.alert is not None or old.error):
859	claims["tls_1_2_refused"] = _claim(True, CLAIMS["tls_1_2_refused"])
860	else:
861	claims["tls_1_2_refused"] = _claim(False, reason="tls12_client_was_served" if old.completed else "inconclusive")
862	claims["alpn_http_1_1"] = (_claim(True, CLAIMS["alpn_http_1_1"]) if default and default.completed and default.alpn == policy.value("terminator.tls_alpn")
863	else _claim(False, reason="probe_failed" if hello is None else "alpn_not_selected"))
864	claims["forward_secrecy"] = (_claim(True, CLAIMS["forward_secrecy"]) if hello is not None and hello.version == TLS_13 and hello.group is not None
865	else _claim(False, reason="probe_failed" if hello is None else "no_ephemeral_group"))
866	if classic is None:
867	claims["classical_fallback"] = _claim(False, reason="not_probed")
868	elif classic.client_hello is not None and classic.client_hello.offers_hybrid:
869	claims["classical_fallback"] = _claim(False, reason="openssl_conf_ignored") # the child still offered hybrid: this runtime does not honour OPENSSL_CONF
870	elif classic.completed and classic.negotiated is not None and not classic.negotiated.hybrid:
871	claims["classical_fallback"] = _claim(True, CLAIMS["classical_fallback"])
872	else:
873	claims["classical_fallback"] = _claim(False, reason="classic_client_refused")
874	if not runtime.hybrid_capable:
875	claims["hybrid_x25519mlkem768"] = _claim(False, reason="openssl_older_than_3_5")
876	elif default is None or default.client_hello is None or not default.client_hello.offers_hybrid:
877	claims["hybrid_x25519mlkem768"] = _claim(False, reason="client_did_not_offer_hybrid")
878	elif hello is not None and hello.group == GROUP_X25519_MLKEM768:
879	claims["hybrid_x25519mlkem768"] = _claim(True, CLAIMS["hybrid_x25519mlkem768"])
880	else:
881	claims["hybrid_x25519mlkem768"] = _claim(False, reason="group_not_negotiated")
882	return {name: claims[name] for name in CLAIMS}
883
884
885	def capability_report(*, probe: bool = True, chain_path: Path \| str \| None = None, key_path: Path \| str \| None = None, runtime: Runtime \| None = None,
886	environ: Mapping[str, str] \| None = None, now: float \| None = None) -> dict[str, Any]:
887	"""The per-OS TLS report of this computer, as the JSON that feeds ``e2e-proof.json``: the runtime, the start decision, the loopback probes and the claims they
888	prove (``allowed_claims``). `runtime` can be replaced to describe another library (tests); the probes always run on the real one. With ``probe=False`` nothing is
889	handshaken and no claim is provable (the cheap form the status file may carry)."""
890	runtime = runtime or Runtime.current()
891	environ = os.environ if environ is None else environ
892	decision = start_decision(runtime)
893	report: dict[str, Any] = {"v": REPORT_VERSION, "kind": "tls_capability_report",
894	"generated_at": (datetime.datetime.now(datetime.timezone.utc) if now is None
895	else datetime.datetime.fromtimestamp(now, datetime.timezone.utc)).strftime("%Y-%m-%dT%H:%M:%SZ"),
896	**runtime.as_dict(),
897	"policy": {"min_version": policy.value("terminator.tls_min_version"), "alpn": policy.value("terminator.tls_alpn")},
898	"terminator": {"can_start": decision["state"] == "ready", **decision},
899	"openssl_conf": {"default_on": OPENSSL_CONF_DEFAULT_ON, "switch": OPENSSL_CONF_SWITCH, "in_environment": bool(environ.get("OPENSSL_CONF")),
900	"effective": None},
901	"probes": {}, "claims": {}, "allowed_claims": [], "never_claimed": dict(NEVER_CLAIMED)}
902	probes: dict[str, Handshake \| None] = {}
903	if probe and runtime.has_tls13 and tls13_available():
904	with tempfile.TemporaryDirectory(prefix="all-tls-report-") as scratch:
905	folder = Path(scratch)
906	if chain_path is None or key_path is None:
907	chain_path, key_path = probe_certificate(folder)
908	context = server_context(chain_path, key_path)
909	minimum_ok = context.minimum_version >= ssl.TLSVersion.TLSv1_3
910	for profile in (HYBRID_CLIENT, CLASSIC_CLIENT, TLS12_CLIENT):
911	try:
912	probes[profile.name] = handshake(context, profile, workdir=folder)
913	except Exception as exc: # noqa: BLE001 - a probe that cannot run is reported as not proven, never a crash of the report
914	log.warning("TLS probe %s failed (%s)", profile.name, type(exc).__name__)
915	probes[profile.name] = None
916	report["probes"] = {name: (shake.as_dict() if shake else {"completed": False, "error": "probe_failed"}) for name, shake in probes.items()}
917	classic = probes.get("classic_client")
918	if classic is not None and classic.client_hello is not None:
919	if runtime.hybrid_capable: # a library that has no hybrid group shows nothing about whether it honoured the file
920	report["openssl_conf"]["effective"] = not classic.client_hello.offers_hybrid
921	report["claims"] = evaluate_claims(runtime, probes, minimum_ok)
922	else:
923	reason = "tls13_required" if not runtime.has_tls13 else "not_probed"
924	report["claims"] = {name: _claim(False, reason=reason) for name in CLAIMS}
925	report["allowed_claims"] = [name for name, claim in report["claims"].items() if claim["provable"]]
926	return report
927
928
929	def validate_report(report: Any) -> list[str]:
930	"""What is wrong with a capability report, as closed words (an empty list is a well-formed one): the check the gate runs before it copies a report into
931	``e2e-proof.json``, and the one the site's TLS sentences rely on."""
932	problems: list[str] = []
933	if not isinstance(report, dict):
934	return ["not_an_object"]
935	if report.get("v") != REPORT_VERSION or report.get("kind") != "tls_capability_report":
936	problems.append("wrong_kind_or_version")
937	for key in ("os", "python", "openssl", "policy", "terminator", "probes", "claims", "allowed_claims", "never_claimed", "generated_at"):
938	if key not in report:
939	problems.append("missing_" + key)
940	claims = report.get("claims") if isinstance(report.get("claims"), dict) else {}
941	if set(claims) != set(CLAIMS):
942	problems.append("claims_not_the_known_set")
943	allowed = report.get("allowed_claims")
944	if not isinstance(allowed, list) or allowed != [name for name in CLAIMS if claims.get(name, {}).get("provable") is True]:
945	problems.append("allowed_claims_do_not_match_claims")
946	openssl = report.get("openssl") if isinstance(report.get("openssl"), dict) else {}
947	if claims.get("hybrid_x25519mlkem768", {}).get("provable") and not openssl.get("hybrid_capable"):
948	problems.append("hybrid_claim_without_openssl_3_5")
949	if claims.get("tls_1_3_minimum", {}).get("provable") and not openssl.get("has_tls13"):
950	problems.append("tls13_claim_without_tls13")
951	for name, claim in claims.items():
952	if not isinstance(claim, dict) or not isinstance(claim.get("provable"), bool):
953	problems.append("claim_malformed")
954	elif claim["provable"] and not claim.get("evidence"):
955	problems.append("claim_without_evidence_" + name)
956	elif not claim["provable"] and not claim.get("reason"):
957	problems.append("claim_without_reason_" + name)
958	return problems
959
960
961	def proof_section(report: Mapping[str, Any]) -> dict[str, Any]:
962	"""The part of a report that ``e2e-proof.json`` carries for this operating system: who it is, which library, what the loopback showed and the claims it
963	allows. The blindness gate adds the ServerHello counts of the real capture next to it (``summarize``)."""
964	observed = (report.get("probes", {}).get("hybrid_client") or {}).get("server")
965	return {"os": report.get("os"), "python": report.get("python"), "openssl": report.get("openssl"), "terminator": report.get("terminator"),
966	"loopback": observed, "allowed_claims": list(report.get("allowed_claims", [])), "generated_at": report.get("generated_at")}
967
968
969	def main(argv: Sequence[str] \| None = None) -> int:
970	parser = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter)
971	parser.add_argument("command", choices=["report"])
972	parser.add_argument("--out", type=Path, help="write the JSON here as well")
973	parser.add_argument("--no-probe", action="store_true", help="runtime facts only: no handshake, no claim proven")
974	args = parser.parse_args(argv)
975	report = capability_report(probe=not args.no_probe)
976	text = json.dumps(report, indent=2, sort_keys=True)
977	if args.out:
978	args.out.write_text(text + "\n", encoding="utf-8")
979	print(text)
980	return 0 if report["terminator"]["can_start"] else 3
981
982
983	if __name__ == "__main__":
984	hide_child_consoles() # Windows: the probe's child interpreter opens no console window
985	sys.exit(main())