diff --git a/README.md b/README.md index b2950e0..d04e4f4 100644 --- a/README.md +++ b/README.md @@ -58,6 +58,12 @@ Status: in progress. The design is in `specs/next-ci-reliability` of `py -3 standards/audit.py [--fail-on error]`, or `report/ci_report.py --audit` to append it to the weekly report. +- `standards/bootstrap.py` (available): applies the organisation's repository settings to a new + repository: branch deletion after merge, secret scanning and push protection where they are free, + Dependabot alerts and security updates, and the `automerge` and `needs-human` labels the merge + train uses. A dry run by default; `--apply` makes the changes. + `py -3 standards/bootstrap.py owner/name [--apply]`. + ## Rules - No secrets, no business logic and no repository-specific values: everything comes in as an input. diff --git a/standards/bootstrap.py b/standards/bootstrap.py new file mode 100644 index 0000000..3d2ac89 --- /dev/null +++ b/standards/bootstrap.py @@ -0,0 +1,226 @@ +"""Apply the organisation's repository settings to a new repository. + + py -3 bootstrap.py owner/name # show what would change (dry run, the default) + py -3 bootstrap.py owner/name --apply # make the changes + +The settings come from the standards audit (audit.py): branch deletion after merge, secret +scanning and push protection where they are free (public repositories), Dependabot alerts and +security updates, and the two labels the merge train uses. Nothing is changed without --apply, and a +setting that already follows the standard is left alone. +""" + +from __future__ import annotations + +import argparse +import json +import os +import re +import subprocess +import sys +import tempfile +from collections.abc import Callable +from dataclasses import dataclass + +LABELS = { + "automerge": ( + "0E8A16", + "The merge train merges this pull request when its checks pass", + ), + "needs-human": ( + "D93F0B", + "A person must resolve this before the merge train retries", + ), +} +REPO_PATTERN = re.compile(r"^[A-Za-z0-9_.-]+/[A-Za-z0-9_.-]+$") + + +class BootstrapError(ValueError): + """The repository cannot be bootstrapped.""" + + +@dataclass(frozen=True) +class Change: + setting: str + current: str + wanted: str + + +def run_gh(command: list[str]) -> str: + # Every command is built in this file from fixed words and the repository name given. + no_window = getattr(subprocess, "CREATE_NO_WINDOW", 0) + return subprocess.run( + command, + check=True, + capture_output=True, + text=True, + encoding="utf-8", + creationflags=no_window, + ).stdout + + +class Github: + def __init__(self, run: Callable[[list[str]], str] = run_gh): + self._run = run + + def repo(self, full_name: str) -> dict: + return json.loads(self._run(["gh", "api", f"repos/{full_name}"]) or "{}") + + def labels(self, full_name: str) -> list[str]: + out = self._run( + [ + "gh", + "api", + f"repos/{full_name}/labels?per_page=100", + "--paginate", + "--slurp", + ] + ) + listed = json.loads(out or "[]") + if listed and isinstance(listed[0], list): + listed = [item for page in listed for item in page] + return [item["name"] for item in listed] + + def alerts_enabled(self, full_name: str) -> bool: + try: + self._run(["gh", "api", f"repos/{full_name}/vulnerability-alerts"]) + except subprocess.CalledProcessError: + return False + return True + + def security_fixes_enabled(self, full_name: str) -> bool: + try: + out = self._run( + [ + "gh", + "api", + f"repos/{full_name}/automated-security-fixes", + "--jq", + ".enabled", + ] + ) + except subprocess.CalledProcessError: + return False + return out.strip() == "true" + + def patch_repo(self, full_name: str, fields: dict) -> None: + with tempfile.NamedTemporaryFile( + "w", suffix=".json", delete=False, encoding="utf-8" + ) as handle: + json.dump(fields, handle) + path = handle.name + try: + self._run( + ["gh", "api", "-X", "PATCH", f"repos/{full_name}", "--input", path] + ) + finally: + os.unlink(path) + + def create_label( + self, full_name: str, name: str, color: str, description: str + ) -> None: + self._run( + [ + "gh", + "api", + "-X", + "POST", + f"repos/{full_name}/labels", + "-f", + f"name={name}", + "-f", + f"color={color}", + "-f", + f"description={description}", + ] + ) + + def enable_alerts(self, full_name: str) -> None: + self._run(["gh", "api", "-X", "PUT", f"repos/{full_name}/vulnerability-alerts"]) + + def enable_security_fixes(self, full_name: str) -> None: + self._run( + ["gh", "api", "-X", "PUT", f"repos/{full_name}/automated-security-fixes"] + ) + + +def plan(gh, full_name: str) -> list[Change]: + repo = gh.repo(full_name) + if repo.get("archived"): + raise BootstrapError(f"{full_name} is archived") + changes: list[Change] = [] + if not repo.get("delete_branch_on_merge"): + changes.append(Change("delete_branch_on_merge", "off", "on")) + if not repo.get("private", True): + security = repo.get("security_and_analysis") or {} + for key in ("secret_scanning", "secret_scanning_push_protection"): + if (security.get(key) or {}).get("status") != "enabled": + changes.append(Change(key, "off", "on")) + existing = set(gh.labels(full_name)) + for name in LABELS: + if name not in existing: + changes.append(Change(f"label:{name}", "missing", "created")) + if not gh.alerts_enabled(full_name): + changes.append(Change("vulnerability_alerts", "off", "on")) + if not gh.security_fixes_enabled(full_name): + changes.append(Change("automated_security_fixes", "off", "on")) + return changes + + +def apply(gh, full_name: str, changes: list[Change], dry_run: bool) -> None: + if dry_run: + return + fields: dict = {} + security: dict = {} + for change in changes: + if change.setting == "delete_branch_on_merge": + fields["delete_branch_on_merge"] = True + elif change.setting in ("secret_scanning", "secret_scanning_push_protection"): + security[change.setting] = {"status": "enabled"} + if security: + fields["security_and_analysis"] = security + if fields: + gh.patch_repo(full_name, fields) + for change in changes: + if change.setting.startswith("label:"): + name = change.setting.removeprefix("label:") + color, description = LABELS[name] + gh.create_label(full_name, name, color, description) + elif change.setting == "vulnerability_alerts": + gh.enable_alerts(full_name) + elif change.setting == "automated_security_fixes": + gh.enable_security_fixes(full_name) + + +def render(changes: list[Change], full_name: str, dry_run: bool) -> str: + mode = "dry run: nothing changed" if dry_run else "applied" + lines = [f"# Bootstrap of {full_name} ({mode})", ""] + if not changes: + lines.append(f"{full_name} already follows the standard.") + for change in changes: + lines.append(f"- {change.setting}: {change.current} -> {change.wanted}") + return "\n".join(lines) + + +def main(argv: list[str] | None = None) -> int: + parser = argparse.ArgumentParser(description=__doc__.splitlines()[0]) + parser.add_argument("repo", help="owner/name") + parser.add_argument( + "--apply", action="store_true", help="make the changes (default: dry run)" + ) + args = parser.parse_args(argv) + if not REPO_PATTERN.match(args.repo): + print("repo must look like owner/name", file=sys.stderr) + return 2 + gh = Github() + try: + changes = plan(gh, args.repo) + except BootstrapError as exc: + print(f"bootstrap error: {exc}", file=sys.stderr) + return 2 + apply(gh, args.repo, changes, dry_run=not args.apply) + print(render(changes, args.repo, dry_run=not args.apply)) + return 0 + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/tests/test_bootstrap.py b/tests/test_bootstrap.py new file mode 100644 index 0000000..b20a044 --- /dev/null +++ b/tests/test_bootstrap.py @@ -0,0 +1,203 @@ +"""Tests for standards/bootstrap.py: the plan of settings for a new repository and its application.""" + +from __future__ import annotations + +import json +import subprocess +import sys +import unittest +from pathlib import Path + +sys.path.insert(0, str(Path(__file__).resolve().parent.parent / "standards")) + +import bootstrap + +REPO = "naindev-contrast3d/new-app" + + +def settings(**overrides): + base = { + "private": True, + "archived": False, + "delete_branch_on_merge": False, + "security_and_analysis": { + "secret_scanning": {"status": "disabled"}, + "secret_scanning_push_protection": {"status": "disabled"}, + }, + } + base.update(overrides) + return base + + +class FakeGithub: + def __init__(self, repo=None, labels=(), alerts=False, fixes=False): + self._repo = repo or settings() + self._labels = list(labels) + self._alerts = alerts + self._fixes = fixes + self.calls = [] + + def repo(self, full_name): + return self._repo + + def labels(self, full_name): + return self._labels + + def alerts_enabled(self, full_name): + return self._alerts + + def security_fixes_enabled(self, full_name): + return self._fixes + + def patch_repo(self, full_name, fields): + self.calls.append(("patch", fields)) + + def create_label(self, full_name, name, color, description): + self.calls.append(("label", name)) + + def enable_alerts(self, full_name): + self.calls.append(("alerts",)) + + def enable_security_fixes(self, full_name): + self.calls.append(("fixes",)) + + +def names(changes): + return {c.setting for c in changes} + + +class Plan(unittest.TestCase): + def test_a_fresh_private_repository_needs_branch_deletion_labels_and_security_updates( + self, + ): + changes = bootstrap.plan(FakeGithub(), REPO) + self.assertEqual( + names(changes), + { + "delete_branch_on_merge", + "label:automerge", + "label:needs-human", + "vulnerability_alerts", + "automated_security_fixes", + }, + ) + + def test_secret_scanning_is_asked_for_only_on_public_repositories(self): + public = FakeGithub(settings(private=False)) + self.assertIn("secret_scanning", names(bootstrap.plan(public, REPO))) + self.assertIn( + "secret_scanning_push_protection", names(bootstrap.plan(public, REPO)) + ) + self.assertNotIn("secret_scanning", names(bootstrap.plan(FakeGithub(), REPO))) + + def test_settings_already_in_place_are_not_planned(self): + done = settings(delete_branch_on_merge=True) + gh = FakeGithub( + done, labels=["automerge", "needs-human"], alerts=True, fixes=True + ) + self.assertEqual(bootstrap.plan(gh, REPO), []) + + def test_an_archived_repository_is_refused(self): + with self.assertRaises(bootstrap.BootstrapError): + bootstrap.plan(FakeGithub(settings(archived=True)), REPO) + + def test_every_change_says_what_it_is_now_and_what_it_will_be(self): + change = next( + c + for c in bootstrap.plan(FakeGithub(), REPO) + if c.setting == "delete_branch_on_merge" + ) + self.assertEqual((change.current, change.wanted), ("off", "on")) + + +class Apply(unittest.TestCase): + def test_nothing_changes_without_apply(self): + gh = FakeGithub() + bootstrap.apply(gh, REPO, bootstrap.plan(gh, REPO), dry_run=True) + self.assertEqual(gh.calls, []) + + def test_apply_makes_each_planned_change_once(self): + gh = FakeGithub() + bootstrap.apply(gh, REPO, bootstrap.plan(gh, REPO), dry_run=False) + kinds = [call[0] for call in gh.calls] + self.assertEqual(sorted(kinds), ["alerts", "fixes", "label", "label", "patch"]) + patch = next(call for call in gh.calls if call[0] == "patch") + self.assertEqual(patch[1], {"delete_branch_on_merge": True}) + + def test_public_security_settings_go_in_one_patch(self): + gh = FakeGithub(settings(private=False, delete_branch_on_merge=True)) + bootstrap.apply(gh, REPO, bootstrap.plan(gh, REPO), dry_run=False) + patch = next(call for call in gh.calls if call[0] == "patch") + self.assertEqual( + patch[1], + { + "security_and_analysis": { + "secret_scanning": {"status": "enabled"}, + "secret_scanning_push_protection": {"status": "enabled"}, + } + }, + ) + + +class Rendering(unittest.TestCase): + def test_the_plan_is_printed_as_a_list_and_says_when_nothing_is_needed(self): + text = bootstrap.render(bootstrap.plan(FakeGithub(), REPO), REPO, dry_run=True) + self.assertIn("dry run", text.lower()) + self.assertIn("delete_branch_on_merge: off -> on", text) + done = FakeGithub( + settings(delete_branch_on_merge=True), + labels=["automerge", "needs-human"], + alerts=True, + fixes=True, + ) + self.assertIn( + "already follows", + bootstrap.render(bootstrap.plan(done, REPO), REPO, dry_run=True), + ) + + +class RealCommands(unittest.TestCase): + def setUp(self): + self.commands = [] + self.outputs = {} + self.errors = {} + self.gh = bootstrap.Github(self.fake_run) + + def fake_run(self, command): + self.commands.append(command) + key = " ".join(command) + for needle, error in self.errors.items(): + if needle in key: + raise subprocess.CalledProcessError(1, command, stderr=error) + for needle, output in self.outputs.items(): + if needle in key: + return output + return "" + + def test_alerts_are_enabled_when_the_endpoint_answers_and_not_when_it_is_404(self): + self.assertTrue(self.gh.alerts_enabled(REPO)) + self.errors["vulnerability-alerts"] = "HTTP 404" + self.assertFalse(self.gh.alerts_enabled(REPO)) + + def test_labels_are_read_by_name(self): + self.outputs["labels"] = json.dumps([{"name": "automerge"}, {"name": "bug"}]) + self.assertEqual(self.gh.labels(REPO), ["automerge", "bug"]) + + def test_the_repository_is_patched_with_a_json_body(self): + self.gh.patch_repo(REPO, {"delete_branch_on_merge": True}) + command = self.commands[0] + self.assertIn("PATCH", command) + self.assertIn("--input", command) + + def test_security_updates_and_alerts_use_their_endpoints(self): + self.gh.enable_alerts(REPO) + self.gh.enable_security_fixes(REPO) + joined = [" ".join(c) for c in self.commands] + self.assertTrue(any("PUT" in c and "vulnerability-alerts" in c for c in joined)) + self.assertTrue( + any("PUT" in c and "automated-security-fixes" in c for c in joined) + ) + + +if __name__ == "__main__": + unittest.main()