From eddef8bb52f5fc04c4f6fe7b8bac8a130de8d576 Mon Sep 17 00:00:00 2001 From: Tee8z Date: Tue, 20 Oct 2026 17:35:25 -0400 Subject: [PATCH] feat(ci): add central reviewed-changes verifier --- .github/workflows/README.md | 18 ++ .github/workflows/reviewed-changes.yml | 279 +++++++++++++++++++++++++ 2 files changed, 297 insertions(+) create mode 100644 .github/workflows/README.md create mode 100644 .github/workflows/reviewed-changes.yml diff --git a/.github/workflows/README.md b/.github/workflows/README.md new file mode 100644 index 0000000..33f92f2 --- /dev/null +++ b/.github/workflows/README.md @@ -0,0 +1,18 @@ +# Reviewed changes workflow + +This workflow is the central source for Voltage's required approval-equivalence +check. Organization rulesets select it from this repository's protected `master` +branch. Keep native stale-review dismissal enabled on this source repository. + +The Python source, tests, generator, consumer event helpers, and deployment +instructions live in +[voltage-terraform](https://github.com/voltagecloud/voltage-terraform/tree/master/v2/environments/github/reviewed_changes) +and its [GitHub environment README](https://github.com/voltagecloud/voltage-terraform/blob/master/v2/environments/github/README.md). +Regenerate the workflow there, then review the generated update in this repository. + +Merging this workflow does not activate approval retention. Install and verify the +consumer helpers and required organization ruleset before changing native review +settings. The workflow blocks merging when it cannot establish equivalence. + +The source repository is excluded from this gate and uses native review +protection. The verifier job skips source-repository PRs. diff --git a/.github/workflows/reviewed-changes.yml b/.github/workflows/reviewed-changes.yml new file mode 100644 index 0000000..fe13ac9 --- /dev/null +++ b/.github/workflows/reviewed-changes.yml @@ -0,0 +1,279 @@ +# Generated by v2/environments/github/reviewed_changes/generate.py; edit its Python sources. +name: Reviewed changes +on: + pull_request: + merge_group: +permissions: + contents: read + pull-requests: read +jobs: + verify: + name: Verify reviewed changes + # The source repository keeps native stale-review dismissal and is excluded + # from approval retention. Consumers must always run the verifier. + if: github.repository != 'voltagecloud/.github' + runs-on: ubuntu-latest + timeout-minutes: 15 + steps: + - name: Compare approved and current content + env: + GH_TOKEN: ${{ github.token }} + shell: python + run: | + #!/usr/bin/env python3 + """Validate approval content without checking out or executing pull-request code.""" + + import base64 + import json + import os + import re + import subprocess + import sys + import tempfile + import urllib.error + import urllib.parse + import urllib.request + from pathlib import Path + + + class ReviewRequired(Exception): + pass + + + def sha(value): + if not isinstance(value, str) or not re.fullmatch(r"[0-9a-f]{40}", value): + raise ReviewRequired("Missing or invalid commit identity.") + return value + + + class GitHub: + def __init__(self, token, repository): + if not re.fullmatch(r"voltagecloud/[A-Za-z0-9_.-]+", repository): + raise ReviewRequired("Unexpected repository.") + self.token = token + self.repository = repository + + def request(self, path, data=None): + url = "https://api.github.com" + path + request = urllib.request.Request( + url, + data=None if data is None else json.dumps(data).encode(), + headers={ + "Authorization": "Bearer " + self.token, + "Accept": "application/vnd.github+json", + "X-GitHub-Api-Version": "2022-11-28", + "Content-Type": "application/json", + }, + ) + try: + with urllib.request.urlopen(request, timeout=30) as response: + body = response.read() + return json.loads(body) if body else None + except urllib.error.HTTPError as error: + raise ReviewRequired(f"GitHub returned HTTP {error.code}; cannot verify approval.") from None + + def pages(self, path): + separator = "&" if "?" in path else "?" + for page in range(1, 101): + result = self.request(f"{path}{separator}per_page=100&page={page}") + if not isinstance(result, list): + raise ReviewRequired("Unexpected paginated GitHub response.") + yield from result + if len(result) < 100: + return + raise ReviewRequired("Review history exceeded the pagination limit.") + + def snapshot(self, number): + prefix = f"/repos/{self.repository}" + pr = self.request(f"{prefix}/pulls/{number}") + reviews = list(self.pages(f"{prefix}/pulls/{number}/reviews")) + owner, name = self.repository.split("/") + result = self.request("/graphql", { + "query": """query($owner:String!,$name:String!,$number:Int!) { + repository(owner:$owner,name:$name) { + pullRequest(number:$number) { reviewDecision } + } + }""", + "variables": {"owner": owner, "name": name, "number": number}, + }) + if result.get("errors"): + raise ReviewRequired("Cannot read GitHub's required-review decision.") + decision = result["data"]["repository"]["pullRequest"]["reviewDecision"] + opinions = latest_opinions(reviews) + permissions = {} + for review in opinions.values(): + user = review["user"] + if user["type"] == "User" and user["login"] != pr["user"]["login"]: + login = urllib.parse.quote(user["login"], safe="") + permissions[user["login"]] = self.request( + f"{prefix}/collaborators/{login}/permission" + )["permission"] + return { + "id": pr["id"], "state": pr["state"], "draft": pr["draft"], + "author": pr["user"]["login"], + "head": sha(pr["head"]["sha"]), "base": sha(pr["base"]["sha"]), + "base_ref": pr["base"]["ref"], + "base_repository": pr["base"]["repo"]["full_name"], + "decision": decision, "opinions": opinions, "permissions": permissions, + } + + + def latest_opinions(reviews): + """A comment does not withdraw an approval; a dismissal or rejection does.""" + result = {} + for review in sorted(reviews, key=lambda item: item["id"]): + if review["state"] in ("COMMENTED", "PENDING"): + continue + if review["state"] not in ("APPROVED", "DISMISSED", "CHANGES_REQUESTED"): + raise ReviewRequired("Unknown review state.") + if not review.get("user"): + raise ReviewRequired("A review has no identifiable author.") + result[review["user"]["login"]] = { + "id": review["id"], "state": review["state"], + "commit_id": review.get("commit_id"), + "user": {"login": review["user"]["login"], "type": review["user"]["type"]}, + } + return result + + + def approvals(snapshot): + if snapshot["state"] != "open" or snapshot["draft"]: + raise ReviewRequired("The pull request must be open and ready for review.") + if snapshot["decision"] != "APPROVED": + raise ReviewRequired("GitHub's required reviews, including code-owner reviews, are not satisfied.") + result = [] + for login, review in snapshot["opinions"].items(): + if login == snapshot["author"] or review["user"]["type"] != "User": + continue + if snapshot["permissions"].get(login) not in ("write", "maintain", "admin"): + continue + if review["state"] == "CHANGES_REQUESTED": + raise ReviewRequired("An eligible reviewer has requested changes.") + if review["state"] == "APPROVED": + sha(review["commit_id"]) + result.append(review) + if not result: + raise ReviewRequired("A human reviewer with write access must approve the changes.") + return result + + + class GitTrees: + def __init__(self, directory, token=None, repository=None): + self.directory = str(directory) + self.environment = dict(os.environ) + # Ignore ambient drivers, hooks, replacement objects, and configuration. + for key in list(self.environment): + if key.startswith("GIT_"): + del self.environment[key] + self.environment.update({ + "GIT_CONFIG_NOSYSTEM": "1", "GIT_CONFIG_GLOBAL": os.devnull, + "GIT_ATTR_NOSYSTEM": "1", "GIT_TERMINAL_PROMPT": "0", + "GIT_NO_REPLACE_OBJECTS": "1", + }) + if token is not None: + credential = base64.b64encode(("x-access-token:" + token).encode()).decode() + self.environment.update({ + "GIT_CONFIG_COUNT": "1", "GIT_CONFIG_KEY_0": "http.extraheader", + "GIT_CONFIG_VALUE_0": "AUTHORIZATION: basic " + credential, + }) + self.git("init", "--bare", "--quiet", self.directory, outside=True) + self.git("remote", "add", "origin", f"https://github.com/{repository}.git") + self.trees = {} + + def git(self, *arguments, outside=False): + command = ["git", "-c", "core.hooksPath=" + os.devnull] + if not outside: + command += ["-C", self.directory] + result = subprocess.run( + command + list(arguments), env=self.environment, capture_output=True, + text=True, timeout=180, + ) + if result.returncode: + # Do not echo commands, credentials, conflict contents, or untrusted paths. + raise ReviewRequired("Git could not establish a clean, unambiguous merge; request a fresh review.") + return result.stdout.strip() + + def fetch(self, commits): + for commit in sorted(set(commits)): + self.git("fetch", "--quiet", "--no-tags", "--no-recurse-submodules", "origin", sha(commit)) + if self.git("cat-file", "-t", commit) != "commit": + raise ReviewRequired("A review does not identify an available commit.") + + def merged_tree(self, commit, base): + key = (sha(commit), sha(base)) + if key not in self.trees: + bases = self.git("merge-base", "--all", commit, base).splitlines() + if len(bases) != 1: + raise ReviewRequired("The comparison has no unique merge base; request a fresh review.") + self.trees[key] = sha(self.git("merge-tree", "--write-tree", commit, base).splitlines()[0]) + return self.trees[key] + + + def verify(snapshot, trees): + reviews = approvals(snapshot) + current_tree = trees.merged_tree(snapshot["head"], snapshot["base"]) + evidence = [] + # Every effective human approval must cover the content. This also prevents + # an obsolete code-owner approval satisfying GitHub alongside a fresh review. + for review in reviews: + approved_tree = trees.merged_tree(review["commit_id"], snapshot["base"]) + if approved_tree != current_tree: + raise ReviewRequired( + f"Review {review['id']} covers different content. Its reviewer must reapprove, " + "or an authorized maintainer must dismiss it before the required fresh review." + ) + evidence.append({ + "review_id": review["id"], "reviewer": review["user"]["login"], + "approved_commit": review["commit_id"], "head_commit": snapshot["head"], + "base_commit": snapshot["base"], "merged_tree": current_tree, + }) + return evidence + + + def run(github, event, trees_factory): + if "pull_request" not in event: + raise ReviewRequired("Only individual pull requests are supported; merge queues remain blocked.") + trigger = event["pull_request"] + number = trigger["number"] + if type(number) is not int or number < 1: + raise ReviewRequired("Invalid pull request number.") + snapshot = github.snapshot(number) + # Reruns after retargeting use the current base and current native review + # decision. Only the PR identity and head must match the original event. + if (snapshot["id"] != trigger["id"] or snapshot["head"] != sha(trigger["head"]["sha"]) + or snapshot["base_repository"] != github.repository): + raise ReviewRequired("The pull request changed since this run was triggered; use its latest run.") + reviews = approvals(snapshot) + with tempfile.TemporaryDirectory(prefix="reviewed-changes-") as directory: + trees = trees_factory(directory) + trees.fetch([snapshot["base"], snapshot["head"]] + [r["commit_id"] for r in reviews]) + evidence = verify(snapshot, trees) + # A late review, dismissal, permission change, base update, or push invalidates + # this result. The refresh workflow reruns using current API state. + if github.snapshot(number) != snapshot: + raise ReviewRequired("The pull request or its reviews changed during verification; rerun the check.") + return evidence + + + def main(): + token = os.environ["GH_TOKEN"] + repository = os.environ["GITHUB_REPOSITORY"] + event = json.loads(Path(os.environ["GITHUB_EVENT_PATH"]).read_text()) + github = GitHub(token, repository) + try: + evidence = run(github, event, lambda path: GitTrees(path, token, repository)) + except (ReviewRequired, KeyError, TypeError, ValueError, OSError, + subprocess.SubprocessError, urllib.error.URLError) as error: + message = str(error) if isinstance(error, ReviewRequired) else "Verification failed; approval equivalence is unknown." + print(message) + return 1 + report = "Reviewed changes are equivalent.\n\n```json\n" + json.dumps(evidence, indent=2) + "\n```\n" + print(report) + if os.environ.get("GITHUB_STEP_SUMMARY"): + with open(os.environ["GITHUB_STEP_SUMMARY"], "a") as summary: + summary.write(report) + return 0 + + + if __name__ == "__main__": + sys.exit(main())