From 4d822cddfca47d5dc07a9b72b76e2521beb1aa7b Mon Sep 17 00:00:00 2001 From: ProtocolWarden <32967198+ProtocolWarden@users.noreply.github.com> Date: Tue, 14 Jul 2026 12:11:53 -0400 Subject: [PATCH 1/2] feat(loop): C2 launch-time committed-truth check (keyless council anchor) When the loop config has no signed reference, compare the live pseudo_operator section against the committed copy on origin/main and run the committed copy on drift (drift_unsigned), the keyless analogue of Track C restore-by-consumption. --require-committed makes it a fail-closed gate (refuses on drift and when it cannot confirm committed truth). Signed reference present -> Track C wins. Council spec COUNCIL_VERDICT.md Phase 1 (C2). Co-Authored-By: Claude Opus 4.8 (1M context) --- .console/log.md | 19 ++ src/context_lifecycle/cli/loop.py | 21 +- .../pseudo_operator/committed.py | 133 +++++++++++ .../pseudo_operator/config.py | 89 ++++++-- tests/test_committed.py | 212 ++++++++++++++++++ 5 files changed, 454 insertions(+), 20 deletions(-) create mode 100644 src/context_lifecycle/pseudo_operator/committed.py create mode 100644 tests/test_committed.py diff --git a/.console/log.md b/.console/log.md index db7de19..175ea6e 100644 --- a/.console/log.md +++ b/.console/log.md @@ -1,4 +1,23 @@ # Log +## 2026-07-14 — feat: C2 launch-time committed-truth check (keyless council anchor) + +Council spec (COUNCIL_VERDICT.md) Phase 1. When the loop config has NO signed +reference, `cl loop run` now compares the live `pseudo_operator:` section +against the committed copy on `origin/main` (`git show origin/main:` +after a bounded `git fetch`) — the keyless analogue of Track C's +restore-by-consumption. Match → run live (it IS committed truth); drift → run +the COMMITTED copy (`signed_status="drift_unsigned"`, no YAML rewrite) and flag +loudly; unreachable origin → skip with a loud note (degrade-never-halt). New +launcher flag `--require-committed` (parity with `--require-signed`, lives +outside agent-reachable config) turns it into a fail-closed gate: refuses to +start on drift AND when it cannot confirm committed truth (offline) — a +fail-open skip there would let anyone bypass the gate by cutting network. A +signed reference present ⇒ Track C wins unchanged; C2 is never consulted. New +module `pseudo_operator/committed.py` reuses signing's `_load_section` + +`canonical_bytes` so the committed scope is byte-identical to what Track C +signs. 28 tests (test_committed.py + signing/loop regression); T1 satisfied by +a `CommittedResult` isinstance assertion. + ## 2026-07-07 — fix: surface hook stderr in the loop log on success Hooks log their actions via logging → stderr (e.g. OC self-update's diff --git a/src/context_lifecycle/cli/loop.py b/src/context_lifecycle/cli/loop.py index 9b7a873..360c49b 100644 --- a/src/context_lifecycle/cli/loop.py +++ b/src/context_lifecycle/cli/loop.py @@ -57,16 +57,29 @@ def run_cmd( help="Refuse to start unless the config is anchored (signed reference " "verifies). Launchers pass this once the operator has signed.", ), + require_committed: bool = typer.Option( + False, + "--require-committed", + help="Keyless (C2) parity with --require-signed: when the config is NOT " + "signed, refuse to start if the live section drifts from origin/main " + "instead of running the committed copy. Ignored when a signed reference " + "is present (Track C wins).", + ), ) -> NoReturn: """Run the loop (foreground). Refuses to start on an invalid config. Trust anchor (Track C): with a signed reference present, the engine runs the VERIFIED section — on drift it consumes the reference (not the live section) and flags the divergence; a bad signature always refuses. + + Committed-truth check (C2): with NO signed reference, the engine compares + the live section against the committed copy on ``origin/main`` — on drift it + runs the committed copy (``drift_unsigned``) unless ``--require-committed`` + refuses; offline/unreachable launches degrade to unsigned with a loud note. """ try: cfg, signed_status, detail = load_verified_config( - config, require_signed=require_signed + config, require_signed=require_signed, require_committed=require_committed ) except (ValueError, OSError) as exc: typer.echo(f"config error: {exc}", err=True) @@ -77,6 +90,12 @@ def run_cmd( f"reference (restore-by-consumption): {detail}", err=True, ) + elif signed_status == "drift_unsigned": + typer.echo( + f"[TRUST] live config DIVERGES from origin/main — running the COMMITTED " + f"copy (keyless committed-truth check, restore-by-consumption): {detail}", + err=True, + ) elif signed_status == "unsigned": typer.echo( f"[TRUST] loop config is NOT anchored ({detail}) — running unsigned; " diff --git a/src/context_lifecycle/pseudo_operator/committed.py b/src/context_lifecycle/pseudo_operator/committed.py new file mode 100644 index 0000000..bf999db --- /dev/null +++ b/src/context_lifecycle/pseudo_operator/committed.py @@ -0,0 +1,133 @@ +# SPDX-License-Identifier: AGPL-3.0-or-later +# Copyright (C) 2026 ProtocolWarden +"""Committed-truth check (C2) — the keyless analogue of Track C's anchor. + +When no signed reference exists the loop has no cryptographic anchor, but the +committed history is still an authority the fleet cannot rewrite from inside a +session: ``origin/main`` is protected by branch rules + CODEOWNERS, so the +*committed* copy of the config is a trustworthy "restore" target even without a +signature. C2 makes "restore vs. change" decidable the same way Track C does, +keylessly: + +- read the committed ``pseudo_operator:`` section from ``origin/main`` + (``git show origin/main:`` after a bounded ``git fetch``); +- drift (live section != committed section) → the caller runs the COMMITTED + copy and flags the divergence — restore-by-consumption, no YAML rewriting; +- match → the live section IS the committed truth, run it as normal unsigned. + +Degrade-never-halt: an ordinary launch that cannot reach ``origin`` (offline, +fetch failure, config absent on origin, git unavailable) SKIPS the check with a +loud note rather than refusing — the keyless check is a guardrail, not a gate. +The gate form is opt-in via the launcher's ``--require-committed`` flag, which +lives outside agent-reachable config exactly like ``--require-signed``. + +Signed reference present ⇒ Track C wins; C2 is not consulted (see +``config.load_verified_config``). +""" + +from __future__ import annotations + +import subprocess +from dataclasses import dataclass +from pathlib import Path + +import yaml + +from .signing import _load_section, canonical_bytes + +COMMITTED_REF = "origin/main" +# Bounded so an offline/slow launch degrades quickly instead of hanging. +FETCH_TIMEOUT_SECONDS = 20 +GIT_TIMEOUT_SECONDS = 15 + + +@dataclass(frozen=True) +class CommittedResult: + """Outcome of comparing the live section against ``origin/main``.""" + + status: str # "match" | "drift" | "skip" + detail: str + committed: dict | None # the committed section (match/drift), else None + + +def _run_git(args: list[str], cwd: Path, timeout: int) -> subprocess.CompletedProcess[str]: + return subprocess.run( # noqa: S603 — fixed argv, no shell + ["git", *args], + cwd=cwd, + capture_output=True, + text=True, + timeout=timeout, + ) + + +def _section_from_yaml_text(text: str) -> dict: + """Parse the committed YAML bytes to the same section Track C anchors.""" + data = yaml.safe_load(text) + if not isinstance(data, dict): + raise ValueError("committed config is not a YAML mapping") + section = data.get("pseudo_operator", data) + if not isinstance(section, dict): + raise ValueError("committed config has no pseudo_operator section") + return section + + +def verify_committed(config_file: Path) -> CommittedResult: + """Compare the live section against the committed copy on ``origin/main``. + + Never raises on reachability/parse failures — those return ``skip`` (the + caller runs the live section unsigned, loudly). Only genuine drift returns + ``drift`` with the committed section for the caller to consume. + """ + cwd = config_file.resolve().parent + try: + fetch = _run_git(["fetch", "--quiet", "origin", "main"], cwd, FETCH_TIMEOUT_SECONDS) + except (OSError, subprocess.SubprocessError) as exc: + return CommittedResult("skip", f"origin fetch failed: {exc}", None) + if fetch.returncode != 0: + return CommittedResult("skip", f"origin fetch failed: {fetch.stderr.strip()}", None) + + try: + top = _run_git(["rev-parse", "--show-toplevel"], cwd, GIT_TIMEOUT_SECONDS) + except (OSError, subprocess.SubprocessError) as exc: + return CommittedResult("skip", f"git rev-parse failed: {exc}", None) + if top.returncode != 0: + return CommittedResult("skip", f"not a git repo: {top.stderr.strip()}", None) + try: + relpath = config_file.resolve().relative_to(Path(top.stdout.strip())) + except ValueError as exc: + return CommittedResult("skip", f"config file outside git repo: {exc}", None) + + try: + show = _run_git( + ["show", f"{COMMITTED_REF}:{relpath.as_posix()}"], cwd, GIT_TIMEOUT_SECONDS + ) + except (OSError, subprocess.SubprocessError) as exc: + return CommittedResult("skip", f"git show failed: {exc}", None) + if show.returncode != 0: + return CommittedResult( + "skip", f"config absent on {COMMITTED_REF}: {show.stderr.strip()}", None + ) + + try: + committed = _section_from_yaml_text(show.stdout) + except ValueError as exc: + return CommittedResult("skip", f"committed config unreadable: {exc}", None) + try: + live = _load_section(config_file) + except ValueError as exc: + return CommittedResult("skip", f"live config unreadable: {exc}", None) + + if canonical_bytes(live) != canonical_bytes(committed): + return CommittedResult( + "drift", + f"live pseudo_operator section diverges from {COMMITTED_REF}", + committed, + ) + return CommittedResult("match", f"live section matches {COMMITTED_REF}", committed) + + +__all__ = [ + "COMMITTED_REF", + "CommittedResult", + "verify_committed", +] diff --git a/src/context_lifecycle/pseudo_operator/config.py b/src/context_lifecycle/pseudo_operator/config.py index 3de0a82..09916a8 100644 --- a/src/context_lifecycle/pseudo_operator/config.py +++ b/src/context_lifecycle/pseudo_operator/config.py @@ -235,43 +235,94 @@ def load_pseudo_operator_config(config_file: Path) -> PseudoOperatorConfig: return PseudoOperatorConfig.model_validate(section) +def _with_repo_root(section: dict, config_file: Path) -> dict: + """Inject the default ``repo_root`` into a section loaded from anchor bytes.""" + section = dict(section) + if "repo_root" not in section: + parent = config_file.resolve().parent + section["repo_root"] = str( + parent.parent if parent.name in {".context", ".console"} else parent + ) + return section + + def load_verified_config( - config_file: Path, *, require_signed: bool = False + config_file: Path, *, require_signed: bool = False, require_committed: bool = False ) -> tuple[PseudoOperatorConfig, str, str]: - """Load the loop config through the trust anchor (Track C). + """Load the loop config through the trust anchor (Track C) or, when it is + not anchored, the keyless committed-truth check (C2). Returns ``(config, signed_status, detail)`` where signed_status is - ok | drift | unsigned. Enforcement: + ok | drift | unsigned | drift_unsigned. Enforcement: - - ``ok``: run the live section (it IS the reference). + - ``ok``: run the live section (it IS the signed reference). - ``drift``: run the SIGNED REFERENCE, not the live section — restore-by- consumption; the caller logs the divergence loudly. - ``bad_signature``: raise — a present-but-unverifiable reference means the anchor is compromised or corrupt; never fall back to the live section. - - ``unsigned``: raise when ``require_signed``; otherwise run the live - section (pre-anchoring mode; the caller warns). + - no signed reference (``unsigned``): raise when ``require_signed``; + otherwise fall through to the committed-truth check (C2, below). + + C2 — when no signed reference exists (Track C wins unchanged whenever one is + present, so this runs only in the unsigned case): + + - committed match: run the live section (it IS the committed truth) as + normal unsigned. + - ``drift_unsigned``: the live section diverges from ``origin/main`` — run + the COMMITTED copy (restore-by-consumption, no YAML rewriting) and flag it + loudly, OR raise when ``require_committed``. + - skip (offline / fetch failure / config absent on origin): run the live + section unsigned with a loud note — degrade-never-halt for ordinary + launches; but raise when ``require_committed`` — the gate cannot confirm + the live section matches committed truth, and a fail-open skip would let + anyone bypass ``--require-committed`` by cutting network access (parity + with ``--require-signed``, which likewise refuses when it cannot confirm). """ + from .committed import verify_committed from .signing import verify_reference result = verify_reference(config_file) if result.status == "bad_signature": raise ValueError(f"signed loop config FAILED verification: {result.detail}") - if result.status == "unsigned": - if require_signed: + if result.status == "drift": + section = _with_repo_root(result.reference or {}, config_file) + return PseudoOperatorConfig.model_validate(section), "drift", result.detail + if result.status == "ok": + return load_pseudo_operator_config(config_file), "ok", result.detail + + # No signed reference — Track C does not apply. + if require_signed: + raise ValueError( + f"--require-signed set but the loop config is not anchored ({result.detail}); " + "sign it with `cl loop sign-config`" + ) + + # C2: keyless committed-truth check against origin/main. + committed = verify_committed(config_file) + if committed.status == "drift": + if require_committed: raise ValueError( - f"--require-signed set but the loop config is not anchored ({result.detail}); " - "sign it with `cl loop sign-config`" + "--require-committed set but the live loop config DIVERGES from " + f"origin/main ({committed.detail}); commit the change on origin/main " + "or relaunch without --require-committed" ) - return load_pseudo_operator_config(config_file), "unsigned", result.detail - if result.status == "drift": - section = dict(result.reference or {}) - if "repo_root" not in section: - parent = config_file.resolve().parent - section["repo_root"] = str( - parent.parent if parent.name in {".context", ".console"} else parent + section = _with_repo_root(committed.committed or {}, config_file) + return PseudoOperatorConfig.model_validate(section), "drift_unsigned", committed.detail + if committed.status == "skip": + if require_committed: + raise ValueError( + "--require-committed set but the live loop config could not be " + f"verified against origin/main ({committed.detail}); the gate " + "cannot confirm committed truth — resolve connectivity or " + "relaunch without --require-committed" ) - return PseudoOperatorConfig.model_validate(section), "drift", result.detail - return load_pseudo_operator_config(config_file), "ok", result.detail + return ( + load_pseudo_operator_config(config_file), + "unsigned", + f"{result.detail}; committed-truth check skipped: {committed.detail}", + ) + # match — live section is the committed truth. + return load_pseudo_operator_config(config_file), "unsigned", committed.detail __all__ = [ diff --git a/tests/test_committed.py b/tests/test_committed.py new file mode 100644 index 0000000..08dfd48 --- /dev/null +++ b/tests/test_committed.py @@ -0,0 +1,212 @@ +# SPDX-License-Identifier: AGPL-3.0-or-later +# Copyright (C) 2026 ProtocolWarden +"""Committed-truth check (C2) — the keyless analogue of Track C's anchor. + +No signed reference: the live section is compared against the committed copy on +origin/main. Match runs live; drift runs the COMMITTED copy (drift_unsigned) or +refuses under --require-committed; an unreachable origin degrades to unsigned +with a loud note (but --require-committed refuses — it cannot confirm truth). +A signed reference present ⇒ Track C wins, C2 is not consulted. +""" + +from __future__ import annotations + +import subprocess +from pathlib import Path + +import pytest +from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey +from cryptography.hazmat.primitives.serialization import ( + Encoding, + NoEncryption, + PrivateFormat, +) +from typer.testing import CliRunner + +from context_lifecycle.cli import loop as loop_cli +from context_lifecycle.pseudo_operator import signing +from context_lifecycle.pseudo_operator.committed import CommittedResult, verify_committed +from context_lifecycle.pseudo_operator.config import load_verified_config +from context_lifecycle.pseudo_operator.signing import sign_reference + +runner = CliRunner() + +_CONFIG = ( + "pseudo_operator:\n" + " loop_name: t\n" + " session_prompt_file: prompt.txt\n" + " max_iterations: 7\n" +) + + +def _git(cwd: Path, *args: str) -> None: + subprocess.run( + ["git", *args], + cwd=cwd, + check=True, + capture_output=True, + text=True, + ) + + +def _repo(tmp_path: Path) -> Path: + """A git work-repo whose origin/main holds a committed loop config. + + Returns the path to the working config file. ``repo_root`` is injected by + the loader (defaults to the config file's parent), so it is not committed. + """ + work = tmp_path / "work" + origin = tmp_path / "origin.git" + work.mkdir() + _git(work, "init", "-b", "main") + _git(work, "config", "user.email", "t@example.com") + _git(work, "config", "user.name", "t") + (work / "prompt.txt").write_text("x", encoding="utf-8") + cfg = work / "config.yaml" + cfg.write_text(_CONFIG, encoding="utf-8") + _git(work, "add", "-A") + _git(work, "commit", "-m", "init") + origin.mkdir() + _git(origin, "init", "--bare", "-b", "main") + _git(work, "remote", "add", "origin", str(origin)) + _git(work, "push", "-u", "origin", "main") + return cfg + + +def _drift(cfg: Path) -> None: + """Loosen a guardrail in the working config without committing it.""" + cfg.write_text( + cfg.read_text(encoding="utf-8").replace("max_iterations: 7", "max_iterations: 999999"), + encoding="utf-8", + ) + + +# ── verify_committed ────────────────────────────────────────────────────────── +def test_match_when_live_equals_committed(tmp_path: Path): + cfg = _repo(tmp_path) + result = verify_committed(cfg) + assert isinstance(result, CommittedResult) + assert result.status == "match" + assert result.committed is not None + + +def test_drift_detected_and_committed_section_returned(tmp_path: Path): + cfg = _repo(tmp_path) + _drift(cfg) + result = verify_committed(cfg) + assert result.status == "drift" + assert result.committed["max_iterations"] == 7 + + +def test_skip_when_origin_unreachable(tmp_path: Path): + cfg = _repo(tmp_path) + # Repoint origin at a path that does not exist → fetch fails → skip. + _git(cfg.parent, "remote", "set-url", "origin", str(tmp_path / "gone.git")) + result = verify_committed(cfg) + assert result.status == "skip" + assert "fetch failed" in result.detail + + +def test_skip_when_config_absent_on_origin(tmp_path: Path): + cfg = _repo(tmp_path) + other = cfg.parent / "not_committed.yaml" + other.write_text(_CONFIG, encoding="utf-8") + result = verify_committed(other) + assert result.status == "skip" + assert "absent on origin/main" in result.detail + + +# ── load_verified_config (C2 wiring) ───────────────────────────────────────── +def test_no_signed_plus_match_runs_live_unsigned(tmp_path: Path): + cfg = _repo(tmp_path) + loaded, status, _ = load_verified_config(cfg) + assert status == "unsigned" + assert loaded.max_iterations == 7 + + +def test_no_signed_plus_drift_runs_committed_and_flags(tmp_path: Path): + cfg = _repo(tmp_path) + _drift(cfg) + loaded, status, _ = load_verified_config(cfg) + assert status == "drift_unsigned" + assert loaded.max_iterations == 7 # the COMMITTED cap, not the tampered one + + +def test_no_signed_plus_drift_plus_require_committed_refuses(tmp_path: Path): + cfg = _repo(tmp_path) + _drift(cfg) + with pytest.raises(ValueError, match="require-committed"): + load_verified_config(cfg, require_committed=True) + + +def test_offline_skips_check_and_runs_live_loud(tmp_path: Path): + cfg = _repo(tmp_path) + _drift(cfg) # would be drift, but origin is unreachable → skip, run live + _git(cfg.parent, "remote", "set-url", "origin", str(tmp_path / "gone.git")) + loaded, status, detail = load_verified_config(cfg) + assert status == "unsigned" + assert loaded.max_iterations == 999999 # ran the live section, check skipped + assert "committed-truth check skipped" in detail + + +def test_require_committed_refuses_when_offline(tmp_path: Path): + cfg = _repo(tmp_path) + _drift(cfg) + _git(cfg.parent, "remote", "set-url", "origin", str(tmp_path / "gone.git")) + # Fail-closed gate: an unreachable origin means the gate cannot confirm the + # live section matches committed truth, so --require-committed must refuse + # (parity with --require-signed). A fail-open skip here would let anyone + # bypass the gate by cutting network access. + with pytest.raises(ValueError, match="could not be verified against origin/main"): + load_verified_config(cfg, require_committed=True) + + +def test_signed_reference_present_bypasses_c2(tmp_path: Path): + cfg = _repo(tmp_path) + key = Ed25519PrivateKey.generate() + priv = cfg.parent / "operator_priv.pem" + priv.write_bytes(key.private_bytes(Encoding.PEM, PrivateFormat.PKCS8, NoEncryption())) + (cfg.parent / signing.PUBKEY_NAME).write_text( + key.public_key().public_bytes_raw().hex() + "\n", encoding="utf-8" + ) + sign_reference(cfg, priv) + _drift(cfg) + # Track C wins: status is signed "drift" (not the keyless "drift_unsigned"), + # even though the working tree also diverges from origin/main. + loaded, status, _ = load_verified_config(cfg, require_committed=True) + assert status == "drift" + assert loaded.max_iterations == 7 + + +# ── CLI surface ─────────────────────────────────────────────────────────────── +def test_cli_run_drift_unsigned_marks_state(tmp_path: Path, monkeypatch): + cfg = _repo(tmp_path) + _drift(cfg) + + captured: dict = {} + + class FakeEngine: + def __init__(self, c): + captured["cfg"] = c + self.signed_status = "unsigned" + + def run(self): + captured["signed_status"] = self.signed_status + return 0 + + monkeypatch.setattr(loop_cli, "PseudoOperatorEngine", FakeEngine) + res = runner.invoke(loop_cli.app, ["run", "--config", str(cfg)]) + assert res.exit_code == 0 + assert "DIVERGES from origin/main" in res.output + assert captured["cfg"].max_iterations == 7 + assert captured["signed_status"] == "drift_unsigned" + + +def test_cli_run_require_committed_refuses_on_drift(tmp_path: Path): + cfg = _repo(tmp_path) + _drift(cfg) + res = runner.invoke( + loop_cli.app, ["run", "--config", str(cfg), "--require-committed"] + ) + assert res.exit_code == 2 + assert "require-committed" in res.output From ab69399ba6adff19eef74da1591d848b42803cb2 Mon Sep 17 00:00:00 2001 From: ProtocolWarden <32967198+ProtocolWarden@users.noreply.github.com> Date: Tue, 14 Jul 2026 16:37:36 -0400 Subject: [PATCH 2/2] refactor(loop): C2 shares one public section-extractor with signing committed.py imported the private signing._load_section and duplicated its section-extraction. Promote to public signing.section_of(data) + load_section(file) so the signed and committed anchors resolve the SAME scope with no duplication, and catch yaml.YAMLError in verify_committed (its never-raises contract missed malformed committed YAML). Resolves the reviewer's code_quality concern on this PR. Co-Authored-By: Claude Opus 4.8 (1M context) --- .console/log.md | 13 ++++++++++ .../pseudo_operator/committed.py | 19 +++----------- .../pseudo_operator/signing.py | 26 ++++++++++++++----- tests/test_signing.py | 18 +++++++++++++ 4 files changed, 54 insertions(+), 22 deletions(-) diff --git a/.console/log.md b/.console/log.md index 175ea6e..df3acfb 100644 --- a/.console/log.md +++ b/.console/log.md @@ -1,4 +1,17 @@ # Log +## 2026-07-14 — refactor: C2 shares one public section-extractor with signing (reviewer code_quality) + +Addressed the C2 reviewer's code_quality concern + removed a real smell: +committed.py imported the PRIVATE `signing._load_section` AND duplicated its +section-extraction as `_section_from_yaml_text`. Promoted to a single public +`signing.section_of(data)` (the one source of truth for what the anchor covers) ++ `signing.load_section(file)`; committed.py now calls both, so the signed and +committed checks provably resolve the SAME scope with no duplication. Also +caught a latent bug: the committed-YAML parse could raise `yaml.YAMLError`, +which `verify_committed` (contract: never raises) did not catch — now returns +`skip` on malformed committed YAML too. 408 pass; a same-scope test pins that +`load_section` and `section_of` resolve identically (T1). + ## 2026-07-14 — feat: C2 launch-time committed-truth check (keyless council anchor) Council spec (COUNCIL_VERDICT.md) Phase 1. When the loop config has NO signed diff --git a/src/context_lifecycle/pseudo_operator/committed.py b/src/context_lifecycle/pseudo_operator/committed.py index bf999db..5da9e39 100644 --- a/src/context_lifecycle/pseudo_operator/committed.py +++ b/src/context_lifecycle/pseudo_operator/committed.py @@ -33,7 +33,7 @@ import yaml -from .signing import _load_section, canonical_bytes +from .signing import canonical_bytes, load_section, section_of COMMITTED_REF = "origin/main" # Bounded so an offline/slow launch degrades quickly instead of hanging. @@ -60,17 +60,6 @@ def _run_git(args: list[str], cwd: Path, timeout: int) -> subprocess.CompletedPr ) -def _section_from_yaml_text(text: str) -> dict: - """Parse the committed YAML bytes to the same section Track C anchors.""" - data = yaml.safe_load(text) - if not isinstance(data, dict): - raise ValueError("committed config is not a YAML mapping") - section = data.get("pseudo_operator", data) - if not isinstance(section, dict): - raise ValueError("committed config has no pseudo_operator section") - return section - - def verify_committed(config_file: Path) -> CommittedResult: """Compare the live section against the committed copy on ``origin/main``. @@ -109,11 +98,11 @@ def verify_committed(config_file: Path) -> CommittedResult: ) try: - committed = _section_from_yaml_text(show.stdout) - except ValueError as exc: + committed = section_of(yaml.safe_load(show.stdout), source=f"{COMMITTED_REF}:committed") + except (ValueError, yaml.YAMLError) as exc: return CommittedResult("skip", f"committed config unreadable: {exc}", None) try: - live = _load_section(config_file) + live = load_section(config_file) except ValueError as exc: return CommittedResult("skip", f"live config unreadable: {exc}", None) diff --git a/src/context_lifecycle/pseudo_operator/signing.py b/src/context_lifecycle/pseudo_operator/signing.py index 24a876f..1322505 100644 --- a/src/context_lifecycle/pseudo_operator/signing.py +++ b/src/context_lifecycle/pseudo_operator/signing.py @@ -55,18 +55,28 @@ def _paths(config_file: Path) -> tuple[Path, Path, Path]: return d / REFERENCE_NAME, d / SIGNATURE_NAME, d / PUBKEY_NAME -def _load_section(config_file: Path) -> dict: - from context_lifecycle.io.yaml_io import load_yaml_safe +def section_of(data: object, *, source: str = "config") -> dict: + """Extract the anchored ``pseudo_operator`` section from parsed YAML data. - data = load_yaml_safe(config_file, default=None) + The single source of truth for what a signed/committed anchor covers, so the + file path (:func:`load_section`) and any already-parsed source (e.g. the + committed bytes from ``git show``) resolve the SAME scope. + """ if not isinstance(data, dict): - raise ValueError(f"{config_file}: not a YAML mapping") + raise ValueError(f"{source}: not a YAML mapping") section = data.get("pseudo_operator", data) if not isinstance(section, dict): - raise ValueError(f"{config_file}: no pseudo_operator section") + raise ValueError(f"{source}: no pseudo_operator section") return section +def load_section(config_file: Path) -> dict: + """Load and extract the anchored section from a YAML config file.""" + from context_lifecycle.io.yaml_io import load_yaml_safe + + return section_of(load_yaml_safe(config_file, default=None), source=str(config_file)) + + def sign_reference(config_file: Path, private_key_path: Path) -> Path: """Operator tool: snapshot + sign the live ``pseudo_operator:`` section. @@ -74,7 +84,7 @@ def sign_reference(config_file: Path, private_key_path: Path) -> Path: keygen). Writes the canonical snapshot + signature beside the config file; commit both. Never called by any fleet path. """ - section = _load_section(config_file) + section = load_section(config_file) payload = canonical_bytes(section) data = private_key_path.read_bytes() key = load_pem_private_key(data, password=None) @@ -129,7 +139,7 @@ def verify_reference(config_file: Path) -> VerifyResult: except ValueError as exc: return VerifyResult("bad_signature", f"reference is not valid JSON: {exc}", None) try: - live = _load_section(config_file) + live = load_section(config_file) except ValueError as exc: return VerifyResult("drift", f"live config unreadable: {exc}", reference) if canonical_bytes(live) != canonical_bytes(reference): @@ -147,6 +157,8 @@ def verify_reference(config_file: Path) -> VerifyResult: "SIGNATURE_NAME", "VerifyResult", "canonical_bytes", + "load_section", + "section_of", "sign_reference", "verify_reference", ] diff --git a/tests/test_signing.py b/tests/test_signing.py index 040600c..55a181d 100644 --- a/tests/test_signing.py +++ b/tests/test_signing.py @@ -70,6 +70,24 @@ def _repo(tmp_path: Path) -> tuple[Path, Path]: return cfg, priv +def test_section_extractors_resolve_same_scope(tmp_path: Path): + """load_section (from file) and section_of (from parsed data) must resolve + the SAME anchored section — the single source of truth the signed (Track C) + and committed (C2) checks both compare, so their scopes can never diverge.""" + cfg, _ = _repo(tmp_path) + import yaml + + from_file = signing.load_section(cfg) + from_data = signing.section_of(yaml.safe_load(cfg.read_text(encoding="utf-8"))) + assert from_file == from_data + assert from_file["max_iterations"] == 7 + # A mapping without a pseudo_operator key falls back to the whole mapping; + # a non-mapping raises (the never-anchor-junk guard). + assert signing.section_of({"loop_name": "x"}) == {"loop_name": "x"} + with pytest.raises(ValueError, match="not a YAML mapping"): + signing.section_of(["not", "a", "mapping"]) + + def test_unsigned_when_no_reference(tmp_path: Path): cfg, _ = _repo(tmp_path) assert verify_reference(cfg).status == "unsigned"