mosaicstack/stack
1
0
Fork 0
Code Issues 81 Pull Requests 1 Actions Packages Projects Releases 15 Wiki Activity

fix(827): harden pinned launcher validation

Browse Source
This commit is contained in:
ms-lead-reviewer
2026-07-18 16:48:53 -05:00
parent 7ff63cd5c1
commit 6164dc0794
1 changed files with 18 additions and 14 deletions
Download Patch File Download Diff File Expand all files Collapse all files

32
docs/compaction-refresh/probes/p3_d4_focused_run.py
View File

@@ -292,11 +292,17 @@ def resolve_gated_wi_root() -> Path:
try: try:
if not gated_root.is_dir(): if not gated_root.is_dir():
raise RuntimeError("D4 precondition: GATED_WI_ROOT is not a directory") raise RuntimeError("D4 precondition: GATED_WI_ROOT is not a directory")
is_worktree = subprocess.check_output(
["git", "-C", str(gated_root), "rev-parse", "--is-inside-work-tree"],
text=True,
).strip()
head = subprocess.check_output( head = subprocess.check_output(
["git", "-C", str(gated_root), "rev-parse", "HEAD"], text=True ["git", "-C", str(gated_root), "rev-parse", "HEAD"], text=True
).strip() ).strip()
except (OSError, subprocess.CalledProcessError) as error: except (OSError, subprocess.CalledProcessError) as error:
raise RuntimeError("D4 precondition: GATED_WI_ROOT is not a git worktree") from error raise RuntimeError("D4 precondition: GATED_WI_ROOT is not a git worktree") from error
if is_worktree != "true":
raise RuntimeError("D4 precondition: GATED_WI_ROOT is not a git worktree")
if head != GATED_WI_HEAD: if head != GATED_WI_HEAD:
raise RuntimeError(f"D4 precondition: gated WI head mismatch: {head}") raise RuntimeError(f"D4 precondition: gated WI head mismatch: {head}")
return gated_root return gated_root
@@ -353,23 +359,21 @@ def gated_launcher_precondition(
except UnicodeDecodeError as error: except UnicodeDecodeError as error:
raise RuntimeError("D4 precondition: pinned launcher is not UTF-8") from error raise RuntimeError("D4 precondition: pinned launcher is not UTF-8") from error
register = launcher_source.find('"action": "register_anchor"') # The exact launcher digest above is the trust anchor. These marker checks
initialize = launcher_source.find("initialize_runtime_generation(generation_file, generation)") # are diagnostic belt-and-suspenders only, never a substitute for the pin.
execute = launcher_source.find("execute(command[0], command, environment)") behavior_markers = (
fixture_socket = launcher_source.find('source_environment["MOSAIC_LEASE_BROKER_SOCKET"]') '"action": "register_anchor"',
fixture_state = launcher_source.find('socket_path.parent / f"generation-{session_id}.state"') "initialize_runtime_generation(generation_file, generation)",
exported_state = launcher_source.find('environment["MOSAIC_LEASE_GENERATION_FILE"]') "execute(command[0], command, environment)",
generation_api = ( 'source_environment["MOSAIC_LEASE_BROKER_SOCKET"]',
'socket_path.parent / f"generation-{session_id}.state"',
'environment["MOSAIC_LEASE_GENERATION_FILE"]',
)
if not all(marker in launcher_source for marker in behavior_markers) or not (
"def read_runtime_generation" in generation_source "def read_runtime_generation" in generation_source
and "def bump_runtime_generation" in generation_source and "def bump_runtime_generation" in generation_source
)
if (
min(register, initialize, execute, fixture_socket, fixture_state, exported_state) < 0
or register >= initialize
or initialize >= execute
or not generation_api
): ):
raise RuntimeError("D4 precondition: launcher is not file-generation fixture-bound") raise RuntimeError("D4 precondition: pinned launcher lacks file-generation markers")
fixture_launcher_dir = root / "pinned-lease-broker" fixture_launcher_dir = root / "pinned-lease-broker"
fixture_launcher_dir.mkdir(mode=0o700) fixture_launcher_dir.mkdir(mode=0o700)