feat: Add Red Hat data source (Fixes #2331, #2367) (#2368)

* Fixes #2331 
* Fixes #2367

Author: anthonyharrison

README.md

@@ -179,7 +179,7 @@ CVE Data Download:
   <a href="https://github.com/intel/cve-bin-tool/blob/main/doc/MANUAL.md#--nvd-api-key-nvd_api_key">--nvd-api-key NVD_API_KEY</a>
                         specify NVD API key (used to improve NVD rate limit)
   <a href="https://github.com/intel/cve-bin-tool/blob/main/doc/MANUAL.md#-d-nvdosvgad-nvdosvgad----disable-data-source-nvdosvgad-nvdosvgad-">-d {NVD,OSV} [{NVD,OSV} ...], --disable-data-source {NVD,OSV} [{NVD,OSV} ...]</a>
-                        specify data sources that should be disabled
+                        comma-separated list of data sources (GAD, NVD, OSV, REDHAT) to disable (default: NONE)
 
 Input:
   <a href="https://github.com/intel/cve-bin-tool/blob/main/doc/MANUAL.md#directory-positional-argument">directory</a>             directory to scan
@@ -289,18 +289,26 @@ This data source consists of majority of the CVE entries and is essential to pro
 
 ### [Open Source Vulnerability Database](https://osv.dev/) (OSV)
 
-This data source is based on the OSV schema from google, and consists of CVEs from different ecosystems that might not be covered by NVD.
+This data source is based on the OSV schema from Google, and consists of CVEs from different ecosystems that might not be covered by NVD.
 NVD is given priority if there are duplicate CVEs as some CVEs from OSV may not contain CVSS scores.
 Using OSV will increase number of CVEs and time taken to update the database but searching database for vulnerabilities will have similar performance.
 
 ### [Gitlab Advisory Database](https://advisories.gitlab.com/) (GAD)
 
 This data source consists of security advisories used by the GitLab dependency scanner.
-Amount of CVEs added from this data source is similar to OSV.
+The number of CVEs added from this data source is similar to OSV.
 
-### [RedHat Advisory Database](https://access.redhat.com/security/data) (RSD)
+### [RedHat Security Database](https://access.redhat.com/security/data) (REDHAT)
 
-This data source contains CVEs pertaining to RedHat Products. Many are also included in NVD (duplicate CVEs are not reported multiple times).
+This data source contains CVEs pertaining to RedHat Products.
+
+Access to the data is subject to [Legal Notice](https://access.redhat.com/documentation/en-us/red_hat_security_data_api/1.0/html/red_hat_security_data_api/legal-notice).
+
+### [RedHat Security Database](https://access.redhat.com/security/data) (REDHAT)
+
+This data source contains CVEs pertaining to RedHat Products.
+
+Access to the data is subject to [Legal Notice](https://access.redhat.com/documentation/en-us/red_hat_security_data_api/1.0/html/red_hat_security_data_api/legal-notice).
 
 ## Binary checker list
 

cve_bin_tool/cli.py

@@ -41,11 +41,12 @@
 from cve_bin_tool.cve_scanner import CVEScanner
 from cve_bin_tool.cvedb import CVEDB, OLD_CACHE_DIR
 from cve_bin_tool.data_sources import (
+    DataSourceSupport,
     curl_source,
     gad_source,
     nvd_source,
     osv_source,
-    rsd_source,
+    redhat_source,
 )
 from cve_bin_tool.error_handler import (
     CVEDataMissing,
@@ -106,7 +107,7 @@ def main(argv=None):
         )
         + "\n\n"
         + textwrap.fill(
-            "Available data-sources: National Vulnerability Database (NVD), Open Source Vulnerability Database (OSV), Gitlab Advisory Database (GAD)"
+            f'Available data sources: {", ".join(DataSourceSupport.available_data_sources())}'
         )
         + "\n\n"
         + textwrap.fill(
@@ -141,13 +142,12 @@ def main(argv=None):
         default="",
         help="specify NVD API key (used to improve NVD rate limit)",
     )
+    data_source_disable_help = f'comma-separated list of data sources ({", ".join(DataSourceSupport.available_data_sources())}) to disable (default: NONE)'
     data_sources_group.add_argument(
         "-d",
         "--disable-data-source",
-        nargs="+",
-        choices=["NVD", "OSV", "GAD", "RSD"],
-        type=str,
-        help="specify data sources that should be disabled",
+        action=StringToListAction,
+        help=data_source_disable_help,
         default=[],
     )
 
@@ -493,8 +493,27 @@ def main(argv=None):
     )
 
     # list of sources that can be disabled but are not disabled
+    # Data Source Processing
+    # Ensure data source names are all upper case before validating list
+    disable_data_sources = [d.upper() for d in args["disable_data_source"]]
+    # Validate data source choices
+    data_sources = DataSourceSupport()
+    valid_data_sources = data_sources.get_data_sources()
+    disabled_sources = []
+    if len(disable_data_sources) > 0:
+        LOGGER.debug(f"Processing disabled data sources {disable_data_sources}")
+        for data_source in disable_data_sources:
+            if data_source not in valid_data_sources:
+                LOGGER.warning(
+                    f"Argument --disable-data-source: invalid choice: {data_source} (choose from {valid_data_sources})"
+                )
+            else:
+                LOGGER.info(f"Disabling data source {data_source}")
+                disabled_sources.append(data_source)
+        LOGGER.debug(f"Accepted disabled data sources {disabled_sources}")
+
+    # Maintain list of sources that are used
     enabled_sources = []
-    disabled_sources = args["disable_data_source"]
 
     if "OSV" not in disabled_sources:
         source_osv = osv_source.OSV_Source(incremental_update=incremental_db_update)
@@ -506,12 +525,11 @@ def main(argv=None):
         )
         enabled_sources.append(source_gad)
 
-    if "RSD" not in disabled_sources:
-        source_rsd = rsd_source.RSD_Source(incremental_update=incremental_db_update)
-        enabled_sources.append(source_rsd)
-
-    for data_source in disabled_sources:
-        LOGGER.info(f"Disabling data source {data_source}")
+    if "REDHAT" not in disabled_sources:
+        source_redhat = redhat_source.REDHAT_Source(
+            incremental_update=incremental_db_update
+        )
+        enabled_sources.append(source_redhat)
 
     default_sources = [source_nvd, curl_source.Curl_Source()]
     default_sources.extend(enabled_sources)
@@ -584,7 +602,7 @@ def main(argv=None):
         "%d %B %Y at %H:%M:%S", time.localtime(cvedb_orig.get_db_update_date())
     )
     LOGGER.info(
-        "CVE database contains CVEs from National Vulnerability Database (NVD), Open Source Vulnerability Database (OSV), Gitlab Advisory Database (GAD)"
+        "CVE database contains CVEs from National Vulnerability Database (NVD), Open Source Vulnerability Database (OSV), Gitlab Advisory Database (GAD) and RedHat"
     )
     LOGGER.info(f"CVE database last updated on {db_date}")
 

cve_bin_tool/cvedb.py

@@ -283,9 +283,7 @@ def populate_db(self) -> None:
         for cve_data, source_name in self.data:
 
             if source_name != "NVD" and cve_data[0] is not None:
-                if source_name != "RSD":
-                    cve_data = self.update_vendors(cve_data)
-                cve_data = self.filter_duplicate(cve_data, source_name)
+                cve_data = self.update_vendors(cve_data)
 
             severity_data, affected_data = cve_data
 
@@ -492,39 +490,6 @@ def update_vendors(self, cve_data):
 
         return updated_severity, updated_affected
 
-    def filter_duplicate(self, cve_data, source):
-        """Filter out duplicate CVEs in CVE data."""
-        updated_severity = []
-        updated_affected = []
-
-        severity_data, affected_data = cve_data
-
-        cursor = self.db_open_and_get_cursor()
-
-        query = """
-        SELECT cve_number, data_source FROM cve_severity
-        WHERE cve_number=?
-        """
-
-        sel_cve = set()
-
-        for affected in affected_data:
-            cursor.execute(query, [affected["cve_id"]])
-            result = cursor.fetchall()
-            cve = list(map(lambda x: x[0], result))
-
-            if len(cve) == 0 or result[0][1] == source:
-                updated_affected.append(affected)
-                sel_cve.add(affected["cve_id"])
-
-        for cve in severity_data:
-            if cve["ID"] in sel_cve:
-                updated_severity.append(cve)
-
-        self.db_close()
-
-        return updated_severity, updated_affected
-
     def db_open_and_get_cursor(self) -> sqlite3.Cursor:
         """Opens connection to sqlite database, returns cursor object."""
 

cve_bin_tool/data_sources/__init__.py

@@ -1,9 +1,17 @@
 # Copyright (C) 2022 Intel Corporation
 # SPDX-License-Identifier: GPL-3.0-or-later
 
+from __future__ import annotations
+
+import sys
 from abc import ABC, abstractmethod
 from pathlib import Path
 
+if sys.version_info >= (3, 9):
+    import importlib.resources as resources
+else:
+    import importlib_resources as resources
+
 USER_HOME = Path("~")
 
 # database defaults
@@ -18,3 +26,31 @@ class Data_Source(ABC):
     @abstractmethod
     async def get_cve_data(self):
         pass
+
+
+class DataSourceSupport:
+
+    # Supported Data Sources
+    DATA_SOURCES_ENTRYPOINT = "cve_bin_tool.data_sources"
+
+    def __init__(self):
+        self.data_sources = self.available_data_sources()
+
+    @classmethod
+    def available_data_sources(cls) -> list[str]:
+        """Find Data Sources"""
+        data_sources_directory = resources.files(cls.DATA_SOURCES_ENTRYPOINT)
+        sources = data_sources_directory.iterdir()
+        disable_source = []
+        disable_source.append("curl_source")
+        disable_source.append("__init__")
+        data_sources = []
+        for data_source in sources:
+            if str(data_source).endswith(".py"):
+                source = data_source.name.replace(".py", "")
+                if source not in disable_source:
+                    data_sources.append(source.replace("_source", "").upper())
+        return sorted(data_sources)
+
+    def get_data_sources(self) -> list[str]:
+        return self.data_sources