From 3ab79acacbe969a5ef68231fd8cf0b7f3fd77845 Mon Sep 17 00:00:00 2001 From: maksimtech <268127343+maksimtech@users.noreply.github.com> Date: Sat, 17 Oct 2026 18:40:16 +0200 Subject: [PATCH] fix(watchlist): vendor and product_id are the vendor source's, as inventory.py meant inventory.py named the watchlist columns before they existed: vendor, product_id, installed_version, channel, cpe. There a vendor source is a feed that answers by version (Chrome's, Edge's, MSRC, Mozilla's) and product_id is the product it needs; NVD by CPE is the other route. #58 filled vendor and product_id with the CPE's fields instead - a misreading of the design it rebuilt - which would have had the inventory import write one over the other. No release carried it. - The CPE's vendor and product are read from `cpe` (WatchEntry.cpe_vendor, cpe_product, installed()), not stored. - set_source records the vendor source, product and channel a person confirmed, for the import to call; it never adds a name. - DESIGN.md section 5.1 says which column is which, and why. Tests first, seen failing; the mutants that swap source and product or drop the version from installed() are killed. Co-Authored-By: Claude Opus 5.5 (1M context) --- docs/DESIGN.md | 14 ++++-- src/patchradar/db/database.py | 44 +++++++++++++---- tests/test_api_versioned.py | 3 +- tests/test_vendor_source_columns.py | 76 +++++++++++++++++++++++++++++ tests/test_watchlist_versions.py | 6 +-- 5 files changed, 126 insertions(+), 17 deletions(-) create mode 100644 tests/test_vendor_source_columns.py diff --git a/docs/DESIGN.md b/docs/DESIGN.md index a8d17a5..ec2e100 100644 --- a/docs/DESIGN.md +++ b/docs/DESIGN.md @@ -47,13 +47,21 @@ 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` | +| `vendor` | the vendor source that answers by version — `inventory.py`'s *source*: `chrome`, `edge`, `msrc`, `mozilla` | the inventory import, once confirmed (§5.6) | +| `product_id` | the product that source needs: `chrome`, `Edge` | the inventory import, once confirmed | | `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) | +| `channel` | the release channel the source answers for (stable, ESR, LTS) | the inventory import, once confirmed | An existing database gets them on the next start, as `cves` got its EPSS columns. +Two routes to an answer by version, kept apart: the CPE, through NVD's +configurations, which the scan judges with today; and a vendor source, which +`inventory.py` proposes and a person confirms. The CPE's own vendor and product +(`notepad-plus-plus`, `notepad\+\+`) are read from `cpe` when the scan needs +them, not stored. Until 2026-10-17 they were stored in `vendor` and `product_id`, +by a misreading of this table that `inventory.py` contradicted; no release +carried that. + ### 5.2 Nothing is guessed The CPE is typed by a person, never derived from the name. Measured on a real diff --git a/src/patchradar/db/database.py b/src/patchradar/db/database.py index 5928757..b727544 100644 --- a/src/patchradar/db/database.py +++ b/src/patchradar/db/database.py @@ -5,7 +5,7 @@ import aiosqlite -from patchradar.cpe import Cpe +from patchradar.cpe import Cpe, parse_cpe from patchradar.priority import as_flag, sql_order logger = logging.getLogger(__name__) @@ -209,7 +209,13 @@ async def add_to_watchlist(name: str) -> bool: @dataclass(frozen=True) class WatchEntry: - """A watchlist row with what is known of the installed product.""" + """A watchlist row with what is known of the installed product. + + `vendor`, `product_id` and `channel` are the vendor source's — the feed + that answers by version and the product it needs, as `inventory.py` names + them (docs/DESIGN.md §5.1). The CPE's vendor and product are another thing, + read from `cpe`: NVD by CPE is the other route. + """ name: str installed_version: str | None = None @@ -218,12 +224,21 @@ class WatchEntry: product_id: str | None = None channel: str | None = None + @property + def cpe_vendor(self) -> str | None: + return parse_cpe(self.cpe).vendor if self.cpe else None + + @property + def cpe_product(self) -> str | None: + return parse_cpe(self.cpe).product if self.cpe else 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 + installed — the version to compare, and the CPE's vendor and product + NVD's `cpeMatch` is read by — and None when any of them is missing.""" + if self.installed_version and self.cpe: + cpe = parse_cpe(self.cpe) + return self.installed_version, cpe.vendor, cpe.product return None @property @@ -244,17 +259,28 @@ async def set_installed(name: str, *, version: str | None, cpe: Cpe | None) -> b 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] = {} + fields: dict[str, str | None] = {} 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) + fields["cpe"] = cpe.text + return await _update_entry(name, fields) + + +async def set_source(name: str, *, vendor: str, product_id: str, channel: str | None) -> bool: + """Record the vendor source that answers by version for an entry already on + the watchlist — `inventory.py`'s source, product and channel, once a person + has confirmed them. False when the name is not on the watchlist.""" + return await _update_entry(name, {"vendor": vendor, "product_id": product_id, "channel": channel}) + + +async def _update_entry(name: str, fields: dict[str, str | None]) -> bool: 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. + # Column names from literal keys in the callers, 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())) diff --git a/tests/test_api_versioned.py b/tests/test_api_versioned.py index f331bb3..dcbf121 100644 --- a/tests/test_api_versioned.py +++ b/tests/test_api_versioned.py @@ -46,8 +46,7 @@ async def test_the_version_and_cpe_are_recorded_over_http(client): assert response.status_code == 200, response.text assert response.json()["installed_version"] == "8.5.0" [entry] = await database.get_watch_entries() - assert (entry.installed_version, entry.vendor, entry.product_id) == ("8.5.0", "notepad-plus-plus", - "notepad\\+\\+") + assert entry.installed() == ("8.5.0", "notepad-plus-plus", "notepad\\+\\+") async def test_a_cpe_nvd_would_match_nothing_with_is_refused_with_its_form(client): diff --git a/tests/test_vendor_source_columns.py b/tests/test_vendor_source_columns.py new file mode 100644 index 0000000..7986280 --- /dev/null +++ b/tests/test_vendor_source_columns.py @@ -0,0 +1,76 @@ +""" +PatchRadar — `vendor` and `product_id` belong to the vendor source (docs/DESIGN.md §5.1) + +`inventory.py` named the watchlist columns before they existed: `vendor`, +`product_id`, `installed_version`, `channel`, `cpe`. In it, a vendor source is a +feed that answers by version — Chrome's, Edge's, MSRC, Mozilla's — and +`product_id` is the product that feed needs ("chrome", "Edge"); the CPE is the +separate route, NVD by CPE. #58 filled `vendor` and `product_id` with the CPE's +fields instead, which would have had `import` write a vendor source over the +CPE's vendor, or the other way round. + +The CPE's vendor and product are now read from `cpe` itself, and the two +columns are left to the vendor source, written by `set_source`. +""" +from __future__ import annotations + +import pytest + +from patchradar.cpe import parse_cpe +from patchradar.db import database + +NOTEPADPP = "cpe:2.3:a:notepad-plus-plus:notepad\\+\\+" + + +@pytest.mark.asyncio +async def test_a_cpe_does_not_fill_the_vendor_source_columns(): + await database.add_to_watchlist("notepad") + await database.set_installed("notepad", version="8.5.0", cpe=parse_cpe(NOTEPADPP)) + + [entry] = await database.get_watch_entries() + + assert (entry.vendor, entry.product_id) == (None, None) + assert (entry.cpe_vendor, entry.cpe_product) == ("notepad-plus-plus", "notepad\\+\\+") + assert entry.installed() == ("8.5.0", "notepad-plus-plus", "notepad\\+\\+") + + +@pytest.mark.asyncio +async def test_the_vendor_source_is_stored_beside_the_cpe(): + await database.add_to_watchlist("google chrome") + await database.set_installed("google chrome", version="141.0.7390.65", + cpe=parse_cpe("cpe:2.3:a:google:chrome")) + + assert await database.set_source("google chrome", vendor="chrome", product_id="chrome", channel="stable") + + [entry] = await database.get_watch_entries() + assert (entry.vendor, entry.product_id, entry.channel) == ("chrome", "chrome", "stable") + assert entry.cpe == "cpe:2.3:a:google:chrome" + assert entry.installed() == ("141.0.7390.65", "google", "chrome") + + +@pytest.mark.asyncio +async def test_a_vendor_source_alone_does_not_make_an_entry_judgeable(): + """The scan judges with NVD's configurations, which need the CPE.""" + await database.add_to_watchlist("microsoft edge") + await database.set_installed("microsoft edge", version="141.0.3537.71", cpe=None) + await database.set_source("microsoft edge", vendor="edge", product_id="Edge", channel=None) + + [entry] = await database.get_watch_entries() + assert (entry.vendor, entry.product_id) == ("edge", "Edge") + assert not entry.judgeable + + +@pytest.mark.asyncio +async def test_a_cpe_without_a_version_is_not_judgeable_either(): + await database.add_to_watchlist("notepad") + await database.set_installed("notepad", version=None, cpe=parse_cpe(NOTEPADPP)) + + [entry] = await database.get_watch_entries() + assert entry.installed() is None + + +@pytest.mark.asyncio +async def test_set_source_never_adds_a_name(): + assert not await database.set_source("thunderbird", vendor="mozilla", product_id="thunderbird", + channel="release") + assert await database.get_watchlist() == [] diff --git a/tests/test_watchlist_versions.py b/tests/test_watchlist_versions.py index 8aa3885..a51a3e3 100644 --- a/tests/test_watchlist_versions.py +++ b/tests/test_watchlist_versions.py @@ -94,7 +94,7 @@ async def test_migrating_twice_is_not_an_error(old_database): # ─── storing what is installed ─────────────────────────────────────────────── @pytest.mark.asyncio -async def test_version_and_cpe_are_stored_with_vendor_and_product(): +async def test_version_and_cpe_are_stored_and_the_cpe_names_vendor_and_product(): await database.add_to_watchlist("notepad") assert await database.set_installed("notepad", version="8.5.0", cpe=parse_cpe(NOTEPADPP)) @@ -102,8 +102,8 @@ async def test_version_and_cpe_are_stored_with_vendor_and_product(): [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.cpe_vendor == "notepad-plus-plus" + assert entry.cpe_product == "notepad\\+\\+" assert entry.channel is None assert entry.judgeable