diff --git a/README.md b/README.md index 1d90633..9502ce5 100644 --- a/README.md +++ b/README.md @@ -55,6 +55,28 @@ Both the promoter and independent observer download the exact two raw provider review envelopes and require the immutable parent Kernel to reproduce the candidate's `ALLOW` or `DENY` permit byte for byte; a candidate-authored summary cannot stand in for reduction evidence. +Successor ratification structurally parses the candidate workflow with +Ruby/Psych into an AST-preserving projection before it evaluates authority +semantics. That preserves YAML scalar spellings (including the `on` trigger +key) and rejects duplicate keys, aliases, merge keys, and custom tags. The +previous immutable workflow is the complete parent-owned allowlist: candidate +top-level keys, triggers, permissions, absence of defaults/concurrency, +known-job graph, job execution fields, ordered steps, action SHAs, inputs, +environment, scripts, and artifact paths must all match it exactly. The only +permitted successor differences are the three declared Kernel checkout refs +and the matching `prepare` `KERNEL_SHA`. The approval chain is also modeled +explicitly: its immutable broker checkout and verifier build remain ordered, +the approval-only App action SHA and step ID are exact, and that App token has +one consumer only—the exact-head approval step. Comments, repeated text, or +an unreviewed execution surface cannot constitute authority evidence. +The parent verifier also inventories the complete `.github/workflows` Git tree: +the candidate must retain the same paths and regular-file modes as its +immutable parent. Only `ci.yml` is semantically reviewed under the four +declared Kernel substitutions; every other workflow blob is byte-identical. +Added, deleted, nested, symlinked, gitlink, mode-changed, or altered workflows +fail closed. The authority repository's source-owned gate profile repeats that +inventory before it runs a candidate command, so an extra workflow cannot hide +outside the `ci.yml` semantic exception. Credentialed jobs also bind the pinned token action's live App slug and installation ID before use; the approval broker independently checks exact repository scope and the persisted GitHub review actor. diff --git a/config/autonomous-release-authority.json b/config/autonomous-release-authority.json index c8cf4ed..0a784a0 100644 --- a/config/autonomous-release-authority.json +++ b/config/autonomous-release-authority.json @@ -1,6 +1,6 @@ { "adversarial_corpus_sha256": "a5572c4fcd50bc331043d90e6e97f7cd383baeb278c9d653b173019eeee41d5f", - "gate_profiles_sha256": "56ddccd1f312599a828ca53cb0d78189bba2b75f2346722ef904ccc4314d1720", + "gate_profiles_sha256": "2de164ead3ea1b2680468a781810b6bee64688d9c2ca17399ae6f780d5937bed", "generation": 2, "kernel_sha": "83cc3eeb1cf512bed44b560254b11a342cee5b15", "parent": { diff --git a/config/autonomous-release-gates.json b/config/autonomous-release-gates.json index 594ff27..e292b6d 100644 --- a/config/autonomous-release-gates.json +++ b/config/autonomous-release-gates.json @@ -6,8 +6,6 @@ ["make", "test"] ], "protected_files": { - ".github/workflows/ci.yml": "a4e14a559ed51489d4afa39cceb24fee4a4ad2adf8ceebb3ee3255c9d66717e9", - ".github/workflows/promote-authority.yml": "097460ffa3f2673f7f2d67a45a57bbb963acb231da3131a929a48d84aa6c6c39", "Makefile": "4559934af490cb54f61405c8471fca1e9d287ea5c719ff90192d33c1ec180925", "config/autonomous-release-bootstrap-v1.json": "d5b2012cadd4fbf8e24bfe4f8ed69c620cf9d6481dc9f488852aa855fe1712b6", "config/autonomous-release-control-plane.json": "62332956bdad7554e4b093c8abb95cf984a300c55f10381dee906f1466818992", @@ -17,9 +15,9 @@ "scripts/bootstrap_authority.py": "8ab8500dd56e6bc295e260fec5d30915b79de39e8d9c67bf3837d8b162d66cf9", "scripts/configure_machine_approval_gates.py": "ba5109198c69e2d2a0ce56bef18e43c0a201e2eba0f79c4ef5c7a4a759e4e923", "scripts/observe_authority_promotion.py": "04a4767e4a641fdcca809e6a185a502dcb53cdd6c47519df979a0f08fdcf773a", - "scripts/run_autonomous_release_gates.py": "e9ab0ed2d91923e93df3e7c148f2ea15381cff9ceb6f4360d3b1fd258ae26f31", + "scripts/run_autonomous_release_gates.py": "80a21f602748b39310f74c56e1ca6ccb2466dfab91d7de4ffd8120ad56b40839", "scripts/submit_machine_approval.py": "1a2743342026928c6a346df88ad0864b034155ce9320e6e45a598146398c986f", - "scripts/verify_authority_promotion.py": "4d5f079ffc21ee350799eb93374a7ce95e0926b151a10b2015105642c7622f5b", + "scripts/verify_authority_promotion.py": "3afa53d363fbab363c19e481954895bf5270e16b97c5df1f0919e012ddfce8b2", "scripts/verify_control_plane.py": "78ad97bffd7f065737c6839ea66475a98ca4583dfbedb60673648f86ef893416", "scripts/wait_for_authority_canary.py": "6a6524639590e5dac432f0bc25a135712310a4fa1a87bc2f3aff19cc3f31662f", "scripts/wait_for_authority_suite.py": "b97933597c2b5d9f132a488a3a63a6b3ccec299bbe80ad3d2e085ad549343613", @@ -31,12 +29,34 @@ "tests/test_observe_authority_promotion.py": "3ea6c9824b7c3d20738b2c31d198d618a59a985c688dfa39694713756fb5aadd", "tests/test_configure_machine_approval_gates.py": "645d2acdb754fe77da9e10920e0bc9cf65a421e742c426f540e3c2f5b2a2cf6d", "tests/test_submit_machine_approval.py": "0a2887be1d996f83d66e6b2ce29da4915d21c16d2badb614e0bc016d74e8d38d", - "tests/test_verify_authority_promotion.py": "9c5e4e8732984faa2375738df908273614f40dcb49d127a0f96a8be3acf503af", + "tests/test_run_autonomous_release_gates.py": "2d031593af88385e0813a0b5e0cb73114131791065fecf59379b6c5fb68f4fdf", + "tests/test_verify_authority_promotion.py": "2e80ff9bf49abeabfc422fe6df22586c5f81a1e6696cd9113fdb3c2db0bb0291", "tests/test_verify_control_plane.py": "36620d9e7486582c9786d2c05479491e731243298338b8fe0e8302c31979ac88", "tests/test_wait_for_authority_canary.py": "6ec843374d9e2adb7063e15baa41b7a9bad2a49ec9c0f6a760ad17f39f449ce2", "tests/test_wait_for_authority_suite.py": "28297e2d191f21a85d46d5abe4c6403663a942e560299e8c0d549576176369b1", "tools/offline-attest/attest.mjs": "21c544e689f472039cbd5fd621fab34b801bdd40b92fb68745948253ce621144", "tools/offline-attest/package-lock.json": "a4b898bac65ba7ccbf1b6871d5ec5c6fdcf83d8d38fa976b52ad11bad29ac135" + }, + "workflow_inventory": { + "root": ".github/workflows", + "files": { + ".github/workflows/ci.yml": { + "mode": "100644", + "semantic": true + }, + ".github/workflows/docs-truth-public.yml": { + "mode": "100644", + "sha256": "5422a1d0f49f9edf711e5c96ca7c98ebb423cbeb71ef9e57c8a935744b547ea3" + }, + ".github/workflows/docs-truth.yml": { + "mode": "100644", + "sha256": "2cc9bb7b1271b554664ce2f9e8a426b98528b7022187eb6feed9714230a0be2d" + }, + ".github/workflows/promote-authority.yml": { + "mode": "100644", + "sha256": "097460ffa3f2673f7f2d67a45a57bbb963acb231da3131a929a48d84aa6c6c39" + } + } } }, "Mindburn-Labs/app-helm-console": { diff --git a/docs/adr/0002-autonomous-release-permit.md b/docs/adr/0002-autonomous-release-permit.md index 2d16013..6669adb 100644 --- a/docs/adr/0002-autonomous-release-permit.md +++ b/docs/adr/0002-autonomous-release-permit.md @@ -70,7 +70,12 @@ no command-discovery or Makefile fallback. The profile names an exact command vector and SHA-256 digests for every build file that can alter those commands; missing, changed, non-regular, or symlinked protected files fail before either model runs. Authority-critical profile changes use the same two-generation -ratification rule as workflow changes. +ratification rule as workflow changes. The authority repository additionally +uses a closed `.github/workflows` inventory before any candidate command: each +allowed path has an exact regular-file mode, `ci.yml` is the sole semantic +entry, and every other workflow has a source-owned SHA-256 digest. An added, +deleted, nested, non-regular, mode-changed, or digest-drifting workflow fails +before a candidate gate command or model review. Every context also binds `config/autonomous-release-authority.json`. The manifest names a monotonically increasing generation, the exact Kernel SHA, @@ -79,6 +84,30 @@ the immediately preceding workflow generation. Generation 1 is the explicit bootstrap; every later generation must name exactly generation N-1 and cannot name its own workflow SHA as its parent. +The parent-generation promotion verifier structurally parses the candidate +workflow with Ruby/Psych into an AST-preserving projection before evaluating +it. It rejects malformed YAML, duplicate mapping keys, aliases, merge keys, +and custom tags while preserving scalar spellings, including the `on` trigger +key that YAML 1.1 loaders can otherwise coerce. The previous immutable +workflow is the complete parent-owned allowlist: the candidate must match its +exact top-level keys and triggers, top-level permissions and absence of +defaults/concurrency, all known job IDs and dependencies, every job execution +field, and every ordered step, action SHA, input, environment value, shell +script, and artifact path. The only permitted successor substitutions are the +three declared Kernel checkout refs and the matching `prepare` `KERNEL_SHA`. +It also reads the complete workflow directory directly from the immutable +parent and candidate Git trees. Their path and mode inventories must be +identical; only the regular `ci.yml` blob proceeds to the semantic check, while +every other workflow blob must be byte-identical to the parent. Thus an added, +deleted, nested, symlinked, gitlink, mode-changed, or otherwise altered +workflow cannot create an authority surface outside the `ci.yml` exception. +The machine-approval chain is part of that closed-world profile: the immutable +approval-broker checkout and source-owned verifier build remain ordered, the +approval-only App action SHA and step ID remain exact, and its minted token has +exactly one consumer, the exact-head approval step. Comments, repeated +substrings, alternate steps, or any new execution surface do not constitute +authority evidence. + The existing `local-validation` check name and push-to-`main` validation remain intact while the permit is evaluated. The permit records GitHub's actual `workflow_ref` and immutable `workflow_sha`; it must not assert a task-branch diff --git a/docs/runbooks/autonomous-release-bootstrap.md b/docs/runbooks/autonomous-release-bootstrap.md index d8191c9..fca16ff 100644 --- a/docs/runbooks/autonomous-release-bootstrap.md +++ b/docs/runbooks/autonomous-release-bootstrap.md @@ -26,6 +26,22 @@ The local credential may execute those inputs but cannot make a DENY acceptable. providers. - `release-permit-verify` was built from the exact Kernel SHA named by the generation-1 authority manifest. +- The candidate `.github/workflows/ci.yml` passes the parent verifier's + Ruby/Psych closed-world check. Its AST-preserving parse rejects duplicate + keys, aliases, merge keys, and custom tags without coercing the `on` trigger + key. It must match the previous immutable workflow's complete execution + profile: top-level triggers, permissions, and no defaults/concurrency; all + known jobs and dependencies; every execution field; and every ordered step, + action SHA, input, environment value, shell script, and artifact path. Only + the three declared Kernel checkout refs and matching `prepare` `KERNEL_SHA` + may change. The ordered approval-broker checkout/build and sole + approval-step use of the isolated App token are included in that profile. + The verifier also compares the full parent and candidate + `.github/workflows` Git-tree inventories: paths and regular modes are exact, + `ci.yml` is the only semantic exception, and every other workflow blob must + be byte-identical. Added, deleted, nested, symlinked, gitlink, mode-changed, + or altered workflows fail. Comments and repeated text do not satisfy this + check. - The permit, Sigstore bundle, and trusted context were downloaded from the same GitHub Actions run. - The output directory does not exist. Evidence directories are immutable. diff --git a/scripts/run_autonomous_release_gates.py b/scripts/run_autonomous_release_gates.py index 3f76daa..35ee7a2 100644 --- a/scripts/run_autonomous_release_gates.py +++ b/scripts/run_autonomous_release_gates.py @@ -9,6 +9,7 @@ import os from pathlib import Path import re +import stat import subprocess import sys from typing import Any @@ -67,6 +68,55 @@ def validate_protected_path(value: Any, *, label: str) -> str: return value +def validate_workflow_inventory(value: Any, *, label: str) -> dict[str, Any]: + if not isinstance(value, dict): + raise GateProfileError(f"{label} must be an object") + require_exact_keys(value, {"root", "files"}, label) + root = validate_protected_path(value["root"], label=f"{label} root") + if root != ".github/workflows": + raise GateProfileError(f"{label} root must be .github/workflows") + files = value["files"] + if not isinstance(files, dict) or not files or len(files) > 32: + raise GateProfileError(f"{label} must declare 1-32 workflow files") + + validated_files: dict[str, dict[str, Any]] = {} + semantic_paths: list[str] = [] + prefix = f"{root}/" + for path, entry in files.items(): + validated_path = validate_protected_path(path, label=f"{label} workflow path") + if not validated_path.startswith(prefix): + raise GateProfileError(f"{label} workflow path escapes its root: {validated_path}") + if not isinstance(entry, dict): + raise GateProfileError(f"{label} entry for {validated_path} must be an object") + mode = entry.get("mode") + if mode != "100644": + raise GateProfileError( + f"{label} entry for {validated_path} must declare regular mode 100644", + ) + entry_keys = set(entry) + if entry_keys == {"mode", "semantic"}: + if entry["semantic"] is not True: + raise GateProfileError( + f"{label} semantic entry for {validated_path} must be true", + ) + validated_files[validated_path] = {"mode": mode, "semantic": True} + semantic_paths.append(validated_path) + elif entry_keys == {"mode", "sha256"}: + digest = entry["sha256"] + if not isinstance(digest, str) or not SHA256_PATTERN.fullmatch(digest): + raise GateProfileError( + f"{label} digest for {validated_path!r} is invalid", + ) + validated_files[validated_path] = {"mode": mode, "sha256": digest} + else: + raise GateProfileError( + f"{label} entry for {validated_path} must be semantic or digest-locked", + ) + if semantic_paths != [f"{root}/ci.yml"]: + raise GateProfileError(f"{label} may make only {root}/ci.yml semantic") + return {"root": root, "files": validated_files} + + def load_profiles(path: Path) -> dict[str, dict[str, Any]]: document = parse_json_strict(path) if not isinstance(document, dict): @@ -83,11 +133,15 @@ def load_profiles(path: Path) -> dict[str, dict[str, Any]]: raise GateProfileError(f"invalid profile repository: {repository!r}") if not isinstance(profile, dict): raise GateProfileError(f"profile {repository} must be an object") - require_exact_keys( - profile, - {"commands", "protected_files"}, - f"profile {repository}", - ) + profile_keys = set(profile) + required_keys = {"commands", "protected_files"} + permitted_keys = required_keys | {"workflow_inventory"} + if profile_keys != required_keys and profile_keys != permitted_keys: + raise GateProfileError( + f"profile {repository} keys invalid; " + f"missing={sorted(required_keys - profile_keys)}, " + f"unexpected={sorted(profile_keys - permitted_keys)}", + ) commands = profile["commands"] if not isinstance(commands, list) or not commands or len(commands) > 32: raise GateProfileError(f"profile {repository} must contain 1-32 commands") @@ -130,9 +184,16 @@ def load_profiles(path: Path) -> dict[str, dict[str, Any]]: f"profile {repository} protected digest for {validated_path!r} is invalid", ) validated_protected_files[validated_path] = digest + workflow_inventory = None + if "workflow_inventory" in profile: + workflow_inventory = validate_workflow_inventory( + profile["workflow_inventory"], + label=f"profile {repository} workflow inventory", + ) profiles[repository] = { "commands": validated_commands, "protected_files": validated_protected_files, + "workflow_inventory": workflow_inventory, } return profiles @@ -152,6 +213,66 @@ def verify_protected_files(target: Path, protected_files: dict[str, str]) -> Non ) +def verify_workflow_inventory(target: Path, inventory: dict[str, Any]) -> None: + root = inventory["root"] + workflow_root = target.joinpath(*root.split("/")) + try: + root_status = os.lstat(workflow_root) + except OSError as exc: + raise GateProfileError(f"workflow inventory root is unavailable: {root}") from exc + if not stat.S_ISDIR(root_status.st_mode): + raise GateProfileError(f"workflow inventory root is missing or not a directory: {root}") + + observed: dict[str, tuple[int, str]] = {} + for directory, directories, filenames in os.walk(workflow_root, followlinks=False): + current = Path(directory) + for name in directories: + candidate = current / name + if stat.S_ISLNK(os.lstat(candidate).st_mode): + raise GateProfileError( + "workflow inventory contains a symlinked directory: " + f"{candidate.relative_to(target).as_posix()}", + ) + for name in filenames: + candidate = current / name + relative_path = candidate.relative_to(target).as_posix() + try: + candidate_status = os.lstat(candidate) + except OSError as exc: + raise GateProfileError( + f"workflow inventory file is unavailable: {relative_path}", + ) from exc + if not stat.S_ISREG(candidate_status.st_mode): + raise GateProfileError( + f"workflow inventory file is not regular: {relative_path}", + ) + observed[relative_path] = ( + candidate_status.st_mode, + hashlib.sha256(candidate.read_bytes()).hexdigest(), + ) + + expected_files = inventory["files"] + expected_paths = set(expected_files) + observed_paths = set(observed) + if observed_paths != expected_paths: + raise GateProfileError( + "workflow inventory differs from its source-owned profile; " + f"missing={sorted(expected_paths - observed_paths)}, " + f"unexpected={sorted(observed_paths - expected_paths)}", + ) + for path, expected in expected_files.items(): + observed_mode, observed_digest = observed[path] + if f"{observed_mode:06o}" != expected["mode"]: + raise GateProfileError( + f"workflow inventory file mode changed before its source-owned profile: {path}", + ) + expected_digest = expected.get("sha256") + if expected_digest is not None and observed_digest != expected_digest: + raise GateProfileError( + f"workflow inventory file changed before its source-owned profile: {path}", + ) + + def resolve_commands( repository: str, target: Path, @@ -164,6 +285,8 @@ def resolve_commands( "repository has no immutable source-owned gate profile; Makefile fallback is forbidden", ) profile = profiles[repository] + if profile["workflow_inventory"] is not None: + verify_workflow_inventory(target, profile["workflow_inventory"]) verify_protected_files(target, profile["protected_files"]) return "explicit", profile["commands"] diff --git a/scripts/verify_authority_promotion.py b/scripts/verify_authority_promotion.py index abd5940..82628b9 100644 --- a/scripts/verify_authority_promotion.py +++ b/scripts/verify_authority_promotion.py @@ -4,6 +4,7 @@ from __future__ import annotations import argparse +import copy import hashlib import json from pathlib import Path @@ -45,6 +46,166 @@ "reasons", ) +PARENT_WORKFLOW_DIRECTORY = ".github/workflows" +PARENT_WORKFLOW_PATH = f"{PARENT_WORKFLOW_DIRECTORY}/ci.yml" +WORKFLOW_REGULAR_MODE = "100644" + +# The parent checkout is the trusted authority. Its complete workflow is the +# allowlist: the candidate may differ only at the four bindings that move the +# Kernel verifier to the candidate authority generation. The following static +# shape makes the authority surface explicit and rejects a parent checkout that +# has drifted from the reviewed protocol without this verifier being updated. +AUTHORITY_TOP_LEVEL_KEYS = ("name", "on", "permissions", "jobs") +AUTHORITY_TRIGGERS: dict[str, Any] = { + "pull_request": { + "types": ["opened", "reopened", "synchronize", "ready_for_review", "edited"] + }, + "push": {"branches": ["main"]}, +} +AUTHORITY_JOB_IDS = ( + "local-validation", + "workflow-provenance", + "repository-gates", + "prepare", + "model-review", + "permit", + "machine-approval", +) +AUTHORITY_JOB_NEEDS: dict[str, Any] = { + "local-validation": None, + "workflow-provenance": None, + "repository-gates": None, + "prepare": "repository-gates", + "model-review": "prepare", + "permit": ["repository-gates", "prepare", "model-review"], + "machine-approval": "permit", +} + +CHECKOUT_ACTION = "actions/checkout@11bd71901bbe5b1630ceea73d27597364c9af683" +KERNEL_REPOSITORY = "Mindburn-Labs/helm-ai-kernel" +PREPARE_KERNEL_STEP = "Checkout pinned Kernel verifier source for isolated review" +PERMIT_KERNEL_STEP = "Checkout pinned Kernel verifier" +MACHINE_APPROVAL_KERNEL_STEP = "Checkout pinned Kernel verifier" +PREPARE_BUNDLE_STEP = "Prepare commit-bound review bundle" +KERNEL_BINDINGS = ( + ("prepare", PREPARE_KERNEL_STEP, "with", "ref"), + ("permit", PERMIT_KERNEL_STEP, "with", "ref"), + ("machine-approval", MACHINE_APPROVAL_KERNEL_STEP, "with", "ref"), + ("prepare", PREPARE_BUNDLE_STEP, "env", "KERNEL_SHA"), +) + +APPROVER_TOKEN_STEP = "Mint isolated approval-only App token" +APPROVER_TOKEN_ACTION = ( + "actions/create-github-app-token@bcd2ba49218906704ab6c1aa796996da409d3eb1" +) +APPROVER_TOKEN_ID = "approver-token" +APPROVER_TOKEN_REFERENCE = "${{ steps.approver-token.outputs.token }}" +APPROVAL_STEP = "Approve only the exact signed head" +IMMUTABLE_WORKFLOW_REF = "${{ github.workflow_sha }}" +MACHINE_APPROVAL_VERIFIER_BUILD = ( + 'go build -trimpath -o "$GITHUB_WORKSPACE/release-permit-verify" ' + "./cmd/release-permit-verify" +) +MACHINE_APPROVAL_STEP_NAMES = ( + "Download signed ALLOW permit", + "Download permit context", + "Checkout immutable approval broker", + "Checkout pinned Kernel verifier", + "Set up pinned Kernel toolchain", + "Build source-owned permit verifier", + "Mint isolated approval-only App token", + "Bind exact approval App identity", + "Approve only the exact signed head", + "Retain approval receipt", +) + +# Psych is already a repository prerequisite. It is used here rather than a +# permissive YAML loader so comments, duplicate keys, aliases, merge keys, or +# YAML 1.1 coercion (notably on -> true) cannot manufacture an authority +# workflow that looks equivalent only after parsing. The AST projection keeps +# scalar spellings intact while rejecting non-standard tags. +STRICT_WORKFLOW_YAML_PARSER = r''' +require "json" +require "psych" + +def reject_unsafe_yaml(node) + if node.respond_to?(:anchor) && node.anchor + raise "YAML aliases or anchors are not allowed" + end + + if node.respond_to?(:tag) && node.tag + raise "YAML tags are not allowed" + end + + case node + when Psych::Nodes::Scalar + nil + when Psych::Nodes::Alias + raise "YAML aliases or anchors are not allowed" + when Psych::Nodes::Mapping + unless node.children.length.even? + raise "YAML mapping has an incomplete key/value pair" + end + keys = {} + node.children.each_slice(2) do |key, value| + unless key.is_a?(Psych::Nodes::Scalar) + raise "YAML mapping keys must be scalars" + end + key_name = key.value + if key_name == "<<" + raise "YAML merge keys are not allowed" + end + if keys.key?(key_name) + raise "duplicate YAML key: #{key_name}" + end + keys[key_name] = true + reject_unsafe_yaml(key) + reject_unsafe_yaml(value) + end + when Psych::Nodes::Sequence, Psych::Nodes::Document + node.children.each { |child| reject_unsafe_yaml(child) } + end +end + +def project_yaml(node) + case node + when Psych::Nodes::Scalar + node.value + when Psych::Nodes::Sequence + node.children.map { |child| project_yaml(child) } + when Psych::Nodes::Mapping + object = {} + node.children.each_slice(2) do |key, value| + object[key.value] = project_yaml(value) + end + object + else + raise "unsupported YAML node: #{node.class}" + end +end + +begin + source = STDIN.read + stream = Psych.parse_stream(source) + unless stream.children.length == 1 + raise "YAML document must contain exactly one document" + end + document_node = stream.children.first + unless document_node.is_a?(Psych::Nodes::Document) && document_node.children.length == 1 + raise "YAML document must contain exactly one root node" + end + reject_unsafe_yaml(document_node) + document = project_yaml(document_node.children.first) + unless document.is_a?(Hash) + raise "YAML document must be an object" + end + STDOUT.write(JSON.generate(document)) +rescue StandardError => error + warn error.message + exit 1 +end +''' + def load_json_bytes(content: bytes, *, label: str) -> Any: try: @@ -54,6 +215,320 @@ def load_json_bytes(content: bytes, *, label: str) -> Any: return parse_json_strict(text, label=label) +def require_mapping(value: Any, *, label: str) -> dict[str, Any]: + if not isinstance(value, dict): + raise PermitInputError(f"{label} must be an object") + return value + + +def parse_strict_workflow_yaml( + content: str, + *, + label: str = "candidate workflow", +) -> dict[str, Any]: + try: + process = subprocess.run( + ["ruby", "-e", STRICT_WORKFLOW_YAML_PARSER], + input=content.encode("utf-8"), + check=False, + stdout=subprocess.PIPE, + stderr=subprocess.PIPE, + ) + except OSError as exc: + raise PermitInputError( + f"{label} YAML parser is unavailable: {exc}" + ) from exc + if process.returncode != 0: + detail = process.stderr.decode("utf-8", errors="replace").strip() + raise PermitInputError(f"{label} YAML rejected: {detail}") + parsed = load_json_bytes(process.stdout, label=f"{label} YAML") + return require_mapping(parsed, label=label) + + +def find_named_step(job: dict[str, Any], *, name: str, label: str) -> dict[str, Any]: + steps = job.get("steps") + if not isinstance(steps, list): + raise PermitInputError(f"{label} steps must be an array") + matches = [ + step + for step in steps + if isinstance(step, dict) and step.get("name") == name + ] + if len(matches) != 1: + raise PermitInputError( + f"{label} must contain exactly one {name!r} step" + ) + return matches[0] + + +def require_ordered_keys( + value: dict[str, Any], + *, + expected: tuple[str, ...], + label: str, +) -> None: + if tuple(value) != expected: + raise PermitInputError( + f"{label} keys must exactly match the parent-owned authority contract" + ) + + +def validate_authority_workflow_shape( + workflow: dict[str, Any], + *, + label: str, +) -> None: + require_ordered_keys( + workflow, + expected=AUTHORITY_TOP_LEVEL_KEYS, + label=label, + ) + if workflow["on"] != AUTHORITY_TRIGGERS: + raise PermitInputError(f"{label} triggers differ from the authority contract") + if workflow["permissions"] != {}: + raise PermitInputError( + f"{label} top-level permissions differ from the authority contract" + ) + + jobs = require_mapping(workflow["jobs"], label=f"{label} jobs") + require_ordered_keys(jobs, expected=AUTHORITY_JOB_IDS, label=f"{label} jobs") + for job_id, expected_needs in AUTHORITY_JOB_NEEDS.items(): + job = require_mapping(jobs[job_id], label=f"{label} {job_id} job") + observed_needs = job.get("needs") + if expected_needs is None: + if "needs" in job: + raise PermitInputError( + f"{label} {job_id} job must not declare dependencies" + ) + elif observed_needs != expected_needs: + raise PermitInputError( + f"{label} {job_id} dependency graph differs from the authority contract" + ) + + +def find_value_paths(value: Any, *, target: str, path: str) -> list[str]: + if isinstance(value, dict): + paths: list[str] = [] + for key, child in value.items(): + paths.extend(find_value_paths(child, target=target, path=f"{path}.{key}")) + return paths + if isinstance(value, list): + paths = [] + for index, child in enumerate(value): + paths.extend(find_value_paths(child, target=target, path=f"{path}[{index}]")) + return paths + return [path] if value == target else [] + + +def validate_machine_approval_chain( + workflow: dict[str, Any], + *, + label: str, +) -> None: + jobs = require_mapping(workflow.get("jobs"), label=f"{label} jobs") + machine_approval = require_mapping( + jobs.get("machine-approval"), + label=f"{label} machine-approval job", + ) + steps = machine_approval.get("steps") + if not isinstance(steps, list): + raise PermitInputError(f"{label} machine-approval steps must be an array") + if tuple( + step.get("name") if isinstance(step, dict) else None for step in steps + ) != MACHINE_APPROVAL_STEP_NAMES: + raise PermitInputError( + f"{label} machine-approval steps differ from the authority contract" + ) + + broker_step = find_named_step( + machine_approval, + name="Checkout immutable approval broker", + label=f"{label} machine-approval job", + ) + broker_checkout = require_mapping( + broker_step.get("with"), + label=f"{label} immutable approval broker checkout", + ) + if ( + broker_step.get("uses") != CHECKOUT_ACTION + or broker_checkout.get("repository") != "Mindburn-Labs/.github" + or broker_checkout.get("ref") != IMMUTABLE_WORKFLOW_REF + or broker_checkout.get("persist-credentials") != "false" + or broker_checkout.get("path") != "policy" + ): + raise PermitInputError( + f"{label} immutable approval broker checkout differs from the authority contract" + ) + + kernel_step = find_named_step( + machine_approval, + name=MACHINE_APPROVAL_KERNEL_STEP, + label=f"{label} machine-approval job", + ) + kernel_checkout = require_mapping( + kernel_step.get("with"), + label=f"{label} machine-approval Kernel checkout", + ) + if ( + kernel_step.get("uses") != CHECKOUT_ACTION + or kernel_checkout.get("repository") != KERNEL_REPOSITORY + or kernel_checkout.get("persist-credentials") != "false" + or kernel_checkout.get("path") != "kernel" + ): + raise PermitInputError( + f"{label} machine-approval Kernel checkout differs from the authority contract" + ) + require_sha( + kernel_checkout.get("ref"), + label=f"{label} machine-approval Kernel checkout ref", + length=40, + ) + + verifier_build = find_named_step( + machine_approval, + name="Build source-owned permit verifier", + label=f"{label} machine-approval job", + ) + if ( + verifier_build.get("working-directory") != "kernel/core" + or verifier_build.get("run") != MACHINE_APPROVAL_VERIFIER_BUILD + ): + raise PermitInputError( + f"{label} machine-approval verifier build differs from the authority contract" + ) + + token_step = find_named_step( + machine_approval, + name=APPROVER_TOKEN_STEP, + label=f"{label} machine-approval job", + ) + if token_step.get("id") != APPROVER_TOKEN_ID: + raise PermitInputError(f"{label} approval token step ID is not exact") + if token_step.get("uses") != APPROVER_TOKEN_ACTION: + raise PermitInputError(f"{label} approval token action SHA is not exact") + + approval_step = find_named_step( + machine_approval, + name=APPROVAL_STEP, + label=f"{label} machine-approval job", + ) + approval_environment = require_mapping( + approval_step.get("env"), + label=f"{label} exact-head approval environment", + ) + if ( + approval_environment.get("HELM_AUTHORITY_APPROVER_TOKEN") + != APPROVER_TOKEN_REFERENCE + ): + raise PermitInputError( + f"{label} exact-head approval does not consume the isolated App token" + ) + token_paths = find_value_paths( + workflow, + target=APPROVER_TOKEN_REFERENCE, + path="workflow", + ) + expected_token_path = ( + "workflow.jobs.machine-approval.steps[8].env." + "HELM_AUTHORITY_APPROVER_TOKEN" + ) + if token_paths != [expected_token_path]: + raise PermitInputError( + f"{label} isolated App token must have exactly one approval-step consumer" + ) + + +def expected_candidate_workflow(kernel_sha: str) -> dict[str, Any]: + require_sha(kernel_sha, label="candidate authority kernel_sha", length=40) + try: + parent_text = parent_workflow_bytes().decode("utf-8") + except UnicodeDecodeError as exc: + raise PermitInputError( + f"parent authority workflow is invalid UTF-8: {exc}" + ) from exc + parent = parse_strict_workflow_yaml(parent_text, label="parent authority workflow") + validate_authority_workflow_shape(parent, label="parent authority workflow") + validate_machine_approval_chain(parent, label="parent authority workflow") + + expected = copy.deepcopy(parent) + jobs = require_mapping(expected["jobs"], label="parent authority workflow jobs") + for job_id, step_name, section_name, field_name in KERNEL_BINDINGS: + job = require_mapping( + jobs[job_id], + label=f"parent authority workflow {job_id} job", + ) + step = find_named_step( + job, + name=step_name, + label=f"parent authority workflow {job_id} job", + ) + section = require_mapping( + step.get(section_name), + label=f"parent authority workflow {job_id} {step_name} {section_name}", + ) + baseline_value = section.get(field_name) + require_sha( + baseline_value, + label=( + "parent authority workflow " + f"{job_id} {step_name} {section_name}.{field_name}" + ), + length=40, + ) + section[field_name] = kernel_sha + return expected + + +def first_workflow_difference( + expected: Any, + candidate: Any, + *, + path: str = "workflow", +) -> str | None: + if type(expected) is not type(candidate): + return path + if isinstance(expected, dict): + if tuple(expected) != tuple(candidate): + return f"{path} keys/order" + for key, expected_value in expected.items(): + difference = first_workflow_difference( + expected_value, + candidate[key], + path=f"{path}.{key}", + ) + if difference is not None: + return difference + return None + if isinstance(expected, list): + if len(expected) != len(candidate): + return f"{path} length" + for index, (expected_value, candidate_value) in enumerate( + zip(expected, candidate) + ): + difference = first_workflow_difference( + expected_value, + candidate_value, + path=f"{path}[{index}]", + ) + if difference is not None: + return difference + return None + return None if expected == candidate else path + + +def validate_candidate_workflow(workflow: str, *, kernel_sha: str) -> None: + candidate = parse_strict_workflow_yaml(workflow) + validate_authority_workflow_shape(candidate, label="candidate workflow") + validate_machine_approval_chain(candidate, label="candidate workflow") + expected = expected_candidate_workflow(kernel_sha) + difference = first_workflow_difference(expected, candidate) + if difference is not None: + raise PermitInputError( + "candidate workflow differs from the complete parent-owned authority " + f"contract at {difference}" + ) + + def run_git(repository: Path, *arguments: str) -> bytes: process = subprocess.run( ["git", "-C", str(repository), *arguments], @@ -75,6 +550,124 @@ def git_blob(repository: Path, candidate_sha: str, path: str) -> bytes: return run_git(repository, "show", f"{candidate_sha}:{path}") +def parent_repository() -> Path: + return Path(__file__).resolve().parents[1] + + +def git_workflow_inventory( + repository: Path, + revision: str, + *, + label: str, +) -> dict[str, tuple[str, str, str]]: + records = run_git( + repository, + "ls-tree", + "-r", + "-z", + revision, + "--", + PARENT_WORKFLOW_DIRECTORY, + ).split(b"\0") + inventory: dict[str, tuple[str, str, str]] = {} + prefix = f"{PARENT_WORKFLOW_DIRECTORY}/" + for record in records: + if not record: + continue + try: + metadata, raw_path = record.split(b"\t", maxsplit=1) + mode, object_type, object_id = metadata.split(b" ", maxsplit=2) + path = raw_path.decode("utf-8") + parsed_mode = mode.decode("ascii") + parsed_type = object_type.decode("ascii") + parsed_object_id = object_id.decode("ascii") + except (UnicodeDecodeError, ValueError) as exc: + raise PermitInputError(f"{label} has an invalid Git workflow-tree entry") from exc + if not path.startswith(prefix): + raise PermitInputError(f"{label} escaped the workflow directory: {path!r}") + if path in inventory: + raise PermitInputError(f"{label} contains duplicate workflow path: {path}") + require_sha(parsed_object_id, label=f"{label} object for {path}", length=40) + inventory[path] = (parsed_mode, parsed_type, parsed_object_id) + if not inventory: + raise PermitInputError(f"{label} has no workflow files") + return inventory + + +def parent_workflow_inventory() -> dict[str, tuple[str, str, str]]: + inventory = git_workflow_inventory( + parent_repository(), + "HEAD", + label="parent authority workflow inventory", + ) + if PARENT_WORKFLOW_PATH not in inventory: + raise PermitInputError("parent authority workflow inventory is missing ci.yml") + for path, (mode, object_type, _) in inventory.items(): + if mode != WORKFLOW_REGULAR_MODE or object_type != "blob": + raise PermitInputError( + "parent authority workflow inventory contains a non-regular file: " + f"{path}", + ) + return inventory + + +def parent_workflow_bytes() -> bytes: + parent_workflow_inventory() + return git_blob(parent_repository(), "HEAD", PARENT_WORKFLOW_PATH) + + +def validate_candidate_workflow_inventory( + candidate_repository: Path, + candidate_sha: str, + *, + kernel_sha: str, +) -> None: + parent_inventory = parent_workflow_inventory() + candidate_inventory = git_workflow_inventory( + candidate_repository, + candidate_sha, + label="candidate workflow inventory", + ) + parent_paths = set(parent_inventory) + candidate_paths = set(candidate_inventory) + if candidate_paths != parent_paths: + missing = sorted(parent_paths - candidate_paths) + unexpected = sorted(candidate_paths - parent_paths) + raise PermitInputError( + "candidate workflow inventory differs from the immutable parent; " + f"missing={missing}, unexpected={unexpected}", + ) + + parent_repository_path = parent_repository() + for path in sorted(parent_paths): + parent_mode, parent_type, _ = parent_inventory[path] + candidate_mode, candidate_type, _ = candidate_inventory[path] + if (candidate_mode, candidate_type) != (parent_mode, parent_type): + raise PermitInputError( + "candidate workflow inventory mode/type differs from the immutable " + f"parent at {path}: expected {parent_mode} {parent_type}, " + f"observed {candidate_mode} {candidate_type}", + ) + if path == PARENT_WORKFLOW_PATH: + continue + if git_blob(candidate_repository, candidate_sha, path) != git_blob( + parent_repository_path, + "HEAD", + path, + ): + raise PermitInputError( + "candidate non-authority workflow differs from the immutable parent: " + f"{path}", + ) + + workflow_bytes = git_blob(candidate_repository, candidate_sha, PARENT_WORKFLOW_PATH) + try: + workflow = workflow_bytes.decode("utf-8") + except UnicodeDecodeError as exc: + raise PermitInputError(f"candidate workflow is invalid UTF-8: {exc}") from exc + validate_candidate_workflow(workflow, kernel_sha=kernel_sha) + + def validate_authority_shape(authority: Any, *, label: str) -> dict[str, Any]: if not isinstance(authority, dict): raise PermitInputError(f"{label} must be an object") @@ -294,23 +887,11 @@ def verify(args: argparse.Namespace) -> dict[str, Any]: if candidate_authority[field] != observed: raise PermitInputError(f"candidate {field} does not match {path}") - workflow = git_blob( + validate_candidate_workflow_inventory( candidate_repository, candidate_sha, - ".github/workflows/ci.yml", - ).decode("utf-8") - kernel_ref = f"ref: {candidate_authority['kernel_sha']}" - if workflow.count(kernel_ref) != 3: - raise PermitInputError( - "candidate workflow does not pin the declared Kernel exactly three times" - ) - if ( - "--authority-manifest policy/config/autonomous-release-authority.json" - not in workflow - ): - raise PermitInputError( - "candidate workflow does not bind the authority manifest" - ) + kernel_sha=candidate_authority["kernel_sha"], + ) return { "schema": "mindburn.release-authority-promotion/v1", diff --git a/tests/test_run_autonomous_release_gates.py b/tests/test_run_autonomous_release_gates.py index e99f277..aa3910c 100644 --- a/tests/test_run_autonomous_release_gates.py +++ b/tests/test_run_autonomous_release_gates.py @@ -38,6 +38,35 @@ def test_source_profiles_cover_public_authority_and_product_repositories(self) - "Mindburn-Labs/tempora", }, ) + authority_inventory = profiles["Mindburn-Labs/.github"]["workflow_inventory"] + self.assertIsNotNone(authority_inventory) + self.assertEqual( + set(authority_inventory["files"]), + { + ".github/workflows/ci.yml", + ".github/workflows/docs-truth-public.yml", + ".github/workflows/docs-truth.yml", + ".github/workflows/promote-authority.yml", + }, + ) + + def workflow_inventory_fixture(self, target: Path) -> dict[str, object]: + workflows = target / ".github" / "workflows" + workflows.mkdir(parents=True) + ci = workflows / "ci.yml" + immutable = workflows / "docs-truth.yml" + ci.write_text("name: ci\n", encoding="utf-8") + immutable.write_text("name: docs truth\n", encoding="utf-8") + return { + "root": ".github/workflows", + "files": { + ".github/workflows/ci.yml": {"mode": "100644", "semantic": True}, + ".github/workflows/docs-truth.yml": { + "mode": "100644", + "sha256": hashlib.sha256(immutable.read_bytes()).hexdigest(), + }, + }, + } def test_profile_parser_rejects_duplicate_keys_and_shell_executables(self) -> None: for content, message in ( @@ -66,6 +95,45 @@ def test_profile_parser_rejects_duplicate_keys_and_shell_executables(self) -> No with self.assertRaisesRegex(MODULE.GateProfileError, message): MODULE.load_profiles(path) + def test_profile_parser_rejects_malformed_or_duplicate_workflow_inventory(self) -> None: + malformed = { + "schema": "mindburn.autonomous-release-gates/v2", + "profiles": { + "Mindburn-Labs/example": { + "commands": [["tool", "check"]], + "protected_files": {"gate.json": "0" * 64}, + "workflow_inventory": { + "root": ".github/workflows", + "files": { + ".github/workflows/other.yml": { + "mode": "100644", + "semantic": True, + }, + }, + }, + }, + }, + } + duplicate = ( + '{"schema":"mindburn.autonomous-release-gates/v2","profiles":' + '{"Mindburn-Labs/example":{"commands":[["tool"]],' + '"protected_files":{"gate.json":"' + + "0" * 64 + + '"},"workflow_inventory":{"root":".github/workflows",' + '"files":{".github/workflows/ci.yml":{"mode":"100644",' + '"semantic":true},".github/workflows/ci.yml":{"mode":"100644",' + '"semantic":true}}}}}' + ) + for content, message in ( + (json.dumps(malformed), "may make only"), + (duplicate, "duplicate JSON key"), + ): + with self.subTest(message=message), tempfile.TemporaryDirectory() as tmpdir: + path = Path(tmpdir) / "profiles.json" + path.write_text(content, encoding="utf-8") + with self.assertRaisesRegex(MODULE.GateProfileError, message): + MODULE.load_profiles(path) + def test_repository_without_source_owned_profile_fails_closed(self) -> None: with tempfile.TemporaryDirectory() as tmpdir: target = Path(tmpdir) @@ -89,6 +157,50 @@ def test_changed_or_symlinked_gate_definition_fails_closed(self) -> None: with self.assertRaisesRegex(MODULE.GateProfileError, "not regular"): MODULE.verify_protected_files(target, expected) + def test_workflow_inventory_allows_exact_tree_and_rejects_extra_mode_hash_and_symlink(self) -> None: + with tempfile.TemporaryDirectory() as tmpdir: + target = Path(tmpdir) + inventory = self.workflow_inventory_fixture(target) + MODULE.verify_workflow_inventory(target, inventory) + + extra = target / ".github" / "workflows" / "unreviewed.yml" + extra.write_text("name: unreviewed\n", encoding="utf-8") + with self.assertRaisesRegex(MODULE.GateProfileError, "unexpected"): + MODULE.verify_workflow_inventory(target, inventory) + extra.unlink() + + immutable = target / ".github" / "workflows" / "docs-truth.yml" + immutable.write_text("name: changed\n", encoding="utf-8") + with self.assertRaisesRegex(MODULE.GateProfileError, "file changed"): + MODULE.verify_workflow_inventory(target, inventory) + immutable.write_text("name: docs truth\n", encoding="utf-8") + immutable.chmod(0o755) + with self.assertRaisesRegex(MODULE.GateProfileError, "file mode changed"): + MODULE.verify_workflow_inventory(target, inventory) + immutable.chmod(0o644) + immutable.unlink() + immutable.symlink_to("ci.yml") + with self.assertRaisesRegex(MODULE.GateProfileError, "not regular"): + MODULE.verify_workflow_inventory(target, inventory) + + def test_workflow_inventory_extra_fails_before_any_gate_command(self) -> None: + with tempfile.TemporaryDirectory() as tmpdir: + target = Path(tmpdir) + inventory = self.workflow_inventory_fixture(target) + (target / ".github" / "workflows" / "unreviewed.yml").write_text( + "name: unreviewed\n", + encoding="utf-8", + ) + profiles = { + "Mindburn-Labs/example": { + "commands": [["tool", "check"]], + "protected_files": {"gate.json": "0" * 64}, + "workflow_inventory": inventory, + }, + } + with self.assertRaisesRegex(MODULE.GateProfileError, "unexpected"): + MODULE.resolve_commands("Mindburn-Labs/example", target, profiles) + def test_explicit_profile_runs_argv_without_a_shell(self) -> None: with tempfile.TemporaryDirectory() as tmpdir: root = Path(tmpdir) diff --git a/tests/test_verify_authority_promotion.py b/tests/test_verify_authority_promotion.py index 9da1ec0..79a43de 100644 --- a/tests/test_verify_authority_promotion.py +++ b/tests/test_verify_authority_promotion.py @@ -5,6 +5,7 @@ import importlib.util import json from pathlib import Path +import shutil import subprocess import sys import tempfile @@ -22,7 +23,7 @@ PARENT_SHA = "1" * 40 PARENT_KERNEL_SHA = "2" * 40 -CANDIDATE_KERNEL_SHA = "3" * 40 +CANDIDATE_KERNEL_SHA = "3" * 39 + "a" CONTEXT_SHA = "4" * 64 RESPONSE_A = "5" * 64 RESPONSE_B = "6" * 64 @@ -35,11 +36,48 @@ def git(repository: Path, *arguments: str) -> str: ).strip() -def build_candidate(root: Path) -> tuple[Path, str, str]: +def replace_once(source: str, old: str, new: str) -> str: + if source.count(old) != 1: + raise AssertionError(f"expected exactly one occurrence of {old!r}") + return source.replace(old, new, 1) + + +def replace_after(source: str, anchor: str, old: str, new: str) -> str: + before, separator, after = source.partition(anchor) + if not separator: + raise AssertionError(f"missing anchor {anchor!r}") + if old not in after: + raise AssertionError(f"missing text after anchor {old!r}") + return before + separator + after.replace(old, new, 1) + + +def authority_workflow_fixture(kernel_sha: str = CANDIDATE_KERNEL_SHA) -> str: + authority = json.loads( + (ROOT / "config" / "autonomous-release-authority.json").read_text( + encoding="utf-8" + ) + ) + parent_kernel_sha = authority["kernel_sha"] + workflow = (ROOT / ".github" / "workflows" / "ci.yml").read_text( + encoding="utf-8" + ) + if workflow.count(parent_kernel_sha) != 4: + raise AssertionError("authority workflow must expose exactly four Kernel bindings") + return workflow.replace(parent_kernel_sha, kernel_sha) + + +def build_candidate( + root: Path, + *, + workflow: str | None = None, +) -> tuple[Path, str, str]: repository = root / "candidate" (repository / "config").mkdir(parents=True) (repository / "tests" / "fixtures").mkdir(parents=True) - (repository / ".github" / "workflows").mkdir(parents=True) + shutil.copytree( + ROOT / ".github" / "workflows", + repository / ".github" / "workflows", + ) gates = b'{"profiles":{}}\n' corpus = b'{"schema":"mindburn.release-permit-adversarial/v1","cases":[]}\n' (repository / "config" / "autonomous-release-gates.json").write_bytes(gates) @@ -59,18 +97,13 @@ def build_candidate(root: Path) -> tuple[Path, str, str]: encoding="utf-8", ) (repository / ".github" / "workflows" / "ci.yml").write_text( - f"""name: permit -steps: - - ref: {CANDIDATE_KERNEL_SHA} - - ref: {CANDIDATE_KERNEL_SHA} - - ref: {CANDIDATE_KERNEL_SHA} - - run: python script.py --authority-manifest policy/config/autonomous-release-authority.json -""", + workflow or authority_workflow_fixture(), encoding="utf-8", ) subprocess.run(["git", "-C", str(repository), "init", "-b", "main"], check=True) subprocess.run( - ["git", "-C", str(repository), "config", "user.name", "Permit Test"], check=True + ["git", "-C", str(repository), "config", "user.name", "Permit Test"], + check=True, ) subprocess.run( ["git", "-C", str(repository), "config", "user.email", "permit@example.test"], @@ -84,6 +117,18 @@ def build_candidate(root: Path) -> tuple[Path, str, str]: return repository, candidate_sha, git(repository, "rev-parse", "HEAD^{tree}") +def commit_candidate(repository: Path, *, message: str = "candidate mutation") -> tuple[str, str]: + subprocess.run(["git", "-C", str(repository), "add", "-A"], check=True) + subprocess.run( + ["git", "-C", str(repository), "commit", "-m", message], + check=True, + ) + return ( + git(repository, "rev-parse", "HEAD"), + git(repository, "rev-parse", "HEAD^{tree}"), + ) + + def build_permit(candidate_sha: str, candidate_tree: str) -> dict[str, object]: permit: dict[str, object] = { "schema": "mindburn.release-permit/v2", @@ -168,6 +213,30 @@ def verify_args( class AuthorityPromotionTests(unittest.TestCase): + def assert_inventory_rejected(self, mutation) -> None: + with tempfile.TemporaryDirectory() as tmpdir: + root = Path(tmpdir) + repository, _, _ = build_candidate(root) + mutation(repository) + candidate_sha, candidate_tree = commit_candidate(repository) + permit_path = root / "permit.json" + permit_path.write_text( + json.dumps(build_permit(candidate_sha, candidate_tree)) + "\n", + encoding="utf-8", + ) + with self.assertRaisesRegex( + MODULE.PermitInputError, + "workflow inventory|non-authority workflow", + ): + MODULE.verify( + verify_args( + repository, + candidate_sha, + permit_path, + build_fake_verifier(root), + ), + ) + def test_previous_generation_permit_ratifies_exact_successor(self) -> None: with tempfile.TemporaryDirectory() as tmpdir: root = Path(tmpdir) @@ -235,6 +304,345 @@ def test_kernel_rejection_fails_closed(self) -> None: ), ) + def test_candidate_exact_ci_cannot_add_unreviewed_authority_workflow(self) -> None: + def add_unreviewed_workflow(repository: Path) -> None: + workflow = repository / ".github" / "workflows" / "unreviewed-main.yml" + workflow.write_text( + "name: Unreviewed authority surface\n" + "on:\n" + " push:\n" + " branches: [main]\n" + "jobs:\n" + " execute:\n" + " runs-on: ubuntu-latest\n" + " environment: authority-promotion\n" + " steps:\n" + " - run: true\n", + encoding="utf-8", + ) + + self.assert_inventory_rejected(add_unreviewed_workflow) + + def test_candidate_workflow_inventory_rejects_add_delete_mutation_and_symlink(self) -> None: + def add_nested_workflow(repository: Path) -> None: + nested = repository / ".github" / "workflows" / "nested" + nested.mkdir() + (nested / "unreviewed.yml").write_text("name: unreviewed\n", encoding="utf-8") + + def delete_workflow(repository: Path) -> None: + (repository / ".github" / "workflows" / "docs-truth.yml").unlink() + + def mutate_workflow(repository: Path) -> None: + workflow = repository / ".github" / "workflows" / "docs-truth.yml" + workflow.write_text( + workflow.read_text(encoding="utf-8") + "\n# candidate mutation\n", + encoding="utf-8", + ) + + def symlink_workflow(repository: Path) -> None: + workflow = repository / ".github" / "workflows" / "docs-truth.yml" + workflow.unlink() + workflow.symlink_to("docs-truth-public.yml") + + for name, mutation in ( + ("nested", add_nested_workflow), + ("delete", delete_workflow), + ("mutation", mutate_workflow), + ("symlink", symlink_workflow), + ): + with self.subTest(name=name): + self.assert_inventory_rejected(mutation) + + def test_candidate_workflow_inventory_rejects_gitlink(self) -> None: + with tempfile.TemporaryDirectory() as tmpdir: + root = Path(tmpdir) + repository, _, _ = build_candidate(root) + path = ".github/workflows/promote-authority.yml" + subprocess.run( + ["git", "-C", str(repository), "rm", "--cached", path], + check=True, + ) + subprocess.run( + [ + "git", + "-C", + str(repository), + "update-index", + "--add", + "--cacheinfo", + f"160000,{'f' * 40},{path}", + ], + check=True, + ) + subprocess.run( + ["git", "-C", str(repository), "commit", "-m", "candidate gitlink"], + check=True, + ) + candidate_sha = git(repository, "rev-parse", "HEAD") + candidate_tree = git(repository, "rev-parse", "HEAD^{tree}") + permit_path = root / "permit.json" + permit_path.write_text( + json.dumps(build_permit(candidate_sha, candidate_tree)) + "\n", + encoding="utf-8", + ) + with self.assertRaisesRegex(MODULE.PermitInputError, "workflow inventory"): + MODULE.verify( + verify_args( + repository, + candidate_sha, + permit_path, + build_fake_verifier(root), + ), + ) + + +class CandidateWorkflowContractTests(unittest.TestCase): + def assert_rejected(self, workflow: str) -> None: + with self.assertRaisesRegex( + MODULE.PermitInputError, + "(authority contract|machine-approval|triggers|isolated App token)", + ): + MODULE.validate_candidate_workflow( + workflow, + kernel_sha=CANDIDATE_KERNEL_SHA, + ) + + def test_valid_complete_parent_owned_fixture(self) -> None: + MODULE.validate_candidate_workflow( + authority_workflow_fixture(), + kernel_sha=CANDIDATE_KERNEL_SHA, + ) + + def test_repository_candidate_workflow_is_semantically_valid(self) -> None: + authority = json.loads( + (ROOT / "config" / "autonomous-release-authority.json").read_text( + encoding="utf-8" + ) + ) + MODULE.validate_candidate_workflow( + (ROOT / ".github" / "workflows" / "ci.yml").read_text(encoding="utf-8"), + kernel_sha=authority["kernel_sha"], + ) + + def test_only_declared_kernel_bindings_may_differ(self) -> None: + wrong_kernel = authority_workflow_fixture().replace( + CANDIDATE_KERNEL_SHA, + "d" * 40, + 1, + ) + self.assert_rejected(wrong_kernel) + + def test_comments_cannot_supply_authority_evidence(self) -> None: + wrong_kernel = authority_workflow_fixture().replace( + CANDIDATE_KERNEL_SHA, + "d" * 40, + 1, + ) + self.assert_rejected(wrong_kernel + f"# ref: {CANDIDATE_KERNEL_SHA}\n") + + def test_strict_yaml_rejects_duplicate_keys_aliases_and_tags(self) -> None: + workflow = authority_workflow_fixture() + duplicate = workflow.replace( + " permit:\n", + " prepare:\n steps: []\n permit:\n", + 1, + ) + alias = "shared: &shared\n ignored: true\nalias: *shared\n" + workflow + tagged = workflow.replace( + "name: HELM Autonomous Release Permit", + 'name: !unsafe "HELM Autonomous Release Permit"', + 1, + ) + for name, malformed, message in ( + ("duplicate", duplicate, "duplicate YAML key: prepare"), + ("alias", alias, "YAML aliases or anchors"), + ("tag", tagged, "YAML tags are not allowed"), + ): + with self.subTest(name=name): + with self.assertRaisesRegex(MODULE.PermitInputError, message): + MODULE.validate_candidate_workflow( + malformed, + kernel_sha=CANDIDATE_KERNEL_SHA, + ) + + def test_strict_yaml_rejects_mapping_and_sequence_tags(self) -> None: + for name, malformed in ( + ("mapping", "!unsafe {name: authority}\n"), + ("sequence", "!unsafe [authority]\n"), + ): + with self.subTest(name=name): + with self.assertRaisesRegex(MODULE.PermitInputError, "YAML tags are not allowed"): + MODULE.parse_strict_workflow_yaml(malformed) + + def test_yaml_on_key_is_not_coerced_to_boolean(self) -> None: + self.assert_rejected( + replace_once(authority_workflow_fixture(), "\non:\n", "\ntrue:\n") + ) + + def test_complete_contract_rejects_reviewed_bypass_mutations(self) -> None: + workflow = authority_workflow_fixture() + policy_anchor = " - name: Checkout pinned policy helpers\n" + approval_broker_anchor = " - name: Checkout immutable approval broker\n" + permit_anchor = " permit:\n name: HELM Autonomous Release Permit\n" + machine_anchor = " machine-approval:\n" + approval_step = " - name: Approve only the exact signed head\n" + cases = { + "top-level permission escalation": replace_once( + workflow, + "permissions: {}\n", + "permissions:\n contents: write\n", + ), + "top-level defaults": replace_once( + workflow, + "permissions: {}\n", + "permissions: {}\n" + "defaults:\n" + " run:\n" + " shell: /bin/zsh\n", + ), + "top-level concurrency": replace_once( + workflow, + "permissions: {}\n", + "permissions: {}\n" + "concurrency:\n" + " group: authority-bypass\n", + ), + "false condition": replace_once( + workflow, + "needs.model-review.result == 'success'", + "false", + ), + "job graph rebind": replace_after( + workflow, + machine_anchor, + " needs: permit\n", + " needs: prepare\n", + ), + "job working directory": replace_after( + workflow, + permit_anchor, + " runs-on: ubuntu-latest\n", + " runs-on: ubuntu-latest\n" + " defaults:\n" + " run:\n" + " working-directory: /tmp\n", + ), + "step working directory": replace_after( + workflow, + " permit:\n", + " working-directory: kernel/core\n", + " working-directory: /tmp\n", + ), + "custom shell": replace_after( + workflow, + " permit:\n", + ' run: go build -trimpath -o "$GITHUB_WORKSPACE/release-permit-verify" ./cmd/release-permit-verify\n', + " shell: /bin/zsh\n" + ' run: go build -trimpath -o "$GITHUB_WORKSPACE/release-permit-verify" ./cmd/release-permit-verify\n', + ), + "continue on error": replace_after( + workflow, + " - name: Validate repository contract and workspace truth\n", + " run: REQUIRE_WORKSPACE_CONTEXT=0 make lint test\n", + " continue-on-error: true\n" + " run: REQUIRE_WORKSPACE_CONTEXT=0 make lint test\n", + ), + "policy checkout repoint": replace_after( + workflow, + policy_anchor, + " repository: Mindburn-Labs/.github\n", + " repository: Mindburn-Labs/attacker-policy\n", + ), + "workflow SHA rebind": replace_after( + workflow, + approval_broker_anchor, + " ref: ${{ github.workflow_sha }}\n", + " ref: ${{ github.sha }}\n", + ), + "caps rebind": replace_once( + workflow, + ' MAX_PATCH_BYTES: "524288"\n', + ' MAX_PATCH_BYTES: "999999"\n', + ), + "approval Kernel repoint": replace_after( + workflow, + machine_anchor, + f" ref: {CANDIDATE_KERNEL_SHA}\n", + f" ref: {'d' * 40}\n", + ), + "approval verifier build redirect": replace_after( + workflow, + machine_anchor, + " working-directory: kernel/core\n", + " working-directory: /tmp\n", + ), + "raw approval after App token": replace_after( + workflow, + " - name: Bind exact approval App identity\n", + ' test "$INSTALLATION_ID" = "146576964"\n', + ' test "$INSTALLATION_ID" = "146576964"\n' + ' gh pr review --approve "$GITHUB_REPOSITORY"\n', + ), + "extra approval step": replace_once( + workflow, + approval_step, + " - name: Extra approval bypass\n" + " run: echo unexpected\n" + + approval_step, + ), + "extra trigger": replace_once( + workflow, + " push:\n", + " workflow_dispatch:\n push:\n", + ), + "extra job": workflow + + "\n rogue:\n runs-on: ubuntu-latest\n steps: []\n", + "git clone": replace_once( + workflow, + " run: REQUIRE_WORKSPACE_CONTEXT=0 make lint test\n", + " run: |\n" + " REQUIRE_WORKSPACE_CONTEXT=0 make lint test\n" + " git clone https://example.invalid/rogue.git\n", + ), + "escaped permit filename": replace_once( + workflow, + " --permit approval/release-permit.json \\\n", + " --permit ../approval/release-permit.json \\\n", + ), + "action SHA rebind": replace_after( + workflow, + " - name: Checkout repository\n", + f" uses: {MODULE.CHECKOUT_ACTION}\n", + " uses: actions/checkout@v4\n", + ), + "environment rebind": replace_once( + workflow, + ' CI: "true"\n', + ' CI: "false"\n', + ), + } + for name, mutated in cases.items(): + with self.subTest(name=name): + self.assert_rejected(mutated) + + def test_machine_app_token_has_exactly_one_approval_consumer(self) -> None: + self.assert_rejected( + replace_after( + authority_workflow_fixture(), + " - name: Bind exact approval App identity\n", + " APP_SLUG: ${{ steps.approver-token.outputs.app-slug }}\n", + " APP_SLUG: ${{ steps.approver-token.outputs.app-slug }}\n" + " TOKEN: ${{ steps.approver-token.outputs.token }}\n", + ) + ) + + def test_minimal_or_lexical_fixture_cannot_satisfy_complete_contract(self) -> None: + with self.assertRaisesRegex(MODULE.PermitInputError, "authority contract"): + MODULE.validate_candidate_workflow( + "name: permit\njobs: {}\n", + kernel_sha=CANDIDATE_KERNEL_SHA, + ) + if __name__ == "__main__": unittest.main()