From 4f5a963ef681923b84ced5d5bc0a650d77b29636 Mon Sep 17 00:00:00 2001 From: Erich Blume Date: Sat, 18 Apr 2026 16:39:21 -0700 Subject: [PATCH] Add API token auth and git remote detection to runner-logs runner-logs now always authenticates with the Forgejo API token (via --token flag, FORGEJO_TOKEN env, or 1Password) so it works on private repos. The --repo default is auto-detected from the git remote origin URL instead of being hardcoded. Co-Authored-By: Claude Opus 4.6 (1M context) --- docs/changelog.d/+runner-logs-auth.feature.md | 1 + mise-tasks/runner-logs | 94 ++++++++++++++++--- 2 files changed, 81 insertions(+), 14 deletions(-) create mode 100644 docs/changelog.d/+runner-logs-auth.feature.md diff --git a/docs/changelog.d/+runner-logs-auth.feature.md b/docs/changelog.d/+runner-logs-auth.feature.md new file mode 100644 index 0000000..7329bc6 --- /dev/null +++ b/docs/changelog.d/+runner-logs-auth.feature.md @@ -0,0 +1 @@ +runner-logs now authenticates with Forgejo API token (works on private repos) and auto-detects the repo from git remote. diff --git a/mise-tasks/runner-logs b/mise-tasks/runner-logs index 4db203d..dddeaa1 100755 --- a/mise-tasks/runner-logs +++ b/mise-tasks/runner-logs @@ -7,8 +7,9 @@ #USAGE arg "[run_number]" help="Run number to show jobs for (omit to list recent runs)" #USAGE flag "--job -j " help="Job index (0-based) to fetch logs for" #USAGE flag "--runner -r " help="Filter listing by runner: indri, ringtail, or all" -#USAGE flag "--repo " help="Forge repo (owner/name), default eblume/blumeops" +#USAGE flag "--repo " help="Forge repo (owner/name), default: detected from git remote" #USAGE flag "--limit -n " help="Max runs to display (0 for all)" +#USAGE flag "--token " help="Forgejo API token (default: read from 1Password)" """List recent Forgejo Actions runs and fetch job logs. Usage: @@ -20,6 +21,9 @@ Usage: mise run runner-logs 474 -j 1 # fetch logs for job 1 of run 474 """ +import os +import re +import subprocess import sys from typing import Annotated @@ -30,6 +34,7 @@ from rich.table import Table FORGE_URL = "https://forge.ops.eblu.me" FORGE_API = f"{FORGE_URL}/api/v1" +OP_TOKEN_REF = "op://vg6xf6vvfmoh5hqjjhlhbeoaie/w3663ffnvkewbftncqxtcpeavy/api-token" # Workflows using the ringtail nix-container-builder runner; everything else # runs on the indri k8s runner. @@ -38,11 +43,55 @@ RINGTAIL_WORKFLOWS = {"build-container-nix.yaml"} app = typer.Typer(add_completion=False) +def resolve_token(explicit_token: str | None, console: Console) -> str: + """Resolve Forgejo API token: explicit flag > FORGEJO_TOKEN env > 1Password.""" + if explicit_token: + return explicit_token + env_token = os.environ.get("FORGEJO_TOKEN", "").strip() + if env_token: + return env_token + console.print("[dim]Reading Forgejo API token from 1Password...[/dim]") + result = subprocess.run( + ["op", "read", OP_TOKEN_REF], + capture_output=True, + text=True, + check=True, + ) + return result.stdout.strip() + + +def detect_repo_from_git() -> str | None: + """Sniff owner/repo from the git remote 'origin' URL.""" + try: + result = subprocess.run( + ["git", "remote", "get-url", "origin"], + capture_output=True, + text=True, + check=True, + ) + except (subprocess.CalledProcessError, FileNotFoundError): + return None + url = result.stdout.strip() + # Match SSH (git@host:owner/repo.git) or HTTPS (https://host/owner/repo.git) + m = re.search(r"[/:]([^/]+/[^/]+?)(?:\.git)?$", url) + if not m: + return None + candidate = m.group(1) + # Only use it if the remote points at our forge + if "forge.ops.eblu.me" in url or "forge.eblu.me" in url: + return candidate + return None + + def runner_for_workflow(workflow_id: str) -> str: return "ringtail" if workflow_id in RINGTAIL_WORKFLOWS else "indri" -def fetch_tasks(repo: str) -> list[dict]: +def auth_headers(token: str) -> dict[str, str]: + return {"Authorization": f"token {token}"} + + +def fetch_tasks(repo: str, token: str) -> list[dict]: """Fetch all tasks from the Forgejo API, paginating if needed.""" tasks: list[dict] = [] page = 1 @@ -50,6 +99,7 @@ def fetch_tasks(repo: str) -> list[dict]: resp = httpx.get( f"{FORGE_API}/repos/{repo}/actions/tasks", params={"page": page, "limit": 50}, + headers=auth_headers(token), timeout=15, ) resp.raise_for_status() @@ -61,9 +111,9 @@ def fetch_tasks(repo: str) -> list[dict]: return tasks -def list_runs(runner: str, repo: str, limit: int, console: Console) -> None: +def list_runs(runner: str, repo: str, limit: int, token: str, console: Console) -> None: """List recent workflow runs, grouped by run number.""" - tasks = fetch_tasks(repo) + tasks = fetch_tasks(repo, token) # Group tasks by run_number runs: dict[int, list[dict]] = {} @@ -120,9 +170,9 @@ def list_runs(runner: str, repo: str, limit: int, console: Console) -> None: console.print("[dim] mise run runner-logs -j N to fetch logs for job N[/dim]") -def show_jobs(run_number: int, repo: str, console: Console) -> None: +def show_jobs(run_number: int, repo: str, token: str, console: Console) -> None: """Show the jobs within a specific run.""" - tasks = fetch_tasks(repo) + tasks = fetch_tasks(repo, token) jobs = sorted( [t for t in tasks if t["run_number"] == run_number], @@ -152,10 +202,10 @@ def show_jobs(run_number: int, repo: str, console: Console) -> None: console.print(f"\n[dim]Use: mise run runner-logs {run_number} -j N to fetch logs for job N[/dim]") -def fetch_log(run_number: int, job_index: int, repo: str) -> None: +def fetch_log(run_number: int, job_index: int, repo: str, token: str) -> None: """Fetch logs for a specific job via the Forgejo web endpoint.""" url = f"{FORGE_URL}/{repo}/actions/runs/{run_number}/jobs/{job_index}/attempt/1/logs" - resp = httpx.get(url, timeout=30, follow_redirects=True) + resp = httpx.get(url, headers=auth_headers(token), timeout=30, follow_redirects=True) if resp.status_code == 404: typer.echo( f"Error: No logs found for run #{run_number} job {job_index}", @@ -182,13 +232,17 @@ def main( typer.Option("--runner", "-r", help="Filter listing by runner: indri, ringtail, or all"), ] = "all", repo: Annotated[ - str, - typer.Option("--repo", help="Forge repo (owner/name)"), - ] = "eblume/blumeops", + str | None, + typer.Option("--repo", help="Forge repo (owner/name), default: detected from git remote"), + ] = None, limit: Annotated[ int, typer.Option("--limit", "-n", help="Max runs to display (0 for all)"), ] = 15, + token: Annotated[ + str | None, + typer.Option("--token", help="Forgejo API token (default: read from 1Password)"), + ] = None, ) -> None: """List recent Forgejo Actions runs or fetch logs for a specific job.""" if runner not in ("indri", "ringtail", "all"): @@ -197,15 +251,27 @@ def main( console = Console() + if repo is None: + repo = detect_repo_from_git() + if repo is None: + typer.echo( + "Error: could not detect repo from git remote; use --repo owner/name", + err=True, + ) + raise typer.Exit(1) + console.print(f"[dim]Detected repo: {repo}[/dim]") + + resolved_token = resolve_token(token, console) + if run_number is None: if job is not None: typer.echo("Error: --job requires a run number", err=True) raise typer.Exit(1) - list_runs(runner, repo, limit, console) + list_runs(runner, repo, limit, resolved_token, console) elif job is None: - show_jobs(run_number, repo, console) + show_jobs(run_number, repo, resolved_token, console) else: - fetch_log(run_number, job, repo) + fetch_log(run_number, job, repo, resolved_token) if __name__ == "__main__":