path: root/github-action-approve.py

#!/usr/bin/env python3
#
# Approve pending GitHub Actions workflow runs of a pull request.
#
# Workflow runs of pull requests coming from a fork are held in the
# "action_required" state until a maintainer approves them. GitHub does
# offer an official REST endpoint for this:
#
#   POST /repos/{owner}/{repo}/actions/runs/{run_id}/approve
#
# but the web interface only shows an approval button for the newest run
# of a pull request, so all older runs stay pending forever. This script
# approves every pending run of a pull request instead.
#
# The GitHub token is taken from the GitHub CLI ("gh auth token"), run
# "gh auth login" once. The account needs write access to the actions of
# the repository.
#
# Usage: github-action-approve.py [-r OWNER/REPO] [-n] [-d SECONDS] PR_NUMBER

import argparse
import json
import subprocess
import sys
import time
import urllib.error
import urllib.parse
import urllib.request

API_ROOT = "https://api.github.com"
DEFAULT_REPO = "openwrt/openwrt"

# States a workflow run is in while it waits for a maintainer. A run of a
# fork pull request uses "action_required", a run held back by an
# environment protection rule uses "waiting".
PENDING_STATES = ("action_required", "waiting")

# Upper bound for the fallback scan in collect_runs(), 100 runs per page.
MAX_FALLBACK_PAGES = 20


class ApiError(Exception):
    """A request to the GitHub API failed."""

    def __init__(self, message, status=None):
        super().__init__("HTTP %s: %s" % (status, message) if status else message)
        self.status = status


def gh_token():
    """Get an API token from the GitHub CLI."""
    try:
        gh = subprocess.run(["gh", "auth", "token"], check=True,
                            capture_output=True, text=True)
    except FileNotFoundError:
        sys.exit("The GitHub CLI (gh) was not found, install it and run 'gh auth login'.")
    except subprocess.CalledProcessError as error:
        sys.exit("Could not get a token from gh, run 'gh auth login':\n%s"
                 % error.stderr.strip())

    token = gh.stdout.strip()
    if not token:
        sys.exit("The GitHub CLI returned an empty token, run 'gh auth login'.")
    return token


def request(token, method, url, data=None):
    """Send one API request, return the decoded body and the headers."""
    if url.startswith("/"):
        url = API_ROOT + url

    body = json.dumps(data).encode() if data is not None else None
    req = urllib.request.Request(url, data=body, method=method)
    req.add_header("Authorization", "Bearer %s" % token)
    req.add_header("Accept", "application/vnd.github+json")
    req.add_header("X-GitHub-Api-Version", "2022-11-28")
    if body is not None:
        req.add_header("Content-Type", "application/json")

    try:
        with urllib.request.urlopen(req) as response:
            raw = response.read()
            return (json.loads(raw) if raw else None), response.headers
    except urllib.error.HTTPError as error:
        raw = error.read()
        try:
            message = json.loads(raw).get("message", "")
        except (ValueError, AttributeError):
            message = raw.decode(errors="replace").strip()
        raise ApiError(message or error.reason, error.code)
    except urllib.error.URLError as error:
        raise ApiError("could not reach %s: %s" % (API_ROOT, error.reason))


def next_page(headers):
    """Extract the URL of the next page from the Link header."""
    for link in headers.get("Link", "").split(","):
        parts = link.split(";")
        if len(parts) > 1 and 'rel="next"' in parts[1]:
            return parts[0].strip().strip("<>")
    return None


def paginate(token, url, key=None, max_pages=None):
    """Yield all items of a list endpoint, following the Link headers."""
    pages = 0
    while url:
        payload, headers = request(token, "GET", url)
        for item in (payload.get(key, []) if key else payload):
            yield item

        pages += 1
        url = next_page(headers)
        if max_pages is not None and pages >= max_pages and url:
            print("Warning: stopped after %d pages, older runs were not checked."
                  % pages, file=sys.stderr)
            return


def needs_approval(run):
    """Check if a workflow run is waiting for a maintainer."""
    return (run.get("status") in PENDING_STATES
            or run.get("conclusion") == "action_required")


def collect_runs(token, repo, branch):
    """Collect all workflow runs of a branch which wait for approval."""
    runs = {}

    for state in PENDING_STATES:
        query = urllib.parse.urlencode({"branch": branch, "status": state,
                                        "per_page": 100})
        for run in paginate(token, "/repos/%s/actions/runs?%s" % (repo, query),
                            key="workflow_runs"):
            runs[run["id"]] = run

    if runs:
        return runs

    # The server side status filter found nothing. GitHub sometimes reports
    # a pending run with a different status and only sets the conclusion to
    # "action_required", so walk the branch unfiltered as a fallback. This
    # can be a lot of runs when the fork branch is named like a branch of
    # the base repository, hence the page limit.
    query = urllib.parse.urlencode({"branch": branch, "per_page": 100})
    for run in paginate(token, "/repos/%s/actions/runs?%s" % (repo, query),
                        key="workflow_runs", max_pages=MAX_FALLBACK_PAGES):
        if needs_approval(run):
            runs[run["id"]] = run

    return runs


def pr_commit_shas(token, repo, number):
    """Get the commit hashes which are currently part of a pull request."""
    query = urllib.parse.urlencode({"per_page": 100})
    return {commit["sha"] for commit
            in paginate(token, "/repos/%s/pulls/%d/commits?%s" % (repo, number, query))}


def belongs_to_pr(run, number, head_repo, head_branch, shas):
    """Check if a workflow run was triggered by the given pull request."""
    # A pull request from a branch of the repository itself is referenced
    # directly by the run.
    for pull in run.get("pull_requests") or []:
        if pull.get("number") == number:
            return True

    # A run triggered from a fork has an empty "pull_requests" list, match
    # it on the source repository and branch instead. This also catches
    # runs of commits which were replaced by a force push and are not part
    # of the pull request any more.
    head = run.get("head_repository") or {}
    if head_repo and (head.get("full_name") or "").lower() == head_repo.lower():
        if run.get("head_branch") == head_branch:
            return True

    return run.get("head_sha") in shas


def approve_deployments(token, repo, run):
    """Approve a run which is held back by an environment protection rule."""
    url = "/repos/%s/actions/runs/%d/pending_deployments" % (repo, run["id"])
    pending, _ = request(token, "GET", url)

    environments = [entry["environment"]["id"] for entry in pending or []
                    if entry.get("current_user_can_approve")
                    and (entry.get("environment") or {}).get("id")]
    if not environments:
        raise ApiError("no environment of this run is waiting for your review")

    request(token, "POST", url, {"environment_ids": environments,
                                 "state": "approved", "comment": ""})
    return "deployment approved"


def approve_run(token, repo, run):
    """Approve a single workflow run."""
    if run.get("status") == "waiting":
        return approve_deployments(token, repo, run)

    request(token, "POST", "/repos/%s/actions/runs/%d/approve" % (repo, run["id"]))
    return "approved"


def describe(run):
    """Build a one line description of a workflow run."""
    return "%s %-24.24s %-12.12s %s" % ((run.get("created_at") or "")[:19].replace("T", " "),
                                        run.get("name") or "workflow",
                                        run.get("head_sha") or "",
                                        run.get("html_url") or "")


def main():
    parser = argparse.ArgumentParser(
        description="Approve all pending GitHub Actions workflow runs of a pull request.",
        epilog="The GitHub token is taken from the GitHub CLI, run 'gh auth login' once.")
    parser.add_argument("pr", type=int, metavar="PR_NUMBER",
                        help="number of the pull request to approve the runs of")
    parser.add_argument("-r", "--repo", default=DEFAULT_REPO, metavar="OWNER/REPO",
                        help="repository to work on (default: %(default)s)")
    parser.add_argument("-d", "--delay", type=float, default=1.0, metavar="SECONDS",
                        help="wait time after each approval (default: %(default)s)")
    parser.add_argument("-n", "--dry-run", action="store_true",
                        help="only list the runs which would be approved")
    args = parser.parse_args()

    if args.repo.count("/") != 1 or not all(args.repo.split("/")):
        sys.exit("The repository has to be given as OWNER/REPO.")

    token = gh_token()

    try:
        pull, _ = request(token, "GET", "/repos/%s/pulls/%d" % (args.repo, args.pr))
    except ApiError as error:
        sys.exit("Could not fetch pull request %s#%d: %s" % (args.repo, args.pr, error))

    head = pull.get("head") or {}
    head_branch = head.get("ref")
    head_repo = ((head.get("repo") or {}).get("full_name"))

    print("%s#%d: %s" % (args.repo, args.pr, pull.get("title") or ""))
    print("head:  %s:%s" % (head_repo or "<deleted fork>", head_branch))

    if not head_branch:
        sys.exit("The pull request has no head branch, it can not be matched to runs.")

    try:
        candidates = collect_runs(token, args.repo, head_branch)
        shas = pr_commit_shas(token, args.repo, args.pr) if candidates else set()
    except ApiError as error:
        sys.exit("Could not list the workflow runs: %s" % error)

    runs = [run for run in candidates.values()
            if belongs_to_pr(run, args.pr, head_repo, head_branch, shas)]

    # Oldest run first, so the history is worked off in the order it happened.
    runs.sort(key=lambda run: ((run.get("created_at") or ""), run["id"]))

    if not runs:
        print("No workflow run of this pull request is waiting for approval.")
        return 0

    print("Found %d workflow run(s) waiting for approval:" % len(runs))

    failed = 0
    for index, run in enumerate(runs, 1):
        print("[%d/%d] %s ... " % (index, len(runs), describe(run)), end="", flush=True)

        if args.dry_run:
            print("skipped (dry run)")
            continue

        try:
            print(approve_run(token, args.repo, run), flush=True)
        except ApiError as error:
            print("FAILED: %s" % error, flush=True)
            failed += 1
            continue

        # Only start the next request once the server confirmed this one.
        if index < len(runs):
            time.sleep(args.delay)

    if failed:
        print("%d of %d workflow run(s) could not be approved." % (failed, len(runs)),
              file=sys.stderr)
        return 1

    return 0


if __name__ == "__main__":
    try:
        sys.exit(main())
    except KeyboardInterrupt:
        sys.exit(130)