Fix NVD download of CVE data

fixes #906

Author: imsahil007

cve_bin_tool/async_utils.py

@@ -46,7 +46,9 @@ def run_coroutine(coro):
 
 async def aio_run_command(args):
     process = await asyncio.create_subprocess_exec(
-        *args, stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.PIPE,
+        *args,
+        stdout=asyncio.subprocess.PIPE,
+        stderr=asyncio.subprocess.PIPE,
     )
     stdout, stderr = await process.communicate()
     return stdout, stderr  # binary encoded

cve_bin_tool/cli.py

@@ -86,7 +86,11 @@ def main(argv=None):
     )
 
     input_group.add_argument(
-        "-i", "--input-file", action="store", default="", help="provide input filename",
+        "-i",
+        "--input-file",
+        action="store",
+        default="",
+        help="provide input filename",
     )
     input_group.add_argument(
         "-C", "--config", action="store", default="", help="provide config file"
@@ -144,7 +148,10 @@ def main(argv=None):
         help="update schedule for NVD database (default: daily)",
     )
     parser.add_argument(
-        "-x", "--extract", action="store_true", help="autoextract compressed files",
+        "-x",
+        "--extract",
+        action="store_true",
+        help="autoextract compressed files",
     )
     parser.add_argument(
         "--disable-version-check",

cve_bin_tool/cve_scanner.py

@@ -48,7 +48,7 @@ def __init__(
         self.all_cve_data = defaultdict(CVEData)
 
     def get_cves(self, product_info: ProductInfo, triage_data: TriageData):
-        """ Get CVEs against a specific version of a product.
+        """Get CVEs against a specific version of a product.
 
         Example:
             nvd.get_cves('haxx', 'curl', '7.34.0')

cve_bin_tool/cvedb.py

@@ -47,7 +47,8 @@ class CVEDB:
     LOGGER = LOGGER.getChild("CVEDB")
     NVDCVE_FILENAME_TEMPLATE = "nvdcve-1.1-{}.json.gz"
     CURL_CVE_FILENAME_TEMPLATE = "curlcve-{}.json"
-    META_REGEX = re.compile(r"https://.*/json/.*-[0-9]*\.[0-9]*-[0-9]*\.meta")
+    META_LINK = "https://nvd.nist.gov"
+    META_REGEX = re.compile(r"\/feeds\/json\/.*-[0-9]*\.[0-9]*-[0-9]*\.meta")
     RANGE_UNSET = ""
 
     def __init__(
@@ -91,7 +92,10 @@ async def nist_scrape(self, session):
             json_meta_links = self.META_REGEX.findall(page)
             return dict(
                 await asyncio.gather(
-                    *[self.getmeta(session, meta_url) for meta_url in json_meta_links]
+                    *[
+                        self.getmeta(session, f"{self.META_LINK}{meta_url}")
+                        for meta_url in json_meta_links
+                    ]
                 )
             )
 
@@ -561,7 +565,7 @@ def curl_versions(self):
         regex = re.compile(r"curlcve-(\d+.\d+.\d).json")
         return [
             regex.search(filename).group(1)
-            for filename in glob.glob(os.path.join(self.cachedir, "curl-*.json"))
+            for filename in glob.glob(os.path.join(self.cachedir, "curlcve-*.json"))
         ]
 
     def clear_cached_data(self):

cve_bin_tool/egg_updater.py

@@ -46,7 +46,8 @@ def update_egg():
                         )
                         for filename in os.listdir(
                             os.path.join(
-                                os.path.abspath(os.path.dirname(__file__)), "checkers",
+                                os.path.abspath(os.path.dirname(__file__)),
+                                "checkers",
                             )
                         )
                         if filename.endswith(".py") and "__init__" not in filename

cve_bin_tool/output_engine/html.py

@@ -92,13 +92,17 @@ def output_html(
 
     # Chart configuration for product_pie
     product_pie.update_layout(
-        autosize=True, legend_orientation="h",
+        autosize=True,
+        legend_orientation="h",
     )
     product_pie.update_traces(
         hoverinfo="label+percent",
         textinfo="value",
         textfont_size=14,
-        marker=dict(colors=["#d80032", "#1a936f"], line=dict(color="white", width=2),),
+        marker=dict(
+            colors=["#d80032", "#1a936f"],
+            line=dict(color="white", width=2),
+        ),
     )
 
     # dash graph2: Product CVE's Graph
@@ -117,7 +121,9 @@ def output_html(
             )
 
     # Chart configuration for cve_bar
-    cve_bar.update_layout(yaxis_title="Number of CVE's",)
+    cve_bar.update_layout(
+        yaxis_title="Number of CVE's",
+    )
 
     all_paths = defaultdict(list)
 
@@ -134,19 +140,34 @@ def output_html(
             hid = f"{product_info.vendor}{product_info.product}{''.join(product_info.version.split('.'))}"
 
             new_cves = render_cves(
-                hid, cve_row, "NEW", cve_by_remark[Remarks.NewFound],
+                hid,
+                cve_row,
+                "NEW",
+                cve_by_remark[Remarks.NewFound],
             )
             mitigated_cves = render_cves(
-                hid, cve_row, "MITIGATED", cve_by_remark[Remarks.Mitigated],
+                hid,
+                cve_row,
+                "MITIGATED",
+                cve_by_remark[Remarks.Mitigated],
             )
             confirmed_cves = render_cves(
-                hid, cve_row, "CONFIRMED", cve_by_remark[Remarks.Confirmed],
+                hid,
+                cve_row,
+                "CONFIRMED",
+                cve_by_remark[Remarks.Confirmed],
             )
             unexplored_cves = render_cves(
-                hid, cve_row, "UNEXPLORED", cve_by_remark[Remarks.Unexplored],
+                hid,
+                cve_row,
+                "UNEXPLORED",
+                cve_by_remark[Remarks.Unexplored],
             )
             ignored_cves = render_cves(
-                hid, cve_row, "IGNORED", cve_by_remark[Remarks.Ignored],
+                hid,
+                cve_row,
+                "IGNORED",
+                cve_by_remark[Remarks.Ignored],
             )
 
             analysis_data = Counter(cve.severity for cve in cve_data["cves"])
@@ -173,7 +194,10 @@ def output_html(
                 hoverinfo="label+percent",
                 textinfo="value",
                 textfont_size=14,
-                marker=dict(colors=colors, line=dict(color="white", width=2),),
+                marker=dict(
+                    colors=colors,
+                    line=dict(color="white", width=2),
+                ),
             )
             analysis_pie.update_layout(
                 autosize=True,

test/test_cli.py

@@ -187,34 +187,28 @@ def test_update(self, caplog):
         with caplog.at_level(logging.INFO):
             main(["cve-bin-tool", "-u", "daily", test_path])
         assert (
-            (
-                "cve_bin_tool.CVEDB",
-                logging.INFO,
-                "Using cached CVE data (<24h old). Use -u now to update immediately.",
-            )
-            in caplog.record_tuples
-            or (
-                "cve_bin_tool.CVEDB",
-                logging.INFO,
-                "Updating CVE data. This will take a few minutes.",
-            )
-            in caplog.record_tuples
-        )
+            "cve_bin_tool.CVEDB",
+            logging.INFO,
+            "Using cached CVE data (<24h old). Use -u now to update immediately.",
+        ) in caplog.record_tuples or (
+            "cve_bin_tool.CVEDB",
+            logging.INFO,
+            "Updating CVE data. This will take a few minutes.",
+        ) in caplog.record_tuples
         caplog.clear()
 
         with caplog.at_level(logging.INFO):
             main(["cve-bin-tool", "-u", "now", test_path])
         db_path = DISK_LOCATION_DEFAULT
         assert (
-            ("cve_bin_tool.CVEDB", logging.WARNING, f"Deleting cachedir {db_path}",)
-            in caplog.record_tuples
-            and (
-                "cve_bin_tool.CVEDB",
-                logging.INFO,
-                "Updating CVE data. This will take a few minutes.",
-            )
-            in caplog.record_tuples
-        )
+            "cve_bin_tool.CVEDB",
+            logging.WARNING,
+            f"Deleting cachedir {db_path}",
+        ) in caplog.record_tuples and (
+            "cve_bin_tool.CVEDB",
+            logging.INFO,
+            "Updating CVE data. This will take a few minutes.",
+        ) in caplog.record_tuples
         caplog.clear()
 
         with caplog.at_level(logging.INFO):

test/test_data/icecast.py

@@ -1,5 +1,9 @@
 mapping_test_data = [
-    {"product": "icecast", "version": "2.3.0", "version_strings": ["Icecast 2.3.0"],}
+    {
+        "product": "icecast",
+        "version": "2.3.0",
+        "version_strings": ["Icecast 2.3.0"],
+    }
 ]
 package_test_data = [
     {

test/test_json.py

@@ -37,7 +37,8 @@ def test_json_validation(self, year):
         """ Validate latest nvd json file against their published schema """
         # Open the latest nvd file on disk
         with gzip.open(
-            os.path.join(DISK_LOCATION_DEFAULT, f"nvdcve-1.1-{year}.json.gz"), "rb",
+            os.path.join(DISK_LOCATION_DEFAULT, f"nvdcve-1.1-{year}.json.gz"),
+            "rb",
         ) as json_file:
             nvd_json = json.loads(json_file.read())
             LOGGER.info(f"Loaded json for year {year}: nvdcve-1.1-{year}.json.gz")