feat: improved triage process (#4279)

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

Author: mastersans

cve_bin_tool/cli.py

@@ -75,6 +75,7 @@
 from cve_bin_tool.util import ProductInfo
 from cve_bin_tool.version import VERSION
 from cve_bin_tool.version_scanner import VersionScanner
+from cve_bin_tool.vex_manager.parse import VEXParse
 
 sys.excepthook = excepthook  # Always install excepthook for entrypoint module.
 
@@ -180,12 +181,6 @@ def main(argv=None):
         default="",
         help="provide input filename",
     )
-    input_group.add_argument(
-        "--triage-input-file",
-        action="store",
-        default="",
-        help="provide input filename for triage data",
-    )
     input_group.add_argument(
         "-C", "--config", action="store", default="", help="provide config file"
     )
@@ -358,7 +353,7 @@ def main(argv=None):
     output_group.add_argument(
         "--vex-type",
         action="store",
-        default="cyclonedx",
+        default="",
         choices=["cyclonedx", "csaf", "openvex"],
         help="specify type of vulnerability exchange (vex) to generate (default: cyclonedx)",
     )
@@ -380,6 +375,19 @@ def main(argv=None):
         default="",
         help="Vendor/Supplier of Product",
     )
+    output_group.add_argument(
+        "-rr",
+        "--revision-reason",
+        action="store",
+        default="",
+        help="a reason for the update to the vex document should be specified in double quotes",
+    )
+    output_group.add_argument(
+        "--filter-triage",
+        action="store_true",
+        default=False,
+        help="Filter cves based on triage data from Vex file",
+    )
     parser.add_argument(
         "-e",
         "--exclude",
@@ -917,6 +925,7 @@ def main(argv=None):
         and not args["package_list"]
         and not args["merge"]
         and not args["sbom_file"]
+        and not args["vex_file"]
     ):
         parser.print_usage()
         with ErrorHandler(logger=LOGGER, mode=ErrorMode.NoTrace):
@@ -1009,7 +1018,7 @@ def main(argv=None):
         triage_data: TriageData
         total_files: int = 0
         parsed_data: dict[ProductInfo, TriageData] = {}
-
+        vex_product_info: dict[str, str] = {}
         # Package List parsing
         if args["package_list"]:
             sbom_root = args["package_list"]
@@ -1021,18 +1030,6 @@ def main(argv=None):
                 LOGGER.debug(f"{product_info}, {triage_data}")
                 cve_scanner.get_cves(product_info, triage_data)
 
-        if args["triage_input_file"]:
-            input_engine = InputEngine(
-                args["triage_input_file"],
-                logger=LOGGER,
-                error_mode=error_mode,
-                filetype="vex",
-            )
-            parsed_data = input_engine.parse_input()
-            for product_info, triage_data in parsed_data.items():
-                LOGGER.debug(f"{product_info}, {triage_data}")
-                cve_scanner.get_cves(product_info, triage_data)
-
         if args["input_file"]:
             input_engine = InputEngine(
                 args["input_file"], logger=LOGGER, error_mode=error_mode
@@ -1092,6 +1089,41 @@ def main(argv=None):
                 LOGGER.debug(f"{product_info}, {triage_data}")
                 cve_scanner.get_cves(product_info, triage_data)
 
+        if args["vex_file"]:
+            # for now use cyclonedx as auto detection is not implemented in latest pypi package of lib4vex
+            vexdata = VEXParse(
+                filename=args["vex_file"],
+                vextype="cyclonedx",
+                logger=LOGGER,
+            )
+            parsed_vex_data = vexdata.parse_vex()
+            vex_product_info = vexdata.vex_product_info
+            if not parsed_data:
+                # assume the vex file being scanned is a standalone file
+                args["filter_triage"] = False
+                parsed_data = parsed_vex_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)
+            else:
+                LOGGER.info(
+                    f"VEX file {args['vex_file']} is not a standalone file and will be used as a triage file"
+                )
+                # need to do validation on the sbom part
+                # need to implement is_linked() function which will check the linkage.
+                if args["sbom_file"]:
+                    LOGGER.warning(
+                        f"SBOM file: {args['sbom_file']} is not linked to VEX file: {args['vex_file']}."
+                    )
+                for product_info, triage_data in parsed_vex_data.items():
+                    LOGGER.debug(f"{product_info}, {triage_data}")
+                    if product_info in parsed_data:
+                        cve_scanner.get_cves(product_info, triage_data)
+                    else:
+                        LOGGER.info(
+                            f"Product: {product_info.product} with Version: {product_info.version} not found in Parsed Data, is valid vex file being used?"
+                        )
+
         LOGGER.info("Overall CVE summary: ")
         LOGGER.info(
             f"There are {cve_scanner.products_with_cve} products with known CVEs detected"
@@ -1105,19 +1137,29 @@ def main(argv=None):
                 )
             )
             LOGGER.info(f"Known CVEs in {affected_string}:")
-        vex_product_info: dict[str, str] = {}
-        if args["vex_output"]:
+        if args["vex_type"] or args["vex_output"]:
+            # If vex_type is provided, then use it, else use cyclonedx as default vex_output should be provide in this case
+            # If vex_output is provided, then use it, else use product, release and vendor to generate the vex file.
+            if args["vex_output"] and not args["vex_type"]:
+                # default vex_type is cyclonedx
+                args["vex_type"] = "cyclonedx"
             if args["product"] and args["release"] and args["vendor"]:
                 vex_product_info = {
                     "product": args["product"],
                     "release": args["release"],
                     "vendor": args["vendor"],
+                    "revision_reason": args["revision_reason"],
                 }
+            elif args["vex_file"]:
+                vex_product_info["revision_reason"] = args["revision_reason"]
             else:
                 LOGGER.error(
                     "Please provide --product, --release and --vendor for VEX generation"
                 )
                 return ERROR_CODES[InsufficientArgs]
+
+        if args["vex_file"] and args["filter_triage"]:
+            cve_scanner.filter_triage_data()
         # Creates an Object for OutputEngine
         output = OutputEngine(
             all_cve_data=cve_scanner.all_cve_data,

cve_bin_tool/cve_scanner.py

@@ -82,12 +82,42 @@ def get_cves(self, product_info: ProductInfo, triage_data: TriageData):
             return
 
         if product_info in self.all_cve_data:
-            # If product_info already in all_cve_data no need to fetch cves from database again
-            # We just need to update paths.
+            # If product_info already in all_cve_data, no need to fetch CVEs from the database again.
+            # We just need to update paths and triage data.
             self.logger.debug(
-                f"{product_info} already processed. Update path {triage_data['paths']}"
+                f"{product_info} already processed. Update paths {triage_data['paths']}"
             )
-            # self.products_with_cve += 1
+
+            # Update the triage data
+            cve_data = self.all_cve_data[product_info]["cves"]
+            new_cve_data = []
+
+            for cve in cve_data:
+                cve_number = cve.cve_number
+                if cve_number in triage_data:
+                    for key in [
+                        "remarks",
+                        "comments",
+                        "response",
+                        "justification",
+                        "severity",
+                    ]:
+                        data = triage_data[cve_number].get(key)
+                        if data:
+                            if (
+                                key == "severity"
+                                and self.check_exploits
+                                and cve_number in self.exploits_list
+                            ):
+                                data += "-EXPLOIT"
+
+                            self.logger.debug(f"Setting field {key} to: {data}")
+                            cve = cve._replace(**{key: data})
+                new_cve_data.append(cve)
+
+            self.all_cve_data[product_info]["cves"] = new_cve_data
+
+            # Update paths
             self.all_cve_data[product_info]["paths"] |= set(triage_data["paths"])
             return
 
@@ -350,6 +380,43 @@ def get_cves(self, product_info: ProductInfo, triage_data: TriageData):
         if product_info not in self.all_product_data:
             self.all_product_data[product_info] = len(cves)
 
+    def filter_triage_data(self):
+        """
+        Filter out triage data that is not relevant to the CVEs found,
+        specifically those marked as NotAffected or FalsePositives.
+        """
+        to_delete: List[ProductInfo] = []
+
+        for product_info, cve_data in self.all_cve_data.items():
+            original_cves = cve_data["cves"]
+            filtered_cves = []
+            filtered_out_cves = []
+
+            for cve in original_cves:
+                if cve.remarks not in {Remarks.NotAffected, Remarks.FalsePositive}:
+                    filtered_cves.append(cve)
+                else:
+                    filtered_out_cves.append(cve)
+
+            for cve in filtered_out_cves:
+                self.logger.info(
+                    f"Filtered CVE: {cve.cve_number} for Product: {product_info.product}"
+                )
+
+            if filtered_cves:
+                cve_data["cves"] = filtered_cves
+                self.logger.debug(
+                    f"Filtered triage data for {product_info.product}: {[cve.cve_number for cve in filtered_cves]}"
+                )
+            else:
+                to_delete.append(product_info)
+
+        for product_info in to_delete:
+            del self.all_cve_data[product_info]
+            self.logger.debug(
+                f"Removed product info for {product_info.product} due to no relevant CVEs"
+            )
+
     def affected(self):
         """Returns list of vendor.product and version tuples identified from
         scan"""