home-network-k3s-impl/ci/approval-override.py

#!/usr/bin/env python3
"""ci/approval-override.py — select the admin-override on a PR, if there is a valid one.

This script decides NOTHING about authority. It parses a PR's comments and reviews and returns
the single override *candidate* that ci/approval-policy.sh should then try to authorise against
the live Gitea permission API. Keeping the two apart is the point:

  * parsing is pure, offline, and therefore TESTABLE (ci/tests/approval-override.test.sh);
  * authorisation needs the API and must stay in the stage that fails closed.

A candidate is a comment OR a review whose body contains, AT THE START OF A LINE, the directive
followed by a non-empty reason:

    /approve-override the gate cannot gate itself and I am the only human reviewer

TWO SOURCES, AND WHY THAT IS NOT BELT-AND-BRACES. Measured 2026-10-02 on PR #38: the override was
posted as an ordinary PR comment, was well-formed, was newer than the head commit — and the stage
still blocked, because CI could not READ it. `SETTINGS_API_TOKEN` is minted with scopes
`[read:repository, read:admin]`, and Gitea puts /issues/{n}/comments behind the *issue* scope, so
the request came back as an error OBJECT rather than an array. `/pulls/{n}/reviews` is under
read:repository and was readable in the same run. So:

  * a REVIEW body works with the token CI has today (leave a review of type "Comment");
  * a COMMENT body works once the token carries read:issue (see infra/gitea/set-ci-token.sh).

Accepting both means the hatch is not hostage to one API scope. A source that cannot be read
contributes no candidates and is reported on stderr; it can only cause a false BLOCK, never a
false pass, which is why degrading it is safe and blocking on it was not necessary.

WHY EACH RULE IS HERE

  * START OF A LINE, not "anywhere in the body". Otherwise quoting someone else's override
    ("you could just /approve-override this") re-arms it. Lines beginning with '>' are DROPPED
    rather than unquoted, for exactly that reason.
  * A REASON IS MANDATORY. A bare directive records that a review was waived but not why, which
    is the difference between an audit trail and a rubber stamp.
  * STALE OVERRIDES ARE DISCARDED. An override earned at commit A must not silently cover commit
    B. Any candidate created at or before the head commit's timestamp is dropped — the same
    semantics Gitea's own `dismiss_stale_approvals` applies to approvals. Push new commits and
    you must re-override.
  * A PENDING REVIEW IS NOT SUBMITTED. It is a private draft visible only to its author, so it is
    not an act of anything and is dropped.
  * THE LATEST CANDIDATE WINS, across both sources, so a later override supersedes an earlier one.

Output: one TSV line (login, timestamp, source, reason) on stdout, or nothing at all when there is
no valid candidate. Never exits non-zero for "no override" — absence is a normal answer and the
caller must not be able to confuse it with a parse crash.

Input, all via env so nothing lands in `ps`:
  COMMENTS_JSON  the Gitea /issues/{n}/comments array (or an error object; or absent)
  REVIEWS_JSON   the Gitea /pulls/{n}/reviews array   (or an error object; or absent)
  HEAD_TIME      ISO-8601 timestamp of the PR head commit
  DIRECTIVE      the override directive, e.g. /approve-override
"""
import json
import os
import re
import sys
from datetime import datetime, timezone


def parse_ts(value):
    """Gitea emits RFC3339. Return an aware datetime, or None if it is unparseable.

    Unparseable is NOT 'treat as fresh': the caller drops such a candidate, because a
    timestamp we cannot read is a staleness check we cannot perform.
    """
    if not value:
        return None
    text = str(value).strip()
    # Python < 3.11 cannot read a trailing 'Z' in fromisoformat.
    if text.endswith(("Z", "z")):
        text = text[:-1] + "+00:00"
    try:
        parsed = datetime.fromisoformat(text)
    except ValueError:
        return None
    if parsed.tzinfo is None:
        parsed = parsed.replace(tzinfo=timezone.utc)
    return parsed


def extract_reason(body, directive):
    """Return the reason from the first line that STARTS with the directive, else None."""
    pattern = re.compile(
        r"^[ \t]*" + re.escape(directive) + r"(?=[ \t:]|$)[ \t:]*(.*)$"
    )
    for raw in (body or "").splitlines():
        # A quoted line is somebody repeating an override, not casting one.
        if raw.lstrip().startswith(">"):
            continue
        match = pattern.match(raw)
        if match:
            reason = match.group(1).strip()
            return reason or None
    return None


def load_source(env_name, label):
    """Return a list of items for one source, or [] with a stderr note if it is unusable."""
    raw = os.environ.get(env_name)
    if not raw or not raw.strip():
        return []
    try:
        parsed = json.loads(raw)
    except (ValueError, TypeError):
        print(
            "approval-override: %s is not valid JSON — ignoring that source" % label,
            file=sys.stderr,
        )
        return []
    if isinstance(parsed, list):
        return parsed
    # An error object, typically {"message": "token does not have ... scope"}. Say which,
    # because "no override found" and "could not read the overrides" look identical from
    # the outside and the first wasted a real debugging session.
    detail = ""
    if isinstance(parsed, dict):
        detail = str(parsed.get("message") or parsed.get("errors") or "")[:200]
    print(
        "approval-override: %s is not a list (Gitea returned %s%s) — ignoring that source"
        % (label, type(parsed).__name__, ": " + detail if detail else ""),
        file=sys.stderr,
    )
    return []


def candidates_from(items, source, directive, head_time):
    """Yield (when, login, reason, raw_when, source) for every valid override in one source."""
    found = []
    for item in items:
        if not isinstance(item, dict):
            continue
        login = ((item.get("user") or {}).get("login") or "").strip()
        reason = extract_reason(item.get("body"), directive)
        if not login or not reason:
            continue
        # A PENDING review is an unsubmitted private draft — not an act of anything.
        state = str(item.get("state") or "").upper()
        if source == "review" and state == "PENDING":
            print(
                "approval-override: dropping PENDING (unsubmitted) review by '%s'" % login,
                file=sys.stderr,
            )
            continue
        raw_when = (
            item.get("submitted_at") or item.get("created_at") or item.get("updated_at") or ""
        )
        when = parse_ts(raw_when)
        if when is None:
            print(
                "approval-override: dropping %s override by '%s' — unreadable timestamp %r"
                % (source, login, raw_when),
                file=sys.stderr,
            )
            continue
        # An override must post-date the commit it covers.
        if head_time is not None and when <= head_time:
            print(
                "approval-override: dropping STALE %s override by '%s' (%s) — not newer than "
                "head commit (%s). Re-post it to override the current commit."
                % (source, login, when.isoformat(), head_time.isoformat()),
                file=sys.stderr,
            )
            continue
        found.append((when, login, reason, raw_when, source))
    return found


def main():
    directive = os.environ.get("DIRECTIVE", "/approve-override").strip()
    head_time = parse_ts(os.environ.get("HEAD_TIME", ""))

    found = []
    found += candidates_from(
        load_source("COMMENTS_JSON", "COMMENTS_JSON"), "comment", directive, head_time
    )
    found += candidates_from(
        load_source("REVIEWS_JSON", "REVIEWS_JSON"), "review", directive, head_time
    )

    if not found:
        return 0

    # Latest wins, across both sources: a later override supersedes an earlier one.
    when, login, reason, raw_when, source = max(found, key=lambda c: c[0])
    # Keep the reason on one line — this becomes a TSV field.
    reason = " ".join(reason.split())
    print("%s\t%s\t%s\t%s" % (login, raw_when, source, reason))
    return 0


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