feat: adding EPSS probability filter (#3273)

fixes: #3243

README.md

@@ -256,7 +256,9 @@ Output:
                         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)
+  minimum EPSS percentile of CVE range between 0 to 100 to report (input value can also be floating point) (default: 0)
+  <a>--epss-probability</a>
+  minimum EPSS probability of CVE range between 0 to 100 to report (input value can also be floating point) (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

@@ -272,6 +272,12 @@ def main(argv=None):
         help="minimum epss percentile of CVE range between 0 to 100 to report (default: 0)",
         default=0,
     )
+    output_group.add_argument(
+        "--epss-probability",
+        action="store",
+        help="minimum epss probability of CVE range between 0 to 100 to report (default: 0)",
+        default=0,
+    )
     output_group.add_argument(
         "--no-0-cve-report",
         action="store_true",
@@ -573,8 +579,14 @@ def main(argv=None):
         score = int(args["cvss"])
 
     epss_percentile = 0
-    if float(args["epss_percentile"]) > 0:
+    if float(args["epss_percentile"]) > 0 or float(args["epss_percentile"]) < 100:
         epss_percentile = float(args["epss_percentile"]) / 100
+        LOGGER.debug(f"epss percentile stored {epss_percentile}")
+
+    epss_probability = 0
+    if float(args["epss_probability"]) > 0 or float(args["epss_probability"]) < 100:
+        epss_probability = float(args["epss_probability"]) / 100
+        LOGGER.debug(f"epss probability stored {epss_probability}")
 
     config_generate = set(args["generate_config"].split(","))
     config_generate = [config_type.strip() for config_type in config_generate]
@@ -877,6 +889,7 @@ def main(argv=None):
     with CVEScanner(
         score=score,
         epss_percentile=epss_percentile,
+        epss_probability=epss_probability,
         check_exploits=args["exploits"],
         exploits_list=cvedb_orig.get_exploits_list(),
         disabled_sources=disabled_sources,

cve_bin_tool/cve_scanner.py

@@ -41,6 +41,7 @@ def __init__(
         self,
         score: int = 0,
         epss_percentile: float = 0.0,
+        epss_probability: float = 0.0,
         logger: Logger = None,
         error_mode: ErrorMode = ErrorMode.TruncTrace,
         check_exploits: bool = False,
@@ -51,6 +52,7 @@ def __init__(
         self.error_mode = error_mode
         self.score = score
         self.epss_percentile = epss_percentile
+        self.epss_probability = epss_probability
         self.products_with_cve = 0
         self.products_without_cve = 0
         self.all_cve_data = defaultdict(CVEData)
@@ -68,9 +70,7 @@ def get_cves(self, product_info: ProductInfo, triage_data: TriageData):
 
         # Prevent any queries resulting in CVEs with UNKNOWN score value
         # being reported
-        if self.score > 10:
-            return
-        if self.epss_percentile > 100:
+        if self.score > 10 or self.epss_probability > 1.0 or self.epss_percentile > 1.0:
             return
 
         if product_info.vendor == "UNKNOWN":
@@ -262,7 +262,9 @@ def get_cves(self, product_info: ProductInfo, triage_data: TriageData):
                         )
                         # executing query to get metric for CVE
                         metric_result = self.metric(
-                            (row["cve_number"],), self.epss_percentile
+                            (row["cve_number"],),
+                            self.epss_percentile,
+                            self.epss_probability,
                         )
                         # row_dict doesnt have metric as key. As it based on result from query on cve_severity table
                         # declaring row_dict[metric]
@@ -274,7 +276,7 @@ def get_cves(self, product_info: ProductInfo, triage_data: TriageData):
                                 value[1],
                             ]
                         # checking if epss percentile filter is applied
-                        if self.epss_percentile:
+                        if self.epss_percentile > 0.0 or self.epss_probability > 0.0:
                             # 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
@@ -370,7 +372,7 @@ def affected(self):
             for cve_data in self.all_cve_data
         )
 
-    def metric(self, cve_number, epss_percentile):
+    def metric(self, cve_number, epss_percentile, epss_probability):
         """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.
@@ -391,15 +393,29 @@ def metric(self, cve_number, epss_percentile):
             # 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
+
+                # checks if both epss percentile and epss probaility are given. And if given they are greater than found in current CVE. if not it break loops and skips that CVE
+                if (
+                    epss_probability
+                    and epss_percentile
+                    and (
+                        float(metric_field) < float(epss_percentile)
+                        or float(metric_score) < float(epss_probability)
+                    )
+                ):
+                    break
+                # checks if only epss percentile is given and if given then it should be higher than found epss percentile in current CVE. if not it break loops and skips that CVE
+                elif epss_percentile and float(metric_field) < epss_percentile:
+                    break
+                # checks if only epss probability is given and if given then it should be higher than found epss probability in current CVE. if not it break loops and skips that CVE
+                elif epss_probability and float(metric_score) < epss_probability:
+                    break
+
+            self.logger.debug(f"metrics found in CVE {cve_number}  is {met}")
             met[metric_name] = [
                 metric_score,
                 metric_field,
             ]
-        self.logger.debug(f"metrics found in CVE {cve_number}  is {met}")
         cur.close()
         conn.close()
         return met

doc/MANUAL.md

@@ -40,6 +40,7 @@
     - [-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)
+    - [--epss-probability](#epss-probability)
     - [-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)
@@ -128,7 +129,9 @@ which is useful if you're trying the latest code from
                             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)
+                            (input value can also be floating point)(default: 0)
+      --epss-probability     minimum EPSS probability of CVE range between 0 to 100 to report 
+                            (input value can also be floating point)(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
@@ -949,7 +952,11 @@ This option specifies the minimum CVSS score (as integer in range 0 to 10) of th
 
 ### --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.
+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.
+
+### --epss-probability
+
+This option specifies the minimum EPSS probability of CVE range between o to 100 to report. The default value is 0 which result in all CVEs being reported.
 
 ### -S {low,medium,high,critical}, --severity {low,medium,high,critical}
 

test/test_cli.py

@@ -484,6 +484,58 @@ def test_CVSS_score(self, capsys, caplog):
             my_test_filename_pathlib.unlink()
         caplog.clear()
 
+    def test_EPSS_probability(self, capsys, caplog):
+        """scan with EPSS probability to ensure only CVEs above score threshold are reported
+        Checks cannot placed on epss probability value as the value changes everyday
+        """
+
+        my_test_filename = "epss_probability.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-probability",
+                    "-12",
+                    "-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 probability value as the value changes everyday.
+        assert self.check_string_in_file(my_test_filename, "MEDIUM")
+        caplog.clear()
+        if my_test_filename_pathlib.exists():
+            my_test_filename_pathlib.unlink()
+        with caplog.at_level(logging.DEBUG):
+            main(
+                [
+                    "cve-bin-tool",
+                    "-x",
+                    "--epss-probability",
+                    "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()
+
     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
@@ -514,7 +566,7 @@ def test_EPSS_percentile(self, capsys, caplog):
         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)
+        # Check command line parameters. >10 results in no CVEs being reported (Maximum EPSS percentile is 100)
         if my_test_filename_pathlib.exists():
             my_test_filename_pathlib.unlink()
         with caplog.at_level(logging.DEBUG):