diff --git a/CHANGELOG.md b/CHANGELOG.md index 49bc466..8773478 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -26,6 +26,40 @@ 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`. +- **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`). +- **`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. +- **`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. +- **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 d3e195c..09a09ce 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 @@ -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 @@ -98,8 +99,13 @@ 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 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 @@ -144,8 +150,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 337d3c8..82e3250 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 @@ -78,9 +86,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. @@ -104,18 +117,35 @@ 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`). + +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** — KEV and MSRC entries judged through NVD's record of the same - CVE; the verdict and the KEV fields stored; the API. -3. **PR-01c** — the web UI; `patchradar import` writing the mappings a person - confirmed, `channel` included. +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** — `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 7e524c2..5582d89 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 | @@ -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/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..2c2bfd4 100644 --- a/src/patchradar/api/main.py +++ b/src/patchradar/api/main.py @@ -15,17 +15,29 @@ 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 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.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 +from patchradar.priority import merge_by_cve, priority logger = logging.getLogger("patchradar") @@ -194,6 +206,19 @@ 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 safe["cvss_score"] = cve.get("cvss_score") safe["epss_score"] = cve.get("epss_score") safe["epss_percentile"] = cve.get("epss_percentile") @@ -201,8 +226,13 @@ 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). 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", @@ -243,14 +273,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 +317,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 @@ -311,6 +357,15 @@ async def _scan_one(sw: str, days: int, errors: list[dict]) -> int: # 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 @@ -321,6 +376,12 @@ 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. + 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 +395,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 +406,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 +420,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/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/src/patchradar/cli.py b/src/patchradar/cli.py index f80333c..d38782f 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 @@ -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 @@ -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 @@ -399,6 +400,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,17 +426,8 @@ 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 - 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: @@ -444,6 +444,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] — " @@ -631,24 +636,37 @@ 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. + 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 [] + 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, ) console.print(table) @@ -724,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 @@ -801,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/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/src/patchradar/db/database.py b/src/patchradar/db/database.py index a02ec01..d2e9822 100644 --- a/src/patchradar/db/database.py +++ b/src/patchradar/db/database.py @@ -5,7 +5,8 @@ import aiosqlite -from patchradar.cpe import Cpe +from patchradar.cpe import Cpe, parse_cpe +from patchradar.priority import as_flag 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 @@ -65,6 +77,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 @@ -141,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 ) """) @@ -148,11 +177,34 @@ 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) + # 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. + # 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() @@ -170,7 +222,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 @@ -179,12 +237,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 @@ -205,20 +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] = {} + 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.update(cpe=cpe.text, vendor=cpe.vendor, product_id=cpe.product) + 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, [( + "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. + + 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 the literal keys above, 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 @@ -272,24 +357,48 @@ 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"))) + # 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 @@ -301,6 +410,27 @@ async def save_cve(cve: dict) -> bool: logger.debug("save_cve failure", exc_info=True) return False +# 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). +# +# 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. + + +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 @@ -308,17 +438,39 @@ async def get_cves(software: str | None = None, limit: int = 50) -> list[dict]: # 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 = ?) - ORDER BY published_at DESC LIMIT ? - """, (software.lower(), software.lower(), limit)) as cursor: + 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 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 [{**dict(row), "software": software.lower()} for row in rows] + return [{**_read(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(""" + 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 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 [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_api_versioned.py b/tests/test_api_versioned.py new file mode 100644 index 0000000..dcbf121 --- /dev/null +++ b/tests/test_api_versioned.py @@ -0,0 +1,122 @@ +""" +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() == ("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_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"] 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 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..c74f2ad 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,18 @@ 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 + 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 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) 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"] == "" 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_verdict_stored.py b/tests/test_verdict_stored.py new file mode 100644 index 0000000..48f35cc --- /dev/null +++ b/tests/test_verdict_stored.py @@ -0,0 +1,179 @@ +""" +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 + + +@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 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