feat: epss percentile filter (#3244)

Co-authored-by: Terri Oda <[email protected]>

README.md

@@ -255,6 +255,8 @@ Output:
                         specify multiple output formats by using comma (',') as a separator
                         note: don't use spaces between comma (',') and the output formats.
   <a href="https://github.com/intel/cve-bin-tool/blob/main/doc/MANUAL.md#-c-cvss---cvss-cvss">-c CVSS, --cvss CVSS</a>  minimum CVSS score (as integer in range 0 to 10) to report (default: 0)
+  <a>--epss-percentile</a>
+  minimum EPSS percentile of CVE range between 0 to 100 to report (default: 0)
   <a href="https://github.com/intel/cve-bin-tool/blob/main/doc/MANUAL.md#-s-lowmediumhighcritical---severity-lowmediumhighcritical">-S {low,medium,high,critical}, --severity {low,medium,high,critical}</a>
                         minimum CVE severity to report (default: low)
   --no-0-cve-report     only produce report when CVEs are found

cve_bin_tool/cli.py

@@ -266,6 +266,12 @@ def main(argv=None):
         help="minimum CVE severity to report (default: low)",
         default="low",
     )
+    output_group.add_argument(
+        "--epss-percentile",
+        action="store",
+        help="minimum epss percentile of CVE range between 0 to 100 to report (default: 0)",
+        default=0,
+    )
     output_group.add_argument(
         "--no-0-cve-report",
         action="store_true",
@@ -563,6 +569,10 @@ def main(argv=None):
     if int(args["cvss"]) > 0:
         score = int(args["cvss"])
 
+    epss_percentile = 0
+    if float(args["epss_percentile"]) > 0:
+        epss_percentile = float(args["epss_percentile"]) / 100
+
     config_generate = set(args["generate_config"].split(","))
     config_generate = [config_type.strip() for config_type in config_generate]
     for config_type in config_generate:
@@ -863,6 +873,7 @@ def main(argv=None):
 
     with CVEScanner(
         score=score,
+        epss_percentile=epss_percentile,
         check_exploits=args["exploits"],
         exploits_list=cvedb_orig.get_exploits_list(),
         disabled_sources=disabled_sources,

cve_bin_tool/cve_scanner.py

@@ -40,6 +40,7 @@ class CVEScanner:
     def __init__(
         self,
         score: int = 0,
+        epss_percentile: float = 0.0,
         logger: Logger = None,
         error_mode: ErrorMode = ErrorMode.TruncTrace,
         check_exploits: bool = False,
@@ -49,6 +50,7 @@ def __init__(
         self.logger = logger or LOGGER.getChild(self.__class__.__name__)
         self.error_mode = error_mode
         self.score = score
+        self.epss_percentile = epss_percentile
         self.products_with_cve = 0
         self.products_without_cve = 0
         self.all_cve_data = defaultdict(CVEData)
@@ -68,6 +70,8 @@ def get_cves(self, product_info: ProductInfo, triage_data: TriageData):
         # being reported
         if self.score > 10:
             return
+        if self.epss_percentile > 100:
+            return
 
         if product_info.vendor == "UNKNOWN":
             # Add product
@@ -257,16 +261,24 @@ def get_cves(self, product_info: ProductInfo, triage_data: TriageData):
                             row_dict["cvss_version"] or row["cvss_version"]
                         )
                         # executing query to get metric for CVE
-                        metric_result = self.metric((row["cve_number"],))
+                        metric_result = self.metric(
+                            (row["cve_number"],), self.epss_percentile
+                        )
                         # row_dict doesnt have metric as key. As it based on result from query on cve_severity table
                         # declaring row_dict[metric]
                         row_dict["metric"] = {}
-                        # # looping for result of query for metrics.
+                        # looping for result of query for metrics.
                         for key, value in metric_result.items():
                             row_dict["metric"][key] = [
                                 value[0],
                                 value[1],
                             ]
+                        # checking if epss percentile filter is applied
+                        if self.epss_percentile:
+                            # if epss filter is applied and condition is failed to satisfy row_dict["metric"] will be empty
+                            if not row_dict["metric"]:
+                                # continue to not include that particular cve
+                                continue
                         self.logger.debug(
                             f'metrics found in CVE {row_dict["cve_number"]}  is {row_dict["metric"]}'
                         )
@@ -358,7 +370,7 @@ def affected(self):
             for cve_data in self.all_cve_data
         )
 
-    def metric(self, cve_number):
+    def metric(self, cve_number, epss_percentile):
         """The query needs to be executed separately because if it is executed using the same cursor, the search stops.
         We need to create a separate connection and cursor for the query to be executed independently.
         Finally, the function should return a dictionary with the metrics of a given CVE.
@@ -376,6 +388,13 @@ def metric(self, cve_number):
         # looping for result of query for metrics.
         for result in metric_result:
             metric_name, metric_score, metric_field = result
+            # if metric is EPSS if metric field must represent EPSS percentile
+            if metric_name == "EPSS":
+                # comparing if EPSS percentile found in CVE is less then EPSS percentile return
+                if float(metric_field) < epss_percentile:
+                    cur.close()
+                    conn.close()
+                    return met
             met[metric_name] = [
                 metric_score,
                 metric_field,

doc/MANUAL.md

@@ -39,6 +39,7 @@
     - [--html-theme HTML_THEME](#--html-theme-html_theme)
     - [-f {csv,json,console,html}, --format {csv,json,console,html}](#-f-csvjsonconsolehtml---format-csvjsonconsolehtml)
     - [-c CVSS, --cvss CVSS](#-c-cvss---cvss-cvss)
+    - [--epss-percentile](#epss-percentile)
     - [-S {low,medium,high,critical}, --severity {low,medium,high,critical}](#-s-lowmediumhighcritical---severity-lowmediumhighcritical)
     - [-A \[\<distro_name\>-\<distro_version_name\>\], --available-fix \[\<distro_name\>-\<distro_version_name\>\]](#-a-distro_name-distro_version_name---available-fix-distro_name-distro_version_name)
     - [-b \[\<distro_name\>-\<distro_version_name\>\], --backport-fix \[\<distro_name\>-\<distro_version_name\>\]](#-b-distro_name-distro_version_name---backport-fix-distro_name-distro_version_name)
@@ -126,6 +127,8 @@ which is useful if you're trying the latest code from
                             specify multiple output formats by using comma (',') as a separator
                             note: don't use spaces between comma (',') and the output formats.
       -c CVSS, --cvss CVSS  minimum CVSS score (as integer in range 0 to 10) to report (default: 0)
+      --epss-percentile     minimum EPSS percentile of CVE range between 0 to 100 to report 
+                            (default: 0)
       -S {low,medium,high,critical}, --severity {low,medium,high,critical}
                             minimum CVE severity to report (default: low)
       --no-0-cve-report     only produce report when CVEs are found
@@ -930,6 +933,10 @@ Note: Please don't use spaces between comma (',') and the output formats.
 
 This option specifies the minimum CVSS score (as integer in range 0 to 10) of the CVE to report. The default value is 0 which results in all CVEs being reported.
 
+### --epss-percentile
+
+this option specifies the minimum EPSS percentile of CVE range between 0 to 100 to report. The default value is 0 which results in all CVEs being reported.
+
 ### -S {low,medium,high,critical}, --severity {low,medium,high,critical}
 
 This option specifies the minimum CVE severity to report. The default value is low which results in all CVEs being reported.

test/test_cli.py

@@ -484,6 +484,60 @@ def test_CVSS_score(self, capsys, caplog):
             my_test_filename_pathlib.unlink()
         caplog.clear()
 
+    def test_EPSS_percentile(self, capsys, caplog):
+        """scan with EPSS percentile to ensure only CVEs above score threshold are reported
+        Checks cannot placed on epss percentile value as the value changes everyday
+        """
+
+        my_test_filename = "epss_percentile.csv"
+        my_test_filename_pathlib = Path(my_test_filename)
+
+        # Check command line parameters. Less than 0 result in default behaviour.
+        if my_test_filename_pathlib.exists():
+            my_test_filename_pathlib.unlink()
+        with caplog.at_level(logging.DEBUG):
+            main(
+                [
+                    "cve-bin-tool",
+                    "-x",
+                    "--epss-percentile",
+                    "-1",
+                    "-f",
+                    "csv",
+                    "-o",
+                    my_test_filename,
+                    str(Path(self.tempdir) / CURL_7_20_0_RPM),
+                ]
+            )
+        # Verify that some CVEs with a severity of Medium are reported
+        # Checks cannot placed on epss percentile value as the value changes everyday.
+        assert self.check_string_in_file(my_test_filename, "MEDIUM")
+        caplog.clear()
+
+        # Check command line parameters. >10 results in no CVEs being reported (Maximum CVSS score is 10)
+        if my_test_filename_pathlib.exists():
+            my_test_filename_pathlib.unlink()
+        with caplog.at_level(logging.DEBUG):
+            main(
+                [
+                    "cve-bin-tool",
+                    "-x",
+                    "--epss-percentile",
+                    "110",
+                    "-f",
+                    "csv",
+                    "-o",
+                    my_test_filename,
+                    str(Path(self.tempdir) / CURL_7_20_0_RPM),
+                ]
+            )
+        # Verify that no CVEs are reported
+        with open(my_test_filename_pathlib) as fd:
+            assert not fd.read().split("\n")[1]
+        caplog.clear()
+        if my_test_filename_pathlib.exists():
+            my_test_filename_pathlib.unlink()
+
     @pytest.mark.skip(reason="Needs database rebuild. Temporary fix.")
     def test_SBOM(self, caplog):
         # check sbom file option