diff --git a/.github/workflows/p022-merge-gate.yml b/.github/workflows/p022-merge-gate.yml new file mode 100644 index 00000000..db536d73 --- /dev/null +++ b/.github/workflows/p022-merge-gate.yml @@ -0,0 +1,70 @@ +name: P-022 merge gate + +# The merge order for the #263 preregistration, as a check rather than an +# agreement. `mergegate.py` is run against the commit a merge would PRODUCE — +# for a pull_request event that is `github.sha`, the merge ref, not the branch +# head — because a PR number proves someone pressed a button and says nothing +# about what the merged tree contains. +# +# Two jobs, and the gate needs the controls: a gate whose own controls are +# failing must not be the thing that says a merge is safe. A failed controls job +# leaves the gate job unreported, which a required check reads as unsatisfied. +# +# This workflow is one half of the wiring. The other half is a ruleset requiring +# these two checks; without it the job reports and nothing is prevented. The +# names below are the contexts that ruleset must name: +# +# P-022 merge gate controls +# P-022 merge gate +# +# Least privilege as elsewhere in this repository: read-only, and every +# third-party `uses:` pinned to a commit SHA with its version in a comment. + +permissions: + contents: read + +on: + pull_request: + workflow_dispatch: + +jobs: + controls: + name: P-022 merge gate controls + runs-on: ubuntu-latest + steps: + - uses: actions/checkout@34e114876b0b11c390a56381ad16ebd13914f8d5 # v4 + - uses: actions/setup-python@a26af69be951a213d495a4c3e4e4022e16d87065 # v5 + with: + python-version: "3.11" + - name: the gate's own controls (fixtures ship the real tools and mutate them) + run: python tests/test_step7_mergegate.py + - name: the wiring's controls (applicability and co-change) + run: python tests/test_step7_mergegate_ci.py + + gate: + name: P-022 merge gate + needs: controls + runs-on: ubuntu-latest + steps: + # Full history: the co-change rule compares the gate's own files against + # the base commit, which a shallow clone would not carry. + - uses: actions/checkout@34e114876b0b11c390a56381ad16ebd13914f8d5 # v4 + with: + fetch-depth: 0 + - uses: actions/setup-python@a26af69be951a213d495a4c3e4e4022e16d87065 # v5 + with: + python-version: "3.11" + - name: the gate, against the commit this merge would produce + env: + # Read through the environment rather than interpolated into the + # script body, so nothing from the event can be read as shell. + MERGE_COMMIT: ${{ github.sha }} + BASE_COMMIT: ${{ github.event.pull_request.base.sha }} + run: | + set -euo pipefail + if [ -n "${BASE_COMMIT}" ]; then + python scripts/step7/mergegate_ci.py --repo . \ + --commit "${MERGE_COMMIT}" --base "${BASE_COMMIT}" + else + python scripts/step7/mergegate_ci.py --repo . --commit "${MERGE_COMMIT}" + fi diff --git a/scripts/step7/mergegate.py b/scripts/step7/mergegate.py new file mode 100644 index 00000000..2ef635f3 --- /dev/null +++ b/scripts/step7/mergegate.py @@ -0,0 +1,366 @@ +#!/usr/bin/env python3 +"""P-022 — the merge gate: does the world a freeze enters satisfy its references? + +The frozen T0 names things that live outside itself: an accepted harness +identity, the step-7 machinery that enforces the campaign join, the T0-0 +amendment that revoked automatic collection authority. On the branch where T0 +was frozen, some of those referents arrive through *other* pull requests. That +is fine while the merge order holds — and merge order held by agreement is one +stray click away from becoming an archaeological artifact, with a document +saying FROZEN in a tree where its referent does not exist yet. + +So the order is checked rather than promised, and it is checked against the +**target tree**, never against pull-request numbers. A PR number proves that +someone pressed a button; it proves nothing about what the merged tree contains. + +Run it on the commit a merge would produce: + + python scripts/step7/mergegate.py --repo . --commit + +Exit 0 only when every predicate holds. Any failure means the frozen contract +would become reachable from a tree that cannot satisfy it, and the merge is +refused however mergeable the forge believes it to be. + +The digest is not hard-coded here: it is read out of the frozen T0 and then +RECOMPUTED from the instrument sources in the target tree, by the frozen +formula, without importing the instrument. A gate that trusted a constant in +its own source would be checking itself. +""" + +from __future__ import annotations + +import argparse +import json +import re +import subprocess +import sys +import tempfile +from pathlib import Path + +import execbinding as eb # the single implementation of the harness-digest formula + +T0_PATH = "docs/notes/p022-263-t0-protocol-freeze.md" +STEP7_NOTE = "docs/notes/p022-263a-step7-environment-capture.md" +# Each artifact binds the digest at its own exact path. Named here so the check +# reads the binding rather than searching the file for a hopeful substring. +BINDING_ARTIFACTS = { + "docs/evidence/calibration/p022-263a-policy-freeze.json": + ("measurement_harness_digest",), + "docs/evidence/calibration/p022-263a-design-constants.json": + ("bound_measurement_harness_digest",), + "docs/evidence/calibration/p022-263a-training-preregistration.json": + ("bindings", "measurement_harness_digest"), +} +STEP7_TOOLS = ("scripts/step7/envcapture.py", "scripts/step7/hostqual.py", + "scripts/step7/execbinding.py") +DIGEST_RE = re.compile(r"measurement_harness_digest\s*\n?\s*([0-9a-f]{64})") + + +def blob(repo: Path, commit: str, path: str) -> bytes | None: + proc = subprocess.run(["git", "-C", str(repo), "cat-file", "blob", f"{commit}:{path}"], + capture_output=True, check=False) + return proc.stdout if proc.returncode == 0 else None + + +# The witness runs INSIDE the candidate tree's own modules, in a subprocess, so +# that importing them cannot bind to this gate's copies and a passing result +# cannot come from anything but the code being merged. +WITNESS_SOURCE = r""" +import json, sys, subprocess +from pathlib import Path +sys.path.insert(0, "scripts/step7") +import hostqual as hq +import execbinding as eb + +tmp = Path(sys.argv[1]) +POWER = {"platform": "windows", "plan_guid": hq.WIN_ACCEPTED_PLANS[0], + "processor_min_ac": 100, "processor_max_ac": 100, + "processor_min_dc": 100, "processor_max_dc": 100} +hq.power_snapshot = lambda: dict(POWER) +failures = [] + +def repo_with(name, files): + root = tmp / name + root.mkdir(parents=True, exist_ok=True) + run = lambda *a: subprocess.run(["git", "-C", str(root), *a], capture_output=True, check=True) + if not (root / ".git").exists(): + run("init", "-q"); run("config", "user.email", "w@x"); run("config", "user.name", "w") + for rel, text in files.items(): + f = root / rel; f.parent.mkdir(parents=True, exist_ok=True) + f.write_text(text, encoding="utf-8") + run("add", "-A"); run("commit", "-q", "-m", "w") + return root, subprocess.run(["git", "-C", str(root), "rev-parse", "HEAD"], + capture_output=True, text=True, check=True).stdout.strip() + +def w(name, doc): + f = tmp / name; f.write_text(json.dumps(doc), encoding="utf-8"); return f + +# --- W3/W4: the authority state must bite on both readers ------------------- +for label, status, flag, must_pass in (("FROZEN+false", "FROZEN.", "false", False), + ("NOT_FROZEN+true", "NOT_FROZEN.", "true", False), + ("NOT_FROZEN+false", "NOT_FROZEN.", "false", False), + ("FROZEN+true", "FROZEN.", "true", True)): + body = "```text\nStatus:\n %s\n collection_authorized: %s\n```\n" % (status, flag) + root, commit = repo_with("t0-" + flag + status[:4], {"t0.md": body}) + got = hq.bind_t0(root, "t0.md", commit)[1]["result"] == "pass" + if got != must_pass: + failures.append("hostqual accepted " + label if got else + "hostqual refused " + label) + try: + eb.t0_at(root, "t0.md", commit); accepted = True + except eb.BindingRefused: + accepted = False + if accepted != must_pass: + failures.append("execbinding accepted " + label if accepted else + "execbinding refused " + label) + +# --- W1/W2: the campaign link must be verified, not merely mentioned -------- +froot, fcommit = repo_with("freeze", {"d7/payload.json": '{"kind": "d7"}\n', + "d7/attestation.json": '{"kind": "att"}\n'}) +payload, att = froot / "d7/payload.json", froot / "d7/attestation.json" +blob = subprocess.run(["git", "-C", str(froot), "rev-parse", fcommit + ":d7/payload.json"], + capture_output=True, text=True, check=True).stdout.strip() +binding = w("binding.json", {"kind": "own.net/p022/execution-binding"}) +link = {"kind": hq.CAMPAIGN_LINK_SCHEMA, "schema": 1, + "execution_binding_sha256": hq.sha256_file(binding), + "d7_payload": {"path": "d7/payload.json", "sha256": hq.sha256_file(payload), + "blob_sha": blob, "commit": fcommit}, + "d7_attestation": {"path": "d7/attestation.json", "sha256": hq.sha256_file(att)}, + "recorded_at": "w"} +if hq.check_campaign_link(link, binding, froot)["result"] != "pass": + failures.append("a correct campaign link was refused") +wrong = dict(link, execution_binding_sha256="f" * 64) +if hq.check_campaign_link(wrong, binding, froot)["result"] != "fail": + failures.append("a link naming another execution binding was accepted") +payload.write_text('{"kind": "d7", "edited": true}\n', encoding="utf-8") +if hq.check_campaign_link(link, binding, froot)["result"] != "fail": + failures.append("a freeze edited after linking was accepted") + +# --- W5: a campaign cannot be swapped between preflight and postflight ------ +QUAL = {"kind": hq.QUALIFICATION_SCHEMA, "schema": 1, "stratum": "linux", + "t0": {"commit": "c" * 40, "path": "t0.md", "blob_sha": "b" * 40, + "sha256": "f" * 64, "status": "FROZEN"}, + "environment_id": "env-1", "host_fingerprint": "sha256:abc", + "provisioning": {"sha256": "0" * 64}, "environment_manifest": {"sha256": "1" * 64}, + "qualification_tool": {"sha256": "2" * 64}, "power_snapshot": dict(POWER), + "predicate": {k: "pass" for k in hq.PREDICATE_KEYS}, + "memory_metric": hq.STRATUM_METRIC["linux"], "qualified": True, "qualified_at": "w"} +MANIFEST = {"schema": hq.ENVCAPTURE_SCHEMA, + "identity": {"environment_id": {"status": "observed", "value": "env-1"}, + "host_fingerprint": {"status": "observed", "value": "sha256:abc"}}, + "provenance": {"ci": False}} +QUAL["environment_identity_sha256"] = hq.canonical_sha256(MANIFEST["identity"]) +CAND = b"candidate bytes" +cand = tmp / "cand.bin"; cand.write_bytes(CAND) +qpath = w("qual.json", QUAL) +IDENT = QUAL["environment_identity_sha256"] +stratum_block = lambda m, extra: {"qualification_sha256": hq.sha256_file(qpath), + "environment_id": "env-1", + "host_fingerprint": "sha256:abc", + "environment_identity_sha256": IDENT, + "candidate_sha256": hq.sha256_bytes(CAND), + "candidate_bytes": len(CAND), "memory_metric": m, **extra} +BINDING = {"kind": eb.BINDING_SCHEMA, "schema": 1, + "t0": {"commit": "c" * 40, "path": "t0.md", "blob_sha": "b" * 40, "sha256": "f" * 64}, + "instrument": {"accepted_commit": "a" * 40, "harness_digest": "d" * 64}, + "workloads": {"path": eb.WORKLOAD_MANIFEST, "manifest_sha256": "9" * 64}, + "linux": stratum_block(hq.STRATUM_METRIC["linux"], {}), + "windows": stratum_block(hq.STRATUM_METRIC["windows"], {})} +bpath = w("exec-binding.json", BINDING) +mpath = w("manifest.json", MANIFEST) +dpath = w("declaration.json", {"kind": hq.DECLARATION_SCHEMA, "schema": 1, + "no_campaign_workload": True, + "no_interactive_user_workload": True, + "no_prohibited_background_job_active": True, + "operator": "w", "recorded_at": "w"}) +probe = tmp / "probe.json"; probe.write_text("{}", encoding="utf-8") +quiet = {"eligible": True, "reason": "", "samples": [0.01], "mean": 0.01, "max": 0.01} + +def link_for(binding_file, root, commit_sha, payload_file, att_file, blob_sha): + return {"kind": hq.CAMPAIGN_LINK_SCHEMA, "schema": 1, + "execution_binding_sha256": hq.sha256_file(binding_file), + "d7_payload": {"path": "d7/payload.json", "sha256": hq.sha256_file(payload_file), + "blob_sha": blob_sha, "commit": commit_sha}, + "d7_attestation": {"path": "d7/attestation.json", "sha256": hq.sha256_file(att_file)}, + "recorded_at": "w"} + +r2, c2 = repo_with("freeze2", {"d7/payload.json": '{"kind": "d7", "campaign": "two"}\n', + "d7/attestation.json": '{"kind": "att"}\n'}) +p2, a2 = r2 / "d7/payload.json", r2 / "d7/attestation.json" +b2 = subprocess.run(["git", "-C", str(r2), "rev-parse", c2 + ":d7/payload.json"], + capture_output=True, text=True, check=True).stdout.strip() +r3, c3 = repo_with("freeze3", {"d7/payload.json": '{"kind": "d7", "campaign": "three"}\n', + "d7/attestation.json": '{"kind": "att"}\n'}) +p3, a3 = r3 / "d7/payload.json", r3 / "d7/attestation.json" +b3 = subprocess.run(["git", "-C", str(r3), "rev-parse", c3 + ":d7/payload.json"], + capture_output=True, text=True, check=True).stdout.strip() +l2 = w("link2.json", link_for(bpath, r2, c2, p2, a2, b2)) +l3 = w("link3.json", link_for(bpath, r3, c3, p3, a3, b3)) + +pre = hq.session_eligibility(bpath, qpath, mpath, dpath, cand, l2, r2, quiesce_result=quiet) +if not pre["eligible"]: + failures.append("a correctly linked session was refused: " + str(pre["reasons"])[:120]) +ppath = w("preflight.json", pre) +# The positive half first. Without it a postflight hard-wired to inadmissible +# would satisfy the negative half, and this witness would read "everything is +# refused" as "the swap is refused" — a broken closing probe standing in for the +# check it was meant to prove. +kept = hq.session_admissibility(bpath, qpath, ppath, mpath, cand, probe, l2, r2) +if not kept["admissible"]: + failures.append("an unchanged campaign was inadmissible at postflight: " + + str(kept.get("reasons"))[:140]) +after = hq.session_admissibility(bpath, qpath, ppath, mpath, cand, probe, l3, r3) +if after["admissible"]: + failures.append("a campaign swapped between preflight and postflight was admissible") +elif not any("campaign link" in str(r) for r in after.get("reasons", ())): + failures.append("the swapped campaign was refused, but by no reason naming the campaign " + "link: " + str(after.get("reasons"))[:140]) + +print(json.dumps(failures)) +""" + + +def witness(repo: Path, commit: str) -> list[str]: + """Run the target tree's own tools and require them to refuse. + + A name in a source file proves nothing: a comment, a dead function or + `def check_campaign_link(): pass` all satisfy a lexical check, and an earlier + revision of this gate was satisfied by exactly that — its own fixture shipped + the no-op and called the world compliant. So the tools are extracted from the + target tree, imported in a subprocess, and driven against attacks they are + claimed to stop. + """ + with tempfile.TemporaryDirectory() as raw: + work = Path(raw) + for tool in STEP7_TOOLS: + data = blob(repo, commit, tool) + if data is None: + return [f"{tool} is absent"] + target = work / tool + target.parent.mkdir(parents=True, exist_ok=True) + target.write_bytes(data) + script = work / "witness.py" + script.write_text(WITNESS_SOURCE, encoding="utf-8") + fixtures = work / "fixtures" + fixtures.mkdir() + proc = subprocess.run([sys.executable, str(script), str(fixtures)], + capture_output=True, text=True, cwd=str(work), check=False) + if proc.returncode != 0: + tail = (proc.stderr or proc.stdout).strip().splitlines()[-1:] or [""] + return [f"the machinery could not be exercised: {tail[0][:160]}"] + try: + return list(json.loads(proc.stdout.strip().splitlines()[-1])) + except (ValueError, IndexError): + return [f"the witness produced no verdict: {proc.stdout[:120]!r}"] + +def check(name: str, ok: bool, detail: str) -> dict[str, object]: + return {"check": name, "result": "pass" if ok else "fail", "detail": detail} + + +def gate(repo: Path, commit: str) -> list[dict[str, object]]: + results: list[dict[str, object]] = [] + + raw_t0 = blob(repo, commit, T0_PATH) + if raw_t0 is None: + return [check("t0_present", False, f"{T0_PATH} does not exist at {commit}")] + text = raw_t0.decode("utf-8", "replace") + frozen = re.search(r"^\s*(NOT_FROZEN|FROZEN)\.?\s*$", text, re.MULTILINE) + flag = re.search(r"^\s*collection_authorized:\s*(true|false)\s*$", text, re.MULTILINE) + declared = frozen.group(1) if frozen else "" + authorized = (flag.group(1) == "true") if flag else None + results.append(check( + "t0_frozen_and_authorized", declared == "FROZEN" and authorized is True, + f"T0 declares {declared}, collection_authorized={json.dumps(authorized)}")) + + named = DIGEST_RE.search(text) + results.append(check("t0_names_a_digest", named is not None, + f"T0 names {named.group(1)[:12] if named else ''} as the accepted " + "harness identity")) + if named is None: + return results + expected = named.group(1) + + # Recomputed from the target tree's own sources, by the frozen formula. + live = eb.harness_digest_at(repo, commit) + results.append(check( + "instrument_matches_t0", live == expected, + f"the instrument at {commit[:12]} hashes to {live[:12] if live else ''}; " + f"T0 names {expected[:12]}. The frozen contract would otherwise point at an identity " + "this tree does not contain")) + + rebound = [] + for path, field in BINDING_ARTIFACTS.items(): + raw = blob(repo, commit, path) + if raw is None: + rebound.append(f"{Path(path).name}: absent") + continue + try: + doc = json.loads(raw.decode("utf-8")) + except ValueError as exc: + rebound.append(f"{Path(path).name}: not readable JSON ({exc})") + continue + # The exact field, by its own path in that artifact. A substring search + # would accept the right digest sitting in any passing field while the + # real binding stayed old — which is the same defect one level down. + node: object = doc + for key in field: + node = node.get(key) if isinstance(node, dict) else None + if node != expected: + got = str(node)[:12] if node is not None else "" + rebound.append(f"{Path(path).name}: {'.'.join(field)} is {got}, not {expected[:12]}") + results.append(check( + "steps_4_5_6_rebound", not rebound, + "; ".join(rebound) if rebound else + f"the policy freeze, the design constants and the training preregistration all bind " + f"{expected[:12]}")) + + missing_tools = [t for t in STEP7_TOOLS if blob(repo, commit, t) is None] + if missing_tools: + results.append(check("step7_machinery_enforces", False, f"missing: {missing_tools}")) + else: + refusals = witness(repo, commit) + results.append(check( + "step7_machinery_enforces", not refusals, + "; ".join(refusals) if refusals else + "the target tree's own tools were run: both readers refused all three forbidden " + "authority states and accepted FROZEN+true; a link naming another binding and a " + "freeze edited after linking were refused; an unchanged campaign survived " + "preflight to postflight and a swapped one was refused by campaign-link " + "continuity")) + + note = (blob(repo, commit, STEP7_NOTE) or b"").decode("utf-8", "replace") + revoked = "AUTOMATIC AUTHORISATION OF THE FIRST STEP-7 COLLECTION IS REVOKED" in note + lingering = "SINGLE STEP-7 COLLECTION AUTHORISED automatically" in note + results.append(check( + "t0_zero_prerequisite", revoked and not lingering, + "the step-7 status block revokes the automatic collection authority" + if revoked and not lingering else + "the step-7 note still authorises the first collection automatically, so hosts plus a " + "binding would again be enough to start a clock")) + return results + + +def main(argv: list[str] | None = None) -> int: + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("--repo", type=Path, default=Path(__file__).resolve().parents[2]) + parser.add_argument("--commit", default="HEAD") + parser.add_argument("--json", action="store_true") + args = parser.parse_args(argv) + + results = gate(args.repo, args.commit) + failed = [r for r in results if r["result"] != "pass"] + if args.json: + print(json.dumps({"commit": args.commit, "checks": results, + "merge_allowed": not failed}, indent=2)) + else: + for r in results: + print(f"{'ok ' if r['result'] == 'pass' else 'FAIL'} [{r['check']}] {r['detail']}") + print() + print("merge gate: allowed" if not failed else + f"merge gate: REFUSED, {len(failed)} predicate(s) unsatisfied — the frozen " + "contract must not become reachable from this tree") + return 1 if failed else 0 + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/scripts/step7/mergegate_ci.py b/scripts/step7/mergegate_ci.py new file mode 100644 index 00000000..1d839e9c --- /dev/null +++ b/scripts/step7/mergegate_ci.py @@ -0,0 +1,136 @@ +#!/usr/bin/env python3 +"""P-022 — the merge gate as a status check. + +`mergegate.py` answers one question: may the frozen contract become reachable +from this tree? It answers it about a tree that *has* a frozen contract. A +required status check runs on every pull request in the repository, including +the ones that have nothing to do with P-022, so two rules have to be settled +before that gate can be wired to a branch protection rule. + +**Applicability.** If the frozen T0 does not exist at the merge commit, no merge +can make it reachable and there is nothing to protect. The check passes and says +so. This is not a way around the gate: the moment a tree carries T0, every +predicate applies in full. + +**Co-change.** A required check that a pull request can rewrite is decoration. +The workflow that runs on a pull request is the one *on that pull request*, so a +branch may carry both the freeze and a weakened gate and be judged by the gate +it brought with it. So: **the gate's own files may not change in the same merge +that introduces or changes the frozen contract.** Repairing the gate is still +ordinary work; doing it in the same breath as the freeze is not. + +Neither rule is enforcement isolation. A repository administrator can turn the +protection off, and this script cannot see that. What it removes is the quiet +path — a weakened gate arriving as part of the change it was meant to judge. + +Exit codes: 0 the merge may proceed (or the gate does not apply), 1 it is +refused, 2 the question could not be asked. + + python scripts/step7/mergegate_ci.py --repo . --commit \ + --base + +`--base` is what the co-change rule compares against. Without it the rule cannot +run, and the output says that rather than implying it passed. +""" + +from __future__ import annotations + +import argparse +import subprocess +import sys +from pathlib import Path + +sys.path.insert(0, str(Path(__file__).resolve().parent)) + +import mergegate as mg + +# The gate's own surface: the predicates, this wrapper, and the workflow that +# runs them. A change to any of these changes what "the gate passed" means. +GATE_FILES = ("scripts/step7/mergegate.py", + "scripts/step7/mergegate_ci.py", + ".github/workflows/p022-merge-gate.yml") + + +def blob_sha(repo: Path, commit: str, path: str) -> str | None: + """The blob a path resolves to, or None when the path is not there.""" + proc = subprocess.run(["git", "-C", str(repo), "rev-parse", f"{commit}:{path}"], + capture_output=True, text=True, check=False) + return proc.stdout.strip() if proc.returncode == 0 else None + + +def applies(repo: Path, commit: str) -> bool: + return blob_sha(repo, commit, mg.T0_PATH) is not None + + +def is_commit(repo: Path, rev: str) -> bool: + return subprocess.run(["git", "-C", str(repo), "rev-parse", f"{rev}^{{commit}}"], + capture_output=True, check=False).returncode == 0 + + +def co_change(repo: Path, commit: str, base: str) -> list[str]: + """Which gate files this merge changes, if it also touches the contract. + + Empty when the contract is untouched: repairing the gate is ordinary work. + """ + t0_now, t0_base = blob_sha(repo, commit, mg.T0_PATH), blob_sha(repo, base, mg.T0_PATH) + if t0_now == t0_base: + return [] + return [path for path in GATE_FILES + if blob_sha(repo, commit, path) != blob_sha(repo, base, path)] + + +def main(argv: list[str] | None = None) -> int: + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("--repo", type=Path, default=Path(".")) + parser.add_argument("--commit", required=True, + help="the commit a merge would produce, not the branch head") + parser.add_argument("--base", help="the base commit the co-change rule compares against") + args = parser.parse_args(argv) + + if not is_commit(args.repo, args.commit): + print(f"merge gate: MISUSE — {args.commit} is not a commit in {args.repo}") + return 2 + # A base that is not in the clone would make every gate file look changed, + # and the co-change rule would refuse for a reason that is about the checkout + # rather than about the merge. A question that cannot be asked is not a no. + if args.base is not None and not is_commit(args.repo, args.base): + print(f"merge gate: MISUSE — the base {args.base} is not a commit in {args.repo}; " + "the co-change rule cannot be evaluated, and a shallow checkout must not be " + "reported as a rewritten gate") + return 2 + + if not applies(args.repo, args.commit): + print(f"merge gate: not applicable — {mg.T0_PATH} does not exist at {args.commit[:12]}, " + "so no merge here can make the frozen contract reachable. Every predicate " + "applies the moment a tree carries it.") + return 0 + + if args.base is None: + print("merge gate: the co-change rule did not run — no --base was given, so this pass " + "cannot say whether the gate itself changed alongside the contract") + else: + changed = co_change(args.repo, args.commit, args.base) + if changed: + print("merge gate: REFUSED — this merge changes the frozen contract and the gate " + f"that judges it in one act: {', '.join(changed)}. Repair the gate in its own " + "merge; a guard that arrives with what it guards is not a guard.") + return 1 + print("ok [gate_unchanged] the contract moves and the gate does not; this merge is " + "judged by the gate already on the base") + + results = mg.gate(args.repo, args.commit) + for row in results: + mark = "ok " if row["result"] == "pass" else "FAIL" + print(f"{mark} [{row['check']}] {row['detail']}") + bad = [r for r in results if r["result"] != "pass"] + print() + if bad: + print(f"merge gate: REFUSED, {len(bad)} predicate(s) unsatisfied — the frozen contract " + "must not become reachable from this tree") + return 1 + print("merge gate: allowed") + return 0 + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/tests/test_step7_mergegate.py b/tests/test_step7_mergegate.py new file mode 100644 index 00000000..1593235f --- /dev/null +++ b/tests/test_step7_mergegate.py @@ -0,0 +1,499 @@ +#!/usr/bin/env python3 +"""P-022 — controls on the merge gate. + + mergegate-complete-tree a world that satisfies the freeze is allowed + mergegate-missing-instrument the digest T0 names must exist in the tree + mergegate-stale-bindings steps 4/5/6 must be re-bound to that digest + mergegate-exact-binding-field the digest is read at each artifact's own path + mergegate-decoy-digest the right digest in a field nobody binds is not a binding + mergegate-noop-campaign-link a permissive link check must not satisfy the gate + mergegate-dead-authority hostqual authorising anything must not satisfy it + mergegate-dead-authority-eb nor must execbinding — the claim covers both readers + mergegate-dead-continuity nor must a postflight that stops comparing campaign links + mergegate-always-inadmissible nor must one that refuses every campaign + mergegate-missing-machinery the tools must be there at all + mergegate-lingering-auto the revoked automatic authority must stay revoked + mergegate-unfrozen-t0 an unfrozen or unauthorised T0 is not merged as frozen + mergegate-reads-t0-digest the expected digest comes from T0, not from a constant + control-inventory-complete this list and the executed set are the same set + +Fixtures ship the real tools, and four controls mutate one enforcement point +each. An earlier revision shipped a stub whose check_campaign_link was a bare +pass and called that world compliant: the fixture demonstrated the false +positive it was meant to exclude. Checking for the name of a mechanism is not +checking the mechanism, and a witness that only ever sees refusals is not +checking one either — hence a positive control beside each negative one. + +Failures print FAIL[]: ; nothing stops at the first one. + +Run: python tests/test_step7_mergegate.py +""" + +from __future__ import annotations + +import contextlib +import io +import json +import re +import subprocess +import sys +import tempfile +from collections.abc import Callable +from pathlib import Path + +ROOT = Path(__file__).resolve().parent.parent +sys.path.insert(0, str(ROOT / "scripts")) +sys.path.insert(0, str(ROOT / "scripts" / "step7")) + +import execbinding as eb # noqa: E402 +import mergegate as mg # noqa: E402 + +NL = chr(10) + +_FAILURES: list[tuple[str, str]] = [] +_PASSES: list[str] = [] + + +def fail(check: str, detail: str) -> None: + _FAILURES.append((check, detail)) + print(f"FAIL[{check}]: {detail}") + + +def ok(check: str, detail: str = "") -> None: + _PASSES.append(check) + print(f"ok[{check}]: {detail}" if detail else f"ok[{check}]") + + +def guarded(check: str, control: Callable[[], None]) -> None: + try: + control() + except Exception as exc: + fail(check, f"the control raised {type(exc).__name__}: {exc}") + + +NOTE_REVOKED = """Status: + AUTOMATIC AUTHORISATION OF THE FIRST STEP-7 COLLECTION IS REVOKED (T0-0). +""" +NOTE_AUTO = """Status: + SINGLE STEP-7 COLLECTION AUTHORISED automatically after that gate. +""" + +T0_TEMPLATE = """```text +Status: + {status} + collection_authorized: {flag} +``` + + measurement_harness_digest + {digest} +""" + +# The digest sits at a different exact path in each artifact, so a fixture that +# wrote one flat shape everywhere would be exercising a gate that does not exist. +BINDING_SHAPES = { + "docs/evidence/calibration/p022-263a-policy-freeze.json": + lambda d: {"measurement_harness_digest": d}, + "docs/evidence/calibration/p022-263a-design-constants.json": + lambda d: {"bound_measurement_harness_digest": d}, + "docs/evidence/calibration/p022-263a-training-preregistration.json": + lambda d: {"bindings": {"measurement_harness_digest": d}}, +} + +HOSTQUAL, EXECBINDING = mg.STEP7_TOOLS[1], mg.STEP7_TOOLS[2] + +# Mutations of the real tools, one enforcement point each. Appending a definition +# is enough where a later definition wins; the continuity check lives inside a +# larger function, so that one is a targeted edit of its condition. +NOOP_LINK = ''' + +def check_campaign_link(link, binding_path, repo): + return check("campaign_link", True, "stubbed: what a lexical gate accepted") +''' + +DEAD_AUTHORITY = ''' + +def bind_t0(repo, path, commit): + block = {"commit": commit, "path": path, "blob_sha": "b" * 40, + "sha256": "f" * 64, "status": "FROZEN", "collection_authorized": True} + return block, check("t0", True, "stubbed: always authorised") +''' + +# The second reader. The gate's message claims enforcement on both, so a control +# that only breaks the first leaves half of that claim unproved. +DEAD_AUTHORITY_EB = ''' + +def t0_at(repo, path, commit): + return {"commit": commit, "path": path, "blob_sha": "b" * 40, "sha256": "f" * 64, + "status": "FROZEN", "collection_authorized": True} +''' + +# The mutation the positive postflight control exists to catch: a tool that +# refuses every campaign satisfies a negative-only witness, which would then be +# reading "nothing is admissible" as "the swap was caught". +ALWAYS_INADMISSIBLE = ''' + +def session_admissibility(*args, **kwargs): + return {"admissible": False, "reasons": ["stubbed: nothing is ever admissible"]} +''' + +CONTINUITY_CONDITION = ('preflight.get("campaign_link_sha256") != ' + "sha256_file(campaign_link_path)") + + +def dead_continuity(src: str) -> str: + """Leave check_campaign_link working and remove only the comparison of the + preflight's campaign link with this pass's.""" + return src.replace(CONTINUITY_CONDITION, "False") + + +def real_tools() -> dict[str, str]: + """The tools as they actually are. A fixture that shipped a stub would prove + only that the gate accepts stubs.""" + return {t: (ROOT / t).read_text(encoding="utf-8") for t in mg.STEP7_TOOLS} + + +def commit_tree(repo: Path, files: dict[str, str]) -> str: + repo.mkdir(parents=True, exist_ok=True) + + def run(*args: str) -> None: + subprocess.run(["git", "-C", str(repo), *args], capture_output=True, check=True) + + if not (repo / ".git").exists(): + run("init", "-q") + run("config", "user.email", "control@example.invalid") + run("config", "user.name", "control") + for name, text in files.items(): + path = repo / name + path.parent.mkdir(parents=True, exist_ok=True) + path.write_text(text, encoding="utf-8") + run("add", "-A") + run("commit", "-q", "-m", "fixture") + return subprocess.run(["git", "-C", str(repo), "rev-parse", "HEAD"], + capture_output=True, text=True, check=True).stdout.strip() + + +def world(tmp: Path, *, frozen: bool = True, authorized: bool = True, + instrument: str = "print('instrument')" + NL, rebound: bool = True, + machinery: bool = True, auto_authority: bool = False, decoy: bool = False, + wrong_field: bool = False, mutate: dict[str, object] | None = None, + name: str = "w") -> tuple[Path, str]: + """A synthetic target tree, and the commit a merge into it would produce.""" + repo = tmp / name + files = {eb.INSTRUMENT_SOURCES[0]: instrument, + eb.INSTRUMENT_SOURCES[1]: '{"decisive": []}' + NL, + mg.STEP7_NOTE: NOTE_AUTO if auto_authority else NOTE_REVOKED} + if machinery: + tools = real_tools() + for tool, change in (mutate or {}).items(): + before = tools[tool] + tools[tool] = change(before) if callable(change) else before + str(change) + if tools[tool] == before: + # A mutation that lands nowhere turns an attack control into a + # second positive control that nobody reads as one. + raise AssertionError(f"the mutation for {tool} changed nothing; the source it " + "targets has moved") + files.update(tools) + else: + files[mg.STEP7_TOOLS[0]] = "# capture only" + NL + probe = commit_tree(repo, files) + # T0 names whatever THIS tree's own sources hash to, so a fixture cannot pass + # by agreeing with a constant this file also wrote. + digest = eb.harness_digest_at(repo, probe) or "" + files[mg.T0_PATH] = T0_TEMPLATE.format( + status="FROZEN." if frozen else "NOT_FROZEN.", + flag="true" if authorized else "false", + digest=digest) + bound = digest if rebound else "0" * 64 + for artifact, shape in BINDING_SHAPES.items(): + doc = shape(bound) + if decoy: # the right digest, in a field nobody binds + doc["decoy"] = digest + if wrong_field and "training" in artifact: # right digest, wrong exact path + doc = {"measurement_harness_digest": digest} + files[artifact] = json.dumps(doc, indent=2) + NL + return repo, commit_tree(repo, files) + + +def verdicts(repo: Path, commit: str) -> dict[str, str]: + return {str(r["check"]): str(r["result"]) for r in mg.gate(repo, commit)} + + +def details(repo: Path, commit: str) -> dict[str, str]: + return {str(r["check"]): str(r["detail"]) for r in mg.gate(repo, commit)} + + +def control_complete_tree() -> None: + with tempfile.TemporaryDirectory() as raw: + repo, commit = world(Path(raw)) + bad = sorted(k for k, v in verdicts(repo, commit).items() if v != "pass") + if bad: + seen = details(repo, commit) + fail("mergegate-complete-tree", + f"a satisfying world was refused on {bad}: {[seen[k][:90] for k in bad]}") + return + # Captured: a green run that prints a refusal teaches readers to skim past + # refusals, which is how the word stops meaning anything. + with contextlib.redirect_stdout(io.StringIO()): + rc = mg.main(["--repo", str(repo), "--commit", commit]) + if rc != 0: + fail("mergegate-complete-tree", "the CLI refused a satisfying world") + return + ok("mergegate-complete-tree", + "a tree carrying the named instrument, the exactly re-bound artifacts, the real tools " + "and the revoked automatic authority is allowed") + + +def control_missing_instrument() -> None: + with tempfile.TemporaryDirectory() as raw: + repo, _commit = world(Path(raw)) + # the instrument moves after T0 named it — the exact intermediate state a + # stray merge produces when the repair PR is not in yet + drifted = commit_tree(repo, {eb.INSTRUMENT_SOURCES[0]: "print('older')" + NL}) + if verdicts(repo, drifted).get("instrument_matches_t0") != "fail": + fail("mergegate-missing-instrument", + "a tree whose instrument does not hash to the digest T0 names was allowed") + return + noise = io.StringIO() + with contextlib.redirect_stdout(noise): + rc = mg.main(["--repo", str(repo), "--commit", drifted]) + if rc == 0 or "REFUSED" not in noise.getvalue(): + fail("mergegate-missing-instrument", "the CLI allowed it") + return + ok("mergegate-missing-instrument", + "the frozen contract cannot become reachable from a tree whose instrument is not the " + "one it names") + + +def control_stale_bindings() -> None: + with tempfile.TemporaryDirectory() as raw: + repo, commit = world(Path(raw), rebound=False) + if verdicts(repo, commit).get("steps_4_5_6_rebound") != "fail": + fail("mergegate-stale-bindings", "artifacts bound to an older digest were allowed") + return + ok("mergegate-stale-bindings", + "steps 4, 5 and 6 must be re-bound to the digest T0 names, not merely present") + + +def control_exact_binding_field() -> None: + """Each artifact binds at its own path, and only that path counts.""" + with tempfile.TemporaryDirectory() as raw: + tmp = Path(raw) + repo, commit = world(tmp, name="exact") + if verdicts(repo, commit).get("steps_4_5_6_rebound") != "pass": + fail("mergegate-exact-binding-field", + "correct exact fields were refused: " + + details(repo, commit)["steps_4_5_6_rebound"]) + return + moved_repo, moved = world(tmp, wrong_field=True, name="wrongfield") + if verdicts(moved_repo, moved).get("steps_4_5_6_rebound") != "fail": + fail("mergegate-exact-binding-field", + "the training preregistration bound at the wrong path was accepted; the gate " + "is searching the file rather than reading the binding") + return + ok("mergegate-exact-binding-field", + "policy freeze at measurement_harness_digest, design constants at " + "bound_measurement_harness_digest, training preregistration at " + "bindings.measurement_harness_digest — and nowhere else") + + +def control_decoy_digest() -> None: + with tempfile.TemporaryDirectory() as raw: + repo, commit = world(Path(raw), rebound=False, decoy=True) + if verdicts(repo, commit).get("steps_4_5_6_rebound") != "fail": + fail("mergegate-decoy-digest", + "a stale binding passed because the right digest sat in a decoy field") + return + ok("mergegate-decoy-digest", + "the right digest in a field nobody binds is not a binding, and a substring search " + "would have called it one") + + +def control_noop_campaign_link() -> None: + """The attack the previous revision of this gate could not see.""" + with tempfile.TemporaryDirectory() as raw: + repo, commit = world(Path(raw), mutate={HOSTQUAL: NOOP_LINK}, name="noop") + if verdicts(repo, commit).get("step7_machinery_enforces") != "fail": + fail("mergegate-noop-campaign-link", + "a permissive check_campaign_link satisfied the gate; the name of a mechanism " + "is not the mechanism") + return + detail = details(repo, commit)["step7_machinery_enforces"] + if "accepted" not in detail: + fail("mergegate-noop-campaign-link", f"refused for an unrelated reason: {detail}") + return + ok("mergegate-noop-campaign-link", + "a tree whose link check always passes is refused, and the refusal names the attack " + "that got through") + + +def _dead_authority(check_name: str, tool: str, stub: str, reader: str) -> None: + """Both readers carry the authority state machine, and the gate says so. + A control that breaks one of them leaves the other half of that claim + standing on nothing.""" + with tempfile.TemporaryDirectory() as raw: + repo, commit = world(Path(raw), mutate={tool: stub}, name="deadauth") + if verdicts(repo, commit).get("step7_machinery_enforces") != "fail": + fail(check_name, f"a {reader} authority check that always passes satisfied the gate") + return + detail = details(repo, commit)["step7_machinery_enforces"] + wrongly = [s for s in ("FROZEN+false", "NOT_FROZEN+true", "NOT_FROZEN+false") + if f"{reader} accepted {s}" in detail] + if len(wrongly) != 3: + fail(check_name, f"refused, but named {wrongly} rather than all three forbidden " + f"states: {detail}") + return + ok(check_name, + f"a tree whose {reader} authorises anything is refused, and the refusal names every " + "forbidden state it accepted: FROZEN+false, NOT_FROZEN+true, NOT_FROZEN+false") + + +def control_dead_authority() -> None: + _dead_authority("mergegate-dead-authority", HOSTQUAL, DEAD_AUTHORITY, "hostqual") + + +def control_dead_authority_eb() -> None: + _dead_authority("mergegate-dead-authority-eb", EXECBINDING, DEAD_AUTHORITY_EB, "execbinding") + + +def control_dead_continuity() -> None: + """The campaign-swap witness needs its own sensitivity proof: a postflight + that refuses everything would satisfy a negative-only control.""" + with tempfile.TemporaryDirectory() as raw: + repo, commit = world(Path(raw), mutate={HOSTQUAL: dead_continuity}, name="deadcont") + if verdicts(repo, commit).get("step7_machinery_enforces") != "fail": + fail("mergegate-dead-continuity", + "a postflight that no longer compares the preflight's campaign link with " + "this pass's satisfied the gate; the swap witness proves nothing") + return + detail = details(repo, commit)["step7_machinery_enforces"] + if "swapped" not in detail: + fail("mergegate-dead-continuity", f"refused for an unrelated reason: {detail}") + return + ok("mergegate-dead-continuity", + "removing only the preflight/postflight campaign-link comparison, and leaving " + "check_campaign_link intact, is caught and named as the swap becoming admissible") + + +def control_always_inadmissible() -> None: + """A refusal is only evidence if acceptance was possible.""" + with tempfile.TemporaryDirectory() as raw: + repo, commit = world(Path(raw), mutate={HOSTQUAL: ALWAYS_INADMISSIBLE}, name="noadmit") + if verdicts(repo, commit).get("step7_machinery_enforces") != "fail": + fail("mergegate-always-inadmissible", + "a postflight that refuses every campaign satisfied the gate; the swap " + "witness was reading a blanket refusal as enforcement") + return + detail = details(repo, commit)["step7_machinery_enforces"] + if "unchanged campaign was inadmissible" not in detail: + fail("mergegate-always-inadmissible", f"refused for another reason: {detail}") + return + ok("mergegate-always-inadmissible", + "the unchanged campaign must survive preflight to postflight, so a tool that refuses " + "everything cannot pose as one that caught the swap") + + +def control_missing_machinery() -> None: + with tempfile.TemporaryDirectory() as raw: + repo, commit = world(Path(raw), machinery=False) + if verdicts(repo, commit).get("step7_machinery_enforces") != "fail": + fail("mergegate-missing-machinery", + "a tree without the qualification and binding tools was allowed") + return + ok("mergegate-missing-machinery", + "without the tools that enforce the campaign link and the authority state, the freeze " + "would enter a world that cannot keep its promises") + + +def control_lingering_auto() -> None: + with tempfile.TemporaryDirectory() as raw: + repo, commit = world(Path(raw), auto_authority=True) + if verdicts(repo, commit).get("t0_zero_prerequisite") != "fail": + fail("mergegate-lingering-auto", + "a tree still authorising the first collection automatically was allowed") + return + ok("mergegate-lingering-auto", + "the revoked automatic authority must still be revoked in the target tree, or hosts " + "plus a binding are again enough to start a clock") + + +def control_unfrozen_t0() -> None: + states = (("not frozen", {"frozen": False}), + ("frozen but unauthorised", {"authorized": False}), + ("unfrozen but authorised", {"frozen": False, "authorized": True})) + with tempfile.TemporaryDirectory() as raw: + tmp = Path(raw) + for i, (label, kwargs) in enumerate(states): + repo, commit = world(tmp, name=f"state{i}", **kwargs) + if verdicts(repo, commit).get("t0_frozen_and_authorized") != "fail": + fail("mergegate-unfrozen-t0", f"{label} was treated as a freeze") + return + ok("mergegate-unfrozen-t0", + "only FROZEN together with collection_authorized: true is merged as a freeze; the other " + "three states are refused here as they are at the session gates") + + +def control_reads_t0_digest() -> None: + """The gate must not carry the expected digest in its own source.""" + source = (ROOT / "scripts" / "step7" / "mergegate.py").read_text(encoding="utf-8") + hardcoded = re.findall("(? None: + listed = set(re.findall("^ ([a-z0-9-]+) +[^ ]", __doc__ or "", re.MULTILINE)) + executed = {name for name, _ in CONTROLS} + if listed != executed: + fail("control-inventory-complete", + f"listed but not executed: {sorted(listed - executed)}; executed but not listed: " + f"{sorted(executed - listed)}") + return + ok("control-inventory-complete", + f"{len(executed)} controls listed, {len(executed)} executed, same names in both") + + +CONTROLS: list[tuple[str, Callable[[], None]]] = [ + ("mergegate-complete-tree", control_complete_tree), + ("mergegate-missing-instrument", control_missing_instrument), + ("mergegate-stale-bindings", control_stale_bindings), + ("mergegate-exact-binding-field", control_exact_binding_field), + ("mergegate-decoy-digest", control_decoy_digest), + ("mergegate-noop-campaign-link", control_noop_campaign_link), + ("mergegate-dead-authority", control_dead_authority), + ("mergegate-dead-authority-eb", control_dead_authority_eb), + ("mergegate-dead-continuity", control_dead_continuity), + ("mergegate-always-inadmissible", control_always_inadmissible), + ("mergegate-missing-machinery", control_missing_machinery), + ("mergegate-lingering-auto", control_lingering_auto), + ("mergegate-unfrozen-t0", control_unfrozen_t0), + ("mergegate-reads-t0-digest", control_reads_t0_digest), + ("control-inventory-complete", control_inventory_complete), +] + + +def run() -> int: + for name, control in CONTROLS: + guarded(name, control) + print() + print(f"merge gate controls: {len(_PASSES)} passed, {len(_FAILURES)} failed") + return 1 if _FAILURES else 0 + + +if __name__ == "__main__": + sys.exit(run()) diff --git a/tests/test_step7_mergegate_ci.py b/tests/test_step7_mergegate_ci.py new file mode 100644 index 00000000..60de8d8b --- /dev/null +++ b/tests/test_step7_mergegate_ci.py @@ -0,0 +1,302 @@ +#!/usr/bin/env python3 +"""P-022 — controls on the wiring that makes the merge gate a status check. + + ci-not-applicable a tree without the frozen T0 passes, and says why + ci-applies-in-full a tree with it is judged by every predicate + ci-refusal-survives a refusal from the gate is still a refusal here + ci-co-change-refused the gate may not change in the merge that brings the contract + ci-gate-repair-allowed repairing the gate on its own is ordinary work + ci-no-base-is-stated without a base the co-change rule says it did not run + ci-misuse-is-two a commit that does not exist is misuse, not a verdict + ci-absent-base-is-two nor may an unavailable base be reported as a rewritten gate + ci-gate-files-exist every file the co-change rule watches is really there + ci-workflow-names the workflow names the contexts a ruleset must require + ci-workflow-derives-t0 the workflow hard-codes no path the tools already own + control-inventory this list and the executed set are the same set + +The applicability rule is the one that makes a required check possible at all: +without it the gate refuses every pull request in the repository, including the +one that adds it. The co-change rule is the one that makes it worth requiring: +a pull request runs its own copy of the workflow, so a branch could otherwise +carry the freeze and a weakened gate together and be judged by the gate it +brought with it. + +Run: python tests/test_step7_mergegate_ci.py +""" + +from __future__ import annotations + +import contextlib +import io +import sys +import tempfile +from collections.abc import Callable +from pathlib import Path + +ROOT = Path(__file__).resolve().parent.parent +sys.path.insert(0, str(ROOT / "scripts" / "step7")) +sys.path.insert(0, str(ROOT / "tests")) + +import mergegate_ci as ci # noqa: E402 +import test_step7_mergegate as mgt # noqa: E402 + +WORKFLOW = ROOT / ".github/workflows/p022-merge-gate.yml" +CONTEXTS = ("P-022 merge gate controls", "P-022 merge gate") + +_FAILURES: list[tuple[str, str]] = [] +_PASSES: list[str] = [] + + +def fail(check: str, detail: str) -> None: + _FAILURES.append((check, detail)) + print(f"FAIL[{check}]: {detail}") + + +def ok(check: str, detail: str = "") -> None: + _PASSES.append(check) + print(f"ok[{check}]: {detail}" if detail else f"ok[{check}]") + + +def guarded(check: str, control: Callable[[], None]) -> None: + try: + control() + except Exception as exc: + fail(check, f"the control raised {type(exc).__name__}: {exc}") + + +def invoke(*args: str) -> tuple[int, str]: + """The wrapper's exit code and everything it said, captured. A green run that + prints REFUSED teaches readers to skim past refusals. + + Not named `run`: that name belongs to this module's own entry point, which + tests/run_tests.py calls.""" + out = io.StringIO() + with contextlib.redirect_stdout(out): + rc = ci.main(list(args)) + return rc, out.getvalue() + + +def wired(tmp: Path, name: str, **kwargs: object) -> tuple[Path, str, str]: + """A repository, the commit that carries T0, and the commit before it. + + `world()` builds its tree in two commits: the first carries the instrument + and the tools, the second adds T0 and the bindings. That first commit is + exactly the shape of a base branch that does not yet carry the contract. + """ + repo, head = mgt.world(tmp, name=name, **kwargs) + return repo, head, head + "^" + + +def control_not_applicable() -> None: + with tempfile.TemporaryDirectory() as raw: + repo, _, before = wired(Path(raw), "na") + rc, said = invoke("--repo", str(repo), "--commit", before) + if rc != 0: + fail("ci-not-applicable", f"a tree with no frozen contract was refused: {said[:160]}") + return + if "not applicable" not in said: + fail("ci-not-applicable", f"it passed without saying why: {said[:160]}") + return + ok("ci-not-applicable", + "with no T0 in the tree no merge can make the frozen contract reachable, so the check " + "passes and names the reason — otherwise it would refuse every pull request in the repo") + + +def control_applies_in_full() -> None: + with tempfile.TemporaryDirectory() as raw: + repo, head, before = wired(Path(raw), "full") + rc, said = invoke("--repo", str(repo), "--commit", head, "--base", before) + if rc != 0: + fail("ci-applies-in-full", f"a satisfying tree was refused: {said[:200]}") + return + missing = [c for c in ("t0_frozen_and_authorized", "instrument_matches_t0", + "steps_4_5_6_rebound", "step7_machinery_enforces") + if c not in said] + if missing: + fail("ci-applies-in-full", f"the wrapper did not report {missing}") + return + ok("ci-applies-in-full", + "the moment a tree carries T0 every predicate runs and is reported by name") + + +def control_refusal_survives() -> None: + with tempfile.TemporaryDirectory() as raw: + repo, head, before = wired(Path(raw), "stale", rebound=False) + rc, said = invoke("--repo", str(repo), "--commit", head, "--base", before) + if rc != 1 or "REFUSED" not in said: + fail("ci-refusal-survives", + f"a stale binding came back as rc={rc}: {said[:200]}") + return + ok("ci-refusal-survives", + "the wrapper adds rules, it does not soften the ones underneath: a stale binding is " + "still exit 1") + + +def control_co_change_refused() -> None: + with tempfile.TemporaryDirectory() as raw: + repo, head, before = wired(Path(raw), "cochange") + # the branch brings the contract and its own copy of the gate at once + attacked = mgt.commit_tree(repo, { + ci.GATE_FILES[0]: "# a gate that says yes\n", + ci.GATE_FILES[2]: "name: P-022 merge gate\n"}) + rc, said = invoke("--repo", str(repo), "--commit", attacked, "--base", before) + if rc != 1: + fail("ci-co-change-refused", + "a merge carrying both the freeze and a rewritten gate was allowed") + return + if ci.GATE_FILES[0] not in said: + fail("ci-co-change-refused", f"refused without naming the changed file: {said[:200]}") + return + _ = head + ok("ci-co-change-refused", + "a guard that arrives together with what it guards is not a guard; the refusal names " + "every gate file the merge rewrites") + + +def control_gate_repair_allowed() -> None: + with tempfile.TemporaryDirectory() as raw: + repo, head, _ = wired(Path(raw), "repair") + # the contract is already in the base and does not move; only the gate does + repaired = mgt.commit_tree(repo, {ci.GATE_FILES[0]: "# a repaired gate\n"}) + rc, said = invoke("--repo", str(repo), "--commit", repaired, "--base", head) + if rc != 0: + fail("ci-gate-repair-allowed", + f"an ordinary repair of the gate was refused: {said[:200]}") + return + if "gate_unchanged" not in said: + fail("ci-gate-repair-allowed", f"it passed without saying so: {said[:200]}") + return + ok("ci-gate-repair-allowed", + "when the contract does not move, changing the gate is ordinary work and stays " + "possible — the rule is about co-change, not about freezing the gate forever") + + +def control_no_base_is_stated() -> None: + with tempfile.TemporaryDirectory() as raw: + repo, head, _ = wired(Path(raw), "nobase") + rc, said = invoke("--repo", str(repo), "--commit", head) + if rc != 0: + fail("ci-no-base-is-stated", f"a satisfying tree was refused: {said[:200]}") + return + if "did not run" not in said: + fail("ci-no-base-is-stated", + f"a pass with no base did not say the co-change rule was skipped: {said[:200]}") + return + ok("ci-no-base-is-stated", + "a manual run cannot pose as a full one: without a base the wrapper says the co-change " + "rule did not run rather than implying it held") + + +def control_misuse_is_two() -> None: + with tempfile.TemporaryDirectory() as raw: + repo, _, _ = wired(Path(raw), "misuse") + rc, said = invoke("--repo", str(repo), "--commit", "f" * 40) + if rc != 2: + fail("ci-misuse-is-two", f"a commit that does not exist returned {rc}: {said[:160]}") + return + ok("ci-misuse-is-two", + "a question that could not be asked is exit 2, never a verdict — the same three codes " + "the step-7 tools already use") + + +def control_absent_base_is_two() -> None: + """A shallow checkout must not be able to manufacture a refusal.""" + with tempfile.TemporaryDirectory() as raw: + repo, head, _ = wired(Path(raw), "absentbase") + rc, said = invoke("--repo", str(repo), "--commit", head, "--base", "e" * 40) + if rc == 1: + fail("ci-absent-base-is-two", + "a base that is not in the clone was reported as a rewritten gate; every gate " + "file looks changed against a base nobody has") + return + if rc != 2: + fail("ci-absent-base-is-two", f"an unavailable base returned {rc}: {said[:160]}") + return + ok("ci-absent-base-is-two", + "an unreachable base is exit 2 and says so; the co-change rule refuses to answer rather " + "than answering with the checkout's depth") + + +def control_gate_files_exist() -> None: + missing = [p for p in ci.GATE_FILES if not (ROOT / p).exists()] + if missing: + fail("ci-gate-files-exist", + f"the co-change rule watches {missing}, which are not in this repository; a " + "renamed file would be watched by nobody") + return + ok("ci-gate-files-exist", + f"all {len(ci.GATE_FILES)} watched files exist: {', '.join(ci.GATE_FILES)}") + + +def control_workflow_names() -> None: + if not WORKFLOW.exists(): + fail("ci-workflow-names", f"{WORKFLOW.name} does not exist") + return + text = WORKFLOW.read_text(encoding="utf-8") + absent = [c for c in CONTEXTS if f"name: {c}" not in text] + if absent: + fail("ci-workflow-names", + f"the workflow declares no job named {absent}; a ruleset requiring that context " + "would wait forever on a check nothing reports") + return + if "mergegate_ci.py" not in text: + fail("ci-workflow-names", "the workflow does not run the wrapper") + return + ok("ci-workflow-names", + f"the workflow declares exactly the contexts a ruleset must require: {', '.join(CONTEXTS)}") + + +def control_workflow_derives_t0() -> None: + """The workflow must not carry a second copy of a path the tools own.""" + import mergegate as mg + text = WORKFLOW.read_text(encoding="utf-8") + if mg.T0_PATH in text: + fail("ci-workflow-derives-t0", + f"the workflow hard-codes {mg.T0_PATH}; applicability would then be decided in two " + "places that can drift apart") + return + ok("ci-workflow-derives-t0", + "applicability is decided once, in the tool that owns T0_PATH; the workflow only runs it") + + +def control_inventory() -> None: + import re + listed = set(re.findall("^ (ci-[a-z0-9-]+|control-inventory) +[^ ]", __doc__ or "", + re.MULTILINE)) + executed = {name for name, _ in CONTROLS} + if listed != executed: + fail("control-inventory", + f"listed but not executed: {sorted(listed - executed)}; executed but not listed: " + f"{sorted(executed - listed)}") + return + ok("control-inventory", f"{len(executed)} controls listed, {len(executed)} executed") + + +CONTROLS: list[tuple[str, Callable[[], None]]] = [ + ("ci-not-applicable", control_not_applicable), + ("ci-applies-in-full", control_applies_in_full), + ("ci-refusal-survives", control_refusal_survives), + ("ci-co-change-refused", control_co_change_refused), + ("ci-gate-repair-allowed", control_gate_repair_allowed), + ("ci-no-base-is-stated", control_no_base_is_stated), + ("ci-misuse-is-two", control_misuse_is_two), + ("ci-absent-base-is-two", control_absent_base_is_two), + ("ci-gate-files-exist", control_gate_files_exist), + ("ci-workflow-names", control_workflow_names), + ("ci-workflow-derives-t0", control_workflow_derives_t0), + ("control-inventory", control_inventory), +] + + +def run() -> int: + # `run`, not `main`: tests/run_tests.py imports every test_*.py and calls + # run() on it, and a module without one is a suite that reports nothing. + for name, control in CONTROLS: + guarded(name, control) + print() + print(f"merge gate wiring controls: {len(_PASSES)} passed, {len(_FAILURES)} failed") + return 1 if _FAILURES else 0 + + +if __name__ == "__main__": + sys.exit(run())