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)