#!/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) # advisory-scan supplies the source-fetching + changelog-attribution used by --security-sources _aspec = importlib.util.spec_from_file_location("advisory_scan", os.path.join(HERE, "advisory-scan.py")) A = importlib.util.module_from_spec(_aspec) _aspec.loader.exec_module(A) 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 security_source_audit(recipe: str) -> list[dict]: """Per source: are its CVEs USABLE, or merely visible? The nginx lesson. nginx publishes no GitHub advisories; all its CVEs live in nginx.org/en/CHANGES. The scan saw them and could do nothing with them, because nothing said which release fixed which CVE — so every nginx bump in the fleet reported 0. Attribution (advisory-scan §2b) fixed that for changelogs organised by release, but a page that lists CVEs with NO release structure is still a blind spot: visible, uncountable. This finds those. Per source: `advisory-feed` (structured, best), `changelog` (CVEs attributable to a release), `unattributable` (CVEs present but no release structure — BLIND), or `no-cve-data`. """ urls, _ = _registry_urls(recipe) out = [] # NVD CPE entries are a first-class source: for projects publishing nothing machine-readable # (mattermost, mumble) they are the ONLY structured source, and omitting them here made two # recipes look permanently blind after they had been fixed. for key, cpe in A.registry_cpes(recipe, REGISTRY_DIR): e = A.nvd_advisories(cpe, key) n = len(e.get("advisories") or []) out.append({"source": e["source"] + f" ({cpe.split(':')[4]}/{cpe.split(':')[3]})", "kind": "advisory-feed" if n else "no-cve-data", "status": e["status"], "cves": n, "usable": n}) for entry in A.github_advisories(urls): out.append({"source": entry["source"], "kind": "advisory-feed", "status": entry["status"], "cves": len(entry.get("advisories") or []), "usable": len(entry.get("advisories") or [])}) for entry in A.vendor_pages(urls): if entry["status"].startswith("skipped"): continue n = len(entry.get("cves") or []) attributed = len(entry.get("fixed_in") or {}) kind = ("no-cve-data" if n == 0 else "changelog" if attributed else "unattributable") out.append({"source": entry["source"], "kind": kind, "status": entry["status"], "cves": n, "usable": attributed}) return out 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") ap.add_argument("--security-sources", action="store_true", help="audit whether each recipe's CVE sources are USABLE (structured advisory " "feed / release-attributable changelog) or merely visible") a = ap.parse_args() recipes = a.recipes or all_recipes() if a.security_sources: # What matters is whether the RECIPE can see CVEs at all — not whether some individual page # is unparseable. A page with no release structure is harmless when the same project also # publishes an advisory feed (redis, gitea, minio, clickhouse all do); it is only a blind # spot when nothing else covers that project. blind_recipes, noisy = [], 0 for r in recipes: rows = security_source_audit(r) feeds = [x for x in rows if x["kind"] == "advisory-feed" and x["cves"] > 0] logs = [x for x in rows if x["kind"] == "changelog"] unattr = [x for x in rows if x["kind"] == "unattributable"] noisy += len(unattr) usable = len(feeds) + len(logs) if usable == 0: blind_recipes.append(r) print(f"!! {r}: NO USABLE CVE SOURCE — {len(unattr)} unparseable page(s), " f"0 advisory feeds, 0 attributable changelogs") for x in rows: print(f" {x['kind']:15} {x['source'][:64]} ({x['cves']} CVEs)") else: print(f"OK {r}: {len(feeds)} advisory-feed(s), {len(logs)} changelog(s)" + (f", {len(unattr)} unparseable page(s) (redundant — covered by a feed)" if unattr else "")) for x in logs: print(f" changelog {x['source'][:62]} ({x['usable']}/{x['cves']})") print(f"\n{len(recipes)} recipes · {len(blind_recipes)} with NO usable CVE source" + (f": {', '.join(blind_recipes)}" if blind_recipes else "") + f" · {noisy} unparseable page(s) elsewhere (harmless where a feed covers them)") return 0 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())