autonomic-bot
65bf3c095b
The image name was packed into the value, so the flag needed a hand-rolled KEY=FROM:TO parser with its own malformed-input branch, and 'window' named the wrong thing — the tool has two kinds of window (version ranges and, on the date fallback, real date windows) and the flag meant only the first. Now each part is its own argument: --image redis 7.4 8.10, repeatable, all in one call. argparse enforces the arity, so the string parsing and its error path are deleted. 'windows' survives internally as the computed-range concept. Counts unchanged: discourse 128 with redis / 123 without, gitea 2.
523 lines
26 KiB
Python
Executable File
523 lines
26 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}")
|
|
# Leading-version-component jump that means the scheme changed (semver → calver).
|
|
SCHEME_JUMP = 100
|
|
|
|
# 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,
|
|
images: list[tuple[str, str, str]] | None = None) -> 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.
|
|
# ── Classification ────────────────────────────────────────────────────────────────────────
|
|
# A recipe upgrades SEVERAL images (app + redis/postgres/nginx sidecars), each with its OWN
|
|
# version window. Judging every advisory by the app's window is how discourse once reported a
|
|
# false 133 (34 of them redis CVEs, incl. one patched in redis 6.0.11 in 2021). So each source
|
|
# is classified against ITS OWN window, and the count is the union across windows.
|
|
#
|
|
# --from/--to → the PRIMARY app repo (first github source in the registry)
|
|
# --image NAME FROM TO → any other source whose name contains NAME (repeatable),
|
|
# e.g. --image redis 7.4 8.10
|
|
#
|
|
# A source with no window is not classified: its advisories are listed as unclassified so they
|
|
# stay visible without inflating the count.
|
|
gh_sources = [x["source"] for x in report["sources"] if x["source"].startswith("github-advisories:")]
|
|
primary = gh_sources[0] if (gh_sources and (v_from or v_to)) else None
|
|
report["primary_source"] = primary
|
|
|
|
windows = {} # source name -> (from, to)
|
|
if primary:
|
|
windows[primary] = (v_from, v_to)
|
|
for key, wf, wt in (images or []):
|
|
for src in gh_sources:
|
|
if key.lower() in src.lower() and src not in windows:
|
|
windows[src] = (wf, wt)
|
|
report["windows"] = {k: {"from": f, "to": t} for k, (f, t) in windows.items()}
|
|
|
|
def _classify_window(src, wf, wt):
|
|
"""Return (set_of_fixed_cves, method, date_window|None, unresolved:boolean) for one source."""
|
|
kf, kt = _vkey(wf), _vkey(wt)
|
|
# A version-SCHEME change (semver 3.5.3 → calver 2026.7.1) makes numeric ordering
|
|
# meaningless: 2025.12.2 compares "newer" than 3.5.3 while shipping earlier.
|
|
scheme = bool(kf and kt and abs(kt[0] - kf[0]) >= SCHEME_JUMP)
|
|
if not scheme:
|
|
got = set()
|
|
for cve, e in report["cves"].items():
|
|
if src not in e["sources"]:
|
|
continue
|
|
cands = [_vkey(t) for t in re.findall(r"\d+(?:\.\d+)*", e.get("patched") or "")]
|
|
if kf and kt and any(kf < c <= kt for c in cands):
|
|
got.add(cve)
|
|
return got, "patched version ranges", None, False
|
|
# DATE FALLBACK: release DATES always order, even across a scheme change. Resolve both
|
|
# versions to git tag dates and count advisories PUBLISHED in that window — the method a
|
|
# hand count used to establish discourse 3.5.3 (2025-12-30) → 2026.7.1 (2026-07-31) = 123.
|
|
owner, _, repo = src.split(":", 1)[1].partition("/")
|
|
d_from, d_to = _tag_date(owner, repo, wf), _tag_date(owner, repo, wt)
|
|
if d_from and d_to and d_from < d_to:
|
|
got = {cve for cve, e in report["cves"].items()
|
|
if src in e["sources"] and e.get("published_at")
|
|
and d_from < e["published_at"] <= d_to}
|
|
return got, "advisory publish date (version scheme changed)", (d_from, d_to), False
|
|
return set(), "unresolved", None, True
|
|
|
|
fixed_set, methods, date_windows, unresolved_any = set(), {}, {}, False
|
|
for src, (wf, wt) in windows.items():
|
|
got, method, dw, unresolved = _classify_window(src, wf, wt)
|
|
methods[src] = method
|
|
if dw:
|
|
date_windows[src] = {"from": dw[0], "to": dw[1]}
|
|
if unresolved:
|
|
unresolved_any = True
|
|
for cve in got:
|
|
report["cves"][cve]["classification"] = f"fixed-by-this-upgrade ({method}) via {src}"
|
|
fixed_set.add(cve)
|
|
|
|
unknown = []
|
|
for cve, e in report["cves"].items():
|
|
if cve in fixed_set:
|
|
continue
|
|
if not any(src in e["sources"] for src in windows):
|
|
e["classification"] = "unclassified: no versions given for this image"
|
|
unknown.append(cve)
|
|
else:
|
|
e.setdefault("classification", "outside-window")
|
|
if e["classification"] == "outside-window":
|
|
pass
|
|
else:
|
|
unknown.append(cve)
|
|
|
|
report["classified_by"] = methods
|
|
if date_windows:
|
|
report["date_window"] = date_windows
|
|
report["fixed_by_this_upgrade"] = sorted(fixed_set)
|
|
report["unclassified"] = sorted(unknown)
|
|
# NEVER report 0 for something we could not determine — a 0 asserts safety. If ANY requested
|
|
# window could not be ordered at all, the total is UNKNOWN rather than a partial number.
|
|
report["count_known"] = not unresolved_any
|
|
report["cve_count_fixed"] = len(fixed_set) if not unresolved_any 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")
|
|
cb = rep.get("classified_by") or {}
|
|
if isinstance(cb, dict) and cb:
|
|
for src, method in cb.items():
|
|
dw = (rep.get("date_window") or {}).get(src)
|
|
win = (rep.get("windows") or {}).get(src, {})
|
|
span = f"{win.get('from')} → {win.get('to')}"
|
|
extra = (f" (dates {dw['from'][:10]} → {dw['to'][:10]})" if dw else "")
|
|
L.append(f"_{src.split(':',1)[-1]}: {span} — counted by {method}{extra}._")
|
|
L.append("")
|
|
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)
|
|
ap.add_argument("--image", action="append", default=[], nargs=3,
|
|
metavar=("NAME", "FROM", "TO"),
|
|
help="a sidecar image and the versions it moved between, e.g. "
|
|
"--image redis 7.4 8.10 (repeatable). NAME matches a source repo name; "
|
|
"its advisories are then counted against ITS OWN versions instead of "
|
|
"being left unclassified.")
|
|
a = ap.parse_args()
|
|
rep = scan(a.recipe, a.v_from, a.v_to, a.registry, [tuple(x) for x in a.image])
|
|
print(json.dumps(rep, indent=2) if a.json else markdown(rep))
|
|
return 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
sys.exit(main())
|