diff --git a/README.md b/README.md index b294d6f..239a8ef 100644 --- a/README.md +++ b/README.md @@ -113,6 +113,32 @@ admission remains enabled. Same-repository Dependabot updates to existing action SHA pins can enter automatically after full immutable workflow blobs prove that only pins and version comments changed. Runtime code, permissions, renamed or new workflows, and other automation changes still require deliberate admission. +A package repository can opt in to dispatched source-PR result reporting with +`package-source-checks`, a JSON object mapping each configured workflow filename +to its required job names. Include every protected context exactly once and +include `dco.yml` with `DCO sign-off`. Keep this map aligned with branch protection. +The source adapter accepts only open, non-draft, same-repository PRs from the +GitHub Actions bot on `automation/package-updates-` branches. It proves +one actual parent on protected default-branch history, changes limited to the +selected `pkgs/by-name///` directory and `docs/catalog.md`, and +workflow blobs identical to current protected main. It executes no PR code. + +Source statuses link actual successful required jobs only after every configured +workflow and required job passes on the exact source SHA. A blocked approval run +can be ignored only when its jobs endpoint proves zero jobs. Failed, pending, +skipped, missing, stale and untrusted evidence never becomes success; newer +attempts invalidate older source adapter results. Actual failed required jobs +receive failure statuses linking the current failed job. Normal required checks and +expected-head merge-queue admission still apply. Unchanged completed runs reuse +only this bot's previously verified job attestations for the same run and attempt; +latest-run metadata remains live. Immutable Git objects and workflow identity are +cached only within one reconciliation. Opted-in callers should also reconcile +`workflow_run` requested and in-progress events to invalidate rerun results early. +These events inspect only their exact source head; initial events without prior +results do not poll validation, and unrelated scans skip already-admitted heads. +Keep hourly scheduled recovery for missed events. This opt-in adds no token +permissions and leaves existing synthetic queue result reporting unchanged. + Set `base-sha-workflows: '["ci.yml"]'` for a workflow that declares a `base_sha` dispatch input. This supplies the queue commit's first parent for derivation comparison without teaching the shared library repository names. diff --git a/actions/reconcile-queue/action.yml b/actions/reconcile-queue/action.yml index f436211..6d94fba 100644 --- a/actions/reconcile-queue/action.yml +++ b/actions/reconcile-queue/action.yml @@ -7,6 +7,9 @@ inputs: automation-branch: description: Optional same-repository github-actions bot branch eligible for admission. default: '' + package-source-checks: + description: Opt-in JSON workflow-to-required-job map for same-repository package updater source PRs. + default: '{}' base-sha-workflows: description: JSON array of workflows accepting the base_sha dispatch input. default: '[]' @@ -19,10 +22,13 @@ runs: - name: Validate queue inputs shell: bash env: + PACKAGE_SOURCE_CHECKS: ${{ inputs.package-source-checks }} AUTOMATION_BRANCH: ${{ inputs.automation-branch }} BASE_SHA_WORKFLOWS: ${{ inputs.base-sha-workflows }} QUEUE_WORKFLOWS: ${{ inputs.workflows }} SOURCE_RUN_ID: ${{ inputs.source-run-id }} + SOURCE_HEAD_SHA: ${{ github.event.workflow_run.head_sha }} + SOURCE_ACTIVITY: ${{ github.event.action }} run: | set -euo pipefail if [[ -n "$SOURCE_RUN_ID" && ! "$SOURCE_RUN_ID" =~ ^[0-9]+$ ]]; then @@ -33,19 +39,32 @@ runs: printf 'Invalid automation-branch input: %s\n' "$AUTOMATION_BRANCH" >&2 exit 1 fi - python3 - "$QUEUE_WORKFLOWS" "$BASE_SHA_WORKFLOWS" <<'PY' + python3 - "$QUEUE_WORKFLOWS" "$BASE_SHA_WORKFLOWS" "$PACKAGE_SOURCE_CHECKS" <<'PY' import json import re import sys workflow_pattern = re.compile(r"^[A-Za-z0-9._/-]+\.yml$") - for label, value in zip(("workflows", "base-sha-workflows"), sys.argv[1:]): + for label, value in zip(("workflows", "base-sha-workflows"), sys.argv[1:3]): entries = json.loads(value) if not isinstance(entries, list) or not all( isinstance(entry, str) and workflow_pattern.fullmatch(entry) for entry in entries ): raise SystemExit(f"Invalid {label} input") + source = json.loads(sys.argv[3]) + if not isinstance(source, dict) or any( + not workflow_pattern.fullmatch(key) or not isinstance(jobs, list) or not jobs + or any(not isinstance(job, str) or not job or len(job) > 256 for job in jobs) + or len(jobs) != len(set(jobs)) + for key, jobs in source.items() + ): + raise SystemExit("Invalid package-source-checks input") + contexts = [job for jobs in source.values() for job in jobs] + if len(contexts) != len(set(contexts)): + raise SystemExit("Source required job names must be unique") + if source and (set(source) != set(json.loads(sys.argv[1])) or "DCO sign-off" not in source.get("dco.yml", [])): + raise SystemExit("Source checks must cover every workflow, including actual source DCO") PY - name: Set up queue validation Python uses: actions/setup-python@5fda3b95a4ea91299a34e894583c3862153e4b97 # v7 @@ -66,8 +85,11 @@ runs: GH_TOKEN: ${{ github.token }} GH_REPO: ${{ github.repository }} DEFAULT_BRANCH: ${{ github.event.repository.default_branch }} + PACKAGE_SOURCE_CHECKS: ${{ inputs.package-source-checks }} AUTOMATION_BRANCH: ${{ inputs.automation-branch }} BASE_SHA_WORKFLOWS: ${{ inputs.base-sha-workflows }} QUEUE_WORKFLOWS: ${{ inputs.workflows }} SOURCE_RUN_ID: ${{ inputs.source-run-id }} + SOURCE_HEAD_SHA: ${{ github.event.workflow_run.head_sha }} + SOURCE_ACTIVITY: ${{ github.event.action }} run: python3 "$ACTION_PATH/../../scripts/reconcile-queue.py" diff --git a/docs/releases/v2.11.0.md b/docs/releases/v2.11.0.md new file mode 100644 index 0000000..1272d36 --- /dev/null +++ b/docs/releases/v2.11.0.md @@ -0,0 +1,41 @@ +# Nix Forge CI v2.11.0 + +## Changelog + +- Add opt-in reporting and normal queue admission for verified package source PRs + whose explicitly dispatched validation is missing from GitHub's PR rollup. +- Prove the exact GitHub Actions bot identity, same-repository package branch, + actual parent ancestry, selected package/catalog paths, and current protected + workflow blobs before trusting source validation. The reconciler executes no PR code. +- Report actual failed required jobs with their exact job URLs. Missing, running, + skipped, stale, or untrusted evidence never becomes success. Producer changes + during collection or publication invalidate old results and refuse admission. +- Reuse verified unchanged-run attestations and cache only immutable Git data and + stable workflow identity within one reconciliation. Latest-run state remains + live; exact-head events avoid polling unrelated source PRs. + +Existing callers keep source reporting disabled and retain their current queue +behavior. Update general shared references coherently to the v2.11.0 release +commit. Repository checks and SLSA builders retain separately reviewed pins. +The package caller opts in through `package-source-checks`, mapping every +configured workflow to all of its protected required job names, including actual +source `DCO sign-off`. Add `requested`, `in_progress`, and `completed` workflow-run +notifications and retain hourly scheduled recovery. Regenerate existing source +branches from reviewed protected main so their workflow blobs match the caller. + +The change adds no token permissions and passes no privileged token to PR builds. +Expected-head queue admission and full merge-group validation remain required. +Regression coverage includes source identity and ownership, workflow drift, +approval runs with zero jobs, actual job run/attempt provenance, failure URLs, +producer races, stale-result invalidation, and six-PR request budgets. All 133 +repository tests and offline workflow/security checks passed locally; required +hosted PR and merge-group validation and representative caller checks must pass +before the release tag is published. + +The functional source was reviewed at +`dc803bb28ea7ae9f2c576d2d03bb78a7f497e72b`. The signed tag and release manifest identify +the final reviewed source commit including these notes. Publication uses +`nix-forge/ci/.github/workflows/release.yml`, which verifies the archive checksum, +all provenance subjects, exact pinned builder identity, and protected-main tag +ancestry. Restore the v2.10.0 pin and remove the source opt-in for rollback. +No support end date is scheduled for compatible v2.x consumers. diff --git a/scripts/reconcile-queue.py b/scripts/reconcile-queue.py index d36cc27..2fc8752 100644 --- a/scripts/reconcile-queue.py +++ b/scripts/reconcile-queue.py @@ -28,6 +28,12 @@ BASE_SHA_WORKFLOWS = json.loads(os.environ.get("BASE_SHA_WORKFLOWS", "[]")) QUEUE_PREFIX = f"refs/heads/gh-readonly-queue/{DEFAULT_BRANCH}/" ADMISSION = "Merge queue admission" +SOURCE_CHECKS = json.loads(os.environ.get("PACKAGE_SOURCE_CHECKS", "{}")) +SOURCE_REPORT = "Mirrored package source job" +SOURCE_HEAD = os.environ.get("SOURCE_HEAD_SHA", "") +SOURCE_ACTIVITY = os.environ.get("SOURCE_ACTIVITY", "") +SOURCE_CACHE: dict[str, Any] = {} +SOURCE_EVIDENCE: dict[str, dict[str, Any]] = {} PAGE_SIZE = 100 GH = shutil.which("gh") or "/usr/bin/gh" @@ -76,10 +82,31 @@ def admit(pr: dict[str, Any]) -> bool: Whether this trusted PR is now queued and may need a validation ref. """ - trusted = pr["user"]["login"] == "dependabot[bot]" or ( - bool(AUTOMATION_BRANCH) - and pr["user"]["login"] == "github-actions[bot]" - and pr["head"]["ref"] == AUTOMATION_BRANCH + source = bool(SOURCE_CHECKS) and source_identity(pr) + source_statuses = None + if source: + sha = pr["head"]["sha"] + if SOURCE_HEAD and SOURCE_HEAD != sha: + return False + source_statuses = read_statuses(sha) + if not SOURCE_HEAD and any( + status["context"] == ADMISSION for status in source_statuses.values() + ): + state = graphql( + "query($id:ID!) { node(id:$id) { ... on PullRequest { mergeQueueEntry { id } } } }", + id=pr["node_id"], + )["node"] + return bool(state["mergeQueueEntry"]) + if not reconcile_source(pr, source_statuses): + return False + trusted = ( + source + or pr["user"]["login"] == "dependabot[bot]" + or ( + bool(AUTOMATION_BRANCH) + and pr["user"]["login"] == "github-actions[bot]" + and pr["head"]["ref"] == AUTOMATION_BRANCH + ) ) if ( not trusted @@ -114,7 +141,11 @@ def admit(pr: dict[str, Any]) -> bool: return True # Once admitted, a rejected head needs a fix or deliberate requeue. # Never keep rebuilding the same failing merge group indefinitely. - statuses = api(f"repos/{REPOSITORY}/commits/{sha}/status")["statuses"] + statuses = ( + list(source_statuses.values()) + if source_statuses is not None + else api(f"repos/{REPOSITORY}/commits/{sha}/status")["statuses"] + ) if any(status["context"] == ADMISSION for status in statuses): print(f"PR #{number}: already admitted this head; leave rejection for review") return False @@ -378,6 +409,380 @@ def front_queue_entry() -> dict[str, Any] | None: return entries[0] if entries else None +def source_identity(pr: dict[str, Any]) -> bool: + """Limit opt-in source reconciliation to the package updater's bot PRs. + + Returns: + Whether identity and namespace are eligible for immutable inspection. + + """ + return ( + pr.get("state") == "open" + and not pr.get("draft", True) + and pr.get("user", {}).get("id") == 41898282 + and pr["user"].get("type") == "Bot" + and pr["user"].get("login") == "github-actions[bot]" + and (pr.get("head", {}).get("repo") or {}).get("full_name") == REPOSITORY + and (pr.get("base", {}).get("repo") or {}).get("full_name") == REPOSITORY + and pr["base"].get("ref") == DEFAULT_BRANCH + and re.fullmatch( + r"automation/package-updates-[a-z0-9][a-z0-9+._-]*", + pr["head"].get("ref", ""), + ) + is not None + and re.fullmatch(r"[0-9a-f]{40}", pr["head"].get("sha", "")) is not None + ) + + +def source_read(path: str) -> Any: # ruff: ignore[any-type] + """Cache immutable source objects and stable workflow identity within one pass. + + Returns: + The authoritative response; current refs and latest runs are never cached. + + """ + if path not in SOURCE_CACHE: + SOURCE_CACHE[path] = api(path) + return SOURCE_CACHE[path] + + +def source_contract(pr: dict[str, Any]) -> str | None: + """Prove the real source parent, owned paths and current workflow identities. + + Returns: + The exact protected base, or None for an untrusted or stale source. + + """ + sha = pr["head"]["sha"] + commit = source_read(f"repos/{REPOSITORY}/git/commits/{sha}") + parents = commit.get("parents", []) + if len(parents) != 1: + return None + base = parents[0]["sha"] + if re.fullmatch(r"[0-9a-f]{40}", base) is None: + return None + protected = api( + f"repos/{REPOSITORY}/git/ref/heads/{quote(DEFAULT_BRANCH, safe='')}" + )["object"]["sha"] + comparison = source_read(f"repos/{REPOSITORY}/compare/{base}...{protected}") + if comparison.get("status") not in {"ahead", "identical"}: + return None + changes = source_read(f"repos/{REPOSITORY}/compare/{base}...{sha}") + files = changes.get("files", []) + package = pr["head"]["ref"].removeprefix("automation/package-updates-") + prefix = f"pkgs/by-name/{package[:2]}/{package}/" + if ( + changes.get("status") != "ahead" + or changes.get("total_commits") != 1 + or not files + or len(files) >= 300 + or not any(item["filename"].startswith(prefix) for item in files) + or any( + item.get("status") == "renamed" + or not ( + item["filename"].startswith(prefix) + or item["filename"] == "docs/catalog.md" + ) + for item in files + ) + ): + return None + trees = [] + for revision in (sha, protected): + tree = source_read(f"repos/{REPOSITORY}/git/trees/{revision}?recursive=1") + if tree.get("truncated") is not False: + return None + trees.append({item["path"]: item for item in tree["tree"]}) + for workflow in WORKFLOWS: + path = f".github/workflows/{workflow}" + old, current = (tree.get(path, {}) for tree in trees) + if ( + not old.get("sha") + or old.get("sha") != current.get("sha") + or old.get("mode") != "100644" + or current.get("mode") != "100644" + or old.get("type") != "blob" + or current.get("type") != "blob" + ): + return None + return base + + +def source_run( + pr: dict[str, Any], workflow: str +) -> tuple[dict[str, Any], dict[str, Any] | None] | None: + """Select the latest real exact-head run, proving approval-only exclusions. + + Returns: + Current run metadata and optional already-read jobs, or None. + + """ + sha, branch = pr["head"]["sha"], pr["head"]["ref"] + metadata = source_read(f"repos/{REPOSITORY}/actions/workflows/{workflow}") + path = f".github/workflows/{workflow}" + if metadata.get("path") != path or metadata.get("state") != "active": + return None + response = api( + f"repos/{REPOSITORY}/actions/workflows/{workflow}/runs?head_sha={sha}&per_page={PAGE_SIZE}" + ) + if ( + not isinstance(response.get("total_count"), int) + or not 0 <= response["total_count"] <= PAGE_SIZE + or not isinstance(response.get("workflow_runs"), list) + ): + return None + for run in response["workflow_runs"]: + if not ( + run.get("workflow_id") == metadata["id"] + and run.get("path") == path + and run.get("head_sha") == sha + and run.get("head_branch") == branch + and run.get("repository", {}).get("full_name") == REPOSITORY + and run.get("head_repository", {}).get("full_name") == REPOSITORY + and run.get("event") in {"pull_request", "workflow_dispatch"} + and isinstance(run.get("id"), int) + and run["id"] > 0 + and isinstance(run.get("run_attempt"), int) + and run["run_attempt"] > 0 + ): + continue + batch = None + if ( + run.get("status") == "completed" + and run.get("conclusion") == "action_required" + ): + batch = api( + f"repos/{REPOSITORY}/actions/runs/{run['id']}/jobs?filter=latest&per_page={PAGE_SIZE}" + ) + if batch.get("total_count") == 0 and batch.get("jobs") == []: + continue + return run.copy(), batch + return None + + +def source_runs_current(pr: dict[str, Any]) -> bool: + """Reject any producer run or attempt that changed after collection. + + Returns: + Whether every collected run still has its exact inspected identity/state. + + """ + for workflow, expected in SOURCE_EVIDENCE[pr["head"]["sha"]]["runs"].items(): + selected = source_run(pr, workflow) + if selected is None or any( + selected[0].get(key) != expected.get(key) + for key in ("id", "run_attempt", "head_sha", "status", "conclusion") + ): + return False + return True + + +def source_jobs( + pr: dict[str, Any], statuses: dict[str, Any] | None = None +) -> list[dict[str, Any]] | None: + """Collect actual current jobs, or reuse our verified unchanged-run attestation. + + Returns: + All required successful jobs, or None for incomplete/failed evidence. + + """ + sha = pr["head"]["sha"] + evidence: dict[str, Any] = {"runs": {}, "failures": []} + SOURCE_EVIDENCE[sha] = evidence + jobs: list[dict[str, Any]] = [] + failures = {"failure", "timed_out", "cancelled"} + for workflow in WORKFLOWS: + selected = source_run(pr, workflow) + if selected is None: + return None + run, batch = selected + evidence["runs"][workflow] = run + if run.get("status") != "completed": + return None + previous = ( + [statuses.get(context, {}) for context in SOURCE_CHECKS[workflow]] + if statuses + else [] + ) + description = f"{SOURCE_REPORT} result; attempt {run['run_attempt']}" + prefix = f"https://github.com/{REPOSITORY}/actions/runs/{run['id']}/job/" + if ( + run.get("conclusion") == "success" + and previous + and all( + status.get("state") == "success" + and status.get("description") == description + and status.get("creator", {}).get("id") == 41898282 + and re.fullmatch( + re.escape(prefix) + r"[0-9]+", status.get("target_url", "") + ) + for status in previous + ) + ): + jobs.extend( + { + "name": status["context"], + "html_url": status["target_url"], + "attempt": run["run_attempt"], + } + for status in previous + ) + continue + if batch is None: + batch = api( + f"repos/{REPOSITORY}/actions/runs/{run['id']}/jobs?filter=latest&per_page={PAGE_SIZE}" + ) + if ( + not isinstance(batch.get("total_count"), int) + or not 0 <= batch["total_count"] <= PAGE_SIZE + or not isinstance(batch.get("jobs"), list) + or batch["total_count"] != len(batch["jobs"]) + ): + return None + selected_jobs = batch["jobs"] + if any( + job.get("head_sha") != sha + or job.get("status") != "completed" + or job.get("run_id") != run["id"] + or job.get("run_attempt") != run["run_attempt"] + for job in selected_jobs + ): + return None + current_jobs = [] + for context in SOURCE_CHECKS[workflow]: + matches = [job for job in selected_jobs if job.get("name") == context] + if len(matches) != 1: + return None + job = matches[0] + if ( + not isinstance(job.get("id"), int) + or job["id"] <= 0 + or job.get("html_url") != f"{prefix}{job['id']}" + ): + return None + record = {**job, "attempt": run["run_attempt"]} + current_jobs.append(record) + if job.get("conclusion") in failures: + evidence["failures"].append(record) + if run.get("conclusion") != "success" or any( + job.get("conclusion") != "success" for job in current_jobs + ): + return None + jobs.extend(current_jobs) + return jobs + + +def reconcile_source( + pr: dict[str, Any], + statuses: dict[str, Any] | None = None, + *, + allow_success: bool = True, +) -> bool: + """Report real source outcomes and keep producer races from admission. + + Returns: + Whether complete required source evidence remained successful and current. + + """ + if ( + set(SOURCE_CHECKS) != set(WORKFLOWS) + or "DCO sign-off" not in SOURCE_CHECKS.get("dco.yml", []) + or not source_identity(pr) + ): + return False + sha, ref = pr["head"]["sha"], f"refs/heads/{pr['head']['ref']}" + statuses = read_statuses(sha) if statuses is None else statuses + SOURCE_EVIDENCE[sha] = {"runs": {}, "failures": []} + initial_event = SOURCE_HEAD == sha and SOURCE_ACTIVITY in { + "requested", + "in_progress", + } + if initial_event and not any( + status.get("description", "").startswith(SOURCE_REPORT) + for status in statuses.values() + ): + return False + jobs = ( + source_jobs(pr, statuses) + if not initial_event and source_contract(pr) is not None + else None + ) + if SOURCE_EVIDENCE[sha]["runs"] and not source_runs_current(pr): + source_jobs(pr, statuses) + if SOURCE_EVIDENCE[sha]["runs"] and not source_runs_current(pr): + SOURCE_EVIDENCE[sha]["failures"] = [] + jobs = None + if not allow_success: + jobs = None + if jobs and source_contract(pr) is None: + jobs = None + SOURCE_EVIDENCE[sha]["failures"] = [] + failed = SOURCE_EVIDENCE[sha]["failures"] + results = { + job["name"]: { + "context": job["name"], + "state": "success", + "target_url": job["html_url"], + "attempt": job["attempt"], + } + for job in jobs or [] + } + results.update( + { + job["name"]: { + "context": job["name"], + "state": "failure", + "target_url": job["html_url"], + "attempt": job["attempt"], + } + for job in failed + } + ) + for status in statuses.values(): + if ( + status.get("description", "").startswith(SOURCE_REPORT) + and not jobs + and status["context"] not in results + ): + results[status["context"]] = { + "context": status["context"], + "state": "pending", + "target_url": status["target_url"], + "attempt": 0, + } + wrote = False + for result in results.values(): + previous = statuses.get(result["context"], {}) + description = f"{SOURCE_REPORT} result; attempt {result['attempt']}" + if ( + previous.get("state") == result["state"] + and previous.get("target_url") == result["target_url"] + and previous.get("description") == description + ): + continue + fresh = api(f"repos/{REPOSITORY}/pulls/{pr['number']}") + if not source_identity(fresh) or fresh["head"]["sha"] != sha: + return False + current = api(f"repos/{REPOSITORY}/git/ref/{ref.removeprefix('refs/')}") + if current["object"]["sha"] != sha: + return False + wrote = True + api( + f"repos/{REPOSITORY}/statuses/{sha}", + "POST", + { + "context": result["context"], + "state": result["state"], + "target_url": result["target_url"], + "description": description, + }, + ) + if jobs and wrote and (not source_runs_current(pr) or source_contract(pr) is None): + reconcile_source(pr, allow_success=False) + return False + return bool(jobs) + + def main() -> None: """Reconcile open automation PRs, then start missing queue validation. @@ -385,6 +790,8 @@ def main() -> None: subprocess.CalledProcessError: An API error other than a deleted ref occurred. """ + SOURCE_CACHE.clear() + SOURCE_EVIDENCE.clear() wait_for_source_run(os.environ.get("SOURCE_RUN_ID", "")) # Pagination is bounded by GitHub's repository PR limit, not a waiting loop. diff --git a/site/mkdocs.yml b/site/mkdocs.yml index 74fe4c1..59324ff 100644 --- a/site/mkdocs.yml +++ b/site/mkdocs.yml @@ -36,6 +36,7 @@ nav: - OpenSSF baseline: openssf-baseline.md - Releases: releases.md - Release runbook: releases/README.md + - v2.11.0 release notes: releases/v2.11.0.md - v2.10.0 release notes: releases/v2.10.0.md - v2.9.0 release notes: releases/v2.9.0.md - v2.8.0 release notes: releases/v2.8.0.md diff --git a/tests/test_reconcile_queue.py b/tests/test_reconcile_queue.py index f80253b..5c7500b 100644 --- a/tests/test_reconcile_queue.py +++ b/tests/test_reconcile_queue.py @@ -890,5 +890,566 @@ def test_rejects_invalid_run_id(self) -> None: api.assert_not_called() +class PackageSourceTests(unittest.TestCase): + """Exercise opted-in source proof with real REST response shapes.""" + + def setUp(self) -> None: + self.sha, self.base = "a" * 40, "b" * 40 + self.branch = "automation/package-updates-example" + self.pr = { + "state": "open", + "draft": False, + "number": 7, + "node_id": "PR_7", + "user": {"login": "github-actions[bot]", "type": "Bot", "id": 41898282}, + "head": { + "sha": self.sha, + "ref": self.branch, + "repo": {"full_name": queue.REPOSITORY}, + }, + "base": { + "ref": queue.DEFAULT_BRANCH, + "repo": {"full_name": queue.REPOSITORY}, + }, + } + self.checks = {"ci.yml": ["Build", "Lint"], "dco.yml": ["DCO sign-off"]} + self.runs = {} + self.jobs = {} + for index, workflow in enumerate(self.checks, 1): + self.runs[workflow] = [ + { + "id": index, + "workflow_id": index, + "path": f".github/workflows/{workflow}", + "head_sha": self.sha, + "head_branch": self.branch, + "event": "workflow_dispatch", + "repository": {"full_name": queue.REPOSITORY}, + "head_repository": {"full_name": queue.REPOSITORY}, + "status": "completed", + "conclusion": "success", + "run_attempt": 1, + } + ] + self.jobs[index] = [ + { + "id": index * 10 + number, + "run_id": index, + "run_attempt": 1, + "name": name, + "head_sha": self.sha, + "status": "completed", + "conclusion": "success", + "html_url": f"https://github.com/{queue.REPOSITORY}/actions/runs/{index}/job/{index * 10 + number}", + } + for number, name in enumerate(self.checks[workflow]) + ] + self.paths = [ + {"filename": "pkgs/by-name/ex/example/source.nix", "status": "modified"}, + {"filename": "docs/catalog.md", "status": "modified"}, + ] + self.parents = [{"sha": self.base}] + self.ancestry = "ahead" + self.tree = { + "truncated": False, + "tree": [ + { + "path": f".github/workflows/{workflow}", + "sha": "c" * 40, + "mode": "100644", + "type": "blob", + } + for workflow in self.checks + ], + } + self.drift = False + self.head = self.sha + self.statuses = [] + self.writes = [] + self.fail_jobs_api = False + self.new_attempt = False + self.new_run = False + self.history_reads = 0 + self.workflow_reads = {} + + def _api(self, path: str, method: str = "GET", data: object = None) -> Any: + if method != "GET": + self.writes.append((path, data)) + return None + if "/git/commits/" in path: + return {"parents": self.parents} + if f"compare/{self.base}...{self.base}" in path: + return {"status": self.ancestry} + if "/compare/" in path: + return {"status": "ahead", "total_commits": 1, "files": self.paths} + if "/git/trees/" in path: + if self.drift and self.sha not in path: + return {"truncated": False, "tree": []} + return self.tree + if "/git/ref/heads/" in path: + return {"object": {"sha": self.head if self.branch in path else self.base}} + if path.endswith(f"/pulls/{self.pr['number']}"): + return self.pr + if "/status?" in path or path.endswith("/status"): + return {"statuses": self.statuses} + if "/files?" in path: + return self.paths + for index, workflow in enumerate(self.checks, 1): + if path.endswith(f"/workflows/{workflow}"): + return { + "id": index, + "path": f".github/workflows/{workflow}", + "state": "active", + } + if f"/workflows/{workflow}/runs?" in path: + self.history_reads += 1 + self.workflow_reads[workflow] = self.workflow_reads.get(workflow, 0) + 1 + if self.new_attempt and self.workflow_reads[workflow] == 2: + self.runs[workflow][0] = { + **self.runs[workflow][0], + "run_attempt": 2, + "status": "in_progress", + "conclusion": None, + } + if self.new_run and self.workflow_reads[workflow] == 2: + self.runs[workflow].insert( + 0, + { + **self.runs[workflow][0], + "id": 3, + "status": "in_progress", + "conclusion": None, + }, + ) + self.jobs[3] = [] + return { + "total_count": len(self.runs[workflow]), + "workflow_runs": self.runs[workflow], + } + if "/jobs?" in path: + if self.fail_jobs_api: + raise subprocess.CalledProcessError(1, ["gh", "api"]) + run_id = int(path.split("/runs/")[1].split("/")[0]) + return {"total_count": len(self.jobs[run_id]), "jobs": self.jobs[run_id]} + if "/actions/runs/" in path: + run_id = int(path.rsplit("/", 1)[1]) + for runs in self.runs.values(): + for run in runs: + if run["id"] == run_id: + return {**run, "run_attempt": 2} if self.new_attempt else run + raise AssertionError(path) + + def _report(self) -> bool: + with ( + patch.object(queue, "api", side_effect=self._api), + patch.object(queue, "SOURCE_CACHE", {}), + patch.object(queue, "SOURCE_EVIDENCE", {}), + patch.object(queue, "SOURCE_CHECKS", self.checks), + patch.object(queue, "WORKFLOWS", list(self.checks)), + ): + return queue.reconcile_source(self.pr) + + def test_real_required_jobs_become_linked_results(self) -> None: + self.assertTrue(self._report()) + self.assertEqual( + [data["context"] for _, data in self.writes], + ["Build", "Lint", "DCO sign-off"], + ) + self.assertTrue(all(data["state"] == "success" for _, data in self.writes)) + self.assertEqual(self.writes[-1][1]["target_url"], self.jobs[2][0]["html_url"]) + + def test_actions_source_admission_requires_explicit_opt_in(self) -> None: + result = subprocess.CompletedProcess([], 0, '[{"bucket":"pass"}]') + with ( + patch.object(queue, "api", side_effect=self._api), + patch.object(queue, "SOURCE_CACHE", {}), + patch.object(queue, "SOURCE_EVIDENCE", {}), + patch.object(queue, "SOURCE_CHECKS", self.checks), + patch.object(queue, "WORKFLOWS", list(self.checks)), + patch.object( + queue, "graphql", return_value={"node": {"mergeQueueEntry": None}} + ) as graphql, + patch.object(queue.subprocess, "run", return_value=result), + ): + self.assertTrue(queue.admit(self.pr)) + self.assertEqual(graphql.call_count, 2) + self.assertEqual(self.writes[-1][1]["context"], queue.ADMISSION) + self.writes.clear() + with ( + patch.object(queue, "SOURCE_CHECKS", {}), + patch.object(queue, "api") as api, + ): + self.assertFalse(queue.admit(self.pr)) + api.assert_not_called() + + def test_wrong_author_fork_draft_and_namespace_are_ineligible(self) -> None: + for key, value in (("id", 9), ("type", "User"), ("login", "someone")): + with self.subTest(key=key): + old = self.pr["user"][key] + self.pr["user"][key] = value + self.assertFalse(queue.source_identity(self.pr)) + self.pr["user"][key] = old + self.pr["head"]["repo"]["full_name"] = "attacker/example" + self.assertFalse(queue.source_identity(self.pr)) + self.pr["head"]["repo"]["full_name"] = queue.REPOSITORY + self.pr["draft"] = True + self.assertFalse(queue.source_identity(self.pr)) + self.pr["draft"] = False + self.pr["head"]["ref"] = "automation/package-updates/example" + self.assertFalse(queue.source_identity(self.pr)) + + def test_wrong_parent_ancestry_scope_and_workflow_are_rejected(self) -> None: + for invalid in ( + "parent", + "ancestry", + "scope", + "rename", + "workflow", + "truncated", + ): + with self.subTest(invalid=invalid): + self.setUp() + if invalid == "parent": + self.parents.append({"sha": "d" * 40}) + if invalid == "ancestry": + self.ancestry = "diverged" + if invalid == "scope": + self.paths.append({"filename": ".github/workflows/dco.yml"}) + if invalid == "rename": + self.paths[0]["status"] = "renamed" + if invalid == "workflow": + self.drift = True + if invalid == "truncated": + self.tree["truncated"] = True + self.assertFalse(self._report()) + self.assertEqual(self.writes, []) + + def test_real_failed_pending_skipped_missing_and_wrong_head_jobs_never_pass( + self, + ) -> None: + for invalid in ( + "failure", + "pending", + "skipped", + "missing", + "head", + "url", + "attempt", + ): + with self.subTest(invalid=invalid): + self.setUp() + if invalid == "failure": + self.runs["dco.yml"][0]["conclusion"] = "failure" + if invalid == "pending": + self.runs["dco.yml"][0]["status"] = "in_progress" + if invalid == "skipped": + self.jobs[2][0]["conclusion"] = "skipped" + if invalid == "missing": + self.jobs[2] = [] + if invalid == "head": + self.jobs[2][0]["head_sha"] = "other" + if invalid == "url": + self.jobs[2][0]["html_url"] = "https://attacker.example/job" + if invalid == "attempt": + self.new_attempt = True + if invalid == "new_run": + self.new_run = True + if invalid == "job_attempt": + self.jobs[2][0]["run_attempt"] = 2 + if invalid == "job_run": + self.jobs[2][0]["run_id"] = 9 + self.assertFalse(self._report()) + self.assertEqual(self.writes, []) + + def test_zero_job_approval_run_does_not_suppress_real_dispatch(self) -> None: + self.runs["dco.yml"].insert( + 0, + { + **self.runs["dco.yml"][0], + "id": 3, + "event": "pull_request", + "conclusion": "action_required", + }, + ) + self.jobs[3] = [] + self.assertTrue(self._report()) + self.jobs[3] = [{**self.jobs[2][0], "conclusion": "failure"}] + self.writes.clear() + self.assertFalse(self._report()) + self.assertEqual(self.writes, []) + + def test_api_error_does_not_write_or_enqueue(self) -> None: + self.fail_jobs_api = True + with self.assertRaises(subprocess.CalledProcessError): + self._report() + self.assertEqual(self.writes, []) + + def test_failed_new_attempt_invalidates_previous_source_success(self) -> None: + self.statuses = [ + { + "context": "DCO sign-off", + "state": "success", + "target_url": self.jobs[2][0]["html_url"], + "description": f"{queue.SOURCE_REPORT} result", + } + ] + self.runs["dco.yml"][0]["conclusion"] = "failure" + self.assertFalse(self._report()) + self.assertEqual([data["state"] for _, data in self.writes], ["pending"]) + + def _attest(self) -> None: + self.statuses = [ + { + "context": job["name"], + "state": "success", + "target_url": job["html_url"], + "description": f"{queue.SOURCE_REPORT} result; attempt 1", + "creator": {"id": 41898282}, + } + for jobs in self.jobs.values() + for job in jobs + ] + + def test_real_failed_new_job_replaces_old_success_with_actual_failure_url( + self, + ) -> None: + self._attest() + self.runs["dco.yml"][0].update(run_attempt=2, conclusion="failure") + self.jobs[2][0].update( + id=25, + run_attempt=2, + conclusion="failure", + html_url=f"https://github.com/{queue.REPOSITORY}/actions/runs/2/job/25", + ) + self.assertFalse(self._report()) + actual = {data["context"]: data for _, data in self.writes} + self.assertEqual(actual["DCO sign-off"]["state"], "failure") + self.assertEqual( + actual["DCO sign-off"]["target_url"], self.jobs[2][0]["html_url"] + ) + self.assertEqual(actual["Build"]["state"], "pending") + + def test_producer_rerun_after_collection_invalidates_greens_and_refuses_admission( + self, + ) -> None: + self._attest() + self.new_attempt = True + with ( + patch.object(queue, "api", side_effect=self._api), + patch.object(queue, "SOURCE_CACHE", {}), + patch.object(queue, "SOURCE_EVIDENCE", {}), + patch.object(queue, "SOURCE_CHECKS", self.checks), + patch.object(queue, "WORKFLOWS", list(self.checks)), + patch.object(queue, "graphql") as graphql, + patch.object(queue.subprocess, "run") as command, + ): + self.assertFalse(queue.admit(self.pr)) + self.assertEqual([data["state"] for _, data in self.writes], ["pending"] * 3) + graphql.assert_not_called() + command.assert_not_called() + + def test_rerun_during_status_publication_is_bounded_and_invalidates_new_greens( + self, + ) -> None: + def api(path: str, method: str = "GET", data: object = None) -> Any: + result = self._api(path, method, data) + if method == "POST": + self.statuses.append({**data, "creator": {"id": 41898282}}) + if len(self.writes) == 1: + self.runs["ci.yml"][0].update( + run_attempt=2, status="in_progress", conclusion=None + ) + return result + + with ( + patch.object(queue, "api", side_effect=api), + patch.object(queue, "SOURCE_CACHE", {}), + patch.object(queue, "SOURCE_EVIDENCE", {}), + patch.object(queue, "SOURCE_CHECKS", self.checks), + patch.object(queue, "WORKFLOWS", list(self.checks)), + ): + self.assertFalse(queue.reconcile_source(self.pr)) + self.assertEqual( + [data["state"] for _, data in self.writes], + ["success"] * 3 + ["pending"] * 3, + ) + + def test_untrusted_cached_status_cannot_replace_actual_job_evidence(self) -> None: + self._attest() + self.statuses[-1]["creator"]["id"] = 9 + self.fail_jobs_api = True + with self.assertRaises(subprocess.CalledProcessError): + self._report() + self.assertEqual(self.writes, []) + + def test_changed_ref_and_missing_dco_configuration_do_not_report(self) -> None: + self.head = "d" * 40 + self.assertFalse(self._report()) + self.assertEqual(self.writes, []) + self.setUp() + self.checks["dco.yml"] = ["other"] + self.assertFalse(self._report()) + self.assertEqual(self.writes, []) + + +class SourceBudgetTests(unittest.TestCase): + """Count API requests for six callers with the actual sixteen-job contract.""" + + def _fixtures(self) -> list[PackageSourceTests]: + checks = { + "ci.yml": [ + "Build (x86_64-linux)", + "Build (aarch64-linux)", + "Build (aarch64-darwin)", + "Lint and security", + "Flake checks (x86_64-linux)", + "Flake checks (aarch64-linux)", + "Flake checks (aarch64-darwin)", + "NUR compatibility / NUR (locked)", + "NUR compatibility / NUR (unstable)", + "Flake lock health", + "Build (x86_64-linux 2/2)", + "Build (aarch64-linux 2/2)", + ], + "codeql.yml": ["Analyze (python) / analyze"], + "dependency-review.yml": ["dependency-review / review"], + "dco.yml": ["DCO sign-off"], + "fuzz.yml": ["Fuzz font path validation"], + } + fixtures = [] + for number in range(7, 13): + fixture = PackageSourceTests() + fixture.setUp() + fixture.checks = checks + fixture.sha = f"{number:040x}" + fixture.head = fixture.sha + fixture.branch = f"automation/package-updates-example{number}" + fixture.pr["number"] = number + fixture.pr["head"].update(sha=fixture.sha, ref=fixture.branch) + fixture.paths[0]["filename"] = f"pkgs/by-name/ex/example{number}/source.nix" + fixture.tree["tree"] = [ + { + "path": f".github/workflows/{workflow}", + "sha": "c" * 40, + "mode": "100644", + "type": "blob", + } + for workflow in checks + ] + fixture.runs, fixture.jobs = {}, {} + for workflow_id, workflow in enumerate(checks, 1): + run_id = number * 10 + workflow_id + fixture.runs[workflow] = [ + { + "id": run_id, + "workflow_id": workflow_id, + "path": f".github/workflows/{workflow}", + "head_sha": fixture.sha, + "head_branch": fixture.branch, + "event": "workflow_dispatch", + "repository": {"full_name": queue.REPOSITORY}, + "head_repository": {"full_name": queue.REPOSITORY}, + "status": "completed", + "conclusion": "success", + "run_attempt": 1, + } + ] + fixture.jobs[run_id] = [ + { + "id": run_id * 100 + index, + "run_id": run_id, + "run_attempt": 1, + "name": name, + "head_sha": fixture.sha, + "status": "completed", + "conclusion": "success", + "html_url": f"https://github.com/{queue.REPOSITORY}/actions/runs/{run_id}/job/{run_id * 100 + index}", + } + for index, name in enumerate(checks[workflow]) + ] + fixture._attest() + fixtures.append(fixture) + return fixtures + + def _case(self, mode: str) -> dict[str, int]: + fixtures = self._fixtures() + counts = {"get": 0, "write": 0, "jobs": 0} + if mode in {"queued", "admitted_event"}: + for fixture in fixtures: + fixture.statuses.append( + {"context": queue.ADMISSION, "state": "success"} + ) + if mode == "initial_event": + fixtures[0].statuses = [] + + def api(path: str, method: str = "GET", data: object = None) -> Any: + counts["get" if method == "GET" else "write"] += 1 + if "/jobs?" in path: + counts["jobs"] += 1 + for fixture in fixtures: + if ( + fixture.sha in path + or fixture.branch in path + or path.endswith(f"/pulls/{fixture.pr['number']}") + or f"/pulls/{fixture.pr['number']}/files?" in path + ): + return fixture._api(path, method, data) + for run_id in fixture.jobs: + if f"/runs/{run_id}/" in path or path.endswith(f"/runs/{run_id}"): + return fixture._api(path, method, data) + return fixtures[0]._api(path, method, data) + + def graphql(query: str, **variables: str) -> dict[str, Any]: + counts["get"] += 1 + return {"node": {"mergeQueueEntry": {"id": "queue"}}} + + head = fixtures[0].sha if mode.endswith("event") else "" + activity = ( + "in_progress" + if mode in {"admitted_event", "initial_event"} + else "completed" + ) + with ( + patch.object(queue, "api", side_effect=api), + patch.object(queue, "graphql", side_effect=graphql), + patch.object(queue, "SOURCE_CACHE", {}), + patch.object(queue, "SOURCE_EVIDENCE", {}), + patch.object(queue, "SOURCE_CHECKS", fixtures[0].checks), + patch.object(queue, "WORKFLOWS", list(fixtures[0].checks)), + patch.object(queue, "SOURCE_HEAD", head), + patch.object(queue, "SOURCE_ACTIVITY", activity), + ): + for fixture in fixtures: + if mode == "idle": + self.assertTrue(queue.reconcile_source(fixture.pr)) + else: + queue.admit(fixture.pr) + return counts + + def test_six_idle_source_prs_fit_hourly_recovery_budget(self) -> None: + counts = self._case("idle") + self.assertLessEqual(counts["get"], 110) + self.assertEqual(counts["write"], 0) + self.assertEqual(counts["jobs"], 0) + + def test_six_admitted_sources_skip_unrelated_polling(self) -> None: + self.assertEqual(self._case("queued"), {"get": 12, "write": 0, "jobs": 0}) + + def test_source_event_inspects_only_its_exact_head(self) -> None: + counts = self._case("completed_event") + self.assertLessEqual(counts["get"], 25) + self.assertEqual(counts["write"], 0) + self.assertEqual(counts["jobs"], 0) + + def test_rerun_event_invalidates_even_an_admitted_source(self) -> None: + counts = self._case("admitted_event") + self.assertEqual(counts, {"get": 33, "write": 16, "jobs": 0}) + + def test_initial_requested_event_with_no_attestation_stops_immediately( + self, + ) -> None: + self.assertEqual(self._case("initial_event"), {"get": 1, "write": 0, "jobs": 0}) + + if __name__ == "__main__": unittest.main()