awoooi/ops/runner/guard-gitea-runner-pressure.py

#!/usr/bin/env python3
"""Guard Gitea workflows from reintroducing 110 runner pressure triggers.

This is a source guard only. It reads committed `.gitea/workflows` files and
fails when automatic branch events target 110 incident runner labels, or when a
generic runner label is reintroduced.
"""

from __future__ import annotations

import argparse
import re
import sys
from dataclasses import dataclass
from pathlib import Path


AUTO_BRANCH_EVENTS = {"push", "pull_request", "pull_request_target"}
INCIDENT_RUNNER_LABELS = {"awoooi-ubuntu", "awoooi-host"}
GENERIC_LABELS = {"ubuntu-latest", "self-hosted"}

RUNS_ON_RE = re.compile(r"^\s*runs-on:\s*(?P<value>.+?)\s*$")
ON_RE = re.compile(r"^on:\s*(?P<value>.*)$")
ON_CHILD_EVENT_RE = re.compile(r"^(?P<indent>\s+)(?P<event>[A-Za-z_][A-Za-z0-9_-]*)\s*:")


@dataclass(frozen=True)
class WorkflowInfo:
    path: Path
    events: set[str]
    labels: list[tuple[int, str]]


@dataclass(frozen=True)
class Violation:
    path: Path
    line: int
    label: str
    reason: str
    events: set[str]


def parse_args() -> argparse.Namespace:
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("--root", type=Path, default=Path.cwd(), help="Repository root")
    parser.add_argument(
        "--workflow-dir",
        default=".gitea/workflows",
        help="Workflow directory relative to --root",
    )
    return parser.parse_args()


def remove_inline_comment(value: str) -> str:
    return value.split("#", 1)[0].rstrip()


def strip_inline_comment(value: str) -> str:
    return remove_inline_comment(value).strip()


def parse_inline_events(value: str) -> set[str]:
    clean = strip_inline_comment(value).strip()
    if not clean:
        return set()
    if clean.startswith("[") and clean.endswith("]"):
        return {part.strip().strip("'\"") for part in clean[1:-1].split(",") if part.strip()}
    if clean.startswith("{") and clean.endswith("}"):
        return {
            part.split(":", 1)[0].strip().strip("'\"")
            for part in clean[1:-1].split(",")
            if part.strip()
        }
    return {clean.strip().strip("'\"")}


def parse_events(lines: list[str]) -> set[str]:
    events: set[str] = set()
    for index, line in enumerate(lines):
        match = ON_RE.match(remove_inline_comment(line))
        if not match:
            continue
        inline_events = parse_inline_events(match.group("value"))
        if inline_events:
            events.update(inline_events)
            continue

        for child in lines[index + 1 :]:
            if not child.strip() or child.lstrip().startswith("#"):
                continue
            if child[0] not in {" ", "\t"}:
                break
            child_match = ON_CHILD_EVENT_RE.match(remove_inline_comment(child))
            if child_match:
                events.add(child_match.group("event"))
    return events


def normalize_runs_on(value: str) -> list[str]:
    clean = strip_inline_comment(value).strip()
    if clean.startswith("[") and clean.endswith("]"):
        return [part.strip().strip("'\"") for part in clean[1:-1].split(",") if part.strip()]
    return [clean.strip().strip("'\"")]


def parse_workflow(path: Path) -> WorkflowInfo:
    lines = path.read_text(encoding="utf-8").splitlines()
    labels: list[tuple[int, str]] = []
    for line_number, line in enumerate(lines, start=1):
        match = RUNS_ON_RE.match(line)
        if not match:
            continue
        for label in normalize_runs_on(match.group("value")):
            labels.append((line_number, label))
    return WorkflowInfo(path=path, events=parse_events(lines), labels=labels)


def label_is_generic(label: str) -> bool:
    if label in GENERIC_LABELS:
        return True
    return label.startswith("ubuntu-")


def inspect(workflow_dir: Path) -> tuple[list[WorkflowInfo], list[Violation]]:
    workflows = [parse_workflow(path) for path in sorted(workflow_dir.glob("*.y*ml"))]
    violations: list[Violation] = []

    for workflow in workflows:
        auto_events = workflow.events & AUTO_BRANCH_EVENTS
        for line, label in workflow.labels:
            if label_is_generic(label):
                violations.append(
                    Violation(
                        path=workflow.path,
                        line=line,
                        label=label,
                        reason="generic_runner_label_reintroduced",
                        events=workflow.events,
                    )
                )
            if auto_events and label in INCIDENT_RUNNER_LABELS:
                violations.append(
                    Violation(
                        path=workflow.path,
                        line=line,
                        label=label,
                        reason="auto_branch_event_targets_110_incident_runner",
                        events=workflow.events,
                    )
                )
    return workflows, violations


def main() -> int:
    args = parse_args()
    root = args.root.expanduser().resolve()
    workflow_dir = root / args.workflow_dir
    if not workflow_dir.exists():
        print(f"GITEA_RUNNER_PRESSURE_GUARD_BLOCKED workflow_dir_missing={workflow_dir}")
        return 2

    workflows, violations = inspect(workflow_dir)
    if violations:
        print("GITEA_RUNNER_PRESSURE_GUARD_BLOCKED")
        for item in violations:
            rel_path = item.path.relative_to(root)
            events = ",".join(sorted(item.events)) or "none"
            print(
                f"{rel_path}:{item.line} label={item.label} "
                f"reason={item.reason} events={events}"
            )
        return 1

    scheduled = sum(1 for workflow in workflows if "schedule" in workflow.events)
    print(
        "GITEA_RUNNER_PRESSURE_GUARD_OK "
        f"workflow_files={len(workflows)} "
        f"scheduled_workflows={scheduled} "
        "auto_branch_events_on_110=0 "
        "generic_runner_labels=0"
    )
    return 0


if __name__ == "__main__":
    sys.exit(main())