Repository navigation
🔒 [security] approval: bind human sign-off to reviewed code #23
There are no files selected for viewing
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
|
Owner
Author
There was a problem hiding this comment. Author explanation Purpose What changed Why it matters |
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,248 @@ | ||
| #!/usr/bin/env python3 | ||
| """Read-only, head-bound human approval shared by the planner and merge gate. | ||
|
|
||
| A trusted request pins the code and human steps BEFORE a human replies `done`. | ||
| Labels only describe UI state. They never authorize a merge. No GitHub writes. | ||
| """ | ||
| from __future__ import annotations | ||
|
|
||
| import argparse | ||
| from datetime import datetime | ||
| import hashlib | ||
| import json | ||
| import re | ||
| import subprocess | ||
| import sys | ||
|
|
||
| PREFIX = "approval-request: " | ||
| SHA = re.compile(r"[0-9a-f]{40}\Z") | ||
| SCOPE = re.compile(r"[0-9a-f]{64}\Z") | ||
| BLOCK = re.compile(r"Reason tag:[^\n]*🙋|^approval-request:", re.M) | ||
| DONE = re.compile(r"\s*done[.!]?\s*\Z", re.I) | ||
|
|
||
|
|
||
| def scope_for(plan): | ||
| # Exclude UI-only `done`; a label change must not change what was approved. | ||
| relevant = {key: plan.get(key, []) for key in ("human", "migrations", "run", "needs_you")} | ||
| return hashlib.sha256(json.dumps(relevant, sort_keys=True, separators=(",", ":")).encode()).hexdigest() | ||
|
|
||
|
|
||
| def request_for(repo, pr, head, scope): | ||
| return {"repo": repo.lower(), "pr": int(pr), "head": head, "scope": scope} | ||
|
|
||
|
|
||
| def marker(request): | ||
| return PREFIX + json.dumps(request, sort_keys=True, separators=(",", ":")) | ||
|
|
||
|
|
||
| def parse_request(comment): | ||
| lines = [line for line in (comment.get("body") or "").splitlines() if line.startswith(PREFIX)] | ||
| if len(lines) != 1: | ||
| return None | ||
| try: | ||
| value = json.loads(lines[0][len(PREFIX):]) | ||
| if (not isinstance(value, dict) or not isinstance(value.get("repo"), str) | ||
| or not re.fullmatch(r"[^/\s]+/[^/\s]+", value["repo"]) | ||
| or type(value.get("pr")) is not int or value["pr"] < 1 | ||
| or not SHA.fullmatch(value.get("head", "")) | ||
| or not SCOPE.fullmatch(value.get("scope", ""))): | ||
| return None | ||
| return value | ||
| except (ValueError, TypeError): | ||
| return None | ||
|
|
||
|
|
||
| def timestamp(value): | ||
| if not isinstance(value, str): | ||
| raise ValueError("missing comment timestamp") | ||
| result = datetime.fromisoformat(value.replace("Z", "+00:00")) | ||
| if result.tzinfo is None: | ||
| raise ValueError("comment timestamp lacks timezone") | ||
| return result | ||
|
|
||
|
|
||
| def trusted_blocks(comments, permission): | ||
| result = [] | ||
| for block in comments: | ||
| if not BLOCK.search(block.get("body") or ""): | ||
| continue | ||
| user = block.get("user") or {} | ||
| if not user.get("login") or user.get("type") not in ("User", "Bot"): | ||
| raise ValueError("requester identity could not be verified") | ||
| if permission(user["login"]) in ("write", "maintain", "admin"): | ||
| result.append(block) | ||
| return result | ||
|
|
||
|
|
||
| def validate(comments, expected, permission): | ||
| """Pure verdict; permission(login) is a current repository permission lookup.""" | ||
| def hold(why): | ||
| return {"approved": False, "why": why} | ||
|
|
||
| if not SHA.fullmatch(expected.get("head", "")): | ||
| return hold("current PR head is unavailable") | ||
| try: | ||
| blocks = trusted_blocks(comments, permission) | ||
| if not blocks: | ||
| return hold("no request from a trusted repository collaborator") | ||
| # Any later human block supersedes the old request, including one with | ||
| # missing/malformed metadata. Equal-second requests are ambiguous: hold. | ||
| latest_time = max(timestamp(c.get("created_at")) for c in blocks) | ||
| latest = [c for c in blocks if timestamp(c.get("created_at")) == latest_time] | ||
| if len(latest) != 1: | ||
| return hold("ambiguous newest human request") | ||
| request_comment = latest[0] | ||
| request = parse_request(request_comment) | ||
| if request is None or any(request.get(k) != v for k, v in expected.items()): | ||
| return hold("the latest request does not cover the current code and human steps") | ||
| if timestamp(request_comment.get("updated_at")) != latest_time: | ||
| return hold("approval request was edited; post a fresh request") | ||
|
Comment on lines
+96
to
+99
Owner
Author
There was a problem hiding this comment. Author explanation Purpose What changed Why it matters |
||
| requester = request_comment.get("user") or {} | ||
| if not requester.get("login") or permission(requester["login"]) not in ("write", "maintain", "admin"): | ||
| return hold("requester authority could not be verified") | ||
| if not isinstance(request_comment.get("source"), int) or not request_comment.get("id"): | ||
| return hold("request identity is missing") | ||
| for comment in comments: | ||
| if (not comment.get("id") or not DONE.fullmatch(comment.get("body") or "") | ||
| or comment.get("source") != request_comment.get("source")): | ||
| continue | ||
| created = timestamp(comment.get("created_at")) | ||
| if created <= latest_time or timestamp(comment.get("updated_at")) != created: | ||
| continue | ||
| author = comment.get("user") or {} | ||
| if (author.get("type") == "User" and author.get("login") | ||
| and permission(author["login"]) in ("write", "maintain", "admin")): | ||
| return {"approved": True, "why": "trusted human confirmed the current request", | ||
| "request": request, "requestId": request_comment.get("id"), | ||
| "approvalId": comment.get("id"), "approver": author["login"]} | ||
|
Comment on lines
+105
to
+117
Owner
Author
There was a problem hiding this comment. Author explanation Purpose What changed Why it matters |
||
| return hold("waiting for a trusted human to comment done on the current request") | ||
| except (ValueError, TypeError, KeyError): | ||
| return hold("approval evidence is incomplete or unreadable") | ||
|
|
||
|
|
||
| class GitHub: | ||
| def __init__(self, repo): | ||
| if not re.fullmatch(r"[A-Za-z0-9_.-]+/[A-Za-z0-9_.-]+", repo): | ||
| raise ValueError("missing or invalid repository") | ||
| self.repo = repo.lower() | ||
| self.permissions = {} | ||
|
|
||
| @staticmethod | ||
| def read(*args): | ||
| result = subprocess.run(["gh", *args], capture_output=True, text=True, timeout=30) | ||
| if result.returncode: | ||
| raise ValueError("GitHub evidence could not be read") | ||
| return json.loads(result.stdout) | ||
|
|
||
| def permission(self, login): | ||
| if not re.fullmatch(r"[A-Za-z0-9_-]+(?:\[bot\])?", login): | ||
| return None | ||
| if login not in self.permissions: | ||
| value = self.read("api", f"repos/{self.repo}/collaborators/{login}/permission") | ||
| self.permissions[login] = value.get("permission") | ||
| return self.permissions[login] | ||
|
|
||
| def comments(self, number): | ||
| pages = self.read("api", "--paginate", "--slurp", | ||
| f"repos/{self.repo}/issues/{number}/comments?per_page=100") | ||
| if not isinstance(pages, list) or not pages or any(not isinstance(p, list) for p in pages): | ||
| raise ValueError("unreadable comments") | ||
| if any(not isinstance(c, dict) for page in pages for c in page): | ||
| raise ValueError("unreadable comment") | ||
| return [dict(comment, source=number) for page in pages for comment in page] | ||
|
|
||
| def pr(self, number): | ||
| # Paginate linked issues too: an omitted newer request must never make | ||
| # an older approval authoritative. Remote-repository links are ignored. | ||
| query = '''query($owner:String!,$repo:String!,$pr:Int!,$endCursor:String){ | ||
| repository(owner:$owner,name:$repo){pullRequest(number:$pr){headRefOid state | ||
| closingIssuesReferences(first:100,after:$endCursor){ | ||
| pageInfo{hasNextPage endCursor} nodes{number repository{nameWithOwner}} | ||
| }}}}''' | ||
| owner, name = self.repo.split("/") | ||
| pages = self.read("api", "graphql", "--paginate", "--slurp", "-f", f"query={query}", | ||
| "-F", f"owner={owner}", "-F", f"repo={name}", "-F", f"pr={number}") | ||
| heads, issues = set(), set() | ||
| if not isinstance(pages, list) or not pages: | ||
| raise ValueError("unreadable PR") | ||
| for page in pages: | ||
| if page.get("errors"): | ||
| raise ValueError("incomplete PR response") | ||
| pr = page["data"]["repository"]["pullRequest"] | ||
| if pr["state"] != "OPEN": | ||
| raise ValueError("PR is not open") | ||
| heads.add(pr["headRefOid"]) | ||
| links = pr["closingIssuesReferences"] | ||
| for issue in links["nodes"]: | ||
| if issue["repository"]["nameWithOwner"].lower() == self.repo: | ||
| issues.add(issue["number"]) | ||
| if links["pageInfo"]["hasNextPage"] or len(heads) != 1: | ||
| raise ValueError("PR changed or issue links are incomplete") | ||
| head = heads.pop() | ||
| if not SHA.fullmatch(head): | ||
| raise ValueError("unreadable PR head") | ||
| return head, issues | ||
|
|
||
| def check(self, number, head=None, scope=None, issue=None): | ||
| current_head, issues = self.pr(number) | ||
| if issue is not None and issue not in issues: | ||
| return {"approved": False, "why": "request issue is not linked to this PR"} | ||
| if head is not None and current_head != head: | ||
| return {"approved": False, "headChanged": True, "why": "PR head changed after review"} | ||
| comments = [] | ||
| for source in sorted(issues | {number}): | ||
| comments.extend(self.comments(source)) | ||
| expected = {"repo": self.repo, "pr": number, "head": current_head} | ||
| if scope is not None: | ||
| expected["scope"] = scope | ||
| return validate(comments, expected, self.permission) | ||
|
|
||
|
|
||
| def main(): | ||
| ap = argparse.ArgumentParser(description=__doc__) | ||
| ap.add_argument("--repo", required=True) | ||
| ap.add_argument("--pr", type=int) | ||
| ap.add_argument("--head") | ||
| ap.add_argument("--plan", help="JSON plan from merge-policy") | ||
| ap.add_argument("--request", action="store_true", help="print a pinned request; no network or writes") | ||
| ap.add_argument("--deps", action="store_true", help="enrich the dependency graph on stdin") | ||
| args = ap.parse_args() | ||
| if args.request: | ||
| if not args.pr or not args.head or not SHA.fullmatch(args.head) or not args.plan: | ||
| ap.error("request requires --pr, full --head and --plan") | ||
| print(marker(request_for(args.repo, args.pr, args.head, scope_for(json.loads(args.plan))))) | ||
| return | ||
| if args.deps: | ||
| graph = json.load(sys.stdin) | ||
| github = GitHub(args.repo) if args.repo else None | ||
| for entry in graph.values(): | ||
| entry["needsYouDone"] = False | ||
| entry["approvalRefresh"] = False | ||
| if not entry.get("needsYou") or github is None: | ||
| continue | ||
| try: | ||
| comments = github.comments(entry["number"]) | ||
| blocks = trusted_blocks(comments, github.permission) | ||
| latest = max(blocks, key=lambda c: timestamp(c.get("created_at"))) if blocks else {} | ||
| request = parse_request(latest) | ||
| if request is None or request["repo"] != github.repo: | ||
| continue | ||
| result = github.check(request["pr"], head=request["head"], issue=entry["number"]) | ||
| entry["needsYouDone"] = result["approved"] | ||
| entry["approvalRefresh"] = result.get("headChanged", False) | ||
| entry["approvalWhy"] = result["why"] | ||
| except (ValueError, KeyError, TypeError, OSError, subprocess.SubprocessError): | ||
| entry["approvalWhy"] = "approval evidence could not be verified" | ||
| print(json.dumps(graph)) | ||
| return | ||
| if not args.pr or not args.head or not args.plan: | ||
| ap.error("check requires --pr, --head and --plan") | ||
| try: | ||
| result = GitHub(args.repo).check(args.pr, args.head, scope_for(json.loads(args.plan))) | ||
| except (ValueError, KeyError, TypeError, OSError, subprocess.SubprocessError): | ||
| result = {"approved": False, "why": "approval evidence could not be verified"} | ||
| print(json.dumps(result)) | ||
|
|
||
|
|
||
| if __name__ == "__main__": | ||
| main() | ||
|
Owner
Author
There was a problem hiding this comment. Author explanation Purpose What changed Why it matters |
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
|
Owner
Author
There was a problem hiding this comment. Author explanation Purpose What changed Why it matters |
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Oops, something went wrong.
Oops, something went wrong.
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Author explanation
Purpose
Install Super Board’s tools into a project.
What changed
Copy the new approval checker alongside the existing command helpers.
Why it matters
An installed project needs this checker for the planner and merge checks to use the new approval rules.