188 lines
9.0 KiB
Python
188 lines
9.0 KiB
Python
#!/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())
|