fix(get-cves): Store cpe update values and use to correctly fetch cves

Signed-off-by: Tom Marks <thomas.marks@palindrometech.com>

Author: Tom

cve_bin_tool/cve_scanner.py

@@ -1,6 +1,7 @@
 # Copyright (C) 2021 Intel Corporation
 # SPDX-License-Identifier: GPL-3.0-or-later
 
+# import itertools
 import sqlite3
 import sys
 from collections import defaultdict
@@ -143,15 +144,39 @@ def get_cves(self, product_info: ProductInfo, triage_data: TriageData):
         # Check for anything directly marked
         query = """
         SELECT CVE_number FROM cve_range
-        WHERE vendor=? AND product=? AND version=?
+        WHERE (
+          vendor=:vendor
+          AND product=:product
+          AND version=:version
+          AND data_source != "NVD"
+        ) OR (
+          vendor=:vendor
+          AND product=:product
+          AND (version || versionUpdate) = :version
+          AND data_source = "NVD"
+        )
         """
         # Removing * from vendors that are guessed by the package list parser
         vendor = product_info.vendor.replace("*", "")
 
         # Use our Version class to do version compares
         parsed_version = Version(product_info.version)
 
-        self.cursor.execute(query, [vendor, product_info.product, str(parsed_version)])
+        version = str(parsed_version)
+        # numeric = "0123456789."
+        # numeric_version = "".join(itertools.takewhile(lambda i: i in numeric, parsed_version))
+        # version_update = product_info.version
+
+        self.cursor.execute(
+            query,
+            {
+                "vendor": vendor,
+                "product": product_info.product,
+                "version": version,
+                # "numericVersion": numeric_version,
+                # "versionUpdate": version_update,
+            },
+        )
 
         cve_list = list(map(lambda x: x[0], self.cursor.fetchall()))
 

cve_bin_tool/cvedb.py

@@ -89,6 +89,7 @@ class CVEDB:
             vendor TEXT,
             product TEXT,
             version TEXT,
+            versionUpdate TEXT,
             versionStartIncluding TEXT,
             versionStartExcluding TEXT,
             versionEndIncluding TEXT,
@@ -185,13 +186,14 @@ class CVEDB:
             vendor,
             product,
             version,
+            versionUpdate,
             versionStartIncluding,
             versionStartExcluding,
             versionEndIncluding,
             versionEndExcluding,
             data_source
             )
-            VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)
+            VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
         """,
         "insert_exploit": """
         INSERT or REPLACE INTO cve_exploited (
@@ -611,6 +613,7 @@ def populate_affected(self, affected_data, cursor, data_source):
                         affected["vendor"],
                         affected["product"],
                         affected["version"],
+                        affected["versionUpdate"],
                         affected["versionStartIncluding"],
                         affected["versionStartExcluding"],
                         affected["versionEndIncluding"],

cve_bin_tool/data_sources/nvd_source.py

@@ -69,12 +69,13 @@ def __init__(
         nvd_type: str = "json-mirror",
         incremental_update: bool = False,
         nvd_api_key: str = "",
+        cachedir: Path | None = None,
     ):
         if feed is None:
             self.feed = self.FEED_NVD if nvd_type == "json-nvd" else self.FEED_MIRROR
         else:
             self.feed = feed
-        self.cachedir = self.CACHEDIR
+        self.cachedir = cachedir or self.CACHEDIR
         self.backup_cachedir = self.BACKUPCACHEDIR
         self.error_mode = error_mode
         self.source_name = self.SOURCE
@@ -102,7 +103,10 @@ def __init__(
     async def get_cve_data(self):
         """Retrieves the CVE data from the data source."""
         await self.fetch_cves()
+        return self._load_cve_data()
 
+    def _load_cve_data(self):
+        """Loads the CVE data which has been fetched from the data source."""
         if self.nvd_type == "api2":
             return self.format_data_api2(self.all_cve_entries), self.source_name
 
@@ -211,10 +215,17 @@ def parse_node(self, node: dict[str, list[dict[str, str]]]) -> list[dict[str, st
             for cpe_match in vulnerable_matches:
                 # split on `:` only if it's not escaped
                 cpe_split = re.split(r"(?<!\\):", cpe_match["cpe23Uri"])
+
+                if cpe_split[6] in ("-", "*", ""):
+                    versionUpdate = ""
+                else:
+                    versionUpdate = cpe_split[6]
+
                 affects = {
                     "vendor": cpe_split[3],
                     "product": cpe_split[4],
                     "version": cpe_split[5],
+                    "versionUpdate": versionUpdate,
                 }
 
                 # if we have a range (e.g. version is *) fill it out, and put blanks where needed
@@ -395,10 +406,17 @@ def parse_node_api2(
             for cpe_match in vulnerable_matches:
                 # split on `:` only if it's not escaped
                 cpe_split = re.split(r"(?<!\\):", cpe_match["criteria"])
+
+                if cpe_split[6] in ("-", "*", ""):
+                    versionUpdate = ""
+                else:
+                    versionUpdate = cpe_split[6]
+
                 affects = {
                     "vendor": cpe_split[3],
                     "product": cpe_split[4],
                     "version": cpe_split[5],
+                    "versionUpdate": versionUpdate,
                 }
 
                 # if we have a range (e.g. version is *) fill it out, and put blanks where needed

test/test_cvedb.py

@@ -2,6 +2,7 @@
 # SPDX-License-Identifier: GPL-3.0-or-later
 
 import datetime
+import pathlib
 import shutil
 import tempfile
 from test.utils import EXTERNAL_SYSTEM, LONG_TESTS
@@ -16,8 +17,10 @@
 class TestCVEDB:
     @classmethod
     def setup_class(cls):
-        cls.nvd = nvd_source.NVD_Source(nvd_type="json")
         cachedir = tempfile.mkdtemp(prefix="cvedb-")
+        cls.nvd = nvd_source.NVD_Source(
+            nvd_type="json", cachedir=pathlib.Path(cachedir)
+        )
         cls.exported_data = tempfile.mkdtemp(prefix="exported-data-")
         cls.cvedb = cvedb.CVEDB(sources=[cls.nvd], cachedir=cachedir)
         cls.nvd.cachedir = cachedir
@@ -91,3 +94,85 @@ def test_new_database_schema(self):
             assert all(column in column_names for column in required_columns[table])
 
         self.cvedb.db_close()
+
+    @pytest.mark.asyncio
+    @pytest.mark.skipif(
+        not EXTERNAL_SYSTEM(),
+        reason="NVD tests run only when EXTERNAL_SYSTEM=1",
+    )
+    async def test_populate_foo(self):
+        await self.nvd.get_cve_data()
+        self.cvedb.data = [self.nvd._load_cve_data()]
+        self.cvedb.init_database()
+        self.cvedb.populate_db()
+        self.cvedb.check_cve_entries()
+        assert self.cvedb.cve_count > 0
+
+    def test_populate_db(self):
+        """Test to check whether we are able to fetch and save the nvd entries"""
+        severity_data = [
+            {
+                "ID": "CVE-9999-9999",
+                "description": "Test Vuln",
+                "severity": "UNKNOWN",
+                "score": 7.5,
+                "CVSS_version": 2,
+                "CVSS_vector": "AV:N/AC:L/Au:N/C:P/I:P/A:P",
+                "last_modified": "2025-04-09T00:30:58.490",
+            }
+        ]
+        affected_data = [
+            {
+                "vendor": "test-vendor",
+                "product": "test-product",
+                "version": "6.1",
+                "versionUpdate": "p1",
+                "versionStartIncluding": "",
+                "versionStartExcluding": "",
+                "versionEndIncluding": "",
+                "versionEndExcluding": "",
+                "cve_id": "CVE-9999-9999",
+            }
+        ]
+        source_name = "NVD"
+        test_data = ((severity_data, affected_data), source_name)
+        self.cvedb.data = [test_data]
+
+        self.cvedb.init_database()
+        self.cvedb.populate_db()
+        self.cvedb.check_cve_entries()
+
+        from cve_bin_tool.cve_scanner import CVEScanner
+        from cve_bin_tool.util import CVE, ProductInfo
+
+        class TestableCVEScanner(CVEScanner):
+            dbname: str = str(self.cvedb.dbpath)
+
+        with TestableCVEScanner() as cve_scanner:
+            product_info = ProductInfo(
+                vendor="test-vendor",
+                product="test-product",
+                version="6.1p1",
+            )
+            triage_data = {"paths": []}
+
+            cve_scanner.get_cves(product_info, triage_data)
+            products_with_cve = cve_scanner.products_with_cve == 1
+            results = dict(cve_scanner.all_cve_data)
+
+        assert products_with_cve == 1
+        assert results[product_info] == {
+            "cves": [
+                CVE(
+                    cve_number="CVE-9999-9999",
+                    severity="UNKNOWN",
+                    description="Test Vuln",
+                    score=7.5,
+                    cvss_version=2,
+                    cvss_vector="AV:N/AC:L/Au:N/C:P/I:P/A:P",
+                    data_source="NVD",
+                    last_modified="",
+                ),
+            ],
+            "paths": set(),
+        }