#!/usr/bin/env python3 """Operator-side upstream discovery; output is a review queue, never an update pin. Run this hourly from an operator scheduler with an explicit persistent --state path. It reads only manifests with an explicit supported app.upstream source. The signed app catalog remains the sole authority for installable updates. """ from __future__ import annotations import argparse from concurrent.futures import ThreadPoolExecutor import importlib.util import json import os import re import tempfile import time from pathlib import Path from typing import Any, Callable CHECK_PATH = Path(__file__).with_name("check-upstream-releases.py") spec = importlib.util.spec_from_file_location("upstream_release_check", CHECK_PATH) assert spec and spec.loader checker = importlib.util.module_from_spec(spec) import sys sys.modules[spec.name] = checker spec.loader.exec_module(checker) SUPPORTED = frozenset(checker.FETCHERS) REPO_PART = re.compile(r"[A-Za-z0-9_.-]{1,100}\Z") TAG = re.compile(r"[A-Za-z0-9._+\-]{1,128}\Z") GHSA = re.compile(r"GHSA-[A-Za-z0-9-]{4,64}\Z") def valid_repo(repo: str, github: bool = False) -> bool: parts = repo.split("/") return (len(parts) == 2 if github else 2 <= len(parts) <= 5) and all( part not in {".", ".."} and REPO_PART.fullmatch(part) for part in parts) def published_github_advisories(repo: str) -> set[str]: """Only public, published IDs; an API failure is not an empty result.""" if not valid_repo(repo, github=True): raise ValueError("invalid GitHub repository identifier") rows = checker.http_json( f"https://api.github.com/repos/{repo}/security-advisories?state=published&per_page=30", {"Accept": "application/vnd.github+json"}, ) if not isinstance(rows, list): raise ValueError("invalid advisory response") return {row["ghsa_id"] for row in rows if isinstance(row, dict) and isinstance(row.get("ghsa_id"), str) and GHSA.fullmatch(row["ghsa_id"]) and row.get("state") == "published"} def discover( pins: list[Any], previous: dict[str, Any], max_apps: int, check: Callable[[Any, bool], dict[str, Any]] = checker.check, advisories: Callable[[str], set[str]] = published_github_advisories, offline: bool = False, ) -> tuple[dict[str, Any], list[dict[str, str]]]: """Advance a bounded round-robin scan and return newly seen review items.""" eligible = sorted((pin for pin in pins if pin.upstream.get("kind") in SUPPORTED and valid_repo(str(pin.upstream.get("repo", "")))), key=lambda pin: pin.app_id) state = {"version": 1, "cursor": 0, "apps": dict(previous.get("apps", {})), "pending_review": list(previous.get("pending_review", [])), "degraded_sources": []} if not eligible: return state, [] cursor = int(previous.get("cursor", 0)) % len(eligible) notices: list[dict[str, str]] = [] selected = [eligible[(cursor + offset) % len(eligible)] for offset in range(min(max_apps, len(eligible)))] def scan(pin: Any) -> tuple[dict[str, Any], set[str] | None, bool]: row = check(pin, offline) if pin.upstream["kind"] != "github" or offline: return row, None, False try: return row, advisories(pin.upstream["repo"]), False except (OSError, ValueError, checker.RateLimited): return row, None, True # Bounded parallel I/O avoids a slow upstream consuming the entire hourly # discovery window. State and review notices are still merged in source order. with ThreadPoolExecutor(max_workers=min(4, len(selected))) as pool: scanned = list(pool.map(scan, selected)) for pin, (row, current, advisory_failed) in zip(selected, scanned): record = dict(state["apps"].get(pin.app_id, {})) if row.get("status") in {"ERROR", "RATE-LIMITED"}: state["degraded_sources"].append({"app": pin.app_id, "reason": "release_source_unavailable"}) if advisory_failed: state["degraded_sources"].append({"app": pin.app_id, "reason": "advisory_source_unavailable"}) latest, shipped = row.get("latest", ""), row.get("shipped", "") if (row.get("status") in {"BEHIND", "UNCOMPARABLE"} and isinstance(latest, str) and TAG.fullmatch(latest) and latest != shipped and latest != record.get("release")): notices.append({"type": "upstream_release", "app": pin.app_id, "release": latest, "shipped": str(shipped)[:128], "status": row["status"], "action": "review_only"}) if isinstance(latest, str) and TAG.fullmatch(latest) and row.get("status") not in {"ERROR", "RATE-LIMITED"}: record["release"] = latest if current is not None: known = set(record.get("advisories", [])) for ghsa in sorted(current - known): notices.append({"type": "security_advisory", "app": pin.app_id, "id": ghsa, "status": "published", "action": "review_only"}) record["advisories"] = sorted(known | current)[-256:] state["apps"][pin.app_id] = record state["cursor"] = (cursor + len(selected)) % len(eligible) queued = {json.dumps(item, sort_keys=True) for item in state["pending_review"]} for item in notices: key = json.dumps(item, sort_keys=True) if key not in queued: state["pending_review"].append(item) queued.add(key) return state, notices def write_state(path: Path, state: dict[str, Any]) -> None: path.parent.mkdir(mode=0o700, parents=True, exist_ok=True) fd, temporary = tempfile.mkstemp(prefix=".upstream-", dir=path.parent) try: os.fchmod(fd, 0o600) with os.fdopen(fd, "w", encoding="utf-8") as out: json.dump(state, out, sort_keys=True, separators=(",", ":")) out.write("\n") out.flush() os.fsync(out.fileno()) os.replace(temporary, path) parent_fd = os.open(path.parent, os.O_RDONLY | os.O_DIRECTORY) try: os.fsync(parent_fd) finally: os.close(parent_fd) finally: if os.path.exists(temporary): os.unlink(temporary) def main() -> int: parser = argparse.ArgumentParser(description=__doc__) parser.add_argument("--state", type=Path, required=True, help="persistent operator-only JSON state; never put on installed nodes") parser.add_argument("--max-apps", type=int, help="sources per run; default all with GitHub token, four otherwise") parser.add_argument("--interval-minutes", type=int, help="run continuously at this cadence under an operator supervisor (minimum 60)") parser.add_argument("--offline", action="store_true", help="validate source coverage without network or state writes") args = parser.parse_args() has_token = bool(os.environ.get("GITHUB_TOKEN") or os.environ.get("GH_TOKEN")) max_apps = args.max_apps if args.max_apps is not None else (100 if has_token or args.offline else 4) if not 1 <= max_apps <= 100: parser.error("--max-apps must be between 1 and 100") if not has_token and not args.offline and max_apps > 4: parser.error("unauthenticated scans are limited to four sources per hour") if args.interval_minutes is not None and not 60 <= args.interval_minutes <= 1440: parser.error("--interval-minutes must be between 60 and 1440") while True: previous = json.loads(args.state.read_text()) if args.state.exists() else {} if previous and previous.get("version") != 1: parser.error("unknown state version; preserve it for review") pins = checker.load_apps([]) if not pins: parser.error("no app manifests found; use a full source checkout") if not any(pin.upstream.get("kind") in SUPPORTED and valid_repo(str(pin.upstream.get("repo", ""))) for pin in pins): parser.error("no supported upstream feeds found; preserve prior review state") state, notices = discover(pins, previous, max_apps, offline=args.offline) if not args.offline: write_state(args.state, state) print(json.dumps({"notices": notices, "pending_review": len(state["pending_review"]), "scanned": min(max_apps, sum( pin.upstream.get("kind") in SUPPORTED and valid_repo(str(pin.upstream.get("repo", ""))) for pin in pins)), "mode": "offline" if args.offline else "authenticated_full_cycle" if has_token else "degraded_no_github_token", "degraded_sources": state["degraded_sources"], "installable_updates": "signed_catalog_only", "offline": args.offline}, sort_keys=True), flush=True) if args.interval_minutes is None: break time.sleep(args.interval_minutes * 60) return 0 if __name__ == "__main__": raise SystemExit(main())