#!/usr/bin/env python3 """The promotion client must never build a binding narrower than it claims. RED-first against a real defect: ``build_construction`` skipped any normative source it could not read (``except OSError: continue``) and promoted whatever remained. That is not a degraded binding, it is a forged smaller one — the broker recomputes ``h_source`` / ``h_payload`` from the fragments it is *sent* (``daemon.py:602-616``), so an omitted fragment is internally consistent and ``PAYLOAD_BINDING_MISMATCH`` cannot fire. Measured before the fix: with only ``USER.md`` readable (964 bytes on the live host), the client produced a one-fragment construction with ``promotion=True``. The classification under test mirrors the framework's own file ownership, and must keep mirroring it: * framework-owned, reconciled every upgrade (``install.sh`` FRAMEWORK_OWNED / ``config/file-adapter.ts`` FRAMEWORK_OWNED_FILES) plus the per-runtime contract — absence is a broken deployment, so it is REFUSED; * ``SOUL.md`` / ``USER.md`` — install.sh deliberately does not seed them ("generated by `mosaic init`"), so absence is legitimate and ALLOWED. Unreadable is treated separately from absent for *every* source, optional ones included: a file that will not open is not a file that was never configured, and collapsing the two is what let a permission change quietly shrink the law. """ from __future__ import annotations import contextlib import io import os import sys import tempfile import unittest from pathlib import Path TOOLS = Path(__file__).parents[2] / "framework/tools/lease-broker" sys.path.insert(0, str(TOOLS)) import lease_promote # noqa: E402 RUNTIME = "pi" RUNTIME_CONTRACT = f"runtime/{RUNTIME}/RUNTIME.md" ALL_SOURCES = (*lease_promote.FRAGMENT_SOURCES, RUNTIME_CONTRACT) REQUIRED = frozenset(lease_promote.REQUIRED_SOURCES) | {RUNTIME_CONTRACT} # Derived, never listed: a hand-kept second copy is exactly the drift this file # exists to catch. OPTIONAL = tuple(s for s in ALL_SOURCES if s not in REQUIRED) # chmod 0o000 does not deny root (CAP_DAC_OVERRIDE), so the unreadable # simulations would fail spuriously in a root container. runs_unprivileged = unittest.skipIf( os.geteuid() == 0, "chmod 0o000 cannot make a file unreadable to root" ) class PromotionBindingTest(unittest.TestCase): def setUp(self) -> None: self._previous_home = os.environ.get("MOSAIC_HOME") self._temporary = tempfile.TemporaryDirectory() self.root = Path(self._temporary.name) for source_id in ALL_SOURCES: path = self.root / source_id path.parent.mkdir(parents=True, exist_ok=True) path.write_bytes(f"# {source_id}\nnormative bytes\n".encode()) os.environ["MOSAIC_HOME"] = str(self.root) def tearDown(self) -> None: for path in self.root.rglob("*"): if path.is_file(): path.chmod(0o644) self._temporary.cleanup() if self._previous_home is None: os.environ.pop("MOSAIC_HOME", None) else: os.environ["MOSAIC_HOME"] = self._previous_home def reset_home(self) -> None: """Discard the current home and seed a fresh complete one. Each subTest mutates the tree destructively, so it needs a clean start — and the old one must be released, not orphaned. """ self.tearDown() self.setUp() def build(self): return lease_promote.build_construction(RUNTIME) def source_ids(self) -> list[str]: construction, _ = self.build() return [f["source_id"] for f in construction["fragments"]] # --- the binding is complete when the deployment is complete ------------- def test_complete_deployment_binds_every_source(self) -> None: construction, result = self.build() self.assertEqual([f["source_id"] for f in construction["fragments"]], list(ALL_SOURCES)) self.assertTrue(result.promotion) # --- absence: refused for framework-owned, allowed for operator-owned ---- def test_absent_required_source_is_refused(self) -> None: for source_id in sorted(REQUIRED): with self.subTest(source=source_id): self.reset_home() (self.root / source_id).unlink() with self.assertRaises(lease_promote.IncompleteBinding) as caught: self.build() self.assertIn(source_id, str(caught.exception)) def test_absent_operator_source_still_binds_the_rest(self) -> None: for source_id in OPTIONAL: with self.subTest(source=source_id): self.reset_home() (self.root / source_id).unlink() notice = io.StringIO() with contextlib.redirect_stderr(notice): bound = self.source_ids() self.assertNotIn(source_id, bound) for required in lease_promote.REQUIRED_SOURCES: self.assertIn(required, bound) # A silent omission is the original defect in miniature: the # narrower binding must announce itself. self.assertIn(source_id, notice.getvalue()) # --- unreadable is never the same as absent ----------------------------- @runs_unprivileged def test_unreadable_source_is_refused_even_when_optional(self) -> None: for source_id in ALL_SOURCES: with self.subTest(source=source_id): self.reset_home() (self.root / source_id).chmod(0o000) with self.assertRaises(lease_promote.IncompleteBinding) as caught: self.build() self.assertIn(source_id, str(caught.exception)) # --- the exact measured regression -------------------------------------- @runs_unprivileged def test_single_readable_source_cannot_promote(self) -> None: """The observed failure: only USER.md readable produced a valid binding.""" for source_id in ALL_SOURCES: if source_id != "USER.md": (self.root / source_id).chmod(0o000) with self.assertRaises(lease_promote.IncompleteBinding): self.build() @runs_unprivileged def test_no_source_readable_cannot_promote(self) -> None: for source_id in ALL_SOURCES: (self.root / source_id).chmod(0o000) with self.assertRaises(lease_promote.IncompleteBinding): self.build() # --- the classification must not drift from the framework's ------------- def test_required_set_excludes_only_the_unseeded_sources(self) -> None: """`install.sh` decides which files exist; this list must follow it. If a source moves between framework-owned and operator-generated upstream, this fails and forces the classification to be re-read rather than silently inherited. """ self.assertEqual( set(lease_promote.REQUIRED_SOURCES), {"CONSTITUTION.md", "AGENTS.md", "STANDARDS.md"}, "REQUIRED_SOURCES changed — re-read install.sh FRAMEWORK_OWNED and " "config/file-adapter.ts FRAMEWORK_OWNED_FILES before accepting it", ) self.assertTrue( set(lease_promote.REQUIRED_SOURCES) <= set(lease_promote.FRAGMENT_SOURCES), "a required source is not in the binding order", ) if __name__ == "__main__": unittest.main()