|
| 1 | +#!/usr/bin/env python3 |
| 2 | +from __future__ import annotations |
| 3 | +import fnmatch,json,os,sys,urllib.error,urllib.parse,urllib.request |
| 4 | +from dataclasses import dataclass |
| 5 | +API="https://api.github.com"; SAFE={"success","neutral","skipped"} |
| 6 | +class Block(RuntimeError): pass |
| 7 | +class Error(RuntimeError): pass |
| 8 | +def env(n): |
| 9 | + v=os.environ.get(n,"").strip() |
| 10 | + if not v: raise Error(f"{n} is empty") |
| 11 | + return v |
| 12 | +def csv(n): return tuple(x.strip() for x in os.environ.get(n,"").split(",") if x.strip()) |
| 13 | +@dataclass(frozen=True) |
| 14 | +class Policy: |
| 15 | + repo:str; required:tuple[str,...]; groups:tuple[str,...]; paths:tuple[str,...]; method:str; max_files:int=25 |
| 16 | + @classmethod |
| 17 | + def load(cls): |
| 18 | + p=tuple(x.strip() for x in os.environ.get("GOVERNOR_ALLOWED_PATHS","").splitlines() if x.strip()); o=cls(env("GITHUB_REPOSITORY"),csv("GOVERNOR_REQUIRED_WORKFLOWS"),csv("GOVERNOR_ALLOWED_GROUPS"),p,os.environ.get("GOVERNOR_MERGE_METHOD","merge").strip()) |
| 19 | + if not o.required or not o.groups or not o.paths: raise Error("policy lists must not be empty") |
| 20 | + if o.method not in {"merge","rebase","squash"}: raise Error(f"bad merge method {o.method}") |
| 21 | + return o |
| 22 | +class GH: |
| 23 | + def __init__(self,t): self.t=t |
| 24 | + def call(self,m,p,x=None): |
| 25 | + q=urllib.request.Request(API+p,data=None if x is None else json.dumps(x).encode(),method=m,headers={"Accept":"application/vnd.github+json","Authorization":f"Bearer {self.t}","X-GitHub-Api-Version":"2022-11-28","User-Agent":"dependabot-governor"}) |
| 26 | + try: |
| 27 | + with urllib.request.urlopen(q,timeout=30) as r: d=r.read() |
| 28 | + except urllib.error.HTTPError as e: raise Error(f"GitHub API {m} {p} failed {e.code}: {e.read().decode('utf-8','replace')[:500]}") from e |
| 29 | + return json.loads(d) if d else None |
| 30 | + def get(self,p): return self.call("GET",p) |
| 31 | + def put(self,p,x): return self.call("PUT",p,x) |
| 32 | +def event(): |
| 33 | + with open(env("GITHUB_EVENT_PATH"),encoding="utf-8") as f: return json.load(f) |
| 34 | +def resolve(e): |
| 35 | + n=env("GITHUB_EVENT_NAME") |
| 36 | + if n=="workflow_dispatch": |
| 37 | + v=str(e.get("inputs",{}).get("pr_number","")).strip() |
| 38 | + if not v.isdigit(): raise Error("workflow_dispatch requires numeric pr_number") |
| 39 | + return int(v),None,None |
| 40 | + if n!="workflow_run": raise Block(f"{n} is not a governance event") |
| 41 | + r=e.get("workflow_run") or {}; ps=r.get("pull_requests") or [] |
| 42 | + if r.get("event")!="pull_request": raise Block("triggering workflow was not a pull_request run") |
| 43 | + if len(ps)!=1 or not isinstance(ps[0].get("number"),int): raise Block("workflow run is not associated with exactly one PR") |
| 44 | + return ps[0]["number"],r.get("head_sha"),((ps[0].get("base") or {}).get("sha")) |
| 45 | +def routine(t,g): return any(x.lower() in t.lower() for x in g) |
| 46 | +def allowed(p,patterns): return any(fnmatch.fnmatchcase(p,x) for x in patterns) |
| 47 | +def files_ok(items,p): |
| 48 | + n=[str(x.get("filename","")) for x in items] |
| 49 | + if not n: raise Block("PR has no files") |
| 50 | + if len(n)>p.max_files: raise Block(f"PR changes {len(n)} files; limit is {p.max_files}") |
| 51 | + b=[x for x in n if not allowed(x,p.paths)] |
| 52 | + if b: raise Block("non-dependency file scope: "+", ".join(b)) |
| 53 | + return n |
| 54 | +def runs_ok(runs,required,main): |
| 55 | + latest={} |
| 56 | + for r in runs: |
| 57 | + p=str(r.get("path","")) |
| 58 | + if p and p not in latest: latest[p]=r |
| 59 | + m=[p for p in required if p not in latest] |
| 60 | + if m: raise Block("required workflows have not started: "+", ".join(m)) |
| 61 | + for p in required: |
| 62 | + r=latest[p] |
| 63 | + if r.get("status")!="completed": raise Block(f"required workflow still running: {p}") |
| 64 | + if r.get("conclusion")!="success": raise Block(f"required workflow failed: {p}={r.get('conclusion')}") |
| 65 | + ps=r.get("pull_requests") or []; base=((ps[0].get("base") or {}).get("sha")) if len(ps)==1 else None |
| 66 | + if base!=main: raise Block(f"{p} tested obsolete base {base or 'unknown'}") |
| 67 | + for r in runs: |
| 68 | + p=str(r.get("path","<unknown>")) |
| 69 | + if r.get("status")!="completed": raise Block(f"PR workflow still running: {p}") |
| 70 | + if r.get("conclusion") not in SAFE: raise Block(f"PR workflow not green: {p}={r.get('conclusion')}") |
| 71 | +def summary(lines): |
| 72 | + f=os.environ.get("GITHUB_STEP_SUMMARY") |
| 73 | + if f: |
| 74 | + with open(f,"a",encoding="utf-8") as h: h.write("\n".join(lines)+"\n") |
| 75 | +def govern(): |
| 76 | + p=Policy.load(); gh=GH(env("GITHUB_TOKEN")); number,ts,tb=resolve(event()); pr=gh.get(f"/repos/{p.repo}/pulls/{number}") |
| 77 | + if pr.get("state")!="open": raise Block("PR is not open") |
| 78 | + if pr.get("draft"): raise Block("draft PR") |
| 79 | + if (pr.get("user") or {}).get("login")!="dependabot[bot]": raise Block("author is not dependabot[bot]") |
| 80 | + if (pr.get("base") or {}).get("ref")!="main": raise Block("base is not main") |
| 81 | + if ((pr.get("head") or {}).get("repo") or {}).get("full_name")!=p.repo: raise Block("head repo is not this repo") |
| 82 | + head=(pr.get("head") or {}).get("sha") |
| 83 | + if not head: raise Error("missing PR head SHA") |
| 84 | + if ts and ts!=head: raise Block("workflow belongs to obsolete PR head") |
| 85 | + if not routine(str(pr.get("title","")),p.groups): raise Block("not an allowlisted routine group; individual/major migrations stay human-controlled") |
| 86 | + main=((gh.get(f"/repos/{p.repo}/branches/main").get("commit") or {}).get("sha")) |
| 87 | + if not main: raise Error("missing main SHA") |
| 88 | + if tb and tb!=main: raise Block("triggering workflow tested obsolete main; wait for Dependabot rebase") |
| 89 | + names=files_ok(gh.get(f"/repos/{p.repo}/pulls/{number}/files?per_page=100"),p); q=urllib.parse.urlencode({"head_sha":head,"event":"pull_request","per_page":100}); runs=(gh.get(f"/repos/{p.repo}/actions/runs?{q}").get("workflow_runs") or []); runs_ok(runs,p.required,main) |
| 90 | + if pr.get("mergeable") is False: raise Block("GitHub reports PR not mergeable") |
| 91 | + result=gh.put(f"/repos/{p.repo}/pulls/{number}/merge",{"sha":head,"merge_method":p.method,"commit_title":str(pr.get("title","Dependabot routine update")),"commit_message":f"Automatically qualified by the repository Dependabot governor.\n\nPR #{number}; exact head {head}; required workflows: {', '.join(p.required)}."}) |
| 92 | + if not result or not result.get("merged"): raise Error(f"merge rejected: {result}") |
| 93 | + summary(["## Dependabot governor","",f"- Decision: **merged** PR #{number}",f"- Exact head: `{head}`",f"- Merge method: `{p.method}`",f"- Files: {', '.join(f'`{x}`' for x in names)}"]); print(f"merged Dependabot PR #{number} at {head}") |
| 94 | +def self_test(): |
| 95 | + p=Policy("o/r",("ci","security"),("routine-dependencies","routine-actions"),("package.json","package-lock.json",".github/workflows/*.yml"),"merge"); assert routine("bump routine-dependencies group",p.groups) and not routine("bump framework 1 to 2",p.groups); assert allowed(".github/workflows/ci.yml",p.paths) and not allowed("src/app.py",p.paths); assert files_ok([{"filename":"package.json"}],p)==["package.json"] |
| 96 | + try: files_ok([{"filename":"README.md"}],p) |
| 97 | + except Block: pass |
| 98 | + else: raise AssertionError("README must be blocked") |
| 99 | + good=[{"path":"ci","status":"completed","conclusion":"success","pull_requests":[{"base":{"sha":"m"}}]},{"path":"security","status":"completed","conclusion":"success","pull_requests":[{"base":{"sha":"m"}}]}]; runs_ok(good,p.required,"m") |
| 100 | + for c in ("failure","cancelled","timed_out","action_required"): |
| 101 | + bad=[dict(x) for x in good]; bad[1]["conclusion"]=c |
| 102 | + try: runs_ok(bad,p.required,"m") |
| 103 | + except Block: pass |
| 104 | + else: raise AssertionError(c) |
| 105 | + print("dependabot governor self-test: ok") |
| 106 | +def main(): |
| 107 | + if "--self-test" in sys.argv: self_test(); return 0 |
| 108 | + try: govern(); return 0 |
| 109 | + except Block as e: summary(["## Dependabot governor","","- Decision: **no merge**",f"- Reason: {e}"]); print(f"policy no-op: {e}"); return 0 |
| 110 | + except Error as e: print(f"policy error: {e}",file=sys.stderr); return 1 |
| 111 | +if __name__=="__main__": raise SystemExit(main()) |
0 commit comments