From f912b3ad54dfd45630bbc104cc2702aaf2069284 Mon Sep 17 00:00:00 2001 From: maksimtech <268127343+maksimtech@users.noreply.github.com> Date: Sat, 17 Oct 2026 17:02:06 +0200 Subject: [PATCH 1/8] feat(verdict): keep the verdict per product, and judge over HTTP too PR-01a judged each CVE against the installed version in memory; the next `status`, /api/cves and the web UI went back to CVEs by name, and a scan over HTTP did not judge at all, so the two scans disagreed. - cve_software gains verdict, verdict_reason, verdict_fixed_version and judged_version, migrated into existing databases: per product, since one CVE can affect one watched product and not another. The latest scan decides it. - get_cves shows a verdict only while the installed version is still the one it was given for: a stale "affected" sends someone to patch what is patched, a stale "not affected" is worse. - affected.mark is the one place both scans judge. - The API takes version and cpe on POST /api/watchlist/{name} with the CLI's checks, lists entries on GET /api/watchlist, and returns the split per product from POST /api/scan. `status` gets a Verdict column. Tests first, seen failing; 15 of 15 mutants killed. Co-Authored-By: Claude Opus 5.5 (1M context) --- CHANGELOG.md | 6 ++ README.md | 6 +- docs/DESIGN.md | 19 +++- src/patchradar/affected.py | 28 +++++- src/patchradar/api/main.py | 58 ++++++++++-- src/patchradar/cli.py | 20 ++-- src/patchradar/db/database.py | 63 ++++++++++--- tests/test_api_versioned.py | 123 +++++++++++++++++++++++++ tests/test_verdict_stored.py | 168 ++++++++++++++++++++++++++++++++++ 9 files changed, 451 insertions(+), 40 deletions(-) create mode 100644 tests/test_api_versioned.py create mode 100644 tests/test_verdict_stored.py diff --git a/CHANGELOG.md b/CHANGELOG.md index 49bc466..1f7fc72 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -26,6 +26,12 @@ what has already shipped, and for nothing else. See `RELEASING.md`. split, one what an update closes, and the NVD caveat is always printed. Entries without a version are scanned by name, and the scan says which. `list` shows version and CPE. The verdict is not stored yet (§5.5): `status`, the API and the web UI are the next step. +- **The verdict is kept, and the API judges too** (docs/DESIGN.md §5.5). In `cve_software`, per + product, with the installed version it was given for: after a new `--version` the old verdict + is not shown until a scan judges the new one. `patchradar status` and `/api/cves` show it; + `POST /api/watchlist/{name}` takes `version` and `cpe` with the CLI's checks, `GET + /api/watchlist` adds `entries`, and `POST /api/scan` judges with the CLI's function + (`affected.mark`) and returns the split per product in `verdicts`. - **docs/DESIGN.md.** The code referred to "the versionradar design, §5" for these columns; the text was never committed and could not be found. It is reconstructed from what `affected.py`, `inventory.py` and `tests/test_priority_wiring.py` already committed to. diff --git a/README.md b/README.md index d3e195c..146a7a7 100644 --- a/README.md +++ b/README.md @@ -98,8 +98,10 @@ affected ones — the rest stay open because NVD does not state where they were fixed, not because the update falls short — and a count from NVD is always followed by the reminder that it counts identifiers, not flaws. Entries without a version are scanned by name as before, and the scan names them. The verdict -is not stored yet, so `status`, the API and the web UI still show CVEs by name -([docs/DESIGN.md](docs/DESIGN.md) §5.5). +is stored with the version it was given for: `patchradar status` and +`/api/cves` show it while that version is still the installed one, and the API +takes the version and CPE too (`POST /api/watchlist/?version=…&cpe=…`). +The web UI does not show it yet ([docs/DESIGN.md](docs/DESIGN.md) §5.5, §5.6). ### Asking Debian what it is actually doing diff --git a/docs/DESIGN.md b/docs/DESIGN.md index 337d3c8..479e72a 100644 --- a/docs/DESIGN.md +++ b/docs/DESIGN.md @@ -104,16 +104,25 @@ The total line keeps its meaning: CVEs found by name. ### 5.5 What is stored -PR-01a judges in memory, during the scan. Storing the verdict — so that -`status`, the API and the web UI can show it — needs columns in `cves` and is the -migration `tests/test_priority_wiring.py` waits for, together with the KEV fields. +The verdict is kept on the link between a CVE and a product, `cve_software` +(`verdict`, `verdict_reason`, `verdict_fixed_version`, `judged_version`), because +one CVE stored once can be affected for one watched product and not for another. +The latest scan decides it. It is read back only while the installed version is +still `judged_version`: after `add --version` with a new version, the old verdict +is not shown until a scan judges the new one. `status` and `/api/cves` show it; +the API records version and CPE (`POST /api/watchlist/{name}?version=&cpe=`) and +judges with the same function as the CLI (`affected.mark`). + +Still in memory only: the KEV fields that decide a rank — the migration +`tests/test_priority_wiring.py` waits for. ### 5.6 Order of work 1. **PR-01a** — the columns, `add --version --cpe`, the verdict for NVD records in the command-line scan, `--all`, the README. -2. **PR-01b** — KEV and MSRC entries judged through NVD's record of the same - CVE; the verdict and the KEV fields stored; the API. +2. **PR-01b** — in two parts: the verdict stored and the API (01b-1); the KEV + fields stored, with the database-backed listings ranked by priority, and KEV + and MSRC entries judged through NVD's record of the same CVE (01b-2). 3. **PR-01c** — the web UI; `patchradar import` writing the mappings a person confirmed, `channel` included. diff --git a/src/patchradar/affected.py b/src/patchradar/affected.py index 9cc052e..70b8ec6 100644 --- a/src/patchradar/affected.py +++ b/src/patchradar/affected.py @@ -26,11 +26,11 @@ ever assigned an identifier. `Summary.caveat` carries that sentence so a caller cannot print "no known vulnerabilities" without it. -**Wired into the command-line scan** for a watchlist entry that has both an -installed version and a CPE (docs/DESIGN.md §5): `judge` gives each CVE found by -name one of three verdicts — affected, not affected, not evaluable — and -`closed_by` says what one update closes. Storing the verdict, and the API, are -the next step (§5.5); `tests/test_priority_wiring.py` holds that boundary. +**Wired into both scans**, the CLI's and the API's, for a watchlist entry that +has both an installed version and a CPE (docs/DESIGN.md §5): `judge` gives each +CVE found by name one of three verdicts — affected, not affected, not evaluable +— `mark` applies it to a scan's records, and `closed_by` says what one update +closes. The verdict is stored with the version it was given for (§5.5). """ from __future__ import annotations @@ -451,3 +451,21 @@ def judge(record: dict, *, version: str, vendor: str, cpe_product: str) -> Verdi f"compared with {version}") return Verdict(NOT_AFFECTED, f"NVD states the affected versions ({stated}); " f"{version} is not among them") + + +def mark(records: list[dict], installed: tuple[str, str, str] | None) -> None: + """Give each scan record its verdict for `installed` — (version, vendor, + product), or None for an entry matched by name — and take NVD's + configurations off it, which nothing after the verdict reads. + + The one place the CLI and the API judge, so the two cannot disagree. The + record keeps the version it was judged for, which is what lets a stored + verdict be told apart from one about a version no longer installed. + """ + for record in records: + if installed is not None: + version, vendor, product = installed + verdict = judge(record, version=version, vendor=vendor, cpe_product=product) + record.update(verdict=verdict.status, verdict_reason=verdict.reason, + verdict_fixed_version=verdict.fixed_version, judged_version=version) + record.pop("configurations", None) diff --git a/src/patchradar/api/main.py b/src/patchradar/api/main.py index 31fe0d9..97cbf60 100644 --- a/src/patchradar/api/main.py +++ b/src/patchradar/api/main.py @@ -15,6 +15,7 @@ from fastapi.security import APIKeyHeader from fastapi.staticfiles import StaticFiles +from patchradar.affected import AFFECTED, NOT_AFFECTED, NOT_EVALUABLE, mark, orderable from patchradar.collectors.debian import fetch_cves as debian_fetch from patchradar.collectors.errors import CollectorError from patchradar.collectors.kev import fetch_cves as kev_fetch @@ -22,7 +23,17 @@ from patchradar.collectors.nvd import fetch_cves as nvd_fetch from patchradar.collectors.vuldb import api_key as vuldb_api_key from patchradar.collectors.vuldb import fetch_cves as vuldb_fetch -from patchradar.db.database import add_to_watchlist, get_cves, get_watchlist, init_db, remove_from_watchlist +from patchradar.cpe import CpeError, parse_cpe +from patchradar.db.database import ( + WatchEntry, + add_to_watchlist, + get_cves, + get_watch_entries, + get_watchlist, + init_db, + remove_from_watchlist, + set_installed, +) from patchradar.epss import enrich as epss_enrich from patchradar.names import SOFTWARE_NAME_MAX_LENGTH, SOFTWARE_NAME_PATTERN, normalise_software_name from patchradar.priority import merge_by_cve @@ -194,6 +205,9 @@ def sanitize_cve(cve: dict) -> dict: # Numeric fields. The EPSS pair is passed through as stored, including a # missing one: `.get` yields None, and None here means FIRST does not score # the CVE — which a 0.0 would misreport as "almost certainly not exploited". + for field in ("verdict", "verdict_reason", "verdict_fixed_version", "judged_version"): + value = cve.get(field) + safe[field] = _strip_control_chars(value).strip()[:300] if isinstance(value, str) else None safe["cvss_score"] = cve.get("cvss_score") safe["epss_score"] = cve.get("epss_score") safe["epss_percentile"] = cve.get("epss_percentile") @@ -202,7 +216,11 @@ def sanitize_cve(cve: dict) -> dict: @app.get("/api/watchlist") async def api_watchlist(): items = await get_watchlist() - return {"watchlist": items} + # `watchlist` keeps its shape for what already reads it; `entries` adds + # what is installed (docs/DESIGN.md §5.1). + entries = [{"name": e.name, "installed_version": e.installed_version, "cpe": e.cpe} + for e in await get_watch_entries()] + return {"watchlist": items, "entries": entries} @app.post( "/api/watchlist/import", @@ -243,14 +261,29 @@ async def api_watchlist_import(payload: dict): async def api_add( software: str = Path( ..., min_length=1, max_length=SOFTWARE_NAME_MAX_LENGTH, pattern=SOFTWARE_NAME_PATTERN - ) + ), + version: str | None = Query(None, min_length=1, max_length=64), + cpe: str | None = Query(None, min_length=1, max_length=256), ): # Normalise here too, so both routes store the same canonical value. name = normalise_software_name(software) if name is None: raise HTTPException(status_code=422, detail="Invalid software name") + # The same checks as `patchradar add`, before anything is written. + if version is not None and not orderable(version): + raise HTTPException(status_code=422, detail=f"{version} is not a version the scan can compare") + parsed = None + if cpe is not None: + try: + parsed = parse_cpe(cpe) + except CpeError as exc: + raise HTTPException(status_code=422, detail=str(exc)) from None added = await add_to_watchlist(name) - return {"added": added, "software": name} + if version is not None or parsed is not None: + await set_installed(name, version=version, cpe=parsed) + entry = next(e for e in await get_watch_entries() if e.name == name) + return {"added": added, "software": name, + "installed_version": entry.installed_version, "cpe": entry.cpe} @app.delete( @@ -272,7 +305,8 @@ async def api_cves(software: str = Query(None, min_length=1, max_length=100), li sanitized = [sanitize_cve(c) for c in cves] return {"cves": sanitized, "total": len(sanitized)} -async def _scan_one(sw: str, days: int, errors: list[dict]) -> int: +async def _scan_one(sw: str, days: int, errors: list[dict], entry: WatchEntry | None = None, + verdicts: dict | None = None) -> int: """Collect and persist every source for a single package. A source that fails is recorded in `errors` rather than silently counted @@ -321,6 +355,13 @@ async def _scan_one(sw: str, days: int, errors: list[dict]) -> int: logger.warning("scan of %r unranked by EPSS: %s", sw, exc) errors.append({"software": sw, "source": exc.source, "reason": exc.reason, "status": exc.status}) + # Judged by the function the CLI judges with, before anything is stored, so + # the verdict lands with the link and the two scans give one answer. + installed = entry.installed() if entry is not None else None + mark(collected, installed) + if installed is not None and verdicts is not None: + found = [cve["verdict"] for cve in collected] + verdicts[sw] = {status: found.count(status) for status in (AFFECTED, NOT_EVALUABLE, NOT_AFFECTED)} for cve in collected: await save_cve(cve) return len(collected) @@ -334,6 +375,8 @@ async def _scan_one(sw: str, days: int, errors: list[dict]) -> int: ) async def api_scan(days: int = Query(7, ge=1, le=90)): watchlist = await get_watchlist() + entries = {entry.name: entry for entry in await get_watch_entries()} + verdicts: dict[str, dict[str, int]] = {} total = 0 results = {} errors: list[dict] = [] @@ -343,7 +386,7 @@ async def api_scan(days: int = Query(7, ge=1, le=90)): # scan that never returns pins a worker indefinitely. async with asyncio.timeout(SCAN_TIMEOUT_SECONDS): for sw in watchlist: - count = await _scan_one(sw, days, errors) + count = await _scan_one(sw, days, errors, entries.get(sw), verdicts) results[sw] = count total += count except TimeoutError: @@ -357,6 +400,9 @@ async def api_scan(days: int = Query(7, ge=1, le=90)): return { "total": total, "by_software": results, + # Per product judged against its installed version: how many of the + # CVEs found by name affect it, could not be evaluated, do not affect it. + "verdicts": verdicts, "timed_out": timed_out, "scanned": len(results), "watched": len(watchlist), diff --git a/src/patchradar/cli.py b/src/patchradar/cli.py index f80333c..af0bc89 100644 --- a/src/patchradar/cli.py +++ b/src/patchradar/cli.py @@ -14,7 +14,7 @@ from rich.text import Text import patchradar -from patchradar.affected import AFFECTED, CAVEAT, NOT_AFFECTED, NOT_EVALUABLE, closed_by, judge, orderable +from patchradar.affected import AFFECTED, CAVEAT, NOT_AFFECTED, NOT_EVALUABLE, closed_by, mark, orderable from patchradar.collectors.debian import DEBIAN_RELEASE as debian_release from patchradar.collectors.debian import get_tracker_snapshot from patchradar.collectors.errors import CollectorError @@ -419,15 +419,7 @@ async def _scan_target( # listed, unless --all asks for them. installed = entry.installed() if entry is not None else None judged = installed is not None - if installed is not None: - version, vendor, product = installed - for cve in all_cves: - verdict = judge(cve, version=version, vendor=vendor, cpe_product=product) - cve["verdict"] = verdict.status - cve["verdict_reason"] = verdict.reason - cve["verdict_fixed_version"] = verdict.fixed_version - for cve in all_cves: - cve.pop("configurations", None) + mark(all_cves, installed) for cve in all_cves: await save_cve(cve) if collected is not None: @@ -637,11 +629,18 @@ def _print_cves_table(cves: list): table.add_column("Score", justify="center", width=6) table.add_column("Severity", justify="center", width=10) table.add_column("Source", width=6) + # Only when a stored verdict is current for some row: a watchlist matched by + # name alone gets the table it always had. + judged = any(cve.get("verdict") for cve in cves) + if judged: + table.add_column("Verdict") for cve in cves: score_str = _format_score(cve.get("cvss_score")) severity = _normalise_severity(cve.get("severity")) severity_color = SEVERITY_STYLES.get(severity, "white") + verdict = cve.get("verdict") + extra = [Text(verdict or "—", style=VERDICT_STYLES.get(verdict, "dim"))] if judged else [] table.add_row( Text(str(cve.get("id", ""))), @@ -649,6 +648,7 @@ def _print_cves_table(cves: list): score_str, Text(severity, style=severity_color), Text(str(cve.get("source") or "")), + *extra, ) console.print(table) diff --git a/src/patchradar/db/database.py b/src/patchradar/db/database.py index a02ec01..af5aadf 100644 --- a/src/patchradar/db/database.py +++ b/src/patchradar/db/database.py @@ -65,6 +65,18 @@ def normalize_timestamp(value) -> str | None: } +# The verdict on a CVE for one watched product (docs/DESIGN.md §5.5), on the +# link between the two: one CVE stored once is "affected" for one product and +# "not affected" for another. `judged_version` is the installed version it was +# given for, so a verdict about 8.5.0 is never shown as one about 8.9.6.4. +_ADDED_LINK_COLUMNS = { + "verdict": "TEXT", + "verdict_reason": "TEXT", + "verdict_fixed_version": "TEXT", + "judged_version": "TEXT", +} + + async def _add_missing_columns(db, table: str = "cves", columns: dict | None = None) -> None: """Bring an existing table up to the current schema.""" columns = _ADDED_COLUMNS if columns is None else columns @@ -148,11 +160,16 @@ async def init_db(): CREATE TABLE IF NOT EXISTS cve_software ( cve_id TEXT NOT NULL, software TEXT NOT NULL, + verdict TEXT, + verdict_reason TEXT, + verdict_fixed_version TEXT, + judged_version TEXT, PRIMARY KEY (cve_id, software) ) """) await _add_missing_columns(db) await _add_missing_columns(db, "watchlist", _ADDED_WATCHLIST_COLUMNS) + await _add_missing_columns(db, "cve_software", _ADDED_LINK_COLUMNS) await _normalize_stored_timestamps(db) await _link_stored_cves(db) await db.commit() @@ -272,10 +289,20 @@ async def save_cve(cve: dict) -> bool: """ async with aiosqlite.connect(DB_PATH) as db: try: - await db.execute( - "INSERT OR IGNORE INTO cve_software (cve_id, software) VALUES (?, ?)", - (cve["id"], cve["software"]), - ) + # The latest scan decides the verdict, unlike the description: NVD + # analysing a CVE turns "not evaluable" into an answer, and a scan by + # name, with no verdict, says the old one no longer applies. + await db.execute(""" + INSERT INTO cve_software + (cve_id, software, verdict, verdict_reason, verdict_fixed_version, judged_version) + VALUES (?, ?, ?, ?, ?, ?) + ON CONFLICT(cve_id, software) DO UPDATE SET + verdict = excluded.verdict, + verdict_reason = excluded.verdict_reason, + verdict_fixed_version = excluded.verdict_fixed_version, + judged_version = excluded.judged_version + """, (cve["id"], cve["software"], cve.get("verdict"), cve.get("verdict_reason"), + cve.get("verdict_fixed_version"), cve.get("judged_version"))) cursor = await db.execute(""" INSERT INTO cves (id, software, description, cvss_score, cvss_version, @@ -301,24 +328,36 @@ async def save_cve(cve: dict) -> bool: logger.debug("save_cve failure", exc_info=True) return False +# The verdict columns of the link, shown only while the installed version is +# still the one they were given for; NULL otherwise, as for a CVE never judged. +_CURRENT_VERDICT = ", ".join( + f"CASE WHEN link.judged_version = watched.installed_version THEN link.{column} END AS {column}" + for column in _ADDED_LINK_COLUMNS +) + + async def get_cves(software: str | None = None, limit: int = 50) -> list[dict]: async with aiosqlite.connect(DB_PATH) as db: db.row_factory = aiosqlite.Row if software: # The column as well as the links, so a row written without a link # (by hand, or by an older release after the migration) still shows. - async with db.execute(""" - SELECT * FROM cves - WHERE software = ? OR id IN (SELECT cve_id FROM cve_software WHERE software = ?) + async with db.execute(f""" + SELECT cves.*, {_CURRENT_VERDICT} FROM cves + LEFT JOIN cve_software AS link ON link.cve_id = cves.id AND link.software = ? + LEFT JOIN watchlist AS watched ON watched.name = ? + WHERE cves.software = ? OR cves.id IN (SELECT cve_id FROM cve_software WHERE software = ?) ORDER BY published_at DESC LIMIT ? - """, (software.lower(), software.lower(), limit)) as cursor: + """, (software.lower(),) * 4 + (limit,)) as cursor: rows = await cursor.fetchall() # Named after the product asked about, not the first one to find it. return [{**dict(row), "software": software.lower()} for row in rows] else: - async with db.execute( - "SELECT * FROM cves ORDER BY published_at DESC LIMIT ?", - (limit,) - ) as cursor: + async with db.execute(f""" + SELECT cves.*, {_CURRENT_VERDICT} FROM cves + LEFT JOIN cve_software AS link ON link.cve_id = cves.id AND link.software = cves.software + LEFT JOIN watchlist AS watched ON watched.name = cves.software + ORDER BY published_at DESC LIMIT ? + """, (limit,)) as cursor: rows = await cursor.fetchall() return [dict(row) for row in rows] diff --git a/tests/test_api_versioned.py b/tests/test_api_versioned.py new file mode 100644 index 0000000..f331bb3 --- /dev/null +++ b/tests/test_api_versioned.py @@ -0,0 +1,123 @@ +""" +PatchRadar — the API knows what is installed, and judges as the CLI does (DESIGN §5.5) + +A scan over HTTP and a scan from the command line that give different answers +for the same machine are the failure `tests/test_collector_wiring.py` exists +for. PR-01a taught only the CLI to judge CVEs against the installed version; +the API now records the version and CPE, judges with the same function, and +returns what it stored. + +NVD answers from the recording `nvd_notepadpp_configurations.json`; MSRC has no +document (404); Debian refuses (503), so the API lists that one source in +`errors`; KEV is the empty catalogue conftest.py caches. +""" +from __future__ import annotations + +import json +import pathlib + +import httpx +import pytest +import respx +from httpx import ASGITransport, AsyncClient + +import patchradar.api.main as api +from patchradar.collectors import msrc, nvd +from patchradar.db import database + +FIXTURE = pathlib.Path(__file__).parent / "fixtures" / "nvd_notepadpp_configurations.json" +PAYLOAD = json.loads(FIXTURE.read_text(encoding="utf-8")) +NOTEPADPP = "cpe:2.3:a:notepad-plus-plus:notepad\\+\\+" + + +@pytest.fixture +async def client(): + async with AsyncClient(transport=ASGITransport(app=api.app), base_url="http://test") as ac: + yield ac + + +async def _add(client, **params): + return await client.post("/api/watchlist/notepad", params=params) + + +async def test_the_version_and_cpe_are_recorded_over_http(client): + response = await _add(client, version="8.5.0", cpe=NOTEPADPP) + + assert response.status_code == 200, response.text + assert response.json()["installed_version"] == "8.5.0" + [entry] = await database.get_watch_entries() + assert (entry.installed_version, entry.vendor, entry.product_id) == ("8.5.0", "notepad-plus-plus", + "notepad\\+\\+") + + +async def test_a_cpe_nvd_would_match_nothing_with_is_refused_with_its_form(client): + response = await _add(client, version="8.5.0", cpe="cpe:2.3:a:notepad-plus-plus:notepad++") + + assert response.status_code == 422 + assert "notepad\\+\\+" in response.json()["detail"] + assert await database.get_watchlist() == [] + + +async def test_a_version_the_scan_cannot_order_is_refused(client): + response = await _add(client, version="8.x-beta", cpe=NOTEPADPP) + + assert response.status_code == 422 + assert await database.get_watchlist() == [] + + +async def test_adding_again_with_a_version_updates_it(client): + await _add(client) + response = await _add(client, version="8.9.6.4") + + assert response.json()["added"] is False + assert response.json()["installed_version"] == "8.9.6.4" + + +async def test_the_watchlist_lists_what_is_installed_beside_the_names(client): + await _add(client, version="8.5.0", cpe=NOTEPADPP) + + body = (await client.get("/api/watchlist")).json() + + assert body["watchlist"] == ["notepad"] + assert body["entries"] == [{"name": "notepad", "installed_version": "8.5.0", "cpe": NOTEPADPP}] + + +@pytest.fixture +def upstream(): + with respx.mock(assert_all_called=False) as router: + router.get(url__startswith=nvd.NVD_API).mock(return_value=httpx.Response(200, json=PAYLOAD)) + router.get(url__startswith=msrc.MSRC_API).mock(return_value=httpx.Response(404)) + router.route().mock(return_value=httpx.Response(503)) # Debian + yield + + +async def test_a_scan_over_http_gives_the_split_the_cli_gives(client, upstream): + await _add(client, version="8.5.0", cpe=NOTEPADPP) + + body = (await client.post("/api/scan", params={"days": 7})).json() + + assert body["by_software"] == {"notepad": 21} + assert body["verdicts"] == {"notepad": {"affected": 10, "not evaluable": 0, "not affected": 11}} + + +async def test_the_cves_read_back_over_http_carry_their_verdict(client, upstream): + await _add(client, version="8.5.0", cpe=NOTEPADPP) + await client.post("/api/scan", params={"days": 7}) + + rows = (await client.get("/api/cves", params={"software": "notepad"})).json()["cves"] + + assert len(rows) == 21 + verdicts = [row["verdict"] for row in rows] + assert (verdicts.count("affected"), verdicts.count("not affected")) == (10, 11) + assert all(row["verdict_reason"] for row in rows) + + +async def test_an_entry_without_a_version_has_no_verdicts_over_http(client, upstream): + await _add(client) + + body = (await client.post("/api/scan", params={"days": 7})).json() + + assert body["by_software"] == {"notepad": 21} + assert body["verdicts"] == {} + rows = (await client.get("/api/cves", params={"software": "notepad"})).json()["cves"] + assert {row["verdict"] for row in rows} == {None} diff --git a/tests/test_verdict_stored.py b/tests/test_verdict_stored.py new file mode 100644 index 0000000..fedde1b --- /dev/null +++ b/tests/test_verdict_stored.py @@ -0,0 +1,168 @@ +""" +PatchRadar — the verdict outlives the scan (docs/DESIGN.md §5.5) + +PR-01a judged each CVE against the installed version in memory, and the next +`patchradar status`, `/api/cves` and the web UI went back to CVEs by name. The +verdict is now kept beside the link between a CVE and a product, in +`cve_software`, because that is where it belongs: one CVE stored once can be +"affected" for one watched product and "not affected" for another. + +It is kept with the version it was given for. A verdict about 8.5.0 is not a +verdict about 8.9.6.4: after `add notepad --version 8.9.6.4`, the stored one is +not shown until a scan judges the new version — a stale "affected" would send +someone to patch what is already patched, and a stale "not affected" is worse. +""" +from __future__ import annotations + +import asyncio +import json +import pathlib + +import aiosqlite +import httpx +import pytest +import respx +from typer.testing import CliRunner + +from patchradar import cli +from patchradar.collectors import msrc, nvd +from patchradar.cpe import parse_cpe +from patchradar.db import database + +FIXTURE = pathlib.Path(__file__).parent / "fixtures" / "nvd_notepadpp_configurations.json" +PAYLOAD = json.loads(FIXTURE.read_text(encoding="utf-8")) +NOTEPADPP = "cpe:2.3:a:notepad-plus-plus:notepad\\+\\+" +VERDICT_COLUMNS = {"verdict", "verdict_reason", "verdict_fixed_version", "judged_version"} + + +def _record(cve_id="CVE-2023-40031", software="notepad", **verdict): + return {"id": cve_id, "software": software, "description": "x", "cvss_score": 7.8, + "cvss_version": "3.1", "severity": "HIGH", "published_at": "2023-08-25T00:00:00Z", + "source": "NVD", "url": f"https://nvd.nist.gov/vuln/detail/{cve_id}", **verdict} + + +AFFECTED_850 = {"verdict": "affected", "verdict_reason": "last affected version 8.5.6; NVD does " + "not state which release fixed it", "verdict_fixed_version": None, + "judged_version": "8.5.0"} + + +async def _watch(version="8.5.0"): + await database.add_to_watchlist("notepad") + await database.set_installed("notepad", version=version, cpe=parse_cpe(NOTEPADPP)) + + +# ─── the migration ─────────────────────────────────────────────────────────── + +@pytest.fixture +def old_links(tmp_path, monkeypatch): + """`cve_software` as it shipped in 2026.44, written out by hand.""" + path = tmp_path / "old.db" + + async def _build(): + async with aiosqlite.connect(path) as db: + await db.execute(""" + CREATE TABLE cve_software ( + cve_id TEXT NOT NULL, + software TEXT NOT NULL, + PRIMARY KEY (cve_id, software) + ) + """) + await db.execute("INSERT INTO cve_software VALUES ('CVE-2023-40031', 'notepad')") + await db.commit() + + asyncio.run(_build()) + monkeypatch.setattr(database, "DB_PATH", path) + return path + + +@pytest.mark.asyncio +async def test_an_existing_link_table_gains_the_verdict_columns(old_links): + await database.init_db() + + async with aiosqlite.connect(old_links) as db, db.execute("PRAGMA table_info(cve_software)") as cursor: + columns = {row[1] for row in await cursor.fetchall()} + assert columns >= VERDICT_COLUMNS + + +# ─── storing and reading it ────────────────────────────────────────────────── + +@pytest.mark.asyncio +async def test_a_judged_cve_is_read_back_with_its_verdict(): + await _watch() + await database.save_cve(_record(**AFFECTED_850)) + + [row] = await database.get_cves(software="notepad") + + assert row["verdict"] == "affected" + assert row["verdict_reason"].startswith("last affected version 8.5.6") + assert row["judged_version"] == "8.5.0" + + +@pytest.mark.asyncio +async def test_the_listing_of_every_product_carries_the_verdict_too(): + await _watch() + await database.save_cve(_record(**AFFECTED_850)) + + [row] = await database.get_cves() + + assert row["verdict"] == "affected" + + +@pytest.mark.asyncio +async def test_a_verdict_for_another_version_is_not_shown(): + await _watch() + await database.save_cve(_record(**AFFECTED_850)) + await database.set_installed("notepad", version="8.9.6.4", cpe=None) + + [row] = await database.get_cves(software="notepad") + + assert row["verdict"] is None + assert row["verdict_reason"] is None + + +@pytest.mark.asyncio +async def test_a_cve_has_its_own_verdict_for_each_product(): + """Stored once, linked twice: judged for notepad, matched by name for the other.""" + await _watch() + await database.add_to_watchlist("scintilla") + await database.save_cve(_record(**AFFECTED_850)) + await database.save_cve(_record(software="scintilla")) + + [for_notepad] = await database.get_cves(software="notepad") + [for_scintilla] = await database.get_cves(software="scintilla") + + assert for_notepad["verdict"] == "affected" + assert for_scintilla["verdict"] is None + + +@pytest.mark.asyncio +async def test_the_latest_scan_decides_the_verdict(): + """A second scan that judges differently — NVD analysed the CVE since — + replaces the first verdict rather than being ignored like the first writer + of a description.""" + await _watch() + await database.save_cve(_record(verdict="not evaluable", verdict_reason="NVD has not yet stated " + "which products it affects", judged_version="8.5.0")) + await database.save_cve(_record(**AFFECTED_850)) + + [row] = await database.get_cves(software="notepad") + + assert row["verdict"] == "affected" + + +# ─── patchradar status ─────────────────────────────────────────────────────── + +def test_status_shows_the_verdict_the_scan_gave(): + runner = CliRunner() + runner.invoke(cli.app, ["add", "notepad", "--version", "8.5.0", "--cpe", NOTEPADPP]) + with respx.mock(assert_all_called=False) as router: + router.get(url__startswith=nvd.NVD_API).mock(return_value=httpx.Response(200, json=PAYLOAD)) + router.get(url__startswith=msrc.MSRC_API).mock(return_value=httpx.Response(404)) + runner.invoke(cli.app, ["scan", "notepad", "--days", "7"]) + + result = runner.invoke(cli.app, ["status"], env={"COLUMNS": "200"}) + + out = " ".join(result.output.split()) + assert "Verdict" in out + assert "affected" in out + assert "not affected" in out From 284d5de8d25698ada7b4062705f743d1beba2f33 Mon Sep 17 00:00:00 2001 From: maksimtech <268127343+maksimtech@users.noreply.github.com> Date: Sat, 17 Oct 2026 17:12:24 +0200 Subject: [PATCH 2/8] perf(verdict): read the current verdict in the join, one query for the watchlist CodSpeed on #59: /api/watchlist 24% slower, get_cves and the /api/cves listings 14-17%. - /api/watchlist read the names and the entries in two queries; the names now come from the entries. - get_cves picked the current verdict with a CASE per column; the version check is now part of the join, so a stale link simply does not match. Measured locally on the benchmark's 500 seeded CVEs: 1.68 ms -> 1.61 ms (main 1.45 ms). The rest is the join and four more fields per row, which is what reading the verdict costs: the same query with NULL columns was already 1.57 ms. The full listing's stale-verdict case had no test of its own: added, and it kills the mutant that drops the version check from that join. Co-Authored-By: Claude Opus 5.5 (1M context) --- src/patchradar/api/main.py | 11 ++++++----- src/patchradar/db/database.py | 17 +++++++++-------- tests/test_verdict_stored.py | 11 +++++++++++ 3 files changed, 26 insertions(+), 13 deletions(-) diff --git a/src/patchradar/api/main.py b/src/patchradar/api/main.py index 97cbf60..54458a6 100644 --- a/src/patchradar/api/main.py +++ b/src/patchradar/api/main.py @@ -215,12 +215,13 @@ def sanitize_cve(cve: dict) -> dict: @app.get("/api/watchlist") async def api_watchlist(): - items = await get_watchlist() # `watchlist` keeps its shape for what already reads it; `entries` adds - # what is installed (docs/DESIGN.md §5.1). - entries = [{"name": e.name, "installed_version": e.installed_version, "cpe": e.cpe} - for e in await get_watch_entries()] - return {"watchlist": items, "entries": entries} + # what is installed (docs/DESIGN.md §5.1). One query for both: two made the + # endpoint 24% slower (CodSpeed, #59). + entries = await get_watch_entries() + return {"watchlist": [e.name for e in entries], + "entries": [{"name": e.name, "installed_version": e.installed_version, "cpe": e.cpe} + for e in entries]} @app.post( "/api/watchlist/import", diff --git a/src/patchradar/db/database.py b/src/patchradar/db/database.py index af5aadf..04f56f2 100644 --- a/src/patchradar/db/database.py +++ b/src/patchradar/db/database.py @@ -328,12 +328,11 @@ async def save_cve(cve: dict) -> bool: logger.debug("save_cve failure", exc_info=True) return False -# The verdict columns of the link, shown only while the installed version is -# still the one they were given for; NULL otherwise, as for a CVE never judged. -_CURRENT_VERDICT = ", ".join( - f"CASE WHEN link.judged_version = watched.installed_version THEN link.{column} END AS {column}" - for column in _ADDED_LINK_COLUMNS -) +# The verdict columns of the link. The joins below match a link only while the +# installed version is still the one its verdict was given for, so a stale one +# reads as NULL, as for a CVE never judged. In the join rather than as a CASE per +# column: four CASEs made every listing 12% slower (CodSpeed, #59). +_CURRENT_VERDICT = ", ".join(f"link.{column}" for column in _ADDED_LINK_COLUMNS) async def get_cves(software: str | None = None, limit: int = 50) -> list[dict]: @@ -344,8 +343,9 @@ async def get_cves(software: str | None = None, limit: int = 50) -> list[dict]: # (by hand, or by an older release after the migration) still shows. async with db.execute(f""" SELECT cves.*, {_CURRENT_VERDICT} FROM cves - LEFT JOIN cve_software AS link ON link.cve_id = cves.id AND link.software = ? LEFT JOIN watchlist AS watched ON watched.name = ? + LEFT JOIN cve_software AS link ON link.cve_id = cves.id AND link.software = ? + AND link.judged_version = watched.installed_version WHERE cves.software = ? OR cves.id IN (SELECT cve_id FROM cve_software WHERE software = ?) ORDER BY published_at DESC LIMIT ? """, (software.lower(),) * 4 + (limit,)) as cursor: @@ -355,8 +355,9 @@ async def get_cves(software: str | None = None, limit: int = 50) -> list[dict]: else: async with db.execute(f""" SELECT cves.*, {_CURRENT_VERDICT} FROM cves - LEFT JOIN cve_software AS link ON link.cve_id = cves.id AND link.software = cves.software LEFT JOIN watchlist AS watched ON watched.name = cves.software + LEFT JOIN cve_software AS link ON link.cve_id = cves.id AND link.software = cves.software + AND link.judged_version = watched.installed_version ORDER BY published_at DESC LIMIT ? """, (limit,)) as cursor: rows = await cursor.fetchall() diff --git a/tests/test_verdict_stored.py b/tests/test_verdict_stored.py index fedde1b..48f35cc 100644 --- a/tests/test_verdict_stored.py +++ b/tests/test_verdict_stored.py @@ -166,3 +166,14 @@ def test_status_shows_the_verdict_the_scan_gave(): assert "Verdict" in out assert "affected" in out assert "not affected" in out + + +@pytest.mark.asyncio +async def test_the_listing_of_every_product_hides_a_stale_verdict_too(): + await _watch() + await database.save_cve(_record(**AFFECTED_850)) + await database.set_installed("notepad", version="8.9.6.4", cpe=None) + + [row] = await database.get_cves() + + assert row["verdict"] is None From 231c0e0360a5000581188f113ecb8b66a66ce825 Mon Sep 17 00:00:00 2001 From: maksimtech <268127343+maksimtech@users.noreply.github.com> Date: Sat, 17 Oct 2026 17:17:05 +0200 Subject: [PATCH 3/8] feat(scan): a CVE from KEV or MSRC alone is judged through NVD's record KEV and MSRC carry no cpeMatch, so for an entry with an installed version a CVE that reached the scan only through them was "not evaluable" - and KEV's are the ones being exploited. - nvd.records_for asks NVD by id (cveId=), paced like every request to NVD through the pacer fetch_cves now shares. - with_records_for_the_rest merges NVD's record into each such row, configurations and score with it; the CLI and the API call it for an entry they can judge, and only then. - When NVD cannot be asked, the CVE stays not evaluable and the scan says so; the API lists it in `errors`. - The NIS2 document's NVD row says the CVE ids that are sent. Tests first, against the recorded cveId answer for CVE-2026-48092 and a KEV entry derived in KEV's shape; seen failing; 9 of 9 mutants killed. Co-Authored-By: Claude Opus 5.5 (1M context) --- CHANGELOG.md | 6 + README.md | 7 +- docs/DESIGN.md | 11 +- docs/nis2-compliance.md | 2 +- src/patchradar/api/main.py | 11 +- src/patchradar/cli.py | 16 ++- src/patchradar/collectors/nvd.py | 102 ++++++++++++---- tests/test_judge_through_nvd.py | 196 +++++++++++++++++++++++++++++++ 8 files changed, 319 insertions(+), 32 deletions(-) create mode 100644 tests/test_judge_through_nvd.py diff --git a/CHANGELOG.md b/CHANGELOG.md index 1f7fc72..e9f543c 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -32,6 +32,12 @@ what has already shipped, and for nothing else. See `RELEASING.md`. `POST /api/watchlist/{name}` takes `version` and `cpe` with the CLI's checks, `GET /api/watchlist` adds `entries`, and `POST /api/scan` judges with the CLI's function (`affected.mark`) and returns the split per product in `verdicts`. +- **A CVE from KEV or MSRC alone is judged through NVD's record of it.** They carry no + `cpeMatch`, so for an entry with an installed version such a CVE could only be "not + evaluable" — and KEV's are the ones being exploited. For an entry that can be judged, the + scan, CLI and API alike, now asks NVD about each by id (`nvd.records_for`, `cveId=`), paced + like every request to NVD, and merges the record in, score included. When NVD cannot be + asked, the CVE stays not evaluable and the scan says so (the API in `errors`). - **docs/DESIGN.md.** The code referred to "the versionradar design, §5" for these columns; the text was never committed and could not be found. It is reconstructed from what `affected.py`, `inventory.py` and `tests/test_priority_wiring.py` already committed to. diff --git a/README.md b/README.md index 146a7a7..7bb1797 100644 --- a/README.md +++ b/README.md @@ -88,9 +88,10 @@ affected versions: - **affected** — listed in the table, with "fixed in X" only where NVD states it; - **not evaluable** — NVD has not yet said which products the CVE affects (the - usual state for weeks after publication), the CVE came from MSRC or KEV alone, - or NVD's versions belong to another scheme: listed, because "cannot tell" is - not an all-clear; + usual state for weeks after publication), NVD could not be asked about a CVE + that came from KEV or MSRC alone — those are looked up in NVD by id — or NVD's + versions belong to another scheme: listed, because "cannot tell" is not an + all-clear; - **not affected** — counted, and listed only with `--all`. One line gives the split, one says which update closes how many of the diff --git a/docs/DESIGN.md b/docs/DESIGN.md index 479e72a..794d694 100644 --- a/docs/DESIGN.md +++ b/docs/DESIGN.md @@ -78,9 +78,14 @@ name is judged against the installed version with NVD's own `configurations`: version of this product is not among them. - **not evaluable** — anything that cannot be decided: NVD has not yet stated which products the CVE affects (no `configurations`, the usual state of a CVE - for weeks after publication), the CVE came from a source other than NVD and - NVD's record was not part of the scan, or NVD's versions belong to another - scheme than the installed one (`2.07` against `16.0.14334`). + for weeks after publication), NVD could not be asked about a CVE that came + from KEV or MSRC alone, or NVD's versions belong to another scheme than the + installed one (`2.07` against `16.0.14334`). + +A CVE that reached the scan only through KEV or MSRC, which carry no +`cpeMatch`, is looked up in NVD by id (`cveId=`) for an entry that can be judged, +and NVD's record is merged in — configurations and score with it. Paced like +every request to NVD; an entry matched by name costs no lookup. "Not evaluable" never becomes "not affected": for a security tool an answer it cannot give is not an absolution. diff --git a/docs/nis2-compliance.md b/docs/nis2-compliance.md index 7e524c2..ba66570 100644 --- a/docs/nis2-compliance.md +++ b/docs/nis2-compliance.md @@ -107,7 +107,7 @@ cosa ricevono: | Host | Quando | Cosa riceve | | --- | --- | --- | -| `services.nvd.nist.gov` | `scan`, da riga di comando o dall'interfaccia web | il nome di ogni software della watchlist (o quello passato a `scan`) come parola chiave di ricerca, con l'intervallo di date; la chiave `NVD_API_KEY`, se impostata, in un'intestazione | +| `services.nvd.nist.gov` | `scan`, da riga di comando o dall'interfaccia web | il nome di ogni software della watchlist (o quello passato a `scan`) come parola chiave di ricerca, con l'intervallo di date; per un software con versione installata e CPE, l'identificativo di ogni CVE trovata solo da CISA KEV o MSRC (`cveId`); la chiave `NVD_API_KEY`, se impostata, in un'intestazione | | `api.msrc.microsoft.com` | `scan` | il mese dei bollettini richiesti; nessun nome di software, il filtro avviene sulla macchina | | `www.cisa.gov` | `scan` | nulla: scarica il catalogo KEV intero e lo filtra sulla macchina | | `security-tracker.debian.org` | `scan` dall'interfaccia web; `patchradar debian` senza `--file` | nulla: scarica il tracker intero e lo filtra sulla macchina | diff --git a/src/patchradar/api/main.py b/src/patchradar/api/main.py index 54458a6..b2beabd 100644 --- a/src/patchradar/api/main.py +++ b/src/patchradar/api/main.py @@ -21,6 +21,7 @@ from patchradar.collectors.kev import fetch_cves as kev_fetch from patchradar.collectors.msrc import fetch_cves as msrc_fetch from patchradar.collectors.nvd import fetch_cves as nvd_fetch +from patchradar.collectors.nvd import with_records_for_the_rest from patchradar.collectors.vuldb import api_key as vuldb_api_key from patchradar.collectors.vuldb import fetch_cves as vuldb_fetch from patchradar.cpe import CpeError, parse_cpe @@ -346,6 +347,15 @@ async def _scan_one(sw: str, days: int, errors: list[dict], entry: WatchEntry | # and was counted twice — so the two scans gave different totals for the # same machine, which is what keeping them in step was meant to prevent. collected = merge_by_cve(collected) + # As in the CLI: what KEV or MSRC alone found is looked up in NVD by id, + # for an entry that can be judged. + installed = entry.installed() if entry is not None else None + if installed is not None: + collected, _, failure = await with_records_for_the_rest(collected, sw) + if failure is not None: + logger.warning("scan of %r: NVD could not be asked by id: %s", sw, failure) + errors.append({"software": sw, "source": failure.source, + "reason": failure.reason, "status": failure.status}) # FIRST is asked once per package about everything the sources returned, # before any of it is stored, so the EPSS fields land in the same row as the # CVE rather than in a second pass over the database. It cannot change the @@ -358,7 +368,6 @@ async def _scan_one(sw: str, days: int, errors: list[dict], entry: WatchEntry | "reason": exc.reason, "status": exc.status}) # Judged by the function the CLI judges with, before anything is stored, so # the verdict lands with the link and the two scans give one answer. - installed = entry.installed() if entry is not None else None mark(collected, installed) if installed is not None and verdicts is not None: found = [cve["verdict"] for cve in collected] diff --git a/src/patchradar/cli.py b/src/patchradar/cli.py index af0bc89..ba43835 100644 --- a/src/patchradar/cli.py +++ b/src/patchradar/cli.py @@ -23,7 +23,7 @@ from patchradar.collectors.msrc import MAX_DAYS_BACK as msrc_max_days from patchradar.collectors.msrc import fetch_cves as msrc_fetch from patchradar.collectors.nvd import api_key as nvd_api_key -from patchradar.collectors.nvd import fetch_cves +from patchradar.collectors.nvd import fetch_cves, with_records_for_the_rest from patchradar.collectors.vuldb import SOURCE as vuldb_source from patchradar.collectors.vuldb import api_key as vuldb_api_key from patchradar.collectors.vuldb import fetch_cves as vuldb_fetch @@ -399,6 +399,14 @@ async def _scan_target( # `severity: "UNKNOWN"`. The count therefore also changes meaning, from # records returned to CVEs found — which is what the line claims to say. all_cves = merge_by_cve(all_cves) + # For an entry the scan can judge, the CVEs that came from KEV or MSRC alone + # are looked up in NVD by id: without NVD's configurations they could only + # be "not evaluable", and KEV's are the ones being exploited. + installed = entry.installed() if entry is not None else None + unasked, lookup_failure = 0, None + if installed is not None: + with _status(f"[cyan]Asking NVD about what other sources found for {safe_target}...[/cyan]"): + all_cves, unasked, lookup_failure = await with_records_for_the_rest(all_cves, target) # FIRST is asked about the CVEs the scan found, in one request, after the # merge and before the sort: `priority` ranks on the EPSS fields, so # enriching after the sort would order the report on the facts it had a @@ -417,7 +425,6 @@ async def _scan_target( # each CVE found by name is judged against the version with NVD's own # configurations, and the ones that do not affect it are counted, not # listed, unless --all asks for them. - installed = entry.installed() if entry is not None else None judged = installed is not None mark(all_cves, installed) for cve in all_cves: @@ -436,6 +443,11 @@ async def _scan_target( f"⚠️ [yellow]{escape(str(failure))}[/yellow] — " f"results for [bold]{safe_target}[/bold] are incomplete" ) + if lookup_failure is not None: + console.print(Text( + f"⚠️ NVD could not be asked about {unasked} CVE{'s' if unasked != 1 else ''} " + f"from other sources ({lookup_failure}) — not evaluable for {target}", style="yellow" + )) if epss_failure is not None: console.print( f"⚠️ [yellow]{escape(str(epss_failure))}[/yellow] — " diff --git a/src/patchradar/collectors/nvd.py b/src/patchradar/collectors/nvd.py index 7adc500..f2b95b1 100644 --- a/src/patchradar/collectors/nvd.py +++ b/src/patchradar/collectors/nvd.py @@ -9,6 +9,7 @@ from patchradar.collectors.errors import BAD_PAYLOAD, CollectorError, from_http_error from patchradar.cvss import exploit_maturity +from patchradar.priority import merge_by_cve logger = logging.getLogger(__name__) @@ -336,6 +337,31 @@ async def _fetch_window( detail=f"empty page at {start_index} of {total}") +def _pacer(sleep: Callable[[float], Awaitable[None]], delay: float) -> Callable[[], Awaitable[None]]: + """What one call awaits before each of its requests to NVD.""" + sent = 0 + + async def pace() -> None: + # Inside one call the previous request has only just been answered, so + # the whole interval is waited, as before. Before the first request the + # clock is read: a call that starts inside the interval left by the last + # request this process sent — whichever keyword sent it — waits for + # what is left of it, and one that starts later waits for nothing. A + # request that failed counts too: a 403 is the limit speaking. + global _last_request_at + nonlocal sent + if sent: + await sleep(delay) + elif _last_request_at is not None: + remaining = delay - (_monotonic() - _last_request_at) + if remaining > 0: + await sleep(remaining) + sent += 1 + _last_request_at = _monotonic() + + return pace + + async def fetch_cves( keyword: str, days_back: int = 7, *, wait: Callable[[float], Awaitable[None]] | None = None, @@ -359,30 +385,9 @@ async def fetch_cves( caller hands over its own: a test that is about the pacing passes one that writes down what it was asked to wait, rather than replacing `_pace`. """ - sleep = wait or _pace key = api_key() headers = {"apiKey": key} if key else {} - delay = KEYED_DELAY_SECONDS if key else KEYLESS_DELAY_SECONDS - - sent = 0 - - async def pace() -> None: - # Inside one call the previous request has only just been answered, so - # the whole interval is waited, as before. Before the first request the - # clock is read: a call that starts inside the interval left by the last - # request this process sent — whichever keyword sent it — waits for - # what is left of it, and one that starts later waits for nothing. A - # request that failed counts too: a 403 is the limit speaking. - global _last_request_at - nonlocal sent - if sent: - await sleep(delay) - elif _last_request_at is not None: - remaining = delay - (_monotonic() - _last_request_at) - if remaining > 0: - await sleep(remaining) - sent += 1 - _last_request_at = _monotonic() + pace = _pacer(wait or _pace, KEYED_DELAY_SECONDS if key else KEYLESS_DELAY_SECONDS) results: list[dict] = [] async with httpx.AsyncClient(timeout=30.0, headers=headers) as client: @@ -394,3 +399,56 @@ async def pace() -> None: start += timedelta(days=1) await _fetch_window(client, keyword, start, end, pace, results) return results + + + +async def records_for( + cve_ids: list[str], keyword: str, *, + wait: Callable[[float], Awaitable[None]] | None = None, +) -> list[dict]: + """NVD's records of these CVEs, asked for one id at a time (`cveId=`). + + For the CVEs a scan found through a source that carries no `cpeMatch` — + CISA KEV, MSRC — so that an installed version can be judged against them + (docs/DESIGN.md §5.3). Paced like every other request to NVD, and an id NVD + does not know simply gives nothing. A failure raises with the records + already fetched as `partial`. + """ + key = api_key() + headers = {"apiKey": key} if key else {} + pace = _pacer(wait or _pace, KEYED_DELAY_SECONDS if key else KEYLESS_DELAY_SECONDS) + results: list[dict] = [] + async with httpx.AsyncClient(timeout=30.0, headers=headers) as client: + for cve_id in cve_ids: + await pace() + try: + response = await client.get(NVD_API, params={"cveId": cve_id}) + response.raise_for_status() + except httpx.HTTPError as exc: + raise from_http_error(SOURCE, exc, partial=results) from exc + try: + data = response.json() + except ValueError as exc: + raise CollectorError(SOURCE, BAD_PAYLOAD, status=response.status_code, partial=results) from exc + vulnerabilities = data.get("vulnerabilities") if isinstance(data, dict) else None + if isinstance(vulnerabilities, list): + results.extend(_parse_page(vulnerabilities, keyword)) + return results + + +async def with_records_for_the_rest(records: list[dict], keyword: str) -> tuple[list[dict], int, CollectorError | None]: + """`records`, merged with NVD's record of each one no NVD record came with. + + Returns the merged records, how many CVEs NVD could not be asked about, and + the failure that stopped it, if any. The ones left without a record stay as + they were, and the verdict on them says so. + """ + missing = [str(r["id"]) for r in records if "configurations" not in r and r.get("id")] + if not missing: + return records, 0, None + try: + found = await records_for(missing, keyword) + failure = None + except CollectorError as exc: + found, failure = exc.partial, exc + return merge_by_cve(records + found), (len(missing) - len(found) if failure else 0), failure diff --git a/tests/test_judge_through_nvd.py b/tests/test_judge_through_nvd.py new file mode 100644 index 0000000..bf09d24 --- /dev/null +++ b/tests/test_judge_through_nvd.py @@ -0,0 +1,196 @@ +""" +PatchRadar — a CVE from KEV or MSRC is judged through NVD's record of it (DESIGN §5.3) + +CISA KEV and MSRC carry no `cpeMatch`, so for an entry with an installed version +a CVE that reached the scan only through them was "not evaluable": NVD's record +was not part of the scan. Those are the CVEs that matter most — KEV lists what +is being exploited — and the scan had the means to ask. For an entry it can +judge, it now asks NVD about each such CVE by id (`cveId=`) and merges the record +in, which brings NVD's configurations and its score with it. A CVE NVD cannot be +asked about stays not evaluable, and the scan says why. + +NVD answers from recordings: `nvd_cve_2026_48092.json` (cveId=CVE-2026-48092, +7-Zip, 2026-10-07; affected 9.34 <= v < 26.01) and `nvd_no_results.json` for the +keyword search. The KEV catalogue is **derived**: CVE-2026-48092 is not in KEV, +so one entry is written in KEV's shape (the field names of +`tests/fixtures/kev_sample.json`) to stand for an exploited CVE that NVD's keyword +search did not return. +""" +from __future__ import annotations + +import json +import pathlib +import time + +import httpx +import pytest +import respx +from typer.testing import CliRunner + +from patchradar import cli +from patchradar.api import main as api_main +from patchradar.collectors import kev, msrc, nvd +from patchradar.collectors.errors import CollectorError +from patchradar.db import database + +FIXTURES = pathlib.Path(__file__).parent / "fixtures" +CVE_48092 = json.loads((FIXTURES / "nvd_cve_2026_48092.json").read_text(encoding="utf-8")) +NVD_EMPTY = json.loads((FIXTURES / "nvd_no_results.json").read_text(encoding="utf-8")) +SEVEN_ZIP = "cpe:2.3:a:7-zip:7-zip" + +KEV_DERIVED = { + "catalogVersion": "derived", "count": 1, + "vulnerabilities": [{ + "cveID": "CVE-2026-48092", + "vendorProject": "7-Zip", + "product": "7-Zip", + "vulnerabilityName": "7-Zip Derived Entry For Tests", + "dateAdded": "2026-06-10", + "shortDescription": "Stands for an exploited 7-Zip CVE; not a real KEV entry.", + "requiredAction": "Apply updates per vendor instructions.", + "dueDate": "2026-07-01", + "knownRansomwareCampaignUse": "Unknown", + }], +} + + +@pytest.fixture +def kev_lists_it(): + kev._snapshot = KEV_DERIVED + kev._snapshot_at = time.monotonic() + + +@pytest.fixture +def nvd_router(): + with respx.mock(assert_all_called=False) as router: + router.get(url__startswith=nvd.NVD_API, params__contains={"keywordSearch": "7-zip"}).mock( + return_value=httpx.Response(200, json=NVD_EMPTY)) + by_id = router.get(url__startswith=nvd.NVD_API, params__contains={"cveId": "CVE-2026-48092"}).mock( + return_value=httpx.Response(200, json=CVE_48092)) + router.get(url__startswith=msrc.MSRC_API).mock(return_value=httpx.Response(404)) + router.route().mock(return_value=httpx.Response(503)) # Debian, for the API + yield by_id + + +# ─── nvd.records_for ───────────────────────────────────────────────────────── + +async def test_nvd_is_asked_by_id_and_answers_with_the_configurations(nvd_router): + [record] = await nvd.records_for(["CVE-2026-48092"], "7-zip") + + assert record["id"] == "CVE-2026-48092" + assert record["software"] == "7-zip" + assert record["configurations"] + assert nvd_router.call_count == 1 + + +async def test_an_id_nvd_does_not_know_gives_nothing(): + with respx.mock() as router: + router.get(url__startswith=nvd.NVD_API).mock(return_value=httpx.Response(200, json=NVD_EMPTY)) + assert await nvd.records_for(["CVE-2026-00001"], "7-zip") == [] + + +async def test_nvd_refusing_is_a_collector_error_with_what_it_had(): + with respx.mock() as router: + router.get(url__startswith=nvd.NVD_API).mock(return_value=httpx.Response(503)) + with pytest.raises(CollectorError) as raised: + await nvd.records_for(["CVE-2026-48092"], "7-zip") + + assert raised.value.partial == [] + + +# ─── the CLI scan ──────────────────────────────────────────────────────────── + +def _scan(version="24.09", cpe=SEVEN_ZIP): + runner = CliRunner() + added = ["add", "7-zip"] + (["--version", version] if version else []) + (["--cpe", cpe] if cpe else []) + assert runner.invoke(cli.app, added).exit_code == 0 + result = runner.invoke(cli.app, ["scan", "7-zip", "--days", "7", "--all"], env={"COLUMNS": "200"}) + assert result.exit_code == 0, result.output + return " ".join(result.output.split()) + + +def test_an_exploited_cve_from_kev_alone_is_judged_affected(kev_lists_it, nvd_router): + out = _scan() + + assert "CVE-2026-48092" in out + assert "affecting 24.09: 1; not evaluable: 0" in out + assert "NVD states the fix is in 26.01" in out + + +def test_nvd_s_record_brings_its_score_to_the_kev_entry(kev_lists_it, nvd_router): + """KEV states no severity; NVD's record, merged in, does.""" + out = _scan() + + assert "UNKNOWN" not in out + + +def test_the_fixed_version_is_not_affected(kev_lists_it, nvd_router): + assert "affecting 26.01: 0; not evaluable: 0; not affecting it: 1" in _scan(version="26.01") + + +def test_an_entry_matched_by_name_does_not_cost_a_request(kev_lists_it, nvd_router): + _scan(version=None, cpe=None) + + assert nvd_router.call_count == 0 + + +def test_when_nvd_cannot_be_asked_the_cve_stays_not_evaluable_and_the_scan_says_so(kev_lists_it): + with respx.mock(assert_all_called=False) as router: + router.get(url__startswith=nvd.NVD_API, params__contains={"keywordSearch": "7-zip"}).mock( + return_value=httpx.Response(200, json=NVD_EMPTY)) + router.get(url__startswith=nvd.NVD_API, params__contains={"cveId": "CVE-2026-48092"}).mock( + return_value=httpx.Response(503)) + router.get(url__startswith=msrc.MSRC_API).mock(return_value=httpx.Response(404)) + out = _scan() + + assert "affecting 24.09: 0; not evaluable: 1" in out + assert "NVD could not be asked about 1 CVE" in out + + +# ─── the API scan ──────────────────────────────────────────────────────────── + +async def test_the_api_judges_the_kev_entry_through_nvd_too(kev_lists_it, nvd_router): + from patchradar.cpe import parse_cpe + + await database.add_to_watchlist("7-zip") + await database.set_installed("7-zip", version="24.09", cpe=parse_cpe(SEVEN_ZIP)) + [entry] = await database.get_watch_entries() + verdicts: dict = {} + + await api_main._scan_one("7-zip", 7, [], entry, verdicts) + + assert verdicts == {"7-zip": {"affected": 1, "not evaluable": 0, "not affected": 0}} + + +def test_a_cve_nvd_already_returned_is_not_asked_for_again(kev_lists_it): + """NVD's keyword search found it too, configurations and all: the merged + row has them, and a second request would cost a rate-limit slot for nothing.""" + with respx.mock(assert_all_called=False) as router: + router.get(url__startswith=nvd.NVD_API, params__contains={"keywordSearch": "7-zip"}).mock( + return_value=httpx.Response(200, json=CVE_48092)) + by_id = router.get(url__startswith=nvd.NVD_API, params__contains={"cveId": "CVE-2026-48092"}).mock( + return_value=httpx.Response(200, json=CVE_48092)) + router.get(url__startswith=msrc.MSRC_API).mock(return_value=httpx.Response(404)) + out = _scan() + + assert by_id.call_count == 0 + assert "affecting 24.09: 1" in out + + +async def test_the_api_lists_nvd_refusing_the_lookup_among_its_errors(kev_lists_it): + from patchradar.cpe import parse_cpe + + await database.add_to_watchlist("7-zip") + await database.set_installed("7-zip", version="24.09", cpe=parse_cpe(SEVEN_ZIP)) + [entry] = await database.get_watch_entries() + errors: list[dict] = [] + with respx.mock(assert_all_called=False) as router: + router.get(url__startswith=nvd.NVD_API, params__contains={"keywordSearch": "7-zip"}).mock( + return_value=httpx.Response(200, json=NVD_EMPTY)) + router.get(url__startswith=nvd.NVD_API, params__contains={"cveId": "CVE-2026-48092"}).mock( + return_value=httpx.Response(503)) + router.get(url__startswith=msrc.MSRC_API).mock(return_value=httpx.Response(404)) + router.route().mock(return_value=httpx.Response(503)) + await api_main._scan_one("7-zip", 7, errors, entry, {}) + + assert {"software": "7-zip", "source": "NVD", "reason": "server_error", "status": 503} in errors From e57a387b0f705fcbf7ca6070c809777049ecd8af Mon Sep 17 00:00:00 2001 From: maksimtech <268127343+maksimtech@users.noreply.github.com> Date: Sat, 17 Oct 2026 17:28:02 +0200 Subject: [PATCH 4/8] feat(priority): store what decides a rank, and rank the stored listings CISA's exploitation facts and the CVSS v4.0 exploit maturity travelled in the scan's records and were not columns: the next `status` listed by publication date, and an exploited CVE from 2023 sat under every new 5.3. test_priority_wiring held that boundary to say "now rank the database paths too" the day the columns arrived. - cves gains known_exploited, kev_ransomware, kev_date_added, kev_due_date and cvss_exploit_maturity, migrated in place and refreshed by every scan that brings them: CISA lists a CVE long after NVD publishes it. CISA's "Unknown" is stored as no. - get_cves orders by priority.sql_order, built from the constants priority() decides with; an index on that order makes the full listing 1.0 ms on the benchmark's 500 CVEs, against 1.45 ms by date. - /api/cves rows carry priority, priority_reason and the KEV fields; status has a Priority column and lists most urgent first. - test_priority_wiring now holds the other side of the boundary. Tests first, seen failing; the SQL order is checked against sort_by_priority on rows covering every rank and edge. 15 of 16 mutants killed; the survivor drops the "scored" branch, which COALESCE on the score already orders the same way (equivalent). Co-Authored-By: Claude Opus 5.5 (1M context) --- CHANGELOG.md | 10 ++ README.md | 2 +- docs/DESIGN.md | 14 ++- src/patchradar/api/main.py | 12 +- src/patchradar/cli.py | 7 +- src/patchradar/db/database.py | 67 ++++++++++-- src/patchradar/priority.py | 28 +++++ tests/test_priority_stored.py | 201 ++++++++++++++++++++++++++++++++++ tests/test_priority_wiring.py | 40 +++---- 9 files changed, 342 insertions(+), 39 deletions(-) create mode 100644 tests/test_priority_stored.py diff --git a/CHANGELOG.md b/CHANGELOG.md index e9f543c..f9e35bd 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -38,6 +38,16 @@ what has already shipped, and for nothing else. See `RELEASING.md`. scan, CLI and API alike, now asks NVD about each by id (`nvd.records_for`, `cveId=`), paced like every request to NVD, and merges the record in, score included. When NVD cannot be asked, the CVE stays not evaluable and the scan says so (the API in `errors`). +- **`status` and `/api/cves` rank as a scan does.** CISA's exploitation facts and the CVSS v4.0 + exploit maturity travelled in the scan's records and were not columns, so the next `status` + listed by publication date and an exploited CVE from 2023 sat under every new 5.3. They are + stored now (`known_exploited`, `kev_ransomware`, `kev_date_added`, `kev_due_date`, + `cvss_exploit_maturity`), refreshed by each scan that brings them — CISA lists a CVE long after + NVD publishes it — and the listings are ordered by `priority.sql_order`, built from the rules a + scan ranks by; a test runs one set of rows through both. An index on that order makes the full + listing faster than the date order was (1.0 ms against 1.45 ms on the benchmark's 500 CVEs). + `/api/cves` rows carry `priority`, `priority_reason` and the KEV fields; `status` has a Priority + column. - **docs/DESIGN.md.** The code referred to "the versionradar design, §5" for these columns; the text was never committed and could not be found. It is reconstructed from what `affected.py`, `inventory.py` and `tests/test_priority_wiring.py` already committed to. diff --git a/README.md b/README.md index 7bb1797..b11afd5 100644 --- a/README.md +++ b/README.md @@ -65,7 +65,7 @@ patchradar add notepad --version 8.5.0 --cpe "cpe:2.3:a:notepad-plus-plus:notepa patchradar scan notepad --days 365 # what affects 8.5.0, and what is unknown patchradar scan notepad --days 365 --all # also the CVEs that do not affect it -# Show latest CVEs in terminal +# Show the most urgent stored CVEs, ranked as a scan ranks them patchradar status # Remove software diff --git a/docs/DESIGN.md b/docs/DESIGN.md index 794d694..a8d17a5 100644 --- a/docs/DESIGN.md +++ b/docs/DESIGN.md @@ -118,16 +118,20 @@ is not shown until a scan judges the new one. `status` and `/api/cves` show it; the API records version and CPE (`POST /api/watchlist/{name}?version=&cpe=`) and judges with the same function as the CLI (`affected.mark`). -Still in memory only: the KEV fields that decide a rank — the migration -`tests/test_priority_wiring.py` waits for. +What decides a rank is stored too: `known_exploited`, `kev_ransomware`, +`kev_date_added`, `kev_due_date` and the provider's `cvss_exploit_maturity`, in +`cves`, refreshed by every scan that brings them (CISA lists a CVE long after NVD +publishes it). `status` and `/api/cves` list most urgent first, with the order +`priority.sql_order` builds from the rules a scan ranks by, and an index on that +expression keeps a listing of 50 a read of 50 rows. ### 5.6 Order of work 1. **PR-01a** — the columns, `add --version --cpe`, the verdict for NVD records in the command-line scan, `--all`, the README. -2. **PR-01b** — in two parts: the verdict stored and the API (01b-1); the KEV - fields stored, with the database-backed listings ranked by priority, and KEV - and MSRC entries judged through NVD's record of the same CVE (01b-2). +2. **PR-01b** — in three parts: the verdict stored and the API (01b-1); KEV + and MSRC entries judged through NVD's record of the same CVE (01b-2); the KEV + fields stored, with the database-backed listings ranked by priority (01b-3). 3. **PR-01c** — the web UI; `patchradar import` writing the mappings a person confirmed, `channel` included. diff --git a/src/patchradar/api/main.py b/src/patchradar/api/main.py index b2beabd..2c2bfd4 100644 --- a/src/patchradar/api/main.py +++ b/src/patchradar/api/main.py @@ -37,7 +37,7 @@ ) from patchradar.epss import enrich as epss_enrich from patchradar.names import SOFTWARE_NAME_MAX_LENGTH, SOFTWARE_NAME_PATTERN, normalise_software_name -from patchradar.priority import merge_by_cve +from patchradar.priority import merge_by_cve, priority logger = logging.getLogger("patchradar") @@ -206,6 +206,16 @@ def sanitize_cve(cve: dict) -> dict: # Numeric fields. The EPSS pair is passed through as stored, including a # missing one: `.get` yields None, and None here means FIRST does not score # the CVE — which a 0.0 would misreport as "almost certainly not exploited". + # Why the row is where it is, as the CLI prints it: a rank without its + # reason is a number taken on trust. + rank = priority(cve) + safe["priority"] = rank.label + safe["priority_reason"] = _strip_control_chars(rank.reason)[:300] + for flag in ("known_exploited", "kev_ransomware"): + safe[flag] = cve.get(flag) if isinstance(cve.get(flag), bool) else None + for field in ("kev_date_added", "kev_due_date"): + value = cve.get(field) + safe[field] = _strip_control_chars(value).strip()[:20] if isinstance(value, str) else None for field in ("verdict", "verdict_reason", "verdict_fixed_version", "judged_version"): value = cve.get(field) safe[field] = _strip_control_chars(value).strip()[:300] if isinstance(value, str) else None diff --git a/src/patchradar/cli.py b/src/patchradar/cli.py index ba43835..e9010af 100644 --- a/src/patchradar/cli.py +++ b/src/patchradar/cli.py @@ -635,11 +635,14 @@ def _print_cves(software: str, cves: list, judged: bool = False): console.print(table) def _print_cves_table(cves: list): - table = Table(title="🛡️ PatchRadar — Latest CVEs", box=box.ROUNDED, show_lines=True) + # Most urgent first, in the order the database now returns: a scan's report + # and `status` rank the same rows the same way. + table = Table(title="🛡️ PatchRadar — Most urgent CVEs", box=box.ROUNDED, show_lines=True) table.add_column("CVE ID", style="bold cyan", no_wrap=True) table.add_column("Software", style="blue") table.add_column("Score", justify="center", width=6) table.add_column("Severity", justify="center", width=10) + table.add_column("Priority", width=10) table.add_column("Source", width=6) # Only when a stored verdict is current for some row: a watchlist matched by # name alone gets the table it always had. @@ -653,12 +656,14 @@ def _print_cves_table(cves: list): severity_color = SEVERITY_STYLES.get(severity, "white") verdict = cve.get("verdict") extra = [Text(verdict or "—", style=VERDICT_STYLES.get(verdict, "dim"))] if judged else [] + rank = priority(cve) table.add_row( Text(str(cve.get("id", ""))), Text(str(cve.get("software") or "")), score_str, Text(severity, style=severity_color), + Text(rank.label, style=PRIORITY_STYLES[rank.rank]), Text(str(cve.get("source") or "")), *extra, ) diff --git a/src/patchradar/db/database.py b/src/patchradar/db/database.py index 04f56f2..5928757 100644 --- a/src/patchradar/db/database.py +++ b/src/patchradar/db/database.py @@ -6,6 +6,7 @@ import aiosqlite from patchradar.cpe import Cpe +from patchradar.priority import as_flag, sql_order logger = logging.getLogger(__name__) @@ -50,8 +51,19 @@ def normalize_timestamp(value) -> str | None: _ADDED_COLUMNS = { "epss_score": "REAL", "epss_percentile": "REAL", + # What decides a rank besides the score (priority.py), stored so that the + # listings read from the database can be ranked as a scan is + # (docs/DESIGN.md §5.5). The flags are 1/0, NULL for a CVE KEV never listed. + "known_exploited": "INTEGER", + "kev_ransomware": "INTEGER", + "kev_date_added": "TEXT", + "kev_due_date": "TEXT", + "cvss_exploit_maturity": "TEXT", } +# Stored as 1/0 and read back as booleans; NULL stays None. +_FLAGS = ("known_exploited", "kev_ransomware") + # What is installed, for the scan to judge each CVE against (docs/DESIGN.md # §5.1): the names `inventory.py` was written for. Nullable without a default, # for the same reason as above — a version nobody entered is an absence, and a @@ -153,6 +165,11 @@ async def init_db(): url TEXT, epss_score REAL, epss_percentile REAL, + known_exploited INTEGER, + kev_ransomware INTEGER, + kev_date_added TEXT, + kev_due_date TEXT, + cvss_exploit_maturity TEXT, seen_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP ) """) @@ -170,6 +187,11 @@ async def init_db(): await _add_missing_columns(db) await _add_missing_columns(db, "watchlist", _ADDED_WATCHLIST_COLUMNS) await _add_missing_columns(db, "cve_software", _ADDED_LINK_COLUMNS) + # The listings' order as an index, so a listing of 50 reads 50 rows in + # order instead of ranking every stored CVE first: measured on the + # benchmark's 500, get_cves went from 1.9 ms to 1.0 ms with it, below + # the 1.45 ms it took when it still listed by date. + await db.execute(f"CREATE INDEX IF NOT EXISTS cves_by_priority ON cves ({_PRIORITY_INDEX})") await _normalize_stored_timestamps(db) await _link_stored_cves(db) await db.commit() @@ -303,20 +325,34 @@ async def save_cve(cve: dict) -> bool: judged_version = excluded.judged_version """, (cve["id"], cve["software"], cve.get("verdict"), cve.get("verdict_reason"), cve.get("verdict_fixed_version"), cve.get("judged_version"))) + # The exploitation facts are refreshed like the forecast, and for a + # sharper reason: CISA adds a CVE to KEV long after NVD published + # it, and a row stored on Monday must learn on Wednesday that it is + # being exploited. COALESCE keeps a scan without the KEV record — + # a keyword KEV does not match — from erasing what it said. + flags = [int(as_flag(cve[f])) if cve.get(f) is not None else None for f in _FLAGS] cursor = await db.execute(""" INSERT INTO cves (id, software, description, cvss_score, cvss_version, - severity, published_at, source, url, epss_score, epss_percentile) - VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) + severity, published_at, source, url, epss_score, epss_percentile, + known_exploited, kev_ransomware, kev_date_added, kev_due_date, cvss_exploit_maturity) + VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) ON CONFLICT(id) DO UPDATE SET epss_score = COALESCE(excluded.epss_score, cves.epss_score), - epss_percentile = COALESCE(excluded.epss_percentile, cves.epss_percentile) + epss_percentile = COALESCE(excluded.epss_percentile, cves.epss_percentile), + known_exploited = COALESCE(excluded.known_exploited, cves.known_exploited), + kev_ransomware = COALESCE(excluded.kev_ransomware, cves.kev_ransomware), + kev_date_added = COALESCE(excluded.kev_date_added, cves.kev_date_added), + kev_due_date = COALESCE(excluded.kev_due_date, cves.kev_due_date), + cvss_exploit_maturity = COALESCE(excluded.cvss_exploit_maturity, cves.cvss_exploit_maturity) """, ( cve["id"], cve["software"], cve["description"], cve.get("cvss_score"), cve.get("cvss_version"), cve.get("severity"), normalize_timestamp(cve.get("published_at")), cve.get("source"), cve.get("url"), - cve.get("epss_score"), cve.get("epss_percentile") + cve.get("epss_score"), cve.get("epss_percentile"), + *flags, cve.get("kev_date_added"), cve.get("kev_due_date"), + cve.get("cvss_exploit_maturity"), )) await db.commit() return cursor.rowcount > 0 @@ -335,6 +371,21 @@ async def save_cve(cve: dict) -> bool: _CURRENT_VERDICT = ", ".join(f"link.{column}" for column in _ADDED_LINK_COLUMNS) +# Most urgent first, as a scan orders its report (priority.py); among equals the +# newest first, then by id, which is also the order a scan's ties come in. The +# index in init_db holds the same expression, unqualified, which SQLite matches. +_PRIORITY_ORDER = f"{sql_order('cves')}, published_at DESC, cves.id DESC" +_PRIORITY_INDEX = f"{sql_order()}, published_at DESC, id DESC" + + +def _read(row) -> dict: + record = dict(row) + for flag in _FLAGS: + if record.get(flag) is not None: + record[flag] = bool(record[flag]) + return record + + async def get_cves(software: str | None = None, limit: int = 50) -> list[dict]: async with aiosqlite.connect(DB_PATH) as db: db.row_factory = aiosqlite.Row @@ -347,18 +398,18 @@ async def get_cves(software: str | None = None, limit: int = 50) -> list[dict]: LEFT JOIN cve_software AS link ON link.cve_id = cves.id AND link.software = ? AND link.judged_version = watched.installed_version WHERE cves.software = ? OR cves.id IN (SELECT cve_id FROM cve_software WHERE software = ?) - ORDER BY published_at DESC LIMIT ? + ORDER BY {_PRIORITY_ORDER} LIMIT ? """, (software.lower(),) * 4 + (limit,)) as cursor: rows = await cursor.fetchall() # Named after the product asked about, not the first one to find it. - return [{**dict(row), "software": software.lower()} for row in rows] + return [{**_read(row), "software": software.lower()} for row in rows] else: async with db.execute(f""" SELECT cves.*, {_CURRENT_VERDICT} FROM cves LEFT JOIN watchlist AS watched ON watched.name = cves.software LEFT JOIN cve_software AS link ON link.cve_id = cves.id AND link.software = cves.software AND link.judged_version = watched.installed_version - ORDER BY published_at DESC LIMIT ? + ORDER BY {_PRIORITY_ORDER} LIMIT ? """, (limit,)) as cursor: rows = await cursor.fetchall() - return [dict(row) for row in rows] + return [_read(row) for row in rows] diff --git a/src/patchradar/priority.py b/src/patchradar/priority.py index de0112b..c799f19 100644 --- a/src/patchradar/priority.py +++ b/src/patchradar/priority.py @@ -317,6 +317,34 @@ def merge_by_cve(records: list[dict]) -> list[dict]: return [merged[key] for key in order] + loose +def as_flag(value: object) -> bool: + """A yes or no from a field that may hold CISA's "Known"/"Unknown" (see `_is_true`).""" + return _is_true(value) + + +def sql_order(table: str = "") -> str: + """ORDER BY terms that put stored rows in `sort_by_priority`'s order. + + Built from the constants `priority()` decides with, so the two cannot drift + (tests/test_priority_stored.py runs one set of rows through both), and run + by SQLite so a listing of 50 stays a query for 50. Rank first, then the + score, an absent one below every real one; the caller adds the tie-breaks. + The stored flags are 1/0/NULL, so `= 1` is the "Known" test here. Columns + are qualified with `table` when one is given, bare otherwise (an index). + """ + t = f"{table}." if table else "" + promotes = ", ".join(f"'{maturity}'" for maturity in sorted(_PROMOTES)) + return ( + f"CASE WHEN {t}known_exploited = 1 AND {t}kev_ransomware = 1 THEN {RANK_KEV_RANSOMWARE} " + f"WHEN {t}known_exploited = 1 THEN {RANK_KEV} " + f"WHEN {t}epss_score >= {EPSS_THRESHOLD} AND {t}epss_score <= 1.0 THEN {RANK_EPSS} " + f"WHEN {t}cvss_exploit_maturity IN ({promotes}) THEN {RANK_REPORTED} " + f"WHEN {t}cvss_score IS NOT NULL THEN {RANK_SCORED} " + f"ELSE {RANK_UNSCORED} END DESC, " + f"COALESCE({t}cvss_score, {_NO_SCORE_IN_SORT}) DESC" + ) + + def sort_by_priority(rows: list[dict]) -> list[dict]: """Already-merged rows, most urgent first. diff --git a/tests/test_priority_stored.py b/tests/test_priority_stored.py new file mode 100644 index 0000000..7d8d817 --- /dev/null +++ b/tests/test_priority_stored.py @@ -0,0 +1,201 @@ +""" +PatchRadar — what decides a rank is stored, and the stored listings are ranked (DESIGN §5.5) + +`known_exploited`, `kev_ransomware`, the KEV dates and the provider's CVSS v4.0 +exploit maturity travelled in the scan's records and were not columns, so the +fact that CISA has seen a CVE exploited existed for the length of one scan. +`patchradar status` and `/api/cves` listed by publication date, and an exploited +CVE from 2023 sat under every new 5.3. `tests/test_priority_wiring.py` held that +boundary on purpose, to say "now rank the database paths too" the day the +columns arrived; this is that day. + +The order is computed by SQLite, so a listing of 50 is still a query for 50 — +and it is built from the constants `priority()` uses, so the two cannot +drift: the test below runs one set of rows through both and asks for the same +order. +""" +from __future__ import annotations + +import asyncio + +import aiosqlite +import pytest +from httpx import ASGITransport, AsyncClient +from typer.testing import CliRunner + +import patchradar.api.main as api +from patchradar import cli +from patchradar.db import database +from patchradar.priority import sort_by_priority + +PRIORITY_COLUMNS = {"known_exploited", "kev_ransomware", "kev_date_added", "kev_due_date", + "cvss_exploit_maturity"} + + +def _row(cve_id, published, score=None, **facts): + return {"id": cve_id, "software": "nginx", "description": cve_id, "cvss_score": score, + "cvss_version": "3.1" if score is not None else None, + "severity": "HIGH" if score is not None else "UNKNOWN", + "published_at": published, "source": "NVD", + "url": f"https://nvd.nist.gov/vuln/detail/{cve_id}", **facts} + + +# One row per rank, and the edges between them: ties on score, no score, +# EPSS exactly at the threshold, an EPSS outside [0, 1] that must not promote, +# ransomware "Unknown" (CISA's word for no), and v4.0 maturity both ways. +ROWS = [ + _row("CVE-2026-0001", "2026-10-01T00:00:00Z", 5.3), + _row("CVE-2026-0002", "2026-09-30T00:00:00Z", 9.8), + _row("CVE-2026-0003", "2026-09-29T00:00:00Z", None), + _row("CVE-2026-0004", "2026-09-28T00:00:00Z", 6.1, epss_score=0.1, epss_percentile=0.95), + _row("CVE-2026-0005", "2026-09-27T00:00:00Z", 9.1, epss_score=0.0999, epss_percentile=0.9), + _row("CVE-2026-0006", "2026-09-26T00:00:00Z", 7.5, epss_score=1.5), + _row("CVE-2026-0007", "2026-09-25T00:00:00Z", 4.3, cvss_exploit_maturity="PROOF_OF_CONCEPT"), + _row("CVE-2026-0008", "2026-09-24T00:00:00Z", 8.8, cvss_exploit_maturity="UNREPORTED"), + _row("CVE-2023-0009", "2023-03-01T00:00:00Z", 7.8, known_exploited=True, + kev_date_added="2024-01-10", kev_due_date="2024-01-31", kev_ransomware=False), + _row("CVE-2022-0010", "2022-05-01T00:00:00Z", 6.5, known_exploited=True, + kev_date_added="2023-02-01", kev_ransomware=True), + _row("CVE-2021-0011", "2021-06-01T00:00:00Z", None, known_exploited=True, + kev_date_added="2022-03-03", kev_ransomware="Unknown"), + _row("CVE-2026-0012", "2026-09-23T00:00:00Z", 9.8), + # CVSS 0.0 is a measurement — no impact — and sorts above no score at all, + # in the same rank: the sentinel for "no score" must sit below every real one. + _row("CVE-2020-0014", "2020-01-01T00:00:00Z", 0.0, known_exploited=True, + kev_date_added="2021-11-03"), +] + + +async def _store(rows): + await database.add_to_watchlist("nginx") + for row in rows: + assert await database.save_cve(dict(row)), row["id"] + + +def _expected(rows): + """`priority()`'s order, from rows in the order the database breaks ties.""" + by_date = sorted(rows, key=lambda r: (r["published_at"], r["id"]), reverse=True) + return [r["id"] for r in sort_by_priority(by_date)] + + +# ─── the columns ───────────────────────────────────────────────────────────── + +@pytest.fixture +def old_cves(tmp_path, monkeypatch): + path = tmp_path / "old.db" + + async def _build(): + async with aiosqlite.connect(path) as db: + await db.execute(""" + CREATE TABLE cves ( + id TEXT PRIMARY KEY, software TEXT NOT NULL, description TEXT, + cvss_score REAL, cvss_version TEXT, severity TEXT, published_at TIMESTAMP, + source TEXT, url TEXT, epss_score REAL, epss_percentile REAL, + seen_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP + ) + """) + await db.commit() + + asyncio.run(_build()) + monkeypatch.setattr(database, "DB_PATH", path) + return path + + +@pytest.mark.asyncio +async def test_an_existing_cves_table_gains_the_columns_a_rank_needs(old_cves): + await database.init_db() + + async with aiosqlite.connect(old_cves) as db, db.execute("PRAGMA table_info(cves)") as cursor: + assert {row[1] for row in await cursor.fetchall()} >= PRIORITY_COLUMNS + + +@pytest.mark.asyncio +async def test_the_exploitation_fact_is_read_back(): + await _store([ROWS[8]]) + + [row] = await database.get_cves() + + assert row["known_exploited"] is True + assert row["kev_ransomware"] is False + assert (row["kev_date_added"], row["kev_due_date"]) == ("2024-01-10", "2024-01-31") + + +@pytest.mark.asyncio +async def test_cisas_unknown_is_stored_as_no_ransomware(): + await _store([ROWS[10]]) + + [row] = await database.get_cves() + + assert row["kev_ransomware"] is False + + +@pytest.mark.asyncio +async def test_a_cve_stored_before_kev_listed_it_learns_it_on_the_next_scan(): + """First writer wins for the description; it cannot for this. NVD stored the + row on Monday, CISA added it on Tuesday, Wednesday's scan must say so.""" + await _store([_row("CVE-2026-0013", "2026-09-01T00:00:00Z", 7.2)]) + await database.save_cve(_row("CVE-2026-0013", "2026-09-01T00:00:00Z", None, known_exploited=True, + kev_date_added="2026-10-15", kev_ransomware=True)) + + [row] = await database.get_cves() + + assert row["known_exploited"] is True + assert row["kev_ransomware"] is True + assert row["cvss_score"] == 7.2 + + +@pytest.mark.asyncio +async def test_a_scan_without_the_kev_record_does_not_forget_it(): + await _store([ROWS[8]]) + await database.save_cve(_row("CVE-2023-0009", "2023-03-01T00:00:00Z", 7.8)) + + [row] = await database.get_cves() + + assert row["known_exploited"] is True + + +# ─── the order ─────────────────────────────────────────────────────────────── + +@pytest.mark.asyncio +async def test_the_stored_listing_is_in_the_order_priority_gives(): + await _store(ROWS) + + assert [r["id"] for r in await database.get_cves(limit=100)] == _expected(ROWS) + + +@pytest.mark.asyncio +async def test_the_listing_for_one_product_is_in_the_same_order(): + await _store(ROWS) + + assert [r["id"] for r in await database.get_cves(software="nginx", limit=100)] == _expected(ROWS) + + +@pytest.mark.asyncio +async def test_a_short_listing_is_the_most_urgent_not_the_newest(): + await _store(ROWS) + + assert [r["id"] for r in await database.get_cves(limit=3)] == _expected(ROWS)[:3] + assert (await database.get_cves(limit=1))[0]["id"] == "CVE-2022-0010" + + +# ─── what reads it ─────────────────────────────────────────────────────────── + +async def test_the_api_listing_says_why_each_row_ranks_where_it_does(): + await _store(ROWS) + async with AsyncClient(transport=ASGITransport(app=api.app), base_url="http://test") as client: + rows = (await client.get("/api/cves", params={"limit": 100})).json()["cves"] + + assert [r["id"] for r in rows] == _expected(ROWS) + assert rows[0]["priority"] == "ransomware" + assert "in CISA KEV since 2023-02-01" in rows[0]["priority_reason"] + assert rows[0]["known_exploited"] is True + + +def test_status_lists_the_exploited_cve_above_the_newest(): + asyncio.run(_store([ROWS[0], ROWS[8]])) + + result = CliRunner().invoke(cli.app, ["status"], env={"COLUMNS": "200"}) + + out = result.output + assert out.index("CVE-2023-0009") < out.index("CVE-2026-0001") + assert "exploited" in out diff --git a/tests/test_priority_wiring.py b/tests/test_priority_wiring.py index 1c7c46a..819de0c 100644 --- a/tests/test_priority_wiring.py +++ b/tests/test_priority_wiring.py @@ -6,12 +6,13 @@ dead weight until something calls it, and a report that ranks while the API does not is a drift nobody notices until two people compare their screens. -The second half pins the boundary deliberately. `known_exploited`, -`kev_ransomware` and `kev_due_date` travel in the records the collectors return -and are **not** columns in `cves`, so the exploitation fact exists in memory -during a scan and is gone by the next `patchradar status`. Until the migration of -docs/DESIGN.md §5.5 lands, only the in-memory path can rank — and the day someone adds the -column, this test is what says "now wire the API too". +The second half pinned a boundary deliberately: `known_exploited`, +`kev_ransomware` and the KEV dates travelled in the scan's records and were not +columns in `cves`, so the exploitation fact was gone by the next `patchradar +status`, and the day someone added the columns this file was to say "now wire +the API too". They arrived with docs/DESIGN.md §5.5, and the tests below now +hold the other side: the facts are stored, and the stored listings are ranked +by the same rules (`priority.sql_order`), not by date. The EPSS pair *is* stored, which narrows that gap without closing it: a ranking computed from the database would know the forecast and still not know what CISA @@ -70,24 +71,17 @@ def test_the_report_shows_why_a_row_ranks_where_it_does(): assert "reason" in body, "a rank without its reason is a number taken on trust" -def test_the_database_does_not_store_the_exploitation_fact_yet(): - """The boundary, asserted so it cannot move silently. - - If this fails because a column was added, the ranking is no longer confined - to a single scan: `/cves` and `patchradar status` read from the database and - must order by priority too, or the same machine gets two different answers. - """ +def test_the_database_stores_every_fact_a_rank_is_decided_by(): schema = _function_source(DB_SOURCE, "init_db") - stored = [fact for fact in KEV_FACTS if fact in schema] - assert not stored, ( - "the schema now holds " + ", ".join(stored) - + " — rank the database-backed paths as well (API /cves, status)" - ) + missing = [fact for fact in KEV_FACTS if fact not in schema] + assert not missing, "a rank decided by an unstored fact cannot be repeated by status or /cves" -def test_the_api_listing_is_still_ordered_by_date(): +def test_the_stored_listings_are_ordered_by_priority_not_by_date(): """Named here so the two paths are compared in one place rather than - discovered to differ. `get_cves` orders by publication date, which is the - only order the stored columns support.""" - assert "ORDER BY published_at DESC" in DB_SOURCE - assert "order_by_priority" not in API_SOURCE + discovered to differ: `get_cves`, which `status` and `/api/cves` read, orders + with the expression `priority.sql_order` builds from the scan's own rules.""" + body = _function_source(DB_SOURCE, "get_cves") + assert body.count("ORDER BY {_PRIORITY_ORDER}") == 2 + assert "sql_order(" in DB_SOURCE + assert "ORDER BY published_at DESC" not in body From 3ab79acacbe969a5ef68231fd8cf0b7f3fd77845 Mon Sep 17 00:00:00 2001 From: maksimtech <268127343+maksimtech@users.noreply.github.com> Date: Sat, 17 Oct 2026 18:40:16 +0200 Subject: [PATCH 5/8] fix(watchlist): vendor and product_id are the vendor source's, as inventory.py meant inventory.py named the watchlist columns before they existed: vendor, product_id, installed_version, channel, cpe. There a vendor source is a feed that answers by version (Chrome's, Edge's, MSRC, Mozilla's) and product_id is the product it needs; NVD by CPE is the other route. #58 filled vendor and product_id with the CPE's fields instead - a misreading of the design it rebuilt - which would have had the inventory import write one over the other. No release carried it. - The CPE's vendor and product are read from `cpe` (WatchEntry.cpe_vendor, cpe_product, installed()), not stored. - set_source records the vendor source, product and channel a person confirmed, for the import to call; it never adds a name. - DESIGN.md section 5.1 says which column is which, and why. Tests first, seen failing; the mutants that swap source and product or drop the version from installed() are killed. Co-Authored-By: Claude Opus 5.5 (1M context) --- docs/DESIGN.md | 14 ++++-- src/patchradar/db/database.py | 44 +++++++++++++---- tests/test_api_versioned.py | 3 +- tests/test_vendor_source_columns.py | 76 +++++++++++++++++++++++++++++ tests/test_watchlist_versions.py | 6 +-- 5 files changed, 126 insertions(+), 17 deletions(-) create mode 100644 tests/test_vendor_source_columns.py diff --git a/docs/DESIGN.md b/docs/DESIGN.md index a8d17a5..ec2e100 100644 --- a/docs/DESIGN.md +++ b/docs/DESIGN.md @@ -47,13 +47,21 @@ missing version is an absence, and a default would be a claim: | column | holds | set by | |---|---|---| | `cpe` | the CPE 2.3 name of the product, as NVD writes it: `cpe:2.3:a:notepad-plus-plus:notepad\+\+` | `add --cpe` | -| `vendor` | the CPE vendor field, `notepad-plus-plus` | derived from `cpe` | -| `product_id` | the CPE product field, escaped as NVD escapes it, `notepad\+\+` | derived from `cpe` | +| `vendor` | the vendor source that answers by version — `inventory.py`'s *source*: `chrome`, `edge`, `msrc`, `mozilla` | the inventory import, once confirmed (§5.6) | +| `product_id` | the product that source needs: `chrome`, `Edge` | the inventory import, once confirmed | | `installed_version` | the version installed, `8.5.0` | `add --version` | -| `channel` | the release channel a vendor source answers for (stable, ESR, LTS) | reserved for the inventory import (§5.6) | +| `channel` | the release channel the source answers for (stable, ESR, LTS) | the inventory import, once confirmed | An existing database gets them on the next start, as `cves` got its EPSS columns. +Two routes to an answer by version, kept apart: the CPE, through NVD's +configurations, which the scan judges with today; and a vendor source, which +`inventory.py` proposes and a person confirms. The CPE's own vendor and product +(`notepad-plus-plus`, `notepad\+\+`) are read from `cpe` when the scan needs +them, not stored. Until 2026-10-17 they were stored in `vendor` and `product_id`, +by a misreading of this table that `inventory.py` contradicted; no release +carried that. + ### 5.2 Nothing is guessed The CPE is typed by a person, never derived from the name. Measured on a real diff --git a/src/patchradar/db/database.py b/src/patchradar/db/database.py index 5928757..b727544 100644 --- a/src/patchradar/db/database.py +++ b/src/patchradar/db/database.py @@ -5,7 +5,7 @@ import aiosqlite -from patchradar.cpe import Cpe +from patchradar.cpe import Cpe, parse_cpe from patchradar.priority import as_flag, sql_order logger = logging.getLogger(__name__) @@ -209,7 +209,13 @@ async def add_to_watchlist(name: str) -> bool: @dataclass(frozen=True) class WatchEntry: - """A watchlist row with what is known of the installed product.""" + """A watchlist row with what is known of the installed product. + + `vendor`, `product_id` and `channel` are the vendor source's — the feed + that answers by version and the product it needs, as `inventory.py` names + them (docs/DESIGN.md §5.1). The CPE's vendor and product are another thing, + read from `cpe`: NVD by CPE is the other route. + """ name: str installed_version: str | None = None @@ -218,12 +224,21 @@ class WatchEntry: product_id: str | None = None channel: str | None = None + @property + def cpe_vendor(self) -> str | None: + return parse_cpe(self.cpe).vendor if self.cpe else None + + @property + def cpe_product(self) -> str | None: + return parse_cpe(self.cpe).product if self.cpe else None + def installed(self) -> tuple[str, str, str] | None: """(version, vendor, product) when a scan can judge CVEs against what is - installed — the version to compare, and the vendor and product NVD's - `cpeMatch` is read by — and None when any of the three is missing.""" - if self.installed_version and self.vendor and self.product_id: - return self.installed_version, self.vendor, self.product_id + installed — the version to compare, and the CPE's vendor and product + NVD's `cpeMatch` is read by — and None when any of them is missing.""" + if self.installed_version and self.cpe: + cpe = parse_cpe(self.cpe) + return self.installed_version, cpe.vendor, cpe.product return None @property @@ -244,17 +259,28 @@ async def set_installed(name: str, *, version: str | None, cpe: Cpe | None) -> b Only what is given changes: a new version leaves the CPE as it was. False when the name is not on the watchlist — this never adds one. """ - fields: dict[str, str] = {} + fields: dict[str, str | None] = {} if version is not None: fields["installed_version"] = version if cpe is not None: - fields.update(cpe=cpe.text, vendor=cpe.vendor, product_id=cpe.product) + fields["cpe"] = cpe.text + return await _update_entry(name, fields) + + +async def set_source(name: str, *, vendor: str, product_id: str, channel: str | None) -> bool: + """Record the vendor source that answers by version for an entry already on + the watchlist — `inventory.py`'s source, product and channel, once a person + has confirmed them. False when the name is not on the watchlist.""" + return await _update_entry(name, {"vendor": vendor, "product_id": product_id, "channel": channel}) + + +async def _update_entry(name: str, fields: dict[str, str | None]) -> bool: async with aiosqlite.connect(DB_PATH) as db: async with db.execute("SELECT 1 FROM watchlist WHERE name = ?", (name.lower(),)) as cursor: if await cursor.fetchone() is None: return False if fields: - # Column names from the literal keys above, never from input. + # Column names from literal keys in the callers, never from input. assignments = ", ".join(f"{column} = ?" for column in fields) await db.execute(f"UPDATE watchlist SET {assignments} WHERE name = ?", (*fields.values(), name.lower())) diff --git a/tests/test_api_versioned.py b/tests/test_api_versioned.py index f331bb3..dcbf121 100644 --- a/tests/test_api_versioned.py +++ b/tests/test_api_versioned.py @@ -46,8 +46,7 @@ async def test_the_version_and_cpe_are_recorded_over_http(client): assert response.status_code == 200, response.text assert response.json()["installed_version"] == "8.5.0" [entry] = await database.get_watch_entries() - assert (entry.installed_version, entry.vendor, entry.product_id) == ("8.5.0", "notepad-plus-plus", - "notepad\\+\\+") + assert entry.installed() == ("8.5.0", "notepad-plus-plus", "notepad\\+\\+") async def test_a_cpe_nvd_would_match_nothing_with_is_refused_with_its_form(client): diff --git a/tests/test_vendor_source_columns.py b/tests/test_vendor_source_columns.py new file mode 100644 index 0000000..7986280 --- /dev/null +++ b/tests/test_vendor_source_columns.py @@ -0,0 +1,76 @@ +""" +PatchRadar — `vendor` and `product_id` belong to the vendor source (docs/DESIGN.md §5.1) + +`inventory.py` named the watchlist columns before they existed: `vendor`, +`product_id`, `installed_version`, `channel`, `cpe`. In it, a vendor source is a +feed that answers by version — Chrome's, Edge's, MSRC, Mozilla's — and +`product_id` is the product that feed needs ("chrome", "Edge"); the CPE is the +separate route, NVD by CPE. #58 filled `vendor` and `product_id` with the CPE's +fields instead, which would have had `import` write a vendor source over the +CPE's vendor, or the other way round. + +The CPE's vendor and product are now read from `cpe` itself, and the two +columns are left to the vendor source, written by `set_source`. +""" +from __future__ import annotations + +import pytest + +from patchradar.cpe import parse_cpe +from patchradar.db import database + +NOTEPADPP = "cpe:2.3:a:notepad-plus-plus:notepad\\+\\+" + + +@pytest.mark.asyncio +async def test_a_cpe_does_not_fill_the_vendor_source_columns(): + await database.add_to_watchlist("notepad") + await database.set_installed("notepad", version="8.5.0", cpe=parse_cpe(NOTEPADPP)) + + [entry] = await database.get_watch_entries() + + assert (entry.vendor, entry.product_id) == (None, None) + assert (entry.cpe_vendor, entry.cpe_product) == ("notepad-plus-plus", "notepad\\+\\+") + assert entry.installed() == ("8.5.0", "notepad-plus-plus", "notepad\\+\\+") + + +@pytest.mark.asyncio +async def test_the_vendor_source_is_stored_beside_the_cpe(): + await database.add_to_watchlist("google chrome") + await database.set_installed("google chrome", version="141.0.7390.65", + cpe=parse_cpe("cpe:2.3:a:google:chrome")) + + assert await database.set_source("google chrome", vendor="chrome", product_id="chrome", channel="stable") + + [entry] = await database.get_watch_entries() + assert (entry.vendor, entry.product_id, entry.channel) == ("chrome", "chrome", "stable") + assert entry.cpe == "cpe:2.3:a:google:chrome" + assert entry.installed() == ("141.0.7390.65", "google", "chrome") + + +@pytest.mark.asyncio +async def test_a_vendor_source_alone_does_not_make_an_entry_judgeable(): + """The scan judges with NVD's configurations, which need the CPE.""" + await database.add_to_watchlist("microsoft edge") + await database.set_installed("microsoft edge", version="141.0.3537.71", cpe=None) + await database.set_source("microsoft edge", vendor="edge", product_id="Edge", channel=None) + + [entry] = await database.get_watch_entries() + assert (entry.vendor, entry.product_id) == ("edge", "Edge") + assert not entry.judgeable + + +@pytest.mark.asyncio +async def test_a_cpe_without_a_version_is_not_judgeable_either(): + await database.add_to_watchlist("notepad") + await database.set_installed("notepad", version=None, cpe=parse_cpe(NOTEPADPP)) + + [entry] = await database.get_watch_entries() + assert entry.installed() is None + + +@pytest.mark.asyncio +async def test_set_source_never_adds_a_name(): + assert not await database.set_source("thunderbird", vendor="mozilla", product_id="thunderbird", + channel="release") + assert await database.get_watchlist() == [] diff --git a/tests/test_watchlist_versions.py b/tests/test_watchlist_versions.py index 8aa3885..a51a3e3 100644 --- a/tests/test_watchlist_versions.py +++ b/tests/test_watchlist_versions.py @@ -94,7 +94,7 @@ async def test_migrating_twice_is_not_an_error(old_database): # ─── storing what is installed ─────────────────────────────────────────────── @pytest.mark.asyncio -async def test_version_and_cpe_are_stored_with_vendor_and_product(): +async def test_version_and_cpe_are_stored_and_the_cpe_names_vendor_and_product(): await database.add_to_watchlist("notepad") assert await database.set_installed("notepad", version="8.5.0", cpe=parse_cpe(NOTEPADPP)) @@ -102,8 +102,8 @@ async def test_version_and_cpe_are_stored_with_vendor_and_product(): [entry] = await database.get_watch_entries() assert entry.installed_version == "8.5.0" assert entry.cpe == NOTEPADPP - assert entry.vendor == "notepad-plus-plus" - assert entry.product_id == "notepad\\+\\+" + assert entry.cpe_vendor == "notepad-plus-plus" + assert entry.cpe_product == "notepad\\+\\+" assert entry.channel is None assert entry.judgeable From 45ef35c8f5f189acd64949dcc1039441825f3f96 Mon Sep 17 00:00:00 2001 From: maksimtech <268127343+maksimtech@users.noreply.github.com> Date: Sat, 17 Oct 2026 18:47:30 +0200 Subject: [PATCH 6/8] feat(import): --accept writes the products a person confirmed `import` proposes and a person confirms, and what gets remembered is the confirmation (inventory.py). Until the watchlist had the columns the confirmation had nowhere to go and the command wrote nothing. - `--accept ""`, once per product, any of its registry entries: a product the survey identified completely (Chrome, Edge) is watched with the registry's version and its vendor source in vendor and product_id (set_source). - A source without its product, a component, a program nothing answers for: refused with the survey's own reason, exit 1; what was confirmed beside it is still written. - Without --accept nothing is written, and the closing line says how to accept instead of naming columns this release no longer lacks. - README, the NIS2 document and DESIGN 5.6 say what import writes. Tests first, seen failing; the mutants for each refusal, the version, the exit code and the lookup by any registry entry are killed. Co-Authored-By: Claude Opus 5.5 (1M context) --- CHANGELOG.md | 6 ++ README.md | 9 ++- docs/DESIGN.md | 8 +- docs/nis2-compliance.md | 5 +- src/patchradar/cli.py | 68 +++++++++++++--- tests/test_import_accept.py | 149 ++++++++++++++++++++++++++++++++++++ 6 files changed, 230 insertions(+), 15 deletions(-) create mode 100644 tests/test_import_accept.py diff --git a/CHANGELOG.md b/CHANGELOG.md index f9e35bd..d81dcc9 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -48,6 +48,12 @@ what has already shipped, and for nothing else. See `RELEASING.md`. listing faster than the date order was (1.0 ms against 1.45 ms on the benchmark's 500 CVEs). `/api/cves` rows carry `priority`, `priority_reason` and the KEV fields; `status` has a Priority column. +- **`patchradar import --accept ""` writes what a person confirmed.** `import` proposes and a + person confirms, and until the watchlist had the columns the confirmation had nowhere to go. + Named once per product, a product the survey identified completely — Chrome, Edge — is watched + with the registry's version and its vendor source in `vendor` and `product_id`. One it did not + identify (a source without its product, a component, a program nothing answers for) is refused + with the survey's reason, and the command exits 1; without `--accept` nothing is written. - **docs/DESIGN.md.** The code referred to "the versionradar design, §5" for these columns; the text was never committed and could not be found. It is reconstructed from what `affected.py`, `inventory.py` and `tests/test_priority_wiring.py` already committed to. diff --git a/README.md b/README.md index b11afd5..019a250 100644 --- a/README.md +++ b/README.md @@ -147,8 +147,13 @@ package, because the machine worth surveying is the one without a checkout: run reconfiguration of every installed product, and it changes nothing else either. `patchradar import` then reports, per product, whether a vendor source can answer -by version. It writes nothing and guesses nothing: a name cannot be turned into a -vendor's product id by deduction, so it proposes and a person confirms. +by version. It guesses nothing: a name cannot be turned into a vendor's product id +by deduction, so it proposes and a person confirms. Without `--accept` it writes +nothing; `patchradar import --accept "Google Chrome"`, once per product, +puts a product listed as watchable by version on the watchlist with its version +and source. One the survey could not identify completely — a source without the +product it needs, a component, a program nothing answers for — is refused with +the survey's reason, however it is asked. The file boundary in the middle is the point. The analysis is Python, tested against recorded snapshots, and runs on a machine that never saw the one being diff --git a/docs/DESIGN.md b/docs/DESIGN.md index ec2e100..82e3250 100644 --- a/docs/DESIGN.md +++ b/docs/DESIGN.md @@ -140,8 +140,12 @@ expression keeps a listing of 50 a read of 50 rows. 2. **PR-01b** — in three parts: the verdict stored and the API (01b-1); KEV and MSRC entries judged through NVD's record of the same CVE (01b-2); the KEV fields stored, with the database-backed listings ranked by priority (01b-3). -3. **PR-01c** — the web UI; `patchradar import` writing the mappings a person - confirmed, `channel` included. +3. **PR-01c** — `patchradar import --accept ""` writing the products a + person confirmed, only those the survey identified completely (source and + product), into `vendor`, `product_id` and `installed_version` (01c-1); + the web UI (01c-2). `channel` stays empty until a source the survey + identifies needs one: the two certain today, Chrome and Edge, are matched + without it. ## 6. What PatchRadar does not do diff --git a/docs/nis2-compliance.md b/docs/nis2-compliance.md index ba66570..5582d89 100644 --- a/docs/nis2-compliance.md +++ b/docs/nis2-compliance.md @@ -145,8 +145,9 @@ Su disco PatchRadar scrive: Lo script `inventory.ps1` è il raccoglitore dell'inventario di una macchina Windows: va eseguito a mano, possibilmente come amministratore, e scrive un solo file JSON, nella cartella corrente o nel percorso indicato con `-OutFile`. Non contiene comandi di rete e non invia il file a nessuno; il file -resta dove viene scritto. `patchradar import` lo legge e non scrive nulla né contatta -nessuno. Per verificare le firme dei driver lo script chiede a Windows +resta dove viene scritto. `patchradar import` lo legge e non contatta nessuno; scrive +nella watchlist locale solo i prodotti che una persona conferma con `--accept`, e niente +senza. Per verificare le firme dei driver lo script chiede a Windows `Get-AuthenticodeSignature`, e per farlo Windows può contattare i server delle CA indicate nei certificati: quei contatti li decide la configurazione di Windows, non PatchRadar. diff --git a/src/patchradar/cli.py b/src/patchradar/cli.py index e9010af..d38782f 100644 --- a/src/patchradar/cli.py +++ b/src/patchradar/cli.py @@ -38,6 +38,7 @@ remove_from_watchlist, save_cve, set_installed, + set_source, ) from patchradar.epss import enrich as epss_enrich from patchradar.names import SOFTWARE_NAME_MAX_LENGTH, normalise_software_name @@ -741,15 +742,18 @@ def import_inventory( path: str = typer.Argument(..., help="Inventory snapshot written by `patchradar collector`"), show: int = typer.Option(12, "--show", "-n", help="How many entries to list per group; 0 for all"), + accept: list[str] = typer.Option(None, "--accept", + help="Watch this product, once per product: only one listed " + "as watchable by version"), ): """Read a machine's inventory and report what could be watched by version. - Writes nothing, asks nothing of the network, decides nothing. The mapping - from what the registry writes to what a vendor API wants cannot be deduced - from a name — measured on a real machine, 30% mapped by themselves and a - guess produced `MX5` → a Juniper router — so this proposes and a person - confirms. Storing a confirmed mapping needs the watchlist columns that do - not exist yet. + Asks nothing of the network and decides nothing. The mapping from what the + registry writes to what a vendor API wants cannot be deduced from a name — + measured on a real machine, 30% mapped by themselves and a guess produced + `MX5` → a Juniper router — so this proposes and a person confirms: without + `--accept` it writes nothing, and with it only the products named, and only + those the survey identified completely (docs/DESIGN.md §5.6). """ from patchradar.inventory import CERTAIN, PROPOSED, read_snapshot, survey @@ -818,9 +822,55 @@ def import_inventory( if limit is not None and len(rows) > limit: console.print(f" [dim]…and {len(rows) - limit} more[/dim]") - console.print("\n[dim]Nothing was written. A confirmed mapping needs the watchlist " - "columns (vendor, product_id, installed_version, channel, cpe), " - "which this release does not have.[/dim]") + if not accept: + console.print("\n[dim]Nothing was written. To watch a product listed as watchable by " + "version, confirm it: patchradar import --accept \"\", once per " + "product.[/dim]") + return + written, refused = run(_write_accepted(found, accept)) + console.print(f"\nWritten: {written}. Everything else was only reported.") + if refused: + raise typer.Exit(1) + + +async def _write_accepted(found, names: list[str]) -> tuple[int, int]: + """Write the products a person confirmed; (written, refused). + + Only a product the survey identified completely — the vendor source and the + product it needs — is written. A source without its product, a component, a + program nothing answers for: refused with the survey's own reason, because + writing it would be the guess `import` exists to avoid. + """ + from patchradar.inventory import CERTAIN, Family + + by_name: dict[str, Family] = {} + for family in found.families: + by_name.setdefault(family.name.lower(), family) + for candidate in family.candidates: + by_name.setdefault(candidate.name.lower(), family) + written = refused = 0 + for asked in names: + family = by_name.get(asked.strip().lower()) + if family is None: + console.print(Text(f"❌ {asked} is not in this inventory", style="red")) + refused += 1 + continue + best = family.best + if family.confidence is not CERTAIN or not best.source or not best.product_id: + source = f"the source is {best.source}; " if best.source else "" + console.print(Text(f"❌ {family.name}: not written — {source}{best.reason}", style="red")) + refused += 1 + continue + # A certain product is matched by an exact name ("Google Chrome", + # "Microsoft Edge"), which the watchlist always takes. + name = family.name.strip().lower() + await add_to_watchlist(name) + await set_installed(name, version=best.installed_version, cpe=None) + await set_source(name, vendor=best.source, product_id=best.product_id, channel=None) + version = best.installed_version or "(no version in the registry)" + console.print(Text(f"✅ Watching {name} {version} — {best.source}/{best.product_id}")) + written += 1 + return written, refused POSITION_STYLES = { diff --git a/tests/test_import_accept.py b/tests/test_import_accept.py new file mode 100644 index 0000000..53b0fe3 --- /dev/null +++ b/tests/test_import_accept.py @@ -0,0 +1,149 @@ +""" +PatchRadar — `import --accept` writes what a person confirmed, and only that (DESIGN §5.6) + +`import` proposes and a person confirms, and what gets remembered is the +confirmation (`inventory.py`). Until the watchlist had the columns, the +confirmation had nowhere to go and the command wrote nothing. Now +`--accept ""`, once per product, writes a product the survey identified +completely — the vendor source and the product it needs — with the version the +registry holds, into the watchlist and its `vendor`, `product_id` and +`installed_version` columns. + +What the survey did not identify is not written however it is asked: a source +without its product, a component, a program no source answers for. Writing +those would be the silent guess the command exists to avoid. +""" +from __future__ import annotations + +import asyncio +import json + +from typer.testing import CliRunner + +from patchradar.cli import app +from patchradar.db import database + +runner = CliRunner() + + +def snapshot_file(tmp_path, *programs) -> str: + path = tmp_path / "cavia.json" + path.write_text(json.dumps({ + "schema": "radar.inventory/1", + "takenAt": "2026-09-27T21:15:42.0000000+02:00", + "installedSoftware": list(programs), + }), encoding="utf-8") + return str(path) + + +def entry(name: str, version: str | None = "1.0", publisher: str = "") -> dict: + return {"name": name, "version": version, "publisher": publisher, "hive": "HKLM"} + + +CHROME = entry("Google Chrome", "154.0.8037.58", "Google LLC") +EDGE = entry("Microsoft Edge", "154.0.3912.60", "Microsoft Corporation") +OFFICE = entry("Microsoft Office LTSC Professional Plus 2021", "16.0.14332.21037", "Microsoft Corporation") +WAVEPAD = entry("WavePad Sound Editor", "5.99", "NCH Software") + + +def _import(tmp_path, *accepted, programs=(CHROME, EDGE, OFFICE, WAVEPAD)): + args = ["import", snapshot_file(tmp_path, *programs)] + for name in accepted: + args += ["--accept", name] + result = runner.invoke(app, args) + return result, " ".join(result.output.split()) + + +def _entries(): + return {e.name: e for e in asyncio.run(database.get_watch_entries())} + + +def test_an_accepted_product_is_watched_with_its_version_and_source(tmp_path): + result, out = _import(tmp_path, "Google Chrome") + + assert result.exit_code == 0, result.output + chrome = _entries()["google chrome"] + assert (chrome.installed_version, chrome.vendor, chrome.product_id) == ("154.0.8037.58", "chrome", "chrome") + assert "google chrome" in out and "154.0.8037.58" in out + + +def test_the_name_is_matched_however_it_is_capitalised(tmp_path): + _import(tmp_path, "microsoft edge") + + edge = _entries()["microsoft edge"] + assert (edge.vendor, edge.product_id) == ("edge", "Edge") + + +def test_only_what_was_accepted_is_written(tmp_path): + _import(tmp_path, "Google Chrome") + + assert list(_entries()) == ["google chrome"] + + +def test_without_accept_nothing_is_written_and_the_way_to_accept_is_said(tmp_path): + result, out = _import(tmp_path) + + assert result.exit_code == 0 + assert _entries() == {} + assert "Nothing was written" in out + assert "--accept" in out + assert "which this release does not have" not in out + + +def test_a_source_without_its_product_is_not_written(tmp_path): + """MSRC answers for Office, and lists 13 products for one advisory: which + one this install is has not been stated, so there is nothing to confirm.""" + result, out = _import(tmp_path, "Microsoft Office LTSC Professional Plus 2021") + + assert result.exit_code == 1 + assert _entries() == {} + assert "not written" in out + assert "msrc" in out + + +def test_a_program_no_source_answers_for_is_not_written(tmp_path): + result, out = _import(tmp_path, "WavePad Sound Editor") + + assert result.exit_code == 1 + assert _entries() == {} + assert "no source that answers by version" in out + + +def test_a_name_not_in_the_inventory_is_said_to_be_missing(tmp_path): + result, out = _import(tmp_path, "Firefox") + + assert result.exit_code == 1 + assert "Firefox" in out and "not in this inventory" in out + + +def test_one_refusal_does_not_cancel_what_was_confirmed(tmp_path): + result, out = _import(tmp_path, "Google Chrome", "WavePad Sound Editor") + + assert result.exit_code == 1 + assert list(_entries()) == ["google chrome"] + assert "Written: 1" in out + + +def test_a_product_without_a_version_is_watched_and_said_to_have_none(tmp_path): + result, out = _import(tmp_path, "Google Chrome", programs=(entry("Google Chrome", None, "Google LLC"),)) + + assert result.exit_code == 0, result.output + assert _entries()["google chrome"].installed_version is None + assert "no version" in out + + +def test_accepting_again_updates_the_version(tmp_path): + _import(tmp_path, "Google Chrome") + _import(tmp_path, "Google Chrome", programs=(entry("Google Chrome", "155.0.8100.12", "Google LLC"),)) + + assert _entries()["google chrome"].installed_version == "155.0.8100.12" + + +def test_a_product_can_be_named_by_any_of_its_registry_entries(tmp_path): + """Two rows, one product: the survey groups "Google Chrome (x86)" with + "Google Chrome", and whichever row the person names, the product is Chrome.""" + result, _ = _import(tmp_path, "Google Chrome (x86)", + programs=(CHROME, entry("Google Chrome (x86)", "154.0.8037.58", "Google LLC"))) + + assert result.exit_code == 0, result.output + assert list(_entries()) == ["google chrome"] From 64de8487e96921b20ef30409f3360c6ce060c116 Mon Sep 17 00:00:00 2001 From: maksimtech <268127343+maksimtech@users.noreply.github.com> Date: Sat, 17 Oct 2026 18:53:44 +0200 Subject: [PATCH 7/8] feat(ui): the dashboard shows what is installed and the verdict The CLI and the API learnt the installed version, the verdict on each CVE and the stored priority; the dashboard still listed every CVE found by name, ranked as if each were a job, and could not say what is installed. - The watchlist shows the version beside each name; the add form takes a version and a CPE, sent to POST /api/watchlist/{name} with the CLI's checks, and a refusal shows the server's detail (NVD's form of a CPE). - The table gains Priority (its reason on hover) and Verdict columns; a product matched by name reads "not checked" rather than blank. - The CVEs that do not affect the installed version are counted and hidden behind one button, as `scan` does without --all; with none hidden there is no button. - A scan's message gives the split per judged product. Tests first, the page's real script run in Node by the existing harness; seen failing (8 of 9); 10 of 10 mutants killed. One test's routes were reordered after the first run, as test_ui_errors.py orders them: the harness takes the first matching prefix. Co-Authored-By: Claude Opus 5.5 (1M context) --- CHANGELOG.md | 6 + README.md | 5 +- src/patchradar/api/static/app.js | 68 +++++++++- src/patchradar/api/templates/index.html | 10 ++ tests/test_ui_versions.py | 169 ++++++++++++++++++++++++ 5 files changed, 251 insertions(+), 7 deletions(-) create mode 100644 tests/test_ui_versions.py diff --git a/CHANGELOG.md b/CHANGELOG.md index d81dcc9..8773478 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -54,6 +54,12 @@ what has already shipped, and for nothing else. See `RELEASING.md`. with the registry's version and its vendor source in `vendor` and `product_id`. One it did not identify (a source without its product, a component, a program nothing answers for) is refused with the survey's reason, and the command exits 1; without `--accept` nothing is written. +- **The web UI shows what is installed and the verdict.** The watchlist shows the version beside + each name; the add form takes a version and a CPE, sent with the CLI's checks, and a refused CPE + comes back with NVD's form. The table gains Priority and Verdict columns ("not checked" for a + product matched by name), and the CVEs that do not affect the installed version are counted and + hidden behind one button, as `scan` does without `--all`. A scan's message gives the split per + judged product. - **docs/DESIGN.md.** The code referred to "the versionradar design, §5" for these columns; the text was never committed and could not be found. It is reconstructed from what `affected.py`, `inventory.py` and `tests/test_priority_wiring.py` already committed to. diff --git a/README.md b/README.md index 019a250..09a09ce 100644 --- a/README.md +++ b/README.md @@ -102,7 +102,10 @@ a version are scanned by name as before, and the scan names them. The verdict is stored with the version it was given for: `patchradar status` and `/api/cves` show it while that version is still the installed one, and the API takes the version and CPE too (`POST /api/watchlist/?version=…&cpe=…`). -The web UI does not show it yet ([docs/DESIGN.md](docs/DESIGN.md) §5.5, §5.6). +The web UI shows it too: the version beside each watched name, a version and a +CPE in the add form, Priority and Verdict columns, and the CVEs that do not +affect the installed version counted and one click away +([docs/DESIGN.md](docs/DESIGN.md) §5.5, §5.6). ### Asking Debian what it is actually doing diff --git a/src/patchradar/api/static/app.js b/src/patchradar/api/static/app.js index ff32b5d..da895a6 100644 --- a/src/patchradar/api/static/app.js +++ b/src/patchradar/api/static/app.js @@ -1,6 +1,9 @@ let allCves = []; let currentFilter = 'ALL'; let currentSearch = ''; +// The CVEs that do not affect the installed version are counted and hidden, as +// `patchradar scan` does without --all (docs/DESIGN.md §5.4). +let showUnaffected = false; // The key is only needed for write/scan endpoints; reads stay public. // Stored per-browser so it is never baked into the served HTML. @@ -94,6 +97,11 @@ async function loadWatchlist() { el.innerHTML = '
No software added yet
'; return; } + // What is installed, beside the name: the scan judges each CVE against it. + const installed = {}; + (Array.isArray(data.entries) ? data.entries : []).forEach(e => { + if (e && e.installed_version) installed[e.name] = e.installed_version; + }); // CWE-79 fix: build watchlist via DOM API el.innerHTML = ''; data.watchlist.forEach(sw => { @@ -102,6 +110,12 @@ async function loadWatchlist() { const span = document.createElement('span'); span.textContent = ' ' + sw; // textContent prevents XSS + if (installed[sw]) { + const version = document.createElement('span'); + version.className = 'watch-version'; + version.textContent = ' ' + installed[sw]; + span.appendChild(version); + } const btn = document.createElement('button'); btn.className = 'btn-remove'; @@ -120,11 +134,21 @@ async function addSoftware() { const input = document.getElementById('add-input'); const sw = input.value.trim(); if (!sw) return; - const data = await api(`/api/watchlist/${encodeURIComponent(sw)}`, 'POST'); + // The version and CPE go with the name, checked by the server as the CLI + // checks them; a refusal comes back as a 422 whose detail api() shows. + const versionInput = document.getElementById('add-version'); + const cpeInput = document.getElementById('add-cpe'); + const params = []; + if (versionInput.value.trim()) params.push(`version=${encodeURIComponent(versionInput.value.trim())}`); + if (cpeInput.value.trim()) params.push(`cpe=${encodeURIComponent(cpeInput.value.trim())}`); + const query = params.length ? `?${params.join('&')}` : ''; + const data = await api(`/api/watchlist/${encodeURIComponent(sw)}${query}`, 'POST'); if (!data) return; // api() already reported the failure - if (data.added) { - toast(` Added ${sw}`); + if (data.added || params.length) { + toast(data.added ? ` Added ${sw}` : ` Updated ${sw}`); input.value = ''; + versionInput.value = ''; + cpeInput.value = ''; await loadWatchlist(); await loadStats(); } else { @@ -176,7 +200,10 @@ async function scanAll() { } else if (data.timed_out) { toast(`Scan timed out — ${data.scanned}/${data.watched} scanned, ${data.total} CVEs so far`, '#d29922'); } else if (data.total !== undefined) { - toast(`Found ${data.total} CVEs`); + // Per product judged against its installed version, the split `scan` prints. + const split = Object.entries(data.verdicts || {}).map(([sw, v]) => + `${sw}: ${v['affected']} affected, ${v['not evaluable']} not evaluable, ${v['not affected']} not affected`); + toast(`Found ${data.total} CVEs` + (split.length ? ` — ${split.join('; ')}` : '')); } await loadCves(); await loadStats(); @@ -215,11 +242,17 @@ function scoreClass(score) { function renderTable() { const searchTerm = currentSearch.toLowerCase().trim(); - const filtered = allCves.filter(c => { + const matching = allCves.filter(c => { const matchSeverity = currentFilter === 'ALL' || (c.severity||'').toUpperCase() === currentFilter; const matchSearch = !searchTerm || (c.software||'').toLowerCase().includes(searchTerm) || (c.id||'').toLowerCase().includes(searchTerm); return matchSeverity && matchSearch; }); + const unaffected = matching.filter(c => c.verdict === 'not affected').length; + const filtered = showUnaffected ? matching : matching.filter(c => c.verdict !== 'not affected'); + const toggle = document.getElementById('toggle-unaffected'); + // Nothing hidden, nothing to offer: no button rather than "Show 0". + toggle.textContent = !unaffected ? '' : showUnaffected ? 'Hide not affected' : `Show ${unaffected} not affected`; + toggle.style.display = unaffected ? '' : 'none'; const wrap = document.getElementById('cve-table-wrap'); if (!filtered.length) { @@ -250,7 +283,7 @@ function renderTable() { // Header const thead = document.createElement('thead'); const headerRow = document.createElement('tr'); - ['CVE ID','Software','Score','Severity','Description','Source','Published'].forEach(h => { + ['CVE ID','Software','Score','Severity','Priority','Verdict','Description','Source','Published'].forEach(h => { const th = document.createElement('th'); th.textContent = h; headerRow.appendChild(th); @@ -298,6 +331,25 @@ function renderTable() { tdSev.appendChild(spanSev); tr.appendChild(tdSev); + // Priority the rank the server computed, its reason on hover + const tdPri = document.createElement('td'); + const spanPri = document.createElement('span'); + spanPri.className = 'source-badge'; + spanPri.textContent = cve.priority || ''; + spanPri.title = cve.priority_reason || ''; + tdPri.appendChild(spanPri); + tr.appendChild(tdPri); + + // Verdict against the installed version; an empty one is said, not left blank + const tdVer = document.createElement('td'); + const spanVer = document.createElement('span'); + const verdict = cve.verdict || ''; + spanVer.className = `verdict verdict-${verdict ? verdict.replace(' ', '-') : 'none'}`; + spanVer.textContent = verdict || 'not checked'; + spanVer.title = cve.verdict_reason || 'matched by name; no installed version to check it against'; + tdVer.appendChild(spanVer); + tr.appendChild(tdVer); + // Description textContent prevents XSS const tdDesc = document.createElement('td'); tdDesc.className = 'desc'; @@ -474,6 +526,10 @@ function bindControls() { document.querySelectorAll('[data-filter]').forEach(btn => btn.addEventListener('click', () => setFilter(btn.dataset.filter, btn))); + document.getElementById('toggle-unaffected').addEventListener('click', () => { + showUnaffected = !showUnaffected; + renderTable(); + }); const search = document.getElementById('search-input'); search.addEventListener('input', () => { currentSearch = search.value; renderTable(); }); diff --git a/src/patchradar/api/templates/index.html b/src/patchradar/api/templates/index.html index 9ebcac0..b47169c 100644 --- a/src/patchradar/api/templates/index.html +++ b/src/patchradar/api/templates/index.html @@ -63,6 +63,11 @@ .cve-link:hover { text-decoration: underline; } .desc { color: #8b949e; max-width: 400px; } .source-badge { font-size: 0.7rem; background: #21262d; padding: 2px 6px; border-radius: 4px; color: #8b949e; } + .verdict { font-size: 0.75rem; font-weight: 600; } + .verdict-affected { color: #f85149; } + .verdict-not-evaluable { color: #d29922; } + .verdict-not-affected { color: #3fb950; } + .verdict-none, .watch-version { color: #8b949e; font-weight: 400; } .scanning { opacity: 0.6; } .toast { position: fixed; bottom: 2rem; right: 2rem; background: #238636; color: #fff; padding: 0.8rem 1.2rem; border-radius: 8px; font-size: 0.9rem; display: none; z-index: 100; } .empty { text-align: center; padding: 3rem; color: #8b949e; } @@ -126,6 +131,10 @@

Watchlist

+
+ + +
@@ -155,6 +164,7 @@

CVEs

+
diff --git a/tests/test_ui_versions.py b/tests/test_ui_versions.py new file mode 100644 index 0000000..964a5b5 --- /dev/null +++ b/tests/test_ui_versions.py @@ -0,0 +1,169 @@ +""" +PatchRadar — the web UI shows what is installed and the verdict (docs/DESIGN.md §5.4, §5.6) + +The command line and the API learnt the installed version, the verdict on each +CVE and the priority stored with it (#58–#61); the dashboard still listed every +CVE found by name, ranked as if each were a job, with no way to say what is +installed. Now: + +- the watchlist shows the version beside the name, and the add form takes a + version and a CPE, sent to `POST /api/watchlist/{name}` like the CLI's; +- the table has a Verdict and a Priority column, and hides the CVEs that do + not affect the installed version — counted, one click away — as `scan` + does without `--all`; +- a scan's toast gives the split per judged product. + +Runs the page's real script in Node against scripted API answers +(tests/js/ui_harness.js), as tests/test_ui_errors.py does. +""" +from __future__ import annotations + +import json +import shutil +import subprocess +from pathlib import Path +from urllib.parse import quote + +import pytest + +REPO = Path(__file__).resolve().parents[1] +HTML = REPO / "src" / "patchradar" / "api" / "templates" / "index.html" +HARNESS = REPO / "tests" / "js" / "ui_harness.js" + +pytestmark = pytest.mark.skipif(shutil.which("node") is None, reason="node not installed") + +NOTEPADPP = "cpe:2.3:a:notepad-plus-plus:notepad\\+\\+" + + +def _cve(cve_id, verdict, priority="scored", software="notepad"): + return {"id": cve_id, "software": software, "cvss_score": 7.8, "severity": "HIGH", + "description": cve_id, "source": "NVD", "published_at": "2026-06-26T00:00:00Z", + "verdict": verdict, "verdict_reason": f"reason of {cve_id}" if verdict else None, + "priority": priority, "priority_reason": f"why {cve_id}"} + + +CVES = [ + _cve("CVE-2026-0001", "affected", priority="ransomware"), + _cve("CVE-2026-0002", "not evaluable"), + _cve("CVE-2026-0003", "not affected"), + _cve("CVE-2026-0004", None, software="nginx"), +] + +ROUTES = { + "/api/watchlist": {"status": 200, "json": { + "watchlist": ["nginx", "notepad"], + "entries": [{"name": "nginx", "installed_version": None, "cpe": None}, + {"name": "notepad", "installed_version": "8.5.0", "cpe": NOTEPADPP}]}}, + "/api/stats": {"status": 200, "json": {"total_cves": 4, "watched": 2, + "by_severity": {"HIGH": 4}, "by_software": {"notepad": 3}}}, + "/api/cves": {"status": 200, "json": {"cves": CVES, "total": 4}}, +} + + +def run_page(tmp_path, actions=(), inputs=None, routes_after_init=None): + scenario = {"routes": ROUTES, "actions": list(actions), "inputs": inputs or {}} + if routes_after_init: + # The harness takes the first route that is a prefix of the path, so the + # specific ones go first: "/api/watchlist/notepad" before "/api/watchlist". + merged = {**ROUTES, **routes_after_init} + scenario["routesAfterInit"] = dict(sorted(merged.items(), key=lambda kv: -len(kv[0]))) + spec = tmp_path / "scenario.json" + spec.write_text(json.dumps(scenario), encoding="utf-8") + out = subprocess.run(["node", str(HARNESS), str(HTML), str(spec)], + capture_output=True, text=True, timeout=30, encoding="utf-8", errors="replace") + assert out.returncode == 0, out.stderr + result = json.loads(out.stdout) + assert result["errors"] == [], "\n".join(result["errors"]) + return result + + +def table(result) -> str: + return result["elements"]["cve-table-wrap"]["text"] + + +# ─── the watchlist ─────────────────────────────────────────────────────────── + +def test_the_watchlist_shows_the_installed_version(tmp_path): + text = run_page(tmp_path)["elements"]["watchlist-items"]["text"] + + assert "notepad" in text and "8.5.0" in text + assert "nginx" in text + + +def test_the_add_form_sends_the_version_and_the_cpe(tmp_path): + result = run_page(tmp_path, ['await fire("#add-btn", "click")'], + {"add-input": "notepad", "add-version": "8.5.0", "add-cpe": NOTEPADPP}) + + posts = [r["path"] for r in result["requests"] if r["method"] == "POST"] + assert posts == [f"/api/watchlist/notepad?version=8.5.0&cpe={quote(NOTEPADPP, safe='-_.!~*()')}"] + + +def test_without_a_version_the_add_form_sends_the_name_alone(tmp_path): + result = run_page(tmp_path, ['await fire("#add-btn", "click")'], {"add-input": "redis"}) + + assert [r["path"] for r in result["requests"] if r["method"] == "POST"] == ["/api/watchlist/redis"] + + +def test_a_refused_cpe_shows_the_form_nvd_wants(tmp_path): + refusal = {"status": 422, "json": {"detail": "the product 'notepad++' has characters NVD writes " + "escaped; NVD's form is 'notepad\\+\\+'"}} + result = run_page(tmp_path, ['await fire("#add-btn", "click")'], + {"add-input": "notepad", "add-version": "8.5.0", + "add-cpe": "cpe:2.3:a:notepad-plus-plus:notepad++"}, + routes_after_init={"/api/watchlist/notepad": refusal}) + + assert any("notepad\\+\\+" in toast for toast in result["toasts"]) + + +# ─── the table ─────────────────────────────────────────────────────────────── + +def test_the_table_has_a_verdict_and_a_priority_column(tmp_path): + text = table(run_page(tmp_path)) + + assert "Verdict" in text and "Priority" in text + assert "ransomware" in text + assert "not evaluable" in text + + +def test_what_does_not_affect_the_installed_version_is_hidden_and_counted(tmp_path): + result = run_page(tmp_path) + + text = table(result) + assert "CVE-2026-0001" in text and "CVE-2026-0002" in text and "CVE-2026-0004" in text + assert "CVE-2026-0003" not in text + assert "1 not affected" in result["elements"]["toggle-unaffected"]["text"] + + +def test_one_click_shows_them_and_another_hides_them(tmp_path): + shown = run_page(tmp_path, ['await fire("#toggle-unaffected", "click")']) + hidden = run_page(tmp_path, ['await fire("#toggle-unaffected", "click")', + 'await fire("#toggle-unaffected", "click")']) + + assert "CVE-2026-0003" in table(shown) + assert "CVE-2026-0003" not in table(hidden) + + +def test_a_cve_matched_by_name_says_its_version_was_not_checked(tmp_path): + """nginx has no version: its CVE is listed, and the verdict cell does not + leave the reader to guess why it is empty.""" + assert "not checked" in table(run_page(tmp_path)) + + +# ─── a scan ────────────────────────────────────────────────────────────────── + +def test_a_scan_says_the_split_for_each_judged_product(tmp_path): + scan = {"status": 200, "json": {"total": 21, "scanned": 1, "watched": 1, "timed_out": False, + "errors": [], "by_software": {"notepad": 21}, + "verdicts": {"notepad": {"affected": 10, "not evaluable": 0, + "not affected": 11}}}} + result = run_page(tmp_path, ['await fire("#scan-btn", "click")'], routes_after_init={"/api/scan": scan}) + + assert any("Found 21 CVEs" in t and "notepad: 10 affected, 0 not evaluable, 11 not affected" in t + for t in result["toasts"]), result["toasts"] + + +def test_with_nothing_hidden_there_is_no_button_to_show_it(tmp_path, monkeypatch): + only_affected = {"/api/cves": {"status": 200, "json": {"cves": [CVES[0]], "total": 1}}} + monkeypatch.setitem(ROUTES, "/api/cves", only_affected["/api/cves"]) + + assert run_page(tmp_path)["elements"]["toggle-unaffected"]["text"] == "" From 44401b0a6f06e63e64b626dce0b5c68ef6858076 Mon Sep 17 00:00:00 2001 From: maksimtech <268127343+maksimtech@users.noreply.github.com> Date: Sat, 17 Oct 2026 22:17:38 +0200 Subject: [PATCH 8/8] fix(db): no statement is assembled from a formatted string Bandit's B608 flagged three: the watchlist UPDATE whose column list was joined at run time (#58, open on main as alert 61) and the two listings of get_cves whose select list and ORDER BY were interpolated (#64, alerts 64 and 65). Every interpolated piece was a module constant, so no input reached them - but the project takes no `# nosec`, and a statement written out leaves nothing to vouch for. - set_installed / set_source run fixed UPDATE statements, one per field. - get_cves' two listings and the priority index are written out; their ORDER BY is priority.sql_order() verbatim, and the new test holds the text to it, so the one-source rule stands without building from it. EXPLAIN still shows the listing scanning the priority index. Test first and seen failing: no SELECT/INSERT/UPDATE/DELETE passed to execute() may be an f-string or a concatenation. test_priority_wiring now reads the written-out ORDER BY. Bandit B608: no issues. Co-Authored-By: Claude Opus 5.5 (1M context) --- src/patchradar/db/database.py | 87 +++++++++++++++++++++--------- tests/test_priority_wiring.py | 7 +-- tests/test_sql_is_not_formatted.py | 57 ++++++++++++++++++++ 3 files changed, 122 insertions(+), 29 deletions(-) create mode 100644 tests/test_sql_is_not_formatted.py diff --git a/src/patchradar/db/database.py b/src/patchradar/db/database.py index b727544..d2e9822 100644 --- a/src/patchradar/db/database.py +++ b/src/patchradar/db/database.py @@ -6,7 +6,7 @@ import aiosqlite from patchradar.cpe import Cpe, parse_cpe -from patchradar.priority import as_flag, sql_order +from patchradar.priority import as_flag logger = logging.getLogger(__name__) @@ -191,7 +191,20 @@ async def init_db(): # order instead of ranking every stored CVE first: measured on the # benchmark's 500, get_cves went from 1.9 ms to 1.0 ms with it, below # the 1.45 ms it took when it still listed by date. - await db.execute(f"CREATE INDEX IF NOT EXISTS cves_by_priority ON cves ({_PRIORITY_INDEX})") + # The expression is priority.sql_order() written out, so that no + # statement here is assembled from strings; tests/test_sql_is_not_formatted.py + # holds the two to the same text. + await db.execute(""" + CREATE INDEX IF NOT EXISTS cves_by_priority ON cves (CASE + WHEN known_exploited = 1 AND kev_ransomware = 1 THEN 5 + WHEN known_exploited = 1 THEN 4 + WHEN epss_score >= 0.1 AND epss_score <= 1.0 THEN 3 + WHEN cvss_exploit_maturity IN ('ATTACKED', 'PROOF_OF_CONCEPT') THEN 2 + WHEN cvss_score IS NOT NULL THEN 1 + ELSE 0 END DESC, + COALESCE(cvss_score, -1.0) DESC, + published_at DESC, id DESC) + """) await _normalize_stored_timestamps(db) await _link_stored_cves(db) await db.commit() @@ -259,31 +272,38 @@ async def set_installed(name: str, *, version: str | None, cpe: Cpe | None) -> b Only what is given changes: a new version leaves the CPE as it was. False when the name is not on the watchlist — this never adds one. """ - fields: dict[str, str | None] = {} + updates = [] if version is not None: - fields["installed_version"] = version + updates.append(("UPDATE watchlist SET installed_version = ? WHERE name = ?", (version, name.lower()))) if cpe is not None: - fields["cpe"] = cpe.text - return await _update_entry(name, fields) + updates.append(("UPDATE watchlist SET cpe = ? WHERE name = ?", (cpe.text, name.lower()))) + return await _update_entry(name, updates) async def set_source(name: str, *, vendor: str, product_id: str, channel: str | None) -> bool: """Record the vendor source that answers by version for an entry already on the watchlist — `inventory.py`'s source, product and channel, once a person has confirmed them. False when the name is not on the watchlist.""" - return await _update_entry(name, {"vendor": vendor, "product_id": product_id, "channel": channel}) + return await _update_entry(name, [( + "UPDATE watchlist SET vendor = ?, product_id = ?, channel = ? WHERE name = ?", + (vendor, product_id, channel, name.lower()), + )]) + +async def _update_entry(name: str, updates: list[tuple[str, tuple]]) -> bool: + """Run fixed UPDATE statements on an entry already on the watchlist. -async def _update_entry(name: str, fields: dict[str, str | None]) -> bool: + Fixed statements, one per field, rather than a column list joined at run + time: Bandit's B608 flagged the joined one on #58, and a statement written + out leaves nothing to vouch for. + """ async with aiosqlite.connect(DB_PATH) as db: async with db.execute("SELECT 1 FROM watchlist WHERE name = ?", (name.lower(),)) as cursor: if await cursor.fetchone() is None: return False - if fields: - # Column names from literal keys in the callers, never from input. - assignments = ", ".join(f"{column} = ?" for column in fields) - await db.execute(f"UPDATE watchlist SET {assignments} WHERE name = ?", - (*fields.values(), name.lower())) + for statement, values in updates: + await db.execute(statement, values) + if updates: await db.commit() return True @@ -394,14 +414,13 @@ async def save_cve(cve: dict) -> bool: # installed version is still the one its verdict was given for, so a stale one # reads as NULL, as for a CVE never judged. In the join rather than as a CASE per # column: four CASEs made every listing 12% slower (CodSpeed, #59). -_CURRENT_VERDICT = ", ".join(f"link.{column}" for column in _ADDED_LINK_COLUMNS) - - -# Most urgent first, as a scan orders its report (priority.py); among equals the -# newest first, then by id, which is also the order a scan's ties come in. The +# +# The listings below are ordered most urgent first, as a scan orders its report: +# their ORDER BY is priority.sql_order("cves") written out, then the newest +# first and the id, which is the order a scan's ties come in. Written out, not +# interpolated, so that no statement here is assembled from strings; +# tests/test_sql_is_not_formatted.py holds it to priority's own text, and the # index in init_db holds the same expression, unqualified, which SQLite matches. -_PRIORITY_ORDER = f"{sql_order('cves')}, published_at DESC, cves.id DESC" -_PRIORITY_INDEX = f"{sql_order()}, published_at DESC, id DESC" def _read(row) -> dict: @@ -418,24 +437,40 @@ async def get_cves(software: str | None = None, limit: int = 50) -> list[dict]: if software: # The column as well as the links, so a row written without a link # (by hand, or by an older release after the migration) still shows. - async with db.execute(f""" - SELECT cves.*, {_CURRENT_VERDICT} FROM cves + async with db.execute(""" + SELECT cves.*, link.verdict, link.verdict_reason, link.verdict_fixed_version, + link.judged_version FROM cves LEFT JOIN watchlist AS watched ON watched.name = ? LEFT JOIN cve_software AS link ON link.cve_id = cves.id AND link.software = ? AND link.judged_version = watched.installed_version WHERE cves.software = ? OR cves.id IN (SELECT cve_id FROM cve_software WHERE software = ?) - ORDER BY {_PRIORITY_ORDER} LIMIT ? + ORDER BY CASE + WHEN cves.known_exploited = 1 AND cves.kev_ransomware = 1 THEN 5 + WHEN cves.known_exploited = 1 THEN 4 + WHEN cves.epss_score >= 0.1 AND cves.epss_score <= 1.0 THEN 3 + WHEN cves.cvss_exploit_maturity IN ('ATTACKED', 'PROOF_OF_CONCEPT') THEN 2 + WHEN cves.cvss_score IS NOT NULL THEN 1 + ELSE 0 END DESC, + COALESCE(cves.cvss_score, -1.0) DESC, published_at DESC, cves.id DESC LIMIT ? """, (software.lower(),) * 4 + (limit,)) as cursor: rows = await cursor.fetchall() # Named after the product asked about, not the first one to find it. return [{**_read(row), "software": software.lower()} for row in rows] else: - async with db.execute(f""" - SELECT cves.*, {_CURRENT_VERDICT} FROM cves + async with db.execute(""" + SELECT cves.*, link.verdict, link.verdict_reason, link.verdict_fixed_version, + link.judged_version FROM cves LEFT JOIN watchlist AS watched ON watched.name = cves.software LEFT JOIN cve_software AS link ON link.cve_id = cves.id AND link.software = cves.software AND link.judged_version = watched.installed_version - ORDER BY {_PRIORITY_ORDER} LIMIT ? + ORDER BY CASE + WHEN cves.known_exploited = 1 AND cves.kev_ransomware = 1 THEN 5 + WHEN cves.known_exploited = 1 THEN 4 + WHEN cves.epss_score >= 0.1 AND cves.epss_score <= 1.0 THEN 3 + WHEN cves.cvss_exploit_maturity IN ('ATTACKED', 'PROOF_OF_CONCEPT') THEN 2 + WHEN cves.cvss_score IS NOT NULL THEN 1 + ELSE 0 END DESC, + COALESCE(cves.cvss_score, -1.0) DESC, published_at DESC, cves.id DESC LIMIT ? """, (limit,)) as cursor: rows = await cursor.fetchall() return [_read(row) for row in rows] diff --git a/tests/test_priority_wiring.py b/tests/test_priority_wiring.py index 819de0c..c74f2ad 100644 --- a/tests/test_priority_wiring.py +++ b/tests/test_priority_wiring.py @@ -80,8 +80,9 @@ def test_the_database_stores_every_fact_a_rank_is_decided_by(): def test_the_stored_listings_are_ordered_by_priority_not_by_date(): """Named here so the two paths are compared in one place rather than discovered to differ: `get_cves`, which `status` and `/api/cves` read, orders - with the expression `priority.sql_order` builds from the scan's own rules.""" + by the expression `priority.sql_order` builds from the scan's own rules — + written out in the statement, and held to that text by + tests/test_sql_is_not_formatted.py.""" body = _function_source(DB_SOURCE, "get_cves") - assert body.count("ORDER BY {_PRIORITY_ORDER}") == 2 - assert "sql_order(" in DB_SOURCE + assert body.count("ORDER BY CASE") == 2 assert "ORDER BY published_at DESC" not in body diff --git a/tests/test_sql_is_not_formatted.py b/tests/test_sql_is_not_formatted.py new file mode 100644 index 0000000..4e6c2b7 --- /dev/null +++ b/tests/test_sql_is_not_formatted.py @@ -0,0 +1,57 @@ +""" +PatchRadar — no statement in the database layer is built from a formatted string + +Bandit's B608 flagged three on #58 and #64: the watchlist UPDATE whose column +list was joined at run time, and the two listings of `get_cves` whose select +list and ORDER BY were interpolated. Every interpolated piece was a module +constant, so nothing a user typed could reach them — but the project accepts no +`# nosec`, and the cleaner answer is a statement written out: there is then +nothing for a reader, or a scanner, to trust. + +The listings' ORDER BY is still the order `priority()` ranks by: it is written +out in database.py and checked against `priority.sql_order` here, which keeps +the one-source rule without building the statement from it. +""" +from __future__ import annotations + +import ast +import pathlib +import re + +import patchradar.db.database as database +from patchradar.priority import sql_order + +SOURCE = pathlib.Path(database.__file__).read_text(encoding="utf-8") +DML = re.compile(r"^\s*(SELECT|INSERT|UPDATE|DELETE)\b", re.I) + + +def _formatted_statements() -> list[int]: + lines = [] + for node in ast.walk(ast.parse(SOURCE)): + if not (isinstance(node, ast.Call) and getattr(node.func, "attr", "") in {"execute", "executemany"}): + continue + if not node.args: + continue + statement = node.args[0] + if isinstance(statement, ast.JoinedStr): + text = "".join(v.value for v in statement.values if isinstance(v, ast.Constant)) + if DML.match(text): + lines.append(node.lineno) + elif isinstance(statement, ast.BinOp | ast.Call): + lines.append(node.lineno) + return lines + + +def test_no_select_insert_update_or_delete_is_a_formatted_string(): + assert _formatted_statements() == [] + + +def test_the_listings_are_ordered_by_what_priority_ranks_by(): + order = re.sub(r"\s+", " ", sql_order("cves")) + listings = re.sub(r"\s+", " ", SOURCE) + assert listings.count(f"ORDER BY {order}, published_at DESC, cves.id DESC LIMIT ?") == 2 + + +def test_the_priority_index_holds_the_same_expression(): + order = re.sub(r"\s+", " ", sql_order()) + assert f"ON cves ({order}, published_at DESC, id DESC)" in re.sub(r"\s+", " ", SOURCE)