From e57a387b0f705fcbf7ca6070c809777049ecd8af Mon Sep 17 00:00:00 2001 From: maksimtech <268127343+maksimtech@users.noreply.github.com> Date: Sat, 17 Oct 2026 17:28:02 +0200 Subject: [PATCH] feat(priority): store what decides a rank, and rank the stored listings CISA's exploitation facts and the CVSS v4.0 exploit maturity travelled in the scan's records and were not columns: the next `status` listed by publication date, and an exploited CVE from 2023 sat under every new 5.3. test_priority_wiring held that boundary to say "now rank the database paths too" the day the columns arrived. - cves gains known_exploited, kev_ransomware, kev_date_added, kev_due_date and cvss_exploit_maturity, migrated in place and refreshed by every scan that brings them: CISA lists a CVE long after NVD publishes it. CISA's "Unknown" is stored as no. - get_cves orders by priority.sql_order, built from the constants priority() decides with; an index on that order makes the full listing 1.0 ms on the benchmark's 500 CVEs, against 1.45 ms by date. - /api/cves rows carry priority, priority_reason and the KEV fields; status has a Priority column and lists most urgent first. - test_priority_wiring now holds the other side of the boundary. Tests first, seen failing; the SQL order is checked against sort_by_priority on rows covering every rank and edge. 15 of 16 mutants killed; the survivor drops the "scored" branch, which COALESCE on the score already orders the same way (equivalent). Co-Authored-By: Claude Opus 5.5 (1M context) --- CHANGELOG.md | 10 ++ README.md | 2 +- docs/DESIGN.md | 14 ++- src/patchradar/api/main.py | 12 +- src/patchradar/cli.py | 7 +- src/patchradar/db/database.py | 67 ++++++++++-- src/patchradar/priority.py | 28 +++++ tests/test_priority_stored.py | 201 ++++++++++++++++++++++++++++++++++ tests/test_priority_wiring.py | 40 +++---- 9 files changed, 342 insertions(+), 39 deletions(-) create mode 100644 tests/test_priority_stored.py diff --git a/CHANGELOG.md b/CHANGELOG.md index e9f543c..f9e35bd 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -38,6 +38,16 @@ what has already shipped, and for nothing else. See `RELEASING.md`. scan, CLI and API alike, now asks NVD about each by id (`nvd.records_for`, `cveId=`), paced like every request to NVD, and merges the record in, score included. When NVD cannot be asked, the CVE stays not evaluable and the scan says so (the API in `errors`). +- **`status` and `/api/cves` rank as a scan does.** CISA's exploitation facts and the CVSS v4.0 + exploit maturity travelled in the scan's records and were not columns, so the next `status` + listed by publication date and an exploited CVE from 2023 sat under every new 5.3. They are + stored now (`known_exploited`, `kev_ransomware`, `kev_date_added`, `kev_due_date`, + `cvss_exploit_maturity`), refreshed by each scan that brings them — CISA lists a CVE long after + NVD publishes it — and the listings are ordered by `priority.sql_order`, built from the rules a + scan ranks by; a test runs one set of rows through both. An index on that order makes the full + listing faster than the date order was (1.0 ms against 1.45 ms on the benchmark's 500 CVEs). + `/api/cves` rows carry `priority`, `priority_reason` and the KEV fields; `status` has a Priority + column. - **docs/DESIGN.md.** The code referred to "the versionradar design, §5" for these columns; the text was never committed and could not be found. It is reconstructed from what `affected.py`, `inventory.py` and `tests/test_priority_wiring.py` already committed to. diff --git a/README.md b/README.md index 7bb1797..b11afd5 100644 --- a/README.md +++ b/README.md @@ -65,7 +65,7 @@ patchradar add notepad --version 8.5.0 --cpe "cpe:2.3:a:notepad-plus-plus:notepa patchradar scan notepad --days 365 # what affects 8.5.0, and what is unknown patchradar scan notepad --days 365 --all # also the CVEs that do not affect it -# Show latest CVEs in terminal +# Show the most urgent stored CVEs, ranked as a scan ranks them patchradar status # Remove software diff --git a/docs/DESIGN.md b/docs/DESIGN.md index 794d694..a8d17a5 100644 --- a/docs/DESIGN.md +++ b/docs/DESIGN.md @@ -118,16 +118,20 @@ is not shown until a scan judges the new one. `status` and `/api/cves` show it; the API records version and CPE (`POST /api/watchlist/{name}?version=&cpe=`) and judges with the same function as the CLI (`affected.mark`). -Still in memory only: the KEV fields that decide a rank — the migration -`tests/test_priority_wiring.py` waits for. +What decides a rank is stored too: `known_exploited`, `kev_ransomware`, +`kev_date_added`, `kev_due_date` and the provider's `cvss_exploit_maturity`, in +`cves`, refreshed by every scan that brings them (CISA lists a CVE long after NVD +publishes it). `status` and `/api/cves` list most urgent first, with the order +`priority.sql_order` builds from the rules a scan ranks by, and an index on that +expression keeps a listing of 50 a read of 50 rows. ### 5.6 Order of work 1. **PR-01a** — the columns, `add --version --cpe`, the verdict for NVD records in the command-line scan, `--all`, the README. -2. **PR-01b** — in two parts: the verdict stored and the API (01b-1); the KEV - fields stored, with the database-backed listings ranked by priority, and KEV - and MSRC entries judged through NVD's record of the same CVE (01b-2). +2. **PR-01b** — in three parts: the verdict stored and the API (01b-1); KEV + and MSRC entries judged through NVD's record of the same CVE (01b-2); the KEV + fields stored, with the database-backed listings ranked by priority (01b-3). 3. **PR-01c** — the web UI; `patchradar import` writing the mappings a person confirmed, `channel` included. diff --git a/src/patchradar/api/main.py b/src/patchradar/api/main.py index b2beabd..2c2bfd4 100644 --- a/src/patchradar/api/main.py +++ b/src/patchradar/api/main.py @@ -37,7 +37,7 @@ ) from patchradar.epss import enrich as epss_enrich from patchradar.names import SOFTWARE_NAME_MAX_LENGTH, SOFTWARE_NAME_PATTERN, normalise_software_name -from patchradar.priority import merge_by_cve +from patchradar.priority import merge_by_cve, priority logger = logging.getLogger("patchradar") @@ -206,6 +206,16 @@ def sanitize_cve(cve: dict) -> dict: # Numeric fields. The EPSS pair is passed through as stored, including a # missing one: `.get` yields None, and None here means FIRST does not score # the CVE — which a 0.0 would misreport as "almost certainly not exploited". + # Why the row is where it is, as the CLI prints it: a rank without its + # reason is a number taken on trust. + rank = priority(cve) + safe["priority"] = rank.label + safe["priority_reason"] = _strip_control_chars(rank.reason)[:300] + for flag in ("known_exploited", "kev_ransomware"): + safe[flag] = cve.get(flag) if isinstance(cve.get(flag), bool) else None + for field in ("kev_date_added", "kev_due_date"): + value = cve.get(field) + safe[field] = _strip_control_chars(value).strip()[:20] if isinstance(value, str) else None for field in ("verdict", "verdict_reason", "verdict_fixed_version", "judged_version"): value = cve.get(field) safe[field] = _strip_control_chars(value).strip()[:300] if isinstance(value, str) else None diff --git a/src/patchradar/cli.py b/src/patchradar/cli.py index ba43835..e9010af 100644 --- a/src/patchradar/cli.py +++ b/src/patchradar/cli.py @@ -635,11 +635,14 @@ def _print_cves(software: str, cves: list, judged: bool = False): console.print(table) def _print_cves_table(cves: list): - table = Table(title="🛡️ PatchRadar — Latest CVEs", box=box.ROUNDED, show_lines=True) + # Most urgent first, in the order the database now returns: a scan's report + # and `status` rank the same rows the same way. + table = Table(title="🛡️ PatchRadar — Most urgent CVEs", box=box.ROUNDED, show_lines=True) table.add_column("CVE ID", style="bold cyan", no_wrap=True) table.add_column("Software", style="blue") table.add_column("Score", justify="center", width=6) table.add_column("Severity", justify="center", width=10) + table.add_column("Priority", width=10) table.add_column("Source", width=6) # Only when a stored verdict is current for some row: a watchlist matched by # name alone gets the table it always had. @@ -653,12 +656,14 @@ def _print_cves_table(cves: list): severity_color = SEVERITY_STYLES.get(severity, "white") verdict = cve.get("verdict") extra = [Text(verdict or "—", style=VERDICT_STYLES.get(verdict, "dim"))] if judged else [] + rank = priority(cve) table.add_row( Text(str(cve.get("id", ""))), Text(str(cve.get("software") or "")), score_str, Text(severity, style=severity_color), + Text(rank.label, style=PRIORITY_STYLES[rank.rank]), Text(str(cve.get("source") or "")), *extra, ) diff --git a/src/patchradar/db/database.py b/src/patchradar/db/database.py index 04f56f2..5928757 100644 --- a/src/patchradar/db/database.py +++ b/src/patchradar/db/database.py @@ -6,6 +6,7 @@ import aiosqlite from patchradar.cpe import Cpe +from patchradar.priority import as_flag, sql_order logger = logging.getLogger(__name__) @@ -50,8 +51,19 @@ def normalize_timestamp(value) -> str | None: _ADDED_COLUMNS = { "epss_score": "REAL", "epss_percentile": "REAL", + # What decides a rank besides the score (priority.py), stored so that the + # listings read from the database can be ranked as a scan is + # (docs/DESIGN.md §5.5). The flags are 1/0, NULL for a CVE KEV never listed. + "known_exploited": "INTEGER", + "kev_ransomware": "INTEGER", + "kev_date_added": "TEXT", + "kev_due_date": "TEXT", + "cvss_exploit_maturity": "TEXT", } +# Stored as 1/0 and read back as booleans; NULL stays None. +_FLAGS = ("known_exploited", "kev_ransomware") + # What is installed, for the scan to judge each CVE against (docs/DESIGN.md # §5.1): the names `inventory.py` was written for. Nullable without a default, # for the same reason as above — a version nobody entered is an absence, and a @@ -153,6 +165,11 @@ async def init_db(): url TEXT, epss_score REAL, epss_percentile REAL, + known_exploited INTEGER, + kev_ransomware INTEGER, + kev_date_added TEXT, + kev_due_date TEXT, + cvss_exploit_maturity TEXT, seen_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP ) """) @@ -170,6 +187,11 @@ async def init_db(): await _add_missing_columns(db) await _add_missing_columns(db, "watchlist", _ADDED_WATCHLIST_COLUMNS) await _add_missing_columns(db, "cve_software", _ADDED_LINK_COLUMNS) + # The listings' order as an index, so a listing of 50 reads 50 rows in + # order instead of ranking every stored CVE first: measured on the + # benchmark's 500, get_cves went from 1.9 ms to 1.0 ms with it, below + # the 1.45 ms it took when it still listed by date. + await db.execute(f"CREATE INDEX IF NOT EXISTS cves_by_priority ON cves ({_PRIORITY_INDEX})") await _normalize_stored_timestamps(db) await _link_stored_cves(db) await db.commit() @@ -303,20 +325,34 @@ async def save_cve(cve: dict) -> bool: judged_version = excluded.judged_version """, (cve["id"], cve["software"], cve.get("verdict"), cve.get("verdict_reason"), cve.get("verdict_fixed_version"), cve.get("judged_version"))) + # The exploitation facts are refreshed like the forecast, and for a + # sharper reason: CISA adds a CVE to KEV long after NVD published + # it, and a row stored on Monday must learn on Wednesday that it is + # being exploited. COALESCE keeps a scan without the KEV record — + # a keyword KEV does not match — from erasing what it said. + flags = [int(as_flag(cve[f])) if cve.get(f) is not None else None for f in _FLAGS] cursor = await db.execute(""" INSERT INTO cves (id, software, description, cvss_score, cvss_version, - severity, published_at, source, url, epss_score, epss_percentile) - VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) + severity, published_at, source, url, epss_score, epss_percentile, + known_exploited, kev_ransomware, kev_date_added, kev_due_date, cvss_exploit_maturity) + VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) ON CONFLICT(id) DO UPDATE SET epss_score = COALESCE(excluded.epss_score, cves.epss_score), - epss_percentile = COALESCE(excluded.epss_percentile, cves.epss_percentile) + epss_percentile = COALESCE(excluded.epss_percentile, cves.epss_percentile), + known_exploited = COALESCE(excluded.known_exploited, cves.known_exploited), + kev_ransomware = COALESCE(excluded.kev_ransomware, cves.kev_ransomware), + kev_date_added = COALESCE(excluded.kev_date_added, cves.kev_date_added), + kev_due_date = COALESCE(excluded.kev_due_date, cves.kev_due_date), + cvss_exploit_maturity = COALESCE(excluded.cvss_exploit_maturity, cves.cvss_exploit_maturity) """, ( cve["id"], cve["software"], cve["description"], cve.get("cvss_score"), cve.get("cvss_version"), cve.get("severity"), normalize_timestamp(cve.get("published_at")), cve.get("source"), cve.get("url"), - cve.get("epss_score"), cve.get("epss_percentile") + cve.get("epss_score"), cve.get("epss_percentile"), + *flags, cve.get("kev_date_added"), cve.get("kev_due_date"), + cve.get("cvss_exploit_maturity"), )) await db.commit() return cursor.rowcount > 0 @@ -335,6 +371,21 @@ async def save_cve(cve: dict) -> bool: _CURRENT_VERDICT = ", ".join(f"link.{column}" for column in _ADDED_LINK_COLUMNS) +# Most urgent first, as a scan orders its report (priority.py); among equals the +# newest first, then by id, which is also the order a scan's ties come in. The +# index in init_db holds the same expression, unqualified, which SQLite matches. +_PRIORITY_ORDER = f"{sql_order('cves')}, published_at DESC, cves.id DESC" +_PRIORITY_INDEX = f"{sql_order()}, published_at DESC, id DESC" + + +def _read(row) -> dict: + record = dict(row) + for flag in _FLAGS: + if record.get(flag) is not None: + record[flag] = bool(record[flag]) + return record + + async def get_cves(software: str | None = None, limit: int = 50) -> list[dict]: async with aiosqlite.connect(DB_PATH) as db: db.row_factory = aiosqlite.Row @@ -347,18 +398,18 @@ async def get_cves(software: str | None = None, limit: int = 50) -> list[dict]: LEFT JOIN cve_software AS link ON link.cve_id = cves.id AND link.software = ? AND link.judged_version = watched.installed_version WHERE cves.software = ? OR cves.id IN (SELECT cve_id FROM cve_software WHERE software = ?) - ORDER BY published_at DESC LIMIT ? + ORDER BY {_PRIORITY_ORDER} LIMIT ? """, (software.lower(),) * 4 + (limit,)) as cursor: rows = await cursor.fetchall() # Named after the product asked about, not the first one to find it. - return [{**dict(row), "software": software.lower()} for row in rows] + return [{**_read(row), "software": software.lower()} for row in rows] else: async with db.execute(f""" SELECT cves.*, {_CURRENT_VERDICT} FROM cves LEFT JOIN watchlist AS watched ON watched.name = cves.software LEFT JOIN cve_software AS link ON link.cve_id = cves.id AND link.software = cves.software AND link.judged_version = watched.installed_version - ORDER BY published_at DESC LIMIT ? + ORDER BY {_PRIORITY_ORDER} LIMIT ? """, (limit,)) as cursor: rows = await cursor.fetchall() - return [dict(row) for row in rows] + return [_read(row) for row in rows] diff --git a/src/patchradar/priority.py b/src/patchradar/priority.py index de0112b..c799f19 100644 --- a/src/patchradar/priority.py +++ b/src/patchradar/priority.py @@ -317,6 +317,34 @@ def merge_by_cve(records: list[dict]) -> list[dict]: return [merged[key] for key in order] + loose +def as_flag(value: object) -> bool: + """A yes or no from a field that may hold CISA's "Known"/"Unknown" (see `_is_true`).""" + return _is_true(value) + + +def sql_order(table: str = "") -> str: + """ORDER BY terms that put stored rows in `sort_by_priority`'s order. + + Built from the constants `priority()` decides with, so the two cannot drift + (tests/test_priority_stored.py runs one set of rows through both), and run + by SQLite so a listing of 50 stays a query for 50. Rank first, then the + score, an absent one below every real one; the caller adds the tie-breaks. + The stored flags are 1/0/NULL, so `= 1` is the "Known" test here. Columns + are qualified with `table` when one is given, bare otherwise (an index). + """ + t = f"{table}." if table else "" + promotes = ", ".join(f"'{maturity}'" for maturity in sorted(_PROMOTES)) + return ( + f"CASE WHEN {t}known_exploited = 1 AND {t}kev_ransomware = 1 THEN {RANK_KEV_RANSOMWARE} " + f"WHEN {t}known_exploited = 1 THEN {RANK_KEV} " + f"WHEN {t}epss_score >= {EPSS_THRESHOLD} AND {t}epss_score <= 1.0 THEN {RANK_EPSS} " + f"WHEN {t}cvss_exploit_maturity IN ({promotes}) THEN {RANK_REPORTED} " + f"WHEN {t}cvss_score IS NOT NULL THEN {RANK_SCORED} " + f"ELSE {RANK_UNSCORED} END DESC, " + f"COALESCE({t}cvss_score, {_NO_SCORE_IN_SORT}) DESC" + ) + + def sort_by_priority(rows: list[dict]) -> list[dict]: """Already-merged rows, most urgent first. diff --git a/tests/test_priority_stored.py b/tests/test_priority_stored.py new file mode 100644 index 0000000..7d8d817 --- /dev/null +++ b/tests/test_priority_stored.py @@ -0,0 +1,201 @@ +""" +PatchRadar — what decides a rank is stored, and the stored listings are ranked (DESIGN §5.5) + +`known_exploited`, `kev_ransomware`, the KEV dates and the provider's CVSS v4.0 +exploit maturity travelled in the scan's records and were not columns, so the +fact that CISA has seen a CVE exploited existed for the length of one scan. +`patchradar status` and `/api/cves` listed by publication date, and an exploited +CVE from 2023 sat under every new 5.3. `tests/test_priority_wiring.py` held that +boundary on purpose, to say "now rank the database paths too" the day the +columns arrived; this is that day. + +The order is computed by SQLite, so a listing of 50 is still a query for 50 — +and it is built from the constants `priority()` uses, so the two cannot +drift: the test below runs one set of rows through both and asks for the same +order. +""" +from __future__ import annotations + +import asyncio + +import aiosqlite +import pytest +from httpx import ASGITransport, AsyncClient +from typer.testing import CliRunner + +import patchradar.api.main as api +from patchradar import cli +from patchradar.db import database +from patchradar.priority import sort_by_priority + +PRIORITY_COLUMNS = {"known_exploited", "kev_ransomware", "kev_date_added", "kev_due_date", + "cvss_exploit_maturity"} + + +def _row(cve_id, published, score=None, **facts): + return {"id": cve_id, "software": "nginx", "description": cve_id, "cvss_score": score, + "cvss_version": "3.1" if score is not None else None, + "severity": "HIGH" if score is not None else "UNKNOWN", + "published_at": published, "source": "NVD", + "url": f"https://nvd.nist.gov/vuln/detail/{cve_id}", **facts} + + +# One row per rank, and the edges between them: ties on score, no score, +# EPSS exactly at the threshold, an EPSS outside [0, 1] that must not promote, +# ransomware "Unknown" (CISA's word for no), and v4.0 maturity both ways. +ROWS = [ + _row("CVE-2026-0001", "2026-10-01T00:00:00Z", 5.3), + _row("CVE-2026-0002", "2026-09-30T00:00:00Z", 9.8), + _row("CVE-2026-0003", "2026-09-29T00:00:00Z", None), + _row("CVE-2026-0004", "2026-09-28T00:00:00Z", 6.1, epss_score=0.1, epss_percentile=0.95), + _row("CVE-2026-0005", "2026-09-27T00:00:00Z", 9.1, epss_score=0.0999, epss_percentile=0.9), + _row("CVE-2026-0006", "2026-09-26T00:00:00Z", 7.5, epss_score=1.5), + _row("CVE-2026-0007", "2026-09-25T00:00:00Z", 4.3, cvss_exploit_maturity="PROOF_OF_CONCEPT"), + _row("CVE-2026-0008", "2026-09-24T00:00:00Z", 8.8, cvss_exploit_maturity="UNREPORTED"), + _row("CVE-2023-0009", "2023-03-01T00:00:00Z", 7.8, known_exploited=True, + kev_date_added="2024-01-10", kev_due_date="2024-01-31", kev_ransomware=False), + _row("CVE-2022-0010", "2022-05-01T00:00:00Z", 6.5, known_exploited=True, + kev_date_added="2023-02-01", kev_ransomware=True), + _row("CVE-2021-0011", "2021-06-01T00:00:00Z", None, known_exploited=True, + kev_date_added="2022-03-03", kev_ransomware="Unknown"), + _row("CVE-2026-0012", "2026-09-23T00:00:00Z", 9.8), + # CVSS 0.0 is a measurement — no impact — and sorts above no score at all, + # in the same rank: the sentinel for "no score" must sit below every real one. + _row("CVE-2020-0014", "2020-01-01T00:00:00Z", 0.0, known_exploited=True, + kev_date_added="2021-11-03"), +] + + +async def _store(rows): + await database.add_to_watchlist("nginx") + for row in rows: + assert await database.save_cve(dict(row)), row["id"] + + +def _expected(rows): + """`priority()`'s order, from rows in the order the database breaks ties.""" + by_date = sorted(rows, key=lambda r: (r["published_at"], r["id"]), reverse=True) + return [r["id"] for r in sort_by_priority(by_date)] + + +# ─── the columns ───────────────────────────────────────────────────────────── + +@pytest.fixture +def old_cves(tmp_path, monkeypatch): + path = tmp_path / "old.db" + + async def _build(): + async with aiosqlite.connect(path) as db: + await db.execute(""" + CREATE TABLE cves ( + id TEXT PRIMARY KEY, software TEXT NOT NULL, description TEXT, + cvss_score REAL, cvss_version TEXT, severity TEXT, published_at TIMESTAMP, + source TEXT, url TEXT, epss_score REAL, epss_percentile REAL, + seen_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP + ) + """) + await db.commit() + + asyncio.run(_build()) + monkeypatch.setattr(database, "DB_PATH", path) + return path + + +@pytest.mark.asyncio +async def test_an_existing_cves_table_gains_the_columns_a_rank_needs(old_cves): + await database.init_db() + + async with aiosqlite.connect(old_cves) as db, db.execute("PRAGMA table_info(cves)") as cursor: + assert {row[1] for row in await cursor.fetchall()} >= PRIORITY_COLUMNS + + +@pytest.mark.asyncio +async def test_the_exploitation_fact_is_read_back(): + await _store([ROWS[8]]) + + [row] = await database.get_cves() + + assert row["known_exploited"] is True + assert row["kev_ransomware"] is False + assert (row["kev_date_added"], row["kev_due_date"]) == ("2024-01-10", "2024-01-31") + + +@pytest.mark.asyncio +async def test_cisas_unknown_is_stored_as_no_ransomware(): + await _store([ROWS[10]]) + + [row] = await database.get_cves() + + assert row["kev_ransomware"] is False + + +@pytest.mark.asyncio +async def test_a_cve_stored_before_kev_listed_it_learns_it_on_the_next_scan(): + """First writer wins for the description; it cannot for this. NVD stored the + row on Monday, CISA added it on Tuesday, Wednesday's scan must say so.""" + await _store([_row("CVE-2026-0013", "2026-09-01T00:00:00Z", 7.2)]) + await database.save_cve(_row("CVE-2026-0013", "2026-09-01T00:00:00Z", None, known_exploited=True, + kev_date_added="2026-10-15", kev_ransomware=True)) + + [row] = await database.get_cves() + + assert row["known_exploited"] is True + assert row["kev_ransomware"] is True + assert row["cvss_score"] == 7.2 + + +@pytest.mark.asyncio +async def test_a_scan_without_the_kev_record_does_not_forget_it(): + await _store([ROWS[8]]) + await database.save_cve(_row("CVE-2023-0009", "2023-03-01T00:00:00Z", 7.8)) + + [row] = await database.get_cves() + + assert row["known_exploited"] is True + + +# ─── the order ─────────────────────────────────────────────────────────────── + +@pytest.mark.asyncio +async def test_the_stored_listing_is_in_the_order_priority_gives(): + await _store(ROWS) + + assert [r["id"] for r in await database.get_cves(limit=100)] == _expected(ROWS) + + +@pytest.mark.asyncio +async def test_the_listing_for_one_product_is_in_the_same_order(): + await _store(ROWS) + + assert [r["id"] for r in await database.get_cves(software="nginx", limit=100)] == _expected(ROWS) + + +@pytest.mark.asyncio +async def test_a_short_listing_is_the_most_urgent_not_the_newest(): + await _store(ROWS) + + assert [r["id"] for r in await database.get_cves(limit=3)] == _expected(ROWS)[:3] + assert (await database.get_cves(limit=1))[0]["id"] == "CVE-2022-0010" + + +# ─── what reads it ─────────────────────────────────────────────────────────── + +async def test_the_api_listing_says_why_each_row_ranks_where_it_does(): + await _store(ROWS) + async with AsyncClient(transport=ASGITransport(app=api.app), base_url="http://test") as client: + rows = (await client.get("/api/cves", params={"limit": 100})).json()["cves"] + + assert [r["id"] for r in rows] == _expected(ROWS) + assert rows[0]["priority"] == "ransomware" + assert "in CISA KEV since 2023-02-01" in rows[0]["priority_reason"] + assert rows[0]["known_exploited"] is True + + +def test_status_lists_the_exploited_cve_above_the_newest(): + asyncio.run(_store([ROWS[0], ROWS[8]])) + + result = CliRunner().invoke(cli.app, ["status"], env={"COLUMNS": "200"}) + + out = result.output + assert out.index("CVE-2023-0009") < out.index("CVE-2026-0001") + assert "exploited" in out diff --git a/tests/test_priority_wiring.py b/tests/test_priority_wiring.py index 1c7c46a..819de0c 100644 --- a/tests/test_priority_wiring.py +++ b/tests/test_priority_wiring.py @@ -6,12 +6,13 @@ dead weight until something calls it, and a report that ranks while the API does not is a drift nobody notices until two people compare their screens. -The second half pins the boundary deliberately. `known_exploited`, -`kev_ransomware` and `kev_due_date` travel in the records the collectors return -and are **not** columns in `cves`, so the exploitation fact exists in memory -during a scan and is gone by the next `patchradar status`. Until the migration of -docs/DESIGN.md §5.5 lands, only the in-memory path can rank — and the day someone adds the -column, this test is what says "now wire the API too". +The second half pinned a boundary deliberately: `known_exploited`, +`kev_ransomware` and the KEV dates travelled in the scan's records and were not +columns in `cves`, so the exploitation fact was gone by the next `patchradar +status`, and the day someone added the columns this file was to say "now wire +the API too". They arrived with docs/DESIGN.md §5.5, and the tests below now +hold the other side: the facts are stored, and the stored listings are ranked +by the same rules (`priority.sql_order`), not by date. The EPSS pair *is* stored, which narrows that gap without closing it: a ranking computed from the database would know the forecast and still not know what CISA @@ -70,24 +71,17 @@ def test_the_report_shows_why_a_row_ranks_where_it_does(): assert "reason" in body, "a rank without its reason is a number taken on trust" -def test_the_database_does_not_store_the_exploitation_fact_yet(): - """The boundary, asserted so it cannot move silently. - - If this fails because a column was added, the ranking is no longer confined - to a single scan: `/cves` and `patchradar status` read from the database and - must order by priority too, or the same machine gets two different answers. - """ +def test_the_database_stores_every_fact_a_rank_is_decided_by(): schema = _function_source(DB_SOURCE, "init_db") - stored = [fact for fact in KEV_FACTS if fact in schema] - assert not stored, ( - "the schema now holds " + ", ".join(stored) - + " — rank the database-backed paths as well (API /cves, status)" - ) + missing = [fact for fact in KEV_FACTS if fact not in schema] + assert not missing, "a rank decided by an unstored fact cannot be repeated by status or /cves" -def test_the_api_listing_is_still_ordered_by_date(): +def test_the_stored_listings_are_ordered_by_priority_not_by_date(): """Named here so the two paths are compared in one place rather than - discovered to differ. `get_cves` orders by publication date, which is the - only order the stored columns support.""" - assert "ORDER BY published_at DESC" in DB_SOURCE - assert "order_by_priority" not in API_SOURCE + discovered to differ: `get_cves`, which `status` and `/api/cves` read, orders + with the expression `priority.sql_order` builds from the scan's own rules.""" + body = _function_source(DB_SOURCE, "get_cves") + assert body.count("ORDER BY {_PRIORITY_ORDER}") == 2 + assert "sql_order(" in DB_SOURCE + assert "ORDER BY published_at DESC" not in body