cc-ci-orchestrator/cc-ci-plan/audit-sources.py

#!/usr/bin/env python3
"""audit-sources — are we still looking in the right place for each recipe's updates?

A recipe tracks an image repo and a set of registry URLs. Upstreams move: they rename the image,
switch registry, archive the GitHub repo, or split a community edition out of the original. When that
happens nothing errors — the old repo simply stops receiving tags, and the recipe looks "up to date"
forever while real releases happen somewhere else.

plausible is the worked example. It tracked `plausible/analytics` on Docker Hub; upstream moved to
`ghcr.io/plausible/community-edition`. The old repo still exists and still serves v2.0.0, so every
survey said "no upgrades available" while v3 shipped elsewhere.

This reports the signals that catch that, per image and per registry URL:

  * IMAGE GONE QUIET     — newest tag is older than --quiet-days (default 365). The single strongest
                           signal that releases moved somewhere else.
  * DEPRECATION WORDING  — the registry description says deprecated / moved / no longer maintained.
  * GITHUB REPO ARCHIVED — upstream archived it.
  * GITHUB REPO RENAMED  — the API redirects to a different owner/name than we ask for.
  * GITHUB REPO GONE     — 404.

Everything is a SIGNAL, not a verdict: a genuinely stable image (mumble, custom-html) can be quiet
for good reason. The output is for a human to judge, so each finding says what was measured.

    audit-sources.py [recipe ...] [--ssh HOST] [--quiet-days N] [--json]
"""

from __future__ import annotations

import argparse
import importlib.util
import json
import os
import re
import sys
import urllib.error
import urllib.request
from datetime import datetime, timezone

HERE = os.path.dirname(os.path.abspath(__file__))
_spec = importlib.util.spec_from_file_location("resolve_images", os.path.join(HERE, "resolve-images.py"))
RI = importlib.util.module_from_spec(_spec)
_spec.loader.exec_module(RI)

REGISTRY_DIR = os.environ.get("CCCI_UPSTREAM_REGISTRY", os.path.join(HERE, "upstream"))
USED_RECIPES = os.path.join(HERE, "used-recipes.md")
DEPRECATION_RE = re.compile(
    r"\b(deprecat|no longer maintain|unmaintained|superseded|moved to|migrated to|"
    r"has moved|discontinued|end.of.life|archived)\b", re.I)


def _days_since(iso: str | None) -> int | None:
    if not iso:
        return None
    try:
        d = datetime.fromisoformat(iso.replace("Z", "+00:00"))
    except ValueError:
        return None
    return (datetime.now(timezone.utc) - d).days


def hub_repo_meta(repo: str) -> dict:
    """Docker Hub repo metadata: when it was last pushed to, and how it describes itself."""
    try:
        d = RI._json(f"https://hub.docker.com/v2/repositories/{repo}", RI._hub_auth())
    except urllib.error.HTTPError as e:
        return {"status": f"HTTP {e.code}"}
    except Exception as e:  # noqa: BLE001
        return {"status": f"{type(e).__name__}"}
    text = f"{d.get('description') or ''}\n{d.get('full_description') or ''}"
    m = DEPRECATION_RE.search(text)
    return {"status": "ok", "last_updated": d.get("last_updated"),
            "deprecation_hint": (m.group(0) if m else None),
            "archived": bool(d.get("is_archived") or d.get("status") == "inactive")}


def github_repo_meta(owner: str, repo: str) -> dict:
    """GitHub repo state — archived, renamed (the API answers with the CURRENT full_name), or gone."""
    hdrs = {"Accept": "application/vnd.github+json"}
    tok = RI._gh_token()
    if tok:
        hdrs["Authorization"] = f"Bearer {tok}"
    try:
        d = RI._json(f"https://api.github.com/repos/{owner}/{repo}", hdrs)
    except urllib.error.HTTPError as e:
        return {"status": f"HTTP {e.code}"}
    except Exception as e:  # noqa: BLE001
        return {"status": f"{type(e).__name__}"}
    asked, got = f"{owner}/{repo}".lower(), (d.get("full_name") or "").lower()
    return {"status": "ok", "archived": bool(d.get("archived")), "pushed_at": d.get("pushed_at"),
            "renamed_to": (d.get("full_name") if got and got != asked else None),
            "description": d.get("description") or ""}


def newest_tag_date(registry: str, repo: str, tag: str) -> str | None:
    """When was the repo's newest same-shape tag pushed? Docker Hub only (it dates its tags)."""
    if registry not in ("docker.io", "registry-1.docker.io"):
        return None
    try:
        d = RI._json(f"https://hub.docker.com/v2/repositories/{repo}/tags"
                     f"?page_size=100&ordering=last_updated", RI._hub_auth())
    except Exception:  # noqa: BLE001
        return None
    want = RI.shape(tag)
    for row in d.get("results", []):
        if RI.shape(row.get("name") or "") == want:
            return row.get("last_updated")
    return (d.get("results") or [{}])[0].get("last_updated")


def audit_recipe(recipe: str, ssh: str | None, quiet_days: int) -> dict:
    out = {"recipe": recipe, "findings": [], "images": [], "sources": []}
    try:
        refs = (RI.compose_images_ssh(recipe, ssh, "~/.abra/recipes") if ssh
                else RI.compose_images(recipe, RI.RECIPE_DIR))
    except Exception as e:  # noqa: BLE001
        out["findings"].append({"level": "error", "what": f"could not read compose: {e}"})
        return out

    for ref in refs:
        if "${" in ref:
            continue
        info = RI.parse_ref(ref)
        row = {"ref": ref, "registry": info["registry"], "repo": info["repo"], "tag": info["tag"]}
        if info["registry"] in ("docker.io", "registry-1.docker.io"):
            meta = hub_repo_meta(info["repo"])
            row.update(meta)
            newest = newest_tag_date(info["registry"], info["repo"], info["tag"])
            row["newest_tag_pushed"] = newest
            age = _days_since(newest)
            row["newest_tag_age_days"] = age
            if age is not None and age > quiet_days:
                out["findings"].append({
                    "level": "warn", "what": "image has gone quiet",
                    "detail": f"{info['repo']}: newest {RI.shape(info['tag'])}-shaped tag pushed "
                              f"{age} days ago — releases may have moved elsewhere"})
            if meta.get("deprecation_hint"):
                out["findings"].append({
                    "level": "warn", "what": "registry text suggests deprecation",
                    "detail": f"{info['repo']}: says {meta['deprecation_hint']!r}"})
            if meta.get("archived"):
                out["findings"].append({"level": "warn", "what": "registry repo archived/inactive",
                                        "detail": info["repo"]})
        out["images"].append(row)

    urls, reg_path = ([], None)
    try:
        urls, reg_path = _registry_urls(recipe)
    except Exception:  # noqa: BLE001
        pass
    if reg_path is None:
        out["findings"].append({"level": "warn", "what": "no upstream registry file",
                                "detail": f"cc-ci-plan/upstream/{recipe}.md is missing — the advisory "
                                          f"scan has nowhere to look"})
    seen = set()
    for u in urls:
        m = re.match(r"https?://github\.com/([^/]+)/([^/#?]+)", u)
        if not m:
            continue
        owner, repo = m.group(1), m.group(2).removesuffix(".git")
        if (owner, repo) in seen:
            continue
        seen.add((owner, repo))
        meta = github_repo_meta(owner, repo)
        row = {"repo": f"{owner}/{repo}", **meta}
        age = _days_since(meta.get("pushed_at"))
        row["pushed_age_days"] = age
        out["sources"].append(row)
        if meta.get("status") != "ok":
            out["findings"].append({"level": "warn", "what": "registry source unreachable",
                                    "detail": f"{owner}/{repo}: {meta['status']}"})
            continue
        if meta.get("renamed_to"):
            out["findings"].append({"level": "alert", "what": "GitHub repo has MOVED",
                                    "detail": f"{owner}/{repo} now answers as {meta['renamed_to']}"})
        if meta.get("archived"):
            out["findings"].append({"level": "alert", "what": "GitHub repo is ARCHIVED",
                                    "detail": f"{owner}/{repo} — upstream development has stopped here"})
        if age is not None and age > quiet_days:
            out["findings"].append({"level": "warn", "what": "GitHub repo quiet",
                                    "detail": f"{owner}/{repo}: last push {age} days ago"})
    return out


def _registry_urls(recipe: str):
    path = os.path.join(REGISTRY_DIR, f"{recipe}.md")
    if not os.path.exists(path):
        return [], None
    text = open(path).read()
    urls = []
    for u in re.findall(r"https?://[^\s)|\]]+", text):
        u = u.rstrip("`'\"*.,;:>)")
        if u and u not in urls:
            urls.append(u)
    return urls, path


def all_recipes() -> list[str]:
    out = []
    for ln in open(USED_RECIPES):
        ln = ln.strip()
        if not ln or ln.startswith("#") or ln.startswith("`"):
            continue
        parts = ln.split()
        if len(parts) >= 2 and parts[1] in ("weekly", "external"):
            out.append(parts[0])
    return out


def main() -> int:
    ap = argparse.ArgumentParser(description=__doc__,
                                 formatter_class=argparse.RawDescriptionHelpFormatter)
    ap.add_argument("recipes", nargs="*")
    ap.add_argument("--ssh", default=None)
    ap.add_argument("--quiet-days", type=int, default=365)
    ap.add_argument("--json", action="store_true")
    a = ap.parse_args()

    recipes = a.recipes or all_recipes()
    reports = [audit_recipe(r, a.ssh, a.quiet_days) for r in recipes]
    if a.json:
        print(json.dumps(reports, indent=2))
        return 0
    alerts = 0
    for rep in reports:
        fs = rep["findings"]
        mark = "OK " if not fs else ("!! " if any(f["level"] == "alert" for f in fs) else " ? ")
        print(f"{mark} {rep['recipe']}")
        for f in fs:
            alerts += f["level"] == "alert"
            print(f"      [{f['level']}] {f['what']}: {f.get('detail','')}")
    print(f"\n{len(reports)} recipes audited · "
          f"{sum(len(r['findings']) for r in reports)} findings · {alerts} alerts")
    return 0


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