diff --git a/README.md b/README.md index 00bb792..4940327 100644 --- a/README.md +++ b/README.md @@ -43,6 +43,8 @@ Graduation at each level is merged work that a computer checked. That record is See FOUNDING-BRIEF.md for the full plan: how a project moves through the factory, how the academy works, how the org is funded, and the 90-day plan. See [reviews/](reviews/) for review records: what an attack on a project found, with commands to reproduce it. +The [branch policy](policy/README.md) records required checks and the read-only planning and explicit application steps. Merging the policy changes no live setting; an administrator applies a reviewed plan. + ## Website The public website is a static, buildless GitHub Pages site. Pages publishes the root of `main`; `.nojekyll` serves the files directly. Changes to `index.html` or `assets/` publish automatically after they reach `main`. No package installation, build system, secrets, or third-party scripts are required. diff --git a/bar/tests/test_required_checks.py b/bar/tests/test_required_checks.py new file mode 100644 index 0000000..d68fcb5 --- /dev/null +++ b/bar/tests/test_required_checks.py @@ -0,0 +1,248 @@ +"""Safety tests for the read-only planner and explicit branch-policy apply.""" +import copy +import importlib.util +from pathlib import Path +import tempfile +import unittest + +SOURCE = Path(__file__).resolve().parents[2] / "tools/reconcile_required_checks.py" +spec = importlib.util.spec_from_file_location("policy", SOURCE) +policy = importlib.util.module_from_spec(spec) +spec.loader.exec_module(policy) + + +class FakeGitHub: + def __init__(self): + self.head = "a" * 40 + self.contexts = ["self-test", "lean-self-test", "template / bar"] + self.workflow = ".github/workflows/ci.yml" + self.workflow_exists = True + self.suite_path = self.workflow + self.app = "github-actions" + self.protection = { + "required_status_checks": {"strict": True, "contexts": ["external"], + "checks": [{"context": "external", "app_id": 77}]}, + "required_pull_request_reviews": {"required_approving_review_count": 1, + "dismiss_stale_reviews": True, + "require_code_owner_reviews": True}, + "enforce_admins": {"enabled": True}, + "restrictions": {"users": ["owner"]}, "required_linear_history": {"enabled": True}} + self.writes = [] + self.reads = [] + + def get(self, path): + self.reads.append(path) + if "/contents/" in path: + if not self.workflow_exists: + raise policy.Missing(path) + return {"type": "file", "sha": "workflow-blob"} + if "/actions/runs?" in path: + return {"workflow_runs": [{"head_sha": self.head, "path": self.suite_path, "check_suite_id": 9}]} + if "/check-runs?" in path: + return {"check_runs": [{"name": c, "app": {"id": 15368, "slug": self.app}, + "check_suite": {"id": 9}} for c in self.contexts]} + if path.endswith("/protection"): + if self.protection is None: + raise policy.Missing(path) + return copy.deepcopy(self.protection) + if "/branches/" in path: + return {"commit": {"sha": self.head}} + return {"default_branch": "main"} + + def write(self, path, method, payload): + self.writes.append((path, method, copy.deepcopy(payload))) + + +class ProtectionTest(unittest.TestCase): + def setUp(self): + self.tmp = tempfile.TemporaryDirectory() + self.addCleanup(self.tmp.cleanup) + self.path = Path(self.tmp.name) / "policy.toml" + self.path.write_text('''version = 1 +owner = "Proven-Commons" +[defaults] +minimum_reviews = 1 +enforce_admins = true +dismiss_stale_reviews = true +required_conversation_resolution = true +workflow = ".github/workflows/ci.yml" +contexts = ["self-test", "lean-self-test", "template / bar"] +[[repository]] +name = "mission" +''') + self.api = FakeGitHub() + + def plan(self): + return policy.plan(self.api, self.path, ["mission"]) + + def test_default_planning_never_writes(self): + result = self.plan() + self.assertEqual(result["repositories"][0]["state"], "ready") + self.assertEqual(self.api.writes, []) + + def test_merge_keeps_other_contexts_app_bindings_and_strict_mode(self): + repo = self.plan()["repositories"][0] + self.assertEqual(len(repo["actions"]), 1) + action = repo["actions"][0] + self.assertEqual(action["method"], "PATCH") + self.assertTrue(action["payload"]["strict"]) + self.assertIn({"context": "external", "app_id": 77}, action["payload"]["checks"]) + self.assertEqual(len(action["payload"]["checks"]), 4) + # No PUT of a reduced protection object can erase unrelated settings. + self.assertTrue(action["path"].endswith("/required_status_checks")) + + def test_missing_workflow_is_deferred(self): + self.api.workflow_exists = False + self.assertEqual(self.plan()["repositories"][0]["state"], "deferred") + self.assertEqual(self.api.writes, []) + + def test_unobserved_required_context_is_deferred(self): + self.api.contexts.remove("template / bar") + self.assertEqual(self.plan()["repositories"][0]["state"], "deferred") + + def test_same_context_from_other_workflow_is_not_accepted(self): + self.api.suite_path = ".github/workflows/other.yml" + self.assertEqual(self.plan()["repositories"][0]["state"], "deferred") + + def test_same_context_from_other_app_is_not_accepted(self): + self.api.app = "external-service" + self.assertEqual(self.plan()["repositories"][0]["state"], "deferred") + + def test_existing_binding_to_another_app_is_not_replaced(self): + self.api.protection["required_status_checks"]["checks"].append({"context": "self-test", "app_id": 8}) + repo = self.plan()["repositories"][0] + self.assertEqual(repo["state"], "deferred") + self.assertEqual(repo["actions"], []) + + def test_apply_rejects_changed_head_before_any_write(self): + saved = self.plan() + self.api.head = "b" * 40 + with self.assertRaisesRegex(ValueError, "changed"): + policy.apply_saved(self.api, self.path, saved) + self.assertEqual(self.api.writes, []) + + def test_apply_rejects_changed_settings_before_any_write(self): + saved = self.plan() + self.api.protection["required_status_checks"]["contexts"].append("new-external") + with self.assertRaisesRegex(ValueError, "changed"): + policy.apply_saved(self.api, self.path, saved) + self.assertEqual(self.api.writes, []) + + def test_tampered_plan_cannot_supply_writes(self): + saved = self.plan() + saved["repositories"][0]["actions"][0]["payload"]["checks"] = [] + with self.assertRaisesRegex(ValueError, "changed"): + policy.apply_saved(self.api, self.path, saved) + self.assertEqual(self.api.writes, []) + + def test_policy_change_invalidates_saved_plan(self): + saved = self.plan() + self.path.write_text(self.path.read_text() + "\n# changed policy\n") + with self.assertRaisesRegex(ValueError, "changed"): + policy.apply_saved(self.api, self.path, saved) + self.assertEqual(self.api.writes, []) + + def test_explicit_apply_writes_only_the_planned_narrow_patch(self): + saved = self.plan() + before = copy.deepcopy(self.api.protection) + policy.apply_saved(self.api, self.path, saved) + self.assertEqual(len(self.api.writes), 1) + self.assertEqual(self.api.writes[0][1], "PATCH") + self.assertEqual(self.api.protection, before) + self.assertNotIn("required_pull_request_reviews", self.api.writes[0][2]) + + def test_main_moving_during_apply_aborts_before_write(self): + saved = self.plan() + original_get = self.api.get + branch_reads = 0 + def get(path): + nonlocal branch_reads + if path.endswith("/branches/main"): + branch_reads += 1 + if branch_reads == 2: + self.api.head = "b" * 40 + return original_get(path) + self.api.get = get + with self.assertRaisesRegex(ValueError, "main changed"): + policy.apply_saved(self.api, self.path, saved) + self.assertEqual(self.api.writes, []) + + def test_protection_moving_during_apply_aborts_before_write(self): + saved = self.plan() + original_get = self.api.get + protection_reads = 0 + def get(path): + nonlocal protection_reads + if path.endswith("/protection"): + protection_reads += 1 + if protection_reads == 2: + self.api.protection["required_linear_history"]["enabled"] = False + return original_get(path) + self.api.get = get + with self.assertRaisesRegex(ValueError, "protection changed"): + policy.apply_saved(self.api, self.path, saved) + self.assertEqual(self.api.writes, []) + + def test_already_satisfied_policy_is_unchanged(self): + status = self.api.protection["required_status_checks"] + status["checks"] += [{"context": c, "app_id": 15368} for c in self.api.contexts] + self.assertEqual(self.plan()["repositories"][0]["state"], "unchanged") + + def test_review_minimum_never_lowers_existing_reviews(self): + self.api.protection["required_pull_request_reviews"]["required_approving_review_count"] = 3 + repo = self.plan()["repositories"][0] + self.assertFalse(any(a["path"].endswith("required_pull_request_reviews") for a in repo["actions"])) + + def test_missing_reviews_are_added_without_changing_other_review_options(self): + self.api.protection["required_pull_request_reviews"]["required_approving_review_count"] = 0 + actions = self.plan()["repositories"][0]["actions"] + review = next(a for a in actions if a["path"].endswith("required_pull_request_reviews")) + self.assertEqual(review["payload"], {"required_approving_review_count": 1}) + + def test_unprotected_branch_has_an_explicit_initial_protection_plan(self): + self.api.protection = None + action = self.plan()["repositories"][0]["actions"][0] + self.assertEqual(action["method"], "PUT") + self.assertTrue(action["payload"]["enforce_admins"]) + self.assertEqual(action["payload"]["required_pull_request_reviews"]["required_approving_review_count"], 1) + # The first protection must match the org baseline: an approval on an + # older head does not count, and open conversations block a merge. + self.assertIs(action["payload"]["required_pull_request_reviews"]["dismiss_stale_reviews"], True) + self.assertIs(action["payload"]["required_conversation_resolution"], True) + + def test_initial_protection_settings_must_be_booleans(self): + self.api.protection = None + self.path.write_text(self.path.read_text().replace("dismiss_stale_reviews = true", 'dismiss_stale_reviews = "yes"')) + with self.assertRaisesRegex(ValueError, "dismiss_stale_reviews"): + self.plan() + + def test_any_source_check_is_sent_back_as_minus_one(self): + self.api.protection["required_status_checks"]["checks"].append({"context": "legacy", "app_id": None}) + actions = self.plan()["repositories"][0]["actions"] + patch = next(a for a in actions if a["path"].endswith("required_status_checks")) + self.assertIn({"context": "legacy", "app_id": -1}, patch["payload"]["checks"]) + self.assertNotIn(None, [c["app_id"] for c in patch["payload"]["checks"]]) + + def test_workflow_path_that_is_a_directory_is_deferred(self): + real_get = self.api.get + self.api.get = lambda path: [] if "/contents/" in path else real_get(path) + self.assertEqual(self.plan()["repositories"][0]["state"], "deferred") + + def test_missing_repository_is_deferred_not_fatal(self): + real_get = self.api.get + def get(path): + if path == "repos/Proven-Commons/mission": + raise policy.Missing(path) + return real_get(path) + self.api.get = get + repo = self.plan()["repositories"][0] + self.assertEqual((repo["state"], repo["reason"]), ("deferred", "repository not found")) + + def test_tar_stays_deferred_until_policy_is_reviewed_after_landing(self): + self.path.write_text(self.path.read_text() + '\n[[repository]]\nname = "tar"\ndeferred = "Set B not landed"\n') + result = policy.plan(self.api, self.path, ["tar"]) + self.assertEqual(result["repositories"][0]["state"], "deferred") + self.assertEqual(self.api.reads, []) + with self.assertRaisesRegex(ValueError, "deferred"): + policy.apply_saved(self.api, self.path, result) + self.assertEqual(self.api.writes, []) diff --git a/policy/README.md b/policy/README.md new file mode 100644 index 0000000..37cde2f --- /dev/null +++ b/policy/README.md @@ -0,0 +1,75 @@ +# Required checks on main + +`required-checks.toml` records the intended branch policy. The planner in +`tools/reconcile_required_checks.py` reads GitHub through the authenticated +`gh` CLI. Its default mode cannot write settings. + +Mission requires `self-test`, `lean-self-test` and `template / bar` from +`.github/workflows/ci.yml`. Libraries require `bar / bar` only after their +bar workflow and that GitHub Actions check exist on the exact main commit. +The planner matches checks to the named workflow's run through its check +suite, so an identically named check from another workflow does not qualify. +The policy also records one approving review and administrator enforcement. +Existing stronger review counts and unrelated protection settings stay intact. +When main has no protection yet, the first protection also dismisses stale +approvals and requires conversations to be resolved, as every protected main +in the org does, so an approval counts only on the current head (MERGE-SOP). + +Tar is explicitly deferred until Set B and its bar workflow land. Review a +policy change removing that deferral after the landing. Repositories without +a bar workflow stay deferred; this manifest does not create their workflows. +No live protection has been changed by adding this policy. + +## Review and apply a plan + +Run from the mission checkout: + +```sh +python3 tools/reconcile_required_checks.py --repo mission --write-plan /tmp/mission-protection.json +``` + +Inspect the saved JSON, including the main SHA, workflow blob, current +protection and proposed API requests. A read-only plan prints `ready`, +`unchanged` or `deferred` for each repository. Exit 2 means something was +deferred; exit 1 means a read or validation failed. Select repositories with +repeated `--repo` arguments. An apply plan may contain no deferred repository. + +After a maintainer reviews the exact plan, an administrator can apply it: + +```sh +python3 tools/reconcile_required_checks.py --apply-plan /tmp/mission-protection.json +``` + +This explicit command is the only write path. It rebuilds the plan from the +current policy and live GitHub state before writing. A changed SHA, workflow, +policy, protection or action payload rejects the saved plan. Edited JSON +cannot supply arbitrary API requests. Existing protection uses narrow PATCH +requests for required checks and the review minimum, and POST only to enable +administrator enforcement. A new protection uses PUT only after GitHub has +confirmed that main exists and no protection currently exists, and that PUT +sets the org baseline: stale approvals dismissed and conversations resolved. + +GitHub provides no atomic compare-and-swap for these settings. The script +checks the head again before each write and protection again before each +repository. A network failure or concurrent edit can stop a partial apply. +Inspect live state and make a fresh plan before continuing. Do not rerun an +old plan blindly. No command removes protection or deletes an existing check. + +## Pin a reviewed bar release + +After a release commit on mission's default branch passes the required CI +and independent review, a maintainer chooses its exact SHA. Create a release +tag only through that reviewed release step. Each library should pin both +its reusable workflow reference and `bar-ref` to the same reviewed tag or +commit. This script creates no tags, releases or library edits. + +## Verify changes + +```sh +python3 -m unittest discover -s bar/tests -v +``` + +Tests mock GitHub writes and cover dry runs, missing workflows, false check +names, application bindings, existing settings, plan tampering, stale main +SHAs, concurrent changes and explicit application. The mission self-test job +already discovers this test file. diff --git a/policy/required-checks.toml b/policy/required-checks.toml new file mode 100644 index 0000000..6eca64a --- /dev/null +++ b/policy/required-checks.toml @@ -0,0 +1,131 @@ +version = 1 +owner = "Proven-Commons" + +# Each library waits until its bar workflow and check run exist on main. +# No setting is applied by reading this file or running the planner. +[defaults] +minimum_reviews = 1 +enforce_admins = true +# Used only when main has no protection yet, so the first protection +# matches the org baseline and MERGE-SOP's current-head approval. +dismiss_stale_reviews = true +required_conversation_resolution = true +workflow = ".github/workflows/bar.yml" +contexts = ["bar / bar"] + +[[repository]] +name = "mission" +workflow = ".github/workflows/ci.yml" +contexts = ["self-test", "lean-self-test", "template / bar"] + +[[repository]] +name = "tar" +deferred = "Wait for Set B PRs #4-6 and its bar workflow to land on main." + +[[repository]] +name = "bencode" + +[[repository]] +name = "codec" + +[[repository]] +name = "csv" + +[[repository]] +name = "dns" + +[[repository]] +name = "ip" + +[[repository]] +name = "json" + +[[repository]] +name = "pep440" + +[[repository]] +name = "png" + +[[repository]] +name = "protobuf" + +[[repository]] +name = "rfc" + +[[repository]] +name = "semver" + +[[repository]] +name = "toml" + +[[repository]] +name = "utf" + +[[repository]] +name = "uuid" + +[[repository]] +name = "bech32" + +[[repository]] +name = "cbor" + +[[repository]] +name = "hostname" + +[[repository]] +name = "http" + +[[repository]] +name = "punycode" + +[[repository]] +name = "cedar" + +[[repository]] +name = "hpack" + +[[repository]] +name = "der" + +[[repository]] +name = "messagepack" + +[[repository]] +name = "zip" + +[[repository]] +name = "redis" + +[[repository]] +name = "websocket" + +[[repository]] +name = "sqlite" + +[[repository]] +name = "path" + +[[repository]] +name = "ssh" + +[[repository]] +name = "xml" + +[[repository]] +name = "clienthello" + +[[repository]] +name = "dhcpv4" + +[[repository]] +name = "ntp" + +[[repository]] +name = "mqtt" + +[[repository]] +name = "jws" + +[[repository]] +name = "uri" diff --git a/tools/reconcile_required_checks.py b/tools/reconcile_required_checks.py new file mode 100644 index 0000000..6a25226 --- /dev/null +++ b/tools/reconcile_required_checks.py @@ -0,0 +1,215 @@ +#!/usr/bin/env python3 +"""Plan branch protections; write only a saved, unchanged plan on explicit request.""" +import argparse +import copy +import hashlib +import json +from pathlib import Path +import re +import subprocess +import tomllib +import urllib.parse + +DEFAULT_POLICY = Path(__file__).resolve().parents[1] / "policy/required-checks.toml" + + +class Missing(Exception): + pass + + +class GitHub: + def request(self, path, method="GET", payload=None): + cmd = ["gh", "api", "--method", method, path] + if payload is not None: + cmd += ["--input", "-"] + out = subprocess.run(cmd, input=json.dumps(payload) if payload is not None else None, + text=True, capture_output=True, check=False) + if out.returncode: + if "HTTP 404" in out.stderr: + raise Missing(path) + raise RuntimeError(out.stderr.strip()) + return json.loads(out.stdout) if out.stdout.strip() else None + + def get(self, path): + return self.request(path) + + def write(self, path, method, payload): + return self.request(path, method, payload) + + +def collection(api, path, field): + found = [] + for page in range(1, 1001): + sep = "&" if "?" in path else "?" + batch = api.get(f"{path}{sep}per_page=100&page={page}")[field] + found.extend(batch) + if len(batch) < 100: + return found + raise ValueError("GitHub pagination exceeded the safety limit") + + +def load_policy(path): + raw = Path(path).read_bytes() + data = tomllib.loads(raw.decode()) + if data.get("version") != 1 or not re.fullmatch(r"[A-Za-z0-9_.-]+", data.get("owner", "")): + raise ValueError("invalid policy version or owner") + names = [p["name"] for p in data["repository"]] + if len(set(names)) != len(names) or any(not re.fullmatch(r"[A-Za-z0-9_.-]+", n) for n in names): + raise ValueError("invalid or duplicate repository name") + return data, hashlib.sha256(raw).hexdigest() + + +def plan_repo(api, owner, desired): + name = desired["name"] + result = {"repository": f"{owner}/{name}", "actions": []} + if desired.get("deferred"): + return dict(result, state="deferred", reason=desired["deferred"]) + base = f"repos/{owner}/{name}" + try: + info = api.get(base) + except Missing: + return dict(result, state="deferred", reason="repository not found") + branch = info["default_branch"] + if branch != "main": + return dict(result, state="deferred", reason=f"default branch is {branch}, not main") + branch_path = f"{base}/branches/{urllib.parse.quote(branch, safe='')}" + head = api.get(branch_path)["commit"]["sha"] + result["head"] = head + workflow = desired["workflow"] + if not workflow.startswith(".github/workflows/") or not workflow.endswith((".yml", ".yaml")): + raise ValueError("workflow must name a workflow file") + try: + source = api.get(f"{base}/contents/{workflow}?ref={head}") + except Missing: + return dict(result, state="deferred", reason=f"{workflow} is absent on main") + if not isinstance(source, dict) or source.get("type") != "file": + return dict(result, state="deferred", reason="workflow path is not a file") + result["workflow_blob"] = source["sha"] + runs = collection(api, f"{base}/actions/runs?head_sha={head}", "workflow_runs") + suites = {r["check_suite_id"] for r in runs + if r.get("head_sha") == head and r.get("path", "").split("@", 1)[0] == workflow} + checks = collection(api, f"{base}/commits/{head}/check-runs", "check_runs") + required = [] + for context in desired["contexts"]: + observed = [c for c in checks if c.get("name") == context + and (c.get("app") or {}).get("slug") == "github-actions" + and (c.get("check_suite") or {}).get("id") in suites] + app_ids = {(c.get("app") or {}).get("id") for c in observed} + if len(app_ids) != 1 or None in app_ids: + return dict(result, state="deferred", reason=f"{context!r} is not an observed check of {workflow} on main") + required.append({"context": context, "app_id": app_ids.pop()}) + if not required: + raise ValueError("policy must specify at least one required check") + minimum = desired["minimum_reviews"] + if isinstance(minimum, bool) or not isinstance(minimum, int) or not 1 <= minimum <= 6: + raise ValueError("minimum_reviews must be between 1 and 6") + try: + protection = api.get(branch_path + "/protection") + except Missing: + protection = None + for key in ("enforce_admins", "dismiss_stale_reviews", "required_conversation_resolution"): + if not isinstance(desired.get(key), bool): + raise ValueError(f"{key} must be true or false") + result["protection"] = protection + actions = result["actions"] + if protection is None: + actions.append({"method": "PUT", "path": branch_path + "/protection", "payload": { + "required_status_checks": {"strict": False, "checks": required}, + "enforce_admins": bool(desired["enforce_admins"]), + "required_pull_request_reviews": {"required_approving_review_count": minimum, + "dismiss_stale_reviews": desired["dismiss_stale_reviews"]}, + "required_conversation_resolution": desired["required_conversation_resolution"], + "restrictions": None}}) + else: + status = protection.get("required_status_checks") or {} + merged = copy.deepcopy(status.get("checks") or []) + for check in merged: + # GitHub reports a check that any source may set with app_id null; + # it accepts -1 for that on write. + if check.get("app_id") is None: + check["app_id"] = -1 + for context in status.get("contexts") or []: + if not any(c["context"] == context for c in merged): + merged.append({"context": context, "app_id": -1}) + changed = False + for check in required: + existing = [c for c in merged if c["context"] == check["context"]] + if existing and any(c.get("app_id", -1) not in (-1, None, check["app_id"]) for c in existing): + return dict(result, state="deferred", reason=f"{check['context']} already belongs to a different app", actions=[]) + if not existing: + merged.append(check) + changed = True + if changed: + actions.append({"method": "PATCH", "path": branch_path + "/protection/required_status_checks", + "payload": {"strict": bool(status.get("strict", False)), "checks": merged}}) + reviews = protection.get("required_pull_request_reviews") or {} + if reviews.get("required_approving_review_count", 0) < minimum: + actions.append({"method": "PATCH", "path": branch_path + "/protection/required_pull_request_reviews", + "payload": {"required_approving_review_count": minimum}}) + if desired["enforce_admins"] and not (protection.get("enforce_admins") or {}).get("enabled"): + actions.append({"method": "POST", "path": branch_path + "/protection/enforce_admins", "payload": None}) + return dict(result, state="ready" if actions else "unchanged") + + +def plan(api, policy_path, selected): + policy, digest = load_policy(policy_path) + names = [p["name"] for p in policy["repository"]] + if selected and not set(selected) <= set(names): + raise ValueError("selected repository is not in the policy") + repos = [p for p in policy["repository"] if not selected or p["name"] in selected] + return {"version": 1, "policy_sha256": digest, "selected": sorted(selected or names), + "repositories": [plan_repo(api, policy["owner"], dict(policy["defaults"], **p)) for p in repos]} + + +def apply_saved(api, policy_path, saved): + # Recompute all actions from the policy, never execute payloads supplied + # by an edited plan. No writes occur if any read, SHA or setting drifted. + if not isinstance(saved, dict) or not isinstance(saved.get("selected"), list): + raise ValueError("saved plan must contain a repository selection") + current = plan(api, policy_path, saved["selected"]) + if current != saved: + raise ValueError("live state or policy changed; save and review a fresh dry-run plan") + if any(r["state"] == "deferred" for r in current["repositories"]): + raise ValueError("plan contains deferred repositories; select only ready repositories") + for repo in current["repositories"]: + base = "repos/" + repo["repository"] + "/branches/main" + try: + protection = api.get(base + "/protection") + except Missing: + protection = None + if protection != repo["protection"]: + raise ValueError("protection changed during apply; stop and create a fresh plan") + for action in repo["actions"]: + # Recheck the head before each write. GitHub has no atomic + # compare-and-swap for branch protection; see the operator notes. + base = "repos/" + repo["repository"] + "/branches/main" + if api.get(base)["commit"]["sha"] != repo["head"]: + raise ValueError("main changed during apply; stop and create a fresh plan") + api.write(action["path"], action["method"], action["payload"]) + return current + + +def main(): + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("--policy", type=Path, default=DEFAULT_POLICY) + parser.add_argument("--repo", action="append", default=[], help="repository short name; repeat to select") + parser.add_argument("--write-plan", type=Path, help="save the read-only plan for review") + parser.add_argument("--apply-plan", type=Path, help="explicitly apply an unchanged saved plan") + args = parser.parse_args() + if args.apply_plan and (args.repo or args.write_plan): + parser.error("--apply-plan cannot be combined with --repo or --write-plan") + try: + api = GitHub() + result = (apply_saved(api, args.policy, json.loads(args.apply_plan.read_text())) if args.apply_plan + else plan(api, args.policy, args.repo)) + output = json.dumps(result, indent=2) + "\n" + if args.write_plan: + args.write_plan.write_text(output) + print(output, end="") + return 2 if any(r["state"] == "deferred" for r in result["repositories"]) else 0 + except (ValueError, RuntimeError, Missing, OSError, KeyError, TypeError, AttributeError) as e: + parser.exit(1, f"No further changes applied: {e}\n") + + +if __name__ == "__main__": + raise SystemExit(main())