From bd7196bf9e9d12dfcbd5b8c68e0877fbb585fc12 Mon Sep 17 00:00:00 2001 From: notplants <@notplants> Date: Mon, 2 Nov 2026 19:40:46 +0000 Subject: [PATCH] Improve recipe advisory coverage and report non-CVE fixes --- .claude/skills/cve-check/SKILL.md | 9 +- .claude/skills/recipe-report/SKILL.md | 3 + cc-ci-plan/advisory-scan.py | 168 ++++++++++++++++++++++--- cc-ci-plan/audit-sources.py | 23 ++-- cc-ci-plan/recipe-report.py | 4 +- cc-ci-plan/test-advisory-scan.py | 65 +++++++++- cc-ci-plan/upstream/rustdesk-server.md | 8 +- 7 files changed, 238 insertions(+), 42 deletions(-) diff --git a/.claude/skills/cve-check/SKILL.md b/.claude/skills/cve-check/SKILL.md index c6d13e5..185984d 100644 --- a/.claude/skills/cve-check/SKILL.md +++ b/.claude/skills/cve-check/SKILL.md @@ -113,9 +113,9 @@ A recipe whose sources yield **no CVE data at all** cannot produce a meaningful measured, the same way a missing registry file cannot. Render those as **`?`**, not `0`. Do not assume that the fleet has zero blind recipes. Run the audit every sweep and inspect source -statuses; an HTTP error or rate limit is not evidence of no CVEs. `rustdesk-server` is currently -blind because its upstream GitHub repo publishes no security advisories and no other usable source -has been verified; mark it `?` until that changes. +statuses; an HTTP error or rate limit is not evidence of no CVEs. `rustdesk-server` has an NVD CPE +scoped to the OSS server edition; a successful zero-result query is a checked zero for that source, +not a reason to count Client or Server Pro vulnerabilities against it. Check `test -s /srv/cc-ci/.github-token` before a fleet scan. Without that read-only token, GitHub's anonymous API limit can be exhausted partway through the fleet; HTTP 403 results are @@ -178,7 +178,8 @@ Severity outranks raw count: 2 CVSS-9.8 RCEs matter more than 120 medium plugin - `table[]` — every recipe swept. `recipe`; `change` = the window you scanned, e.g. `1.27.0 → 1.27.1 · redis 7.4 → 8.10`; `status` = `UPTODATE` when nothing is available, else `PENDING` (an upgrade exists and is not yet taken); **`cve`** = the count (integer, `?` only per the - rules below); `notes` = severity mix, whether the number is a floor, and `maintained elsewhere` for + rules below, or `"YES"` when a confirmed security advisory has no CVE ID and no numbered CVEs + were found); `notes` = the advisory ID and severity, whether the number is a floor, and `maintained elsewhere` for `external` rows. **Leave `ci`/`pr` empty — nothing was built and no PR exists.** - `addendum[]` — real anomalies only: registry URLs that failed, recipes whose window could not be established, a scan whose count is a floor with many undetermined advisories. diff --git a/.claude/skills/recipe-report/SKILL.md b/.claude/skills/recipe-report/SKILL.md index cf3c1cf..d0d4242 100644 --- a/.claude/skills/recipe-report/SKILL.md +++ b/.claude/skills/recipe-report/SKILL.md @@ -52,6 +52,9 @@ keeps every weekly edition looking the same regardless of which model writes the A zero for an up-to-date recipe means there is no upgrade window to credit; it does not certify that the running images are free of vulnerabilities. If an upgrade exists but its advisory scan is absent, a source failed, or an image window was unmatched, use `?`. + If a confirmed security advisory is fixed but has no CVE ID, put `"YES"` in `cve` rather + than `0`; describe its advisory ID and severity in Notes. If CVE IDs are also fixed, show + the numeric CVE count and mention the additional non-CVE advisories in Notes. - **`?` means the count is not verified.** Use it when an upgrade has no scan, a source failed, an image window was unmatched, or the scan block says COUNT UNKNOWN. In that case publish `?` and say why in diff --git a/cc-ci-plan/advisory-scan.py b/cc-ci-plan/advisory-scan.py index 5187aba..c69afba 100755 --- a/cc-ci-plan/advisory-scan.py +++ b/cc-ci-plan/advisory-scan.py @@ -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: = `.""" + """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 " diff --git a/cc-ci-plan/audit-sources.py b/cc-ci-plan/audit-sources.py index e1a99e1..4b8afd3 100755 --- a/cc-ci-plan/audit-sources.py +++ b/cc-ci-plan/audit-sources.py @@ -134,7 +134,7 @@ def security_source_audit(recipe: str) -> list[dict]: 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", + "kind": "advisory-feed", "verified_empty": n == 0, "status": e["status"], "cves": n, "usable": n}) for entry in A.github_advisories(urls): out.append({"source": entry["source"], "kind": "advisory-feed", @@ -227,18 +227,7 @@ def audit_recipe(recipe: str, ssh: str | None, quiet_days: int) -> dict: 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 "…" in u or "..." in u: - continue - if u and u not in urls: - urls.append(u) - return urls, path + return A.registry_urls(recipe, REGISTRY_DIR) def all_recipes() -> list[str]: @@ -274,7 +263,8 @@ def main() -> int: 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 + feeds = [x for x in rows if x["kind"] == "advisory-feed" + and (x["cves"] > 0 or x.get("verified_empty")) and x["status"] == "ok"] logs = [x for x in rows if x["kind"] == "changelog" and x["status"] == "ok"] unattr = [x for x in rows if x["kind"] == "unattributable"] @@ -286,13 +276,16 @@ def main() -> int: print(f"!! {r}: NO VERIFIED CVE SOURCE — {len(unattr)} unparseable page(s), " f"0 populated advisory feeds, 0 attributable changelogs, {len(failed)} failed source(s)") for x in rows: - print(f" {x['kind']:15} {x['source'][:64]} ({x['cves']} CVEs; {x['status']})") + unit = "advisories" if x["kind"] == "advisory-feed" else "CVE IDs" + print(f" {x['kind']:15} {x['source'][:64]} ({x['cves']} {unit}; {x['status']})") else: print(f"OK {r}: {len(feeds)} advisory-feed(s), {len(logs)} changelog(s)" + (f", {len(unattr)} unparseable page(s)" if unattr else "") + (f", {len(failed)} failed source(s) — counts incomplete" if failed else "")) for x in logs: print(f" changelog {x['source'][:62]} ({x['usable']}/{x['cves']})") + for x in failed: + print(f" FAILED {x['source'][:90]} ({x['status']})") 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)") diff --git a/cc-ci-plan/recipe-report.py b/cc-ci-plan/recipe-report.py index f114c04..8075531 100755 --- a/cc-ci-plan/recipe-report.py +++ b/cc-ci-plan/recipe-report.py @@ -30,7 +30,7 @@ SPEC SHAPE (the agent writes this JSON): {"date":"YYYY-MM-DD","subtitle":"Week of ", "lead":"", "table":[{"recipe":"x","change":"a → b","status":"GREEN|FAILED|STALE|SKIPPED|UPTODATE", # -> TESTS col - "cve":16, # number of CVEs this PR fixes; 0/omit for none + "cve":16, # CVE count; "YES" for advisory without CVE ID; "?" unknown "ci":"build 154 ✓","ci_url":"…","pr":"#4","pr_url":"…","notes":"…"}], # ROWS SORTED by recommended priority to address — recipes WITH CVEs first, then failures, # then stale-tests, then routine green, then up-to-date/skipped. @@ -232,6 +232,8 @@ def _table(rows, repo_url=None): # int is the confirmed CVE count; 0/omit is a confirmed-clean scan. if isinstance(cve, str) and cve.strip() == "?": cve_cell = '?' + elif isinstance(cve, str) and cve.strip().upper() == "YES": + cve_cell = 'YES' elif isinstance(cve, (int, float)) and cve: cve_cell = f'{int(cve)}' else: diff --git a/cc-ci-plan/test-advisory-scan.py b/cc-ci-plan/test-advisory-scan.py index df527e0..fdf3ed2 100755 --- a/cc-ci-plan/test-advisory-scan.py +++ b/cc-ci-plan/test-advisory-scan.py @@ -52,7 +52,7 @@ def vendor(url, cves=(), status="ok"): def run_scan(gh_entries=(), vendor_entries=(), tag_dates=None, *, v_from=None, v_to=None, - images=None, recipe="fixture", urls=None, releases=None): + images=None, recipe="fixture", urls=None, releases=None, image_sources=None): """scan() with every network call replaced by fixtures. `releases` maps CVE id -> tags whose release notes name it (the third, release-note method).""" @@ -62,6 +62,7 @@ def run_scan(gh_entries=(), vendor_entries=(), tag_dates=None, *, v_from=None, v with unittest.mock.patch.object(A, "registry_urls", lambda r, d: (list(urls), "/fake/reg.md")), \ unittest.mock.patch.object(A, "github_advisories", lambda u: list(gh_entries)), \ unittest.mock.patch.object(A, "vendor_pages", lambda u: list(vendor_entries)), \ + unittest.mock.patch.object(A, "registry_image_sources", lambda r, d: image_sources or {}), \ unittest.mock.patch.object(A, "osv", lambda r, v: None), \ unittest.mock.patch.object(A, "_tag_date", lambda o, r, v: tag_dates.get(v)), \ unittest.mock.patch.object(A, "release_fix_versions", lambda src, cve: list(releases.get(cve, []))): @@ -157,9 +158,23 @@ class TestRegistryUrls(unittest.TestCase): def test_prose_ellipsis_is_not_a_security_source(self): with tempfile.TemporaryDirectory() as directory: urls, _ = self._write(directory, - "https://github.com/cryptpad/cryptpad and example `https://…` URLs") + "https://github.com/cryptpad/cryptpad and example `https://…` URLs " + "http://docspec:3000/conversion http://127.0.0.1:3000/health") self.assertEqual(urls, ["https://github.com/cryptpad/cryptpad"]) + def test_image_source_table_maps_monorepo_images(self): + mapping = A.registry_image_sources("lasuite-drive", str(HERE / "upstream")) + self.assertIn("github-advisories:suitenumerique/drive", mapping["drive-backend"]) + self.assertIn("github-advisories:CollaboraOnline/online", mapping["code"]) + + def test_common_database_cpes_follow_registry_images(self): + self.assertIn(("postgres", A.COMMON_IMAGE_CPES["postgres"]), + A.registry_cpes("immich", str(HERE / "upstream"))) + self.assertIn(("mariadb", A.COMMON_IMAGE_CPES["mariadb"]), + A.registry_cpes("gitea", str(HERE / "upstream"))) + self.assertIn(("rustdesk-server", "cpe:2.3:a:rustdesk:rustdesk_server:*:*:*:*:oss:*:*:*"), + A.registry_cpes("rustdesk-server", str(HERE / "upstream"))) + def test_strips_trailing_markdown_punctuation(self): # Production defect: a captured backtick 404'd the fetch and rendered n8n/immich as '?'. import tempfile @@ -181,6 +196,25 @@ class TestRegistryUrls(unittest.TestCase): self.assertEqual((urls, path), ([], None)) +class TestNvdCompleteness(unittest.TestCase): + def test_truncated_results_are_not_a_clean_zero(self): + payload = json.dumps({"totalResults": 2, "vulnerabilities": [{"cve": {"id": "CVE-2026-0001"}}]}) + with unittest.mock.patch.object(A, "_fetch", return_value=payload): + row = A.nvd_advisories("cpe:2.3:a:example:server:*:*:*:*:*:*:*:*", "server") + self.assertEqual(row["status"], "error: NVD result set truncated") + self.assertEqual(row["advisories"], []) + + def test_database_backport_is_not_credited_again_on_major_upgrade(self): + ranges = [ + {"version": "*", "versionStartIncluding": "15.0", "versionEndExcluding": "15.14"}, + {"version": "*", "versionStartIncluding": "16.0", "versionEndExcluding": "16.5"}, + ] + self.assertIs(A._nvd_affects("15.13", ranges), True) + self.assertIs(A._nvd_affects("15.19", ranges), False) + self.assertIs(A._nvd_affects("18.6", ranges), False) + self.assertIsNone(A._nvd_affects("15-alpine", ranges)) + + # ── C. --image argument parsing ─────────────────────────────────────────────────────────────────── class TestImageArgParsing(unittest.TestCase): @@ -306,6 +340,24 @@ class TestPerImageWindows(unittest.TestCase): self.assertEqual(rep["unmatched_image_windows"], ["nosuchimage"]) self.assertIsNone(rep["cve_count_fixed"]) + def test_registry_maps_image_to_differently_named_source(self): + rep = run_scan([self.APP, gh("suitenumerique/drive", [adv("CVE-DRIVE-1", patched="0.25.3")])], + v_from="3.5.3", v_to="3.5.4", images=[("drive-backend", "0.22.0", "0.25.3")], + urls=self.URLS, + image_sources={"drive-backend": {"github-advisories:suitenumerique/drive"}}) + self.assertEqual(rep.get("unmatched_image_windows"), None) + self.assertIn("CVE-DRIVE-1", rep["fixed_by_this_upgrade"]) + + def test_monorepo_images_share_identical_window(self): + source = "github-advisories:suitenumerique/drive" + rep = run_scan([gh("suitenumerique/drive", [adv("CVE-DRIVE-1", patched="0.25.3")])], + v_from="0.22.0", v_to="0.25.3", + images=[("drive-backend", "0.22.0", "0.25.3")], + image_sources={"drive-backend": {source}}) + self.assertTrue(rep["count_known"]) + self.assertEqual(rep.get("unmatched_image_windows"), None) + self.assertEqual(rep["cve_count_fixed"], 1) + class TestSchemeChangeDateFallback(unittest.TestCase): DATES = {"3.5.3": "2025-12-30T00:00:00Z", "2026.7.1": "2026-07-31T00:00:00Z"} @@ -367,6 +419,15 @@ class TestCountGuarantees(unittest.TestCase): self.assertTrue(rep["count_known"]) self.assertIn("0 identified", A.markdown(rep)) + def test_ghsa_without_cve_is_preserved_as_security_advisory(self): + entry = {"cve": None, "ghsa": "GHSA-aaaa-bbbb-cccc", "severity": "low", + "patched": "1.1", "summary": "security fix", "published_at": None} + rep = run_scan([gh("app/app", [entry])], v_from="1.0", v_to="1.1") + self.assertTrue(rep["count_known"]) + self.assertEqual(rep["cve_count_fixed"], 0) + self.assertEqual(rep["fixed_non_cve_advisories"], ["GHSA-aaaa-bbbb-cccc"]) + self.assertIn("use YES", A.markdown(rep)) + def test_404_advisory_feed_is_benign_not_a_failure(self): rep = run_scan([gh("app/app", [adv("CVE-1", patched="1.1")]), gh("side/car", [], status="no-advisories-published")], diff --git a/cc-ci-plan/upstream/rustdesk-server.md b/cc-ci-plan/upstream/rustdesk-server.md index 3e125d9..a3aaa64 100644 --- a/cc-ci-plan/upstream/rustdesk-server.md +++ b/cc-ci-plan/upstream/rustdesk-server.md @@ -5,11 +5,11 @@ | app | rustdesk/rustdesk-server | https://github.com/rustdesk/rustdesk-server | https://github.com/rustdesk/rustdesk-server/releases | | signal | rustdesk/rustdesk-server | https://github.com/rustdesk/rustdesk-server | https://github.com/rustdesk/rustdesk-server/releases | +- nvd-cpe: rustdesk-server = cpe:2.3:a:rustdesk:rustdesk_server:*:*:*:*:oss:*:*:* + ## Standing notes -- **NOT enrolled** in the cc-ci weekly set (operator 2026-10-30: single client, maintenance contract - pending). Mirror `recipe-maintainers/rustdesk-server` exists + this registry entry so one-off - `/recipe-upgrade` runs and advisory scans work; if the client confirms, enroll via `/recipe-enroll` - (test suite + bridge + used-recipes weekly row). +- Enrolled in the cc-ci weekly set. The NVD CPE above is scoped to the open source server edition; + do not count RustDesk Client or Server Pro CVEs for this recipe. - Both services (`app` = hbbs, `signal` = hbbr) use the SAME image — always bump them together. - Docker Hub has stray `1.4.5-*` arch tags (pushed 2026-01-12, top-level `1.4.5` tag deleted, no GitHub release) — an aborted/alias push, NOT a real version line. The real line is `1.1.x`; -- 2.54.0