diff --git a/schemas/ruleset-workflow-pin-proposal-v1.schema.json b/schemas/ruleset-workflow-pin-proposal-v1.schema.json new file mode 100644 index 0000000..f4bcbc3 --- /dev/null +++ b/schemas/ruleset-workflow-pin-proposal-v1.schema.json @@ -0,0 +1,68 @@ +{ + "$schema": "https://json-schema.org/draft/2020-12/schema", + "$id": "https://agiletec.net/schemas/ruleset-workflow-pin-proposal-v1.json", + "type": "object", + "additionalProperties": false, + "required": [ + "schema_version", + "operation", + "target", + "change", + "canary", + "digest" + ], + "properties": { + "schema_version": { "const": 1 }, + "operation": { "const": "ruleset-workflow-pin" }, + "target": { + "type": "object", + "additionalProperties": false, + "required": ["organization", "ruleset_id", "workflow"], + "properties": { + "organization": { "const": "agiletec-inc" }, + "ruleset_id": { "const": 19456040 }, + "workflow": { + "type": "object", + "additionalProperties": false, + "required": ["repository_id", "path", "ref"], + "properties": { + "repository_id": { "type": "integer" }, + "path": { "const": ".github/workflows/org-quality-gate.yml" }, + "ref": { "const": "refs/heads/main" } + } + } + } + }, + "change": { + "type": "object", + "additionalProperties": false, + "required": ["proposed_sha"], + "properties": { "proposed_sha": { "type": "string", "pattern": "^[0-9a-f]{40}$" } } + }, + "canary": { + "type": "object", + "additionalProperties": false, + "required": [ + "repository", + "pull_request", + "head_sha", + "check_name", + "check_run_id", + "workflow_run_id", + "workflow_path", + "conclusion" + ], + "properties": { + "repository": { "const": "agiletec-inc/github-actions" }, + "pull_request": { "type": "integer", "minimum": 1 }, + "head_sha": { "type": "string", "pattern": "^[0-9a-f]{40}$" }, + "check_name": { "const": "test" }, + "check_run_id": { "type": "integer", "minimum": 1 }, + "workflow_run_id": { "type": "integer", "minimum": 1 }, + "workflow_path": { "const": ".github/workflows/ci.yml" }, + "conclusion": { "const": "success" } + } + }, + "digest": { "type": "string", "pattern": "^sha256:[0-9a-f]{64}$" } + } +} diff --git a/tests/test_ruleset_workflow_pin_proposal.py b/tests/test_ruleset_workflow_pin_proposal.py new file mode 100644 index 0000000..6d68f5d --- /dev/null +++ b/tests/test_ruleset_workflow_pin_proposal.py @@ -0,0 +1,173 @@ +from __future__ import annotations + +import hashlib +import importlib.util +import json +import pathlib +import unittest + + +ROOT = pathlib.Path(__file__).resolve().parents[1] +SCRIPT = ROOT / "tools" / "generate_required_workflow_pin_proposal.py" +SPEC = importlib.util.spec_from_file_location("ruleset_pin_proposal", SCRIPT) +assert SPEC is not None and SPEC.loader is not None +MODULE = importlib.util.module_from_spec(SPEC) +SPEC.loader.exec_module(MODULE) + +PROPOSED_SHA = "2" * 40 +REPOSITORY_ID = 12345 + + +class FakeReader: + def __init__(self, responses: dict[str, dict[str, object]]) -> None: + self.responses = responses + self.paths: list[str] = [] + + def get_object(self, path: str) -> dict[str, object]: + self.paths.append(path) + return self.responses[path] + + +def responses() -> dict[str, dict[str, object]]: + repository_path = "/repos/agiletec-inc/github-actions" + return { + repository_path: {"id": REPOSITORY_ID, "default_branch": "main"}, + f"{repository_path}/commits/{PROPOSED_SHA}": {"sha": PROPOSED_SHA}, + f"{repository_path}/compare/{PROPOSED_SHA}...main": {"status": "ahead"}, + f"{repository_path}/pulls/25": { + "state": "closed", + "merged_at": "2026-07-27T00:00:00Z", + "base": {"ref": "main"}, + "head": {"sha": PROPOSED_SHA}, + }, + f"{repository_path}/commits/{PROPOSED_SHA}/check-runs?filter=latest&per_page=100": { + "check_runs": [ + { + "id": 88, + "name": "test", + "status": "completed", + "conclusion": "success", + "details_url": "https://github.com/agiletec-inc/github-actions/actions/runs/77/job/99", + "app": {"slug": "github-actions"}, + } + ] + }, + f"{repository_path}/actions/runs/77": { + "head_sha": PROPOSED_SHA, + "path": ".github/workflows/ci.yml", + "status": "completed", + "conclusion": "success", + }, + "/repos/agiletec-inc/agiletec/rulesets/19456040": { + "id": 19456040, + "source_type": "Organization", + "source": "agiletec-inc", + "enforcement": "active", + "rules": [ + { + "type": "workflows", + "parameters": { + "workflows": [ + { + "repository_id": REPOSITORY_ID, + "path": ".github/workflows/org-quality-gate.yml", + "ref": "refs/heads/main", + } + ] + }, + } + ], + }, + } + + +class ProposalTest(unittest.TestCase): + def test_build_proposal_is_deterministic_and_digest_covers_unsigned_body( + self, + ) -> None: + canary = { + "repository": "agiletec-inc/github-actions", + "pull_request": 25, + "head_sha": PROPOSED_SHA, + "check_name": "test", + "check_run_id": 88, + "workflow_run_id": 77, + "workflow_path": ".github/workflows/ci.yml", + "conclusion": "success", + } + first = MODULE.build_proposal(REPOSITORY_ID, PROPOSED_SHA, canary) + second = MODULE.build_proposal(REPOSITORY_ID, PROPOSED_SHA, canary) + self.assertEqual(first, second) + unsigned = {key: value for key, value in first.items() if key != "digest"} + expected = hashlib.sha256(MODULE.canonical_json(unsigned).encode()).hexdigest() + self.assertEqual(first["digest"], f"sha256:{expected}") + + def test_validates_candidate_canary_and_effective_ruleset(self) -> None: + reader = FakeReader(responses()) + repository_id = MODULE.validate_candidate(reader, PROPOSED_SHA) + canary = MODULE.validate_canary(reader, 25, PROPOSED_SHA) + ruleset = reader.get_object("/repos/agiletec-inc/agiletec/rulesets/19456040") + MODULE.validate_effective_workflow(ruleset, repository_id) + self.assertEqual(canary["workflow_run_id"], 77) + + def test_rejects_candidate_not_reachable_from_main(self) -> None: + payloads = responses() + payloads[f"/repos/agiletec-inc/github-actions/compare/{PROPOSED_SHA}...main"][ + "status" + ] = "diverged" + with self.assertRaisesRegex(MODULE.ProposalError, "not reachable"): + MODULE.validate_candidate(FakeReader(payloads), PROPOSED_SHA) + + def test_rejects_canary_head_mismatch(self) -> None: + payloads = responses() + payloads["/repos/agiletec-inc/github-actions/pulls/25"]["head"] = { + "sha": "3" * 40 + } + with self.assertRaisesRegex(MODULE.ProposalError, "head does not match"): + MODULE.validate_canary(FakeReader(payloads), 25, PROPOSED_SHA) + + def test_rejects_spoofed_or_unsuccessful_check(self) -> None: + payloads = responses() + checks = payloads[ + f"/repos/agiletec-inc/github-actions/commits/{PROPOSED_SHA}/check-runs?filter=latest&per_page=100" + ]["check_runs"] + assert isinstance(checks, list) + checks[0]["app"] = {"slug": "third-party"} + with self.assertRaisesRegex(MODULE.ProposalError, "Expected one canary check"): + MODULE.validate_canary(FakeReader(payloads), 25, PROPOSED_SHA) + + def test_rejects_wrong_workflow_run(self) -> None: + payloads = responses() + payloads["/repos/agiletec-inc/github-actions/actions/runs/77"]["path"] = ( + ".github/workflows/other.yml" + ) + with self.assertRaisesRegex(MODULE.ProposalError, "does not match"): + MODULE.validate_canary(FakeReader(payloads), 25, PROPOSED_SHA) + + def test_rejects_inactive_or_ambiguous_effective_ruleset(self) -> None: + ruleset = responses()["/repos/agiletec-inc/agiletec/rulesets/19456040"] + ruleset["enforcement"] = "disabled" + with self.assertRaisesRegex(MODULE.ProposalError, "not active"): + MODULE.validate_effective_workflow(ruleset, REPOSITORY_ID) + + def test_schema_and_script_keep_fixed_authority_boundary(self) -> None: + schema = json.loads( + ( + ROOT / "schemas" / "ruleset-workflow-pin-proposal-v1.schema.json" + ).read_text() + ) + self.assertEqual( + schema["properties"]["operation"]["const"], "ruleset-workflow-pin" + ) + target = schema["properties"]["target"]["properties"] + self.assertEqual(target["organization"]["const"], "agiletec-inc") + self.assertEqual(target["ruleset_id"]["const"], 19456040) + source = SCRIPT.read_text() + self.assertNotRegex(source, r'method="(?:POST|PUT|PATCH|DELETE)"') + self.assertNotIn("--token", source) + self.assertNotIn("admin:org", source) + self.assertNotIn("dotenv", source) + + +if __name__ == "__main__": + unittest.main() diff --git a/tools/generate_required_workflow_pin_proposal.py b/tools/generate_required_workflow_pin_proposal.py new file mode 100644 index 0000000..1c318e5 --- /dev/null +++ b/tools/generate_required_workflow_pin_proposal.py @@ -0,0 +1,253 @@ +#!/usr/bin/env python3 +from __future__ import annotations + +import argparse +import hashlib +import json +import os +import re +import sys +import urllib.error +import urllib.request +from typing import Any + + +ORGANIZATION = "agiletec-inc" +SOURCE_REPOSITORY = "github-actions" +CANARY_REPOSITORY = "github-actions" +EFFECTIVE_REPOSITORY = "agiletec" +RULESET_ID = 19456040 +TARGET_WORKFLOW = ".github/workflows/org-quality-gate.yml" +CANARY_WORKFLOW = ".github/workflows/ci.yml" +CANARY_CHECK = "test" + + +class ProposalError(RuntimeError): + pass + + +class GitHubReader: + def __init__(self, base_url: str, token: str) -> None: + self.base_url = base_url.rstrip("/") + self.token = token + + def get_object(self, path: str) -> dict[str, Any]: + request = urllib.request.Request( + f"{self.base_url}{path}", + method="GET", + headers={ + "Accept": "application/vnd.github+json", + "Authorization": f"Bearer {self.token}", + "X-GitHub-Api-Version": "2026-03-10", + }, + ) + try: + with urllib.request.urlopen(request, timeout=30) as response: + payload = response.read() + except (urllib.error.HTTPError, urllib.error.URLError, TimeoutError) as error: + raise ProposalError(f"GitHub API GET {path} failed: {error}") from error + try: + decoded = json.loads(payload) + except (json.JSONDecodeError, UnicodeDecodeError) as error: + raise ProposalError( + f"GitHub API GET {path} returned invalid JSON" + ) from error + if not isinstance(decoded, dict): + raise ProposalError(f"GitHub API GET {path} did not return an object") + return decoded + + +def require_string(record: dict[str, Any], key: str, context: str) -> str: + value = record.get(key) + if not isinstance(value, str) or not value: + raise ProposalError(f"{context} has invalid {key}") + return value + + +def require_integer(record: dict[str, Any], key: str, context: str) -> int: + value = record.get(key) + if not isinstance(value, int) or isinstance(value, bool): + raise ProposalError(f"{context} has invalid {key}") + return value + + +def validate_candidate(reader: GitHubReader, proposed_sha: str) -> int: + repository = reader.get_object(f"/repos/{ORGANIZATION}/{SOURCE_REPOSITORY}") + repository_id = require_integer(repository, "id", "source repository") + if require_string(repository, "default_branch", "source repository") != "main": + raise ProposalError("Source repository default branch is not main") + commit = reader.get_object( + f"/repos/{ORGANIZATION}/{SOURCE_REPOSITORY}/commits/{proposed_sha}" + ) + if require_string(commit, "sha", "candidate commit") != proposed_sha: + raise ProposalError("Candidate commit SHA mismatch") + comparison = reader.get_object( + f"/repos/{ORGANIZATION}/{SOURCE_REPOSITORY}/compare/{proposed_sha}...main" + ) + if comparison.get("status") not in {"ahead", "identical"}: + raise ProposalError("Candidate SHA is not reachable from github-actions main") + return repository_id + + +def validate_canary( + reader: GitHubReader, canary_pr: int, proposed_sha: str +) -> dict[str, object]: + repository_path = f"/repos/{ORGANIZATION}/{CANARY_REPOSITORY}" + pull = reader.get_object(f"{repository_path}/pulls/{canary_pr}") + state = pull.get("state") + merged_at = pull.get("merged_at") + if state != "open" and not (state == "closed" and isinstance(merged_at, str)): + raise ProposalError("Canary pull request is neither open nor merged") + base = pull.get("base") + head = pull.get("head") + if ( + not isinstance(base, dict) + or base.get("ref") != "main" + or not isinstance(head, dict) + ): + raise ProposalError("Canary pull request has an invalid base or head") + head_sha = require_string(head, "sha", "canary pull request head") + if head_sha != proposed_sha: + raise ProposalError("Canary pull request head does not match the proposed SHA") + checks = reader.get_object( + f"{repository_path}/commits/{head_sha}/check-runs?filter=latest&per_page=100" + ) + check_runs = checks.get("check_runs") + if not isinstance(check_runs, list): + raise ProposalError("Canary check-runs response is invalid") + matching = [ + check + for check in check_runs + if isinstance(check, dict) + and check.get("name") == CANARY_CHECK + and isinstance(check.get("app"), dict) + and check["app"].get("slug") == "github-actions" + ] + if len(matching) != 1: + raise ProposalError(f"Expected one canary check, found {len(matching)}") + check = matching[0] + if check.get("status") != "completed" or check.get("conclusion") != "success": + raise ProposalError("Canary check is not successful") + check_id = require_integer(check, "id", "canary check") + details_url = require_string(check, "details_url", "canary check") + run_match = re.search(r"/actions/runs/(\d+)(?:/|$)", details_url) + if run_match is None: + raise ProposalError("Canary check has no workflow run URL") + run_id = int(run_match.group(1)) + run = reader.get_object(f"{repository_path}/actions/runs/{run_id}") + if run.get("head_sha") != head_sha or run.get("path") != CANARY_WORKFLOW: + raise ProposalError( + "Canary workflow run does not match the candidate head or path" + ) + if run.get("conclusion") != "success" or run.get("status") != "completed": + raise ProposalError("Canary workflow run is not successful") + return { + "repository": f"{ORGANIZATION}/{CANARY_REPOSITORY}", + "pull_request": canary_pr, + "head_sha": head_sha, + "check_name": CANARY_CHECK, + "check_run_id": check_id, + "workflow_run_id": run_id, + "workflow_path": CANARY_WORKFLOW, + "conclusion": "success", + } + + +def validate_effective_workflow(ruleset: dict[str, Any], repository_id: int) -> None: + if ruleset.get("id") != RULESET_ID: + raise ProposalError("Effective Ruleset ID mismatch") + if ( + ruleset.get("source_type") != "Organization" + or ruleset.get("source") != ORGANIZATION + ): + raise ProposalError("Effective Ruleset authority mismatch") + if ruleset.get("enforcement") != "active": + raise ProposalError("Effective Ruleset is not active") + rules = ruleset.get("rules") + if not isinstance(rules, list): + raise ProposalError("Effective Ruleset has invalid rules") + matching: list[dict[str, Any]] = [] + for rule in rules: + if not isinstance(rule, dict) or rule.get("type") != "workflows": + continue + parameters = rule.get("parameters") + workflows = ( + parameters.get("workflows") if isinstance(parameters, dict) else None + ) + if not isinstance(workflows, list): + raise ProposalError("Effective workflow rule has invalid workflows") + for workflow in workflows: + if not isinstance(workflow, dict): + raise ProposalError("Effective Ruleset has an invalid workflow entry") + if ( + workflow.get("repository_id") == repository_id + and workflow.get("path") == TARGET_WORKFLOW + and workflow.get("ref") == "refs/heads/main" + ): + matching.append(workflow) + if len(matching) != 1: + raise ProposalError(f"Expected one effective workflow, found {len(matching)}") + + +def canonical_json(value: object) -> str: + return json.dumps(value, ensure_ascii=False, separators=(",", ":"), sort_keys=True) + + +def build_proposal( + repository_id: int, + proposed_sha: str, + canary: dict[str, object], +) -> dict[str, object]: + unsigned: dict[str, object] = { + "schema_version": 1, + "operation": "ruleset-workflow-pin", + "target": { + "organization": ORGANIZATION, + "ruleset_id": RULESET_ID, + "workflow": { + "repository_id": repository_id, + "path": TARGET_WORKFLOW, + "ref": "refs/heads/main", + }, + }, + "change": {"proposed_sha": proposed_sha}, + "canary": canary, + } + digest = hashlib.sha256(canonical_json(unsigned).encode()).hexdigest() + return {**unsigned, "digest": f"sha256:{digest}"} + + +def main() -> int: + parser = argparse.ArgumentParser() + parser.add_argument("--proposed-sha", required=True) + parser.add_argument("--canary-pr", type=int, required=True) + parser.add_argument("--api-base-url", default="https://api.github.com") + args = parser.parse_args() + if not re.fullmatch(r"[0-9a-f]{40}", args.proposed_sha): + raise ProposalError( + "--proposed-sha must be a 40-character lowercase commit SHA" + ) + if args.canary_pr < 1: + raise ProposalError("--canary-pr must be positive") + token = os.environ.get("GH_TOKEN") + if not token: + raise ProposalError("GH_TOKEN is required") + + reader = GitHubReader(args.api_base_url, token) + repository_id = validate_candidate(reader, args.proposed_sha) + canary = validate_canary(reader, args.canary_pr, args.proposed_sha) + ruleset = reader.get_object( + f"/repos/{ORGANIZATION}/{EFFECTIVE_REPOSITORY}/rulesets/{RULESET_ID}" + ) + validate_effective_workflow(ruleset, repository_id) + proposal = build_proposal(repository_id, args.proposed_sha, canary) + print(json.dumps(proposal, ensure_ascii=False, indent=2, sort_keys=True)) + return 0 + + +if __name__ == "__main__": + try: + raise SystemExit(main()) + except ProposalError as error: + print(f"error: {error}", file=sys.stderr) + raise SystemExit(1) from error