feat(ci): add central reviewed-changes verifier

.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.

.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())