Bug Fix: improve Kiuwan SCA parser to support multi component findings (#12753)

* fix(Kiuwan Sca Scan): improve Kiuwan SCA parser to support multi-component findings

* test: update unit tests according to correct logic

* refactor: use better unique id from tool

* fix unit tests

* refactor: back to using unique id from tool like supposed to be

* fix: lint

* fix: tests

* refactor: update logic to set finding title

* refactor: use debug instead of warning for log output

* refactor: use vuln_id_from_tool instead of unique_id_from_tool

Author: mwager

dojo/tools/kiuwan_sca/parser.py

@@ -1,8 +1,11 @@
 import hashlib
 import json
+import logging
 
 from dojo.models import Finding
 
+logger = logging.getLogger(__name__)
+
 __author__ = "mwager"
 
 
@@ -37,50 +40,59 @@ def get_findings(self, filename, test):
             if row["muted"] is True:
                 continue
 
-            finding = Finding(test=test)
-            finding.unique_id_from_tool = row["id"]
-            finding.cve = row["cve"]
-            finding.description = row["description"]
-            finding.severity = self.SEVERITY[row["securityRisk"]]
-
-            if "components" in row and len(row["components"]) > 0:
-                finding.component_name = row["components"][0]["artifact"]
-                finding.component_version = row["components"][0]["version"]
-                finding.title = finding.component_name + " v" + str(finding.component_version)
-
-            if not finding.title:
-                finding.title = row["cve"]
-
-            if "cwe" in row and "CWE-" in row["cwe"]:
-                finding.cwe = int(row["cwe"].replace("CWE-", ""))
-
-            if "epss_score" in row:
-                finding.epss_score = row["epss_score"]
-            if "epss_percentile" in row:
-                finding.epss_percentile = row["epss_percentile"]
-
-            if "cVSSv3BaseScore" in row:
-                finding.cvssv3_score = float(row["cVSSv3BaseScore"])
-
-            finding.references = "See Kiuwan Web UI"
-            finding.mitigation = "See Kiuwan Web UI"
-            finding.static_finding = True
-
-            key = hashlib.sha256(
-                (
-                    finding.description
-                    + "|"
-                    + finding.severity
-                    + "|"
-                    + finding.component_name
-                    + "|"
-                    + finding.component_version
-                    + "|"
-                    + str(finding.cwe)
-                ).encode("utf-8"),
-            ).hexdigest()
-
-            if key not in dupes:
-                dupes[key] = finding
+            components = row.get("components", [])
+            if not components:
+                logger.debug("Insights Finding from Kiuwan does not have a related component - Skipping.")
+                continue
+
+            # We want one unique finding in DD for each component affected:
+            for component in components:
+                finding = Finding(test=test)
+                finding.vuln_id_from_tool = str(row["id"])
+                finding.cve = row["cve"]
+                finding.description = row["description"]
+                finding.severity = self.SEVERITY[row["securityRisk"]]
+
+                if "artifact" in component:
+                    finding.component_name = component["artifact"]
+                if "version" in component:
+                    finding.component_version = component["version"]
+
+                if finding.component_name and finding.component_version:
+                    finding.title = f"{finding.component_name} v{finding.component_version}"
+                else:
+                    finding.title = finding.cve or "Unnamed Finding"
+
+                if "cwe" in row and "CWE-" in row["cwe"]:
+                    finding.cwe = int(row["cwe"].replace("CWE-", ""))
+
+                if "epss_score" in row:
+                    finding.epss_score = row["epss_score"]
+                if "epss_percentile" in row:
+                    finding.epss_percentile = row["epss_percentile"]
+
+                if "cVSSv3BaseScore" in row:
+                    finding.cvssv3_score = float(row["cVSSv3BaseScore"])
+
+                finding.references = "See Kiuwan Web UI"
+                finding.mitigation = "See Kiuwan Web UI"
+                finding.static_finding = True
+
+                key = hashlib.sha256(
+                    (
+                        finding.description
+                        + "|"
+                        + finding.severity
+                        + "|"
+                        + finding.component_name
+                        + "|"
+                        + finding.component_version
+                        + "|"
+                        + str(finding.cwe or "")
+                    ).encode("utf-8"),
+                ).hexdigest()
+
+                if key not in dupes:
+                    dupes[key] = finding
 
         return list(dupes.values())

unittests/tools/test_kiuwan_sca_parser.py

@@ -15,15 +15,15 @@ def test_parse_file_with_two_vuln_has_two_findings(self):
         with (get_unit_tests_scans_path("kiuwan_sca") / "kiuwan_sca_two_vuln.json").open(encoding="utf-8") as testfile:
             parser = KiuwanSCAParser()
             findings = parser.get_findings(testfile, Test())
-            # file contains 3, but we only get 2 as "muted" ones are ignored:
-            self.assertEqual(2, len(findings))
+            # file contains 3 cves, one of them is muted. so we have 2 cves with a total of 5 components
+            self.assertEqual(5, len(findings))
 
     def test_parse_file_with_multiple_vuln_has_multiple_finding(self):
         with (get_unit_tests_scans_path("kiuwan_sca") / "kiuwan_sca_many_vuln.json").open(encoding="utf-8") as testfile:
             parser = KiuwanSCAParser()
             findings = parser.get_findings(testfile, Test())
-            # also tests deduplication as there are 28 findings in the file:
-            self.assertEqual(27, len(findings))
+            # also tests deduplication as there are 28 cves in the file (but some including >1 components!):
+            self.assertEqual(45, len(findings))
 
     def test_correct_mapping(self):
         with (get_unit_tests_scans_path("kiuwan_sca") / "kiuwan_sca_two_vuln.json").open(encoding="utf-8") as testfile:
@@ -37,7 +37,7 @@ def test_correct_mapping(self):
             self.assertEqual(finding1.component_name, "org.apache.cxf:cxf-rt-ws-policy")
             self.assertEqual(finding1.component_version, "3.3.5")
             self.assertEqual(finding1.cwe, 835)
-            self.assertEqual(finding1.unique_id_from_tool, 158713)
+            self.assertEqual(finding1.vuln_id_from_tool, "158713")
             self.assertEqual(finding1.cvssv3_score, 7.5)
             self.assertEqual(finding1.epss_score, 0.1)
             self.assertEqual(finding1.epss_percentile, 0.2)