feat(priority): store what decides a rank, and rank the stored listings

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.

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

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.
 

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

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,
         )

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]

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.