diff --git a/CHANGELOG.md b/CHANGELOG.md index 49bc466..1f7fc72 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -26,6 +26,12 @@ what has already shipped, and for nothing else. See `RELEASING.md`. split, one what an update closes, and the NVD caveat is always printed. Entries without a version are scanned by name, and the scan says which. `list` shows version and CPE. The verdict is not stored yet (§5.5): `status`, the API and the web UI are the next step. +- **The verdict is kept, and the API judges too** (docs/DESIGN.md §5.5). In `cve_software`, per + product, with the installed version it was given for: after a new `--version` the old verdict + is not shown until a scan judges the new one. `patchradar status` and `/api/cves` show it; + `POST /api/watchlist/{name}` takes `version` and `cpe` with the CLI's checks, `GET + /api/watchlist` adds `entries`, and `POST /api/scan` judges with the CLI's function + (`affected.mark`) and returns the split per product in `verdicts`. - **docs/DESIGN.md.** The code referred to "the versionradar design, §5" for these columns; the text was never committed and could not be found. It is reconstructed from what `affected.py`, `inventory.py` and `tests/test_priority_wiring.py` already committed to. diff --git a/README.md b/README.md index d3e195c..146a7a7 100644 --- a/README.md +++ b/README.md @@ -98,8 +98,10 @@ affected ones — the rest stay open because NVD does not state where they were fixed, not because the update falls short — and a count from NVD is always followed by the reminder that it counts identifiers, not flaws. Entries without a version are scanned by name as before, and the scan names them. The verdict -is not stored yet, so `status`, the API and the web UI still show CVEs by name -([docs/DESIGN.md](docs/DESIGN.md) §5.5). +is stored with the version it was given for: `patchradar status` and +`/api/cves` show it while that version is still the installed one, and the API +takes the version and CPE too (`POST /api/watchlist/?version=…&cpe=…`). +The web UI does not show it yet ([docs/DESIGN.md](docs/DESIGN.md) §5.5, §5.6). ### Asking Debian what it is actually doing diff --git a/docs/DESIGN.md b/docs/DESIGN.md index 337d3c8..479e72a 100644 --- a/docs/DESIGN.md +++ b/docs/DESIGN.md @@ -104,16 +104,25 @@ The total line keeps its meaning: CVEs found by name. ### 5.5 What is stored -PR-01a judges in memory, during the scan. Storing the verdict — so that -`status`, the API and the web UI can show it — needs columns in `cves` and is the -migration `tests/test_priority_wiring.py` waits for, together with the KEV fields. +The verdict is kept on the link between a CVE and a product, `cve_software` +(`verdict`, `verdict_reason`, `verdict_fixed_version`, `judged_version`), because +one CVE stored once can be affected for one watched product and not for another. +The latest scan decides it. It is read back only while the installed version is +still `judged_version`: after `add --version` with a new version, the old verdict +is not shown until a scan judges the new one. `status` and `/api/cves` show it; +the API records version and CPE (`POST /api/watchlist/{name}?version=&cpe=`) and +judges with the same function as the CLI (`affected.mark`). + +Still in memory only: the KEV fields that decide a rank — the migration +`tests/test_priority_wiring.py` waits for. ### 5.6 Order of work 1. **PR-01a** — the columns, `add --version --cpe`, the verdict for NVD records in the command-line scan, `--all`, the README. -2. **PR-01b** — KEV and MSRC entries judged through NVD's record of the same - CVE; the verdict and the KEV fields stored; the API. +2. **PR-01b** — in two parts: the verdict stored and the API (01b-1); the KEV + fields stored, with the database-backed listings ranked by priority, and KEV + and MSRC entries judged through NVD's record of the same CVE (01b-2). 3. **PR-01c** — the web UI; `patchradar import` writing the mappings a person confirmed, `channel` included. diff --git a/src/patchradar/affected.py b/src/patchradar/affected.py index 9cc052e..70b8ec6 100644 --- a/src/patchradar/affected.py +++ b/src/patchradar/affected.py @@ -26,11 +26,11 @@ ever assigned an identifier. `Summary.caveat` carries that sentence so a caller cannot print "no known vulnerabilities" without it. -**Wired into the command-line scan** for a watchlist entry that has both an -installed version and a CPE (docs/DESIGN.md §5): `judge` gives each CVE found by -name one of three verdicts — affected, not affected, not evaluable — and -`closed_by` says what one update closes. Storing the verdict, and the API, are -the next step (§5.5); `tests/test_priority_wiring.py` holds that boundary. +**Wired into both scans**, the CLI's and the API's, for a watchlist entry that +has both an installed version and a CPE (docs/DESIGN.md §5): `judge` gives each +CVE found by name one of three verdicts — affected, not affected, not evaluable +— `mark` applies it to a scan's records, and `closed_by` says what one update +closes. The verdict is stored with the version it was given for (§5.5). """ from __future__ import annotations @@ -451,3 +451,21 @@ def judge(record: dict, *, version: str, vendor: str, cpe_product: str) -> Verdi f"compared with {version}") return Verdict(NOT_AFFECTED, f"NVD states the affected versions ({stated}); " f"{version} is not among them") + + +def mark(records: list[dict], installed: tuple[str, str, str] | None) -> None: + """Give each scan record its verdict for `installed` — (version, vendor, + product), or None for an entry matched by name — and take NVD's + configurations off it, which nothing after the verdict reads. + + The one place the CLI and the API judge, so the two cannot disagree. The + record keeps the version it was judged for, which is what lets a stored + verdict be told apart from one about a version no longer installed. + """ + for record in records: + if installed is not None: + version, vendor, product = installed + verdict = judge(record, version=version, vendor=vendor, cpe_product=product) + record.update(verdict=verdict.status, verdict_reason=verdict.reason, + verdict_fixed_version=verdict.fixed_version, judged_version=version) + record.pop("configurations", None) diff --git a/src/patchradar/api/main.py b/src/patchradar/api/main.py index 31fe0d9..54458a6 100644 --- a/src/patchradar/api/main.py +++ b/src/patchradar/api/main.py @@ -15,6 +15,7 @@ from fastapi.security import APIKeyHeader from fastapi.staticfiles import StaticFiles +from patchradar.affected import AFFECTED, NOT_AFFECTED, NOT_EVALUABLE, mark, orderable from patchradar.collectors.debian import fetch_cves as debian_fetch from patchradar.collectors.errors import CollectorError from patchradar.collectors.kev import fetch_cves as kev_fetch @@ -22,7 +23,17 @@ from patchradar.collectors.nvd import fetch_cves as nvd_fetch from patchradar.collectors.vuldb import api_key as vuldb_api_key from patchradar.collectors.vuldb import fetch_cves as vuldb_fetch -from patchradar.db.database import add_to_watchlist, get_cves, get_watchlist, init_db, remove_from_watchlist +from patchradar.cpe import CpeError, parse_cpe +from patchradar.db.database import ( + WatchEntry, + add_to_watchlist, + get_cves, + get_watch_entries, + get_watchlist, + init_db, + remove_from_watchlist, + set_installed, +) from patchradar.epss import enrich as epss_enrich from patchradar.names import SOFTWARE_NAME_MAX_LENGTH, SOFTWARE_NAME_PATTERN, normalise_software_name from patchradar.priority import merge_by_cve @@ -194,6 +205,9 @@ def sanitize_cve(cve: dict) -> dict: # Numeric fields. The EPSS pair is passed through as stored, including a # missing one: `.get` yields None, and None here means FIRST does not score # the CVE — which a 0.0 would misreport as "almost certainly not exploited". + for field in ("verdict", "verdict_reason", "verdict_fixed_version", "judged_version"): + value = cve.get(field) + safe[field] = _strip_control_chars(value).strip()[:300] if isinstance(value, str) else None safe["cvss_score"] = cve.get("cvss_score") safe["epss_score"] = cve.get("epss_score") safe["epss_percentile"] = cve.get("epss_percentile") @@ -201,8 +215,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 +262,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 +306,8 @@ async def api_cves(software: str = Query(None, min_length=1, max_length=100), li sanitized = [sanitize_cve(c) for c in cves] return {"cves": sanitized, "total": len(sanitized)} -async def _scan_one(sw: str, days: int, errors: list[dict]) -> int: +async def _scan_one(sw: str, days: int, errors: list[dict], entry: WatchEntry | None = None, + verdicts: dict | None = None) -> int: """Collect and persist every source for a single package. A source that fails is recorded in `errors` rather than silently counted @@ -321,6 +356,13 @@ async def _scan_one(sw: str, days: int, errors: list[dict]) -> int: logger.warning("scan of %r unranked by EPSS: %s", sw, exc) errors.append({"software": sw, "source": exc.source, "reason": exc.reason, "status": exc.status}) + # Judged by the function the CLI judges with, before anything is stored, so + # the verdict lands with the link and the two scans give one answer. + installed = entry.installed() if entry is not None else None + mark(collected, installed) + if installed is not None and verdicts is not None: + found = [cve["verdict"] for cve in collected] + verdicts[sw] = {status: found.count(status) for status in (AFFECTED, NOT_EVALUABLE, NOT_AFFECTED)} for cve in collected: await save_cve(cve) return len(collected) @@ -334,6 +376,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 +387,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 +401,9 @@ async def api_scan(days: int = Query(7, ge=1, le=90)): return { "total": total, "by_software": results, + # Per product judged against its installed version: how many of the + # CVEs found by name affect it, could not be evaluated, do not affect it. + "verdicts": verdicts, "timed_out": timed_out, "scanned": len(results), "watched": len(watchlist), diff --git a/src/patchradar/cli.py b/src/patchradar/cli.py index f80333c..af0bc89 100644 --- a/src/patchradar/cli.py +++ b/src/patchradar/cli.py @@ -14,7 +14,7 @@ from rich.text import Text import patchradar -from patchradar.affected import AFFECTED, CAVEAT, NOT_AFFECTED, NOT_EVALUABLE, closed_by, judge, orderable +from patchradar.affected import AFFECTED, CAVEAT, NOT_AFFECTED, NOT_EVALUABLE, closed_by, mark, orderable from patchradar.collectors.debian import DEBIAN_RELEASE as debian_release from patchradar.collectors.debian import get_tracker_snapshot from patchradar.collectors.errors import CollectorError @@ -419,15 +419,7 @@ async def _scan_target( # listed, unless --all asks for them. installed = entry.installed() if entry is not None else None judged = installed is not None - if installed is not None: - version, vendor, product = installed - for cve in all_cves: - verdict = judge(cve, version=version, vendor=vendor, cpe_product=product) - cve["verdict"] = verdict.status - cve["verdict_reason"] = verdict.reason - cve["verdict_fixed_version"] = verdict.fixed_version - for cve in all_cves: - cve.pop("configurations", None) + mark(all_cves, installed) for cve in all_cves: await save_cve(cve) if collected is not None: @@ -637,11 +629,18 @@ def _print_cves_table(cves: list): table.add_column("Score", justify="center", width=6) table.add_column("Severity", justify="center", width=10) table.add_column("Source", width=6) + # Only when a stored verdict is current for some row: a watchlist matched by + # name alone gets the table it always had. + judged = any(cve.get("verdict") for cve in cves) + if judged: + table.add_column("Verdict") for cve in cves: score_str = _format_score(cve.get("cvss_score")) severity = _normalise_severity(cve.get("severity")) severity_color = SEVERITY_STYLES.get(severity, "white") + verdict = cve.get("verdict") + extra = [Text(verdict or "—", style=VERDICT_STYLES.get(verdict, "dim"))] if judged else [] table.add_row( Text(str(cve.get("id", ""))), @@ -649,6 +648,7 @@ def _print_cves_table(cves: list): score_str, Text(severity, style=severity_color), Text(str(cve.get("source") or "")), + *extra, ) console.print(table) diff --git a/src/patchradar/db/database.py b/src/patchradar/db/database.py index a02ec01..04f56f2 100644 --- a/src/patchradar/db/database.py +++ b/src/patchradar/db/database.py @@ -65,6 +65,18 @@ def normalize_timestamp(value) -> str | None: } +# The verdict on a CVE for one watched product (docs/DESIGN.md §5.5), on the +# link between the two: one CVE stored once is "affected" for one product and +# "not affected" for another. `judged_version` is the installed version it was +# given for, so a verdict about 8.5.0 is never shown as one about 8.9.6.4. +_ADDED_LINK_COLUMNS = { + "verdict": "TEXT", + "verdict_reason": "TEXT", + "verdict_fixed_version": "TEXT", + "judged_version": "TEXT", +} + + async def _add_missing_columns(db, table: str = "cves", columns: dict | None = None) -> None: """Bring an existing table up to the current schema.""" columns = _ADDED_COLUMNS if columns is None else columns @@ -148,11 +160,16 @@ async def init_db(): CREATE TABLE IF NOT EXISTS cve_software ( cve_id TEXT NOT NULL, software TEXT NOT NULL, + verdict TEXT, + verdict_reason TEXT, + verdict_fixed_version TEXT, + judged_version TEXT, PRIMARY KEY (cve_id, software) ) """) await _add_missing_columns(db) await _add_missing_columns(db, "watchlist", _ADDED_WATCHLIST_COLUMNS) + await _add_missing_columns(db, "cve_software", _ADDED_LINK_COLUMNS) await _normalize_stored_timestamps(db) await _link_stored_cves(db) await db.commit() @@ -272,10 +289,20 @@ async def save_cve(cve: dict) -> bool: """ async with aiosqlite.connect(DB_PATH) as db: try: - await db.execute( - "INSERT OR IGNORE INTO cve_software (cve_id, software) VALUES (?, ?)", - (cve["id"], cve["software"]), - ) + # The latest scan decides the verdict, unlike the description: NVD + # analysing a CVE turns "not evaluable" into an answer, and a scan by + # name, with no verdict, says the old one no longer applies. + await db.execute(""" + INSERT INTO cve_software + (cve_id, software, verdict, verdict_reason, verdict_fixed_version, judged_version) + VALUES (?, ?, ?, ?, ?, ?) + ON CONFLICT(cve_id, software) DO UPDATE SET + verdict = excluded.verdict, + verdict_reason = excluded.verdict_reason, + verdict_fixed_version = excluded.verdict_fixed_version, + judged_version = excluded.judged_version + """, (cve["id"], cve["software"], cve.get("verdict"), cve.get("verdict_reason"), + cve.get("verdict_fixed_version"), cve.get("judged_version"))) cursor = await db.execute(""" INSERT INTO cves (id, software, description, cvss_score, cvss_version, @@ -301,24 +328,37 @@ 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). +_CURRENT_VERDICT = ", ".join(f"link.{column}" for column in _ADDED_LINK_COLUMNS) + + async def get_cves(software: str | None = None, limit: int = 50) -> list[dict]: async with aiosqlite.connect(DB_PATH) as db: db.row_factory = aiosqlite.Row if software: # The column as well as the links, so a row written without a link # (by hand, or by an older release after the migration) still shows. - async with db.execute(""" - SELECT * FROM cves - WHERE software = ? OR id IN (SELECT cve_id FROM cve_software WHERE software = ?) + async with db.execute(f""" + SELECT cves.*, {_CURRENT_VERDICT} FROM cves + LEFT JOIN watchlist AS watched ON watched.name = ? + LEFT JOIN cve_software AS link ON link.cve_id = cves.id AND link.software = ? + AND link.judged_version = watched.installed_version + WHERE cves.software = ? OR cves.id IN (SELECT cve_id FROM cve_software WHERE software = ?) ORDER BY published_at DESC LIMIT ? - """, (software.lower(), software.lower(), limit)) as cursor: + """, (software.lower(),) * 4 + (limit,)) as cursor: rows = await cursor.fetchall() # Named after the product asked about, not the first one to find it. return [{**dict(row), "software": software.lower()} for row in rows] else: - async with db.execute( - "SELECT * FROM cves ORDER BY published_at DESC LIMIT ?", - (limit,) - ) as cursor: + async with db.execute(f""" + SELECT cves.*, {_CURRENT_VERDICT} FROM cves + LEFT JOIN watchlist AS watched ON watched.name = cves.software + LEFT JOIN cve_software AS link ON link.cve_id = cves.id AND link.software = cves.software + AND link.judged_version = watched.installed_version + ORDER BY published_at DESC LIMIT ? + """, (limit,)) as cursor: rows = await cursor.fetchall() return [dict(row) for row in rows] diff --git a/tests/test_api_versioned.py b/tests/test_api_versioned.py new file mode 100644 index 0000000..f331bb3 --- /dev/null +++ b/tests/test_api_versioned.py @@ -0,0 +1,123 @@ +""" +PatchRadar — the API knows what is installed, and judges as the CLI does (DESIGN §5.5) + +A scan over HTTP and a scan from the command line that give different answers +for the same machine are the failure `tests/test_collector_wiring.py` exists +for. PR-01a taught only the CLI to judge CVEs against the installed version; +the API now records the version and CPE, judges with the same function, and +returns what it stored. + +NVD answers from the recording `nvd_notepadpp_configurations.json`; MSRC has no +document (404); Debian refuses (503), so the API lists that one source in +`errors`; KEV is the empty catalogue conftest.py caches. +""" +from __future__ import annotations + +import json +import pathlib + +import httpx +import pytest +import respx +from httpx import ASGITransport, AsyncClient + +import patchradar.api.main as api +from patchradar.collectors import msrc, nvd +from patchradar.db import database + +FIXTURE = pathlib.Path(__file__).parent / "fixtures" / "nvd_notepadpp_configurations.json" +PAYLOAD = json.loads(FIXTURE.read_text(encoding="utf-8")) +NOTEPADPP = "cpe:2.3:a:notepad-plus-plus:notepad\\+\\+" + + +@pytest.fixture +async def client(): + async with AsyncClient(transport=ASGITransport(app=api.app), base_url="http://test") as ac: + yield ac + + +async def _add(client, **params): + return await client.post("/api/watchlist/notepad", params=params) + + +async def test_the_version_and_cpe_are_recorded_over_http(client): + response = await _add(client, version="8.5.0", cpe=NOTEPADPP) + + assert response.status_code == 200, response.text + assert response.json()["installed_version"] == "8.5.0" + [entry] = await database.get_watch_entries() + assert (entry.installed_version, entry.vendor, entry.product_id) == ("8.5.0", "notepad-plus-plus", + "notepad\\+\\+") + + +async def test_a_cpe_nvd_would_match_nothing_with_is_refused_with_its_form(client): + response = await _add(client, version="8.5.0", cpe="cpe:2.3:a:notepad-plus-plus:notepad++") + + assert response.status_code == 422 + assert "notepad\\+\\+" in response.json()["detail"] + assert await database.get_watchlist() == [] + + +async def test_a_version_the_scan_cannot_order_is_refused(client): + response = await _add(client, version="8.x-beta", cpe=NOTEPADPP) + + assert response.status_code == 422 + assert await database.get_watchlist() == [] + + +async def test_adding_again_with_a_version_updates_it(client): + await _add(client) + response = await _add(client, version="8.9.6.4") + + assert response.json()["added"] is False + assert response.json()["installed_version"] == "8.9.6.4" + + +async def test_the_watchlist_lists_what_is_installed_beside_the_names(client): + await _add(client, version="8.5.0", cpe=NOTEPADPP) + + body = (await client.get("/api/watchlist")).json() + + assert body["watchlist"] == ["notepad"] + assert body["entries"] == [{"name": "notepad", "installed_version": "8.5.0", "cpe": NOTEPADPP}] + + +@pytest.fixture +def upstream(): + with respx.mock(assert_all_called=False) as router: + router.get(url__startswith=nvd.NVD_API).mock(return_value=httpx.Response(200, json=PAYLOAD)) + router.get(url__startswith=msrc.MSRC_API).mock(return_value=httpx.Response(404)) + router.route().mock(return_value=httpx.Response(503)) # Debian + yield + + +async def test_a_scan_over_http_gives_the_split_the_cli_gives(client, upstream): + await _add(client, version="8.5.0", cpe=NOTEPADPP) + + body = (await client.post("/api/scan", params={"days": 7})).json() + + assert body["by_software"] == {"notepad": 21} + assert body["verdicts"] == {"notepad": {"affected": 10, "not evaluable": 0, "not affected": 11}} + + +async def test_the_cves_read_back_over_http_carry_their_verdict(client, upstream): + await _add(client, version="8.5.0", cpe=NOTEPADPP) + await client.post("/api/scan", params={"days": 7}) + + rows = (await client.get("/api/cves", params={"software": "notepad"})).json()["cves"] + + assert len(rows) == 21 + verdicts = [row["verdict"] for row in rows] + assert (verdicts.count("affected"), verdicts.count("not affected")) == (10, 11) + assert all(row["verdict_reason"] for row in rows) + + +async def test_an_entry_without_a_version_has_no_verdicts_over_http(client, upstream): + await _add(client) + + body = (await client.post("/api/scan", params={"days": 7})).json() + + assert body["by_software"] == {"notepad": 21} + assert body["verdicts"] == {} + rows = (await client.get("/api/cves", params={"software": "notepad"})).json()["cves"] + assert {row["verdict"] for row in rows} == {None} diff --git a/tests/test_verdict_stored.py b/tests/test_verdict_stored.py new file mode 100644 index 0000000..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