Compare commits

..
10 changed files with 458 additions and 85 deletions
+16 -6
View File
@@ -95,8 +95,9 @@ For each recipe produce **one window per image**: `current pinned tag → newest
the sidecars (redis, postgres, nginx …), not just the app — a sidecar bump is where discourse's only the sidecars (redis, postgres, nginx …), not just the app — a sidecar bump is where discourse's only
CRITICAL came from, and an image with no window is not counted at all. CRITICAL came from, and an image with no window is not counted at all.
- **No upgrade available** → the recipe is `UPTODATE`; its CVE count is **`0`**, not `?`. There is - **No upgrade available** → the recipe is `UPTODATE`; its **CVEs fixed by an available upgrade**
nothing an upgrade could fix. Record it and move on. count is `0`. This does not assert that the running image has no known CVEs. If advisory sources
failed or are blind, use `?` and say why.
- **No output at all is NOT "no upgrade".** An abra call that times out, FATAs, or prints nothing - **No output at all is NOT "no upgrade".** An abra call that times out, FATAs, or prints nothing
leaves the recipe **unverified** — treat it as a distinct outcome, never fold it into up-to-date. leaves the recipe **unverified** — treat it as a distinct outcome, never fold it into up-to-date.
Re-run it, and if it still yields nothing, resolve the versions by direct registry check (box item 4). Re-run it, and if it still yields nothing, resolve the versions by direct registry check (box item 4).
@@ -111,9 +112,17 @@ python3 cc-ci-plan/audit-sources.py --security-sources
A recipe whose sources yield **no CVE data at all** cannot produce a meaningful `0` — nothing was A recipe whose sources yield **no CVE data at all** cannot produce a meaningful `0` — nothing was
measured, the same way a missing registry file cannot. Render those as **`?`**, not `0`. measured, the same way a missing registry file cannot. Render those as **`?`**, not `0`.
**The fleet is currently at zero such recipes.** The last two — `mattermost-lts` (empty advisory Do not assume that the fleet has zero blind recipes. Run the audit every sweep and inspect source
feed, client-side-rendered bulletins) and `mumble` (nothing published anywhere) — were fixed by statuses; an HTTP error or rate limit is not evidence of no CVEs. `rustdesk-server` has an NVD CPE
declaring an NVD CPE in their registry: 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
failed sources, so the affected CVE counts remain unknown.
Two older blind spots — `mattermost-lts` and `mumble` — were addressed by declaring NVD CPEs in
their registries:
``` ```
- nvd-cpe: mattermost-team-edition = cpe:2.3:a:mattermost:mattermost_server:*:*:*:*:*:*:*:* - nvd-cpe: mattermost-team-edition = cpe:2.3:a:mattermost:mattermost_server:*:*:*:*:*:*:*:*
``` ```
@@ -169,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. - `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 `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 `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.** `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 - `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. established, a scan whose count is a floor with many undetermined advisories.
+13 -4
View File
@@ -30,7 +30,9 @@ keeps every weekly edition looking the same regardless of which model writes the
2. **Survey.** `python3 /srv/cc-ci/cc-ci-plan/recipe-report.py survey <DATE> > /tmp/survey.json`. This 2. **Survey.** `python3 /srv/cc-ci/cc-ci-plan/recipe-report.py survey <DATE> > /tmp/survey.json`. This
gives you: the raw `/upgrade-all` summary; for every recipe mirror its open PRs + the `cc-ci/testme` gives you: the raw `/upgrade-all` summary; for every recipe mirror its open PRs + the `cc-ci/testme`
verdict; and each recipe's **per-recipe upgrade notes** (`upgrade_notes_md` — the breaking-change / verdict; and each recipe's **per-recipe upgrade notes** (`upgrade_notes_md` — the breaking-change /
**CVE / security** analysis the upgrade did). Read it all. **CVE / security** analysis the upgrade did). Read it all. Also run
`python3 /srv/cc-ci/cc-ci-plan/audit-sources.py --security-sources`; a recipe with no verified
security source or a failed source must be called out, not presented as a clean CVE scan.
3. **Review & classify.** The page order is: **short lead → the full wire table (priority-sorted) → 3. **Review & classify.** The page order is: **short lead → the full wire table (priority-sorted) →
Addendum → Security Bulletin → per-recipe "What changed".** Same newspaper style; this order. Addendum → Security Bulletin → per-recipe "What changed".** Same newspaper style; this order.
@@ -46,9 +48,16 @@ keeps every weekly edition looking the same regardless of which model writes the
lists **failed sources**, the count is **not** authoritative: render the cve cell as `?` lists **failed sources**, the count is **not** authoritative: render the cve cell as `?`
(unknown), never `none` — a blank that reads as "clean" is exactly how two CVSS-9.8 gitea (unknown), never `none` — a blank that reads as "clean" is exactly how two CVSS-9.8 gitea
RCEs were reported as "none" on 2026-08-07. RCEs were reported as "none" on 2026-08-07.
- **`?` must stay RARE — it means "we tried and could not tell", not "we didn't look".** Use it - The table's **CVEs** column counts known CVEs addressed by the available recipe upgrade.
ONLY when a scan ran and reported genuinely failed sources, **or when the scan block says A zero for an up-to-date recipe means there is no upgrade window to credit; it does not
COUNT UNKNOWN**. In that case the scan's `0` means *not determined*: publish `?` and say so in 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
the notes; publishing `0` would assert a clean bill of health nothing supports. Note a the notes; publishing `0` would assert a clean bill of health nothing supports. Note a
**version-scheme change is no longer a reason for `?`** — the scan resolves semver→calver jumps **version-scheme change is no longer a reason for `?`** — the scan resolves semver→calver jumps
(discourse 3.5.3 → 2026.7.1) by falling back to advisory publish dates and reports a real number. (discourse 3.5.3 → 2026.7.1) by falling back to advisory publish dates and reports a real number.
+25 -6
View File
@@ -143,12 +143,31 @@ ssh cc-ci 'export PATH=/run/current-system/sw/bin:$PATH; R=<r>; \
script -qec "abra recipe upgrade $R -m -n" /dev/null; \ script -qec "abra recipe upgrade $R -m -n" /dev/null; \
git -C ~/.abra/recipes/$R stash pop >/dev/null 2>&1 || true' git -C ~/.abra/recipes/$R stash pop >/dev/null 2>&1 || true'
``` ```
Build `RECIPES_TO_UPGRADE` = recipes with **≥1 available upgrade** (after the stash, so the untracked **Mandatory registry cross-check for EVERY recipe and EVERY image, even if abra says current:** after
overlay doesn't count as dirty). Others go to `SKIPPED_UPFRONT` with a reason (`dirty-worktree` only for reconciling the mirror and refreshing the recipe checkout, run
real TRACKED edits, `up-to-date`, `not-fetchable`). **A `tag and digest … not supported` parse FATA is `python3 /srv/cc-ci/cc-ci-plan/resolve-images.py <recipe> --recipe-dir /root/.abra/recipes`
NOT `not-fetchable`** — per box item 4, catch it and resolve availability via the direct upstream-tag on cc-ci. Read the JSON `images`, `upgrades_available`, `unresolved`, and `all_resolved` fields.
check; only `not-fetchable` when the mirror genuinely can't be fetched (network/auth), never for a The candidate list is the **union** of abra's upgrades and the resolver's upgrades. CryptPad's
digest-pinned image. `version-YYYY.M.D` app tag is a known counterexample: abra only reports a parse warning, while the
resolver finds the newer image. A warning is never evidence that an image is current. Inspect both
`newest_same_shape` and `newest_within_major`; a newer major is still a candidate for planning,
though it must not be applied without a compatibility check.
Build `RECIPES_TO_UPGRADE` from every recipe with at least one candidate. Only label a recipe
`up-to-date` if abra completed successfully **and** the resolver checked every relevant image
(`all_resolved: true`) **and** neither found a candidate. If an image is unresolved, templated,
floating, or lacks comparable tags, investigate it directly and record the evidence. Otherwise
record `unverified` with the image and reason; never fold it into `up-to-date`. `dirty-worktree`
applies only to real tracked edits. A tag+digest parse failure is not `not-fetchable`.
Run `python3 /srv/cc-ci/cc-ci-plan/audit-sources.py --security-sources` once for the full inventory
and include any `NO VERIFIED CVE SOURCE` recipes or failed advisory sources in the weekly summary.
Do not describe their CVE count as clean. This audit is a coverage check, not a substitute for
`advisory-scan.py` on each actual upgrade window.
Before the audit, check `test -s /srv/cc-ci/.github-token`. If it is missing, record that GitHub
advisory checks are subject to the anonymous 60-request/hour limit; any resulting HTTP 403 is a
failed source and must become `?` in the report. Never replace it with `0` or silently retry from
model memory. The file is a read-only GitHub token location; never print its contents.
**Reconcile every candidate's mirror during the survey** (even the up-to-date ones), so merged-upstream **Reconcile every candidate's mirror during the survey** (even the up-to-date ones), so merged-upstream
PRs are closed and every mirror `main` tracks true upstream — fleet-wide, not just where there's a new PRs are closed and every mirror `main` tracks true upstream — fleet-wide, not just where there's a new
+171 -27
View File
@@ -40,6 +40,7 @@ from __future__ import annotations
import argparse import argparse
import gzip import gzip
import ipaddress
import json import json
import os import os
import re import re
@@ -166,6 +167,43 @@ def _vkey(v: str | None) -> tuple:
return tuple(out) 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: def _already_fixed_on_from_line(kf: tuple, cands: list[tuple]) -> bool:
"""Was it ALREADY fixed on the line we are upgrading FROM? """Was it ALREADY fixed on the line we are upgrading FROM?
@@ -231,6 +269,16 @@ def registry_urls(recipe: str, registry_dir: str) -> tuple[list[str], str | None
# recipe render '?' for no real reason — that is what put immich and n8n in the unknown # 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). # column on 2026-08-07 (https://docs.n8n.io/release-notes/` ← note the backtick).
u = u.rstrip("`'\"*.,;:>)") 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: if u and u not in urls:
urls.append(u) urls.append(u)
return urls, path return urls, path
@@ -369,15 +417,57 @@ def vendor_pages(urls: list[str]) -> list[dict]:
NVD_API = "https://services.nvd.nist.gov/rest/json/cves/2.0" 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_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]]: 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") path = os.path.join(registry_dir, f"{recipe}.md")
try: try:
return [(m.group(1), m.group(2)) for m in NVD_CPE_RE.finditer(open(path).read())] text = open(path).read()
except OSError: except OSError:
return [] 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: def nvd_advisories(cpe: str, key: str) -> dict:
@@ -395,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, 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. 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": []} entry = {"source": f"nvd:{key}", "status": "ok", "advisories": []}
url = f"{NVD_API}?resultsPerPage=2000&virtualMatchString={urllib.parse.quote(cpe)}" url = f"{NVD_API}?resultsPerPage=2000&virtualMatchString={urllib.parse.quote(cpe)}"
data = None data = None
@@ -407,15 +499,31 @@ def nvd_advisories(cpe: str, key: str) -> dict:
entry["status"] = f"error: {type(e).__name__}" entry["status"] = f"error: {type(e).__name__}"
return entry return entry
time.sleep(8) 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 {} c = v.get("cve") or {}
cid = c.get("id") cid = c.get("id")
if not cid: if not cid or c.get("vulnStatus") == "Rejected":
continue 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 cfg in c.get("configurations", []):
for node in cfg.get("nodes", []): for node in cfg.get("nodes", []):
for m in node.get("cpeMatch", []): 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"): if m.get("versionEndExcluding"):
fixed.add(m["versionEndExcluding"]) fixed.add(m["versionEndExcluding"])
elif m.get("versionEndIncluding"): elif m.get("versionEndIncluding"):
@@ -433,9 +541,11 @@ def nvd_advisories(cpe: str, key: str) -> dict:
"vulnerable_range": None, "vulnerable_range": None,
"patched": "; ".join(sorted(fixed)) or None, "patched": "; ".join(sorted(fixed)) or None,
"affected_max": "; ".join(sorted(affected_max)) 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}", "url": f"https://nvd.nist.gov/vuln/detail/{cid}",
"published_at": c.get("published"), "description": None, "cvss": None, "published_at": c.get("published"), "description": None, "cvss": None,
}) })
_NVD_CACHE[cpe] = entry
return entry return entry
@@ -737,7 +847,8 @@ def scan(recipe: str, v_from: str | None, v_to: str | None, registry_dir: str,
"vulnerable_range": None, "patched": None, "vulnerable_range": None, "patched": None,
"context": None, "published_at": None, "context": None, "published_at": None,
"description": None, "url": None, "cvss": 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"]: if src not in e["sources"]:
e["sources"].append(src) e["sources"].append(src)
for k, v in extra.items(): for k, v in extra.items():
@@ -748,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"], report["sources"].append({"source": entry["source"], "status": entry["status"],
"found": len(entry.get("advisories", []))}) "found": len(entry.get("advisories", []))})
for a in entry.get("advisories", []): for a in entry.get("advisories", []):
if a.get("cve"): advisory_id = a.get("cve") or a.get("ghsa")
record(a["cve"], entry["source"], severity=a.get("severity"), ghsa=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"), vulnerable_range=a.get("vulnerable_range"), patched=a.get("patched"),
context=a.get("summary"), published_at=a.get("published_at"), context=a.get("summary"), published_at=a.get("published_at"),
description=a.get("description"), url=a.get("url"), cvss=a.get("cvss")) description=a.get("description"), url=a.get("url"), cvss=a.get("cvss"))
@@ -761,6 +873,7 @@ def scan(recipe: str, v_from: str | None, v_to: str | None, registry_dir: str,
for a in entry.get("advisories", []): for a in entry.get("advisories", []):
record(a["cve"], entry["source"], severity=a.get("severity"), record(a["cve"], entry["source"], severity=a.get("severity"),
patched=a.get("patched"), affected_max=a.get("affected_max"), 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"), context=a.get("summary"), published_at=a.get("published_at"),
url=a.get("url")) url=a.get("url"))
@@ -825,18 +938,25 @@ def scan(recipe: str, v_from: str | None, v_to: str | None, registry_dir: str,
if pname in k or k in pname: if pname in k or k in pname:
windows[src] = (v_from, v_to) windows[src] = (v_from, v_to)
window_key[src] = k window_key[src] = k
image_sources = registry_image_sources(recipe, registry_dir)
for key, wf, wt in (images or []): for key, wf, wt in (images or []):
matched = False
explicit = image_sources.get(key.lower(), set())
for src in gh_sources: for src in gh_sources:
if src in windows:
continue
# Match BOTH ways: an image name is often longer than its source repo # Match BOTH ways: an image name is often longer than its source repo
# (`clickhouse/clickhouse-server` vs source `ClickHouse/ClickHouse`) and sometimes # (`clickhouse/clickhouse-server` vs source `ClickHouse/ClickHouse`) and sometimes
# shorter (`redis` vs `redis/redis`). One-directional matching silently dropped the # shorter (`redis` vs `redis/redis`). One-directional matching silently dropped the
# clickhouse window when the key was derived from a compose file. # clickhouse window when the key was derived from a compose file.
k, name = key.lower(), src.split("/")[-1].lower() 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) windows[src] = (wf, wt)
window_key[src] = key window_key[src] = key
matched = True
if not matched:
report.setdefault("unmatched_image_windows", []).append(key)
report["windows"] = {k: {"from": f, "to": t} for k, (f, t) in windows.items()} report["windows"] = {k: {"from": f, "to": t} for k, (f, t) in windows.items()}
def _classify_window(src, wf, wt): def _classify_window(src, wf, wt):
@@ -844,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 `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.""" 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) kf, kt = _vkey(wf), _vkey(wt)
# A version-SCHEME change (semver 3.5.3 → calver 2026.7.1) makes numeric ordering # 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. # meaningless: 2025.12.2 compares "newer" than 3.5.3 while shipping earlier.
@@ -1012,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" report["cves"][cve]["classification"] = "indeterminate: no fix version published"
unknown = [c for c in unknown if c not in fixed_set] unknown = [c for c in unknown if c not in fixed_set]
report["fixed_by_this_upgrade"] = sorted(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["indeterminate"] = sorted(indeterminate)
report["cve_count_indeterminate"] = len(indeterminate) report["cve_count_indeterminate"] = len(indeterminate)
report["unclassified"] = sorted(unknown) report["unclassified"] = sorted(unknown)
@@ -1020,24 +1156,22 @@ def scan(recipe: str, v_from: str | None, v_to: str | None, registry_dir: str,
# A scan with NO usable source has not measured anything, so it must not report a number — # A scan with NO usable source has not measured anything, so it must not report a number —
# least of all 0, which asserts safety. mumble had no cc-ci-plan/upstream/mumble.md at all and # least of all 0, which asserts safety. mumble had no cc-ci-plan/upstream/mumble.md at all and
# still produced "0 identified", which was then published as a clean 0 in a CVE report. # still produced "0 identified", which was then published as a clean 0 in a CVE report.
usable_sources = [ usable_sources = [s for s in report["sources"] if s["status"] == "ok"]
s for s in report["sources"]
if s["status"] == "ok" or s["status"].startswith("no-advisories-published")
]
no_sources = not usable_sources no_sources = not usable_sources
if no_sources: if no_sources:
report["no_usable_sources"] = True report["no_usable_sources"] = True
report["count_known"] = not unresolved_any and not no_sources
report["cve_count_fixed"] = (len(fixed_set)
if (not unresolved_any and not no_sources) 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"] = [ report["sources_failed"] = [
s["source"] s["source"]
for s in report["sources"] for s in report["sources"]
if not (s["status"] == "ok" or s["status"].startswith(("no-advisories-published", "skipped:"))) if not (s["status"] == "ok" or s["status"].startswith(("no-advisories-published", "skipped:")))
] ]
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"] = (sum(x.startswith("CVE-") for x in fixed_set)
if report["count_known"] else None)
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"] = [ report["sources_benign"] = [
s["source"] s["source"]
for s in report["sources"] for s in report["sources"]
@@ -1052,16 +1186,20 @@ def markdown(rep: dict) -> str:
f"{rep.get('from') or '?'} → {rep.get('to') or '?'}"] f"{rep.get('from') or '?'} → {rep.get('to') or '?'}"]
if not rep.get("count_known", True): 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**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 advisory/advisories were identified: "
+ ", ".join(rep["fixed_by_this_upgrade"][:15]))
if rep.get("no_usable_sources"): if rep.get("no_usable_sources"):
L.append("\n⚠ This is NOT zero. **No usable source was checked at all** — the registry " L.append("\n⚠ This is NOT zero. **No usable source was checked at all** — the registry "
"file `cc-ci-plan/upstream/<recipe>.md` is missing or every source failed, so " "file `cc-ci-plan/upstream/<recipe>.md` is missing or every source failed, so "
"nothing was measured. Render this recipe's cve cell as `?`, never `0`, and add " "nothing was measured. Render this recipe's cve cell as `?`, never `0`, and add "
"the registry file.") "the registry file.")
else: else:
L.append("\n⚠ This is NOT zero. A version-scheme change (e.g. semver → calver) makes " L.append("\n⚠ This is NOT zero. At least one version window or advisory source could "
"numeric ordering meaningless across this jump, so no advisory could be " "not be checked completely. Render this recipe's cve cell as `?`, never `0`.")
"classified. Render this recipe's cve cell as `?`, never `0`. Read the vendor's " if rep.get("unmatched_image_windows"):
"release notes for the jump and count by hand.") L.append("\n⚠ image windows with no matching advisory source: "
+ ", ".join(rep["unmatched_image_windows"]))
if rep["unclassified"]: if rep["unclassified"]:
L.append(f"\nAdvisories seen but unclassifiable ({len(rep['unclassified'])}) — includes " L.append(f"\nAdvisories seen but unclassifiable ({len(rep['unclassified'])}) — includes "
f"other images in this recipe: " + ", ".join(rep["unclassified"][:12])) f"other images in this recipe: " + ", ".join(rep["unclassified"][:12]))
@@ -1072,7 +1210,8 @@ def markdown(rep: dict) -> str:
f"reading in the upgrade step._") f"reading in the upgrade step._")
return "\n".join(L) return "\n".join(L)
ind = rep.get("cve_count_indeterminate") or 0 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 "" floor = " (at least — see undetermined below)" if ind else ""
L.append(f"\n**CVEs fixed by this upgrade: {rep['cve_count_fixed']}**{floor}\n") L.append(f"\n**CVEs fixed by this upgrade: {rep['cve_count_fixed']}**{floor}\n")
cb = rep.get("classified_by") or {} cb = rep.get("classified_by") or {}
@@ -1086,12 +1225,17 @@ def markdown(rep: dict) -> str:
L.append("") L.append("")
L.append("| CVE | severity | fixed in | advisory | source |") L.append("| CVE | severity | fixed in | advisory | source |")
L.append("|---|---|---|---|---|") L.append("|---|---|---|---|---|")
for cve in rep["fixed_by_this_upgrade"]: for cve in fixed_cves:
e = rep["cves"][cve] e = rep["cves"][cve]
L.append(f"| {cve} | {e.get('severity') or '?'} | {e.get('patched') or '?'} | " L.append(f"| {cve} | {e.get('severity') or '?'} | {e.get('patched') or '?'} | "
f"{e.get('ghsa') or '-'} | {e['sources'][0]} |") f"{e.get('ghsa') or '-'} | {e['sources'][0]} |")
else: else:
L.append("\n**CVEs fixed by this upgrade: 0 identified by the deterministic scan.**") 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: if ind:
L.append(f"\n⚠ **{ind} advisory/advisories could NOT be judged** — the vendor published no fix " 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 " f"version (GitHub carries `TBD` or a placeholder like `7.4.X`) and the vulnerable "
+15 -18
View File
@@ -134,7 +134,7 @@ def security_source_audit(recipe: str) -> list[dict]:
e = A.nvd_advisories(cpe, key) e = A.nvd_advisories(cpe, key)
n = len(e.get("advisories") or []) n = len(e.get("advisories") or [])
out.append({"source": e["source"] + f" ({cpe.split(':')[4]}/{cpe.split(':')[3]})", 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}) "status": e["status"], "cves": n, "usable": n})
for entry in A.github_advisories(urls): for entry in A.github_advisories(urls):
out.append({"source": entry["source"], "kind": "advisory-feed", out.append({"source": entry["source"], "kind": "advisory-feed",
@@ -227,16 +227,7 @@ def audit_recipe(recipe: str, ssh: str | None, quiet_days: int) -> dict:
def _registry_urls(recipe: str): def _registry_urls(recipe: str):
path = os.path.join(REGISTRY_DIR, f"{recipe}.md") return A.registry_urls(recipe, REGISTRY_DIR)
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]: def all_recipes() -> list[str]:
@@ -272,23 +263,29 @@ def main() -> int:
blind_recipes, noisy = [], 0 blind_recipes, noisy = [], 0
for r in recipes: for r in recipes:
rows = security_source_audit(r) 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"
logs = [x for x in rows if x["kind"] == "changelog"] 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"] unattr = [x for x in rows if x["kind"] == "unattributable"]
failed = [x for x in rows if x["status"].startswith("error:")]
noisy += len(unattr) noisy += len(unattr)
usable = len(feeds) + len(logs) usable = len(feeds) + len(logs)
if usable == 0: if usable == 0:
blind_recipes.append(r) blind_recipes.append(r)
print(f"!! {r}: NO USABLE CVE SOURCE — {len(unattr)} unparseable page(s), " print(f"!! {r}: NO VERIFIED CVE SOURCE — {len(unattr)} unparseable page(s), "
f"0 advisory feeds, 0 attributable changelogs") f"0 populated advisory feeds, 0 attributable changelogs, {len(failed)} failed source(s)")
for x in rows: for x in rows:
print(f" {x['kind']:15} {x['source'][:64]} ({x['cves']} CVEs)") unit = "advisories" if x["kind"] == "advisory-feed" else "CVE IDs"
print(f" {x['kind']:15} {x['source'][:64]} ({x['cves']} {unit}; {x['status']})")
else: else:
print(f"OK {r}: {len(feeds)} advisory-feed(s), {len(logs)} changelog(s)" print(f"OK {r}: {len(feeds)} advisory-feed(s), {len(logs)} changelog(s)"
+ (f", {len(unattr)} unparseable page(s) (redundant — covered by a feed)" + (f", {len(unattr)} unparseable page(s)" if unattr else "")
if unattr else "")) + (f", {len(failed)} failed source(s) — counts incomplete" if failed else ""))
for x in logs: for x in logs:
print(f" changelog {x['source'][:62]} ({x['usable']}/{x['cves']})") 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" print(f"\n{len(recipes)} recipes · {len(blind_recipes)} with NO usable CVE source"
+ (f": {', '.join(blind_recipes)}" if blind_recipes else "") + (f": {', '.join(blind_recipes)}" if blind_recipes else "")
+ f" · {noisy} unparseable page(s) elsewhere (harmless where a feed covers them)") + f" · {noisy} unparseable page(s) elsewhere (harmless where a feed covers them)")
+3 -1
View File
@@ -30,7 +30,7 @@ SPEC SHAPE (the agent writes this JSON):
{"date":"YYYY-MM-DD","subtitle":"Week of <human date>", {"date":"YYYY-MM-DD","subtitle":"Week of <human date>",
"lead":"<ONE short paragraph>", "lead":"<ONE short paragraph>",
"table":[{"recipe":"x","change":"a → b","status":"GREEN|FAILED|STALE|SKIPPED|UPTODATE", # -> TESTS col "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":"…"}], "ci":"build 154 ✓","ci_url":"…","pr":"#4","pr_url":"…","notes":"…"}],
# ROWS SORTED by recommended priority to address — recipes WITH CVEs first, then failures, # ROWS SORTED by recommended priority to address — recipes WITH CVEs first, then failures,
# then stale-tests, then routine green, then up-to-date/skipped. # 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. # int is the confirmed CVE count; 0/omit is a confirmed-clean scan.
if isinstance(cve, str) and cve.strip() == "?": if isinstance(cve, str) and cve.strip() == "?":
cve_cell = '<span class="muted" title="advisory scan incomplete or absent — CVE count unknown">?</span>' cve_cell = '<span class="muted" title="advisory scan incomplete or absent — CVE count unknown">?</span>'
elif isinstance(cve, str) and cve.strip().upper() == "YES":
cve_cell = '<span class="cve">YES</span>'
elif isinstance(cve, (int, float)) and cve: elif isinstance(cve, (int, float)) and cve:
cve_cell = f'<span class="cve">{int(cve)}</span>' cve_cell = f'<span class="cve">{int(cve)}</span>'
else: else:
+76 -16
View File
@@ -163,7 +163,8 @@ def parse_ref(ref: str) -> dict:
return {"registry": host, "repo": repo, "tag": tag, "digest": digest} return {"registry": host, "repo": repo, "tag": tag, "digest": digest}
HUB_RECENT_PAGES = int(os.environ.get("RESOLVE_IMAGES_HUB_PAGES", "10")) HUB_RECENT_PAGES = int(os.environ.get("RESOLVE_IMAGES_HUB_PAGES", "40"))
_HUB_TRUNCATED: set[str] = set()
def _hub_tag_exists(repo: str, tag: str) -> bool: def _hub_tag_exists(repo: str, tag: str) -> bool:
@@ -177,10 +178,10 @@ def _hub_tag_exists(repo: str, tag: str) -> bool:
def _hub_tags(repo: str) -> list[str]: def _hub_tags(repo: str) -> list[str]:
"""Recently-pushed tags, newest first. """Recently-pushed tags, newest first.
Popular Docker Hub repos carry many thousands of tags, so a full enumeration is impractical — Popular Docker Hub repos carry many thousands of tags, so enumerate a bounded recent window.
but it is also unnecessary: a tag NEWER than the one we run must have been pushed AFTER it, so A rebuilt old tag can displace a newer version from that window, so callers MUST treat a
ordering by last_updated and reading a bounded recent window is sufficient to find any upgrade. truncated listing with no detected upgrade as unverified, not current. (ghcr offers no recency
(ghcr offers no ordering, which is why that path needs a different strategy.) ordering, which is why that path needs a different strategy.)
""" """
tags, url = [], (f"https://hub.docker.com/v2/repositories/{repo}/tags" tags, url = [], (f"https://hub.docker.com/v2/repositories/{repo}/tags"
f"?page_size=100&ordering=last_updated") f"?page_size=100&ordering=last_updated")
@@ -191,6 +192,8 @@ def _hub_tags(repo: str) -> list[str]:
url = d.get("next") url = d.get("next")
if not url: if not url:
break break
if url:
_HUB_TRUNCATED.add(repo)
return tags return tags
@@ -236,15 +239,26 @@ def _oci_tags(host: str, repo: str) -> list[str]:
hdrs = _oci_bearer(host, repo) hdrs = _oci_bearer(host, repo)
tags, url = [], f"https://{host}/v2/{repo}/tags/list?n=1000" tags, url = [], f"https://{host}/v2/{repo}/tags/list?n=1000"
for _ in range(MAX_TAG_PAGES): for _ in range(MAX_TAG_PAGES):
req = urllib.request.Request(url, headers={"User-Agent": UA, **hdrs}) for attempt in range(RETRIES):
with urllib.request.urlopen(req, timeout=TIMEOUT) as r: req = urllib.request.Request(url, headers={"User-Agent": UA, **hdrs})
tags += (json.load(r) or {}).get("tags") or [] try:
link = r.headers.get("Link", "") or "" with urllib.request.urlopen(req, timeout=TIMEOUT) as r:
tags += (json.load(r) or {}).get("tags") or []
link = r.headers.get("Link", "") or ""
break
except urllib.error.HTTPError as e:
if e.code not in (429, 503) or attempt == RETRIES - 1:
raise
retry_after = e.headers.get("Retry-After") or ""
delay = min(int(retry_after) if retry_after.isdigit() else 2 ** (attempt + 1), 60)
time.sleep(delay)
m = re.search(r'<([^>]+)>;\s*rel="next"', link) m = re.search(r'<([^>]+)>;\s*rel="next"', link)
if not m: if not m:
break break
nxt = m.group(1) nxt = m.group(1)
url = f"https://{host}{nxt}" if nxt.startswith("/") else nxt url = f"https://{host}{nxt}" if nxt.startswith("/") else nxt
else:
raise RuntimeError(f"tag listing truncated after {MAX_TAG_PAGES} pages")
return tags return tags
@@ -299,6 +313,39 @@ def _release_fallback_repos(registry: str, repo: str) -> list[tuple[str, str]]:
return cands return cands
def _oci_tag_exists(host: str, repo: str, tag: str) -> bool:
"""Verify a release candidate is actually published as an image manifest."""
hdrs = {"User-Agent": UA,
"Accept": "application/vnd.oci.image.index.v1+json, "
"application/vnd.docker.distribution.manifest.list.v2+json, "
"application/vnd.oci.image.manifest.v1+json",
**_oci_bearer(host, repo)}
url = f"https://{host}/v2/{repo}/manifests/{urllib.parse.quote(tag, safe='')}"
try:
with urllib.request.urlopen(urllib.request.Request(url, headers=hdrs, method="HEAD"),
timeout=TIMEOUT) as response:
return response.status == 200
except urllib.error.HTTPError as e:
if e.code == 404:
return False
raise
def _release_manifest_fallback(registry: str, repo: str, current: str) -> tuple[list[str], str] | None:
"""For huge GHCR registries, use release tags only after checking actual manifests."""
for owner, name in _release_fallback_repos(registry, repo):
releases = github_release_tags(owner, name)
if current not in releases:
continue # incomplete release window; it cannot establish a current baseline
candidates = sorted((tag for tag in set(releases)
if shape(tag) == shape(current) and vkey(tag) >= vkey(current)),
key=vkey, reverse=True)
for tag in candidates:
if _oci_tag_exists(registry, repo, tag):
return [current, tag], f"github-releases+registry-manifest:{owner}/{name}"
return None
_TAG_CACHE: dict[tuple[str, str], tuple[list[str], str | None]] = {} _TAG_CACHE: dict[tuple[str, str], tuple[list[str], str | None]] = {}
@@ -335,8 +382,16 @@ def resolve(ref: str) -> dict:
"upgrade_available": False} "upgrade_available": False}
tags, err = list_tags(info["registry"], info["repo"]) tags, err = list_tags(info["registry"], info["repo"])
if err: if err:
out["status"] = f"error: {err}" try:
return out fallback = (_release_manifest_fallback(info["registry"], info["repo"], info["tag"])
if "tag listing truncated" in err else None)
except Exception as e: # noqa: BLE001 — a failed fallback must stay unresolved
out["status"] = f"error: registry {err}; release fallback {type(e).__name__}: {e}"
return out
if not fallback:
out["status"] = f"error: {err}"
return out
tags, out["source"] = fallback
out["tags_seen"] = len(set(tags)) out["tags_seen"] = len(set(tags))
# INTEGRITY CHECK: the tag we are currently running MUST appear in the listing. If it does not, # INTEGRITY CHECK: the tag we are currently running MUST appear in the listing. If it does not,
# the listing is incomplete and any "newest" derived from it is a guess — ghcr paginates to tens # the listing is incomplete and any "newest" derived from it is a guess — ghcr paginates to tens
@@ -367,8 +422,9 @@ def resolve(ref: str) -> dict:
same = [t for t in set(tags) if shape(t) == want and vkey(t)] same = [t for t in set(tags) if shape(t) == want and vkey(t)]
out["candidates"] = len(same) out["candidates"] = len(same)
if not same: if not same:
# Not a failure: digest-only pins and `latest`/`stable` have no comparable siblings. # A floating or unusual tag cannot prove that an image is current. Keep it visible to
out["status"] = "no comparable tags (shape has no numeric siblings)" # callers rather than letting an empty candidate set become a false UPTODATE verdict.
out["status"] = "unverifiable: no comparable numeric tags"
return out return out
newest = max(same, key=vkey) newest = max(same, key=vkey)
out["newest_same_shape"] = newest out["newest_same_shape"] = newest
@@ -377,6 +433,10 @@ def resolve(ref: str) -> dict:
if within: if within:
out["newest_within_major"] = max(within, key=vkey) out["newest_within_major"] = max(within, key=vkey)
out["upgrade_available"] = bool(cur and vkey(newest) > cur) out["upgrade_available"] = bool(cur and vkey(newest) > cur)
if (info["registry"] in ("docker.io", "registry-1.docker.io")
and info["repo"] in _HUB_TRUNCATED and not out["upgrade_available"]):
out["status"] = (f"unverifiable: Docker Hub listing truncated after "
f"{HUB_RECENT_PAGES * 100} recent tags")
return out return out
@@ -436,16 +496,16 @@ def main() -> int:
"recipe": a.recipe, "recipe": a.recipe,
"images": results, "images": results,
"upgrades_available": [r["ref"] for r in results if r["upgrade_available"]], "upgrades_available": [r["ref"] for r in results if r["upgrade_available"]],
"unresolved": [r["ref"] for r in results if r["status"].startswith("error")], "unresolved": [r["ref"] for r in results if r["status"].startswith(("error", "skipped", "unverifiable"))],
# The whole point: distinguish "checked, current" from "could not check". # The whole point: distinguish "checked, current" from "could not check".
"all_resolved": not any(r["status"].startswith("error") for r in results), "all_resolved": bool(results) and all(r["status"] == "ok" for r in results),
} }
if not a.table: if not a.table:
print(json.dumps(report, indent=2)) print(json.dumps(report, indent=2))
return 0 return 0
print(f"{a.recipe} — {len(results)} images") print(f"{a.recipe} — {len(results)} images")
for r in results: for r in results:
flag = "UPGRADE" if r["upgrade_available"] else ("ERROR" if r["status"].startswith("error") else "current") flag = "UPGRADE" if r["upgrade_available"] else ("UNKNOWN" if r["status"] != "ok" else "current")
print(f" [{flag:7}] {r['repo']}:{r['tag']}" + (" (digest-pinned)" if r["digest"] else "")) print(f" [{flag:7}] {r['repo']}:{r['tag']}" + (" (digest-pinned)" if r["digest"] else ""))
print(f" shape={r['shape']} candidates={r['candidates']}" print(f" shape={r['shape']} candidates={r['candidates']}"
f" newest_same_shape={r['newest_same_shape']}" f" newest_same_shape={r['newest_same_shape']}"
+80 -3
View File
@@ -22,6 +22,7 @@ import io
import json import json
import os import os
import pathlib import pathlib
import tempfile
import sys import sys
import unittest import unittest
import unittest.mock import unittest.mock
@@ -51,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, 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. """scan() with every network call replaced by fixtures.
`releases` maps CVE id -> tags whose release notes name it (the third, release-note method).""" `releases` maps CVE id -> tags whose release notes name it (the third, release-note method)."""
@@ -61,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")), \ 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, "github_advisories", lambda u: list(gh_entries)), \
unittest.mock.patch.object(A, "vendor_pages", lambda u: list(vendor_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, "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, "_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, []))): unittest.mock.patch.object(A, "release_fix_versions", lambda src, cve: list(releases.get(cve, []))):
@@ -153,6 +155,26 @@ class TestRegistryUrls(unittest.TestCase):
p.write_text(text) p.write_text(text)
return A.registry_urls("r", tmp) return A.registry_urls("r", tmp)
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 "
"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): def test_strips_trailing_markdown_punctuation(self):
# Production defect: a captured backtick 404'd the fetch and rendered n8n/immich as '?'. # Production defect: a captured backtick 404'd the fetch and rendered n8n/immich as '?'.
import tempfile import tempfile
@@ -174,6 +196,25 @@ class TestRegistryUrls(unittest.TestCase):
self.assertEqual((urls, path), ([], None)) 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 ─────────────────────────────────────────────────────────────────── # ── C. --image argument parsing ───────────────────────────────────────────────────────────────────
class TestImageArgParsing(unittest.TestCase): class TestImageArgParsing(unittest.TestCase):
@@ -290,12 +331,32 @@ class TestPerImageWindows(unittest.TestCase):
self.assertEqual(rep["windows"]["github-advisories:discourse/discourse"], self.assertEqual(rep["windows"]["github-advisories:discourse/discourse"],
{"from": "3.5.3", "to": "3.5.4"}) {"from": "3.5.3", "to": "3.5.4"})
def test_unmatched_image_name_is_silently_ignored(self): def test_unmatched_image_name_makes_count_unknown(self):
# Documents CURRENT behaviour: a typo'd name costs coverage without warning. # A typo'd sidecar name must not silently turn its CVEs into a clean zero.
rep = run_scan([self.APP], v_from="3.5.3", v_to="3.5.4", rep = run_scan([self.APP], v_from="3.5.3", v_to="3.5.4",
images=[("nosuchimage", "1", "2")], urls=self.URLS[:1]) images=[("nosuchimage", "1", "2")], urls=self.URLS[:1])
self.assertEqual(list(rep["windows"]), ["github-advisories:discourse/discourse"]) self.assertEqual(list(rep["windows"]), ["github-advisories:discourse/discourse"])
self.assertFalse(rep["count_known"])
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.assertTrue(rep["count_known"])
self.assertEqual(rep.get("unmatched_image_windows"), None)
self.assertEqual(rep["cve_count_fixed"], 1)
class TestSchemeChangeDateFallback(unittest.TestCase): class TestSchemeChangeDateFallback(unittest.TestCase):
@@ -358,6 +419,15 @@ class TestCountGuarantees(unittest.TestCase):
self.assertTrue(rep["count_known"]) self.assertTrue(rep["count_known"])
self.assertIn("0 identified", A.markdown(rep)) 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): def test_404_advisory_feed_is_benign_not_a_failure(self):
rep = run_scan([gh("app/app", [adv("CVE-1", patched="1.1")]), rep = run_scan([gh("app/app", [adv("CVE-1", patched="1.1")]),
gh("side/car", [], status="no-advisories-published")], gh("side/car", [], status="no-advisories-published")],
@@ -377,6 +447,13 @@ class TestCountGuarantees(unittest.TestCase):
v_from="1.0", v_to="1.1") v_from="1.0", v_to="1.1")
self.assertIn("github-advisories:app/app", rep["sources_failed"]) self.assertIn("github-advisories:app/app", rep["sources_failed"])
self.assertIn("FAILED", A.markdown(rep)) self.assertIn("FAILED", A.markdown(rep))
self.assertFalse(rep["count_known"])
def test_only_404_feed_is_not_evidence_of_clean_scan(self):
rep = run_scan([gh("app/app", [], status="no-advisories-published")],
v_from="1.0", v_to="1.1")
self.assertTrue(rep["no_usable_sources"])
self.assertFalse(rep["count_known"])
def test_no_window_given_classifies_nothing(self): def test_no_window_given_classifies_nothing(self):
rep = run_scan([gh("app/app", [adv("CVE-1", patched="1.1")])]) rep = run_scan([gh("app/app", [adv("CVE-1", patched="1.1")])])
+55
View File
@@ -0,0 +1,55 @@
#!/usr/bin/env python3
"""Offline regressions for the weekly image survey's false-current cases."""
import importlib.util
import pathlib
import unittest
from unittest import mock
HERE = pathlib.Path(__file__).resolve().parent
spec = importlib.util.spec_from_file_location("resolve_images", HERE / "resolve-images.py")
resolver = importlib.util.module_from_spec(spec)
spec.loader.exec_module(resolver)
class TestResolverCoverage(unittest.TestCase):
def test_cryptpad_calver_tag_is_found_independently_of_abra(self):
with mock.patch.object(resolver, "list_tags", return_value=(
["version-2026.5.1", "version-2026.10.0", "opendesk-2026.08.04"], None)):
row = resolver.resolve("cryptpad/cryptpad:version-2026.5.1")
self.assertTrue(row["upgrade_available"])
self.assertEqual(row["newest_same_shape"], "version-2026.10.0")
def test_floating_and_templated_tags_cannot_prove_current(self):
with mock.patch.object(resolver, "list_tags", return_value=(["alpine", "1.2.3"], None)):
floating = resolver.resolve("atmoz/sftp:alpine")
templated = resolver.resolve("ghost:${IMAGE_VERSION}-alpine")
self.assertTrue(floating["status"].startswith("unverifiable:"))
self.assertTrue(templated["status"].startswith("skipped:"))
def test_truncated_registry_listing_cannot_prove_current(self):
with mock.patch.object(resolver, "list_tags", return_value=(["1.2.3"], None)):
with mock.patch.object(resolver, "_HUB_TRUNCATED", {"library/example"}):
row = resolver.resolve("example:1.2.3")
self.assertTrue(row["status"].startswith("unverifiable:"))
def test_huge_ghcr_listing_uses_only_published_release_manifests(self):
ref = "ghcr.io/immich-app/immich-machine-learning:v3.3.5"
with mock.patch.object(resolver, "list_tags", return_value=([], "RuntimeError: tag listing truncated")), \
mock.patch.object(resolver, "github_release_tags",
return_value=["v3.3.6", "v3.3.5"]), \
mock.patch.object(resolver, "_oci_tag_exists", side_effect=lambda h, r, t: t == "v3.3.5"):
row = resolver.resolve(ref)
self.assertEqual(row["status"], "ok")
self.assertFalse(row["upgrade_available"]) # release exists; image does not yet
self.assertIn("registry-manifest", row["source"])
with mock.patch.object(resolver, "list_tags", return_value=([], "RuntimeError: tag listing truncated")), \
mock.patch.object(resolver, "github_release_tags",
return_value=["v3.3.6", "v3.3.5"]), \
mock.patch.object(resolver, "_oci_tag_exists", return_value=True):
upgraded = resolver.resolve(ref)
self.assertTrue(upgraded["upgrade_available"])
if __name__ == "__main__":
unittest.main()
+4 -4
View File
@@ -5,11 +5,11 @@
| app | rustdesk/rustdesk-server | https://github.com/rustdesk/rustdesk-server | https://github.com/rustdesk/rustdesk-server/releases | | 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 | | 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 ## Standing notes
- **NOT enrolled** in the cc-ci weekly set (operator 2026-10-30: single client, maintenance contract - Enrolled in the cc-ci weekly set. The NVD CPE above is scoped to the open source server edition;
pending). Mirror `recipe-maintainers/rustdesk-server` exists + this registry entry so one-off do not count RustDesk Client or Server Pro CVEs for this recipe.
`/recipe-upgrade` runs and advisory scans work; if the client confirms, enroll via `/recipe-enroll`
(test suite + bridge + used-recipes weekly row).
- Both services (`app` = hbbs, `signal` = hbbr) use the SAME image — always bump them together. - 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 - 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`; GitHub release) — an aborted/alias push, NOT a real version line. The real line is `1.1.x`;