recipe-maintainers/cc-ci-orchestrator
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Improve recipe advisory coverage and report non-CVE fixes

Browse Source
This commit is contained in:
notplants
2026-11-02 19:40:46 +00:00
committed by notplants
parent 508d6bb0de
commit bd7196bf9e
7 changed files with 238 additions and 42 deletions
Download Patch File Download Diff File Expand all files Collapse all files
cc-ci-plan/advisory-scan.py
+152 -16
View File
@@ -40,6 +40,7 @@ from __future__ import annotations
import argparse
import gzip
import ipaddress
import json
import os
import re
@@ -166,6 +167,43 @@ def _vkey(v: str | None) -> tuple:
return tuple(out)
def _nvd_affects(version: str | None, ranges: list[dict]) -> bool | None:
"""Whether NVD marks a concrete image version affected; None means tag lacks precision."""
current = _vkey(version)
if not current:
return None
uncertain = False
for row in ranges:
exact = _vkey(row.get("version")) if row.get("version") not in (None, "*") else ()
bounds = [(name, _vkey(row.get(name))) for name in (
"versionStartIncluding", "versionStartExcluding",
"versionEndIncluding", "versionEndExcluding") if row.get(name)]
versions = ([exact] if exact else []) + [value for _, value in bounds]
if any(not value for value in versions):
uncertain = True
continue
if any(len(value) > len(current) and value[:len(current)] == current for value in versions):
uncertain = True # e.g. postgres:13-alpine does not say which 13.x patch is running
continue
def compare(other):
length = max(len(current), len(other))
return (current + (0,) * (length - len(current))) > (other + (0,) * (length - len(other)))
if exact and current != exact:
continue
def outside(name, value):
if name == "versionStartExcluding":
return not compare(value)
if name == "versionStartIncluding":
return not compare(value) and current != value
if name == "versionEndIncluding":
return compare(value)
return compare(value) or current == value # versionEndExcluding
if any(outside(name, value) for name, value in bounds):
continue
return True
return None if uncertain else False
def _already_fixed_on_from_line(kf: tuple, cands: list[tuple]) -> bool:
"""Was it ALREADY fixed on the line we are upgrading FROM?
@@ -233,6 +271,14 @@ def registry_urls(recipe: str, registry_dir: str) -> tuple[list[str], str | None
u = u.rstrip("`'\"*.,;:>)")
if "…" in u or "..." in u:
continue # prose placeholder, not a fetchable security source
host = urllib.parse.urlsplit(u).hostname
if not host or host == "localhost" or "." not in host:
continue # a container/loopback URL in operational notes, not a public advisory source
try:
if not ipaddress.ip_address(host).is_global:
continue
except ValueError:
pass
if u and u not in urls:
urls.append(u)
return urls, path
@@ -371,15 +417,57 @@ def vendor_pages(urls: list[str]) -> list[dict]:
NVD_API = "https://services.nvd.nist.gov/rest/json/cves/2.0"
NVD_CPE_RE = re.compile(r"^\s*[-*]?\s*nvd-cpe:\s*(\S+)\s*=\s*(cpe:2\.3:[^\s`]+)", re.M | re.I)
_NVD_CACHE: dict[str, dict] = {}
COMMON_IMAGE_CPES = {
"postgres": "cpe:2.3:a:postgresql:postgresql:*:*:*:*:*:*:*:*",
"mariadb": "cpe:2.3:a:mariadb:mariadb:*:*:*:*:*:*:*:*",
"mysql": "cpe:2.3:a:oracle:mysql:*:*:*:*:*:*:*:*",
}
def registry_cpes(recipe: str, registry_dir: str) -> list[tuple[str, str]]:
"""[(image-key, cpe)] declared in the recipe's registry as `nvd-cpe: <key> = <cpe>`."""
"""Explicit CPEs plus common database CPEs for images listed in the registry table."""
path = os.path.join(registry_dir, f"{recipe}.md")
try:
return [(m.group(1), m.group(2)) for m in NVD_CPE_RE.finditer(open(path).read())]
text = open(path).read()
except OSError:
return []
out = [(m.group(1), m.group(2)) for m in NVD_CPE_RE.finditer(text)]
explicit = {key for key, _ in out}
for line in text.splitlines():
if not line.startswith("|"):
continue
cells = [cell.strip().strip("`") for cell in line.strip().strip("|").split("|")]
if len(cells) < 2:
continue
image = cells[1].split("/")[-1].lower()
if image in COMMON_IMAGE_CPES and image not in explicit:
out.append((image, COMMON_IMAGE_CPES[image]))
explicit.add(image)
return out
def registry_image_sources(recipe: str, registry_dir: str) -> dict[str, set[str]]:
"""Map image repo basenames to GitHub advisory sources from the registry table."""
path = os.path.join(registry_dir, f"{recipe}.md")
try:
lines = open(path).read().splitlines()
except OSError:
return {}
out: dict[str, set[str]] = {}
for line in lines:
if not line.startswith("|"):
continue
cells = [cell.strip().strip("`") for cell in line.strip().strip("|").split("|")]
if len(cells) < 3:
continue
image = cells[1].split("/")[-1].lower()
match = re.search(r"https?://github\.com/([^/\s|]+)/([^/\s|)]+)", cells[2])
if not match or image in ("image", "---", ""):
continue
source = f"github-advisories:{match.group(1)}/{match.group(2).removesuffix('.git')}"
out.setdefault(image, set()).add(source)
return out
def nvd_advisories(cpe: str, key: str) -> dict:
@@ -397,6 +485,8 @@ def nvd_advisories(cpe: str, key: str) -> dict:
NVD LAGS the vendor (it had neither gitea CVSS-9.8 RCE at publication), so this is a fallback,
never a replacement for 2a/2b. Unauthenticated calls are rate-limited to ~5/30s, hence the retry.
"""
if cpe in _NVD_CACHE:
return {**_NVD_CACHE[cpe], "source": f"nvd:{key}"}
entry = {"source": f"nvd:{key}", "status": "ok", "advisories": []}
url = f"{NVD_API}?resultsPerPage=2000&virtualMatchString={urllib.parse.quote(cpe)}"
data = None
@@ -409,15 +499,31 @@ def nvd_advisories(cpe: str, key: str) -> dict:
entry["status"] = f"error: {type(e).__name__}"
return entry
time.sleep(8)
for v in (data or {}).get("vulnerabilities", []):
vulnerabilities = (data or {}).get("vulnerabilities", [])
if (data or {}).get("totalResults", len(vulnerabilities)) > len(vulnerabilities):
entry["status"] = "error: NVD result set truncated"
return entry
for v in vulnerabilities:
c = v.get("cve") or {}
cid = c.get("id")
if not cid:
if not cid or c.get("vulnStatus") == "Rejected":
continue
fixed, affected_max = set(), set()
fixed, affected_max, affected_ranges = set(), set(), []
cpe_product = ":".join(cpe.split(":")[:5]) + ":"
cpe_edition = cpe.split(":")[9] if len(cpe.split(":")) > 9 else "*"
for cfg in c.get("configurations", []):
for node in cfg.get("nodes", []):
for m in node.get("cpeMatch", []):
criteria = m.get("criteria") or ""
if not m.get("vulnerable") or not criteria.startswith(cpe_product):
continue
edition = criteria.split(":")[9] if len(criteria.split(":")) > 9 else "*"
if cpe_edition != "*" and edition not in (cpe_edition, "*"):
continue
affected_ranges.append({k: m.get(k) for k in (
"versionStartIncluding", "versionStartExcluding",
"versionEndIncluding", "versionEndExcluding") if m.get(k)})
affected_ranges[-1]["version"] = criteria.split(":")[5]
if m.get("versionEndExcluding"):
fixed.add(m["versionEndExcluding"])
elif m.get("versionEndIncluding"):
@@ -435,9 +541,11 @@ def nvd_advisories(cpe: str, key: str) -> dict:
"vulnerable_range": None,
"patched": "; ".join(sorted(fixed)) or None,
"affected_max": "; ".join(sorted(affected_max)) or None,
"affected_ranges": affected_ranges or None,
"url": f"https://nvd.nist.gov/vuln/detail/{cid}",
"published_at": c.get("published"), "description": None, "cvss": None,
})
_NVD_CACHE[cpe] = entry
return entry
@@ -739,7 +847,8 @@ def scan(recipe: str, v_from: str | None, v_to: str | None, registry_dir: str,
"vulnerable_range": None, "patched": None,
"context": None, "published_at": None,
"description": None, "url": None, "cvss": None,
"changelog_fixed_in": None, "affected_max": None})
"changelog_fixed_in": None, "affected_max": None,
"affected_ranges": None})
if src not in e["sources"]:
e["sources"].append(src)
for k, v in extra.items():
@@ -750,8 +859,9 @@ def scan(recipe: str, v_from: str | None, v_to: str | None, registry_dir: str,
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"),
advisory_id = a.get("cve") or a.get("ghsa")
if advisory_id:
record(advisory_id, 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"),
description=a.get("description"), url=a.get("url"), cvss=a.get("cvss"))
@@ -763,6 +873,7 @@ def scan(recipe: str, v_from: str | None, v_to: str | None, registry_dir: str,
for a in entry.get("advisories", []):
record(a["cve"], entry["source"], severity=a.get("severity"),
patched=a.get("patched"), affected_max=a.get("affected_max"),
affected_ranges=a.get("affected_ranges"),
context=a.get("summary"), published_at=a.get("published_at"),
url=a.get("url"))
@@ -827,17 +938,20 @@ def scan(recipe: str, v_from: str | None, v_to: str | None, registry_dir: str,
if pname in k or k in pname:
windows[src] = (v_from, v_to)
window_key[src] = k
image_sources = registry_image_sources(recipe, registry_dir)
for key, wf, wt in (images or []):
matched = False
explicit = image_sources.get(key.lower(), set())
for src in gh_sources:
if src in windows:
continue
# Match BOTH ways: an image name is often longer than its source repo
# (`clickhouse/clickhouse-server` vs source `ClickHouse/ClickHouse`) and sometimes
# shorter (`redis` vs `redis/redis`). One-directional matching silently dropped the
# clickhouse window when the key was derived from a compose file.
k, name = key.lower(), src.split("/")[-1].lower()
if k in src.lower() or name in k:
if src in explicit or k in src.lower() or name in k:
if src in windows:
matched |= windows[src] == (wf, wt)
continue # two images in one monorepo share a source and version window
windows[src] = (wf, wt)
window_key[src] = key
matched = True
@@ -850,6 +964,21 @@ def scan(recipe: str, v_from: str | None, v_to: str | None, registry_dir: str,
`indeterminate` = advisories from this source that the method COULD NOT JUDGE (no usable
patched version, or no publish date). They are neither counted nor dismissed."""
if src.startswith("nvd:"):
got, undecidable = set(), set()
for advisory, evidence in report["cves"].items():
if src not in evidence["sources"]:
continue
if not evidence.get("affected_ranges"):
undecidable.add(advisory)
continue
old = _nvd_affects(wf, evidence["affected_ranges"])
new = _nvd_affects(wt, evidence["affected_ranges"])
if old is True and new is False:
got.add(advisory)
elif old is None or (old is True and new is None):
undecidable.add(advisory)
return got, "NVD affected-version ranges", None, False, undecidable
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.
@@ -1018,6 +1147,7 @@ def scan(recipe: str, v_from: str | None, v_to: str | None, registry_dir: str,
report["cves"][cve]["classification"] = "indeterminate: no fix version published"
unknown = [c for c in unknown if c not in fixed_set]
report["fixed_by_this_upgrade"] = sorted(fixed_set)
report["fixed_non_cve_advisories"] = sorted(x for x in fixed_set if not x.startswith("CVE-"))
report["indeterminate"] = sorted(indeterminate)
report["cve_count_indeterminate"] = len(indeterminate)
report["unclassified"] = sorted(unknown)
@@ -1037,9 +1167,9 @@ def scan(recipe: str, v_from: str | None, v_to: str | None, registry_dir: str,
]
coverage_failed = bool(report.get("unmatched_image_windows")) or bool(v_from and not primary)
report["count_known"] = not unresolved_any and not no_sources and not coverage_failed and not report["sources_failed"]
report["cve_count_fixed"] = (len(fixed_set)
report["cve_count_fixed"] = (sum(x.startswith("CVE-") for x in fixed_set)
if report["count_known"] else None)
report["cve_count_total_seen"] = len(report["cves"])
report["cve_count_total_seen"] = sum(x.startswith("CVE-") for x in 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_benign"] = [
@@ -1057,7 +1187,7 @@ def markdown(rep: dict) -> str:
if not rep.get("count_known", True):
L.append("\n**CVEs fixed by this upgrade: UNKNOWN — the scan could NOT determine a count.**")
if rep.get("fixed_by_this_upgrade"):
L.append(f"\nAt least {len(rep['fixed_by_this_upgrade'])} fixed CVE(s) were identified: "
L.append(f"\nAt least {len(rep['fixed_by_this_upgrade'])} fixed advisory/advisories were identified: "
+ ", ".join(rep["fixed_by_this_upgrade"][:15]))
if rep.get("no_usable_sources"):
L.append("\n⚠ This is NOT zero. **No usable source was checked at all** — the registry "
@@ -1080,7 +1210,8 @@ def markdown(rep: dict) -> str:
f"reading in the upgrade step._")
return "\n".join(L)
ind = rep.get("cve_count_indeterminate") or 0
if rep["fixed_by_this_upgrade"]:
fixed_cves = [x for x in rep["fixed_by_this_upgrade"] if x.startswith("CVE-")]
if fixed_cves:
floor = " (at least — see undetermined below)" if ind else ""
L.append(f"\n**CVEs fixed by this upgrade: {rep['cve_count_fixed']}**{floor}\n")
cb = rep.get("classified_by") or {}
@@ -1094,12 +1225,17 @@ def markdown(rep: dict) -> str:
L.append("")
L.append("| CVE | severity | fixed in | advisory | source |")
L.append("|---|---|---|---|---|")
for cve in rep["fixed_by_this_upgrade"]:
for cve in fixed_cves:
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.get("fixed_non_cve_advisories"):
L.append("\n**Security advisories fixed without CVE IDs: "
+ str(len(rep["fixed_non_cve_advisories"])) + "** — "
+ ", ".join(rep["fixed_non_cve_advisories"][:15])
+ ". In the report CVEs column, use YES when the CVE count is zero.")
if ind:
L.append(f"\n⚠ **{ind} advisory/advisories could NOT be judged** — the vendor published no fix "
f"version (GitHub carries `TBD` or a placeholder like `7.4.X`) and the vulnerable "