ops-docs/scripts/rotation_due.py

#!/usr/bin/env python3
"""Rotation-due watchdog for the co-op credential rotation policy.

Reads the `last rotated` table from the operator token runbook and prints a
reminder ONLY when a credential is due -- i.e. within 30 days of its due date
(last rotation + 12 months), or already past it. Prints NOTHING when nothing is
due, so it can drive a watchdog cron where empty output means "no message".

Deterministic: the output depends only on the table and the current date. It
prints no timestamps of its own, so when nothing is due it produces no output at
all and stays silent across runs.

Table source (first that applies):
  1. --file PATH
  2. $ROTATION_RUNBOOK
  3. the runbook fetched from Gitea's main branch (authoritative, always current)
  4. the local clone ~/repos/ops-docs/operator-token-runbook.md

Exit codes: 0 ok (including "nothing due", no output), 2 could not read the
runbook or found no dated rows -- a loud failure so a watcher is not silently
blind.
"""
import argparse
import calendar
import datetime as dt
import os
import re
import sys
import urllib.request

RUNBOOK_URL = "https://git.inference.coop/co-op/ops-docs/raw/branch/main/operator-token-runbook.md"
LOCAL_FALLBACK = os.path.expanduser("~/repos/ops-docs/operator-token-runbook.md")
DEFAULT_WITHIN_DAYS = 30
SECTION_RE = re.compile(r"^#{1,6}\s+.*last\s+rotated.*$", re.IGNORECASE)
HEADING_RE = re.compile(r"^#{1,6}\s+")
SEP_RE = re.compile(r"[\s|:\-]+")
DATE_RE = re.compile(r"\b(\d{4})-(\d{2})-(\d{2})\b")


def fetch_text(args):
    if args.file:
        try:
            with open(args.file, encoding="utf-8") as fh:
                return fh.read()
        except OSError as exc:
            print(f"rotation_due: could not read --file {args.file} ({exc})", file=sys.stderr)
            raise SystemExit(2)
    env = os.environ.get("ROTATION_RUNBOOK")
    if env and os.path.exists(env):
        try:
            with open(env, encoding="utf-8") as fh:
                return fh.read()
        except OSError as exc:
            print(f"rotation_due: could not read $ROTATION_RUNBOOK {env} ({exc})", file=sys.stderr)
            raise SystemExit(2)
    try:
        req = urllib.request.Request(RUNBOOK_URL, headers={"User-Agent": "rotation_due/1.0"})
        with urllib.request.urlopen(req, timeout=20) as resp:
            return resp.read().decode("utf-8")
    except Exception as exc:  # noqa: BLE001 - fall back, then fail loudly
        if os.path.exists(LOCAL_FALLBACK):
            with open(LOCAL_FALLBACK, encoding="utf-8") as fh:
                return fh.read()
        print(
            f"rotation_due: could not read the runbook from {RUNBOOK_URL} ({exc}) "
            f"and no local copy at {LOCAL_FALLBACK}",
            file=sys.stderr,
        )
        raise SystemExit(2)


def extract_section(text):
    """Return the lines of the first 'last rotated' section."""
    out, inside = [], False
    for line in text.splitlines():
        if SECTION_RE.match(line):
            inside = True
            continue
        if inside and HEADING_RE.match(line):
            break
        if inside:
            out.append(line)
    return out


def parse_rows(section_lines):
    """Yield (name, last_rotated_date) for table rows that carry a date."""
    for line in section_lines:
        if not line.lstrip().startswith("|"):
            continue
        stripped = line.strip().strip("|")
        if not stripped or SEP_RE.fullmatch(line.strip()):
            continue
        cells = [c.strip() for c in stripped.split("|")]
        if not cells:
            continue
        match = DATE_RE.search(line)
        if not match:
            continue
        year, month, day = (int(g) for g in match.groups())
        try:
            last = dt.date(year, month, day)
        except ValueError:
            continue
        yield cells[0], last


def add_months(date, months):
    month_index = date.month - 1 + months
    year = date.year + month_index // 12
    month = month_index % 12 + 1
    day = min(date.day, calendar.monthrange(year, month)[1])
    return dt.date(year, month, day)


def main():
    ap = argparse.ArgumentParser(description="Remind when a credential rotation is due.")
    ap.add_argument("--file", help="read the runbook from this path instead of fetching it")
    ap.add_argument("--within-days", type=int, default=DEFAULT_WITHIN_DAYS,
                    help="days before the due date to start reminding (default 30)")
    ap.add_argument("--as-of", help="evaluate as of this date (YYYY-MM-DD); default today UTC")
    args = ap.parse_args()

    section = extract_section(fetch_text(args))
    if not section:
        print("rotation_due: no 'last rotated' section found in the runbook", file=sys.stderr)
        return 2

    parsed = list(parse_rows(section))
    if not parsed:
        print("rotation_due: the 'last rotated' section has no rows carrying a date", file=sys.stderr)
        return 2

    as_of = dt.date.fromisoformat(args.as_of) if args.as_of else dt.datetime.now(dt.timezone.utc).date()

    due = []
    for name, last in parsed:
        due_date = add_months(last, 12)
        days = (due_date - as_of).days
        if days <= args.within_days:
            due.append((days, name, due_date))

    if not due:
        return 0  # nothing due -> no output at all

    due.sort()
    for days, name, due_date in due:
        if days < 0:
            print(f"credential rotation OVERDUE: {name} -- was due {due_date.isoformat()} ({-days} days ago)")
        else:
            print(f"credential rotation due soon: {name} -- due {due_date.isoformat()} (in {days} days)")
    return 0


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