💄 🪲 Fix Aqua parser severity justification (#12192)

* 💄 🪲 Fix Aqua parser severity justification

* update

* update

Author: manuel-sommer

dojo/tools/aqua/parser.py

@@ -92,6 +92,7 @@ def get_dedupe_fields(self) -> list[str]:
     #         "title",
     #         "description",
     #     ]
+
     def get_scan_types(self):
         return ["Aqua Scan"]
 
@@ -103,9 +104,6 @@ def get_description_for_scan_types(self, scan_type):
 
     def get_findings(self, json_output, test):
         tree = json.load(json_output)
-        return self.get_items(tree, test)
-
-    def get_items(self, tree, test):
         self.items = {}
         if isinstance(tree, list):  # Aqua Scan Report coming from Azure Devops jobs (Windows based image)
             vulnerabilitytree = tree[0]["results"]["resources"] if tree else []
@@ -115,11 +113,15 @@ def get_items(self, tree, test):
             self.vulnerability_tree(vulnerabilitytree, test)
         elif "result" in tree:     # Aqua Scan Report from apiv2
             resulttree = tree["result"]
-            self.result_tree(resulttree, test)
+            for vuln in resulttree:
+                resource = vuln.get("resource")
+                item = self.get_item(resource, vuln, test)
+                unique_key = resource.get("cpe") + vuln.get("name", "None") + resource.get("path", "None")
+                self.items[unique_key] = item
         elif "cves" in tree:       # Aqua Scan Report from apiv1
             for cve in tree["cves"]:
                 unique_key = cve.get("file") + cve.get("name")
-                self.items[unique_key] = get_item_v2(cve, test)
+                self.items[unique_key] = self.get_item_v2(cve, test)
         return list(self.items.values())
 
     def vulnerability_tree(self, vulnerabilitytree, test):
@@ -130,189 +132,172 @@ def vulnerability_tree(self, vulnerabilitytree, test):
             if vulnerabilities is None:
                 vulnerabilities = []
             for vuln in vulnerabilities:
-                item = get_item(resource, vuln, test)
+                item = self.get_item(resource, vuln, test)
                 unique_key = resource.get("cpe") + vuln.get("name", "None") + resource.get("path", "None")
                 self.items[unique_key] = item
             if sensitive_items is None:
                 sensitive_items = []
             for sensitive_item in sensitive_items:
-                item = get_item_sensitive_data(resource, sensitive_item, test)
+                item = self.get_item_sensitive_data(resource, sensitive_item, test)
                 unique_key = resource.get("cpe") + resource.get("path", "None") + str(sensitive_item)
                 self.items[unique_key] = item
 
-    def result_tree(self, resulttree, test):
-        for vuln in resulttree:
-            resource = vuln.get("resource")
-            item = get_item(resource, vuln, test)
-            unique_key = resource.get("cpe") + vuln.get("name", "None") + resource.get("path", "None")
-            self.items[unique_key] = item
-
-
-def get_item(resource, vuln, test):
-    resource_name = resource.get("name", resource.get("path"))
-    resource_version = resource.get("version", "No version")
-    vulnerability_id = vuln.get("name", "No CVE")
-    fix_version = vuln.get("fix_version", "None")
-    description = vuln.get("description", "No description.") + "\n"
-    if resource.get("path"):
-        description += "**Path:** " + resource.get("path") + "\n"
-    cvssv3 = None
-
-    url = ""
-    if "nvd_url" in vuln:
-        url += "\n{}".format(vuln.get("nvd_url"))
-    if "vendor_url" in vuln:
-        url += "\n{}".format(vuln.get("vendor_url"))
-
-    # Take in order of preference (most prio at the bottom of ifs), and put
-    # everything in severity justification anyways.
-    score = 0
-    severity_justification = ""
-    used_for_classification = ""
-    if "aqua_severity" in vuln:
-        score = vuln.get("aqua_severity")
-        severity = aqua_severity_of(score)
-        used_for_classification = (
-            f"Aqua security score ({score}) used for classification.\n"
-        )
-        severity_justification = vuln.get("aqua_severity_classification")
-        if "nvd_score_v3" in vuln:
-            cvssv3 = vuln.get("nvd_vectors_v3")
-    else:
+    def get_item(self, resource, vuln, test):
+        resource_name = resource.get("name", resource.get("path"))
+        resource_version = resource.get("version", "No version")
+        vulnerability_id = vuln.get("name", "No CVE")
+        fix_version = vuln.get("fix_version", "None")
+        description = vuln.get("description", "No description.") + "\n"
+        if resource.get("path"):
+            description += "**Path:** " + resource.get("path") + "\n"
+        cvssv3 = None
+        url = ""
+        if "nvd_url" in vuln:
+            url += "\n{}".format(vuln.get("nvd_url"))
+        if "vendor_url" in vuln:
+            url += "\n{}".format(vuln.get("vendor_url"))
+        # Take in order of preference (most prio at the bottom of ifs), and put
+        # everything in severity justification anyways.
+        score = None
+        severity_justification = ""
+        used_for_classification = ""
+        if "aqua_severity" in vuln:
+            if score is None:
+                score = vuln.get("aqua_severity")
+                used_for_classification = (
+                    f"Aqua severity ({score}) used for classification.\n"
+                )
+            severity_justification += "\nAqua severity classification: {}".format(vuln.get("aqua_severity_classification"))
+            severity_justification += "\nAqua scoring system: {}".format(vuln.get("aqua_scoring_system"))
+            if "nvd_score_v3" in vuln:
+                cvssv3 = vuln.get("nvd_vectors_v3")
         if "aqua_score" in vuln:
-            score = vuln.get("aqua_score")
-            used_for_classification = (
-                f"Aqua score ({score}) used for classification.\n"
-            )
-        elif "vendor_score" in vuln:
-            score = vuln.get("vendor_score")
-            used_for_classification = (
-                f"Vendor score ({score}) used for classification.\n"
-            )
-        elif "nvd_score_v3" in vuln:
-            score = vuln.get("nvd_score_v3")
-            used_for_classification = (
-                f"NVD score v3 ({score}) used for classification.\n"
-            )
-            severity_justification += "\nNVD v3 vectors: {}".format(
-                vuln.get("nvd_vectors_v3"),
-            )
+            if score is None:
+                score = vuln.get("aqua_score")
+                used_for_classification = (
+                    f"Aqua score ({score}) used for classification.\n"
+                )
+            severity_justification += "\nAqua score: {}".format(vuln.get("aqua_score"))
+        if "vendor_score" in vuln:
+            if score is None:
+                score = vuln.get("vendor_score")
+                used_for_classification = (
+                    f"Vendor score ({score}) used for classification.\n"
+                )
+            severity_justification += "\nVendor score: {}".format(vuln.get("vendor_score"))
+        if "nvd_score_v3" in vuln:
+            if score is None:
+                score = vuln.get("nvd_score_v3")
+                used_for_classification = (
+                    f"NVD score v3 ({score}) used for classification.\n"
+                )
+            severity_justification += "\nNVD v3 vectors: {}".format(vuln.get("nvd_vectors_v3"))
             # Add the CVSS3 to Finding
             cvssv3 = vuln.get("nvd_vectors_v3")
-        elif "nvd_score" in vuln:
-            score = vuln.get("nvd_score")
-            used_for_classification = (
-                f"NVD score v2 ({score}) used for classification.\n"
-            )
-            severity_justification += "\nNVD v2 vectors: {}".format(
-                vuln.get("nvd_vectors"),
-            )
-        severity = severity_of(score)
+        if "nvd_score" in vuln:
+            if score is None:
+                score = vuln.get("nvd_score")
+                used_for_classification = (
+                    f"NVD score v2 ({score}) used for classification.\n"
+                )
+            severity_justification += "\nNVD v2 vectors: {}".format(vuln.get("nvd_vectors"))
         severity_justification += f"\n{used_for_classification}"
+        severity = self.severity_of(score)
+        finding = Finding(
+            title=vulnerability_id
+            + " - "
+            + resource_name
+            + " ("
+            + resource_version
+            + ") ",
+            test=test,
+            severity=severity,
+            severity_justification=severity_justification,
+            cwe=0,
+            cvssv3=cvssv3,
+            description=description.strip(),
+            mitigation=fix_version,
+            references=url,
+            component_name=resource.get("name"),
+            component_version=resource.get("version"),
+            impact=severity,
+        )
+        if vulnerability_id != "No CVE":
+            finding.unsaved_vulnerability_ids = [vulnerability_id]
+        if vuln.get("epss_score"):
+            finding.epss_score = vuln.get("epss_score")
+        if vuln.get("epss_percentile"):
+            finding.epss_percentile = vuln.get("epss_percentile")
+        return finding
 
-    finding = Finding(
-        title=vulnerability_id
-        + " - "
-        + resource_name
-        + " ("
-        + resource_version
-        + ") ",
-        test=test,
-        severity=severity,
-        severity_justification=severity_justification,
-        cwe=0,
-        cvssv3=cvssv3,
-        description=description.strip(),
-        mitigation=fix_version,
-        references=url,
-        component_name=resource.get("name"),
-        component_version=resource.get("version"),
-        impact=severity,
-    )
-    if vulnerability_id != "No CVE":
-        finding.unsaved_vulnerability_ids = [vulnerability_id]
-    if vuln.get("epss_score"):
-        finding.epss_score = vuln.get("epss_score")
-    if vuln.get("epss_percentile"):
-        finding.epss_percentile = vuln.get("epss_percentile")
-    return finding
-
-
-def get_item_v2(item, test):
-    vulnerability_id = item["name"]
-    file_path = item["file"]
-    url = item.get("url")
-    severity = severity_of(float(item["score"]))
-    description = item.get("description")
-    solution = item.get("solution")
-    fix_version = item.get("fix_version")
-    if solution:
-        mitigation = solution
-    elif fix_version:
-        mitigation = "Upgrade to " + str(fix_version)
-    else:
-        mitigation = "No known mitigation"
-
-    finding = Finding(
-        title=str(vulnerability_id) + ": " + str(file_path),
-        description=description,
-        url=url,
-        cwe=0,
-        test=test,
-        severity=severity,
-        impact=severity,
-        mitigation=mitigation,
-    )
-    finding.unsaved_vulnerability_ids = [vulnerability_id]
-
-    return finding
-
-
-def get_item_sensitive_data(resource, sensitive_item, test):
-    resource_name = resource.get("name", "None")
-    resource_path = resource.get("path", "None")
-    vulnerability_id = resource_name
-    description = "**Senstive Item:** " + sensitive_item + "\n"
-    description += "**Layer:** " + resource.get("layer", "None") + "\n"
-    description += "**Layer_Digest:** " + resource.get("layer_digest", "None") + "\n"
-    description += "**Path:** " + resource.get("path", "None") + "\n"
-    finding = Finding(
-        title=vulnerability_id
-        + " - "
-        + resource_name
-        + " ("
-        + resource_path
-        + ") ",
-        test=test,
-        severity="Info",
-        description=description.strip(),
-        component_name=resource.get("name"),
-    )
-    if vulnerability_id != "No CVE":
+    def get_item_v2(self, item, test):
+        vulnerability_id = item["name"]
+        file_path = item["file"]
+        url = item.get("url")
+        severity = self.severity_of(float(item["score"]))
+        description = item.get("description")
+        solution = item.get("solution")
+        fix_version = item.get("fix_version")
+        if solution:
+            mitigation = solution
+        elif fix_version:
+            mitigation = "Upgrade to " + str(fix_version)
+        else:
+            mitigation = "No known mitigation"
+        finding = Finding(
+            title=str(vulnerability_id) + ": " + str(file_path),
+            description=description,
+            url=url,
+            cwe=0,
+            test=test,
+            severity=severity,
+            impact=severity,
+            mitigation=mitigation,
+        )
         finding.unsaved_vulnerability_ids = [vulnerability_id]
+        return finding
 
-    return finding
-
-
-def aqua_severity_of(score):
-    if score == "high":
-        return "High"
-    if score == "medium":
-        return "Medium"
-    if score == "low":
-        return "Low"
-    if score == "negligible":
-        return "Info"
-    return "Critical"
-
+    def get_item_sensitive_data(self, resource, sensitive_item, test):
+        resource_name = resource.get("name", "None")
+        resource_path = resource.get("path", "None")
+        vulnerability_id = resource_name
+        description = "**Senstive Item:** " + sensitive_item + "\n"
+        description += "**Layer:** " + resource.get("layer", "None") + "\n"
+        description += "**Layer_Digest:** " + resource.get("layer_digest", "None") + "\n"
+        description += "**Path:** " + resource.get("path", "None") + "\n"
+        finding = Finding(
+            title=vulnerability_id
+            + " - "
+            + resource_name
+            + " ("
+            + resource_path
+            + ") ",
+            test=test,
+            severity="Info",
+            description=description.strip(),
+            component_name=resource.get("name"),
+        )
+        if vulnerability_id != "No CVE":
+            finding.unsaved_vulnerability_ids = [vulnerability_id]
+        return finding
 
-def severity_of(score):
-    if score == 0:
-        return "Info"
-    if score < 4:
-        return "Low"
-    if 4.0 < score < 7.0:
-        return "Medium"
-    if 7.0 < score < 9.0:
-        return "High"
-    return "Critical"
+    def severity_of(self, score):
+        if isinstance(score, str):
+            if score == "high":
+                return "High"
+            if score == "medium":
+                return "Medium"
+            if score == "low":
+                return "Low"
+            if score == "negligible":
+                return "Info"
+            return "Critical"
+        if score == 0:
+            return "Info"
+        if score < 4:
+            return "Low"
+        if 4.0 < score < 7.0:
+            return "Medium"
+        if 7.0 < score < 9.0:
+            return "High"
+        return "Critical"

unittests/tools/test_aqua_parser.py

@@ -28,6 +28,16 @@ def test_aqua_parser_has_one_finding(self):
             self.assertEqual("1.1.20-r4", finding.component_version)
             self.assertEqual(1, len(finding.unsaved_vulnerability_ids))
             self.assertEqual("CVE-2019-14697", finding.unsaved_vulnerability_ids[0])
+            finding_severity_justification = """
+Aqua severity classification: None
+Aqua scoring system: CVSS V2
+Aqua score: 7.5
+Vendor score: 7.5
+NVD v3 vectors: CVSS:3.0/AV:N/AC:L/PR:N/UI:N/S:U/C:H/I:H/A:H
+NVD v2 vectors: AV:N/AC:L/Au:N/C:P/I:P/A:P
+Aqua severity (high) used for classification.
+"""
+            self.assertEqual(finding_severity_justification, finding.severity_justification)
 
     def test_aqua_parser_has_many_findings(self):
         with open(get_unit_tests_scans_path("aqua") / "many_vulns.json", encoding="utf-8") as testfile: