Files
cc-ci-orchestrator/cc-ci-plan/advisory-scan.py
T
autonomic-bot 98a624a13a advisory-scan: paginate, and count by advisory DATE when versions can't be ordered
Answers 'how can the weekly run produce counts like the hand count?' — by doing
exactly what the hand count did, deterministically. Two changes:

1. PAGINATION. The scanner requested per_page=100 and stopped. This endpoint caps at
   100 AND ignores ?page= (it re-returns the same rows — which is how a manual count
   first produced exact triplicates and a bogus 300). Busy projects were silently
   truncated: discourse has 286 advisories, so a single page could not see the window
   at all. Now follows the Link rel=next cursor to exhaustion.

2. DATE-BASED FALLBACK. Version strings cannot be ordered across a scheme change
   (discourse semver 3.5.3 -> calver 2026.7.1), which is why the scan first reported a
   false 133, then correctly refused. Release DATES always order. When the version path
   refuses, the scan now resolves both versions to their git tag dates on the primary
   repo and counts advisories PUBLISHED in that window, labelling the method in the
   output. The version path is still preferred when usable — it is exact rather than
   temporal.

Verified: discourse 3.5.3 -> 2026.7.1 now reports 123, matching the hand count
(1 critical, 16 high, 91 medium, 16 low; window 2025-12-30 -> 2026-07-31); gitea
1.27.0 -> 1.27.1 still reports 2 via the version path.
2026-08-10 21:02:39 +00:00

490 lines
25 KiB
Python
Executable File

#!/usr/bin/env python3
"""Deterministic per-recipe CVE/advisory scan — an ADDITIVE pre-step for /recipe-upgrade.
WHY THIS EXISTS (2026-08-10): gitea 1.27.1 fixed two CVSS-9.8 RCEs (CVE-2026-60004,
CVE-2026-59774). Our weekly report showed gitea's CVE count as "1", then "none". The upgrade
subagent had scanned the GitHub *release notes*, which mention neither; the two CVEs were announced
only in the vendor's blog security section. The report generator then derived security content from
those notes plus model knowledge — and the model's training predates the CVEs. Nothing in the
pipeline ever queried an advisory source, so a critical CVE that is newer than the model and absent
from the changelog was invisible by construction.
WHAT IT DOES NOT DO: it does not replace or alter any existing security analysis. It is a strictly
ADDITIONAL evidence source whose findings are unioned into the CVE count.
SOURCES (measured against the gitea case before being chosen):
1. GitHub Security Advisories API — repos/<owner>/<repo>/security-advisories. PRIMARY: carries
CVE id, GHSA id, severity AND vulnerable/patched version ranges, so "fixed by THIS upgrade" is
computable rather than guessed. Found both gitea CVEs. Derived from the source-repo URLs the
per-recipe registry already records — no new per-recipe config needed.
2. Vendor release/security pages — every URL in cc-ci-plan/upstream/<recipe>.md, fetched and
regex-scanned for CVE ids. This is what would have caught gitea: the vendor blog names both,
while the GitHub releases page names neither. Add vendor security/announcement URLs to the
registry to widen this.
3. OSV.dev — supplementary, best-effort, only when the recipe declares an ecosystem/package
mapping below. NOTE: for gitea, OSV returned only Go *dependency* advisories and 404'd on both
application CVEs; NVD's API had them neither by CPE, CVE id, nor keyword. Advisory databases
lag the vendor — which is exactly why (1) and (2) lead.
Every source reports its own status, so "checked, none found" is never confused with "not checked".
Usage:
advisory-scan.py <recipe> [--from <version>] [--to <version>] [--json] [--registry DIR]
--from/--to are the app versions being upgraded between (e.g. 1.26.2 -> 1.27.1). When given, each
advisory is classified fixed-by-this-upgrade / still-open / older. Without them everything known
is listed unclassified. Exits 0 even when sources fail (informational; failures are reported).
"""
from __future__ import annotations
import argparse
import gzip
import json
import os
import re
import sys
import urllib.error
import urllib.request
REGISTRY_DIR = os.environ.get("CCCI_UPSTREAM_REGISTRY", "/srv/cc-ci/cc-ci-plan/upstream")
UA = "cc-ci-advisory-scan (+https://git.autonomic.zone/recipe-maintainers/cc-ci)"
TIMEOUT = int(os.environ.get("ADVISORY_SCAN_TIMEOUT", "45"))
CVE_RE = re.compile(r"CVE-\d{4}-\d{4,7}")
# Optional OSV mappings: recipe -> (ecosystem, package). Supplementary only (see module docstring).
OSV_PACKAGES: dict[str, tuple[str, str]] = {
"gitea": ("Go", "code.gitea.io/gitea"),
"n8n": ("npm", "n8n"),
}
def _github_token() -> str | None:
"""Read-only GitHub token, for the API rate limit ONLY (60/hr anonymous → 5000/hr with a token).
Env `GITHUB_TOKEN` wins; otherwise the file at `GITHUB_TOKEN_FILE` (default
/srv/cc-ci/.github-token, chmod 600, never in git). Reading PUBLIC security advisories needs NO
scopes at all — create a classic PAT with every box unticked, or a fine-grained token limited to
"Public repositories: read". Do NOT grant repo/write scopes: this tool only ever GETs advisories.
A missing token is not an error — the scan simply runs anonymously and will report sources as
failed once the 60/hr limit bites, which is visible rather than silent.
"""
tok = os.environ.get("GITHUB_TOKEN")
if tok:
return tok.strip()
path = os.environ.get("GITHUB_TOKEN_FILE", "/srv/cc-ci/.github-token")
try:
with open(path) as f:
return f.read().strip() or None
except OSError:
return None
def _gh_paginate(url: str, hdrs: dict, max_pages: int = 20):
"""Yield every row from a GitHub list endpoint, following Link rel=\"next\" cursors."""
seen_keys = set()
for _ in range(max_pages):
req = urllib.request.Request(url, headers={"User-Agent": UA, **hdrs})
with urllib.request.urlopen(req, timeout=TIMEOUT) as r:
rows = json.load(r)
link = r.headers.get("Link", "") or ""
fresh = 0
for a in rows:
k = a.get("ghsa_id") or json.dumps(a, sort_keys=True)[:120]
if k not in seen_keys:
seen_keys.add(k); fresh += 1
yield a
nxt = None
for part in link.split(","):
if 'rel="next"' in part:
nxt = part.split(";")[0].strip().strip("<>")
if not nxt or fresh == 0:
return
url = nxt
def _tag_date(owner: str, repo: str, version: str | None) -> str | None:
"""Publish date of a release tag, for DATE-BASED classification (see classify_by_date).
Version strings cannot be ordered across a scheme change (semver → calver), but tag dates
always can. Tries the common tag spellings; returns an ISO timestamp or None."""
if not version:
return None
hdrs = {"Accept": "application/vnd.github+json"}
tok = _github_token()
if tok:
hdrs["Authorization"] = f"Bearer {tok}"
for tag in (f"v{version}", version):
try:
ref = json.loads(_fetch(f"https://api.github.com/repos/{owner}/{repo}/git/ref/tags/{tag}", hdrs))
obj = ref.get("object", {})
sha, typ = obj.get("sha"), obj.get("type")
if typ == "tag":
t = json.loads(_fetch(f"https://api.github.com/repos/{owner}/{repo}/git/tags/{sha}", hdrs))
if t.get("tagger", {}).get("date"):
return t["tagger"]["date"]
sha = t.get("object", {}).get("sha")
c = json.loads(_fetch(f"https://api.github.com/repos/{owner}/{repo}/commits/{sha}", hdrs))
return c["commit"]["committer"]["date"]
except Exception: # noqa: BLE001 — try the next spelling
continue
return None
def _fetch(url: str, headers: dict | None = None) -> str:
h = {"User-Agent": UA, "Accept-Encoding": "gzip"}
h.update(headers or {})
with urllib.request.urlopen(urllib.request.Request(url, headers=h), timeout=TIMEOUT) as r:
raw = r.read()
if r.headers.get("Content-Encoding") == "gzip":
raw = gzip.decompress(raw)
return raw.decode(errors="replace")
def _vkey(v: str | None) -> tuple:
"""Loose version ordering key: leading integers of each dot-part ('1.27.1-rootless' -> (1,27,1))."""
if not v:
return ()
v = v.strip().lstrip("vV").split("+")[0]
out = []
for part in re.split(r"[.\-_]", v):
m = re.match(r"^\d+", part)
if m:
out.append(int(m.group()))
elif out:
break
return tuple(out)
def registry_urls(recipe: str, registry_dir: str) -> tuple[list[str], str | None]:
path = os.path.join(registry_dir, f"{recipe}.md")
try:
with open(path) as f:
text = f.read()
except OSError:
return [], None
urls = []
for u in re.findall(r"https?://[^\s)|\]]+", text):
# The registry is MARKDOWN: urls appear inside `backticks`, 'quotes', **bold**, and at the
# end of sentences. Trailing punctuation captured into the url makes the fetch 404 and the
# recipe render '?' for no real reason — that is what put immich and n8n in the unknown
# column on 2026-08-07 (https://docs.n8n.io/release-notes/` ← note the backtick).
u = u.rstrip("`'\"*.,;:>)")
if u and u not in urls:
urls.append(u)
return urls, path
def github_advisories(urls: list[str]) -> list[dict]:
"""Query GitHub Security Advisories for every github.com/<owner>/<repo> in the registry."""
seen, results = 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))
api = f"https://api.github.com/repos/{owner}/{repo}/security-advisories?per_page=100"
hdrs = {"Accept": "application/vnd.github+json"}
tok = _github_token()
if tok:
hdrs["Authorization"] = f"Bearer {tok}"
entry = {"source": f"github-advisories:{owner}/{repo}", "status": "ok", "advisories": []}
try:
# PAGINATE. This endpoint caps at 100 per response and IGNORES ?page= — it returns the
# same rows again, which silently truncates busy projects (discourse has 286; a hand
# count on 2026-08-10 found 123 CVEs in one upgrade window that a single page missed).
# Follow the Link rel="next" cursor to exhaustion instead.
for a in _gh_paginate(api, hdrs):
# An advisory carries ONE ENTRY PER PATCHED RELEASE LINE. n8n patches three
# (1.123.32, 2.17.4, 2.18.1); reading only vulnerabilities[0] silently dropped the
# line our deployment is actually on, so CVE-2026-42231/42232 classified as
# out-of-window. Keep them ALL and let the classifier match any of them.
vulns = a.get("vulnerabilities") or []
entry["advisories"].append(
{
"cve": a.get("cve_id"),
"ghsa": a.get("ghsa_id"),
"severity": a.get("severity"),
"summary": (a.get("summary") or "")[:200],
"vulnerable_range": "; ".join(
filter(None, (v.get("vulnerable_version_range") for v in vulns))
) or None,
"patched": "; ".join(
filter(None, (v.get("patched_versions") for v in vulns))
) or None,
"url": a.get("html_url"),
"published_at": a.get("published_at"),
}
)
except urllib.error.HTTPError as e:
# 404 = this repo simply publishes no security advisories (e.g. sidecar images like
# pgautoupgrade). That is a BENIGN ABSENCE, not a failed check — conflating the two
# would push nearly every recipe to "unknown" and make the ? signal meaningless again.
entry["status"] = "no-advisories-published" if e.code == 404 else f"error: HTTP {e.code}"
except Exception as e: # noqa: BLE001 — a genuinely dead source must be REPORTED, never silent
entry["status"] = f"error: {type(e).__name__}: {e}"
results.append(entry)
return results
def vendor_pages(urls: list[str]) -> list[dict]:
"""Fetch each registry URL and regex out CVE ids, with a little surrounding context."""
out = []
for u in urls:
if u.startswith("https://api.github.com"):
continue
if re.search(r"[<>{}]|\bVERSION\b|\bvX\.Y\.Z\b", u):
# Registry entries sometimes carry TEMPLATE urls for humans
# (…/changelog/v<VERSION>/). They are documentation, not fetchable — skipping them is
# correct; counting them as failures would wrongly mark the recipe's count unreliable.
out.append({"source": u, "status": "skipped: template URL (not fetchable)", "cves": [], "context": {}})
continue
entry = {"source": u, "status": "ok", "cves": [], "context": {}}
try:
text = _fetch(u)
plain = re.sub(r"<[^>]+>", " ", text)
for cve in sorted(set(CVE_RE.findall(plain))):
entry["cves"].append(cve)
i = plain.find(cve)
entry["context"][cve] = re.sub(r"\s+", " ", plain[max(0, i - 160) : i + 200]).strip()
except Exception as e: # noqa: BLE001
entry["status"] = f"error: {type(e).__name__}: {e}"
out.append(entry)
return out
def osv(recipe: str, version: str | None) -> dict | None:
pkg = OSV_PACKAGES.get(recipe)
if not pkg or not version:
return None
eco, name = pkg
entry = {"source": f"osv:{eco}/{name}@{version}", "status": "ok", "cves": []}
try:
body = json.dumps({"package": {"name": name, "ecosystem": eco}, "version": version}).encode()
req = urllib.request.Request(
"https://api.osv.dev/v1/query", data=body,
headers={"Content-Type": "application/json", "User-Agent": UA}, method="POST",
)
with urllib.request.urlopen(req, timeout=TIMEOUT) as r:
data = json.load(r)
ids = set()
for v in data.get("vulns", []):
for a in [v.get("id")] + (v.get("aliases") or []):
if a and a.startswith("CVE"):
ids.add(a)
entry["cves"] = sorted(ids)
except Exception as e: # noqa: BLE001
entry["status"] = f"error: {type(e).__name__}: {e}"
return entry
def scan(recipe: str, v_from: str | None, v_to: str | None, registry_dir: str) -> dict:
urls, reg_path = registry_urls(recipe, registry_dir)
report: dict = {
"recipe": recipe,
"from": v_from,
"to": v_to,
"registry": reg_path,
"registry_urls": len(urls),
"sources": [],
"cves": {},
}
if reg_path is None:
report["sources"].append(
{"source": f"registry:{recipe}.md", "status": "error: registry file not found"}
)
def record(cve: str, src: str, **extra):
e = report["cves"].setdefault(cve, {"sources": [], "severity": None, "ghsa": None,
"vulnerable_range": None, "patched": None,
"context": None, "published_at": None})
if src not in e["sources"]:
e["sources"].append(src)
for k, v in extra.items():
if v and not e.get(k):
e[k] = v
for entry in github_advisories(urls):
report["sources"].append({"source": entry["source"], "status": entry["status"],
"found": len(entry.get("advisories", []))})
for a in entry.get("advisories", []):
if a.get("cve"):
record(a["cve"], entry["source"], severity=a.get("severity"), ghsa=a.get("ghsa"),
vulnerable_range=a.get("vulnerable_range"), patched=a.get("patched"),
context=a.get("summary"), published_at=a.get("published_at"))
for entry in vendor_pages(urls):
report["sources"].append({"source": entry["source"], "status": entry["status"],
"found": len(entry.get("cves", []))})
for cve in entry.get("cves", []):
record(cve, entry["source"], context=entry["context"].get(cve))
for version in filter(None, (v_from, v_to)):
o = osv(recipe, version)
if o:
report["sources"].append({"source": o["source"], "status": o["status"],
"found": len(o.get("cves", []))})
for cve in o.get("cves", []):
record(cve, o["source"])
# Classify against the upgrade window when we know it: an advisory is "fixed by this upgrade"
# when its patched version is newer than `from` and no newer than `to`.
#
# TWO HARD-WON CONSTRAINTS (2026-08-10, discourse reported a false 133):
# a) The window belongs to ONE image. Advisories from OTHER repos in the registry (redis,
# postgres, nginx sidecars) must NOT be judged by it — redis CVE-2021-21309, patched in
# redis 6.0.11, scored as "fixed" because 6.0.11 sits numerically inside discourse's
# 3.5.3 → 2026.7.1 window. Only the PRIMARY app repo is classified; every other source is
# reported as unclassified so a human/agent still sees it but it never inflates the count.
# b) A version-SCHEME change (semver → calver, 3.5.3 → 2026.7.1) makes numeric ordering
# meaningless: 2025.12.2 compares "newer" than 3.5.3 while shipping earlier. When the
# leading component jumps by more than SCHEME_JUMP we refuse to classify and say so,
# rather than emitting a confident wrong number.
kf, kt = _vkey(v_from), _vkey(v_to)
SCHEME_JUMP = 100
scheme_change = bool(kf and kt and abs(kt[0] - kf[0]) >= SCHEME_JUMP)
primary = None
if v_from or v_to:
# the app repo = first github source in the registry (registry lists the app service first)
primary = next((s["source"] for s in report["sources"]
if s["source"].startswith("github-advisories:")), None)
report["primary_source"] = primary
report["scheme_change"] = scheme_change
fixed, unknown = [], []
for cve, e in report["cves"].items():
# `patched_versions` is a RANGE EXPRESSION (">= 2.18.1"), not a bare version, and there may
# be several (one per patched release line, joined with ";"). Pull every version-looking
# token and treat the advisory as fixed-by-this-upgrade if ANY of them lands in (from, to].
cands = [_vkey(t) for t in re.findall(r"\d+(?:\.\d+)*", e.get("patched") or "")]
kp = next((c for c in cands if kf and kt and kf < c <= kt), None) or (cands[0] if cands else ())
from_primary = primary is not None and primary in e["sources"]
if scheme_change:
e["classification"] = "unclassified: version-scheme change, cannot order reliably"
unknown.append(cve)
elif not from_primary:
e["classification"] = "unclassified: different image than the given version window"
unknown.append(cve)
elif kf and kt and kp and kf < kp <= kt:
e["classification"] = "fixed-by-this-upgrade"
fixed.append(cve)
else:
e["classification"] = "unclassified" if not (kf and kt and kp) else "outside-window"
if e["classification"] == "unclassified":
unknown.append(cve)
# NEVER report 0 for something we could not determine. When classification was refused, the
# count is UNKNOWN (null) — a 0 would be read as "no CVEs", which is an assertion this scan
# cannot make. Consumers must distinguish "counted 0" from "could not count".
# DATE-BASED FALLBACK (phase datewin). Version strings cannot be ordered across a scheme change,
# but RELEASE DATES always can. Resolve both versions to their tag dates on the primary repo and
# count advisories PUBLISHED inside that window — the method a hand count used on 2026-08-10 to
# establish discourse 3.5.3 (2025-12-30) → 2026.7.1 (2026-07-31) = 123 CVEs, where version
# comparison had produced first a false 133 and then a refusal. Only used when the version path
# refuses; a successful version classification is always preferred (it is exact, not temporal).
date_window = None
if scheme_change and primary and primary.startswith("github-advisories:"):
owner_repo = primary.split(":", 1)[1]
owner, _, repo = owner_repo.partition("/")
d_from, d_to = _tag_date(owner, repo, v_from), _tag_date(owner, repo, v_to)
if d_from and d_to and d_from < d_to:
date_window = (d_from, d_to)
fixed, unknown = [], []
for cve, e in report["cves"].items():
pub = e.get("published_at")
if primary in e["sources"] and pub and d_from < pub <= d_to:
e["classification"] = "fixed-by-this-upgrade (by advisory publish date)"
fixed.append(cve)
else:
e["classification"] = ("outside-window (by date)" if primary in e["sources"]
else "unclassified: different image than the given window")
if primary not in e["sources"]:
unknown.append(cve)
scheme_change = False # resolved by date; a real count is available
report["classified_by"] = "advisory publish date (version scheme changed)"
report["date_window"] = {"from": d_from, "to": d_to}
if not date_window and not scheme_change:
report["classified_by"] = "patched version ranges"
report["fixed_by_this_upgrade"] = sorted(fixed)
report["unclassified"] = sorted(unknown)
report["count_known"] = not scheme_change
report["cve_count_fixed"] = len(fixed) if not scheme_change else None
report["cve_count_total_seen"] = len(report["cves"])
# Only GENUINE failures make a count unreliable. "no-advisories-published" (404: the repo has
# no advisory feed) and "skipped: template URL" are benign and must not degrade the verdict.
report["sources_failed"] = [
s["source"]
for s in report["sources"]
if not (s["status"] == "ok" or s["status"].startswith(("no-advisories-published", "skipped:")))
]
report["sources_benign"] = [
s["source"]
for s in report["sources"]
if s["status"].startswith(("no-advisories-published", "skipped:"))
]
return report
def markdown(rep: dict) -> str:
"""Human/agent-readable block for pasting into the per-recipe upgrade log."""
L = [f"### Advisory scan (deterministic pre-step) — {rep['recipe']} "
f"{rep.get('from') or '?'}{rep.get('to') or '?'}"]
if not rep.get("count_known", True):
L.append("\n**CVEs fixed by this upgrade: UNKNOWN — the scan could NOT determine a count.**")
L.append("\n⚠ This is NOT zero. A version-scheme change (e.g. semver → calver) makes numeric "
"ordering meaningless across this jump, so no advisory could be classified. Render "
"this recipe's cve cell as `?`, never `0`. Read the vendor's release notes for the "
"jump and count by hand.")
if rep["unclassified"]:
L.append(f"\nAdvisories seen but unclassifiable ({len(rep['unclassified'])}) — includes "
f"other images in this recipe: " + ", ".join(rep["unclassified"][:12]))
if rep["sources_failed"]:
L.append(f"\n⚠ sources that FAILED: {', '.join(rep['sources_failed'])}")
L.append(f"\n_Sources checked: {len(rep['sources'])} ({rep['registry_urls']} registry URLs + "
f"advisory APIs). This scan is ADDITIVE — it does not replace the release-note "
f"reading in the upgrade step._")
return "\n".join(L)
if rep["fixed_by_this_upgrade"]:
L.append(f"\n**CVEs fixed by this upgrade: {rep['cve_count_fixed']}**\n")
if rep.get("date_window"):
L.append(f"_Counted by advisory PUBLISH DATE ({rep['date_window']['from'][:10]} → "
f"{rep['date_window']['to'][:10]}) because the version scheme changed across this "
f"jump; version strings cannot be ordered across it._\n")
L.append("| CVE | severity | fixed in | advisory | source |")
L.append("|---|---|---|---|---|")
for cve in rep["fixed_by_this_upgrade"]:
e = rep["cves"][cve]
L.append(f"| {cve} | {e.get('severity') or '?'} | {e.get('patched') or '?'} | "
f"{e.get('ghsa') or '-'} | {e['sources'][0]} |")
else:
L.append("\n**CVEs fixed by this upgrade: 0 identified by the deterministic scan.**")
if rep["unclassified"]:
L.append(f"\nSeen but not version-classified ({len(rep['unclassified'])}) — includes advisories "
f"from OTHER images in this recipe (sidecars), which this window cannot judge: "
+ ", ".join(rep["unclassified"][:12]))
if rep["sources_failed"]:
L.append(f"\n⚠ sources that FAILED (treat counts as incomplete): {', '.join(rep['sources_failed'])}")
L.append(f"\n_Sources checked: {len(rep['sources'])} "
f"({rep['registry_urls']} registry URLs + advisory APIs). This scan is ADDITIVE — it does "
f"not replace the release-note reading in the upgrade step._")
return "\n".join(L)
def main() -> int:
ap = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter)
ap.add_argument("recipe")
ap.add_argument("--from", dest="v_from", default=None)
ap.add_argument("--to", dest="v_to", default=None)
ap.add_argument("--json", action="store_true", help="emit raw JSON instead of markdown")
ap.add_argument("--registry", default=REGISTRY_DIR)
a = ap.parse_args()
rep = scan(a.recipe, a.v_from, a.v_to, a.registry)
print(json.dumps(rep, indent=2) if a.json else markdown(rep))
return 0
if __name__ == "__main__":
sys.exit(main())