advisory-scan: tests, audit, and two real undercounts they found

Adds test-advisory-scan.py (58 offline tests on fixtures + 6 live regressions
against the week-2026-08-07 report) and audit-advisory-scan.py, which re-derives
every count with a SEPARATE semver implementation and its own release fetch and
diffs against the scanner. Both found real defects:

1. Window membership was compared on ragged tuples, so (18,) < (18,0) — a CVE
   patched in 18.0 fell OUTSIDE a window ending at 18. Bare major tags are the
   norm for sidecars (postgres:18, redis:8-alpine). Now zero-padded, which also
   keeps the upper bound conservative (18.5 stays out of a window ending at 18).

2. Advisories with no knowable fix version were silently counted as 'not fixed'.
   Twelve redis advisories say patched_versions 'TBD' or '7.4.X' with an
   open-ended range — six of them high severity. They are now INDETERMINATE:
   not counted, not dismissed, and surfaced in the output.

   All twelve turned out to be genuinely fixed: redis names each in the release
   notes of every branch that got the fix (CVE-2025-32023 -> 6.2.19, 7.2.10,
   7.4.5, 8.0.3, 8.2.0). So a third deterministic method resolves them from
   release notes, with the naming tags recorded as the citation. discourse's
   redis contribution goes 5 -> 17, and its total 128 -> 140.

Pass 2 (--adjudicate) is the model-judged stage for what arithmetic cannot
settle: it hands over each open case's full evidence, plus every verdict pass 1
reached, and takes FIXED/NOT-FIXED/STILL-UNKNOWN with a reason citing that
evidence. It may only raise a count. Vendor-page-only CVEs — the shape of both
gitea CVSS-9.8 RCEs — now reach it instead of being dropped.

Tests cover pass 1 only, by design; pass 2's judgement is a model's. What is
tested there is deterministic: which cases it selects, and that truncation is
announced rather than silent.

SPEC.md rewritten around the two passes.
This commit is contained in:
autonomic-bot
2026-08-11 01:29:11 +00:00
parent 46c4fff1a6
commit 44cb9b6704
6 changed files with 1288 additions and 57 deletions
+3
View File
@@ -52,6 +52,9 @@ keeps every weekly edition looking the same regardless of which model writes the
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
(discourse 3.5.3 → 2026.7.1) by falling back to advisory publish dates and reports a real number.
- **A count with undetermined advisories is a FLOOR.** If the scan block says N advisories
"could NOT be judged", report the number but say in the notes that it is a floor — those
advisories are neither fixed nor safe, they are unmeasured. Never round them away.
- **Counts span every image, each judged by its own window.** A scan block lists one line per
image with its version range and classification method; the headline is their union. So a
recipe's count legitimately includes **sidecar** CVEs (discourse's 128 = 123 app + 5 redis).
+10 -3
View File
@@ -176,9 +176,16 @@ e.g. discourse moving app 3.5.3→2026.7.1 *and* redis 7.4→8.10:
... --from 3.5.3 --to 2026.7.1 --image redis=7.4:8.10
```
→ 128 CVEs (123 app + 5 redis), where the redis five include a **critical** (CVE-2025-49844) that is
invisible if the sidecar is left out. `<name>` is substring-matched against source repo names, so make
it specific enough to hit exactly one.
→ 140 CVEs (123 app + 17 redis), where the redis seventeen include a **critical** (CVE-2025-49844)
that is invisible if the sidecar is left out. `<name>` is substring-matched against source repo names,
so make it specific enough to hit exactly one.
**If the scan reports advisories it could NOT judge, re-run it with `--adjudicate`.** That is a second
pass: it collects each open case's full evidence (advisory prose, references, affected ranges, every
release naming the CVE) and asks YOU to decide FIXED / NOT-FIXED / STILL-UNKNOWN with a reason citing
that evidence. The deterministic number is a **floor** — add every FIXED to the count. Say
STILL-UNKNOWN rather than inferring from memory, and never record an undecided CVE as unaffected.
It also shows what pass 1 already decided; if a verdict looks wrong given its evidence, say so.
It queries, per recipe: the **GitHub Security Advisories API** for every source repo in
`cc-ci-plan/upstream/<recipe>.md` (CVE + GHSA + severity + vulnerable/patched ranges, so
+159 -42
View File
@@ -1,12 +1,14 @@
# Advisory scan — specification
What `cc-ci-plan/advisory-scan.py` does, step by step, and why each step exists. This documents the
implementation as it stands (2026-08-10); if you change the code, change this file in the same commit.
implementation as it stands (2026-08-11); if you change the code, change this file in the same commit.
`cc-ci-plan/test-advisory-scan.py` is the executable half of this spec — every rule below is asserted
there.
**Role.** A deterministic, per-recipe CVE detector run as a **pre-step of `/recipe-upgrade`** (step 2a).
It is **strictly additive**: it never replaces the release-note reading the upgrade agent already does.
The CVE count reported for a recipe is the **union** of what the agent read and what this scan found;
the scan may never *lower* a count established by reading.
**Role.** A per-recipe CVE detector run as a **pre-step of `/recipe-upgrade`** (step 2a). It is
**strictly additive**: it never replaces the release-note reading the upgrade agent already does. The
CVE count reported for a recipe is the **union** of what the agent read and what this scan found; the
scan may never *lower* a count established by reading.
**Why it exists.** gitea 1.27.1 fixed CVE-2026-60004 and CVE-2026-59774 (both CVSS 9.8). The weekly
report printed gitea's CVE count as `1`, then `none`. The upgrade agent had read the GitHub *release
@@ -16,18 +18,35 @@ Nothing in the pipeline queried an advisory source. This scan closes that hole.
---
## Two passes
| | Pass 1 — measure | Pass 2 — judge (`--adjudicate`) |
|---|---|---|
| Who | Pure Python, no model | The calling agent, a model |
| Does | Collects evidence and decides every case it can by arithmetic | Weighs the collected evidence on cases arithmetic cannot settle |
| Output | A count, or `UNKNOWN` | FIXED / NOT-FIXED / STILL-UNKNOWN per open case |
| Rule | Deterministic and reproducible | May only **raise** the count, never lower it |
**Prefer pass 1.** Every case pass 1 decides is one that reproduces identically next week. Pass 2 exists
for evidence that is prose rather than data — a fallback, not a co-equal stage. When a class of case
keeps landing in pass 2, the fix is a new deterministic method in pass 1. §4c is exactly that: it moved
12 redis advisories out of pass 2 and into arithmetic.
---
## Inputs
```
advisory-scan.py <recipe> [--from <version>] [--to <version>]
[--image <name>=<from>:<to>]... [--json] [--registry DIR]
[--image <name>=<from>:<to>]... [--adjudicate] [--json] [--registry DIR]
```
| Input | Meaning |
|---|---|
| `<recipe>` | Recipe name; selects `cc-ci-plan/upstream/<recipe>.md` (the per-recipe URL registry) |
| `--from` / `--to` | The **primary app image's** version window being upgraded across |
| `--image NAME=FROM:TO` | A **sidecar image and the versions it moved between** (repeatable). `NAME` is matched as a substring against source repo names, e.g. `--image redis=7.4:8.10`. Malformed values are warned about on stderr and skipped. Without it that image's advisories stay unclassified. |
| `--image NAME=FROM:TO` | A **sidecar image and the versions it moved between** (repeatable, all in ONE call). `NAME` is substring-matched against source repo names. Malformed values warn on stderr and are skipped. Without it that image's advisories stay unclassified. |
| `--adjudicate` | Run pass 2: append the evidence dossier for judgement |
| `--registry` | Registry dir; also `CCCI_UPSTREAM_REGISTRY` |
| `GITHUB_TOKEN` / `GITHUB_TOKEN_FILE` | Read-only token; **rate limit only** (60/hr anonymous → 5000/hr). Default file `/srv/cc-ci/.github-token`, mode 600. Public advisories need **no scopes**. |
@@ -35,6 +54,8 @@ Exit code is always 0 — this is informational. Failures are *reported*, never
---
# Pass 1 — deterministic
## Step 1 — Collect source URLs from the registry
Read `cc-ci-plan/upstream/<recipe>.md` and extract every `http(s)://…` URL.
@@ -59,8 +80,8 @@ with *"not checked"*.
For every `github.com/<owner>/<repo>` URL in the registry:
`GET /repos/<owner>/<repo>/security-advisories`.
Captured per advisory: `cve_id`, `ghsa_id`, `severity`, `summary`, `published_at`, and **all**
`vulnerabilities[]` entries' `vulnerable_version_range` + `patched_versions` (joined with `;`).
Captured per advisory: `cve_id`, `ghsa_id`, `severity`, `summary`, **`description`**, `published_at`,
and **all** `vulnerabilities[]` entries' `vulnerable_version_range` + `patched_versions` (joined `;`).
- **All entries, not just the first.** An advisory carries one entry **per patched release line** —
n8n patches three (1.123.32, 2.17.4, 2.18.1). Reading only `vulnerabilities[0]` silently dropped the
@@ -68,12 +89,11 @@ Captured per advisory: `cve_id`, `ghsa_id`, `severity`, `summary`, `published_at
- **Pagination via the `Link rel="next"` cursor**, to exhaustion (cap 20 hops). This endpoint returns
at most 100 rows **and ignores `?page=`** — it re-returns the same rows, which silently truncates busy
projects. discourse has 286 advisories; a single page cannot even cover one upgrade window.
- **The description is kept from this response.** It is already present here, and pass 2 needs the
prose; re-fetching it per advisory would cost one request each.
- **HTTP 404 ⇒ `no-advisories-published`** — a benign absence (many sidecar images publish none), **not**
a failure. Conflating the two pushed nearly every recipe to `?` and destroyed the signal.
This source is primary because it carries **severity and version ranges**, making "fixed by *this*
upgrade" computable rather than guessed.
### 2b. Vendor release / security pages
Every other registry URL is fetched, HTML-stripped, and scanned for `CVE-\d{4}-\d{4,7}`, keeping ±160
@@ -83,7 +103,8 @@ URLs containing `<`, `>`, `{`, `}`, `VERSION`, or `vX.Y.Z` are **skipped as temp
human documentation (`…/changelog/v<VERSION>/`), not fetchable, and counting them as failures is wrong.
This is the source that would have caught gitea: the vendor blog names both CVEs, the GitHub release
page names neither.
page names neither. A CVE found **only** here carries no version data, so pass 1 cannot place it — it
goes to pass 2 (§6).
### 2c. OSV.dev — supplementary
@@ -96,7 +117,7 @@ Only when the recipe has an entry in `OSV_PACKAGES` (ecosystem + package) and a
## Step 3 — Union
All findings merge into one CVE map: id → `{sources[], severity, ghsa, vulnerable_range, patched,
published_at, context}`. A CVE seen by several sources keeps them all.
description, published_at, url, cvss, context}`. A CVE seen by several sources keeps them all.
## Step 4 — Classify against the upgrade window
@@ -104,25 +125,33 @@ Two invariants govern this step, both learned from a wrong answer in production.
> **A. Every image is judged by its OWN versions.** The app repo uses `--from/--to`; each sidecar uses
> its own `--image NAME=FROM:TO`. **Pass them all in ONE invocation** — the count is a union across
> images, and the UNKNOWN guarantee in B only holds when a single run sees every one. An image with no window is **not** classified — its advisories are
> listed as unclassified so they stay visible without inflating the count. The reported count is the
> **union across windows**, and each window is classified independently (so one may use version
> ranges while another falls back to dates).
> images, and the UNKNOWN guarantee in B only holds when a single run sees every one. An image with no
> window is **not** classified; its advisories are listed as unclassified so they stay visible without
> inflating the count. Each window is classified independently, so one may use version ranges while
> another falls back to dates.
> *Why:* discourse once reported **133**, of which **34 were redis CVEs** — including
> `CVE-2021-21309`, patched in redis 6.0.11 in 2021 — counted purely because 6.0.11 sits numerically
> inside discourse's `3.5.3 → 2026.7.1` range. The fix is not to ignore sidecars but to give each one
> the versions it actually moved through: with `--image redis=7.4:8.10`, discourse scores
> **128 = 123 (app, by date) + 5 (redis, by version range)** and the redis five include
> `CVE-2025-49844`, **critical**, which was invisible while sidecars went uncounted.
> **140 = 123 (app, by date) + 17 (redis)**, and the redis seventeen include `CVE-2025-49844`,
> **critical**, invisible while sidecars went uncounted.
>
> **B. Never emit a number you cannot justify.** If neither method below can order the window, the
> count is `null` / `UNKNOWN`, never `0`. A `0` in a security column asserts safety.
> **B. Never emit a number you cannot justify.** If no method can order a window, the count is
> `null` / `UNKNOWN`, never `0`. A `0` in a security column asserts safety. Equally, an advisory that
> cannot be judged is **indeterminate** (§4d) — never silently counted as "not fixed".
### 4a. By patched version (preferred — exact)
`patched_versions` is a **range expression** (`">= 2.18.1"`), possibly several joined by `;`. Extract
every version-looking token; the advisory is **fixed-by-this-upgrade** if **any** patched version `p`
satisfies `from < p <= to`, using a loose numeric key (leading integers per dot-part).
satisfies `from < p <= to` — exclusive lower (a fix already in the version you were on is not this
upgrade's doing), inclusive upper.
**Comparison is zero-padded to equal length**, so `18` == `18.0` == `18.0.0` as semver means it.
Without padding, plain tuple order makes `(18,) < (18,0)`, i.e. a fix in 18.0 falls *outside* a window
ending at 18 — and bare major tags are the norm for sidecars (`postgres:18`, `redis:8-alpine`). Padding
is permissive at the lower bound and conservative at the upper: with `to = 18`, a fix in `18.5` is
**not** counted, because nothing proves which 18.x a floating tag resolved to.
### 4b. By advisory publish date (fallback — temporal)
@@ -132,22 +161,50 @@ component jumping by ≥ `SCHEME_JUMP` (100) — e.g. semver `3.5.3` → calver
Version strings are unorderable across such a jump (`2025.12.2` compares "newer" than `3.5.3` while
shipping earlier), but **release dates always order**. So:
1. Resolve `--from` and `--to` to **git tag dates** on the primary repo (tries `v<version>` then
1. Resolve `--from` and `--to` to **git tag dates** on that source (tries `v<version>` then
`<version>`; annotated tag → tagger date, else commit date).
2. An advisory counts as fixed when `date_from < published_at <= date_to`.
2. An advisory counts as fixed when `date_from < published_at <= date_to` — the same exclusive/inclusive
boundaries as 4a, so the two methods agree at the edges.
Both the method and the resolved window appear in the output. This reproduces, automatically, the hand
count that established discourse `3.5.3` (2025-12-30) → `2026.7.1` (2026-07-31) = **123 CVEs**.
This reproduces, automatically, the hand count that established discourse `3.5.3` (2025-12-30) →
`2026.7.1` (2026-07-31) = **123 CVEs**.
*Assumption:* the vendor publishes advisories at fix time (true for discourse). The count includes
**first-party plugin advisories** where the vendor files them on the same repo — which is why a
plugin-rich project scores far higher than a monolith, not a statement about relative security.
### 4c. Otherwise
### 4c. By release notes naming the CVE (rescue — for advisories with no fix version)
`count_known = false`, `cve_count_fixed = null`, and the markdown headline reads
**"CVEs fixed by this upgrade: UNKNOWN — the scan could NOT determine a count"** with an explicit
*"This is NOT zero"*.
Applied to advisories 4a/4b could not decide, **before** giving up on them. Fetch the source repo's
GitHub **releases** (cached per repo, 4 pages) and find every tag whose notes **name the CVE id**. If
any such tag falls inside the window by 4a's rule, the advisory is fixed by this upgrade, and the
naming tags are recorded in `fix_versions_from_release_notes` as the citation.
> **Why this is not optional.** Vendors routinely publish an advisory with `patched_versions: "TBD"`
> and then name the CVE in the release notes of every branch that got the fix. **All 12** redis
> advisories crossed by discourse's redis bump are exactly this shape — `TBD`, or a placeholder like
> `7.4.X`, with an open-ended `vulnerable_version_range` (`All`, `>= 7.0.0`) — yet each is named in
> concrete releases (`CVE-2025-32023` → 6.2.19, 7.2.10, 7.4.5, 8.0.3, 8.2.0). Without this method
> discourse's redis contribution reads 5; with it, 17. Six of the twelve are high severity.
### 4d. Otherwise — indeterminate, not excluded
An advisory from a **windowed** source that none of 4a4c could decide — no usable `patched_versions`,
an open-ended vulnerable range, and no release note naming it — is recorded as **indeterminate**. It is:
- **not** added to the count (nothing justifies counting it), and
- **not** treated as unaffected (nothing justifies dismissing it either).
It is listed prominently, the headline reads *"(at least — see undetermined below)"*, and it becomes an
input to pass 2. Silently excluding these is the same defect class as printing `0` for an unscanned
recipe, one level down.
### 4e. Unorderable window
If neither 4a nor 4b can order a window at all, `count_known = false`, `cve_count_fixed = null`, and the
headline reads **"CVEs fixed by this upgrade: UNKNOWN — the scan could NOT determine a count"** with an
explicit *"This is NOT zero"*. If **any** requested window is unorderable the whole count is suppressed;
a partial number would understate a security figure.
## Step 5 — Output
@@ -155,16 +212,56 @@ Markdown (default) for pasting into the per-recipe upgrade log, or `--json`.
| Field | Meaning |
|---|---|
| `cve_count_fixed` | Union across all windows, or **`null`** if ANY requested window could not be ordered (a partial number would understate) |
| `cve_count_fixed` | Union across all windows, or **`null`** if any window was unorderable |
| `count_known` | Distinguishes "counted zero" from "could not count" |
| `windows` | Every source classified, with its from/to (the internal computed ranges) |
| `classified_by` | **Per source**: `patched version ranges` or `advisory publish date (version scheme changed)` |
| `cve_count_indeterminate` / `indeterminate[]` | Judged by nothing; a **floor marker** on the count |
| `resolved_by_release_notes` | CVE → the tags that named it (4c citations) |
| `windows` | Every source classified, with its from/to |
| `classified_by` | **Per source**: which method decided it |
| `date_window` | **Per source**, when 4b was used |
| `fixed_by_this_upgrade[]` | CVE ids, with severity / GHSA / fixed-in per id |
| `unclassified[]` | Seen but not attributable to this window (incl. other images) |
| `sources[]` | Every source with its own status |
| `sources_failed[]` | **Genuine** failures only |
| `sources_benign[]` | `no-advisories-published`, `skipped: template URL` |
| `unclassified[]` | Seen but not attributable (other images, or vendor-page-only) |
| `sources[]` / `sources_failed[]` / `sources_benign[]` | Per-source status; only genuine failures in `failed` |
---
# Pass 2 — adjudication (`--adjudicate`)
## Step 6 — What gets judged
Two kinds of open case, both real gaps rather than noise:
1. **Indeterminate** (§4d) — from an image *with* a window, but no fix version is knowable anywhere.
2. **Vendor-page-only** — a CVE seen only on a vendor security page, with no structured advisory
behind it. **gitea's two CVSS-9.8 RCEs are this shape.** They carry no version data, so no
arithmetic can place them, but the page prose usually states the fixed release.
## Step 7 — The evidence dossier
Pass 1 collects; pass 2 judges. Nothing in the dossier interprets — it assembles what was *measured*,
so the judgement is made against evidence rather than recollection. That distinction is the whole
point: the original failure was a report leaning on model knowledge that predated the CVEs, with no
source queried at all.
Per open case: severity, CVSS, sources, its window, why it is undecided, `patched_versions` and
`vulnerable_version_range` **as published**, summary, full description, references, affected ranges
with `first_patched_version`, and **every release tag naming the CVE** — whether or not in window, since
the model may reason about branch lines the arithmetic deliberately will not.
**Pass 2 also sees every decision pass 1 made** — a compact table of counted and excluded advisories
with the evidence behind each verdict. A deterministic verdict can still be wrong (a mis-parsed range,
a release note that mentions a CVE without fixing it), and only a reader with the evidence in front of
it can catch that. Silence means agreement.
## Step 8 — The verdict contract
For each open case: **FIXED** / **NOT-FIXED** / **STILL-UNKNOWN**, each with a one-line reason
**citing the evidence shown**. Every FIXED is added to the recipe's count — pass 1's number is a floor,
not a total. If the evidence does not settle it, **STILL-UNKNOWN**: do not infer from memory of the
project, and never record an undecided CVE as unaffected.
**No silent caps.** `MAX_ADJUDICATE` (25) and `MAX_REVIEW_ROWS` (400) bound the output; whenever either
truncates, the block says how many were dropped and that the unshown remain undetermined.
---
@@ -176,21 +273,41 @@ Markdown (default) for pasting into the per-recipe upgrade log, or `--json`.
- **failed sources**, or `UNKNOWN` → **`?`**, never `none` — a blank reads as "clean", which is exactly
how two CVSS-9.8 gitea RCEs were published as `none`;
- **no upgrade this run** → `0`, not `?` — nothing an upgrade could have fixed;
- a count with **indeterminate advisories** → the number is a floor; say so in the notes;
- benign notes → never `?`.
`?` must stay **rare**: it means *we tried and could not tell*, not *we did not look*. A rash of `?` is
a bug to raise in the report's Addendum, not a normal outcome — every instance so far traced to a defect
in this tool or stale registry data.
## Testing
`test-advisory-scan.py` — 58 offline tests (fixtures, no network) plus 6 live regressions against the
counts published in week-2026-08-07. **The offline tier covers pass 1 only, by design**: pass 2's
judgement is a model's and cannot be asserted deterministically. What *is* tested about pass 2 is the
part that stays deterministic — which cases it selects, and that truncation is always announced.
```
python3 test-advisory-scan.py # offline
python3 test-advisory-scan.py --live # + historic report numbers
```
`audit-advisory-scan.py` re-derives the counts with a **separate** semver implementation and its own
release fetch, then diffs against the scanner. Run it after changing classification; it is what caught
the 12 undercounted redis CVEs.
## Known limits
1. **Versions must be supplied per image.** An image with no `--image` is not counted — the scan will not
guess a version range it was not told. `/recipe-upgrade` passes one per image it bumped.
1. **Versions must be supplied per image.** An image with no `--image` is not counted — the scan will
not guess a version range it was not given. `/recipe-upgrade` passes one per image it bumped.
2. **Date-based counts are temporal**, not exact — they assume publish-at-fix-time.
3. **Registry-bound.** Unlisted vendor security pages are invisible; the scan cannot know what it was
never pointed at.
4. **`NAME` matching is substring-against-source-name**, so a short or generic name can attach to more
than one repo (`postgres` matches `discourse/discourse-postgres`). The primary source is claimed
first and cannot be stolen; be specific enough to hit one repo.
5. **Rate limit** without a token is 60/hr — a full weekly sweep will exhaust it and degrade to failed
first and cannot be stolen. A name matching **nothing** is silently ignored — a typo costs coverage
without warning.
5. **Release-note rescue (4c) trusts that naming implies fixing.** A release note that merely mentions a
CVE would be read as fixing it. Pass 2's review table exists partly to catch this.
6. **Rate limit** without a token is 60/hr — a full weekly sweep will exhaust it and degrade to failed
sources (visibly, but degraded).
+342 -12
View File
@@ -53,6 +53,12 @@ TIMEOUT = int(os.environ.get("ADVISORY_SCAN_TIMEOUT", "45"))
CVE_RE = re.compile(r"CVE-\d{4}-\d{4,7}")
# Leading-version-component jump that means the scheme changed (semver → calver).
SCHEME_JUMP = 100
# A patched_versions field that names no usable version. GitHub carries these verbatim from the
# vendor: redis publishes "TBD" for 11 advisories and "6.2.X, 7.2.X, 7.4.X" for another, and their
# vulnerable_version_range is open-ended ("All", ">= 7.0.0"), so the fix version is NOT recoverable.
# Such an advisory must be reported as INDETERMINATE, never silently counted as "not fixed" — that
# would assert an upgrade did not fix something we simply cannot judge.
PLACEHOLDER_RE = re.compile(r"\bTBD\b|\bunknown\b|\bnone\b|\d+\.[Xx]\b|\?", re.I)
# Optional OSV mappings: recipe -> (ecosystem, package). Supplementary only (see module docstring).
OSV_PACKAGES: dict[str, tuple[str, str]] = {
@@ -158,6 +164,21 @@ def _vkey(v: str | None) -> tuple:
return tuple(out)
def _within(kf: tuple, kt: tuple, c: tuple) -> bool:
"""Is patched-version `c` inside the window (kf, kt] — exclusive lower, inclusive upper?
Compares ZERO-PADDED to equal length, so "18" == "18.0" == "18.0.0" the way semver means it.
Without the padding, plain tuple order says (18,) < (18,0), i.e. a CVE patched in 18.0 falls
OUTSIDE a window ending at 18 — and bare major tags are the norm for sidecars (postgres:18,
redis:8-alpine), so that silently dropped real fixes. Padding also keeps the upper bound
conservative: a fix in 18.5 is still outside a window ending at "18", because nothing proves
which 18.x a floating tag resolved to.
"""
n = max(len(kf), len(kt), len(c))
pad = lambda t: t + (0,) * (n - len(t))
return pad(kf) < pad(c) <= pad(kt)
def registry_urls(recipe: str, registry_dir: str) -> tuple[list[str], str | None]:
path = os.path.join(registry_dir, f"{recipe}.md")
try:
@@ -219,6 +240,10 @@ def github_advisories(urls: list[str]) -> list[dict]:
) or None,
"url": a.get("html_url"),
"published_at": a.get("published_at"),
# The list response ALREADY carries the prose. Keep it: the adjudication
# pass needs it, and re-fetching per advisory costs a request each.
"description": (a.get("description") or "")[:4000],
"cvss": ((a.get("cvss") or {}).get("vector_string")),
}
)
except urllib.error.HTTPError as e:
@@ -283,6 +308,244 @@ def osv(recipe: str, version: str | None) -> dict | None:
return entry
_RELEASE_CACHE: dict[str, list[tuple[str, str]]] = {}
def _releases(owner: str, repo: str, max_pages: int = 4) -> list[tuple[str, str]]:
"""[(tag, body)] for a repo's GitHub releases, cached per repo for the process."""
key = f"{owner}/{repo}"
if key in _RELEASE_CACHE:
return _RELEASE_CACHE[key]
hdrs = {"Accept": "application/vnd.github+json"}
tok = _github_token()
if tok:
hdrs["Authorization"] = f"Bearer {tok}"
out: list[tuple[str, str]] = []
try:
for rel in _gh_paginate(
f"https://api.github.com/repos/{owner}/{repo}/releases?per_page=100", hdrs, max_pages
):
out.append((rel.get("tag_name") or "",
f"{rel.get('name') or ''}\n{rel.get('body') or ''}"))
except Exception: # noqa: BLE001 — best effort; absence just leaves advisories undetermined
pass
_RELEASE_CACHE[key] = out
return out
def release_fix_versions(source: str, cve: str) -> list[str]:
"""Release tags whose notes NAME this CVE — a deterministic fix version when the advisory has none.
Vendors routinely publish an advisory with `patched_versions: "TBD"` and then name the CVE in the
release notes of every branch that got the fix. redis does exactly this: all 12 of its advisories
that discourse's redis bump crosses carry TBD, yet each is named in concrete releases
(CVE-2025-32023 → 6.2.19, 7.2.10, 7.4.5, 8.0.3, 8.2.0). Ignoring that evidence undercounted
discourse by 12 CVEs, so this is checked BEFORE giving up on an advisory.
"""
if not source.startswith("github-advisories:"):
return []
owner, _, repo = source.split(":", 1)[1].partition("/")
return [tag for tag, body in _releases(owner, repo) if cve in body]
def advisory_text(ghsa: str, source: str | None = None) -> dict:
"""Full text of one advisory, for the ADJUDICATION pass (see adjudication_block).
The structured `patched_versions` field is often "TBD" while the prose description and the
linked references DO state where the fix landed. That prose is not machine-parseable in general
— which is the point: it is collected here for a MODEL to judge, not for a regex."""
hdrs = {"Accept": "application/vnd.github+json"}
tok = _github_token()
if tok:
hdrs["Authorization"] = f"Bearer {tok}"
out = {"ghsa": ghsa, "status": "ok"}
# Repo-scoped FIRST. Many repository advisories are never mirrored into the global GitHub
# Advisory Database, so /advisories/<ghsa> 404s for them (all 12 redis ones, for instance)
# while /repos/<owner>/<repo>/security-advisories/<ghsa> returns the full record.
cands = []
if source and source.startswith("github-advisories:"):
cands.append(f"https://api.github.com/repos/{source.split(':',1)[1]}/security-advisories/{ghsa}")
cands.append(f"https://api.github.com/advisories/{ghsa}")
try:
a, last = None, None
for u in cands:
try:
a = json.loads(_fetch(u, hdrs)); break
except Exception as ex: # noqa: BLE001 — try the next endpoint
last = ex
if a is None:
raise last or RuntimeError("no advisory endpoint responded")
out.update({
"summary": a.get("summary"),
"description": (a.get("description") or "")[:4000],
"severity": a.get("severity"),
"published_at": a.get("published_at"),
"references": [r for r in (a.get("references") or [])][:12] or (
[a.get("html_url")] if a.get("html_url") else []),
"cvss": (a.get("cvss") or {}).get("vector_string"),
"vulnerabilities": [
{"package": (v.get("package") or {}).get("name"),
"vulnerable_version_range": v.get("vulnerable_version_range"),
"first_patched_version": v.get("first_patched_version")}
for v in (a.get("vulnerabilities") or [])
],
})
except Exception as e: # noqa: BLE001
out["status"] = f"error: {type(e).__name__}: {e}"
return out
MAX_ADJUDICATE = int(os.environ.get("ADVISORY_SCAN_MAX_ADJUDICATE", "25"))
# Compact review rows for advisories pass 1 DID decide. Pass 2 sees these too, so a wrong
# deterministic verdict can be caught rather than inherited.
MAX_REVIEW_ROWS = int(os.environ.get("ADVISORY_SCAN_MAX_REVIEW", "400"))
def needs_judgement(rep: dict) -> list[str]:
"""CVEs the deterministic pass could not decide — the input set for the adjudication pass.
Two kinds, both real gaps rather than noise:
1. INDETERMINATE — from an image WITH a window, but no fix version is knowable (advisory says
`TBD`/`7.4.X`, range is open-ended, and no release note names it).
2. VENDOR-PAGE-ONLY — a CVE seen only on a vendor security page, with no structured advisory
behind it at all. gitea's two CVSS-9.8 RCEs are this shape. They carry no version data, so
no arithmetic can place them, but the page's prose usually states the fixed release.
"""
windows = rep.get("windows") or {}
out = list(rep.get("indeterminate") or [])
for cve in rep.get("unclassified") or []:
srcs = rep["cves"][cve]["sources"]
if not any(s.startswith("github-advisories:") for s in srcs) and cve not in out:
out.append(cve)
return sorted(out)
def evidence_bundle(rep: dict, cve: str) -> dict:
"""EVERY deterministic signal held about one CVE, gathered for a model to weigh.
Pass 1 collects; pass 2 judges. Nothing here interprets — it assembles what was measured, so the
judgement is made against evidence rather than recollection (the exact failure that let two
CVSS-9.8 gitea RCEs be published as "none": the report leaned on model knowledge that predated
them, and no source had been queried at all).
"""
e = rep["cves"][cve]
src = e["sources"][0]
windows = rep.get("windows") or {}
win = next((windows[s] for s in e["sources"] if s in windows), None)
ev = {
"cve": cve,
"severity": e.get("severity"),
"cvss": e.get("cvss"),
"sources": e["sources"],
"window": win,
"why_undecided": ("no fix version published and no release note names it"
if cve in (rep.get("indeterminate") or [])
else "seen only on a vendor page — no structured advisory, no version data"),
"patched_as_published": e.get("patched"),
"vulnerable_range_as_published": e.get("vulnerable_range"),
"summary": e.get("context"),
"description": e.get("description"),
"advisory_url": e.get("url"),
# Release tags NAMING this CVE, whether or not they fall in the window — the model may
# reason about branch lines the arithmetic deliberately would not.
"releases_naming_it": release_fix_versions(src, cve) if src.startswith("github-advisories:") else [],
"references": [],
}
if e.get("ghsa"):
t = advisory_text(e["ghsa"], src)
if t.get("status") == "ok":
ev["references"] = t.get("references") or []
ev["description"] = t.get("description") or ev["description"]
ev["summary"] = t.get("summary") or ev["summary"]
ev["affected"] = t.get("vulnerabilities") or []
else:
ev["detail_fetch"] = t.get("status")
return ev
def adjudication_block(rep: dict) -> str:
"""SECOND PASS: present the collected evidence and ask for a judgement on each open case.
This block decides nothing. The deterministic count stands as a FLOOR; a verdict here may only
ADD to it, matching the rule that this scan raises a count on evidence but never lowers one.
"""
todo = needs_judgement(rep)
if not todo:
return ""
shown, dropped = todo[:MAX_ADJUDICATE], max(0, len(todo) - MAX_ADJUDICATE)
L = ["", "---", "",
f"## Adjudication pass — {len(todo)} advisory/advisories need judgement", "",
"Pass 1 collected the evidence below deterministically and could NOT decide these cases. "
"Weigh the evidence and decide each one.", "",
"**For each: did the version move in its window fix it?** Answer **FIXED** / **NOT-FIXED** / "
"**STILL-UNKNOWN**, each with a one-line reason **citing the evidence shown** — a fixed "
"release named in the text, a branch line, an affected range. Add every FIXED to the "
"recipe's CVE count; the deterministic number is a floor, not a total. If the evidence does "
"not settle it, say STILL-UNKNOWN: do NOT infer from memory of the project, and never "
"record an undecided CVE as unaffected.", ""]
for src, win in (rep.get("windows") or {}).items():
L.append(f"- window: `{src.split(':',1)[-1]}` {win['from']}{win['to']}")
if dropped:
L += ["", f"⚠ Showing the first {MAX_ADJUDICATE} of {len(todo)}; **{dropped} not shown** "
f"(raise ADVISORY_SCAN_MAX_ADJUDICATE). The unshown remain undetermined — do not "
f"treat them as absent."]
L.append("")
for cve in shown:
ev = evidence_bundle(rep, cve)
L.append(f"### {cve}{ev['severity'] or '?'}")
L.append(f"- undecided because: {ev['why_undecided']}")
L.append(f"- source: `{ev['sources'][0]}`"
+ (f" · window {ev['window']['from']}{ev['window']['to']}" if ev["window"] else
" · **no version window** for this image"))
L.append(f"- patched_versions as published: `{ev['patched_as_published']}`")
L.append(f"- vulnerable_range as published: `{ev['vulnerable_range_as_published']}`")
if ev["releases_naming_it"]:
L.append(f"- **releases naming this CVE**: {', '.join(ev['releases_naming_it'][:14])}")
for v in ev.get("affected") or []:
L.append(f"- affects `{v.get('package')}` {v.get('vulnerable_version_range')}"
f"first_patched_version: {v.get('first_patched_version')}")
if ev["references"]:
L.append(f"- references: {', '.join(r.strip() for r in ev['references'][:6])}")
if ev.get("detail_fetch"):
L.append(f"- ⚠ detail fetch failed: {ev['detail_fetch']} (evidence below is from pass 1)")
if ev["summary"]:
L += ["", f"> {ev['summary']}"]
if ev["description"]:
L += ["", "```", (ev["description"] or "").strip()[:2000], "```"]
L.append("")
# ── everything pass 1 DID decide, with the evidence behind each verdict ──────────────────────
# Pass 2 must see the whole picture, not only the leftovers: a deterministic verdict can still
# be wrong (a mis-parsed range, a release note that names a CVE without fixing it), and only a
# reader with the evidence in front of it can catch that.
decided = []
for cve in rep.get("fixed_by_this_upgrade") or []:
e = rep["cves"][cve]
decided.append((cve, "COUNTED", e))
for cve, e in sorted(rep.get("cves", {}).items()):
if e.get("classification") == "outside-window":
decided.append((cve, "excluded (outside window)", e))
if decided:
shown_rows, dropped_rows = decided[:MAX_REVIEW_ROWS], max(0, len(decided) - MAX_REVIEW_ROWS)
L += ["---", "",
f"## Pass 1 decisions — {len(decided)} already judged deterministically", "",
"Review these too. If any verdict looks wrong given its evidence, say so and explain; "
"a correction here changes the count. Silence means you agree.", ""]
if dropped_rows:
L.append(f"⚠ Showing {MAX_REVIEW_ROWS} of {len(decided)}; **{dropped_rows} not shown** "
f"(raise ADVISORY_SCAN_MAX_REVIEW).")
L.append("")
L += ["| CVE | verdict | severity | patched as published | releases naming it | source |",
"|---|---|---|---|---|---|"]
for cve, verdict, e in shown_rows:
rel = e.get("fix_versions_from_release_notes") or []
L.append(f"| {cve} | {verdict} | {e.get('severity') or '?'} | "
f"{(e.get('patched') or '')[:60]} | {', '.join(rel[:5]) or ''} | "
f"{e['sources'][0].split(':',1)[-1]} |")
L.append("")
return "\n".join(L)
def scan(recipe: str, v_from: str | None, v_to: str | None, registry_dir: str,
images: list[tuple[str, str, str]] | None = None) -> dict:
urls, reg_path = registry_urls(recipe, registry_dir)
@@ -303,7 +566,8 @@ def scan(recipe: str, v_from: str | None, v_to: str | None, registry_dir: str,
def record(cve: str, src: str, **extra):
e = report["cves"].setdefault(cve, {"sources": [], "severity": None, "ghsa": None,
"vulnerable_range": None, "patched": None,
"context": None, "published_at": None})
"context": None, "published_at": None,
"description": None, "url": None, "cvss": None})
if src not in e["sources"]:
e["sources"].append(src)
for k, v in extra.items():
@@ -317,7 +581,8 @@ def scan(recipe: str, v_from: str | None, v_to: str | None, registry_dir: str,
if a.get("cve"):
record(a["cve"], entry["source"], severity=a.get("severity"), ghsa=a.get("ghsa"),
vulnerable_range=a.get("vulnerable_range"), patched=a.get("patched"),
context=a.get("summary"), published_at=a.get("published_at"))
context=a.get("summary"), published_at=a.get("published_at"),
description=a.get("description"), url=a.get("url"), cvss=a.get("cvss"))
for entry in vendor_pages(urls):
report["sources"].append({"source": entry["source"], "status": entry["status"],
@@ -372,20 +637,28 @@ def scan(recipe: str, v_from: str | None, v_to: str | None, registry_dir: str,
report["windows"] = {k: {"from": f, "to": t} for k, (f, t) in windows.items()}
def _classify_window(src, wf, wt):
"""Return (set_of_fixed_cves, method, date_window|None, unresolved:boolean) for one source."""
"""Return (fixed, method, date_window|None, unresolved, indeterminate) for one source.
`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."""
kf, kt = _vkey(wf), _vkey(wt)
# A version-SCHEME change (semver 3.5.3 → calver 2026.7.1) makes numeric ordering
# meaningless: 2025.12.2 compares "newer" than 3.5.3 while shipping earlier.
scheme = bool(kf and kt and abs(kt[0] - kf[0]) >= SCHEME_JUMP)
if not scheme:
got = set()
got, undecidable = set(), set()
for cve, e in report["cves"].items():
if src not in e["sources"]:
continue
cands = [_vkey(t) for t in re.findall(r"\d+(?:\.\d+)*", e.get("patched") or "")]
if kf and kt and any(kf < c <= kt for c in cands):
patched = e.get("patched") or ""
cands = [_vkey(t) for t in re.findall(r"\d+(?:\.\d+)*", patched)]
if kf and kt and any(_within(kf, kt, c) for c in cands):
got.add(cve)
return got, "patched version ranges", None, False
elif not patched or PLACEHOLDER_RE.search(patched):
# No fix version published ("TBD") or only a placeholder ("7.4.X" — which could
# be 7.4.1, inside the window). We cannot say either way, so say so.
undecidable.add(cve)
return got, "patched version ranges", None, False, undecidable
# DATE FALLBACK: release DATES always order, even across a scheme change. Resolve both
# versions to git tag dates and count advisories PUBLISHED in that window — the method a
# hand count used to establish discourse 3.5.3 (2025-12-30) → 2026.7.1 (2026-07-31) = 123.
@@ -395,12 +668,16 @@ def scan(recipe: str, v_from: str | None, v_to: str | None, registry_dir: str,
got = {cve for cve, e in report["cves"].items()
if src in e["sources"] and e.get("published_at")
and d_from < e["published_at"] <= d_to}
return got, "advisory publish date (version scheme changed)", (d_from, d_to), False
return set(), "unresolved", None, True
undecidable = {cve for cve, e in report["cves"].items()
if src in e["sources"] and not e.get("published_at")}
return got, "advisory publish date (version scheme changed)", (d_from, d_to), False, undecidable
return set(), "unresolved", None, True, set()
fixed_set, methods, date_windows, unresolved_any = set(), {}, {}, False
indeterminate: set = set()
for src, (wf, wt) in windows.items():
got, method, dw, unresolved = _classify_window(src, wf, wt)
got, method, dw, unresolved, undecidable = _classify_window(src, wf, wt)
indeterminate |= undecidable
methods[src] = method
if dw:
date_windows[src] = {"from": dw[0], "to": dw[1]}
@@ -427,7 +704,37 @@ def scan(recipe: str, v_from: str | None, v_to: str | None, registry_dir: str,
report["classified_by"] = methods
if date_windows:
report["date_window"] = date_windows
# THIRD METHOD: before declaring an advisory undecidable, look for the CVE id in the project's
# own release notes. A tag that names it, inside the window, IS the fix version the advisory
# failed to publish. Deterministic and citable — not a judgement call.
resolved_by_release = {}
for cve in sorted(indeterminate - fixed_set):
e = report["cves"][cve]
for src in e["sources"]:
if src not in windows:
continue
wf, wt = windows[src]
kf, kt = _vkey(wf), _vkey(wt)
if not (kf and kt):
continue
hits = [t for t in release_fix_versions(src, cve) if _within(kf, kt, _vkey(t))]
if hits:
fixed_set.add(cve)
resolved_by_release[cve] = sorted(hits)
e["classification"] = (f"fixed-by-this-upgrade (named in release notes "
f"{', '.join(sorted(hits))}) via {src}")
e["fix_versions_from_release_notes"] = sorted(hits)
break
if resolved_by_release:
report["resolved_by_release_notes"] = resolved_by_release
indeterminate -= fixed_set
for cve in indeterminate:
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["indeterminate"] = sorted(indeterminate)
report["cve_count_indeterminate"] = len(indeterminate)
report["unclassified"] = sorted(unknown)
# NEVER report 0 for something we could not determine — a 0 asserts safety. If ANY requested
# window could not be ordered at all, the total is UNKNOWN rather than a partial number.
@@ -468,8 +775,10 @@ def markdown(rep: dict) -> str:
f"advisory APIs). This scan is ADDITIVE — it does not replace the release-note "
f"reading in the upgrade step._")
return "\n".join(L)
ind = rep.get("cve_count_indeterminate") or 0
if rep["fixed_by_this_upgrade"]:
L.append(f"\n**CVEs fixed by this upgrade: {rep['cve_count_fixed']}**\n")
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 {}
if isinstance(cb, dict) and cb:
for src, method in cb.items():
@@ -487,6 +796,15 @@ def markdown(rep: dict) -> str:
f"{e.get('ghsa') or '-'} | {e['sources'][0]} |")
else:
L.append("\n**CVEs fixed by this upgrade: 0 identified by the deterministic scan.**")
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 "
f"range is open-ended, so neither method can tell whether this upgrade fixed them. "
f"They are NOT included in the count above and must NOT be read as unaffected: "
+ ", ".join(f"{c} ({rep['cves'][c].get('severity') or '?'})"
for c in rep["indeterminate"][:15])
+ ("" if len(rep["indeterminate"]) > 15 else "")
+ "\n\nRe-run with `--adjudicate` for the collected evidence on each, to judge.")
if rep["unclassified"]:
L.append(f"\nSeen but not version-classified ({len(rep['unclassified'])}) — includes advisories "
f"from OTHER images in this recipe (sidecars), which this window cannot judge: "
@@ -505,6 +823,10 @@ def main() -> int:
ap.add_argument("--from", dest="v_from", default=None)
ap.add_argument("--to", dest="v_to", default=None)
ap.add_argument("--json", action="store_true", help="emit raw JSON instead of markdown")
ap.add_argument("--adjudicate", action="store_true",
help="SECOND PASS: for advisories the deterministic pass could not judge (no "
"fix version published), fetch their full text + references and append a "
"block for the agent to judge. Additive: it never changes the count above.")
ap.add_argument("--registry", default=REGISTRY_DIR)
ap.add_argument("--image", action="append", default=[], metavar="NAME=FROM:TO",
help="a sidecar image and the versions it moved between, e.g. "
@@ -521,7 +843,15 @@ def main() -> int:
else:
print(f'ignoring malformed --image {spec!r} (expected NAME=FROM:TO)', file=sys.stderr)
rep = scan(a.recipe, a.v_from, a.v_to, a.registry, images)
print(json.dumps(rep, indent=2) if a.json else markdown(rep))
if a.adjudicate:
rep["adjudication"] = [evidence_bundle(rep, c)
for c in needs_judgement(rep)[:MAX_ADJUDICATE]]
if a.json:
print(json.dumps(rep, indent=2))
else:
print(markdown(rep))
if a.adjudicate:
print(adjudication_block(rep))
return 0
+151
View File
@@ -0,0 +1,151 @@
#!/usr/bin/env python3
"""Independent audit of advisory-scan's counts.
Deliberately does NOT reuse the scanner's classifier. Re-parses patched versions with a separate
semver implementation and re-derives membership, then diffs against what the scanner concluded.
Anything the two disagree on is a miscategorization in one of them.
"""
import importlib.util, json, re, sys
spec = importlib.util.spec_from_file_location("A", "/srv/cc-ci-orch/cc-ci-plan/advisory-scan.py")
A = importlib.util.module_from_spec(spec); spec.loader.exec_module(A)
REG = "/srv/cc-ci-orch/cc-ci-plan/upstream"
def sv(s):
"""Independent semver parse: strict 3-tuple, missing parts are 0."""
m = re.match(r"^\s*v?(\d+)(?:\.(\d+))?(?:\.(\d+))?", s or "")
if not m:
return None
return tuple(int(x) if x else 0 for x in m.groups())
def in_window(f, t, patched_expr):
"""Independent membership: any patched token strictly above f and at most t."""
kf, kt = sv(f), sv(t)
for tok in re.findall(r"\d+(?:\.\d+)*", patched_expr or ""):
c = sv(tok)
if c and kf and kt and kf < c <= kt:
return True
return False
_RELS = {}
def fetch_releases(owner_repo):
"""Independent releases fetch — deliberately NOT the scanner's cache or pagination helper."""
if owner_repo in _RELS:
return _RELS[owner_repo]
import urllib.request
tok = None
try:
tok = open("/srv/cc-ci/.github-token").read().strip()
except OSError:
pass
h = {"User-Agent": "audit", "Accept": "application/vnd.github+json"}
if tok:
h["Authorization"] = f"Bearer {tok}"
out, url, pages = [], f"https://api.github.com/repos/{owner_repo}/releases?per_page=100", 0
while url and pages < 4:
req = urllib.request.Request(url, headers=h)
with urllib.request.urlopen(req, timeout=45) as r:
rows = json.load(r)
link = r.headers.get("Link", "") or ""
out += [(x.get("tag_name") or "", (x.get("body") or "") + " " + (x.get("name") or ""))
for x in rows]
url = None
for part in link.split(","):
if 'rel="next"' in part:
url = part.split(";")[0].strip().strip("<>")
pages += 1
_RELS[owner_repo] = out
return out
def audit(recipe, vf, vt, images=None, label=""):
rep = A.scan(recipe, vf, vt, REG, images)
print(f"\n{'='*78}\n{recipe} {vf}{vt} {label}\n{'='*78}")
print(f"scanner count = {rep['cve_count_fixed']} known={rep['count_known']} "
f"failed_sources={rep['sources_failed']}")
counted = set(rep["fixed_by_this_upgrade"])
by_src = {}
for cve, e in rep["cves"].items():
by_src.setdefault(e["sources"][0], []).append((cve, e))
windows = rep["windows"]
total_mismatch = 0
for src, win in windows.items():
f, t = win["from"], win["to"]
method = rep["classified_by"][src]
rows = by_src.get(src, [])
scanner_here = {c for c, e in rows if c in counted}
print(f"\n ── {src} ({f}{t}) method={method} advisories={len(rows)}")
if "publish date" in method:
dw = rep["date_window"][src]
indep = {c for c, e in rows
if e.get("published_at") and dw["from"] < e["published_at"] <= dw["to"]}
print(f" date window {dw['from'][:10]}{dw['to'][:10]}")
else:
indep = {c for c, e in rows if in_window(f, t, e.get("patched"))}
# Independently redo the release-note method: fetch the repo's releases ourselves and
# confirm a tag NAMING the CVE really does fall inside (f, t].
kf, kt = sv(f), sv(t)
owner_repo = src.split(":", 1)[1]
rels = fetch_releases(owner_repo)
for c, e in rows:
if c in indep:
continue
naming = [tag for tag, body in rels if c in body]
if any(kf < sv(tag) <= kt for tag in naming if sv(tag)):
indep.add(c)
missed = indep - scanner_here
extra = scanner_here - indep
print(f" scanner counted {len(scanner_here)} | independent {len(indep)}"
f" | missed_by_scanner {len(missed)} | over_counted {len(extra)}")
if missed:
print(f" !! MISSED: {sorted(missed)}")
for c in sorted(missed):
print(f" {c} patched={dict(rows)[c].get('patched')!r}")
if extra:
print(f" !! OVER-COUNTED: {sorted(extra)}")
for c in sorted(extra):
print(f" {c} patched={dict(rows)[c].get('patched')!r}")
total_mismatch += len(missed) + len(extra)
# Anything counted that belongs to NO window would be a leak.
leaked = {c for c in counted if not any(s in rep["cves"][c]["sources"] for s in windows)}
if leaked:
print(f"\n !! COUNTED BUT OUTSIDE EVERY WINDOW: {sorted(leaked)}")
total_mismatch += len(leaked)
# Unclassified entries that belong to a WINDOWED source would mean a judged CVE was dropped.
dropped = [c for c in rep["unclassified"]
if any(s in rep["cves"][c]["sources"] for s in windows)]
if dropped:
print(f"\n !! UNCLASSIFIED DESPITE HAVING A WINDOW: {sorted(dropped)[:10]}")
total_mismatch += len(dropped)
unwindowed = {}
for cve in rep["unclassified"]:
unwindowed.setdefault(rep["cves"][cve]["sources"][0], []).append(cve)
if unwindowed:
print("\n unclassified by source (expected: images with no --image given):")
for s, cs in sorted(unwindowed.items()):
print(f" {len(cs):4d} {s}")
print(f"\n VERDICT: {'CLEAN' if total_mismatch == 0 else f'{total_mismatch} DISAGREEMENTS'}")
return total_mismatch, rep
if __name__ == "__main__":
bad = 0
bad += audit("gitea", "1.27.0", "1.27.1")[0]
bad += audit("discourse", "3.5.3", "2026.7.1")[0]
bad += audit("discourse", "3.5.3", "2026.7.1", [("redis", "7.4", "8.10")], "+redis sidecar")[0]
bad += audit("keycloak", "26.7.0", "26.7.1")[0]
bad += audit("mailu", "2024.06.55", "2024.06.57", [("redis", "8.8.0", "8.10.0")], "+redis")[0]
bad += audit("n8n", "1.123.0", "2.18.1")[0]
print(f"\n\n{'#'*78}\nOVERALL: {'CLEAN — no disagreements' if bad == 0 else f'{bad} DISAGREEMENTS'}")
sys.exit(1 if bad else 0)
+623
View File
@@ -0,0 +1,623 @@
#!/usr/bin/env python3
"""Tests for advisory-scan.py.
Two tiers:
OFFLINE (default) — pure logic, fixtures injected in place of the network. Fast, deterministic,
no token, no rate limit. These encode every classification rule and every guarantee the CVE count
makes, including the specific production defects that motivated them.
LIVE (--live) — re-derives the CVE counts published in the week-2026-08-07 report against the real
advisory APIs. Slow, needs network + ideally a GitHub token. Run before changing classification.
Usage:
python3 test-advisory-scan.py # offline only
python3 test-advisory-scan.py --live # offline + historic report regressions
"""
from __future__ import annotations
import importlib.util
import io
import json
import os
import pathlib
import sys
import unittest
import unittest.mock
HERE = pathlib.Path(__file__).resolve().parent
_spec = importlib.util.spec_from_file_location("advisory_scan", HERE / "advisory-scan.py")
A = importlib.util.module_from_spec(_spec)
_spec.loader.exec_module(A)
# ── fixture helpers ───────────────────────────────────────────────────────────────────────────────
def adv(cve, patched=None, published=None, severity="high", ghsa=None):
"""One GitHub advisory row as github_advisories() would emit it."""
return {"cve": cve, "ghsa": ghsa or f"GHSA-fake-{cve[-4:]}", "severity": severity,
"summary": f"summary for {cve}", "vulnerable_range": None, "patched": patched,
"url": None, "published_at": published}
def gh(owner_repo, advisories, status="ok"):
return {"source": f"github-advisories:{owner_repo}", "status": status, "advisories": advisories}
def vendor(url, cves=(), status="ok"):
return {"source": url, "status": status, "cves": list(cves),
"context": {c: f"...{c}..." for c in cves}}
def run_scan(gh_entries=(), vendor_entries=(), tag_dates=None, *, v_from=None, v_to=None,
images=None, recipe="fixture", urls=None, releases=None):
"""scan() with every network call replaced by fixtures.
`releases` maps CVE id -> tags whose release notes name it (the third, release-note method)."""
tag_dates = tag_dates or {}
releases = releases or {}
urls = urls if urls is not None else ["https://github.com/app/app"]
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, "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, []))):
return A.scan(recipe, v_from, v_to, "/fake", images)
def parse_image_args(argv):
"""Drive main()'s --image parsing exactly as the CLI does, returning the tuples scan() receives."""
captured = {}
def fake_scan(recipe, vf, vt, reg, images):
captured["images"] = images
return {"recipe": recipe, "from": vf, "to": vt, "registry": reg, "registry_urls": 0,
"sources": [], "cves": {}, "fixed_by_this_upgrade": [], "unclassified": [],
"sources_failed": [], "sources_benign": [], "count_known": True,
"cve_count_fixed": 0, "windows": {}, "classified_by": {}}
err = io.StringIO()
with unittest.mock.patch.object(A, "scan", fake_scan), \
unittest.mock.patch.object(sys, "argv", ["advisory-scan.py", *argv]), \
unittest.mock.patch.object(sys, "stdout", io.StringIO()), \
unittest.mock.patch.object(sys, "stderr", err):
A.main()
return captured["images"], err.getvalue()
# ── A. version ordering ───────────────────────────────────────────────────────────────────────────
class TestVersionKey(unittest.TestCase):
def test_strips_prefix_and_suffix(self):
self.assertEqual(A._vkey("v1.27.1"), (1, 27, 1))
self.assertEqual(A._vkey("1.27.1-rootless"), (1, 27, 1))
self.assertEqual(A._vkey("2024.06.55"), (2024, 6, 55))
def test_empty_and_none(self):
self.assertEqual(A._vkey(None), ())
self.assertEqual(A._vkey(""), ())
def test_dotted_minor_is_numeric_not_lexical(self):
# The bug this guards: "8.10" must be NEWER than "8.2.3". String compare says otherwise.
self.assertGreater(A._vkey("8.10"), A._vkey("8.2.3"))
self.assertGreater(A._vkey("1.27.10"), A._vkey("1.27.9"))
def test_shorter_prefix_orders_below_its_own_patch(self):
# 7.4 < 7.4.1, so a CVE patched in 7.4.1 IS fixed by moving off a bare 7.4 pin.
self.assertLess(A._vkey("7.4"), A._vkey("7.4.1"))
class TestWindowMembership(unittest.TestCase):
"""(from, to] membership — exclusive lower, inclusive upper, compared zero-padded."""
def _in(self, f, t, c):
return A._within(A._vkey(f), A._vkey(t), A._vkey(c))
def test_bounds(self):
self.assertTrue(self._in("1.27.0", "1.27.1", "1.27.1")) # upper inclusive
self.assertFalse(self._in("1.27.0", "1.27.1", "1.27.0")) # lower exclusive
self.assertFalse(self._in("1.27.0", "1.27.1", "1.26.9"))
self.assertFalse(self._in("1.27.0", "1.27.1", "1.28.0"))
def test_bare_major_upper_bound_includes_its_dot_zero(self):
# Regression: plain tuple order makes (18,) < (18,0), so a fix in 18.0 fell OUTSIDE a
# window ending at 18. Bare major tags are the norm for sidecars (postgres:18, redis:8).
self.assertTrue(self._in("17", "18", "18.0"))
self.assertTrue(self._in("7", "8", "8.0"))
self.assertTrue(self._in("7.4", "8.10", "8.0.4"))
def test_bare_major_upper_bound_excludes_later_patches(self):
# Conservative on the other side: nothing proves which 18.x a floating tag resolved to.
self.assertFalse(self._in("17", "18", "18.5"))
def test_bare_version_is_read_literally_as_dot_zero(self):
# from="8" means 8.0, so a fix in 8.0.4 is inside a window that ends at 9.
self.assertTrue(self._in("8", "9", "8.0.4"))
self.assertFalse(self._in("8", "9", "8.0")) # == the stated lower bound
def test_prefix_lower_bound_still_counts_its_patches(self):
self.assertTrue(self._in("7.4", "8.10", "7.4.1"))
self.assertTrue(self._in("7.4", "8.10", "7.4.6"))
def test_the_false_133_cve_stays_out(self):
self.assertFalse(self._in("7.4", "8.10", "6.0.11"))
# ── B. registry URL extraction ────────────────────────────────────────────────────────────────────
class TestRegistryUrls(unittest.TestCase):
def _write(self, tmp, text):
p = pathlib.Path(tmp) / "r.md"
p.write_text(text)
return A.registry_urls("r", tmp)
def test_strips_trailing_markdown_punctuation(self):
# Production defect: a captured backtick 404'd the fetch and rendered n8n/immich as '?'.
import tempfile
with tempfile.TemporaryDirectory() as tmp:
urls, _ = self._write(tmp, "see `https://docs.n8n.io/release-notes/` and "
"**https://example.com/sec.html**, plus https://a.test/x.")
self.assertIn("https://docs.n8n.io/release-notes/", urls)
self.assertIn("https://example.com/sec.html", urls)
self.assertIn("https://a.test/x", urls)
self.assertFalse([u for u in urls if u.endswith(("`", "*", ".", ","))])
def test_dedupes_and_reports_missing_registry(self):
import tempfile
with tempfile.TemporaryDirectory() as tmp:
urls, path = self._write(tmp, "https://a.test/x https://a.test/x")
self.assertEqual(urls.count("https://a.test/x"), 1)
self.assertTrue(path.endswith("r.md"))
urls, path = A.registry_urls("does-not-exist", "/nonexistent-dir")
self.assertEqual((urls, path), ([], None))
# ── C. --image argument parsing ───────────────────────────────────────────────────────────────────
class TestImageArgParsing(unittest.TestCase):
def test_single_and_repeated(self):
imgs, _ = parse_image_args(["r", "--image", "redis=7.4:8.10"])
self.assertEqual(imgs, [("redis", "7.4", "8.10")])
imgs, _ = parse_image_args(["r", "--image", "redis=7.4:8.10", "--image", "postgres=17:18"])
self.assertEqual(imgs, [("redis", "7.4", "8.10"), ("postgres", "17", "18")])
def test_malformed_is_skipped_with_a_warning_not_a_crash(self):
# It is an ADDITIVE pre-step: one typo must not abort the upgrade's scan step.
for bad in ("redis=7.4", "redis", "=7.4:8.10", "redis=:8.10", "redis=7.4:"):
imgs, err = parse_image_args(["r", "--image", bad])
self.assertEqual(imgs, [], f"{bad!r} should be rejected")
self.assertIn("malformed", err)
def test_good_and_bad_mixed_keeps_the_good(self):
imgs, err = parse_image_args(["r", "--image", "redis=7.4:8.10", "--image", "nope"])
self.assertEqual(imgs, [("redis", "7.4", "8.10")])
self.assertIn("malformed", err)
# ── D. classification ─────────────────────────────────────────────────────────────────────────────
class TestClassificationBoundaries(unittest.TestCase):
def _one(self, patched, v_from="1.27.0", v_to="1.27.1"):
rep = run_scan([gh("app/app", [adv("CVE-2026-0001", patched=patched)])],
v_from=v_from, v_to=v_to)
return rep
def test_patched_at_upper_bound_counts(self):
self.assertEqual(self._one("1.27.1")["fixed_by_this_upgrade"], ["CVE-2026-0001"])
def test_patched_at_lower_bound_does_not_count(self):
# Already fixed in the version we were ON — this upgrade did not fix it.
rep = self._one("1.27.0")
self.assertEqual(rep["fixed_by_this_upgrade"], [])
self.assertEqual(rep["cve_count_fixed"], 0)
def test_patched_below_and_above_window_do_not_count(self):
self.assertEqual(self._one("1.26.0")["fixed_by_this_upgrade"], [])
self.assertEqual(self._one("1.28.0")["fixed_by_this_upgrade"], [])
def test_any_of_several_patched_lines_counts(self):
# n8n regression: one advisory patches several release lines; reading only the first
# dropped the line the deployment was on (CVE-2026-42231/42232 misclassified).
rep = self._one("1.123.32; 2.17.4; 2.18.1", v_from="2.17.0", v_to="2.17.4")
self.assertEqual(rep["fixed_by_this_upgrade"], ["CVE-2026-0001"])
def test_no_patched_data_is_not_counted(self):
self.assertEqual(self._one(None)["fixed_by_this_upgrade"], [])
class TestPerImageWindows(unittest.TestCase):
"""The false-133 family of defects: an image must only ever be judged by its OWN versions."""
APP = gh("discourse/discourse", [adv("CVE-APP-0001", patched="3.5.4", published="2026-03-01T00:00:00Z")])
REDIS = gh("redis/redis", [
adv("CVE-2021-21309", patched="6.0.11", published="2021-02-01T00:00:00Z"),
adv("CVE-2025-49844", patched="7.4.6; 8.0.4; 8.2.2", published="2025-10-01T00:00:00Z",
severity="critical"),
])
URLS = ["https://github.com/discourse/discourse", "https://github.com/redis/redis"]
def test_sidecar_cve_is_not_judged_by_the_app_window(self):
# redis 6.0.11 sits numerically inside discourse 3.5.3 -> 2026.7.1. It must NOT count.
rep = run_scan([self.APP, self.REDIS], v_from="3.5.3", v_to="2026.7.1",
tag_dates={"3.5.3": "2025-12-30T00:00:00Z", "2026.7.1": "2026-07-31T00:00:00Z"},
urls=self.URLS)
self.assertNotIn("CVE-2021-21309", rep["fixed_by_this_upgrade"])
self.assertIn("CVE-2021-21309", rep["unclassified"])
def test_unwindowed_image_is_unclassified_never_counted(self):
rep = run_scan([self.APP, self.REDIS], v_from="3.5.3", v_to="3.5.4", urls=self.URLS)
self.assertEqual(rep["fixed_by_this_upgrade"], ["CVE-APP-0001"])
for cve in ("CVE-2021-21309", "CVE-2025-49844"):
self.assertIn(cve, rep["unclassified"])
def test_sidecar_window_counts_only_what_that_bump_fixed(self):
rep = run_scan([self.APP, self.REDIS], v_from="3.5.3", v_to="3.5.4",
images=[("redis", "7.4", "8.10")], urls=self.URLS)
self.assertIn("CVE-2025-49844", rep["fixed_by_this_upgrade"]) # patched 7.4.6, in window
self.assertNotIn("CVE-2021-21309", rep["fixed_by_this_upgrade"]) # patched 6.0.11, below it
self.assertEqual(rep["cve_count_fixed"], 2) # app 1 + redis 1
def test_count_is_the_union_across_images(self):
rep = run_scan([self.APP, self.REDIS], v_from="3.5.3", v_to="3.5.4",
images=[("redis", "7.4", "8.10")], urls=self.URLS)
self.assertEqual(sorted(rep["fixed_by_this_upgrade"]), ["CVE-2025-49844", "CVE-APP-0001"])
def test_each_image_classified_independently(self):
# App crosses a scheme change (date method); redis does not (version method). Both resolve.
rep = run_scan([self.APP, self.REDIS], v_from="3.5.3", v_to="2026.7.1",
images=[("redis", "7.4", "8.10")], urls=self.URLS,
tag_dates={"3.5.3": "2025-12-30T00:00:00Z", "2026.7.1": "2026-07-31T00:00:00Z"})
methods = rep["classified_by"]
self.assertIn("publish date", methods["github-advisories:discourse/discourse"])
self.assertEqual(methods["github-advisories:redis/redis"], "patched version ranges")
self.assertTrue(rep["count_known"])
def test_image_name_matches_as_substring(self):
rep = run_scan([self.APP, gh("discourse/discourse-postgres", [adv("CVE-PG-1", patched="18.0")])],
v_from="3.5.3", v_to="3.5.4", images=[("postgres", "17", "18")],
urls=["https://github.com/discourse/discourse",
"https://github.com/discourse/discourse-postgres"])
self.assertIn("github-advisories:discourse/discourse-postgres", rep["windows"])
self.assertIn("CVE-PG-1", rep["fixed_by_this_upgrade"])
def test_primary_cannot_be_stolen_by_a_loose_image_name(self):
rep = run_scan([self.APP, gh("discourse/discourse-postgres", [adv("CVE-PG-1", patched="18.0")])],
v_from="3.5.3", v_to="3.5.4", images=[("discourse", "1", "2")],
urls=["https://github.com/discourse/discourse",
"https://github.com/discourse/discourse-postgres"])
self.assertEqual(rep["windows"]["github-advisories:discourse/discourse"],
{"from": "3.5.3", "to": "3.5.4"})
def test_unmatched_image_name_is_silently_ignored(self):
# Documents CURRENT behaviour: a typo'd name costs coverage without warning.
rep = run_scan([self.APP], v_from="3.5.3", v_to="3.5.4",
images=[("nosuchimage", "1", "2")], urls=self.URLS[:1])
self.assertEqual(list(rep["windows"]), ["github-advisories:discourse/discourse"])
self.assertTrue(rep["count_known"])
class TestSchemeChangeDateFallback(unittest.TestCase):
DATES = {"3.5.3": "2025-12-30T00:00:00Z", "2026.7.1": "2026-07-31T00:00:00Z"}
def _rep(self, advisories, dates=None):
return run_scan([gh("discourse/discourse", advisories)], v_from="3.5.3", v_to="2026.7.1",
tag_dates=self.DATES if dates is None else dates)
def test_counts_advisories_published_inside_the_date_window(self):
rep = self._rep([adv("CVE-IN-1", published="2026-03-01T00:00:00Z"),
adv("CVE-OUT-1", published="2025-06-01T00:00:00Z"),
adv("CVE-OUT-2", published="2026-09-01T00:00:00Z")])
self.assertEqual(rep["fixed_by_this_upgrade"], ["CVE-IN-1"])
def test_date_boundaries_match_the_version_rule(self):
# Exclusive lower, inclusive upper — same as 4a, so the two methods agree at the edges.
rep = self._rep([adv("CVE-LOWER", published=self.DATES["3.5.3"]),
adv("CVE-UPPER", published=self.DATES["2026.7.1"])])
self.assertEqual(rep["fixed_by_this_upgrade"], ["CVE-UPPER"])
def test_advisory_without_a_publish_date_is_not_counted(self):
self.assertEqual(self._rep([adv("CVE-NODATE", published=None)])["fixed_by_this_upgrade"], [])
def test_scheme_change_is_detected_not_version_compared(self):
rep = self._rep([adv("CVE-IN-1", patched="2026.1.0", published="2026-03-01T00:00:00Z")])
self.assertIn("publish date", rep["classified_by"]["github-advisories:discourse/discourse"])
self.assertIn("github-advisories:discourse/discourse", rep["date_window"])
def test_small_major_bump_still_uses_version_ranges(self):
rep = run_scan([gh("app/app", [adv("CVE-X", patched="3.0.0")])], v_from="2.9.0", v_to="3.0.0")
self.assertEqual(rep["classified_by"]["github-advisories:app/app"], "patched version ranges")
# ── E. count guarantees ───────────────────────────────────────────────────────────────────────────
class TestCountGuarantees(unittest.TestCase):
def test_unresolvable_window_yields_unknown_never_zero(self):
# Scheme change AND tag dates unresolvable -> must refuse to emit a number.
rep = run_scan([gh("app/app", [adv("CVE-1", published="2026-01-01T00:00:00Z")])],
v_from="3.5.3", v_to="2026.7.1", tag_dates={})
self.assertIs(rep["cve_count_fixed"], None)
self.assertFalse(rep["count_known"])
md = A.markdown(rep)
self.assertIn("UNKNOWN", md)
self.assertIn("NOT zero", md)
def test_one_unresolvable_image_makes_the_whole_count_unknown(self):
# A partial number would understate a security figure, so it is suppressed entirely.
rep = run_scan([gh("app/app", [adv("CVE-APP", patched="1.1")]),
gh("redis/redis", [adv("CVE-REDIS", patched="8.0")])],
v_from="1.0", v_to="1.1", images=[("redis", "7.4", "9999.1")],
tag_dates={}, urls=["https://github.com/app/app", "https://github.com/redis/redis"])
self.assertIs(rep["cve_count_fixed"], None)
self.assertFalse(rep["count_known"])
def test_genuine_zero_is_reported_as_zero(self):
rep = run_scan([gh("app/app", [adv("CVE-1", patched="9.9.9")])], v_from="1.0", v_to="1.1")
self.assertEqual(rep["cve_count_fixed"], 0)
self.assertTrue(rep["count_known"])
self.assertIn("0 identified", 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")],
v_from="1.0", v_to="1.1")
self.assertEqual(rep["sources_failed"], [])
self.assertIn("github-advisories:side/car", rep["sources_benign"])
self.assertEqual(rep["cve_count_fixed"], 1)
def test_template_url_is_benign_not_a_failure(self):
rep = run_scan([gh("app/app", [adv("CVE-1", patched="1.1")])],
[vendor("https://x.test/changelog/v<VERSION>/", status="skipped: template URL")],
v_from="1.0", v_to="1.1")
self.assertEqual(rep["sources_failed"], [])
def test_real_source_failure_is_surfaced(self):
rep = run_scan([gh("app/app", [adv("CVE-1", patched="1.1")], status="error: HTTP 500")],
v_from="1.0", v_to="1.1")
self.assertIn("github-advisories:app/app", rep["sources_failed"])
self.assertIn("FAILED", A.markdown(rep))
def test_no_window_given_classifies_nothing(self):
rep = run_scan([gh("app/app", [adv("CVE-1", patched="1.1")])])
self.assertEqual(rep["fixed_by_this_upgrade"], [])
self.assertIn("CVE-1", rep["unclassified"])
class TestVendorOnlyCves(unittest.TestCase):
"""The gitea case: CVEs named ONLY on a vendor page, absent from the GitHub advisory feed."""
def test_vendor_only_cve_is_recorded_and_surfaced(self):
rep = run_scan([gh("go-gitea/gitea", [])],
[vendor("https://blog.gitea.com/release-1.27.1/", ["CVE-2026-60004"])],
v_from="1.27.0", v_to="1.27.1")
self.assertIn("CVE-2026-60004", rep["cves"])
self.assertIn("CVE-2026-60004", rep["unclassified"])
def test_vendor_only_cve_is_NOT_counted_but_IS_sent_for_judgement(self):
# It carries no version data, so no arithmetic can place it — the deterministic count must
# not include it. It must not be silently dropped either: pass 2 gets it with its evidence.
rep = run_scan([gh("go-gitea/gitea", [])],
[vendor("https://blog.gitea.com/release-1.27.1/", ["CVE-2026-60004"])],
v_from="1.27.0", v_to="1.27.1")
self.assertEqual(rep["fixed_by_this_upgrade"], [])
self.assertEqual(rep["cve_count_fixed"], 0)
self.assertIn("CVE-2026-60004", A.needs_judgement(rep))
def test_cve_in_both_vendor_and_advisory_feed_is_counted_once(self):
rep = run_scan([gh("go-gitea/gitea", [adv("CVE-2026-60004", patched="1.27.1")])],
[vendor("https://blog.gitea.com/x/", ["CVE-2026-60004"])],
v_from="1.27.0", v_to="1.27.1")
self.assertEqual(rep["fixed_by_this_upgrade"], ["CVE-2026-60004"])
self.assertEqual(rep["cve_count_fixed"], 1)
self.assertEqual(len(rep["cves"]["CVE-2026-60004"]["sources"]), 2)
class TestIndeterminateBucket(unittest.TestCase):
"""An advisory with no knowable fix version is neither counted nor dismissed."""
def test_tbd_patched_is_indeterminate_not_excluded(self):
rep = run_scan([gh("redis/redis", [adv("CVE-TBD", patched="TBD")])],
v_from="7.4", v_to="8.10", urls=["https://github.com/redis/redis"])
self.assertEqual(rep["fixed_by_this_upgrade"], [])
self.assertIn("CVE-TBD", rep["indeterminate"])
self.assertEqual(rep["cve_count_indeterminate"], 1)
def test_placeholder_patched_is_indeterminate(self):
# "7.4.X" could be 7.4.1 — inside the window. Extracting a bare 7.4 and excluding it was
# how CVE-2024-46981 (high) went missing.
rep = run_scan([gh("redis/redis", [adv("CVE-X", patched="6.2.X, 7.2.X, 7.4.X")])],
v_from="7.4", v_to="8.10", urls=["https://github.com/redis/redis"])
self.assertIn("CVE-X", rep["indeterminate"])
def test_real_versions_outside_the_window_are_decided_not_indeterminate(self):
rep = run_scan([gh("redis/redis", [adv("CVE-OLD", patched="6.0.11")])],
v_from="7.4", v_to="8.10", urls=["https://github.com/redis/redis"])
self.assertEqual(rep["indeterminate"], [])
self.assertEqual(rep["cve_count_fixed"], 0)
def test_indeterminate_is_surfaced_in_the_markdown_and_not_read_as_zero(self):
rep = run_scan([gh("redis/redis", [adv("CVE-TBD", patched="TBD", severity="critical")])],
v_from="7.4", v_to="8.10", urls=["https://github.com/redis/redis"])
md = A.markdown(rep)
self.assertIn("could NOT be judged", md)
self.assertIn("must NOT be read as unaffected", md)
class TestReleaseNoteResolution(unittest.TestCase):
"""Third method: a release whose notes NAME the CVE supplies the fix version the advisory lacks."""
def test_release_naming_the_cve_inside_the_window_counts_it(self):
rep = run_scan([gh("redis/redis", [adv("CVE-TBD", patched="TBD")])],
v_from="7.4", v_to="8.10", urls=["https://github.com/redis/redis"],
releases={"CVE-TBD": ["6.2.19", "7.2.10", "7.4.5", "8.0.3"]})
self.assertIn("CVE-TBD", rep["fixed_by_this_upgrade"])
self.assertEqual(rep["indeterminate"], [])
self.assertEqual(rep["resolved_by_release_notes"]["CVE-TBD"], ["7.4.5", "8.0.3"])
def test_release_naming_it_only_outside_the_window_stays_indeterminate(self):
rep = run_scan([gh("redis/redis", [adv("CVE-TBD", patched="TBD")])],
v_from="7.4", v_to="8.10", urls=["https://github.com/redis/redis"],
releases={"CVE-TBD": ["6.2.19"]})
self.assertEqual(rep["fixed_by_this_upgrade"], [])
self.assertIn("CVE-TBD", rep["indeterminate"])
def test_release_evidence_is_recorded_for_audit(self):
rep = run_scan([gh("redis/redis", [adv("CVE-TBD", patched="TBD")])],
v_from="7.4", v_to="8.10", urls=["https://github.com/redis/redis"],
releases={"CVE-TBD": ["7.4.5"]})
e = rep["cves"]["CVE-TBD"]
self.assertEqual(e["fix_versions_from_release_notes"], ["7.4.5"])
self.assertIn("named in release notes", e["classification"])
def test_it_does_not_override_a_version_range_decision(self):
# A CVE already counted by patched ranges is untouched; the method only rescues undecided.
rep = run_scan([gh("redis/redis", [adv("CVE-OK", patched="7.4.1")])],
v_from="7.4", v_to="8.10", urls=["https://github.com/redis/redis"],
releases={"CVE-OK": ["7.4.1"]})
self.assertNotIn("CVE-OK", rep.get("resolved_by_release_notes") or {})
class TestAdjudicationEvidenceAssembly(unittest.TestCase):
"""Pass 2's JUDGEMENT is a model's and not testable; what IS testable is what it gets shown."""
def test_selects_indeterminate_and_vendor_only_cases(self):
rep = run_scan([gh("redis/redis", [adv("CVE-TBD", patched="TBD")])],
[vendor("https://blog.test/sec", ["CVE-VENDOR"])],
v_from="7.4", v_to="8.10", urls=["https://github.com/redis/redis"])
todo = A.needs_judgement(rep)
self.assertIn("CVE-TBD", todo)
self.assertIn("CVE-VENDOR", todo)
def test_does_not_re_ask_about_cases_pass_1_settled(self):
rep = run_scan([gh("redis/redis", [adv("CVE-OK", patched="7.4.1")])],
v_from="7.4", v_to="8.10", urls=["https://github.com/redis/redis"])
self.assertNotIn("CVE-OK", A.needs_judgement(rep))
def test_evidence_bundle_carries_the_window_and_published_fields(self):
rep = run_scan([gh("redis/redis", [adv("CVE-TBD", patched="TBD")])],
v_from="7.4", v_to="8.10", urls=["https://github.com/redis/redis"])
with unittest.mock.patch.object(A, "advisory_text", lambda g, s=None: {"status": "skipped"}), \
unittest.mock.patch.object(A, "release_fix_versions", lambda s, c: []):
ev = A.evidence_bundle(rep, "CVE-TBD")
self.assertEqual(ev["window"], {"from": "7.4", "to": "8.10"})
self.assertEqual(ev["patched_as_published"], "TBD")
self.assertIn("no fix version", ev["why_undecided"])
def test_pass_1_decisions_are_included_for_review(self):
rep = run_scan([gh("redis/redis", [adv("CVE-OK", patched="7.4.1"),
adv("CVE-OLD", patched="6.0.11"),
adv("CVE-TBD", patched="TBD")])],
v_from="7.4", v_to="8.10", urls=["https://github.com/redis/redis"])
with unittest.mock.patch.object(A, "advisory_text", lambda g, s=None: {"status": "skipped"}), \
unittest.mock.patch.object(A, "release_fix_versions", lambda s, c: []):
block = A.adjudication_block(rep)
self.assertIn("Pass 1 decisions", block)
self.assertIn("CVE-OK", block) # counted
self.assertIn("CVE-OLD", block) # excluded as outside-window
self.assertIn("CVE-TBD", block) # needs judgement
def test_truncation_is_announced_never_silent(self):
advs = [adv(f"CVE-2026-{1000+i}", patched="TBD") for i in range(30)]
rep = run_scan([gh("redis/redis", advs)], v_from="7.4", v_to="8.10",
urls=["https://github.com/redis/redis"])
with unittest.mock.patch.object(A, "advisory_text", lambda g, s=None: {"status": "skipped"}), \
unittest.mock.patch.object(A, "release_fix_versions", lambda s, c: []), \
unittest.mock.patch.object(A, "MAX_ADJUDICATE", 5):
block = A.adjudication_block(rep)
self.assertIn("not shown", block)
self.assertIn("do not", block.lower())
class TestMarkdownOutput(unittest.TestCase):
def test_lists_every_window_with_its_method(self):
rep = run_scan([gh("discourse/discourse", [adv("CVE-A", patched="3.5.4")]),
gh("redis/redis", [adv("CVE-B", patched="8.0")])],
v_from="3.5.3", v_to="3.5.4", images=[("redis", "7.4", "8.10")],
urls=["https://github.com/discourse/discourse", "https://github.com/redis/redis"])
md = A.markdown(rep)
self.assertIn("discourse/discourse: 3.5.3 → 3.5.4", md)
self.assertIn("redis/redis: 7.4 → 8.10", md)
self.assertIn("**CVEs fixed by this upgrade: 2**", md)
def test_severity_and_fixed_in_are_rendered(self):
rep = run_scan([gh("redis/redis", [adv("CVE-2025-49844", patched="7.4.6; 8.2.2",
severity="critical")])],
v_from="7.4", v_to="8.10", urls=["https://github.com/redis/redis"])
md = A.markdown(rep)
self.assertIn("critical", md)
self.assertIn("7.4.6", md)
# ── F. live regressions against published historic reports ────────────────────────────────────────
class TestHistoricReportNumbers(unittest.TestCase):
"""Re-derive counts published in week-2026-08-07. Network + GitHub token; opt in with --live."""
REGISTRY = str(HERE / "upstream")
@classmethod
def setUpClass(cls):
if not os.environ.get("ADVISORY_SCAN_LIVE"):
raise unittest.SkipTest("live tests: re-run with --live")
def _count(self, recipe, v_from, v_to, images=None):
rep = A.scan(recipe, v_from, v_to, self.REGISTRY, images)
self.assertEqual(rep["sources_failed"], [], f"{recipe}: source failures make the count unsafe")
self.assertTrue(rep["count_known"], f"{recipe}: count came back UNKNOWN")
return rep
def test_gitea_1_27_0_to_1_27_1_is_2(self):
rep = self._count("gitea", "1.27.0", "1.27.1")
self.assertEqual(rep["cve_count_fixed"], 2)
# Both CVSS-9.8 RCEs — the pair whose omission is why this tool exists.
self.assertEqual(set(rep["fixed_by_this_upgrade"]), {"CVE-2026-59774", "CVE-2026-60004"})
def test_discourse_app_only_is_123(self):
rep = self._count("discourse", "3.5.3", "2026.7.1")
self.assertEqual(rep["cve_count_fixed"], 123)
self.assertIn("publish date", rep["classified_by"]["github-advisories:discourse/discourse"])
def test_discourse_with_redis_sidecar_is_140(self):
# 123 app + 17 redis. Five redis advisories carry a usable patched_versions; the other
# twelve say "TBD" and are resolved from the release notes that name them.
rep = self._count("discourse", "3.5.3", "2026.7.1", [("redis", "7.4", "8.10")])
self.assertEqual(rep["cve_count_fixed"], 140)
self.assertEqual(len(rep.get("resolved_by_release_notes") or {}), 12)
self.assertEqual(rep["cve_count_indeterminate"], 0)
# The five redis advisories that a sidecar-blind scan missed, incl. one critical.
for cve in ("CVE-2024-31227", "CVE-2024-31228", "CVE-2024-31449",
"CVE-2025-49844", "CVE-2025-62507"):
self.assertIn(cve, rep["fixed_by_this_upgrade"], f"{cve} missing from discourse+redis")
self.assertEqual(rep["cves"]["CVE-2025-49844"]["severity"], "critical")
def test_discourse_redis_delta_is_exactly_seventeen(self):
app = self._count("discourse", "3.5.3", "2026.7.1")
both = self._count("discourse", "3.5.3", "2026.7.1", [("redis", "7.4", "8.10")])
delta = set(both["fixed_by_this_upgrade"]) - set(app["fixed_by_this_upgrade"])
self.assertEqual(len(delta), 17)
for cve in delta:
self.assertIn("redis", both["cves"][cve]["sources"][0])
def test_mailu_scan_finds_zero_and_says_so_knowingly(self):
# Published as 2 via the UNION with release-note reading; the scan's own contribution is 0,
# and 0 here must mean "checked, none", not "could not check".
rep = self._count("mailu", "2024.06.55", "2024.06.57", [("redis", "8.8.0", "8.10.0")])
self.assertEqual(rep["cve_count_fixed"], 0)
self.assertTrue(rep["count_known"])
def test_keycloak_26_7_0_to_26_7_1_is_7(self):
rep = self._count("keycloak", "26.7.0", "26.7.1")
self.assertEqual(rep["cve_count_fixed"], 7)
def _main():
live = "--live" in sys.argv
if live:
sys.argv.remove("--live")
os.environ["ADVISORY_SCAN_LIVE"] = "1"
unittest.main(verbosity=2)
if __name__ == "__main__":
_main()