From f609a44953f5ae61916805fcb45ca337de00b0b0 Mon Sep 17 00:00:00 2001 From: ms-lead-reviewer Date: Sat, 18 Jul 2026 17:44:56 -0500 Subject: [PATCH] test(827): pin full D4 runtime closure --- .../probes/p3_d4_focused_run.py | 207 ++++++++++++++---- 1 file changed, 168 insertions(+), 39 deletions(-) diff --git a/docs/compaction-refresh/probes/p3_d4_focused_run.py b/docs/compaction-refresh/probes/p3_d4_focused_run.py index 8c68cd0..00f2f96 100644 --- a/docs/compaction-refresh/probes/p3_d4_focused_run.py +++ b/docs/compaction-refresh/probes/p3_d4_focused_run.py @@ -10,6 +10,7 @@ and ``p3_generation_broker.py``. It does not use the broader Gate0 runner. from __future__ import annotations import argparse +import ast import hashlib import json import os @@ -22,6 +23,7 @@ import sys import tempfile import threading import time +from dataclasses import dataclass from pathlib import Path from typing import Any, Callable @@ -32,8 +34,13 @@ HERE = Path(__file__).resolve().parent GATED_WI_ROOT_OVERRIDE = os.environ.get("GATED_WI_ROOT") GATED_WI_BRANCH = "refs/heads/feat/830-compaction-revoke" GATED_WI_HEAD = "f400830738998db105107a2a4c69c7f2a2a6fd5d" +GATED_WI_ANCESTOR = "66b1e0a0" +GATED_BROKER_HEAD = "23c0caca9b5d44002e6184cd7f2b6c837e8795b2" LEASE_BROKER_DIRECTORY = "packages/mosaic/framework/tools/lease-broker" +BROKER_RELATIVE_PATH = "docs/compaction-refresh/probes/p3_generation_broker.py" GATED_LAUNCHER_SHA256 = "e950e4224e280f16979d90cabb89aa1896c5ee28bed2df957e14d018d43cda82" +GATED_GENERATION_SHA256 = "061625402f08488eac47acd23272904e71fd1a71fd15b3bdab158632c801be4c" +GATED_BROKER_SHA256 = "4db4fef1ac6658a8ca79ad5091cefc901d2aa26003265c3d6726c294cf895cad" class PiRpc: @@ -305,22 +312,142 @@ def resolve_gated_wi_root() -> Path: raise RuntimeError("D4 precondition: GATED_WI_ROOT is not a git worktree") if head != GATED_WI_HEAD: raise RuntimeError(f"D4 precondition: gated WI head mismatch: {head}") + try: + forward_contains = subprocess.run( + [ + "git", + "-C", + str(gated_root), + "merge-base", + "--is-ancestor", + GATED_WI_ANCESTOR, + GATED_WI_HEAD, + ], + check=False, + ).returncode == 0 + except OSError as error: + raise RuntimeError("D4 precondition: cannot verify WI-3 ancestry") from error + if not forward_contains: + raise RuntimeError("D4 precondition: gated WI lacks required ancestor") return gated_root -def git_object_bytes(gated_root: Path, relative_path: str) -> bytes: +@dataclass(frozen=True) +class PinnedClosure: + launcher: Path + generation: Path + broker: Path + + +def git_object_bytes(git_root: Path, commit: str, relative_path: str) -> bytes: try: return subprocess.check_output( - ["git", "-C", str(gated_root), "show", f"{GATED_WI_HEAD}:{relative_path}"] + ["git", "-C", str(git_root), "show", f"{commit}:{relative_path}"] ) except (OSError, subprocess.CalledProcessError) as error: raise RuntimeError(f"D4 precondition: missing pinned source {relative_path}") from error +def closure_import_guard(member_sources: dict[str, str]) -> None: + """Refuse an incomplete project-code closure before materializing it.""" + + allowed_nonstdlib = {"lease_generation"} + stdlib = getattr(sys, "stdlib_module_names", frozenset()) + for name, source in member_sources.items(): + try: + tree = ast.parse(source, filename=name) + except SyntaxError as error: + raise RuntimeError(f"D4 precondition: pinned {name} does not parse") from error + for node in ast.walk(tree): + module: str | None = None + if isinstance(node, ast.Import): + for alias in node.names: + module = alias.name.split(".", maxsplit=1)[0] + if module not in stdlib and module not in allowed_nonstdlib: + raise RuntimeError(f"D4 precondition: unpinned import {module} in {name}") + elif isinstance(node, ast.ImportFrom): + if node.level: + raise RuntimeError(f"D4 precondition: relative import in {name}") + if node.module: + module = node.module.split(".", maxsplit=1)[0] + if module not in stdlib and module not in allowed_nonstdlib: + raise RuntimeError(f"D4 precondition: unpinned import {module} in {name}") + + +def write_pinned_file(path: Path, data: bytes) -> None: + descriptor = os.open( + path, + os.O_WRONLY | os.O_CREAT | os.O_EXCL | os.O_CLOEXEC, + 0o600, + ) + try: + remaining = memoryview(data) + while remaining: + written = os.write(descriptor, remaining) + if written <= 0: + raise OSError("pinned write made no progress") + remaining = remaining[written:] + finally: + os.close(descriptor) + + +def materialize_closure(root: Path, gated_root: Path, gate0_root: Path) -> PinnedClosure: + """Pin the complete project-authored runtime closure inside this fixture.""" + + launcher_relative = f"{LEASE_BROKER_DIRECTORY}/launch-runtime.py" + generation_relative = f"{LEASE_BROKER_DIRECTORY}/lease_generation.py" + members = ( + ( + "launch-runtime.py", + gated_root, + GATED_WI_HEAD, + launcher_relative, + GATED_LAUNCHER_SHA256, + ), + ( + "lease_generation.py", + gated_root, + GATED_WI_HEAD, + generation_relative, + GATED_GENERATION_SHA256, + ), + ( + "p3_generation_broker.py", + gate0_root, + GATED_BROKER_HEAD, + BROKER_RELATIVE_PATH, + GATED_BROKER_SHA256, + ), + ) + member_bytes: dict[str, bytes] = {} + member_sources: dict[str, str] = {} + for name, git_root, commit, relative_path, digest in members: + data = git_object_bytes(git_root, commit, relative_path) + if hashlib.sha256(data).hexdigest() != digest: + raise RuntimeError(f"D4 precondition: {name} hash mismatch") + try: + member_sources[name] = data.decode("utf-8") + except UnicodeDecodeError as error: + raise RuntimeError(f"D4 precondition: pinned {name} is not UTF-8") from error + member_bytes[name] = data + closure_import_guard(member_sources) + + pinned = root / "pinned" + pinned.mkdir(mode=0o700) + paths = {name: pinned / name for name, *_ in members} + for name, path in paths.items(): + write_pinned_file(path, member_bytes[name]) + return PinnedClosure( + launcher=paths["launch-runtime.py"], + generation=paths["lease_generation.py"], + broker=paths["p3_generation_broker.py"], + ) + + def gated_launcher_precondition( root: Path, socket_path: Path, environment: dict[str, str] -) -> tuple[Path, Path]: - """Verify the pinned WI-3 source before the adjacent in-place launch guard.""" +) -> PinnedClosure: + """Verify and materialize the full WI-3/probe closure before execution.""" if environment.get("MOSAIC_LEASE_BROKER_SOCKET") != str(socket_path): raise RuntimeError("D4 precondition: lease broker socket is not this fixture") @@ -347,20 +474,11 @@ def gated_launcher_precondition( raise RuntimeError("D4 precondition: fixture socket is outside this run's temporary root") gated_root = resolve_gated_wi_root() - launcher_relative = f"{LEASE_BROKER_DIRECTORY}/launch-runtime.py" - generation_relative = f"{LEASE_BROKER_DIRECTORY}/lease_generation.py" - launcher_bytes = git_object_bytes(gated_root, launcher_relative) - generation_bytes = git_object_bytes(gated_root, generation_relative) - if hashlib.sha256(launcher_bytes).hexdigest() != GATED_LAUNCHER_SHA256: - raise RuntimeError("D4 precondition: gated WI launcher hash mismatch") - try: - launcher_source = launcher_bytes.decode("utf-8") - generation_source = generation_bytes.decode("utf-8") - except UnicodeDecodeError as error: - raise RuntimeError("D4 precondition: pinned launcher is not UTF-8") from error - - # The exact launcher digest above is the trust anchor. These marker checks - # are diagnostic belt-and-suspenders only, never a substitute for the pin. + closure = materialize_closure(root, gated_root, repository_root()) + launcher_source = closure.launcher.read_text(encoding="utf-8") + generation_source = closure.generation.read_text(encoding="utf-8") + # Exact hashes in materialize_closure are the trust anchor. These marker + # checks are belt-and-suspenders diagnostics only. behavior_markers = ( '"action": "register_anchor"', "initialize_runtime_generation(generation_file, generation)", @@ -374,8 +492,7 @@ def gated_launcher_precondition( and "def bump_runtime_generation" in generation_source ): raise RuntimeError("D4 precondition: pinned launcher lacks file-generation markers") - - return gated_root / launcher_relative, gated_root / generation_relative + return closure def launch_verified_pi( @@ -407,12 +524,38 @@ def launch_verified_pi( str(extension), ] # This is deliberately the statement immediately before Popen (inside - # PiRpc): the pinned worktree bytes are re-hashed then executed in place. + # PiRpc): the fixture-pinned launcher bytes are re-hashed then executed. if hashlib.sha256(launcher.read_bytes()).hexdigest() != GATED_LAUNCHER_SHA256: raise RuntimeError("D4 precondition: adjacent launcher hash mismatch") return PiRpc(command, workspace, environment) +def launch_verified_broker( + broker_path: Path, generation_path: Path, socket_path: Path, log_path: Path +) -> subprocess.Popen[str]: + command = [ + sys.executable, + str(broker_path), + "--socket", + str(socket_path), + "--log", + str(log_path), + "--generation-module", + str(generation_path), + ] + if hashlib.sha256(broker_path.read_bytes()).hexdigest() != GATED_BROKER_SHA256: + raise RuntimeError("D4 precondition: pinned broker hash mismatch") + # The final helper re-hash is immediately adjacent to the broker Popen. + if hashlib.sha256(generation_path.read_bytes()).hexdigest() != GATED_GENERATION_SHA256: + raise RuntimeError("D4 precondition: pinned helper hash mismatch") + return subprocess.Popen( + command, + text=True, + stdout=subprocess.PIPE, + stderr=subprocess.STDOUT, + ) + + def assert_d4(records: list[dict[str, Any]]) -> dict[str, object]: def record_where(description: str, candidates: list[dict[str, Any]]) -> dict[str, Any]: if not candidates: @@ -599,26 +742,12 @@ def run_once(index: int) -> Path: environment = isolated_environment(root, index, workspace, socket_path, pi_log) # Must run before the fixture broker or Pi process is launched. It proves # the launcher registers before exec and can only read this fixture socket. - gated_launcher, generation_module = gated_launcher_precondition( - root, socket_path, environment - ) - broker = subprocess.Popen( - [ - sys.executable, - str(HERE / "p3_generation_broker.py"), - "--socket", - str(socket_path), - "--log", - str(generation_log), - "--generation-module", - str(generation_module), - ], - text=True, - stdout=subprocess.PIPE, - stderr=subprocess.STDOUT, + closure = gated_launcher_precondition(root, socket_path, environment) + broker = launch_verified_broker( + closure.broker, closure.generation, socket_path, generation_log ) wait_path(socket_path) - pi = launch_verified_pi(gated_launcher, workspace, sessions, extension, environment) + pi = launch_verified_pi(closure.launcher, workspace, sessions, extension, environment) pi.send({"id": "state", "type": "get_state"}) state = pi.response("state") original_session = state["data"]["sessionFile"]