From df6f19506926224375636c87b04661f1907b6ae5 Mon Sep 17 00:00:00 2001 From: maksimtech <268127343+maksimtech@users.noreply.github.com> Date: Sat, 17 Oct 2026 16:51:39 +0200 Subject: [PATCH] feat(scan): judge each CVE against the installed version A scan found the CVEs that name a product and could not say which affect what is installed: for Notepad++ 8.5.0 the recorded NVD answer holds 21, 11 of them about other versions or another product, and all 21 were listed and ranked. affected.py could answer and waited for the watchlist columns of "the versionradar design, section 5", a text never committed. - docs/DESIGN.md rebuilds that design from what the code committed to. - Five nullable watchlist columns (cpe, vendor, product_id, installed_version, channel), migrated into existing databases. - `add --version --cpe`: the CPE typed by a person and checked for form (notepad++ is refused with NVD's notepad\+\+), a version refused when it cannot be ordered. - affected.judge: affected, not affected or not evaluable, from the NVD record's own configurations, which nvd._parse_item now keeps. Cannot tell never becomes not affected. - scan lists affected and not evaluable, --all the rest; it gives the split, what an update closes, and always the NVD caveat; entries without a version are scanned by name and named. Written test first and seen failing; the guards added after the first green run each kill their mutant, 26 of 26. Co-Authored-By: Claude Opus 5.5 (1M context) --- CHANGELOG.md | 16 ++ README.md | 37 ++++- docs/DESIGN.md | 125 ++++++++++++++++ src/patchradar/affected.py | 131 +++++++++++++--- src/patchradar/cli.py | 127 ++++++++++++++-- src/patchradar/collectors/nvd.py | 7 + src/patchradar/cpe.py | 94 ++++++++++++ src/patchradar/db/database.py | 95 ++++++++++-- src/patchradar/inventory.py | 7 +- tests/test_affected_real_payload.py | 3 +- tests/test_affected_verdict.py | 203 +++++++++++++++++++++++++ tests/test_cli_escaping.py | 20 ++- tests/test_cpe.py | 81 ++++++++++ tests/test_priority_wiring.py | 2 +- tests/test_scan_versioned.py | 188 +++++++++++++++++++++++ tests/test_watchlist_versions.py | 225 ++++++++++++++++++++++++++++ 16 files changed, 1308 insertions(+), 53 deletions(-) create mode 100644 docs/DESIGN.md create mode 100644 src/patchradar/cpe.py create mode 100644 tests/test_affected_verdict.py create mode 100644 tests/test_cpe.py create mode 100644 tests/test_scan_versioned.py create mode 100644 tests/test_watchlist_versions.py diff --git a/CHANGELOG.md b/CHANGELOG.md index c9a1cd0..49bc466 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -14,6 +14,22 @@ what has already shipped, and for nothing else. See `RELEASING.md`. ### Added +- **The scan judges CVEs against the installed version** (docs/DESIGN.md §5). A scan found the + CVEs that name a product and could not say which affect what is installed: for Notepad++ the + recorded NVD answer holds 21, of which 11 concern other versions or another product, and all + 21 were listed and ranked. `patchradar add --version --cpe ` records the + installed version and the product's CPE — typed by a person, never guessed, checked for form + (`notepad++` is refused with NVD's `notepad\+\+`) — in five new watchlist columns, migrated + into existing databases. Each CVE NVD returned is then *affected*, *not affected* or *not + evaluable*, from its own `configurations` (`affected.judge`); "cannot tell" never becomes + "not affected". The table lists the first and the last, `--all` the rest; one line gives the + 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. +- **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. + - **VulDB as a scan source** (`patchradar.collectors.vuldb`), queried only when `VULDB_API_KEY` is set; the key travels in the `X-VulDB-ApiKey` header and the software name as the search term. Pushed straight to main on 2026-10-12 without an entry here. diff --git a/README.md b/README.md index 8abaab0..d3e195c 100644 --- a/README.md +++ b/README.md @@ -9,7 +9,7 @@ > Know when your software is vulnerable — before attackers do. -PatchRadar checks the CVE feeds for the software on your watchlist and tells you which vulnerabilities affect it, most urgent first. No more manually checking NVD, MSRC, or Snyk — just add your software and run a scan. +PatchRadar checks the CVE feeds for the software on your watchlist, most urgent first. By name it finds the CVEs that **name** the software; give it the installed version and the product's CPE and it says which of them **affect** that version, which do not, and which nobody can tell yet. Scans run when you start one — `patchradar scan`, or **Scan** in the web UI. There is no scheduler and no notification inside PatchRadar: to scan on a schedule, run `patchradar scan` from cron or the Windows Task Scheduler. @@ -26,6 +26,7 @@ Scans run when you start one — `patchradar scan`, or **Scan** in the web UI. T - 🔍 **On-demand CVE scans** — `patchradar scan` queries NVD, MSRC and CISA KEV; a scan from the web UI (`POST /api/scan`) also queries the Debian Security Tracker - 🎯 **Ranked by what is actually a threat** — CISA KEV for observed exploitation, FIRST EPSS for the 30-day forecast, CVSS only to break the tie - 📋 **Personal watchlist** — add any software you want to monitor +- 🎯 **Judged against what is installed** — with `--version` and `--cpe`, each CVE is *affected*, *not affected* or *not evaluable*, from NVD's own list of affected versions, and the scan says which update closes how many - 🎨 **Beautiful web UI** — dark theme dashboard with charts and filters - 💻 **CLI first** — full command line interface for automation - 📊 **CVSS scoring** — color-coded severity (Critical / High / Medium / Low) @@ -59,6 +60,11 @@ patchradar list # Scan for CVEs (last 30 days) patchradar scan --days 30 +# Say what is installed, so the scan can judge each CVE against it +patchradar add notepad --version 8.5.0 --cpe "cpe:2.3:a:notepad-plus-plus:notepad\+\+" +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 patchradar status @@ -66,6 +72,35 @@ patchradar status patchradar remove proxmox ``` +### Which CVEs affect the installed version + +A scan by name finds the CVEs that mention the software. For Notepad++ the NVD +answer recorded for the tests holds 21, and 11 of them concern other versions or +another product: by name alone they are all in the table, ranked, as if each +were a job. + +`patchradar add --version --cpe ` records what is installed. +The CPE is typed by you, as NVD writes it — PatchRadar never guesses one from a +name — and it is checked for form: `notepad++` is refused with the escaped +`notepad\+\+` shown, because NVD would match nothing and the scan would read +as clean. The scan then judges every CVE NVD returned with NVD's own list of +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; +- **not affected** — counted, and listed only with `--all`. + +One line gives the split, one says which update closes how many of the +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). + ### Asking Debian what it is actually doing A container scanner prints one line for anything without a published fix — *no diff --git a/docs/DESIGN.md b/docs/DESIGN.md new file mode 100644 index 0000000..337d3c8 --- /dev/null +++ b/docs/DESIGN.md @@ -0,0 +1,125 @@ +# PatchRadar — design + +Written on 2026-10-17. `affected.py`, `inventory.py` and the tests around them +referred to "the versionradar design, §5" for the watchlist columns a +version-aware scan needs; that text was never committed and could not be found. +This document is the design the code refers to, reconstructed from what those +modules already commit to — the column names in `inventory.py`, the rules in +`affected.py`, the boundary in `tests/test_priority_wiring.py` — so that every +reference to §5 now points at something that exists. Sections 1 to 4 describe +what PatchRadar already does; §5 is what the version-aware scan adds. + +## 1. The question + +"Which published vulnerabilities concern the software I run, and which first?" +A scan answers it for each entry of the watchlist, on demand: there is no +scheduler and no notification inside PatchRadar. + +## 2. Sources + +NVD, MSRC and CISA KEV answer every scan; VulDB only with `VULDB_API_KEY`; the +Debian Security Tracker on a scan started from the web UI. FIRST EPSS finds +nothing on its own: it scores the CVEs the sources returned. A source that +fails makes the scan *incomplete*, and only a scan every queried source answered +may say "no CVEs found". + +## 3. Matching by name + +Each source is asked by keyword, and a record is kept when the keyword names +the software as a word (`names.keyword_pattern`); MSRC also through the products +an entry lists as Known Affected. This finds the CVEs that **name** the software. +It cannot say whether they affect the version installed: a CVE fixed in 8.9.6.4 +and a CVE that only ever affected 7.3.3 look the same. §5 is the answer to that. + +## 4. Ranking + +CISA KEV (exploited now) before FIRST EPSS (likely in 30 days, threshold 10%) +before CVSS, which only breaks ties. `priority.py` holds the order and the +reason printed beside each rank. + +## 5. The installed version + +### 5.1 Watchlist columns + +Five columns on `watchlist`, every one nullable and without a default — a +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` | +| `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) | + +An existing database gets them on the next start, as `cves` got its EPSS columns. + +### 5.2 Nothing is guessed + +The CPE is typed by a person, never derived from the name. Measured on a real +inventory, a name-to-CPE guess mapped `MX5` to a Juniper router and +`Visual C++ 2012` to `visual_c++:2008`, and nothing in the output said so +(`inventory.py`). `add` checks the form only: `cpe:2.3:`, a part, a vendor and a +product, neither `*` nor `-`. A product written with a CPE special character +unescaped — `notepad++` instead of `notepad\+\+` — is refused with the escaped +form shown, because NVD would match nothing and the scan would read as clean. + +The version must be one `affected.py` can order. One it cannot read is refused +at `add`, not discovered as "not evaluable" on every CVE of every scan. + +### 5.3 Three verdicts per CVE + +For an entry with both `installed_version` and `cpe`, each CVE the scan finds by +name is judged against the installed version with NVD's own `configurations`: + +- **affected** — a vulnerable `cpeMatch` for the vendor and product contains the + version. "Fixed in X" only where NVD states `versionEndExcluding: X`. +- **not affected** — NVD lists the affected products and versions, and this + 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`). + +"Not evaluable" never becomes "not affected": for a security tool an answer it +cannot give is not an absolution. + +### 5.4 What the scan prints + +- The table lists the CVEs that are **affected** or **not evaluable**, each with + its verdict. The **not affected** ones are counted, not listed; `scan --all` + lists them too, with their verdict. +- One line per entry says the split: found by name, how many affect the + installed version, how many could not be evaluated, how many do not affect it. +- When NVD states fixes: which version closes how many of the affected ones, and + that the rest stay open because NVD does not say where they were fixed — not + because the update falls short (`affected.summarise`). +- Always, for a versioned entry: the caveat that a count from NVD is a count of + identifiers requested, not of flaws (`affected.CAVEAT`). +- An entry without a version is scanned by name as before, and the scan says + once which entries were matched by name only. + +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. + +### 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. + +## 6. What PatchRadar does not do + +It does not read the machine it runs on: the inventory collector is a separate +script whose output is read elsewhere. It does not install, update or patch +anything, and a CVE count from any source is not a statement that the software +is safe. diff --git a/src/patchradar/affected.py b/src/patchradar/affected.py index 9054527..9cc052e 100644 --- a/src/patchradar/affected.py +++ b/src/patchradar/affected.py @@ -26,12 +26,11 @@ ever assigned an identifier. `Summary.caveat` carries that sentence so a caller cannot print "no known vulnerabilities" without it. -Not wired into the scan yet, and deliberately: "fixed in X" is a fact about a -product, and a keyword scan has neither a CPE nor an installed version to ask -about. That needs the watchlist columns of the versionradar design (§5) — -`cpe`, `installed_version`. This module is the half of the feature that can be -written and tested before the migration, and `tests/test_priority_wiring.py` -holds the boundary. +**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. """ from __future__ import annotations @@ -91,6 +90,11 @@ def _parts(text: object) -> _Version | None: return tuple(zip(numbers, letters, strict=True)) +def orderable(version: str) -> bool: + """Whether `version` is one the ranges can be compared with at all.""" + return _parts(version) is not None + + def _comparable(left: _Version, right: _Version) -> tuple[_Version, _Version] | None: if abs(len(left) - len(right)) > _SCHEME_TOLERANCE and left[0][0] != right[0][0]: return None @@ -148,6 +152,28 @@ def contains(self, version: str) -> bool: return False return True + def comparable(self, version: str) -> bool: + """Whether `contains` can give a real answer about `version`. + + `contains` says False both for a version outside the range and for one + it cannot order against the range — another scheme, or no version at + all — and only the first is "not affected". This is what tells them + apart. + """ + have = _parts(version) + if have is None: + return False + if self.unbounded: + return True + bounds = [self.exact] if self.exact is not None else [self.start, self.end] + for bound in bounds: + if bound is None: + continue + limit = _parts(bound) + if limit is None or _comparable(have, limit) is None: + return False + return True + def describe(self) -> str: if self.exact is not None: return f"only {self.exact}" @@ -321,15 +347,42 @@ def affects(payload: dict, *, version: str, vendor: str, return sorted(hits, key=lambda hit: (hit.score is not None, hit.score or 0.0), reverse=True) +def _highest_fix(fixed_versions) -> str | None: + candidates = [(parts, fixed) for fixed in fixed_versions + if fixed and (parts := _parts(fixed)) is not None] + if not candidates: + return None + return max(candidates, key=lambda pair: pair[0])[1] + + def target_version(hits: list[Hit]) -> str | None: """The highest fix any of them declares: where to go to close what can be closed with one update. None when nobody declares one — and then no number is invented.""" - candidates = [(parts, hit.fixed_version) for hit in hits - if hit.fixed_version and (parts := _parts(hit.fixed_version)) is not None] - if not candidates: - return None - return max(candidates, key=lambda pair: pair[0])[1] + return _highest_fix(hit.fixed_version for hit in hits) + + +def closed_by(fixed_versions) -> tuple[str | None, int]: + """(the update to take, how many of these it closes), from the fixed + versions of the CVEs that affect a version — None where NVD states none. + + A CVE without a stated fix is never counted as closed: NVD naming its last + affected release says nothing about which one is sound. + """ + fixes = list(fixed_versions) + target = _highest_fix(fixes) + goal = _parts(target) if target is not None else None + if goal is None: + return target, 0 + closed = 0 + for fixed in fixes: + fix = _parts(fixed) if fixed else None + if fix is None: + continue + pair = _comparable(fix, goal) + if pair is not None and pair[0] <= pair[1]: + closed += 1 + return target, closed def summarise(payload: dict, *, version: str, vendor: str, @@ -342,17 +395,7 @@ def summarise(payload: dict, *, version: str, vendor: str, "3 of 10" without that distinction invites the opposite reading. """ hits = affects(payload, version=version, vendor=vendor, cpe_product=cpe_product) - target = target_version(hits) - closed = 0 - if target is not None: - goal = _parts(target) - for hit in hits: - fix = _parts(hit.fixed_version) if hit.fixed_version else None - if fix is None or goal is None: - continue - pair = _comparable(fix, goal) - if pair is not None and pair[0] <= pair[1]: - closed += 1 + target, closed = closed_by(hit.fixed_version for hit in hits) scores = [hit.score for hit in hits if hit.score is not None] return Summary( version=version, @@ -364,3 +407,47 @@ def summarise(payload: dict, *, version: str, vendor: str, caveat=CAVEAT, hits=tuple(hits), ) + + +# ─── One CVE found by name, judged against the installed version ───────────── + +AFFECTED = "affected" +NOT_AFFECTED = "not affected" +NOT_EVALUABLE = "not evaluable" + + +@dataclass(frozen=True) +class Verdict: + status: str + reason: str + fixed_version: str | None = None + + +def judge(record: dict, *, version: str, vendor: str, cpe_product: str) -> Verdict: + """The verdict on one scan record for the installed `version` (DESIGN §5.3). + + The record carries NVD's `configurations` when NVD returned it — an empty + list when NVD has not analysed the CVE yet — and no such key when it came + from another source alone. Only a statement NVD made about this product + decides: anything short of that is "not evaluable", never "not affected". + """ + configurations = record.get("configurations") + if configurations is None: + return Verdict(NOT_EVALUABLE, "NVD's record of it was not part of this scan") + if not isinstance(configurations, list) or not configurations: + return Verdict(NOT_EVALUABLE, "NVD has not yet stated which products it affects") + + spans = ranges_in({"configurations": configurations}, vendor=vendor, cpe_product=cpe_product) + if not spans: + return Verdict(NOT_AFFECTED, f"NVD lists the products it affects, and " + f"{vendor}:{cpe_product} is not among them") + for span in spans: + if span.contains(version): + return Verdict(AFFECTED, _note_for(span), span.fixed_version) + + stated = "; ".join(span.describe() for span in spans) + if not all(span.comparable(version) for span in spans): + return Verdict(NOT_EVALUABLE, f"NVD's affected versions ({stated}) cannot be " + f"compared with {version}") + return Verdict(NOT_AFFECTED, f"NVD states the affected versions ({stated}); " + f"{version} is not among them") diff --git a/src/patchradar/cli.py b/src/patchradar/cli.py index 9698f16..f80333c 100644 --- a/src/patchradar/cli.py +++ b/src/patchradar/cli.py @@ -14,6 +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.collectors.debian import DEBIAN_RELEASE as debian_release from patchradar.collectors.debian import get_tracker_snapshot from patchradar.collectors.errors import CollectorError @@ -26,7 +27,18 @@ 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 -from patchradar.db.database import add_to_watchlist, get_cves, get_watchlist, init_db, remove_from_watchlist, save_cve +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, + save_cve, + set_installed, +) from patchradar.epss import enrich as epss_enrich from patchradar.names import SOFTWARE_NAME_MAX_LENGTH, normalise_software_name from patchradar.priority import ( @@ -275,8 +287,14 @@ def _print_law_check(law, out=None) -> None: @app.command() -def add(software: str = typer.Argument(..., help="Software to monitor")): - """Add software to your watchlist.""" +def add( + software: str = typer.Argument(..., help="Software to monitor"), + version: str = typer.Option(None, "--version", + help="Installed version, to judge each CVE against (needs --cpe)"), + cpe: str = typer.Option(None, "--cpe", + help="CPE 2.3 name as NVD writes it, e.g. cpe:2.3:a:notepad-plus-plus:notepad\\+\\+"), +): + """Add software to your watchlist, or record what is installed of it.""" # The same rule and the same canonical form as the API: " Proxmox " was # stored as is, beside the "proxmox" the API stores, and a multi-line name # reached the watchlist that every other way in refuses. @@ -287,10 +305,31 @@ def add(software: str = typer.Argument(..., help="Software to monitor")): f"spaces, '-', '_' and '.', up to {SOFTWARE_NAME_MAX_LENGTH} characters" ) raise typer.Exit(1) + # Both checked before anything is written: a version the scan cannot order + # is refused once, here, rather than "not evaluable" on every CVE of every + # scan, and a CPE NVD would match nothing with must not read as clean. + if version is not None and not orderable(version): + console.print(Text(f"❌ {version} is not a version the scan can compare: numbers " + f"separated by dots, as in 8.9.6.4 or 1.1.1t", style="red")) + raise typer.Exit(1) + parsed = None + if cpe is not None: + try: + parsed = parse_cpe(cpe) + except CpeError as exc: + console.print(Text(f"❌ {exc}", style="red")) + raise typer.Exit(1) from None added = run(add_to_watchlist(name)) + if version is not None or parsed is not None: + run(set_installed(name, version=version, cpe=parsed)) + stored = ", ".join(part for part in (f"version {version}" if version else "", + parsed.text if parsed else "") if part) + detail = Text(f" — {stored}") if stored else Text("") safe = escape(name) if added: - console.print(f"✅ [green]Added[/green] [bold]{safe}[/bold] to watchlist") + console.print(Text.from_markup(f"✅ [green]Added[/green] [bold]{safe}[/bold] to watchlist") + detail) + elif stored: + console.print(Text.from_markup(f"✏️ [green]Updated[/green] [bold]{safe}[/bold]") + detail) else: console.print(f"⚠️ [yellow]{safe}[/yellow] is already in your watchlist") @@ -307,14 +346,16 @@ def remove(software: str = typer.Argument(..., help="Software to remove")): @app.command(name="list") def list_watchlist(): """Show your current watchlist.""" - items = run(get_watchlist()) + items = run(get_watch_entries()) if not items: console.print("📭 Your watchlist is empty. Use [bold]patchradar add [/bold]") return table = Table(title="🛡️ PatchRadar Watchlist", box=box.ROUNDED) table.add_column("Software", style="cyan bold") + table.add_column("Version") + table.add_column("CPE") for item in items: - table.add_row(Text(str(item))) + table.add_row(Text(item.name), Text(item.installed_version or "—"), Text(item.cpe or "—")) console.print(table) async def _scan_target( @@ -322,6 +363,8 @@ async def _scan_target( days: int, collected: list | None = None, failed: list | None = None, + entry: WatchEntry | None = None, + show_all: bool = False, ) -> int: """Scan a single target for CVEs and return count. @@ -370,12 +413,32 @@ async def _scan_target( except CollectorError as exc: epss_failure = exc all_cves = sort_by_priority(all_cves) + # What is installed decides what the table lists (docs/DESIGN.md §5.4): + # 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) for cve in all_cves: await save_cve(cve) if collected is not None: collected.extend(all_cves) - if all_cves: - _print_cves(target, all_cves) + shown = all_cves + if judged and not show_all: + shown = [cve for cve in all_cves if cve["verdict"] != NOT_AFFECTED] + if shown: + _print_cves(target, shown, judged=judged) + if installed is not None and all_cves: + _print_judgement(target, installed[0], all_cves, show_all) for failure in failures: console.print( f"⚠️ [yellow]{escape(str(failure))}[/yellow] — " @@ -390,9 +453,31 @@ async def _scan_target( # Only an answer from every source may be reported as an all-clear. if not all_cves and not failures: console.print(f"[green]{safe_target}[/green] — no CVEs found in last {days} days") + if judged: + # Whatever the counts: a CVE exists when someone asked for an identifier. + console.print(Text(f"ℹ️ {CAVEAT[0].upper()}{CAVEAT[1:]}", style="dim")) return len(all_cves) +def _print_judgement(target: str, version: str, cves: list[dict], show_all: bool) -> None: + """The split of what was found by name, and what one update closes.""" + counts = Counter(cve["verdict"] for cve in cves) + unlisted = "" if show_all or not counts[NOT_AFFECTED] else " (not listed; --all lists them)" + console.print(Text( + f"{target} {version} — found by name: {len(cves)}; affecting {version}: " + f"{counts[AFFECTED]}; not evaluable: {counts[NOT_EVALUABLE]}; " + f"not affecting it: {counts[NOT_AFFECTED]}{unlisted}" + )) + update, closed = closed_by(cve["verdict_fixed_version"] for cve in cves if cve["verdict"] == AFFECTED) + if update is None: + return + line = f"Updating to {update} closes {closed} of {counts[AFFECTED]}." + if closed < counts[AFFECTED]: + line += (" The rest stay open because NVD does not state where they were fixed, " + "not because the update falls short.") + console.print(Text(line)) + + # Every source `scan` queries. Named here so the summary can say which ones # answered, without inferring it from the CVEs — a source that answered and had # nothing would otherwise look like one that did not answer. @@ -455,6 +540,8 @@ def scan( # min=1: a window of no days is a usage error, refused here; it used to # reach the NVD collector and end the command in a ValueError traceback. days: int = typer.Option(7, "--days", "-d", min=1, help="Days back to search"), + show_all: bool = typer.Option(False, "--all", + help="Also list the CVEs that do not affect the installed version"), ): """Scan for CVEs affecting your software.""" async def _scan(): @@ -462,12 +549,22 @@ async def _scan(): if not targets: console.print("Nothing to scan. Add software with [bold]patchradar add[/bold]") return + entries = {entry.name: entry for entry in await get_watch_entries()} cves: list[dict] = [] failures: list[CollectorError] = [] - total = sum([await _scan_target(t, days, cves, failures) for t in targets]) + total = sum([await _scan_target(t, days, cves, failures, entries.get(t.lower()), show_all) + for t in targets]) console.print(f"\n Total: [bold]{total}[/bold] CVEs found\n") # After the total, because it is the total it qualifies. _report_silent_sources(failures, list(targets)) + by_name = [t for t in targets if not (t.lower() in entries and entries[t.lower()].judgeable)] + if by_name: + # The total above counts CVEs that name these, not CVEs that affect + # what is installed; the line says which entries that is true of. + console.print(Text( + f"ℹ️ Matched by name only, version not checked: {', '.join(by_name)}. " + f"patchradar add --version --cpe lets the scan judge them." + )) if not vuldb_api_key(): console.print(f"ℹ️ {vuldb_source} was not asked: set VULDB_API_KEY to add it to the scan.") if days > msrc_max_days: @@ -488,7 +585,10 @@ async def _status(): _print_cves_table(cves) run(_status()) -def _print_cves(software: str, cves: list): +VERDICT_STYLES = {AFFECTED: "bold red", NOT_EVALUABLE: "yellow", NOT_AFFECTED: "green"} + + +def _print_cves(software: str, cves: list, judged: bool = False): table = Table( title=f"🚨 CVEs for [bold]{escape(software)}[/bold]", box=box.ROUNDED, @@ -501,6 +601,8 @@ def _print_cves(software: str, cves: list): # trust, so the reason travels with it: which catalogue, since when, and the # remediation date when CISA states one. table.add_column("Priority", max_width=34) + if judged: + table.add_column("Verdict", max_width=34) table.add_column("Description", max_width=52) for cve in cves: @@ -514,11 +616,16 @@ def _print_cves(software: str, cves: list): why = Text(rank.label, style=PRIORITY_STYLES[rank.rank]) why.append("\n" + rank.reason, style="dim") + verdict = [] + if judged: + verdict = [Text(cve["verdict"], style=VERDICT_STYLES[cve["verdict"]])] + verdict[0].append("\n" + cve["verdict_reason"], style="dim") table.add_row( Text(str(cve.get("id", ""))), score_str, Text(severity, style=severity_color), why, + *verdict, Text(_truncate(cve.get("description"), 120)), ) console.print(table) diff --git a/src/patchradar/collectors/nvd.py b/src/patchradar/collectors/nvd.py index 5beb085..7adc500 100644 --- a/src/patchradar/collectors/nvd.py +++ b/src/patchradar/collectors/nvd.py @@ -154,6 +154,7 @@ def _parse_item(item: dict, keyword: str) -> dict | None: # Left off the record entirely when the provider defined no threat metric: # the absence is the fact, and a key holding "X" would read as a verdict. maturity = _extract_exploit_maturity(cve) + configurations = cve.get("configurations") return { **({"cvss_exploit_maturity": maturity} if maturity else {}), "id": cve_id, @@ -167,6 +168,12 @@ def _parse_item(item: dict, keyword: str) -> dict | None: "url": f"https://nvd.nist.gov/vuln/detail/{cve_id}", "patch_available": _extract_patch_available(cve), "confidentiality_impact": _extract_confidentiality(cve), + # Which products and versions NVD says the CVE affects, for the scan to + # judge an installed version with (affected.judge). Always present on + # an NVD record, empty while NVD has not analysed the CVE: a record + # without the key came from another source, which is a different + # "cannot tell". Not a column — the verdict is what gets stored (§5.5). + "configurations": configurations if isinstance(configurations, list) else [], } diff --git a/src/patchradar/cpe.py b/src/patchradar/cpe.py new file mode 100644 index 0000000..e94c0b5 --- /dev/null +++ b/src/patchradar/cpe.py @@ -0,0 +1,94 @@ +"""A CPE 2.3 name as a person types it for `patchradar add --cpe` (DESIGN §5.2). + +Checked for form only. The name is never turned into a CPE by deduction — +measured on a real inventory, a guess mapped `MX5` to a Juniper router and +`Visual C++ 2012` to `visual_c++:2008` — so all this can do is refuse a string +NVD would match nothing with. The commonest is a product written the way it is +read: `notepad++`, where CPE 2.3 and NVD write `notepad\\+\\+`. Passed on as +typed, the scan raises nothing, finds nothing, and reads as clean. +""" +from __future__ import annotations + +from dataclasses import dataclass + +PREFIX = "cpe:2.3:" + +# The parts CPE 2.3 defines: application, operating system, hardware. +_PARTS = {"a", "o", "h"} + +# ANY and NA. A vendor or product of either names no product to judge. +_WILDCARDS = {"*", "-", ""} + + +class CpeError(ValueError): + """Not a CPE 2.3 name the scan could judge a version with.""" + + +@dataclass(frozen=True) +class Cpe: + text: str + part: str + vendor: str + product: str + + +def _fields(text: str) -> list[str]: + """Split on the colons that are not escaped.""" + fields, current, escaped = [], [], False + for char in text: + if escaped: + current.append(char) + escaped = False + elif char == "\\": + current.append(char) + escaped = True + elif char == ":": + fields.append("".join(current)) + current = [] + else: + current.append(char) + fields.append("".join(current)) + return fields + + +def _escaped(value: str) -> str: + """`value` with every special character escaped once, as NVD writes it. + + Letters, digits, `_`, `-` and `.` stand as they are — NVD's own + `.net_framework` — and anything already escaped is left alone. + """ + out, escaped = [], False + for char in value: + if escaped: + out.append(char) + escaped = False + elif char == "\\": + out.append(char) + escaped = True + elif char.isalnum() or char in "_-.": + out.append(char) + else: + out.append("\\" + char) + return "".join(out) + + +def parse_cpe(text: str) -> Cpe: + cleaned = text.strip() + if not cleaned.startswith(PREFIX): + raise CpeError(f"'{cleaned}' is not a CPE 2.3 name: it must start with '{PREFIX}'") + fields = _fields(cleaned) + if len(fields) < 5: + raise CpeError(f"'{cleaned}' names no product: cpe:2.3:::") + part, vendor, product = fields[2], fields[3], fields[4] + if part not in _PARTS: + raise CpeError(f"'{cleaned}': the part is '{part}', and CPE 2.3 knows only a, o and h") + for label, value in (("vendor", vendor), ("product", product)): + if value in _WILDCARDS: + shown = f"'{value}'" if value else "empty" + raise CpeError(f"'{cleaned}': the {label} is {shown}, which names no product") + if _escaped(value) != value: + raise CpeError( + f"'{cleaned}': the {label} '{value}' has characters NVD writes escaped, " + f"and as typed it would match nothing; NVD's form is '{_escaped(value)}'" + ) + return Cpe(text=cleaned, part=part, vendor=vendor, product=product) diff --git a/src/patchradar/db/database.py b/src/patchradar/db/database.py index 277747a..a02ec01 100644 --- a/src/patchradar/db/database.py +++ b/src/patchradar/db/database.py @@ -1,9 +1,12 @@ import logging +from dataclasses import dataclass from datetime import UTC, datetime from pathlib import Path import aiosqlite +from patchradar.cpe import Cpe + logger = logging.getLogger(__name__) # The production location. Kept separate from DB_PATH so the running value can @@ -49,18 +52,31 @@ def normalize_timestamp(value) -> str | None: "epss_percentile": "REAL", } +# 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 +# default would be a claim about the machine. +_ADDED_WATCHLIST_COLUMNS = { + "cpe": "TEXT", + "vendor": "TEXT", + "product_id": "TEXT", + "installed_version": "TEXT", + "channel": "TEXT", +} + -async def _add_missing_columns(db) -> None: - """Bring an existing `cves` table up to the current schema.""" - async with db.execute("PRAGMA table_info(cves)") as cursor: +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 + async with db.execute(f"PRAGMA table_info({table})") as cursor: existing = {row[1] for row in await cursor.fetchall()} - for column, kind in _ADDED_COLUMNS.items(): + for column, kind in columns.items(): if column in existing: continue - # Interpolated, because SQLite takes no parameters in DDL. The names and - # types come from the module constant above and never from input. - await db.execute(f"ALTER TABLE cves ADD COLUMN {column} {kind}") - logger.info("added column %s to the cves table", column) + # Interpolated, because SQLite takes no parameters in DDL. The table, + # names and types come from module constants and never from input. + await db.execute(f"ALTER TABLE {table} ADD COLUMN {column} {kind}") + logger.info("added column %s to the %s table", column, table) async def _normalize_stored_timestamps(db) -> None: @@ -104,7 +120,12 @@ async def init_db(): CREATE TABLE IF NOT EXISTS watchlist ( id INTEGER PRIMARY KEY AUTOINCREMENT, name TEXT UNIQUE NOT NULL, - added_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP + added_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP, + cpe TEXT, + vendor TEXT, + product_id TEXT, + installed_version TEXT, + channel TEXT ) """) await db.execute(""" @@ -131,6 +152,7 @@ async def init_db(): ) """) await _add_missing_columns(db) + await _add_missing_columns(db, "watchlist", _ADDED_WATCHLIST_COLUMNS) await _normalize_stored_timestamps(db) await _link_stored_cves(db) await db.commit() @@ -146,6 +168,61 @@ async def add_to_watchlist(name: str) -> bool: except aiosqlite.IntegrityError: return False +@dataclass(frozen=True) +class WatchEntry: + """A watchlist row with what is known of the installed product.""" + + name: str + installed_version: str | None = None + cpe: str | None = None + vendor: str | None = None + product_id: str | None = None + channel: str | None = 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 + return None + + @property + def judgeable(self) -> bool: + return self.installed() is not None + + +async def get_watch_entries() -> list[WatchEntry]: + async with aiosqlite.connect(DB_PATH) as db, db.execute( + "SELECT name, installed_version, cpe, vendor, product_id, channel FROM watchlist ORDER BY name" + ) as cursor: + return [WatchEntry(*row) for row in await cursor.fetchall()] + + +async def set_installed(name: str, *, version: str | None, cpe: Cpe | None) -> bool: + """Record what is installed for an entry already on the watchlist. + + 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] = {} + if version is not None: + fields["installed_version"] = version + if cpe is not None: + fields.update(cpe=cpe.text, vendor=cpe.vendor, product_id=cpe.product) + 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())) + await db.commit() + return True + + async def get_watchlist() -> list[str]: async with aiosqlite.connect(DB_PATH) as db, db.execute("SELECT name FROM watchlist ORDER BY name") as cursor: rows = await cursor.fetchall() diff --git a/src/patchradar/inventory.py b/src/patchradar/inventory.py index 138b83b..a8eb0a5 100644 --- a/src/patchradar/inventory.py +++ b/src/patchradar/inventory.py @@ -17,10 +17,9 @@ Counting a runtime as an unwatched product reports work that does not exist: it is patched by whatever installed it. -Nothing here writes to the database. Storing a confirmed mapping needs the -watchlist columns the versionradar design sets out (`vendor`, `product_id`, -`installed_version`, `channel`, `cpe`), and that migration is deliberately not -part of this. +Nothing here writes to the database. The watchlist columns a confirmed mapping +goes into (`vendor`, `product_id`, `installed_version`, `channel`, `cpe`) exist +since docs/DESIGN.md §5.1; writing them from an import is §5.6, step 3. """ from __future__ import annotations diff --git a/tests/test_affected_real_payload.py b/tests/test_affected_real_payload.py index 052184a..93d2101 100644 --- a/tests/test_affected_real_payload.py +++ b/tests/test_affected_real_payload.py @@ -4,7 +4,8 @@ `test_affected.py` builds each shape by hand: it proves the rules and it agrees with itself. This one runs the whole module over a real payload — the 21 CVEs NVD returns for `cpe:2.3:a:notepad-plus-plus:notepad\\+\\+` — and pins the three -answers the versionradar design measured on 2026-09-28. +answers the prototype of the version-aware scan measured on 2026-09-28 +(docs/DESIGN.md §5). It earned its place immediately. The first run of this module against this file returned **zero hits for all three versions**, because the tests spelled the diff --git a/tests/test_affected_verdict.py b/tests/test_affected_verdict.py new file mode 100644 index 0000000..f01c9e4 --- /dev/null +++ b/tests/test_affected_verdict.py @@ -0,0 +1,203 @@ +""" +PatchRadar — every CVE found by name gets one of three verdicts (DESIGN §5.3) + +`affects()` answers "which of these affect the version", and is silent about the +rest; a scan needs to say what it is silent about. So each record is judged: + +- affected: a vulnerable `cpeMatch` for the vendor and product contains it; +- not affected: NVD lists the affected products and versions, and this version + of this product is not among them; +- not evaluable: anything that cannot be decided — NVD has not yet said which + products the CVE affects, NVD's record was not part of the scan, or NVD's + versions belong to another scheme than the installed one. + +"Not evaluable" never becomes "not affected": an answer a security tool cannot +give is not an absolution. + +The payload is the recorded NVD answer for Notepad++ that +`test_affected_real_payload.py` already pins; `judge` reads it as the scan does, +one record at a time, through `nvd._parse_item`. +""" +from __future__ import annotations + +import copy +import json +import pathlib + +import pytest + +from patchradar.affected import ( + AFFECTED, + NOT_AFFECTED, + NOT_EVALUABLE, + closed_by, + judge, + summarise, +) +from patchradar.collectors.nvd import _parse_item + +FIXTURE = pathlib.Path(__file__).parent / "fixtures" / "nvd_notepadpp_configurations.json" +VENDOR = "notepad-plus-plus" +PRODUCT = "notepad\\+\\+" + + +@pytest.fixture(scope="module") +def payload() -> dict: + return json.loads(FIXTURE.read_text(encoding="utf-8")) + + +@pytest.fixture(scope="module") +def records(payload) -> list[dict]: + return [_parse_item(item, "notepad") for item in payload["vulnerabilities"]] + + +def verdicts(records, version): + return {r["id"]: judge(r, version=version, vendor=VENDOR, cpe_product=PRODUCT) for r in records} + + +# ─── what NVD's record carries ─────────────────────────────────────────────── + +def test_an_nvd_record_keeps_its_configurations(records): + assert all(isinstance(r["configurations"], list) for r in records) + assert any(r["configurations"] for r in records) + + +def test_an_nvd_record_without_configurations_says_so_with_an_empty_list(): + """Present and empty: NVD's record was read, and NVD has not analysed it + yet. Absent means something else — the record did not come from NVD.""" + record = _parse_item({"cve": {"id": "CVE-2026-99999", "published": "2026-10-16T10:00:00.000"}}, "notepad") + + assert record["configurations"] == [] + + +# ─── the three verdicts on the real payload ────────────────────────────────── + +def test_the_affected_ones_are_exactly_those_affects_finds(payload, records): + """Two readings of one payload must agree: `summarise` through `affects`, + and `judge` record by record.""" + expected = {hit.cve for hit in summarise(payload, version="8.5.0", vendor=VENDOR, + cpe_product=PRODUCT).hits} + + found = {cve for cve, v in verdicts(records, "8.5.0").items() if v.status == AFFECTED} + + assert found == expected + assert len(found) == 10 + + +def test_every_other_record_of_the_payload_is_not_affected(records): + """Every CVE here has its products analysed and every version compares, + so nothing is left undecided.""" + statuses = [v.status for v in verdicts(records, "8.5.0").values()] + + assert statuses.count(NOT_AFFECTED) == 11 + assert NOT_EVALUABLE not in statuses + + +def test_a_newer_version_is_affected_by_none(records): + assert all(v.status == NOT_AFFECTED for v in verdicts(records, "8.9.8").values()) + + +def test_a_product_named_only_as_context_is_not_affected(records): + """CVE-2017-8803: the Notepad++ node is `vulnerable: false`, the context of + a plugin's flaw, and 7.3.3 sits inside the range it names.""" + verdict = verdicts(records, "7.3.3")["CVE-2017-8803"] + + assert verdict.status == NOT_AFFECTED + assert "not among them" in verdict.reason + + +def test_an_affected_verdict_says_where_nvd_states_the_fix(records): + """8.9.5: five, fixed in 8.9.6 (CVE-2026-46710), 8.9.6.1 (three) and + 8.9.6.4 (CVE-2026-52885) — each as its `versionEndExcluding` says.""" + found = verdicts(records, "8.9.5") + fixed = {cve: v.fixed_version for cve, v in found.items() if v.status == AFFECTED} + + assert fixed == { + "CVE-2026-46710": "8.9.6", + "CVE-2026-48770": "8.9.6.1", + "CVE-2026-48778": "8.9.6.1", + "CVE-2026-48800": "8.9.6.1", + "CVE-2026-52885": "8.9.6.4", + } + assert all(found[cve].fixed_version in found[cve].reason for cve in fixed) + + +def test_a_last_affected_version_is_not_turned_into_a_fix(records): + """`versionEndIncluding: X` names the last affected release; the next one + is not a number any source stated.""" + open_ones = [v for v in verdicts(records, "8.5.0").values() + if v.status == AFFECTED and v.fixed_version is None] + + assert len(open_ones) == 4 + assert all("does not state" in v.reason for v in open_ones) + + +# ─── what cannot be decided ────────────────────────────────────────────────── + +def test_a_record_from_another_source_is_not_evaluable(): + """MSRC, KEV and VulDB carry no `cpeMatch`; without NVD's record of the same + CVE there is nothing to judge the version with.""" + verdict = judge({"id": "CVE-2026-12345", "source": "CISA KEV"}, + version="8.5.0", vendor=VENDOR, cpe_product=PRODUCT) + + assert verdict.status == NOT_EVALUABLE + assert "not part of this scan" in verdict.reason + + +def test_a_record_nvd_has_not_analysed_is_not_evaluable(): + verdict = judge({"id": "CVE-2026-99999", "source": "NVD", "configurations": []}, + version="8.5.0", vendor=VENDOR, cpe_product=PRODUCT) + + assert verdict.status == NOT_EVALUABLE + assert "not yet" in verdict.reason + + +def test_versions_of_another_scheme_are_not_evaluable_rather_than_not_affected(records): + """Derived from CVE-2023-40031 of the payload, its range moved to an Office + build number: `2.07` against `16.0.14334.20918` is a comparison with no + meaning, and "not affected" would be a verdict on it.""" + record = copy.deepcopy(next(r for r in records if r["id"] == "CVE-2023-40031")) + for configuration in record["configurations"]: + for node in configuration["nodes"]: + for entry in node["cpeMatch"]: + entry.pop("versionEndIncluding", None) + entry.pop("versionStartIncluding", None) + entry["versionEndExcluding"] = "16.0.14334.20918" + + verdict = judge(record, version="2.07", vendor=VENDOR, cpe_product=PRODUCT) + + assert verdict.status == NOT_EVALUABLE + assert "cannot be compared" in verdict.reason + + +def test_a_version_that_is_not_a_version_is_not_evaluable_wherever_the_product_is_listed(records): + """"latest" orders against nothing. Where NVD lists Notepad++ as affected, + the answer is "cannot tell", never "not affected" — and never "affected".""" + at_850 = verdicts(records, "8.5.0") + latest = verdicts(records, "latest") + + assert AFFECTED not in {v.status for v in latest.values()} + assert all(latest[cve].status == NOT_EVALUABLE + for cve, v in at_850.items() if v.status == AFFECTED) + + +# ─── what one update closes ────────────────────────────────────────────────── + +def test_closed_by_agrees_with_summarise(payload, records): + summary = summarise(payload, version="8.5.0", vendor=VENDOR, cpe_product=PRODUCT) + fixes = [v.fixed_version for v in verdicts(records, "8.5.0").values() if v.status == AFFECTED] + + assert closed_by(fixes) == (summary.target, summary.closed_by_target) == ("8.9.6.4", 6) + + +def test_closed_by_names_no_target_when_nobody_states_a_fix(): + assert closed_by([None, None]) == (None, 0) + + +def test_every_version_is_comparable_with_every_version(): + """A `*` in NVD's criteria: unbounded contains any version, so `judge` never + asks — but `comparable` must not be the one to say otherwise.""" + from patchradar.affected import Range + + assert Range(unbounded=True).comparable("8.5.0") + assert not Range(unbounded=True).comparable("latest") diff --git a/tests/test_cli_escaping.py b/tests/test_cli_escaping.py index 1426650..013bd21 100644 --- a/tests/test_cli_escaping.py +++ b/tests/test_cli_escaping.py @@ -17,11 +17,13 @@ import asyncio from io import StringIO +import aiosqlite import pytest import typer from rich.console import Console import patchradar.cli as cli +from patchradar.db import database from patchradar.db.database import add_to_watchlist, get_watchlist @@ -198,7 +200,9 @@ def test_cli_add_escapes_on_duplicate(): # Markup cannot get this far any more (see above): what the duplicate # message prints is the canonical name, and it prints it verbatim. asyncio.run(add_to_watchlist("windows 10")) - out = render(cli.add, " Windows 10 ") + # Called as a function, `add` gets its options' Typer defaults unless they + # are passed; None is what the command line gives when they are left out. + out = render(cli.add, " Windows 10 ", version=None, cpe=None) assert "windows 10 is already in your watchlist" in out assert asyncio.run(get_watchlist()) == ["windows 10"] @@ -221,11 +225,17 @@ async def fake_remove(name): assert MARKUP_NAME in out -def test_cli_list_escapes_watchlist_entries(monkeypatch): - async def fake_watchlist(): - return [MARKUP_NAME, "nginx"] +def test_cli_list_escapes_watchlist_entries(): + """Every way in refuses brackets now; a row written before that rule, by + an older version, is still in somebody's database. It is put there the way + it got there — straight into the table — and printed verbatim.""" + async def old_rows(): + async with aiosqlite.connect(database.DB_PATH) as db: + await db.executemany("INSERT INTO watchlist (name) VALUES (?)", + [(MARKUP_NAME,), ("nginx",)]) + await db.commit() - monkeypatch.setattr(cli, "get_watchlist", fake_watchlist) + asyncio.run(old_rows()) out = render(cli.list_watchlist) assert MARKUP_NAME in out diff --git a/tests/test_cpe.py b/tests/test_cpe.py new file mode 100644 index 0000000..c7466c4 --- /dev/null +++ b/tests/test_cpe.py @@ -0,0 +1,81 @@ +""" +PatchRadar — a CPE is typed by a person, and checked for form only (DESIGN §5.2) + +The name is never turned into a CPE by deduction: on a real inventory a guess +mapped `MX5` to a Juniper router. What `add --cpe` can do is refuse a string NVD +would match nothing with, and the commonest one is a product written the way a +person reads it: `notepad++`, where CPE 2.3 and NVD write `notepad\\+\\+`. Passed +as is, the scan raises nothing, finds nothing, and reads as clean. +""" +from __future__ import annotations + +import pytest + +from patchradar.cpe import CpeError, parse_cpe + +NOTEPADPP = "cpe:2.3:a:notepad-plus-plus:notepad\\+\\+" + + +def test_the_vendor_and_product_are_read_as_nvd_writes_them(): + cpe = parse_cpe(NOTEPADPP) + + assert cpe.part == "a" + assert cpe.vendor == "notepad-plus-plus" + assert cpe.product == "notepad\\+\\+" + assert cpe.text == NOTEPADPP + + +def test_a_full_cpe_with_version_and_wildcards_keeps_its_product(): + cpe = parse_cpe("cpe:2.3:a:notepad-plus-plus:notepad\\+\\+:8.5.0:*:*:*:*:*:*:*") + + assert (cpe.vendor, cpe.product) == ("notepad-plus-plus", "notepad\\+\\+") + + +def test_dots_dashes_and_underscores_need_no_escape(): + """NVD's own spelling: `.net_framework`, `visual_studio_2022`.""" + cpe = parse_cpe("cpe:2.3:a:microsoft:.net_framework") + + assert cpe.product == ".net_framework" + + +def test_surrounding_spaces_are_not_part_of_it(): + assert parse_cpe(f" {NOTEPADPP} \n").text == NOTEPADPP + + +@pytest.mark.parametrize("text", [ + "notepad-plus-plus:notepad\\+\\+", # no prefix + "cpe:/a:notepad-plus-plus:notepad%2b%2b", # CPE 2.2 URI binding + "cpe:2.3:a:notepad-plus-plus", # no product + "cpe:2.3:x:notepad-plus-plus:notepad", # not a, o or h +]) +def test_what_is_not_a_cpe_2_3_name_is_refused(text): + with pytest.raises(CpeError): + parse_cpe(text) + + +@pytest.mark.parametrize("text", [ + "cpe:2.3:a:*:notepad", + "cpe:2.3:a:notepad-plus-plus:*", + "cpe:2.3:a:-:notepad", + "cpe:2.3:a:notepad-plus-plus:", +]) +def test_a_wildcard_or_empty_vendor_or_product_names_no_product(text): + """`*` is ANY and `-` is NA: a watchlist entry for "any product of any + vendor" is not something a verdict can be given about.""" + with pytest.raises(CpeError): + parse_cpe(text) + + +def test_an_unescaped_special_character_is_refused_with_the_escaped_form(): + with pytest.raises(CpeError) as raised: + parse_cpe("cpe:2.3:a:notepad-plus-plus:notepad++") + + assert "notepad\\+\\+" in str(raised.value) + + +def test_a_character_already_escaped_is_not_escaped_twice(): + with pytest.raises(CpeError) as raised: + parse_cpe("cpe:2.3:a:notepad-plus-plus:notepad\\++") + + assert "notepad\\+\\+" in str(raised.value) + assert "\\\\+" not in str(raised.value) diff --git a/tests/test_priority_wiring.py b/tests/test_priority_wiring.py index 6b0f5ee..1c7c46a 100644 --- a/tests/test_priority_wiring.py +++ b/tests/test_priority_wiring.py @@ -10,7 +10,7 @@ `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 -DESIGN §5 lands, only the in-memory path can rank — and the day someone adds the +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 EPSS pair *is* stored, which narrows that gap without closing it: a ranking diff --git a/tests/test_scan_versioned.py b/tests/test_scan_versioned.py new file mode 100644 index 0000000..5c0384d --- /dev/null +++ b/tests/test_scan_versioned.py @@ -0,0 +1,188 @@ +""" +PatchRadar — `scan` judges what it finds against the installed version (DESIGN §5.4) + +Until now a scan found the CVEs that **name** the software and could not say +which of them affect what is installed: for Notepad++ 8.5.0 the recorded NVD +answer holds 21 CVEs, and 11 of them concern other versions or another product. +The table printed all 21, ranked, as if each were a job. + +For an entry added with `--version` and `--cpe` the scan now judges each CVE: +the table lists the ones that affect the version or could not be evaluated, the +others are counted and listed only with `--all`, one line gives the split, one +says what an update closes, and the caveat that a count from NVD is a count of +identifiers is always there. An entry without them is scanned by name as before, +and the scan says so. + +NVD answers from the recording `nvd_notepadpp_configurations.json` (21 CVEs, +trimmed to the fields under test); MSRC has no document for the months asked +(404); the KEV catalogue is the empty one conftest.py caches. +""" +from __future__ import annotations + +import copy +import json +import pathlib + +import httpx +import pytest +import respx +from typer.testing import CliRunner + +from patchradar import cli +from patchradar.affected import summarise +from patchradar.collectors import msrc, nvd + +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\\+\\+" + +AFFECTING_850 = {hit.cve for hit in summarise(PAYLOAD, version="8.5.0", vendor="notepad-plus-plus", + cpe_product="notepad\\+\\+").hits} +NOT_AFFECTING_850 = {item["cve"]["id"] for item in PAYLOAD["vulnerabilities"]} - AFFECTING_850 + + +def _nvd_answers(payload): + router = respx.mock(assert_all_called=False) + 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)) + return router + + +@pytest.fixture +def nvd_answers_notepadpp(): + with _nvd_answers(PAYLOAD): + yield + + +def _scan(*args, version="8.5.0"): + runner = CliRunner() + added = ["add", "notepad", "--cpe", NOTEPADPP] + (["--version", version] if version else []) + assert runner.invoke(cli.app, added).exit_code == 0 + result = runner.invoke(cli.app, ["scan", "notepad", "--days", "7", *args], env={"COLUMNS": "200"}) + assert result.exit_code == 0, result.output + return result.output + + +def _words(output: str) -> str: + return " ".join(output.split()) + + +def test_the_payload_splits_ten_and_eleven(): + assert (len(AFFECTING_850), len(NOT_AFFECTING_850)) == (10, 11) + + +def test_the_split_is_said_in_one_line(nvd_answers_notepadpp): + out = _words(_scan()) + + assert "found by name: 21; affecting 8.5.0: 10; not evaluable: 0; not affecting it: 11" in out + + +def test_the_table_lists_what_affects_the_version_and_not_the_rest(nvd_answers_notepadpp): + out = _scan() + + assert all(cve in out for cve in AFFECTING_850) + assert not any(cve in out for cve in NOT_AFFECTING_850) + + +def test_each_listed_cve_carries_its_verdict(nvd_answers_notepadpp): + out = _words(_scan()) + + assert out.count("affected") >= 10 + assert "NVD states the fix is in 8.9.6.4" in out + + +def test_all_lists_the_ones_that_do_not_affect_it_with_their_verdict(nvd_answers_notepadpp): + out = _scan("--all") + + assert all(cve in out for cve in AFFECTING_850 | NOT_AFFECTING_850) + assert "not affected" in out + + +def test_without_all_the_scan_says_how_to_see_the_rest(nvd_answers_notepadpp): + assert "--all" in _scan() + + +def test_what_one_update_closes_and_why_the_rest_stay_open(nvd_answers_notepadpp): + out = _words(_scan()) + + assert "Updating to 8.9.6.4 closes 6 of 10" in out + assert "NVD does not state where they were fixed" in out + + +def test_the_caveat_is_always_there(nvd_answers_notepadpp): + assert "Counted in NVD only" in _words(_scan()) + + +def test_a_version_none_of_them_affects_lists_none_and_keeps_the_caveat(nvd_answers_notepadpp): + out = _words(_scan(version="8.9.8")) + + assert "found by name: 21; affecting 8.9.8: 0; not evaluable: 0; not affecting it: 21" in out + assert not any(cve in out for cve in AFFECTING_850 | NOT_AFFECTING_850) + assert "Counted in NVD only" in out + assert "no CVEs found" not in out + + +def test_the_total_keeps_counting_what_was_found_by_name(nvd_answers_notepadpp): + assert "Total: 21 CVEs found" in _words(_scan()) + + +def test_a_cve_nvd_has_not_analysed_is_listed_as_not_evaluable(): + """Derived: the payload above with the `configurations` of CVE-2026-52885 + removed, the state of a CVE in NVD's backlog.""" + payload = copy.deepcopy(PAYLOAD) + for item in payload["vulnerabilities"]: + if item["cve"]["id"] == "CVE-2026-52885": + del item["cve"]["configurations"] + + with _nvd_answers(payload): + out = _words(_scan(version="8.9.8")) + + assert "CVE-2026-52885" in out + assert "not evaluable" in out + assert "affecting 8.9.8: 0; not evaluable: 1; not affecting it: 20" in out + + +# ─── entries without a version ─────────────────────────────────────────────── + +def test_an_entry_without_a_version_is_scanned_by_name_and_said_to_be(nvd_answers_notepadpp): + out = _words(_scan(version=None)) + + assert all(cve in out for cve in AFFECTING_850 | NOT_AFFECTING_850) + assert "Matched by name only, version not checked: notepad" in out + assert "found by name:" not in out + + +def test_a_versioned_entry_is_not_among_those_matched_by_name_only(nvd_answers_notepadpp): + assert "Matched by name only" not in _scan() + + +def test_an_update_that_closes_all_says_nothing_about_a_rest(nvd_answers_notepadpp): + """8.9.5: five affect it, every fix stated, the highest is 8.9.6.4.""" + out = _words(_scan(version="8.9.5")) + + assert "Updating to 8.9.6.4 closes 5 of 5." in out + assert "The rest stay open" not in out + + +def test_when_nothing_is_left_out_the_scan_does_not_point_at_all(): + """Derived: the payload above reduced to the ten CVEs that affect 8.5.0.""" + payload = {**PAYLOAD, "vulnerabilities": [item for item in PAYLOAD["vulnerabilities"] + if item["cve"]["id"] in AFFECTING_850]} + + with _nvd_answers(payload): + out = _words(_scan()) + + assert "not affecting it: 0" in out + assert "--all" not in out + + +def test_the_caveat_stays_when_nvd_finds_nothing(): + """NVD's recorded answer for a keyword with no CVEs: no CVE in NVD is the + case the caveat exists for.""" + empty = json.loads((FIXTURE.parent / "nvd_no_results.json").read_text(encoding="utf-8")) + + with _nvd_answers(empty): + out = _words(_scan()) + + assert "no CVEs found" in out + assert "Counted in NVD only" in out diff --git a/tests/test_watchlist_versions.py b/tests/test_watchlist_versions.py new file mode 100644 index 0000000..8aa3885 --- /dev/null +++ b/tests/test_watchlist_versions.py @@ -0,0 +1,225 @@ +""" +PatchRadar — the watchlist learns what is installed (DESIGN §5.1, §5.2) + +Five columns, nullable and without a default: `cpe`, `vendor`, `product_id`, +`installed_version`, `channel` — the names `inventory.py` has been waiting for. +A missing version is an absence; a default would be a claim. + +The migration is half the feature, as it was for EPSS: `CREATE TABLE IF NOT +EXISTS` does nothing to a database that already has a watchlist, so the old +database below is written out by hand, in the shape it shipped in. +""" +from __future__ import annotations + +import asyncio + +import aiosqlite +import pytest +from typer.testing import CliRunner + +from patchradar import cli +from patchradar.cpe import parse_cpe +from patchradar.db import database + +NOTEPADPP = "cpe:2.3:a:notepad-plus-plus:notepad\\+\\+" +FIVE = {"cpe", "vendor", "product_id", "installed_version", "channel"} + +# The watchlist as it shipped until 2026-10-17, written out rather than +# generated: a migration test whose "old" database comes from today's code +# tests nothing. +OLD_WATCHLIST = """ + CREATE TABLE watchlist ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + name TEXT UNIQUE NOT NULL, + added_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP + ) +""" + + +@pytest.fixture +def old_database(tmp_path, monkeypatch): + path = tmp_path / "old.db" + + async def _build(): + async with aiosqlite.connect(path) as db: + await db.execute(OLD_WATCHLIST) + await db.execute("INSERT INTO watchlist (name) VALUES ('notepad')") + await db.commit() + + asyncio.run(_build()) + monkeypatch.setattr(database, "DB_PATH", path) + return path + + +async def _watchlist_columns(path) -> set[str]: + async with aiosqlite.connect(path) as db, db.execute("PRAGMA table_info(watchlist)") as cursor: + return {row[1] for row in await cursor.fetchall()} + + +def _out(result) -> str: + """Rich wraps long lines to the runner's width; the words are what count.""" + return " ".join(result.output.split()) + + +# ─── the migration ─────────────────────────────────────────────────────────── + +@pytest.mark.asyncio +async def test_an_existing_watchlist_gains_the_five_columns(old_database): + assert not FIVE & await _watchlist_columns(old_database) + + await database.init_db() + + assert await _watchlist_columns(old_database) >= FIVE + + +@pytest.mark.asyncio +async def test_an_entry_from_before_keeps_its_name_and_knows_no_version(old_database): + await database.init_db() + + [entry] = await database.get_watch_entries() + assert entry.name == "notepad" + assert entry.installed_version is None + assert entry.cpe is None + assert not entry.judgeable + + +@pytest.mark.asyncio +async def test_migrating_twice_is_not_an_error(old_database): + await database.init_db() + await database.init_db() + + assert await _watchlist_columns(old_database) >= FIVE + + +# ─── storing what is installed ─────────────────────────────────────────────── + +@pytest.mark.asyncio +async def test_version_and_cpe_are_stored_with_vendor_and_product(): + await database.add_to_watchlist("notepad") + + assert await database.set_installed("notepad", version="8.5.0", cpe=parse_cpe(NOTEPADPP)) + + [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.channel is None + assert entry.judgeable + + +@pytest.mark.asyncio +async def test_a_version_without_a_cpe_is_kept_and_cannot_be_judged(): + """The version alone says nothing NVD can be asked about: the vendor and + product are what `cpeMatch` is read by.""" + await database.add_to_watchlist("notepad") + await database.set_installed("notepad", version="8.5.0", cpe=None) + + [entry] = await database.get_watch_entries() + assert entry.installed_version == "8.5.0" + assert not entry.judgeable + + +@pytest.mark.asyncio +async def test_setting_one_field_leaves_the_other_alone(): + await database.add_to_watchlist("notepad") + await database.set_installed("notepad", version="8.5.0", cpe=parse_cpe(NOTEPADPP)) + + await database.set_installed("notepad", version="8.9.6.4", cpe=None) + + [entry] = await database.get_watch_entries() + assert entry.installed_version == "8.9.6.4" + assert entry.cpe == NOTEPADPP + + +@pytest.mark.asyncio +async def test_a_name_not_on_the_watchlist_is_not_added_by_the_back_door(): + assert not await database.set_installed("notepad", version="8.5.0", cpe=None) + assert await database.get_watchlist() == [] + + +@pytest.mark.asyncio +async def test_get_watchlist_still_answers_names(): + """The API and the scan read names from it; their shape does not change.""" + await database.add_to_watchlist("notepad") + await database.set_installed("notepad", version="8.5.0", cpe=parse_cpe(NOTEPADPP)) + + assert await database.get_watchlist() == ["notepad"] + + +# ─── patchradar add / list ─────────────────────────────────────────────────── + +def test_add_with_version_and_cpe_says_what_it_stored(): + result = CliRunner().invoke(cli.app, ["add", "notepad", "--version", "8.5.0", "--cpe", NOTEPADPP]) + + assert result.exit_code == 0, result.output + assert "Added" in result.output + assert "8.5.0" in _out(result) + [entry] = asyncio.run(database.get_watch_entries()) + assert (entry.installed_version, entry.cpe) == ("8.5.0", NOTEPADPP) + + +def test_add_on_an_existing_entry_updates_what_is_installed(): + runner = CliRunner() + runner.invoke(cli.app, ["add", "notepad"]) + + result = runner.invoke(cli.app, ["add", "notepad", "--version", "8.9.6.4", "--cpe", NOTEPADPP]) + + assert result.exit_code == 0, result.output + assert "Updated" in result.output + assert "already in your watchlist" not in result.output + [entry] = asyncio.run(database.get_watch_entries()) + assert entry.installed_version == "8.9.6.4" + + +def test_add_without_options_on_an_existing_entry_is_unchanged(): + runner = CliRunner() + runner.invoke(cli.app, ["add", "notepad", "--version", "8.5.0"]) + + result = runner.invoke(cli.app, ["add", "notepad"]) + + assert "already in your watchlist" in result.output + [entry] = asyncio.run(database.get_watch_entries()) + assert entry.installed_version == "8.5.0" + + +def test_a_version_the_scan_cannot_order_is_refused_at_add(): + """Refused once, here, rather than "not evaluable" on every CVE of every scan.""" + result = CliRunner().invoke(cli.app, ["add", "notepad", "--version", "8.x-beta", "--cpe", NOTEPADPP]) + + assert result.exit_code == 1 + assert "8.x-beta" in result.output + assert asyncio.run(database.get_watchlist()) == [] + + +def test_a_cpe_with_an_unescaped_plus_is_refused_with_the_form_nvd_wants(): + result = CliRunner().invoke(cli.app, ["add", "notepad", "--version", "8.5.0", + "--cpe", "cpe:2.3:a:notepad-plus-plus:notepad++"]) + + assert result.exit_code == 1 + assert "notepad\\+\\+" in result.output + assert asyncio.run(database.get_watchlist()) == [] + + +def test_list_shows_the_version_and_the_cpe(): + runner = CliRunner() + runner.invoke(cli.app, ["add", "notepad", "--version", "8.5.0", "--cpe", NOTEPADPP]) + runner.invoke(cli.app, ["add", "nginx"]) + + result = runner.invoke(cli.app, ["list"]) + + out = _out(result) + assert "8.5.0" in out + assert "notepad-plus-plus" in out + assert "nginx" in out + + +@pytest.mark.asyncio +async def test_setting_nothing_changes_nothing(): + await database.add_to_watchlist("notepad") + await database.set_installed("notepad", version="8.5.0", cpe=parse_cpe(NOTEPADPP)) + + assert await database.set_installed("notepad", version=None, cpe=None) + + [entry] = await database.get_watch_entries() + assert (entry.installed_version, entry.cpe) == ("8.5.0", NOTEPADPP)