CVE detection engine — correctness fixes, spec, tests, audit #2
@@ -42,10 +42,30 @@ keeps every weekly edition looking the same regardless of which model writes the
|
||||
Advisories + vendor security pages + OSV, with severities and fixed-in versions). Treat its
|
||||
CVE list as a **further source** and report the **UNION** of it and what you found by reading.
|
||||
Its entries are machine-derived with advisory IDs, so prefer them for CVE ids / severities /
|
||||
fixed-in versions, and cite the GHSA where present in the Security Bulletin. If a recipe has
|
||||
no scan block, or the block 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 RCEs were reported as "none" on 2026-08-07.
|
||||
fixed-in versions, and cite the GHSA where present in the Security Bulletin. If the block
|
||||
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
|
||||
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
|
||||
ONLY when a scan ran and reported genuinely failed sources, **or when the scan block says
|
||||
COUNT UNKNOWN**. In that case the scan's `0` means *not determined*: publish `?` and say so in
|
||||
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).
|
||||
When a sidecar contributes a critical/high, name the image in the bulletin — CVE-2025-49844 is
|
||||
a redis flaw, not a discourse one, and an operator reading "discourse" needs to know that.
|
||||
(The scan headline itself now says `UNKNOWN` rather than a number in that case.)
|
||||
In particular: a recipe with **no upgrade this run** (up-to-date/skipped) has nothing an
|
||||
upgrade could have fixed — report `0`, not `?`. A recipe with a clean scan reports its number (including `0`). Benign notes in a scan
|
||||
block (`no-advisories-published`, `skipped: template URL`) are NOT failures and must not
|
||||
trigger `?`. If you find yourself rendering `?` for many recipes, that is a bug to report in
|
||||
the Addendum, not a normal outcome.
|
||||
Anything **critical/high** also gets a `security` bulletin entry (recipe · CVE id(s) + severity ·
|
||||
what it fixes · PR link); be specific about severity and what's exposed if not merged.
|
||||
- **Lead — ONE short paragraph.** A tight, concrete opener in opus's voice: fleet state in a sentence
|
||||
|
||||
@@ -163,9 +163,30 @@ Run the deterministic scanner for the exact upgrade window and **paste its markd
|
||||
into the per-recipe log**:
|
||||
|
||||
```
|
||||
python3 /srv/cc-ci/cc-ci-plan/advisory-scan.py <recipe> --from <old-app-version> --to <new-app-version>
|
||||
python3 /srv/cc-ci/cc-ci-plan/advisory-scan.py <recipe> --from <old-app-version> --to <new-app-version> \
|
||||
[--image <name>=<old>:<new>]...
|
||||
```
|
||||
|
||||
**Pass an `--image` for EVERY sidecar you upgraded** (redis, postgres, nginx …), not just the app —
|
||||
each image is judged by its own versions, and an image you don't name is not counted at all. Repeat
|
||||
the flag for each one and pass them **all in a single call** (the count is a union across images).
|
||||
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
|
||||
```
|
||||
|
||||
→ 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
|
||||
"fixed by THIS upgrade" is computed, not guessed), every **vendor release/security URL** in that
|
||||
|
||||
@@ -0,0 +1,313 @@
|
||||
# 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-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 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
|
||||
notes*, which name neither — both were announced only in the vendor's blog security section — and the
|
||||
report then derived security content from those notes plus model knowledge, which predates the CVEs.
|
||||
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>]... [--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, 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**. |
|
||||
|
||||
Exit code is always 0 — this is informational. Failures are *reported*, never raised.
|
||||
|
||||
---
|
||||
|
||||
# Pass 1 — deterministic
|
||||
|
||||
## Step 1 — Collect source URLs from the registry
|
||||
|
||||
Read `cc-ci-plan/upstream/<recipe>.md` and extract every `http(s)://…` URL.
|
||||
|
||||
**Trailing markdown punctuation is stripped** (`` ` `` `'` `"` `*` `.` `,` `;` `:` `>` `)`). The registry
|
||||
is markdown, so URLs appear inside backticks and quotes; capturing the punctuation produced fetches of
|
||||
`https://docs.n8n.io/release-notes/\`` which 404, and made immich and n8n render `?` for no real reason.
|
||||
|
||||
> **Registry hygiene matters.** The scan can only look where the registry points. Two classes of defect
|
||||
> have been found and fixed by running it: a **wrong URL** (`pgautoupgrade/pgautoupgrade`, which 404s —
|
||||
> the repo is `pgautoupgrade/docker-pgautoupgrade`) and a **missing** one (gitea's CVEs are announced at
|
||||
> `blog.gitea.com`, which the registry didn't list). When a vendor publishes security notes somewhere
|
||||
> the registry lacks, add it.
|
||||
|
||||
## Step 2 — Query the sources
|
||||
|
||||
Three source classes, each recording **its own status** so *"checked, none found"* is never confused
|
||||
with *"not checked"*.
|
||||
|
||||
### 2a. GitHub Security Advisories — PRIMARY
|
||||
|
||||
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`, **`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
|
||||
line a deployment was actually on, and misclassified CVE-2026-42231/42232 as out-of-window.
|
||||
- **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.
|
||||
|
||||
### 2b. Vendor release / security pages
|
||||
|
||||
Every other registry URL is fetched, HTML-stripped, and scanned for `CVE-\d{4}-\d{4,7}`, keeping ±160
|
||||
characters of context per hit.
|
||||
|
||||
URLs containing `<`, `>`, `{`, `}`, `VERSION`, or `vX.Y.Z` are **skipped as templates** — they are
|
||||
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. 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
|
||||
|
||||
Only when the recipe has an entry in `OSV_PACKAGES` (ecosystem + package) and a version is given.
|
||||
|
||||
> **Measured, not assumed.** For the two gitea CVEs, OSV **404'd on both** and returned only Go
|
||||
> *dependency* advisories for the package; NVD's API had them by neither CPE, CVE id, nor keyword.
|
||||
> **Advisory databases lag the vendor**, which is why 2a and 2b lead and this is supplementary.
|
||||
|
||||
## Step 3 — Union
|
||||
|
||||
All findings merge into one CVE map: id → `{sources[], severity, ghsa, vulnerable_range, patched,
|
||||
description, published_at, url, cvss, context}`. A CVE seen by several sources keeps them all.
|
||||
|
||||
## Step 4 — Classify against the upgrade window
|
||||
|
||||
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. 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
|
||||
> **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 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` — 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)
|
||||
|
||||
Used **only** when 4a cannot be trusted: a **version-scheme change**, detected as the leading version
|
||||
component jumping by ≥ `SCHEME_JUMP` (100) — e.g. semver `3.5.3` → calver `2026.7.1`.
|
||||
|
||||
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 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` — the same exclusive/inclusive
|
||||
boundaries as 4a, so the two methods agree at the edges.
|
||||
|
||||
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. By release notes naming the CVE (rescue — for advisories with no fix version)
|
||||
|
||||
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 4a–4c 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
|
||||
|
||||
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 window was unorderable |
|
||||
| `count_known` | Distinguishes "counted zero" from "could not count" |
|
||||
| `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 (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.
|
||||
|
||||
---
|
||||
|
||||
## How consumers must read it
|
||||
|
||||
`/recipe-report` renders the `cve` column from the **union** of this scan and the agent's own reading:
|
||||
|
||||
- a clean scan → its number, **including `0`**;
|
||||
- **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 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. 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).
|
||||
+536
-24
@@ -51,6 +51,14 @@ REGISTRY_DIR = os.environ.get("CCCI_UPSTREAM_REGISTRY", "/srv/cc-ci/cc-ci-plan/u
|
||||
UA = "cc-ci-advisory-scan (+https://git.autonomic.zone/recipe-maintainers/cc-ci)"
|
||||
TIMEOUT = int(os.environ.get("ADVISORY_SCAN_TIMEOUT", "45"))
|
||||
CVE_RE = re.compile(r"CVE-\d{4}-\d{4,7}")
|
||||
# Leading-version-component jump that means the scheme changed (semver → calver).
|
||||
SCHEME_JUMP = 100
|
||||
# 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]] = {
|
||||
@@ -80,6 +88,57 @@ def _github_token() -> str | None:
|
||||
return None
|
||||
|
||||
|
||||
def _gh_paginate(url: str, hdrs: dict, max_pages: int = 20):
|
||||
"""Yield every row from a GitHub list endpoint, following Link rel=\"next\" cursors."""
|
||||
seen_keys = set()
|
||||
for _ in range(max_pages):
|
||||
req = urllib.request.Request(url, headers={"User-Agent": UA, **hdrs})
|
||||
with urllib.request.urlopen(req, timeout=TIMEOUT) as r:
|
||||
rows = json.load(r)
|
||||
link = r.headers.get("Link", "") or ""
|
||||
fresh = 0
|
||||
for a in rows:
|
||||
k = a.get("ghsa_id") or json.dumps(a, sort_keys=True)[:120]
|
||||
if k not in seen_keys:
|
||||
seen_keys.add(k); fresh += 1
|
||||
yield a
|
||||
nxt = None
|
||||
for part in link.split(","):
|
||||
if 'rel="next"' in part:
|
||||
nxt = part.split(";")[0].strip().strip("<>")
|
||||
if not nxt or fresh == 0:
|
||||
return
|
||||
url = nxt
|
||||
|
||||
|
||||
def _tag_date(owner: str, repo: str, version: str | None) -> str | None:
|
||||
"""Publish date of a release tag, for DATE-BASED classification (see classify_by_date).
|
||||
|
||||
Version strings cannot be ordered across a scheme change (semver → calver), but tag dates
|
||||
always can. Tries the common tag spellings; returns an ISO timestamp or None."""
|
||||
if not version:
|
||||
return None
|
||||
hdrs = {"Accept": "application/vnd.github+json"}
|
||||
tok = _github_token()
|
||||
if tok:
|
||||
hdrs["Authorization"] = f"Bearer {tok}"
|
||||
for tag in (f"v{version}", version):
|
||||
try:
|
||||
ref = json.loads(_fetch(f"https://api.github.com/repos/{owner}/{repo}/git/ref/tags/{tag}", hdrs))
|
||||
obj = ref.get("object", {})
|
||||
sha, typ = obj.get("sha"), obj.get("type")
|
||||
if typ == "tag":
|
||||
t = json.loads(_fetch(f"https://api.github.com/repos/{owner}/{repo}/git/tags/{sha}", hdrs))
|
||||
if t.get("tagger", {}).get("date"):
|
||||
return t["tagger"]["date"]
|
||||
sha = t.get("object", {}).get("sha")
|
||||
c = json.loads(_fetch(f"https://api.github.com/repos/{owner}/{repo}/commits/{sha}", hdrs))
|
||||
return c["commit"]["committer"]["date"]
|
||||
except Exception: # noqa: BLE001 — try the next spelling
|
||||
continue
|
||||
return None
|
||||
|
||||
|
||||
def _fetch(url: str, headers: dict | None = None) -> str:
|
||||
h = {"User-Agent": UA, "Accept-Encoding": "gzip"}
|
||||
h.update(headers or {})
|
||||
@@ -105,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:
|
||||
@@ -114,8 +188,12 @@ def registry_urls(recipe: str, registry_dir: str) -> tuple[list[str], str | None
|
||||
return [], None
|
||||
urls = []
|
||||
for u in re.findall(r"https?://[^\s)|\]]+", text):
|
||||
u = u.rstrip(".,;")
|
||||
if u not in urls:
|
||||
# The registry is MARKDOWN: urls appear inside `backticks`, 'quotes', **bold**, and at the
|
||||
# end of sentences. Trailing punctuation captured into the url makes the fetch 404 and the
|
||||
# recipe render '?' for no real reason — that is what put immich and n8n in the unknown
|
||||
# column on 2026-08-07 (https://docs.n8n.io/release-notes/` ← note the backtick).
|
||||
u = u.rstrip("`'\"*.,;:>)")
|
||||
if u and u not in urls:
|
||||
urls.append(u)
|
||||
return urls, path
|
||||
|
||||
@@ -138,7 +216,11 @@ def github_advisories(urls: list[str]) -> list[dict]:
|
||||
hdrs["Authorization"] = f"Bearer {tok}"
|
||||
entry = {"source": f"github-advisories:{owner}/{repo}", "status": "ok", "advisories": []}
|
||||
try:
|
||||
for a in json.loads(_fetch(api, hdrs)):
|
||||
# PAGINATE. This endpoint caps at 100 per response and IGNORES ?page= — it returns the
|
||||
# same rows again, which silently truncates busy projects (discourse has 286; a hand
|
||||
# count on 2026-08-10 found 123 CVEs in one upgrade window that a single page missed).
|
||||
# Follow the Link rel="next" cursor to exhaustion instead.
|
||||
for a in _gh_paginate(api, hdrs):
|
||||
# An advisory carries ONE ENTRY PER PATCHED RELEASE LINE. n8n patches three
|
||||
# (1.123.32, 2.17.4, 2.18.1); reading only vulnerabilities[0] silently dropped the
|
||||
# line our deployment is actually on, so CVE-2026-42231/42232 classified as
|
||||
@@ -157,6 +239,11 @@ def github_advisories(urls: list[str]) -> list[dict]:
|
||||
filter(None, (v.get("patched_versions") for v in vulns))
|
||||
) 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:
|
||||
@@ -221,7 +308,246 @@ def osv(recipe: str, version: str | None) -> dict | None:
|
||||
return entry
|
||||
|
||||
|
||||
def scan(recipe: str, v_from: str | None, v_to: str | None, registry_dir: str) -> dict:
|
||||
_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)
|
||||
report: dict = {
|
||||
"recipe": recipe,
|
||||
@@ -239,7 +565,9 @@ 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})
|
||||
"vulnerable_range": None, "patched": 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():
|
||||
@@ -253,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"))
|
||||
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"],
|
||||
@@ -271,24 +600,146 @@ def scan(recipe: str, v_from: str | None, v_to: str | None, registry_dir: str) -
|
||||
|
||||
# Classify against the upgrade window when we know it: an advisory is "fixed by this upgrade"
|
||||
# when its patched version is newer than `from` and no newer than `to`.
|
||||
kf, kt = _vkey(v_from), _vkey(v_to)
|
||||
fixed, unknown = [], []
|
||||
#
|
||||
# TWO HARD-WON CONSTRAINTS (2026-08-10, discourse reported a false 133):
|
||||
# a) The window belongs to ONE image. Advisories from OTHER repos in the registry (redis,
|
||||
# postgres, nginx sidecars) must NOT be judged by it — redis CVE-2021-21309, patched in
|
||||
# redis 6.0.11, scored as "fixed" because 6.0.11 sits numerically inside discourse's
|
||||
# 3.5.3 → 2026.7.1 window. Only the PRIMARY app repo is classified; every other source is
|
||||
# reported as unclassified so a human/agent still sees it but it never inflates the count.
|
||||
# b) A version-SCHEME change (semver → calver, 3.5.3 → 2026.7.1) makes numeric ordering
|
||||
# meaningless: 2025.12.2 compares "newer" than 3.5.3 while shipping earlier. When the
|
||||
# leading component jumps by more than SCHEME_JUMP we refuse to classify and say so,
|
||||
# rather than emitting a confident wrong number.
|
||||
# ── Classification ────────────────────────────────────────────────────────────────────────
|
||||
# A recipe upgrades SEVERAL images (app + redis/postgres/nginx sidecars), each with its OWN
|
||||
# version window. Judging every advisory by the app's window is how discourse once reported a
|
||||
# false 133 (34 of them redis CVEs, incl. one patched in redis 6.0.11 in 2021). So each source
|
||||
# is classified against ITS OWN window, and the count is the union across windows.
|
||||
#
|
||||
# --from/--to → the PRIMARY app repo (first github source in the registry)
|
||||
# --image NAME=FROM:TO → any other source whose name contains NAME (repeatable),
|
||||
# e.g. --image redis=7.4:8.10
|
||||
#
|
||||
# A source with no window is not classified: its advisories are listed as unclassified so they
|
||||
# stay visible without inflating the count.
|
||||
gh_sources = [x["source"] for x in report["sources"] if x["source"].startswith("github-advisories:")]
|
||||
primary = gh_sources[0] if (gh_sources and (v_from or v_to)) else None
|
||||
report["primary_source"] = primary
|
||||
|
||||
windows = {} # source name -> (from, to)
|
||||
if primary:
|
||||
windows[primary] = (v_from, v_to)
|
||||
for key, wf, wt in (images or []):
|
||||
for src in gh_sources:
|
||||
if key.lower() in src.lower() and src not in windows:
|
||||
windows[src] = (wf, wt)
|
||||
report["windows"] = {k: {"from": f, "to": t} for k, (f, t) in windows.items()}
|
||||
|
||||
def _classify_window(src, wf, wt):
|
||||
"""Return (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, undecidable = set(), set()
|
||||
for cve, e in report["cves"].items():
|
||||
if src not in e["sources"]:
|
||||
continue
|
||||
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)
|
||||
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.
|
||||
owner, _, repo = src.split(":", 1)[1].partition("/")
|
||||
d_from, d_to = _tag_date(owner, repo, wf), _tag_date(owner, repo, wt)
|
||||
if d_from and d_to and d_from < d_to:
|
||||
got = {cve for cve, e in report["cves"].items()
|
||||
if src in e["sources"] and e.get("published_at")
|
||||
and d_from < e["published_at"] <= d_to}
|
||||
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, undecidable = _classify_window(src, wf, wt)
|
||||
indeterminate |= undecidable
|
||||
methods[src] = method
|
||||
if dw:
|
||||
date_windows[src] = {"from": dw[0], "to": dw[1]}
|
||||
if unresolved:
|
||||
unresolved_any = True
|
||||
for cve in got:
|
||||
report["cves"][cve]["classification"] = f"fixed-by-this-upgrade ({method}) via {src}"
|
||||
fixed_set.add(cve)
|
||||
|
||||
unknown = []
|
||||
for cve, e in report["cves"].items():
|
||||
# `patched_versions` is a RANGE EXPRESSION (">= 2.18.1"), not a bare version, and there may
|
||||
# be several (one per patched release line, joined with ";"). Pull every version-looking
|
||||
# token and treat the advisory as fixed-by-this-upgrade if ANY of them lands in (from, to].
|
||||
cands = [_vkey(t) for t in re.findall(r"\d+(?:\.\d+)*", e.get("patched") or "")]
|
||||
kp = next((c for c in cands if kf and kt and kf < c <= kt), None) or (cands[0] if cands else ())
|
||||
if kf and kt and kp and kf < kp <= kt:
|
||||
e["classification"] = "fixed-by-this-upgrade"
|
||||
fixed.append(cve)
|
||||
if cve in fixed_set:
|
||||
continue
|
||||
if not any(src in e["sources"] for src in windows):
|
||||
e["classification"] = "unclassified: no versions given for this image"
|
||||
unknown.append(cve)
|
||||
else:
|
||||
e["classification"] = "unclassified" if not (kf and kt and kp) else "outside-window"
|
||||
if e["classification"] == "unclassified":
|
||||
e.setdefault("classification", "outside-window")
|
||||
if e["classification"] == "outside-window":
|
||||
pass
|
||||
else:
|
||||
unknown.append(cve)
|
||||
report["fixed_by_this_upgrade"] = sorted(fixed)
|
||||
|
||||
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)
|
||||
report["cve_count_fixed"] = len(fixed)
|
||||
# NEVER report 0 for something we could not determine — a 0 asserts safety. If ANY requested
|
||||
# window could not be ordered at all, the total is UNKNOWN rather than a partial number.
|
||||
report["count_known"] = not unresolved_any
|
||||
report["cve_count_fixed"] = len(fixed_set) if not unresolved_any else None
|
||||
report["cve_count_total_seen"] = len(report["cves"])
|
||||
# Only GENUINE failures make a count unreliable. "no-advisories-published" (404: the repo has
|
||||
# no advisory feed) and "skipped: template URL" are benign and must not degrade the verdict.
|
||||
@@ -309,8 +760,34 @@ def markdown(rep: dict) -> str:
|
||||
"""Human/agent-readable block for pasting into the per-recipe upgrade log."""
|
||||
L = [f"### Advisory scan (deterministic pre-step) — {rep['recipe']} "
|
||||
f"{rep.get('from') or '?'} → {rep.get('to') or '?'}"]
|
||||
if not rep.get("count_known", True):
|
||||
L.append("\n**CVEs fixed by this upgrade: UNKNOWN — the scan could NOT determine a count.**")
|
||||
L.append("\n⚠ This is NOT zero. A version-scheme change (e.g. semver → calver) makes numeric "
|
||||
"ordering meaningless across this jump, so no advisory could be classified. Render "
|
||||
"this recipe's cve cell as `?`, never `0`. Read the vendor's release notes for the "
|
||||
"jump and count by hand.")
|
||||
if rep["unclassified"]:
|
||||
L.append(f"\nAdvisories seen but unclassifiable ({len(rep['unclassified'])}) — includes "
|
||||
f"other images in this recipe: " + ", ".join(rep["unclassified"][:12]))
|
||||
if rep["sources_failed"]:
|
||||
L.append(f"\n⚠ sources that FAILED: {', '.join(rep['sources_failed'])}")
|
||||
L.append(f"\n_Sources checked: {len(rep['sources'])} ({rep['registry_urls']} registry URLs + "
|
||||
f"advisory APIs). This scan is ADDITIVE — it does not replace the release-note "
|
||||
f"reading in the upgrade step._")
|
||||
return "\n".join(L)
|
||||
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():
|
||||
dw = (rep.get("date_window") or {}).get(src)
|
||||
win = (rep.get("windows") or {}).get(src, {})
|
||||
span = f"{win.get('from')} → {win.get('to')}"
|
||||
extra = (f" (dates {dw['from'][:10]} → {dw['to'][:10]})" if dw else "")
|
||||
L.append(f"_{src.split(':',1)[-1]}: {span} — counted by {method}{extra}._")
|
||||
L.append("")
|
||||
L.append("| CVE | severity | fixed in | advisory | source |")
|
||||
L.append("|---|---|---|---|---|")
|
||||
for cve in rep["fixed_by_this_upgrade"]:
|
||||
@@ -319,8 +796,18 @@ 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'])}): "
|
||||
L.append(f"\nSeen but not version-classified ({len(rep['unclassified'])}) — includes advisories "
|
||||
f"from OTHER images in this recipe (sidecars), which this window cannot judge: "
|
||||
+ ", ".join(rep["unclassified"][:12]))
|
||||
if rep["sources_failed"]:
|
||||
L.append(f"\n⚠ sources that FAILED (treat counts as incomplete): {', '.join(rep['sources_failed'])}")
|
||||
@@ -336,10 +823,35 @@ 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. "
|
||||
"--image redis=7.4:8.10 (repeatable). NAME matches a source repo name; "
|
||||
"its advisories are then counted against ITS OWN versions instead of "
|
||||
"being left unclassified.")
|
||||
a = ap.parse_args()
|
||||
rep = scan(a.recipe, a.v_from, a.v_to, a.registry)
|
||||
print(json.dumps(rep, indent=2) if a.json else markdown(rep))
|
||||
images = []
|
||||
for spec in a.image:
|
||||
name, _, rng = spec.partition('=')
|
||||
vf, _, vt = rng.partition(':')
|
||||
if name and vf and vt:
|
||||
images.append((name, vf, vt))
|
||||
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)
|
||||
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
|
||||
|
||||
|
||||
|
||||
Executable
+151
@@ -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)
|
||||
Executable
+623
@@ -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()
|
||||
@@ -2,7 +2,7 @@
|
||||
|
||||
| service | image | source repo | releases / changelog |
|
||||
|----------|-------------------------------------------|---------------------------------------------------|-------------------------------------------------------------------|
|
||||
| app | mattermost/mattermost-team-edition | https://github.com/mattermost/mattermost | https://docs.mattermost.com/about/mattermost-changelog.html |
|
||||
| app | mattermost/mattermost-team-edition | https://github.com/mattermost/mattermost | https://docs.mattermost.com/deploy/mattermost-changelog.html |
|
||||
| postgres | postgres | https://github.com/postgres/postgres | https://www.postgresql.org/docs/release/ |
|
||||
|
||||
## Standing notes
|
||||
|
||||
Reference in New Issue
Block a user