fix: handle disabled_sources in get_vendor_product_pairs

get_vendor_product_pairs function doesn't handle disabled sources passed
by the user. As a result, the user can't disable a datasource
(e.g., OSV) when parsing a python PKG-INFO file.

Fix this by passing enabled_sources from cli to version_scanner and then
to cvedb. To achieve this functionality, source_nvd must also be added
to enabled_sources when appropriate.

Signed-off-by: Fabrice Fontaine <[email protected]>

Author: ffontaine

cve_bin_tool/cli.py

@@ -734,6 +734,7 @@ def main(argv=None):
             nvd_api_key=args["nvd_api_key"],
             error_mode=error_mode,
         )
+        enabled_sources.append(source_nvd)
         default_sources = [source_nvd]
         default_sources.extend(enabled_sources)
     else:
@@ -1024,6 +1025,7 @@ def main(argv=None):
                 exclude_folders=args["exclude"],
                 error_mode=error_mode,
                 validate=not args["disable_validation_check"],
+                sources=enabled_sources,
             )
             version_scanner.remove_skiplist(skips)
             LOGGER.info(f"Number of checkers: {version_scanner.number_of_checkers()}")

cve_bin_tool/cvedb.py

@@ -679,12 +679,15 @@ def get_vendor_product_pairs(self, package_names) -> list[dict[str, str]]:
         vendor_package_pairs = []
         query = """
         SELECT DISTINCT vendor FROM cve_range
-        WHERE product=?
-        """
+        WHERE product=? AND data_source IN (%s)
+        """ % ",".join(
+            "?" for i in self.sources
+        )
 
+        data_sources = list(map(lambda x: x.source_name, self.sources))
         # For python package checkers we don't need the progress bar running
         if type(package_names) is not list:
-            cursor.execute(query, [package_names])
+            cursor.execute(query, [package_names] + data_sources)
             vendors = list(map(lambda x: x[0], cursor.fetchall()))
 
             for vendor in vendors:
@@ -703,7 +706,7 @@ def get_vendor_product_pairs(self, package_names) -> list[dict[str, str]]:
             for package_name in track(
                 package_names, description="Processing the given list...."
             ):
-                cursor.execute(query, [package_name["name"].lower()])
+                cursor.execute(query, [package_name["name"].lower()] + data_sources)
                 vendors = list(map(lambda x: x[0], cursor.fetchall()))
                 for vendor in vendors:
                     if vendor != "":

cve_bin_tool/version_scanner.py

@@ -54,6 +54,7 @@ def __init__(
         error_mode: ErrorMode = ErrorMode.TruncTrace,
         score: int = 0,
         validate: bool = True,
+        sources=None,
     ):
         self.logger = logger or LOGGER.getChild(self.__class__.__name__)
         # Update egg if installed in development mode
@@ -76,7 +77,7 @@ def __init__(
         self.should_extract = should_extract
         self.file_stack: list[str] = []
         self.error_mode = error_mode
-        self.cve_db = CVEDB()
+        self.cve_db = CVEDB(sources=sources)
         self.validate = validate
         # self.logger.info("Checkers loaded: %s" % (", ".join(self.checkers.keys())))
         self.language_checkers = self.available_language_checkers()