#!/usr/bin/env python3 """Pre-flight for the CI relay's GitHub App credential. Proves, in order, that: the private key signs a usable JWT, the App is installed, an installation token can be minted, the installation can list its repositories, and the CI query returns a verdict for each *recently active* one. Every step that can fail prints why, so a missing permission is named rather than surfacing later as an empty feed. curl -O https://vitalsbar.org/verify-github-app.py export GITHUB_APP_ID=123456 export GITHUB_APP_PRIVATE_KEY_PATH=~/Downloads/vitalsbar-ci.…private-key.pem uv run --with "pyjwt[crypto]" --with httpx python verify-github-app.py Run it after creating the App and before wiring the relay up: every failure it can hit is one you can cause while clicking through GitHub's UI, so it names the problem rather than leaving it to surface later as an empty feed. This file is served from the website (website/public/), because that's where the people setting up a relay can reach it — the repo isn't public. `bun relay:verify` runs this same copy. GITHUB_APP_INSTALLATION_ID is optional — without it the script lists the installations it can see and uses the only one, which is the common case. """ import os import sys import time from datetime import datetime, timedelta, timezone from pathlib import Path import httpx import jwt API = "https://api.github.com" # Only repositories pushed within this window are checked. `pushed_at` comes back # with the installation listing we already make, so narrowing to active repos is # free — no commit history is read to decide what's active. ACTIVE_DAYS = 7 HISTORY_DEPTH = 20 # matches what the relay will use # Repositories per GraphQL call. The rate-limit cost barely moves with this (cost # counts connections, not rows), but the *work* does — ask for a few hundred repos' # histories at once and GitHub's gateway returns a 502 rather than an answer. # Chunking keeps each call small enough to serve. CHUNK = 20 SETTLED = {"SUCCESS", "FAILURE", "ERROR"} def die(message: str) -> None: print(f"\n✗ {message}", file=sys.stderr) sys.exit(1) def private_key() -> str: inline = os.environ.get("GITHUB_APP_PRIVATE_KEY", "").strip() if inline: return inline path = os.environ.get("GITHUB_APP_PRIVATE_KEY_PATH", "").strip() if not path: die("set GITHUB_APP_PRIVATE_KEY_PATH (the .pem you downloaded) or GITHUB_APP_PRIVATE_KEY") pem = Path(path).expanduser() if not pem.is_file(): die(f"no such file: {pem}") return pem.read_text() def app_jwt(app_id: str, pem: str) -> str: now = int(time.time()) # iat backdated a minute for clock skew; GitHub rejects exp more than 10 min out. return jwt.encode({"iat": now - 60, "exp": now + 540, "iss": app_id}, pem, algorithm="RS256") def check(response: httpx.Response, what: str) -> dict: if response.status_code >= 400: die(f"{what} failed — HTTP {response.status_code}: {response.text.strip()[:300]}") return response.json() def main() -> None: app_id = os.environ.get("GITHUB_APP_ID", "").strip() if not app_id: die("set GITHUB_APP_ID (Settings → Developer settings → GitHub Apps → your app)") try: token_jwt = app_jwt(app_id, private_key()) except Exception as exc: # noqa: BLE001 — surfacing the real reason is the point die(f"could not sign a JWT with that key: {exc}") with httpx.Client(timeout=15) as http: headers = {"Authorization": f"Bearer {token_jwt}", "Accept": "application/vnd.github+json"} app = check(http.get(f"{API}/app", headers=headers), "GET /app (is the App ID right?)") print(f"✓ App {app['name']} (slug: {app['slug']})") perms = app.get("permissions", {}) print(f" permissions {perms}") for needed in ("checks", "contents", "statuses"): got = perms.get(needed) mark = "✓" if got in ("read", "write") else "✗" print(f" {mark} {needed}: {got or 'MISSING'}") if not all(perms.get(p) in ("read", "write") for p in ("checks", "contents", "statuses")): die("add the missing permissions, then accept the request on the installation") installation_id = os.environ.get("GITHUB_APP_INSTALLATION_ID", "").strip() if not installation_id: installs = check(http.get(f"{API}/app/installations", headers=headers), "GET /app/installations") if not installs: die("the App isn't installed anywhere yet — install it on the org") for inst in installs: print(f" installation {inst['id']} on {inst['account']['login']}") if len(installs) > 1: die("several installations — set GITHUB_APP_INSTALLATION_ID to the one you want") installation_id = str(installs[0]["id"]) print(f"✓ Installation {installation_id}") minted = check( http.post(f"{API}/app/installations/{installation_id}/access_tokens", headers=headers), "minting an installation token", ) token = minted["token"] auth = {"Authorization": f"Bearer {token}", "Accept": "application/vnd.github+json"} print(f"✓ Token expires {minted['expires_at']}") granted: list[dict] = [] page = 1 while True: payload = check( http.get(f"{API}/installation/repositories?per_page=100&page={page}", headers=auth), "GET /installation/repositories", ) granted += payload.get("repositories", []) if len(granted) >= payload.get("total_count", 0) or not payload.get("repositories"): break page += 1 if not granted: die("the installation can see no repositories — grant it some") def pushed_at(repo: dict) -> datetime: raw = (repo.get("pushed_at") or "").replace("Z", "+00:00") try: return datetime.fromisoformat(raw) except ValueError: # never pushed return datetime.min.replace(tzinfo=timezone.utc) cutoff = datetime.now(timezone.utc) - timedelta(days=ACTIVE_DAYS) archived = [r for r in granted if r.get("archived")] active = sorted( (r for r in granted if not r.get("archived") and pushed_at(r) >= cutoff), key=pushed_at, reverse=True, ) repos = [r["full_name"] for r in active] print(f"✓ Granted {len(granted)} repositories ({len(archived)} archived)") if not repos: die(f"none of them were pushed in the last {ACTIVE_DAYS} days — widen ACTIVE_DAYS") shown = ", ".join(repos[:8]) if len(repos) > 8: shown += f", … and {len(repos) - 8} more" print(f"✓ Active {len(repos)} pushed in the last {ACTIVE_DAYS} days: {shown}") fragment = """ fragment ci on Repository { nameWithOwner databaseId defaultBranchRef { name target { ... on Commit { history(first: %d) { nodes { oid statusCheckRollup { state } } } } } } }""" % HISTORY_DEPTH verdicts: dict[str, str] = {} total_cost = 0 remaining = limit = None for offset in range(0, len(repos), CHUNK): batch = repos[offset:offset + CHUNK] aliases = "\n".join( f' r{i}: repository(owner: "{name.split("/")[0]}", name: "{name.split("/")[1]}") {{ ...ci }}' for i, name in enumerate(batch) ) query = "query {\n rateLimit { cost remaining limit }\n" + aliases + "\n}\n" + fragment response = http.post(f"{API}/graphql", headers=auth, json={"query": query}) if response.status_code >= 400: print(f" ! batch {offset}–{offset + len(batch)} failed: HTTP {response.status_code} " f"({response.text.strip()[:120]})") for name in batch: verdicts[name] = "query failed" continue result = response.json() for error in result.get("errors", []): print(f" ! {error.get('message')}") data = result.get("data") or {} rate = data.get("rateLimit") or {} total_cost += rate.get("cost") or 0 remaining, limit = rate.get("remaining"), rate.get("limit") for i, name in enumerate(batch): repo = data.get(f"r{i}") if not repo: verdicts[name] = "not readable" continue ref = repo.get("defaultBranchRef") or {} nodes = (((ref.get("target") or {}).get("history")) or {}).get("nodes") or [] verdict, pending = "no CI on this branch", 0 for node in nodes: state = (node.get("statusCheckRollup") or {}).get("state") if state in SETTLED: verdict = f"{state} ({node['oid'][:8]}" verdict += f", {pending} newer building)" if pending else ")" break if state: pending += 1 else: if pending: verdict = f"all {pending} recent commits still building" verdicts[name] = f"{ref.get('name', '?')}: {verdict}" calls = (len(repos) + CHUNK - 1) // CHUNK print(f"✓ Query {total_cost} point(s) across {calls} call(s), {remaining}/{limit} left this hour") interesting = {n: v for n, v in verdicts.items() if not v.endswith("no CI on this branch")} print(f"\n {len(interesting)} of the {len(repos)} active repositories have CI on their default branch:\n") for name, verdict in sorted(interesting.items()): print(f" {name:45s} {verdict}") print("\n✓ Everything the relay needs is in place.") if __name__ == "__main__": main()