ci(signed-commits): check that GitHub verifies every pull request commit

.github/workflows/signed-commits.yml

@@ -0,0 +1,39 @@
+# Every commit a pull request carries must be signed and verified by GitHub.
+# Branch protection holds that rule at merge time; this reports it as a named
+# check on every push, lists each offending commit, and fails closed when the
+# commit list cannot be read in full.
+name: Signed commits
+
+on:
+  pull_request:
+    branches: [main]
+    types: [opened, synchronize, reopened]
+
+permissions:
+  contents: read
+  pull-requests: read
+
+jobs:
+  signed:
+    name: Signed commits
+    runs-on: ubuntu-latest
+    steps:
+      - name: Harden runner
+        uses: step-security/harden-runner@e14015d583714f6e62063499dc959a02595150a1 # v2
+        with:
+          egress-policy: audit
+
+      - uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 # v7.0.1
+        with:
+          persist-credentials: false
+
+      - name: Test the check
+        run: python3 scripts/test_check_signed_commits.py
+
+      - name: Check every commit's signature
+        env:
+          GH_TOKEN: ${{ github.token }}
+          REPO: ${{ github.repository }}
+          PR_NUMBER: ${{ github.event.pull_request.number }}
+          HEAD_SHA: ${{ github.event.pull_request.head.sha }}
+        run: python3 scripts/check_signed_commits.py --repo "$REPO" --pr "$PR_NUMBER" --head-sha "$HEAD_SHA"

scripts/check_signed_commits.py

@@ -0,0 +1,154 @@
+#!/usr/bin/env python3
+# Copyright 2026 Ori Nexus Systems LTD
+# SPDX-License-Identifier: Apache-2.0
+"""Refuse a pull request unless GitHub verifies the signature on every commit.
+
+Branch protection holds the same rule at merge time. This reports it as a named
+check on every push, lists each commit GitHub does not verify with GitHub's
+reason, and fails closed on anything it cannot read in full.
+
+    python3 scripts/check_signed_commits.py --repo OWNER/NAME --pr N --head-sha SHA
+"""
+
+from __future__ import annotations
+
+import argparse
+import json
+import subprocess
+import sys
+
+# The pull request commits endpoint lists at most this many, paginated or not.
+API_COMMIT_LIMIT = 250
+SIGNING_DOCS = "https://docs.github.com/en/authentication/managing-commit-signature-verification/signing-commits"
+
+
+class CheckError(Exception):
+    """The commit list could not be established in full."""
+
+
+def _gh_api(*args: str) -> str:
+    try:
+        completed = subprocess.run(
+            ["gh", "api", *args],
+            capture_output=True,
+            text=True,
+            check=False,
+        )
+    except OSError as exc:
+        raise CheckError(f"could not run gh: {exc}") from exc
+    if completed.returncode != 0:
+        raise CheckError(
+            f"gh api {' '.join(args)} exited {completed.returncode}: "
+            f"{completed.stderr.strip()}"
+        )
+    return completed.stdout
+
+
+def _json_values(text: str) -> list[object]:
+    """Every JSON value in a stream; --paginate prints one array per page."""
+    decoder = json.JSONDecoder()
+    values: list[object] = []
+    index = 0
+    while True:
+        while index < len(text) and text[index].isspace():
+            index += 1
+        if index == len(text):
+            return values
+        try:
+            value, index = decoder.raw_decode(text, index)
+        except json.JSONDecodeError as exc:
+            raise CheckError(f"the API returned malformed JSON: {exc}") from exc
+        values.append(value)
+
+
+def _expected_commits(repo: str, pr: int, head_sha: str) -> int:
+    values = _json_values(_gh_api(f"repos/{repo}/pulls/{pr}"))
+    if len(values) != 1 or not isinstance(values[0], dict):
+        raise CheckError("the pull request response is not a single object")
+    pull = values[0]
+    head = pull.get("head")
+    current = head.get("sha") if isinstance(head, dict) else None
+    if current != head_sha:
+        raise CheckError(
+            f"the pull request head is {current!r}, not the {head_sha} this run "
+            "was started for; the run for the newer head is the one that counts"
+        )
+    count = pull.get("commits")
+    if type(count) is not int or count < 1:
+        raise CheckError(f"the pull request reports {count!r} commits")
+    if count > API_COMMIT_LIMIT:
+        raise CheckError(
+            f"the pull request has {count} commits; GitHub lists at most "
+            f"{API_COMMIT_LIMIT}, so the rest cannot be checked. Split it."
+        )
+    return count
+
+
+def _list_commits(repo: str, pr: int) -> list[object]:
+    pages = _json_values(
+        _gh_api("--paginate", f"repos/{repo}/pulls/{pr}/commits?per_page=100")
+    )
+    commits: list[object] = []
+    for page in pages:
+        if not isinstance(page, list):
+            raise CheckError("a page of the commit list is not an array")
+        commits.extend(page)
+    return commits
+
+
+def unverified(commits: list[object]) -> list[tuple[str, str]]:
+    """Each commit GitHub does not verify, with the reason it gives."""
+    failures: list[tuple[str, str]] = []
+    for entry in commits:
+        sha = entry.get("sha") if isinstance(entry, dict) else None
+        if not isinstance(sha, str):
+            failures.append(("<no sha>", "the API entry carries no sha"))
+            continue
+        commit = entry.get("commit") if isinstance(entry, dict) else None
+        verification = commit.get("verification") if isinstance(commit, dict) else None
+        if not isinstance(verification, dict):
+            failures.append((sha, "no verification object"))
+            continue
+        if verification.get("verified") is not True:
+            failures.append((sha, str(verification.get("reason", "no reason given"))))
+    return failures
+
+
+def check(repo: str, pr: int, head_sha: str) -> int:
+    try:
+        expected = _expected_commits(repo, pr, head_sha)
+        commits = _list_commits(repo, pr)
+        if len(commits) != expected:
+            raise CheckError(
+                f"the API listed {len(commits)} commits, the pull request reports "
+                f"{expected}"
+            )
+        shas = {entry.get("sha") for entry in commits if isinstance(entry, dict)}
+        if head_sha not in shas:
+            raise CheckError(f"the head {head_sha} is not among the listed commits")
+    except CheckError as exc:
+        print(f"Cannot establish the pull request's commits: {exc}")
+        return 1
+
+    failures = unverified(commits)
+    if failures:
+        print(f"{len(failures)} of {len(commits)} commits are not verified by GitHub:")
+        for sha, reason in failures:
+            print(f"   {sha}  {reason}")
+        print(f"Sign every commit and push again. See {SIGNING_DOCS}")
+        return 1
+    print(f"All {len(commits)} commits are verified by GitHub.")
+    return 0
+
+
+def main() -> int:
+    parser = argparse.ArgumentParser()
+    parser.add_argument("--repo", required=True)
+    parser.add_argument("--pr", required=True, type=int)
+    parser.add_argument("--head-sha", required=True)
+    args = parser.parse_args()
+    return check(args.repo, args.pr, args.head_sha)
+
+
+if __name__ == "__main__":
+    sys.exit(main())

scripts/test_check_signed_commits.py

@@ -0,0 +1,186 @@
+#!/usr/bin/env python3
+# Copyright 2026 Ori Nexus Systems LTD
+# SPDX-License-Identifier: Apache-2.0
+"""Drive the signed-commits check through its command line against a fake gh."""
+
+from __future__ import annotations
+
+import json
+import os
+import subprocess
+import sys
+import tempfile
+import unittest
+from pathlib import Path
+
+SCRIPT = Path(__file__).resolve().with_name("check_signed_commits.py")
+HEAD = "b" * 40
+
+FAKE_GH = """#!/usr/bin/env python3
+import os, sys
+key = "COMMITS" if "--paginate" in sys.argv else "PULL"
+sys.stdout.write(os.environ["FAKE_" + key])
+sys.stderr.write(os.environ.get("FAKE_" + key + "_ERR", ""))
+sys.exit(int(os.environ.get("FAKE_" + key + "_RC", "0")))
+"""
+
+
+def commit(
+    sha: str, verified: object = True, reason: str = "valid"
+) -> dict[str, object]:
+    return {
+        "sha": sha,
+        "commit": {"verification": {"verified": verified, "reason": reason}},
+    }
+
+
+def pull(commits: object = 2, head: str = HEAD) -> str:
+    return json.dumps({"commits": commits, "head": {"sha": head}})
+
+
+GOOD = [commit("a" * 40), commit(HEAD)]
+
+
+class SignedCommitsTest(unittest.TestCase):
+    def setUp(self) -> None:
+        self._dir = tempfile.TemporaryDirectory()
+        self.addCleanup(self._dir.cleanup)
+        gh = Path(self._dir.name) / "gh"
+        gh.write_text(FAKE_GH, encoding="utf-8")
+        gh.chmod(0o755)
+
+    def run_check(
+        self, pull_json: str, commits_out: str, **extra: str
+    ) -> subprocess.CompletedProcess[str]:
+        env = dict(os.environ)
+        env["PATH"] = self._dir.name + os.pathsep + env.get("PATH", "")
+        env["FAKE_PULL"] = pull_json
+        env["FAKE_COMMITS"] = commits_out
+        env.update(extra)
+        return subprocess.run(
+            [
+                sys.executable,
+                str(SCRIPT),
+                "--repo",
+                "o/r",
+                "--pr",
+                "7",
+                "--head-sha",
+                HEAD,
+            ],
+            capture_output=True,
+            text=True,
+            env=env,
+            check=False,
+        )
+
+    def assert_refused(
+        self, result: subprocess.CompletedProcess[str], text: str
+    ) -> None:
+        self.assertEqual(result.returncode, 1, result.stdout + result.stderr)
+        self.assertIn(text, result.stdout)
+        self.assertNotIn("Traceback", result.stderr)
+
+    def test_every_commit_verified_passes(self) -> None:
+        result = self.run_check(pull(), json.dumps(GOOD))
+        self.assertEqual(result.returncode, 0, result.stdout + result.stderr)
+        self.assertIn("All 2 commits are verified", result.stdout)
+
+    def test_pages_are_concatenated(self) -> None:
+        result = self.run_check(
+            pull(), json.dumps(GOOD[:1]) + "\n" + json.dumps(GOOD[1:])
+        )
+        self.assertEqual(result.returncode, 0, result.stdout + result.stderr)
+
+    def test_unverified_commit_is_named_with_its_reason(self) -> None:
+        bad = [commit("a" * 40, False, "unsigned"), commit(HEAD)]
+        result = self.run_check(pull(), json.dumps(bad))
+        self.assert_refused(result, "a" * 40 + "  unsigned")
+        self.assertIn("signing-commits", result.stdout)
+
+    def test_truthy_but_not_true_is_refused(self) -> None:
+        result = self.run_check(
+            pull(), json.dumps([commit("a" * 40, "true"), commit(HEAD)])
+        )
+        self.assert_refused(result, "a" * 40)
+
+    def test_missing_verification_object_is_refused(self) -> None:
+        bad = [{"sha": "a" * 40, "commit": {}}, commit(HEAD)]
+        self.assert_refused(
+            self.run_check(pull(), json.dumps(bad)), "no verification object"
+        )
+
+    def test_entry_without_sha_is_refused(self) -> None:
+        bad = [{"commit": {"verification": {"verified": True}}}, commit(HEAD)]
+        self.assert_refused(self.run_check(pull(), json.dumps(bad)), "<no sha>")
+
+    def test_empty_list_is_refused(self) -> None:
+        self.assert_refused(self.run_check(pull(), "[]"), "listed 0 commits")
+
+    def test_no_output_is_refused(self) -> None:
+        self.assert_refused(self.run_check(pull(), ""), "listed 0 commits")
+
+    def test_zero_commit_pull_request_is_refused(self) -> None:
+        self.assert_refused(self.run_check(pull(0), "[]"), "reports 0 commits")
+
+    def test_list_api_error_is_refused(self) -> None:
+        result = self.run_check(
+            pull(), "", FAKE_COMMITS_RC="1", FAKE_COMMITS_ERR="HTTP 502"
+        )
+        self.assert_refused(result, "HTTP 502")
+
+    def test_pull_api_error_is_refused(self) -> None:
+        result = self.run_check(
+            "", json.dumps(GOOD), FAKE_PULL_RC="1", FAKE_PULL_ERR="HTTP 404"
+        )
+        self.assert_refused(result, "HTTP 404")
+
+    def test_malformed_json_is_refused(self) -> None:
+        self.assert_refused(
+            self.run_check(pull(), json.dumps(GOOD)[:-1]), "malformed JSON"
+        )
+
+    def test_non_array_page_is_refused(self) -> None:
+        self.assert_refused(self.run_check(pull(), '{"message": "x"}'), "not an array")
+
+    def test_over_the_api_limit_is_refused(self) -> None:
+        self.assert_refused(self.run_check(pull(251), json.dumps(GOOD)), "at most 250")
+
+    def test_truncated_listing_is_refused(self) -> None:
+        self.assert_refused(
+            self.run_check(pull(3), json.dumps(GOOD)), "listed 2 commits"
+        )
+
+    def test_moved_head_is_refused(self) -> None:
+        self.assert_refused(
+            self.run_check(pull(head="c" * 40), json.dumps(GOOD)), "not the"
+        )
+
+    def test_head_absent_from_listing_is_refused(self) -> None:
+        stale = [commit("a" * 40), commit("c" * 40)]
+        self.assert_refused(
+            self.run_check(pull(), json.dumps(stale)), "not among the listed"
+        )
+
+    def test_missing_gh_is_refused(self) -> None:
+        result = subprocess.run(
+            [
+                sys.executable,
+                str(SCRIPT),
+                "--repo",
+                "o/r",
+                "--pr",
+                "7",
+                "--head-sha",
+                HEAD,
+            ],
+            capture_output=True,
+            text=True,
+            env={"PATH": self._dir.name + "-absent"},
+            check=False,
+        )
+        self.assert_refused(result, "could not run gh")
+
+
+if __name__ == "__main__":
+    unittest.main()