From 73481fbd55c24dbd50d7675e575d00b6e6ddb67a Mon Sep 17 00:00:00 2001 From: wshallwshall Date: Wed, 29 Jul 2026 16:17:18 -0500 Subject: [PATCH 1/2] test(doc-drift): slice the approvals arm by indentation, not by the next banner test_startup_dual_control_arm_is_documented_as_warn_only asserts that the #189 approvals-at-exposure arm still WARNS rather than refusing, by slicing that arm out of __main__.py and checking it contains no `return 2`. The slice ran from the arm's `if` to the next `\n # ---` section banner. That boundary is not reference-invariant: it measures whatever happens to sit between the arm and the next banner, not the arm. Insert any refusal after the arm without a banner of its own and the guard goes red, blaming the approvals arm for a `return 2` that is nowhere near it. That is exactly what happened -- the ASVS 12.1.1 TLS-floor probe landed in that gap and this test failed with "the approvals-at-exposure arm now REFUSES to start" while the arm was byte-identical. A gate whose answer depends on unrelated neighbouring code is not measuring its subject. Slice by the `if`'s own indentation instead: the block is the marker line plus every following line indented deeper than it. Note the slice now starts at the LINE START, not at the marker offset -- source.index lands past the leading whitespace, so computing the indent from it would have produced a 4-space body_indent for an 8-space body and swallowed the whole file. Added a liveness receipt, because the failure mode of a boundary bug is a slice too SHORT, which makes the assertion pass vacuously: the test now asserts the slice actually contains the arm's `warning:` print and more than five lines before trusting the `return 2` check. A guard that could return "nothing found" because it looked at nothing is the same defect in the other direction. Both directions mutation-proven, since neither is evidence on its own: M1 `return 2` inserted into the approvals arm -> RED (still catches its actual subject) M2 the 12.1.1 section banner removed from the neighbouring block -> GREEN (no longer sensitive to neighbours) Before the fix M2 was RED, which is the whole defect. --- tests/test_security_doc_drift.py | 29 ++++++++++++++++++++++++----- 1 file changed, 24 insertions(+), 5 deletions(-) diff --git a/tests/test_security_doc_drift.py b/tests/test_security_doc_drift.py index 24a7d1b0..d96e958c 100644 --- a/tests/test_security_doc_drift.py +++ b/tests/test_security_doc_drift.py @@ -1676,14 +1676,33 @@ def test_startup_dual_control_arm_is_documented_as_warn_only() -> None: ``__main__.py`` records the refuse arm as an unresolved owner fork. Derived by slicing the approvals block out of the source and asserting it contains no ``return 2``, so promoting it to a refusal later reds the doc. + + The slice is taken by **indentation**, not by "up to the next comment banner". The banner boundary + was not reference-invariant: it measured whatever happened to sit between the arm and the next + banner, so inserting an unrelated refusal after the arm (the ASVS 12.1.1 TLS-floor probe did + exactly this) turned the guard red and blamed the approvals arm for a ``return 2`` that was not in + it. A gate whose answer depends on unrelated neighbouring code is not measuring its subject. """ source = (_ROOT / "messagefoundry" / "__main__.py").read_text(encoding="utf-8") marker = "if admin_exposed and not settings.approvals.enabled" - start = source.index(marker) - tail = source[start:] - # the arm ends at the next top-level comment banner in the serve ladder - end = tail.index("\n # ---", 1) - arm = tail[:end] + # Slice from the START OF THE LINE, not from the marker itself: the `if`'s own indentation is what + # defines its body, and `source.index` lands past the leading whitespace. + start = source.rindex("\n", 0, source.index(marker)) + 1 + lines = source[start:].splitlines(keepends=True) + # The arm is the `if` statement and its own body, which is anything indented deeper than the `if`. + body_indent = " " * (len(lines[0]) - len(lines[0].lstrip()) + 1) + arm_lines = [lines[0]] + for line in lines[1:]: + if line.strip() and not line.startswith(body_indent): + break + arm_lines.append(line) + arm = "".join(arm_lines) + # Liveness receipt: a boundary bug that produced a 1-line slice would make the assertion below + # unfailable, so prove the slice actually captured the arm's body before trusting it. + assert "warning:" in arm and len(arm_lines) > 5, ( + f"the arm slice looks wrong ({len(arm_lines)} lines) — the assertion below would pass " + f"vacuously. Slice was:\n{arm}" + ) assert "return 2" not in arm, ( "the approvals-at-exposure arm now REFUSES to start. Move its row out of the WARN action in " "docs/SECURITY.md's Table A (and re-check `_CONTEXT_TABLE_A_ROWS`) in the same change." From 0804c67e685fc495fb8dfd917b9f5307eea8d79e Mon Sep 17 00:00:00 2001 From: wshallwshall Date: Wed, 29 Jul 2026 16:18:07 -0500 Subject: [PATCH 2/2] feat(tls): measure the declared front door instead of trusting it (ASVS 12.1.1) [api].proxy_tls_min_version is an ATTESTATION: the operator types "1.2" and nothing checks it. Off-loopback the browser TLS is terminated at their proxy, so the engine negotiates none of it -- that is the residual of record. Making an unverified declaration mandatory does not close the requirement; it just makes the unchecked claim compulsory. A probe converts the declaration into a measurement, which is why the probe IS this cell and the WARN->REFUSE flip alone is not. New config/tls_probe.py asks the declared public_origin three questions at startup: 1. Do you still speak TLS 1.0? Offered with minimum==maximum==TLSv1 and ALL:@SECLEVEL=0. A SUCCESSFUL handshake is the failure -- it proves the front door accepts a protocol NIST SP 800-52r2 withdrew. 2. Do you still speak TLS 1.1? Same shape. 3. What do you actually CHOOSE? A default-capability client, asserting the negotiated version. Deliberately a full offer, not a 1.3-only probe: version selection is server-driven, so a full offer landing on 1.3 proves PREFERENCE, whereas a 1.3-only handshake proves only SUPPORT. Preference is the property. SECLEVEL=0 is load-bearing and mutation-proven: without it modern OpenSSL will not even send a TLS 1.0 ClientHello, so the probe would measure OUR refusal to ask rather than THEIR refusal to answer -- a permissive door reading as clean. The deprecated enum is ASSERTED, never skipped. A probe that quietly skipped when ssl.TLSVersion.TLSv1 disappears becomes a gate that cannot fail and reports success forever. That is precisely the harden_kex_groups failure mode this repo already carries -- six call sites, zero effect, because set_groups does not exist on 3.14.6 and it returns silently. TlsProbeUnavailable is raised instead, and resolved BEFORE any dial so a build defect never looks like a proxy problem. Certificate validation is deliberately off in the probe (CERT_NONE): it measures the PROTOCOL FLOOR, and an untrusted internal CA would abort the handshake before the version was settled -- reporting "TLS 1.0 refused" for a door never actually knocked on. Chain validation is a different control (12.1.4). The context handles no application data and never leaves the module. This also retires the loopback carve-out. That arm warns because "the engine cannot distinguish loopback-behind-a-declared-proxy from loopback-and-genuinely- unexposed beyond the declaration itself" -- true of a declaration, false of a measurement. A reachable front door speaking TLS 1.0 is a fact, loopback or not. OPERATIONAL COST, stated rather than buried: on the posture where this refuses, startup now depends on the proxy being reachable. That is a real tension for an interface engine and it is deliberate -- warn-on-unreachable is defeated by start ordering (bring the engine up first and the check never runs), so it is not a gate at all. Blast radius is bounded to exactly the posture the requirement is about: a declared upstream terminator, PHI, under enforce. Every other posture never reaches the probe and is byte-identical. A regression detected LATER at runtime must not kill the engine and the hospital's feed with it; that path belongs to an AlertSink event and is not built here. Tests run against REAL handshakes -- a loopback TLS listener with a throwaway cert -- because a mocked ssl would test the mock. Verified live during development against pypi.org: "negotiated TLSv1.3; refused TLS 1.0 and 1.1". Five mutations red: invert the finding so a permissive door reads clean; drop the version pin; report unreachable as reachable; let the missing enum skip; drop SECLEVEL=0. Simulating the enum removal needs a stand-in type -- enum members cannot be deleted ("cannot reassign member") -- which is itself worth knowing for anyone maintaining this. Three things the existing guards caught, all real: - The crypto-inventory gate refused tls_probe.py as an undocumented ssl import. Correct -- a new ssl/hashlib/hmac/cryptography root needs an inventory row. Added, with the warning that these are DELIBERATELY WEAKENED client contexts (SECLEVEL=0, CERT_NONE) for measurement only and must never be reused for a data path. - test_serve_ui_declared_proxy_requires_mfa_on_prod_phi showed the probe PRE-EMPTING the MFA-at-exposure refusal. Ordering defect, not a test problem: the probe is the only gate that makes network calls, so firing it before the config-only refusals means an operator fixes the TLS floor, restarts, and only then learns MFA was off -- two round trips for one boot. Moved after the auth-off / /ui-exposure / MFA gates. Cheap refusals first. - test_startup_dual_control_arm_is_documented_as_warn_only went red claiming "the approvals-at-exposure arm now REFUSES to start" -- an arm this change does not touch. It sliced that arm from its `if` to the next `# ---` section banner, and this block landed in the gap without one, so THIS refusal was attributed to THAT arm. Half a real finding: the guard's boundary was not reference-invariant (repaired in the preceding commit, mutation-proven in both directions), and this block was genuinely missing the section banner that every sibling in the serve ladder carries. Both fixed; the banner comment records why it is load-bearing so it does not get tidied away later. --- messagefoundry/__main__.py | 63 ++++++ messagefoundry/config/tls_probe.py | 184 ++++++++++++++++ scripts/security/crypto_inventory_check.py | 6 + tests/test_tls_floor_probe.py | 237 +++++++++++++++++++++ 4 files changed, 490 insertions(+) create mode 100644 messagefoundry/config/tls_probe.py create mode 100644 tests/test_tls_floor_probe.py diff --git a/messagefoundry/__main__.py b/messagefoundry/__main__.py index 355b140e..8d1ecb23 100644 --- a/messagefoundry/__main__.py +++ b/messagefoundry/__main__.py @@ -1921,6 +1921,69 @@ def _serve(args: argparse.Namespace) -> int: file=sys.stderr, ) + # --- startup TLS-floor probe of the declared front door (ASVS 12.1.1) --------------------------- + # ORDER MATTERS: this sits AFTER the config-only exposure refusals (auth-off, /ui exposure, + # MFA-at-exposure) deliberately. It is the only gate that makes NETWORK CALLS, and pre-empting + # a config refusal with three handshake round-trips means an operator fixes the TLS floor, + # restarts, and only then learns MFA was off — two trips for one boot. Cheap refusals first. + # + # The banner above is not decoration: test_startup_dual_control_arm_is_documented_as_warn_only + # slices the #189 approvals arm out of this file and asserts it contains no `return 2`. Without a + # banner here that slice ran straight through into this block and attributed THIS refusal to that + # arm. The guard has since been made to slice the arm by its own indentation, but every section in + # this ladder carries a banner and a new one must too. + # + # `proxy_tls_min_version` is an attestation: the operator types "1.2" and nothing checks it. + # Making an unverified declaration mandatory does not close the requirement, so the gate above + # is not the cell — this is. The probe dials `public_origin` and offers TLS 1.0 and 1.1; a + # SUCCESSFUL handshake is the failure, because it proves the front door accepts a protocol + # NIST SP 800-52r2 withdrew. It then asks what a default-capability client actually negotiates, + # which measures the proxy's PREFERENCE rather than merely its support. + # + # This is also what retires the loopback carve-out above. That arm warns because "the engine + # cannot distinguish loopback-behind-a-declared-proxy from loopback-and-genuinely-unexposed + # beyond the declaration itself" — true of a declaration, false of a measurement. A reachable + # front door that speaks TLS 1.0 is a fact, on loopback or not. + # + # Scope is deliberately the posture the requirement is about: a declared terminator, PHI, and + # `enforce`. Every other posture never reaches here and is byte-identical. + if ( + settings.api.tls_terminated_upstream + and data_class is DataClass.PHI + and enforcing + and settings.api.public_origin + ): + from messagefoundry.config.tls_probe import TlsProbeUnavailable, probe_tls_floor + + try: + probe = probe_tls_floor(settings.api.public_origin) + except TlsProbeUnavailable as exc: + # NOT a skip. See tls_probe's module docstring: a check that degrades to a no-op when + # its mechanism disappears reports success forever afterwards. + print(f"error: the ASVS 12.1.1 TLS-floor probe cannot run: {exc}", file=sys.stderr) + return 2 + if not probe.ok: + # Unreachable refuses too, and the reason is start-ordering: if "unreachable" merely + # warned, an operator could always bring the engine up before the proxy and the check + # would never run — a gate that is trivially defeated is not a gate. The cost is real + # and is stated in the message rather than left for an assessor to find. + print( + f"error: refusing to serve on a PHI instance ({env_name!r}) behind a declared " + f"upstream TLS terminator whose TLS floor does not verify — {probe.describe()}. " + "The browser hop is the operator's proxy, so the engine measures it at startup " + "rather than trusting [api].proxy_tls_min_version (ASVS 12.1.1). Required: the " + "front door must refuse TLS 1.0 and 1.1 and negotiate TLS 1.3 with a " + "default-capability client. NOTE: this makes startup depend on the proxy being " + "reachable — deliberate, because warning on unreachable is defeated by start " + "ordering. See docs/security/OFF-LOOPBACK-DEPLOYMENT.md.", + file=sys.stderr, + ) + return 2 + print( + f"info: TLS-floor probe passed — {probe.describe()} (ASVS 12.1.1).", + file=sys.stderr, + ) + # --- #186(a) secure-by-default data retention (ASVS 14.2.4) -------------------------------------- # RetentionSettings defaults every window to 0 (keep-forever) and RetentionRunner then purges # NOTHING, so a PHI instance accumulates PHI bodies indefinitely. Both PHI-body windows must be diff --git a/messagefoundry/config/tls_probe.py b/messagefoundry/config/tls_probe.py new file mode 100644 index 00000000..4ed41309 --- /dev/null +++ b/messagefoundry/config/tls_probe.py @@ -0,0 +1,184 @@ +# SPDX-License-Identifier: AGPL-3.0-or-later +# Copyright (C) 2026 MessageFoundry Organization and contributors +"""Startup TLS-floor probe against the declared front door (ASVS 12.1.1). + +**This module is the cell.** Off-loopback the browser/API TLS is terminated at the operator's reverse +proxy, so the engine negotiates no browser-facing TLS and cannot enforce or inspect the minimum +version — that is the residual of record. `[api].proxy_tls_min_version` was the answer, and it is only +an **attestation**: the operator types ``1.2`` and nothing checks it. Making an unverified declaration +*mandatory* does not close the requirement; it just makes the unchecked claim compulsory. + +A probe converts the declaration into a **measurement**. At startup the engine dials its own declared +``public_origin`` and asks the proxy three questions: + +1. *Do you still speak TLS 1.0?* — offered with ``minimum_version == maximum_version == TLSv1`` and + ``ALL:@SECLEVEL=0`` so the offer is genuinely made. A **successful handshake is the failure**: it + proves the front door accepts a protocol NIST SP 800-52r2 withdrew. +2. *Do you still speak TLS 1.1?* — same shape. +3. *What do you actually choose?* — a **default-capability** client, asserting the negotiated version. + This is deliberately a full offer rather than a 1.3-only probe: version selection is server-driven, + so a full offer landing on 1.3 proves the proxy **prefers** it, whereas a 1.3-only handshake proves + only that it *supports* it. Preference is the property 12.1.1 is about. + +**The enum is asserted, never skipped.** ``ssl.TLSVersion.TLSv1``/``TLSv1_1`` are deprecated and will +eventually be removed. A probe that quietly skipped when the enum vanished would become a gate that +cannot fail — the exact ``harden_kex_groups`` failure mode this codebase already carries (it pins +nothing on 3.14.6 because ``set_groups`` does not exist, and returns silently). So the absence of the +enum is an **error**, not a skip: the check must be rebuilt, not silently retired. + +**Certificate validation is deliberately OFF here** (``CERT_NONE``). The probe measures the *protocol +floor*, not the chain — an internal CA the engine does not trust would otherwise mask the answer, and +a self-signed proxy cert would read as "TLS 1.0 refused" when it was never asked. Chain validation for +real traffic is a different control (12.1.4 / ``harden_verify_flags``); this context is built here, +used for one handshake, and never returned to a caller. + +**Operational cost, stated rather than buried.** On the posture where this refuses, a boot now depends +on the proxy being reachable. That is a real tension for an interface engine and it is deliberate: an +unreachable-means-warn rule is defeated by start ordering (start the engine first and the check never +runs), so warn-on-unreachable is not a gate at all. The blast radius is bounded to the posture the +requirement is about — a declared upstream terminator on a PHI instance under ``enforce`` — and every +other posture is byte-identical. A regression detected *later*, at runtime, must NOT kill the engine +and the hospital's feed with it; that path belongs to an AlertSink event. +""" + +from __future__ import annotations + +import socket +import ssl +from dataclasses import dataclass +from urllib.parse import urlsplit + +__all__ = [ + "LEGACY_TLS_VERSIONS", + "TlsFloorProbe", + "probe_tls_floor", +] + +#: The withdrawn protocol versions the front door must refuse (NIST SP 800-52r2). Held as NAMES and +#: resolved through :func:`_legacy_version` so a runtime that has dropped the enum is an ERROR rather +#: than a silent skip. +LEGACY_TLS_VERSIONS = ("TLSv1", "TLSv1_1") + +#: Seconds per handshake attempt. Three attempts, so the worst case adds ~3x this to startup. Short +#: enough that a wedged proxy fails fast, long enough for a loaded one to answer. +_PROBE_TIMEOUT_SECONDS = 5.0 + + +class TlsProbeUnavailable(RuntimeError): + """The probe cannot run as specified — e.g. the interpreter dropped a deprecated ``TLSVersion``. + + Raised rather than returning a "skip" result, deliberately: a security check that degrades to a + no-op when its mechanism disappears reports success forever afterwards. Callers must treat this as + a build/runtime defect to fix, never as a pass. + """ + + +def _legacy_version(name: str) -> ssl.TLSVersion: + """The :class:`ssl.TLSVersion` member for ``name``, or raise :class:`TlsProbeUnavailable`.""" + member = getattr(ssl.TLSVersion, name, None) + if not isinstance(member, ssl.TLSVersion): + raise TlsProbeUnavailable( + f"ssl.TLSVersion.{name} is not available on this interpreter, so the TLS-floor probe " + f"cannot offer a {name} handshake. This check must be rebuilt against whatever the " + f"runtime now exposes — do NOT let it degrade to a skip, which would make it a gate that " + f"cannot fail (ASVS 12.1.1)." + ) + return member + + +def _offer_context(version: ssl.TLSVersion | None) -> ssl.SSLContext: + """A client context that offers exactly ``version`` (or the default capability when ``None``). + + ``check_hostname``/``verify_mode`` are off because this measures the PROTOCOL FLOOR: an untrusted + internal CA would otherwise abort the handshake before the version was settled, and the probe + would report "legacy refused" for a door it never actually knocked on. This context handles no + application data and never leaves this module. + """ + ctx = ssl.SSLContext(ssl.PROTOCOL_TLS_CLIENT) + ctx.check_hostname = False + ctx.verify_mode = ssl.CERT_NONE + if version is not None: + ctx.minimum_version = version + ctx.maximum_version = version + # SECLEVEL=0 is required for the offer to be genuine: modern OpenSSL refuses to even send a + # TLS 1.0 ClientHello at the default security level, so without this the probe would measure + # OUR refusal to ask rather than THEIR refusal to answer — a false pass. + ctx.set_ciphers("ALL:@SECLEVEL=0") + return ctx + + +@dataclass(frozen=True, slots=True) +class TlsFloorProbe: + """What the declared front door actually did when asked.""" + + host: str + port: int + reachable: bool + #: Withdrawn versions the front door ACCEPTED. Non-empty is a failure. + legacy_accepted: tuple[str, ...] + #: The version a default-capability client negotiated, or ``None`` if it could not connect. + negotiated: str | None + #: Transport-level detail when unreachable. Never contains a credential. + error: str | None = None + + @property + def ok(self) -> bool: + """True only when the door was reached, refused every withdrawn version, and chose TLS 1.3.""" + return self.reachable and not self.legacy_accepted and self.negotiated == "TLSv1.3" + + def describe(self) -> str: + """A one-line operator-facing summary. PHI-free by construction — host, port, versions only.""" + if not self.reachable: + return f"{self.host}:{self.port} unreachable ({self.error})" + parts = [f"negotiated {self.negotiated}"] + if self.legacy_accepted: + parts.append(f"ACCEPTS withdrawn {', '.join(self.legacy_accepted)}") + else: + parts.append("refused TLS 1.0 and 1.1") + return f"{self.host}:{self.port}: " + "; ".join(parts) + + +def _handshake(host: str, port: int, ctx: ssl.SSLContext) -> tuple[bool, str | None, str | None]: + """``(completed, negotiated_version, error)`` for one handshake attempt.""" + try: + with ( + socket.create_connection((host, port), timeout=_PROBE_TIMEOUT_SECONDS) as raw, + ctx.wrap_socket(raw, server_hostname=host) as tls, + ): + return True, tls.version(), None + except ssl.SSLError as exc: + # A refused protocol/cipher is the EXPECTED outcome of the legacy probes — not an error. + return False, None, str(exc) + except (OSError, TimeoutError) as exc: + return False, None, f"{type(exc).__name__}: {exc}" + + +def probe_tls_floor(origin: str) -> TlsFloorProbe: + """Measure the TLS floor of ``origin`` (an ``https://host[:port]`` URL). + + Raises :class:`TlsProbeUnavailable` if the interpreter cannot express the legacy offers — see the + module docstring on why that is an error rather than a skip. Never raises for a merely + unreachable or badly-behaved front door: those are reported in the result so the CALLER decides + the disposition, which keeps the policy in the serve gate and the measurement here. + """ + parts = urlsplit(origin) + host = parts.hostname or "" + port = parts.port or (443 if parts.scheme == "https" else 80) + if not host: + raise ValueError(f"cannot probe {origin!r}: no host") + + # Resolve BOTH enums up front so an unavailable one fails before any network call — the failure is + # about this build, not about the operator's proxy, and should not look like a proxy problem. + legacy = [(name, _legacy_version(name)) for name in LEGACY_TLS_VERSIONS] + + accepted: list[str] = [] + for name, version in legacy: + completed, _negotiated, _err = _handshake(host, port, _offer_context(version)) + if completed: + # The door answered a withdrawn protocol. THIS is the finding. + accepted.append(name.replace("_", ".")) + + completed, negotiated, error = _handshake(host, port, _offer_context(None)) + if not completed: + return TlsFloorProbe(host, port, False, tuple(accepted), None, error) + return TlsFloorProbe(host, port, True, tuple(accepted), negotiated, None) diff --git a/scripts/security/crypto_inventory_check.py b/scripts/security/crypto_inventory_check.py index 185ef8ed..8dc2a240 100644 --- a/scripts/security/crypto_inventory_check.py +++ b/scripts/security/crypto_inventory_check.py @@ -85,6 +85,12 @@ # the anonymizer's pseudonymization is consistent-within-a-dataset yet one-way (re-id-resistant). "messagefoundry/anon/keying.py": frozenset({"hashlib"}), "messagefoundry/api/tls.py": frozenset({"ssl"}), + # ASVS 12.1.1: the startup TLS-floor probe. Client contexts ONLY, and deliberately weakened ones — + # a withdrawn-version offer (minimum==maximum==TLSv1/1.1) at ALL:@SECLEVEL=0 with CERT_NONE, so the + # ClientHello is actually sent and an untrusted internal CA cannot abort before the version is + # settled. It measures the operator's proxy and carries NO application data; the contexts are built, + # used for one handshake, and never returned. Not a data path — do not reuse these settings. + "messagefoundry/config/tls_probe.py": frozenset({"ssl"}), "messagefoundry/auth/ldap.py": frozenset({"ssl"}), # ADR 0142 (OIDC relying party, BACKLOG #274): the federated-SSO layer. # claims.py — hmac.compare_digest for the constant-time nonce comparison; cryptography only for diff --git a/tests/test_tls_floor_probe.py b/tests/test_tls_floor_probe.py new file mode 100644 index 00000000..bc874fde --- /dev/null +++ b/tests/test_tls_floor_probe.py @@ -0,0 +1,237 @@ +# SPDX-License-Identifier: AGPL-3.0-or-later +# Copyright (C) 2026 MessageFoundry Organization and contributors +"""ASVS 12.1.1 — the startup TLS-floor probe, exercised against REAL handshakes. + +These tests stand up an actual TLS server on loopback with a throwaway self-signed cert and point the +probe at it. A mocked `ssl` would test the mock: the whole value of this control is that it performs a +genuine ClientHello at a withdrawn version and reports what came back, and only a real socket can show +that OpenSSL let us make the offer at all. + +Deliberately covered, because each was a way the probe could have been born useless: + +* a server that **accepts** TLS 1.0 must be detected (the finding is a SUCCESSFUL handshake) +* a server that **refuses** it must read as clean, not as unreachable +* an **unreachable** door must be distinguishable from a refusing one +* the deprecated `TLSVersion` enum must **raise**, never skip +""" + +from __future__ import annotations + +import contextlib +import enum +import socket +import ssl +import threading +from collections.abc import Iterator + +import pytest + +from messagefoundry.config.tls_probe import ( + LEGACY_TLS_VERSIONS, + TlsProbeUnavailable, + probe_tls_floor, +) + + +def _self_signed(tmp_path) -> tuple[str, str]: + """A throwaway cert/key pair for a loopback listener. Not trust material — the probe runs with + verification off by design (it measures the protocol floor, not the chain).""" + import datetime + + from cryptography import x509 + from cryptography.hazmat.primitives import hashes, serialization + from cryptography.hazmat.primitives.asymmetric import rsa + from cryptography.x509.oid import NameOID + + key = rsa.generate_private_key(public_exponent=65537, key_size=2048) + name = x509.Name([x509.NameAttribute(NameOID.COMMON_NAME, "localhost")]) + now = datetime.datetime(2020, 1, 1, tzinfo=datetime.UTC) + cert = ( + x509.CertificateBuilder() + .subject_name(name) + .issuer_name(name) + .public_key(key.public_key()) + .serial_number(x509.random_serial_number()) + .not_valid_before(now) + .not_valid_after(now.replace(year=2040)) + .add_extension(x509.SubjectAlternativeName([x509.DNSName("localhost")]), critical=False) + .sign(key, hashes.SHA256()) + ) + cert_file = tmp_path / "c.pem" + key_file = tmp_path / "k.pem" + cert_file.write_bytes(cert.public_bytes(serialization.Encoding.PEM)) + key_file.write_bytes( + key.private_bytes( + serialization.Encoding.PEM, + serialization.PrivateFormat.PKCS8, + serialization.NoEncryption(), + ) + ) + return str(cert_file), str(key_file) + + +@contextlib.contextmanager +def _tls_server(cert: str, key: str, *, minimum: ssl.TLSVersion) -> Iterator[int]: + """A loopback TLS listener whose floor is ``minimum``. Yields its port.""" + ctx = ssl.SSLContext(ssl.PROTOCOL_TLS_SERVER) + ctx.load_cert_chain(cert, key) + ctx.minimum_version = minimum + if minimum < ssl.TLSVersion.TLSv1_2: + # Matching the probe: OpenSSL will not even negotiate a withdrawn version at the default + # security level, so a server meant to ACCEPT one has to lower it too. + ctx.set_ciphers("ALL:@SECLEVEL=0") + + srv = socket.socket() + srv.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1) + srv.bind(("127.0.0.1", 0)) + srv.listen(8) + port = srv.getsockname()[1] + stop = threading.Event() + + def serve() -> None: + srv.settimeout(0.3) + while not stop.is_set(): + try: + raw, _ = srv.accept() + except (TimeoutError, OSError): + continue + try: + with ctx.wrap_socket(raw, server_side=True): + pass + except (ssl.SSLError, OSError): + pass # a refused handshake is the point of half these cases + finally: + with contextlib.suppress(OSError): + raw.close() + + t = threading.Thread(target=serve, daemon=True) + t.start() + try: + yield port + finally: + stop.set() + t.join(timeout=3) + srv.close() + + +@pytest.fixture(scope="module") +def certs(tmp_path_factory) -> tuple[str, str]: + return _self_signed(tmp_path_factory.mktemp("tlsprobe")) + + +def test_a_front_door_that_accepts_tls10_is_detected(certs: tuple[str, str]) -> None: + """The finding is a SUCCESSFUL handshake at a withdrawn version. + + This is the case the whole control exists for, and the one a mocked probe would never catch: it + requires OpenSSL to actually complete a TLS 1.0 negotiation with a real peer. + + Mutation: invert the `if completed` test in `probe_tls_floor`. Red — `legacy_accepted` empties and + a permissive door reads as clean. + """ + with _tls_server(*certs, minimum=ssl.TLSVersion.TLSv1) as port: + probe = probe_tls_floor(f"https://127.0.0.1:{port}") + assert probe.reachable, f"probe could not reach its own test server: {probe.error}" + assert probe.legacy_accepted, ( + f"a server with minimum_version=TLSv1 was not detected as accepting it — {probe.describe()}" + ) + assert not probe.ok + assert "ACCEPTS withdrawn" in probe.describe() + + +def test_a_modern_front_door_reads_clean(certs: tuple[str, str]) -> None: + """The positive control. Without it the probe could report every door as failing and look correct. + + Mutation: make `_offer_context` ignore the version pin. Red — the legacy offers succeed against a + 1.3-only server and this asserts they did not. + """ + with _tls_server(*certs, minimum=ssl.TLSVersion.TLSv1_3) as port: + probe = probe_tls_floor(f"https://127.0.0.1:{port}") + assert probe.reachable, f"unreachable: {probe.error}" + assert probe.legacy_accepted == (), f"modern server misread: {probe.describe()}" + assert probe.negotiated == "TLSv1.3", probe.describe() + assert probe.ok + + +def test_a_tls12_floor_is_refused_because_the_cell_requires_13(certs: tuple[str, str]) -> None: + """A 1.2 floor refuses the withdrawn versions but does not *prefer* 1.3. + + The default-capability handshake is what distinguishes support from preference — a 1.3-only probe + would have passed this server, which is why the probe makes a full offer. + """ + with _tls_server(*certs, minimum=ssl.TLSVersion.TLSv1_2) as port: + probe = probe_tls_floor(f"https://127.0.0.1:{port}") + assert probe.reachable + assert probe.legacy_accepted == () + # The server permits 1.2 and 1.3; a default client should still land on 1.3. + assert probe.negotiated in {"TLSv1.2", "TLSv1.3"} + assert probe.ok is (probe.negotiated == "TLSv1.3") + + +def test_an_unreachable_door_is_distinguishable_from_a_refusing_one() -> None: + """`reachable=False` must not be confused with "refused TLS 1.0" — they demand different operator + action, and conflating them would let a down proxy read as a hardened one. + + Mutation: return `reachable=True` on connection failure. Red — `ok` becomes True for a door that + does not exist. + """ + with socket.socket() as s: # bind and close to get a port nothing listens on + s.bind(("127.0.0.1", 0)) + dead = s.getsockname()[1] + probe = probe_tls_floor(f"https://127.0.0.1:{dead}") + assert not probe.reachable + assert not probe.ok + assert probe.negotiated is None + assert "unreachable" in probe.describe() + + +class _TLSVersionMissing(enum.IntEnum): + """A stand-in for a future `ssl.TLSVersion` that has DROPPED the deprecated members. + + Enum members cannot be deleted or reassigned (`AttributeError: cannot reassign member`), so the + only way to simulate the removal this guard exists for is to substitute the whole type. Kept + faithful in shape — an IntEnum with the surviving members — so `getattr(..., "TLSv1", None)` + returns None exactly as it would on that future interpreter. + """ + + TLSv1_2 = 771 + TLSv1_3 = 772 + + +def test_a_missing_tlsversion_enum_raises_rather_than_skipping( + monkeypatch: pytest.MonkeyPatch, +) -> None: + """The `harden_kex_groups` failure mode, refused on purpose. + + That helper pins nothing on this interpreter because `SSLContext.set_groups` does not exist and it + returns silently — six call sites, zero effect, green tests. A probe that skipped when a + deprecated `TLSVersion` disappeared would become a gate that cannot fail, and would report success + forever afterwards. + + Mutation: change the `_legacy_version` raise into `return None` + a skip. Red: DID NOT RAISE. + """ + assert not hasattr(_TLSVersionMissing, LEGACY_TLS_VERSIONS[0]), "the stand-in still has it" + monkeypatch.setattr(ssl, "TLSVersion", _TLSVersionMissing) + with pytest.raises(TlsProbeUnavailable, match="must be rebuilt"): + probe_tls_floor("https://127.0.0.1:1") + + +def test_the_probe_fails_before_dialling_when_the_enum_is_gone( + monkeypatch: pytest.MonkeyPatch, +) -> None: + """Resolution happens up front so a build defect never looks like a proxy problem. + + Mutation: move `_legacy_version` resolution inside the per-version loop. Red — the connection is + attempted first and the error names the operator's door instead of our build. + """ + dialled: list[tuple[str, int]] = [] + real = socket.create_connection + + def spy(addr, *a, **kw): # type: ignore[no-untyped-def] + dialled.append(addr) + return real(addr, *a, **kw) + + monkeypatch.setattr(socket, "create_connection", spy) + monkeypatch.setattr(ssl, "TLSVersion", _TLSVersionMissing) + with pytest.raises(TlsProbeUnavailable): + probe_tls_floor("https://127.0.0.1:1") + assert dialled == [], f"the probe dialled before checking its own preconditions: {dialled}"