feat: Update MergeReport based HTML Report triage (#1204)

* feat: Update MergeReport and add split intermediate graph in html reports

* feat: Add severity count traves in intermediate html reports

* Update docstring quotes in html.py

* update docstring and add new line in html

Author: imsahil007

cve_bin_tool/cli.py

@@ -262,17 +262,28 @@ def main(argv=None):
                           **********************************************
                           """
         LOGGER.warning(warning_nolinux)
+
+    # CSVScanner related settings
+    score = 0
+    if args["severity"]:
+        # Set minimum CVSS score based on severity
+        cvss_score = {"low": 0, "medium": 4, "high": 7, "critical": 9}
+        score = cvss_score[args["severity"]]
+    if int(args["cvss"]) > 0:
+        score = int(args["cvss"])
+
+    merged_reports = None
     if args["merge"]:
         LOGGER.info(
             "You can use -f --format and -o --output-file for saving merged intermediate reports in a file"
         )
-        merged_cves = MergeReports(merge_files=args["merge"])
+        merged_reports = MergeReports(merge_files=args["merge"], score=score)
         if args["input_file"]:
             LOGGER.warning(
                 "Ignoring -i --input-file while merging intermediate reports"
             )
-        args["input_file"] = merged_cves.merge_reports()
-        # Creates a Object for OutputEngine
+            args["input_file"] = None
+        merge_cve_scanner = merged_reports.merge_intermediate()
 
     # Database update related settings
     # Connect to the database
@@ -311,7 +322,12 @@ def main(argv=None):
     cvedb_orig.remove_cache_backup()
 
     # Input validation
-    if not args["directory"] and not args["input_file"] and not args["package_list"]:
+    if (
+        not args["directory"]
+        and not args["input_file"]
+        and not args["package_list"]
+        and not args["merge"]
+    ):
         parser.print_usage()
         with ErrorHandler(logger=LOGGER, mode=ErrorMode.NoTrace):
             raise InsufficientArgs(
@@ -343,15 +359,6 @@ def main(argv=None):
             )
         )
 
-    # CSVScanner related settings
-    score = 0
-    if args["severity"]:
-        # Set minimum CVSS score based on severity
-        cvss_score = {"low": 0, "medium": 4, "high": 7, "critical": 9}
-        score = cvss_score[args["severity"]]
-    if int(args["cvss"]) > 0:
-        score = int(args["cvss"])
-
     with CVEScanner(score=score) as cve_scanner:
         triage_data: TriageData
         total_files: int = 0
@@ -394,6 +401,9 @@ def main(argv=None):
                     cve_scanner.get_cves(product_info, triage_data)
             total_files = version_scanner.total_scanned_files
 
+        if args["merge"]:
+            cve_scanner = merge_cve_scanner
+
         LOGGER.info("")
         LOGGER.info("Overall CVE summary: ")
         if args["input_file"] or args["package_list"]:
@@ -426,10 +436,8 @@ def main(argv=None):
                 total_files=total_files,
                 is_report=args["report"],
                 append=args["append"],
+                merge_report=merged_reports,
             )
-            if args["merge"] and args["input_file"]:
-                # remove the merged json from .cache
-                os.remove(args["input_file"])
 
             if not args["quiet"]:
                 output.output_file(args["format"])

cve_bin_tool/merge.py

@@ -3,10 +3,13 @@
 
 import json
 import os
+from collections import defaultdict
 from datetime import datetime
 from logging import Logger
 from typing import Dict, List
 
+from cve_bin_tool.cve_scanner import CVEScanner
+
 from .cvedb import DISK_LOCATION_DEFAULT
 from .error_handler import (
     ErrorHandler,
@@ -15,8 +18,9 @@
     InvalidJsonError,
     MissingFieldsError,
 )
+from .input_engine import TriageData
 from .log import LOGGER
-from .util import DirWalk
+from .util import DirWalk, ProductInfo, Remarks
 
 REQUIRED_INTERMEDIATE_METADATA = {
     "scanned_dir",
@@ -35,9 +39,11 @@ def __init__(
         logger: Logger = None,
         error_mode=ErrorMode.TruncTrace,
         cache_dir=DISK_LOCATION_DEFAULT,
+        score=0,
     ):
         self.logger = logger or LOGGER.getChild(self.__class__.__name__)
         self.merge_files = merge_files
+        self.intermediate_cve_data = []
         self.all_cve_data = []
         self.file_stack = []
         self.error_mode = error_mode
@@ -46,6 +52,8 @@ def __init__(
         self.products_with_cve = 0
         self.products_without_cve = 0
         self.cache_dir = cache_dir
+        self.merged_files = ["tag"]
+        self.score = score
 
         self.walker = DirWalk(
             pattern=";".join(
@@ -97,6 +105,9 @@ def scan_intermediate_file(self, filename):
                                 f"Adding data from {os.path.basename(filename)} with timestamp {inter_data['metadata']['timestamp']}"
                             )
                             self.total_inter_files += 1
+                            inter_data["metadata"]["severity"] = get_severity_count(
+                                inter_data["report"]
+                            )
                             return inter_data
 
             if missing_fields != set():
@@ -106,41 +117,41 @@ def scan_intermediate_file(self, filename):
             with ErrorHandler(mode=self.error_mode):
                 raise InvalidIntermediateJsonError(filename)
 
-    def merge_reports(self):
+    def merge_intermediate(self):
         """Merge valid intermediate dictionaries"""
 
         for inter_file in self.recursive_scan(self.merge_files):
-            # Remove duplicate paths from cve-entries
-            self.all_cve_data.append(self.scan_intermediate_file(inter_file))
-
-        if self.all_cve_data:
+            # Create a list of intermediate files dictionary
+            self.intermediate_cve_data.append(self.scan_intermediate_file(inter_file))
+
+        if self.intermediate_cve_data:
+            # sort on basis of timestamp and scans
+            self.intermediate_cve_data.sort(
+                key=lambda inter: datetime.strptime(
+                    inter["metadata"]["timestamp"], "%Y-%m-%d.%H-%M-%S"
+                )
+            )
             self.all_cve_data = self.remove_intermediate_duplicates()
-            merged_file_path = self.save_merged_intermediate()
-            return merged_file_path
-
-        self.logger.error("No valid Intermediate reports found!")
-        return ""
-
-    def save_merged_intermediate(self):
-        """Save a temporary merged report in .cache/cve-bin-tool"""
+            merged_cve_scanner = self.get_intermediate_cve_scanner(
+                [self.all_cve_data], self.score
+            )[0]
+            merged_cve_scanner.products_with_cve = self.products_with_cve
+            merged_cve_scanner.products_without_cve = self.products_without_cve
 
-        if not os.path.isdir(self.cache_dir):
-            os.makedirs(self.cache_dir)
+            return merged_cve_scanner
 
-        now = datetime.now().strftime("%Y-%m-%d.%H-%M-%S")
-        filename = os.path.join(self.cache_dir, f"merged-{now}.json")
-        with open(filename, "w") as f:
-            json.dump(self.all_cve_data, f, indent="    ")
-
-        return filename
+        self.logger.error("No valid Intermediate reports found!")
+        return {}
 
     def remove_intermediate_duplicates(self) -> List[Dict[str, str]]:
         """Returns a list of dictionary with same format as cve-bin-tool json output"""
 
         output = {}
-        for inter_data in self.all_cve_data:
-            self.products_with_cve += inter_data["metadata"]["products_with_cve"]
-            self.products_without_cve += inter_data["metadata"]["products_without_cve"]
+        for inter_data in self.intermediate_cve_data:
+            self.products_with_cve += int(inter_data["metadata"]["products_with_cve"])
+            self.products_without_cve += int(
+                inter_data["metadata"]["products_without_cve"]
+            )
             for cve in inter_data["report"]:
                 if cve["cve_number"] != "UNKNOWN":
                     if cve["cve_number"] not in output:
@@ -156,3 +167,60 @@ def remove_intermediate_duplicates(self) -> List[Dict[str, str]]:
                         output[cve["cve_number"]]["path"] = path_list
 
         return list(output.values())
+
+    @staticmethod
+    def get_intermediate_cve_scanner(cve_data_list, score) -> List[CVEScanner]:
+        """Returns a list of CVEScanner parsed objects when a list of cve_data json like list is passed"""
+        cve_scanner_list = []
+        for inter_data in cve_data_list:
+            with CVEScanner(score=score) as cve_scanner:
+                triage_data: TriageData
+                parsed_data: Dict[ProductInfo, TriageData] = {}
+
+                parsed_data = parse_data_from_json(
+                    inter_data["report"] if "report" in inter_data else inter_data
+                )
+
+                for product_info, triage_data in parsed_data.items():
+                    LOGGER.debug(f"{product_info}, {triage_data}")
+                    cve_scanner.get_cves(product_info, triage_data)
+
+                cve_scanner_list.append(cve_scanner)
+
+        return cve_scanner_list
+
+
+def parse_data_from_json(
+    json_data: List[Dict[str, str]]
+) -> Dict[ProductInfo, TriageData]:
+    """Parse CVE JSON dictionary to Dict[ProductInfo, TriageData]"""
+
+    parsed_data: Dict[ProductInfo, TriageData] = defaultdict(dict)
+
+    for row in json_data:
+        product_info = ProductInfo(
+            row["vendor"].strip(), row["product"].strip(), row["version"].strip()
+        )
+        parsed_data[product_info][row.get("cve_number", "").strip() or "default"] = {
+            "remarks": Remarks(str(row.get("remarks", "")).strip()),
+            "comments": row.get("comments", "").strip(),
+            "severity": row.get("severity", "").strip(),
+        }
+
+        parsed_data[product_info]["paths"] = set(
+            map(lambda x: x.strip(), row.get("paths", "").split(","))
+        )
+    return parsed_data
+
+
+def get_severity_count(reports: List[Dict[str, str]] = []) -> Dict[str, int]:
+    """Returns a list of Severity counts for intermediate report"""
+    severity_count = {"LOW": 0, "MEDIUM": 0, "HIGH": 0, "CRITICAL": 0, "UNKNOWN": 0}
+
+    for cve in reports:
+        if "severity" in cve and cve["severity"] in severity_count:
+            severity_count[cve["severity"]] += 1
+        else:
+            severity_count["UNKNOWN"] += 1
+
+    return severity_count

cve_bin_tool/output_engine/__init__.py

@@ -6,7 +6,7 @@
 import os
 import time
 from logging import Logger
-from typing import IO, Dict, Union
+from typing import IO, Dict, List, Union
 
 from ..cve_scanner import CVEData
 from ..cvedb import CVEDB
@@ -200,6 +200,7 @@ def __init__(
         total_files: int = 0,
         is_report: bool = False,
         append: Union[str, bool] = False,
+        merge_report: Union[None, List[str]] = None,
     ):
         self.logger = logger or LOGGER.getChild(self.__class__.__name__)
         self.all_cve_data = all_cve_data
@@ -213,6 +214,7 @@ def __init__(
         self.time_of_last_update = time_of_last_update
         self.append = append
         self.tag = tag
+        self.merge_report = merge_report
 
     def output_cves(self, outfile, output_type="console"):
         """Output a list of CVEs
@@ -236,6 +238,7 @@ def output_cves(self, outfile, output_type="console"):
                 self.total_files,
                 self.products_with_cve,
                 self.products_without_cve,
+                self.merge_report,
                 self.logger,
                 outfile,
             )