Feature request 91 (#115)

* FR-91: Add cli arg only fixable vulnerability; use the variable in get_vuln_counts

* Revert "FR-91: Add cli arg only fixable vulnerability; use the variable in get_vuln_counts"

This reverts commit bc532d4.

* FR-91: Add cli arg only fixable vulnerability; use the variable in get_vuln_counts

* FR-91: Fix unit tests

* FR-91: Fix typo in unit tests

* Revert "FR-91: Fix typo in unit tests"

This reverts commit e645542.

* Revert "FR-91: Fix unit tests"

This reverts commit f9157c9.

* Revert "FR-91: Add cli arg only fixable vulnerability; use the variable in get_vuln_counts"

This reverts commit 812c685.

* FR-91: Change orchestrator to only find fixed vulnerabilities if flag show-only-fixed-vulnerabilities is present

* FR-91: Fixed missing variable

* FR-91: Fixed typo

* FR-91: Fixed typo

* FR-91: Another fix

* FR-91: Another fix

* FR-91: Another fix

* FR-91: Another fix

* FR-91: Another fix

* FR-91: Another fix

* FR-91: Another fix

* Add unit test for get_vuln_count

* Fix unit test for get_vuln_count

---------

Co-authored-by: Maria Carolina Conceição <carolina.bento@floy.com>

Author: CarolMebiom

entrypoint/entrypoint/cli.py

@@ -50,6 +50,7 @@ def init(sys_argv=None) -> argparse.Namespace:
                         help="Specifies one or more files and/or directories that should NOT be inventoried.")
     parser.add_argument("--timeout", type=str, default="600",
                         help="The amount of time in seconds that inspector-sbomgne will run. When this timeout is exceeded, sbomgen will gracefully conclude and present any findings discovered up to that point.")
+    parser.add_argument("--show-only-fixed-vulnerabilities", action="store_true", help="If set, this program will only show fixed vulnerabilities in the GitHub Actions job summary page.")
 
     parser.add_argument("--platform", type=str,
                         help="Specifies the OS and CPU arch of the container image you wish to scan. Valid inputs are "

entrypoint/entrypoint/orchestrator.py

@@ -166,7 +166,7 @@ def invoke_sbomgen(args) -> int:
 
     elif "archive" in args.artifact_type.lower():
         args.artifact_type = "archive"
-        path_arg = "--path"
+        path_arg = "--path"        
 
     else:
         logging.error(
@@ -233,18 +233,27 @@ def invoke_inspector_scan(src_sbom, dst_scan):
 
 
 def get_scan_result(args) -> tuple[bool, exporter.InspectorScanResult]:
-    succeeded, criticals, highs, mediums, lows, others = get_vuln_counts(args.out_scan)
-    if succeeded is False:
-        return False, None
+    if args.show_only_fixed_vulnerabilities == True:
+        succeeded, criticals, highs, mediums, lows, others = get_fixed_vuln_counts(args.out_scan)
+        if succeeded is False:
+            return False, None
+    else:
+        succeeded, criticals, highs, mediums, lows, others = get_vuln_counts(args.out_scan)
+        if succeeded is False:
+            return False, None
 
     try:
         with open(args.out_scan, "r") as f:
             inspector_scan = json.load(f)
             vulns = pkg_vuln.parse_inspector_scan_result(inspector_scan)
+            if args.show_only_fixed_vulnerabilities:
+                for vuln in vulns:
+                    if vuln.fixed_ver == "null":
+                        vulns.remove(vuln)
     except Exception as e:
         logging.error(e)
         return False, None
-
+    
     scan_result = exporter.InspectorScanResult(
         vulnerabilities=vulns,
         artifact_name=args.artifact_path,
@@ -270,7 +279,7 @@ def set_github_actions_output(key, value):
     return
 
 
-def get_vuln_counts(inspector_scan_path: str):
+def get_vuln_counts(inspector_scan_path: str) -> tuple[bool, int, int, int, int, int]:
     # vuln severities
     criticals = 0
     highs = 0
@@ -331,6 +340,78 @@ def get_vuln_counts(inspector_scan_path: str):
 
     return True, criticals, highs, mediums, lows, others
 
+def get_fixed_vuln_counts(inspector_scan_path: str) -> tuple[bool, int, int, int, int, int]:
+    criticals_fixed = 0
+    highs_fixed = 0
+    mediums_fixed = 0
+    lows_fixed = 0
+    others_fixed = 0
+
+    scan_contents = ""
+    try:
+        with open(inspector_scan_path, 'r') as f:
+            scan_contents = json.load(f)
+    except Exception as e:
+        logging.error(e)
+        return False, criticals, highs, mediums, lows, others
+
+    # find the sbom->metadata->properties object
+    scan_contents = scan_contents.get("sbom")
+    if scan_contents is None:
+        logging.error(
+            f"expected Inspector scan results with 'sbom' as root object, but it was not found in file {inspector_scan_path}")
+        return False, criticals, highs, mediums, lows, others
+
+    vulnerabilities = scan_contents.get("vulnerabilities")
+
+    if vulnerabilities is None:
+        # no vulnerabilities found
+        return True, criticals_fixed, highs_fixed, mediums_fixed, lows_fixed, others_fixed
+    
+    for vuln in vulnerabilities:
+        props = vuln.get("properties")
+        if props is None:
+            logging.error(f"expected vulnerability with 'properties' key but none was found in file {inspector_scan_path}")
+            continue
+
+        for prop in props:
+            name = prop.get("name")
+            if name is None:
+                logging.error(f"expected vulnerability with 'name' key but none was found in file {inspector_scan_path}")
+                continue
+            if "amazon:inspector:sbom_scanner:fixed_version:comp-" in name:
+                rating = vuln.get("ratings")
+                if rating is None:
+                    logging.error(f"expected vulnerability with 'rating' key but none was found in file {inspector_scan_path}")
+                    continue
+
+                previous_score = 0
+                previous_severity = ""
+                for each_rating in rating:
+                    severity = each_rating.get("severity")
+                    if severity is None:
+                        logging.error(f"expected vulnerability with 'severity' key but none was found in file {inspector_scan_path}")
+                        continue
+                    score = each_rating.get("score")
+                    if score is None:
+                        logging.error(f"expected vulnerability with 'score' key but none was found in file {inspector_scan_path}")
+                        continue
+                    if previous_score < score:
+                        previous_score = score
+                        previous_severity = severity
+                
+                if previous_severity == "none":
+                    others_fixed += 1
+                elif previous_severity == "critical":
+                    criticals_fixed += 1
+                elif previous_severity == "high":
+                    highs_fixed += 1
+                elif previous_severity == "medium":
+                    mediums_fixed += 1
+                elif previous_severity == "low":
+                    lows_fixed += 1
+    
+    return True, criticals_fixed, highs_fixed, mediums_fixed, lows_fixed, others_fixed
 
 def install_sbomgen(args):
     os_name = platform.system()

entrypoint/tests/test_orchestrator.py

@@ -125,6 +125,7 @@ def test_system_against_dockerfile_findings(self):
                     "out_scan_markdown",
                     "out_dockerfile_scan_csv",
                     "out_dockerfile_scan_md",
+                    "show_only_fixed_vulnerabilities",
                 ],
             )
             args = ArgMock(
@@ -135,6 +136,7 @@ def test_system_against_dockerfile_findings(self):
                 out_scan_markdown="/tmp/out_scan.md",
                 out_dockerfile_scan_csv="/tmp/out_dockerfile_scan.csv",
                 out_dockerfile_scan_md="/tmp/out_dockerfile_scan.md",
+                show_only_fixed_vulnerabilities=False,
             )
 
             succeeded, scan_result = orchestrator.get_scan_result(args)
@@ -201,6 +203,36 @@ def test_is_valid_container_platform(self):
         for each_test in test_cases:
             result = orchestrator.is_valid_container_platform(each_test["input"])
             self.assertEqual(result, each_test["expected"])
+            
+    def test_get_fixed_vuln_counts(self):
+        # verify we can successfully parse all known-valid Inspector scans
+        test_dir = "tests/test_data/scans/"
+        file_list = os.listdir(test_dir)
+        for file in file_list:
+            path = os.path.join(test_dir, file)
+            succeeded, _, _, _, _, _ = orchestrator.get_fixed_vuln_counts(path)
+            self.assertTrue(succeeded)
+
+        # verify our fixed vulnerability counts are correct for alpine:3.18.2.json.scan
+        succeeded, criticals_fixed, highs_fixed, mediums_fixed, lows_fixed, others_fixed = orchestrator.get_fixed_vuln_counts(
+            "tests/test_data/scans/alpine:3.18.2.json.scan"
+        )
+        self.assertTrue(succeeded)
+        self.assertEqual(criticals, 1)
+        self.assertEqual(highs, 1)
+        self.assertEqual(mediums, 7)
+        self.assertEqual(lows, 0)
+        self.assertEqual(others, 3)
+
+        succeeded, criticals_fixed, highs_fixed, mediums_fixed, lows_fixed, others_fixed = orchestrator.get_fixed_vuln_counts(
+            "tests/test_data/scans/debian:latest.json.scan"
+        )
+        self.assertTrue(succeeded)
+        self.assertEqual(criticals_fixed, 0)
+        self.assertEqual(highs_fixed, 0)
+        self.assertEqual(mediums_fixed, 0)
+        self.assertEqual(lows_fixed, 0)
+        self.assertEqual(others_fixed, 6)
 
 if __name__ == "__main__":
     unittest.main()