✨ merge clair and clair klar together

docs/content/en/getting_started/upgrading/2.31.md

@@ -2,6 +2,14 @@
 title: 'Upgrading to DefectDojo Version 2.31.x'
 toc_hide: true
 weight: -20240102
-description: No special instructions.
+description: breaking change
 ---
-There are no special instructions for upgrading to 2.31.x. Check the [Release Notes](https://github.com/DefectDojo/django-DefectDojo/releases/tag/2.31.0) for the contents of the release.
+
+To reduce the number of parsers, multiple parsers were merged together:
+- OpenVAS XML and OpenVAS CSV were merged to OpenVAS Parser. There is a migration process built into the upgrade that will automatically convert exiting OpenVAS XML and OpenVAS CSV findings into OpenVAS Parser findings.
+- Clair Scan and Clair Klar Scan were merged to Clair Scan. There is a migration process built into the upgrade that will automatically convert exiting Clair Klar Scan findings to Clair Scan findings.
+
+**Breaking Change**
+ - If there is any use of the above mentioned parsers in automated fashion via the import and reimport API endpoints, the `scan-type` parameter needs to be updated accordingly.
+
+For all other changes, check the [Release Notes](https://github.com/DefectDojo/django-DefectDojo/releases/tag/2.31.0) for the contents of the release.

dojo/db_migrations/0197_parser_merge.py

@@ -0,0 +1,98 @@
+from django.db import migrations
+import logging
+
+
+logger = logging.getLogger(__name__)
+
+
+OPENVAS_REFERENCES = ['OpenVAS CSV', 'OpenVAS XML']
+CLAIRKLAR_REFERENCES = ['Clair Klar Scan']
+
+
+# update the test type object as well as the scan type name
+def update_openvas_test(test, openvas_test_type) -> None:
+    if test.test_type.name in OPENVAS_REFERENCES or test.scan_type in OPENVAS_REFERENCES:
+        test.test_type = openvas_test_type
+        test.scan_type = openvas_test_type.name
+        test.save()
+
+
+def update_clairklar_test(test, clairklar_test_type) -> None:
+    if test.test_type.name in CLAIRKLAR_REFERENCES or test.scan_type in CLAIRKLAR_REFERENCES:
+        test.test_type = clairklar_test_type
+        test.scan_type = clairklar_test_type.name
+        test.save()
+
+
+# Update the found_by field to remove OpenVAS CSV/ OpenVAS XML and add OpenVAS Parser
+def update_openvas_finding(finding, openvas_test_type, openvascsv_test_type, openvasxml_test_type) -> None:
+    # Check if nessus is in found by list and remove
+    if openvascsv_test_type in finding.found_by.all():
+        finding.found_by.remove(openvascsv_test_type.id)
+    # Check if nessus WAS is in found by list and remove
+    if openvasxml_test_type in finding.found_by.all():
+        finding.found_by.remove(openvasxml_test_type.id)
+    # Check if tenable is already in list somehow before adding it
+    if openvas_test_type not in finding.found_by.all():
+        finding.found_by.add(openvas_test_type.id)
+    finding.save()
+
+
+# Update the found_by field to remove Clair Klar Scan and add Clair Scan
+def update_clairklar_finding(finding, clair_test_type, clairklar_test_type) -> None:
+    # Check if nessus is in found by list and remove
+    if clairklar_test_type in finding.found_by.all():
+        finding.found_by.remove(clairklar_test_type.id)
+    # Check if tenable is already in list somehow before adding it
+    if clair_test_type not in finding.found_by.all():
+        finding.found_by.add(clair_test_type.id)
+    finding.save()
+
+
+# Update all finding objects that came from OpenVAS CSV /XML reports
+def migrate_openvas_parsers(apps, schema_editor):
+    finding_model = apps.get_model('dojo', 'Finding')
+    test_type_model = apps.get_model('dojo', 'Test_Type')
+    # Get or create OpenVAS Test Type and fetch the OpenVAS XML and OpenVAS CSV test types
+    openvas_test_type, _ = test_type_model.objects.get_or_create(name="OpenVAS Parser", active=True)
+    openvascsv_test_type = test_type_model.objects.filter(name="OpenVAS CSV").first()
+    openvasxml_test_type = test_type_model.objects.filter(name="OpenVAS XML").first()
+    # Get all the findings found by Nessus and Nessus WAS
+    findings = finding_model.objects.filter(test__scan_type__in=OPENVAS_REFERENCES)
+    logger.warning(f'We identified {findings.count()} OpenVAS CSV/ OpenVAS XML findings to migrate to OpenVAS Parser findings')
+    # Iterate over all findings and change
+    for finding in findings:
+        # Update the found by field
+        update_openvas_finding(finding, openvas_test_type, openvascsv_test_type, openvasxml_test_type)
+        # Update the test object
+        update_openvas_test(finding.test, openvas_test_type)
+
+
+# Update all finding objects that came from Clair Klar reports
+def migrate_clairklar_parsers(apps, schema_editor):
+    finding_model = apps.get_model('dojo', 'Finding')
+    test_type_model = apps.get_model('dojo', 'Test_Type')
+    # Get or create Clair Scan Test Type and fetch the Clair Klar Scan test types
+    clair_test_type, _ = test_type_model.objects.get_or_create(name="Clair Scan", active=True)
+    clairklar_test_type = test_type_model.objects.filter(name="Clair Klar Scan").first()
+    # Get all the findings found by Clair Klar Scan
+    findings = finding_model.objects.filter(test__scan_type__in=OPENVAS_REFERENCES)
+    logger.warning(f'We identified {findings.count()} Clair Klar Scan findings to migrate to Clair Scan findings')
+    # Iterate over all findings and change
+    for finding in findings:
+        # Update the found by field
+        update_clairklar_finding(finding, clair_test_type, clairklar_test_type)
+        # Update the test object
+        update_clairklar_test(finding.test, clair_test_type)
+
+
+class Migration(migrations.Migration):
+
+    dependencies = [
+        ('dojo', '0196_notifications_sla_breach_combined'),
+    ]
+
+    operations = [
+        migrations.RunPython(migrate_openvas_parsers),
+        migrations.RunPython(migrate_clairklar_parsers),
+    ]

dojo/tools/clair/parser.py

@@ -1,6 +1,7 @@
 import json
-
+import logging
 from dojo.models import Finding
+logger = logging.getLogger(__name__)
 
 
 class ClairParser(object):
@@ -11,75 +12,179 @@ def get_label_for_scan_types(self, scan_type):
         return scan_type  # no custom label for now
 
     def get_description_for_scan_types(self, scan_type):
-        return "Import JSON reports of Docker image vulnerabilities."
+        return "Import JSON reports of Docker image vulnerabilities from clair or clair klar client."
 
     def get_findings(self, json_output, test):
         tree = self.parse_json(json_output)
-        return self.get_items(tree, test)
+        if tree:
+            if self.scanner == "clair":
+                return self.get_items_clair(tree, test)
+            elif self.scanner == "clairklar":
+                items = list()
+                clair_severities = [
+                    "Unknown",
+                    "Negligible",
+                    "Low",
+                    "Medium",
+                    "High",
+                    "Critical",
+                    "Defcon1",
+                ]
+                for clair_severity in clair_severities:
+                    items.extend(
+                        self.set_items_for_severity(tree, test, clair_severity)
+                    )
+                return items
+        else:
+            return list()
 
     def parse_json(self, json_output):
-        data = json_output.read()
         try:
-            tree = json.loads(str(data, "utf-8"))
+            data = json_output.read()
+            try:
+                tree = json.loads(str(data, "utf-8"))
+            except BaseException:
+                tree = json.loads(data)
+            if tree.get('image'):
+                self.scanner = "clair"
+                subtree = tree.get("vulnerabilities")
+            elif tree.get('LayerCount'):
+                self.scanner = "clairklar"
+                subtree = tree.get("Vulnerabilities")
         except BaseException:
-            tree = json.loads(data)
-        return tree.get("vulnerabilities")
+            raise ValueError("Invalid format")
+        return subtree
 
-    def get_items(self, tree, test):
-        items = {}
+    def set_items_for_severity(self, tree, test, severity):
+        items = list()
+        tree_severity = tree.get(severity)
+        if tree_severity:
+            for data in self.get_items_clairklar(tree_severity, test):
+                items.append(data)
+            logger.debug("Appended findings for severity " + severity)
+        else:
+            logger.debug("No findings for severity " + severity)
+        return items
 
+    def get_items_clair(self, tree, test):
+        items = {}
         for node in tree:
-            item = get_item(node, test)
+            item = self.get_item_clair(node, test)
             unique_key = str(node["vulnerability"]) + str(node["featurename"])
             items[unique_key] = item
-
         return list(items.values())
 
+    def get_item_clair(self, item_node, test):
+        if (
+            item_node["severity"] == "Negligible"
+            or item_node["severity"] == "Unknown"
+        ):
+            severity = "Info"
+        else:
+            severity = item_node["severity"]
+
+        finding = Finding(
+            title=item_node["vulnerability"]
+            + " - "
+            + "("
+            + item_node["featurename"]
+            + ", "
+            + item_node["featureversion"]
+            + ")",
+            test=test,
+            severity=severity,
+            description=item_node["description"]
+            + "\n Vulnerable feature: "
+            + item_node["featurename"]
+            + "\n Vulnerable Versions: "
+            + str(item_node["featureversion"])
+            + "\n Fixed by: "
+            + str(item_node["fixedby"])
+            + "\n Namespace: "
+            + str(item_node["namespace"])
+            + "\n CVE: "
+            + str(item_node["vulnerability"]),
+            mitigation=item_node["fixedby"],
+            references=item_node["link"],
+            component_name=item_node["featurename"],
+            component_version=item_node["featureversion"],
+            false_p=False,
+            duplicate=False,
+            out_of_scope=False,
+            mitigated=None,
+            static_finding=True,
+            dynamic_finding=False,
+            impact="No impact provided",
+        )
+        if item_node["vulnerability"]:
+            finding.unsaved_vulnerability_ids = [item_node["vulnerability"]]
+        return finding
+
+    def get_items_clairklar(self, tree_severity, test):
+        items = {}
+        for node in tree_severity:
+            item = self.get_item_clairklar(node, test)
+            unique_key = str(node["Name"]) + str(node["FeatureName"])
+            items[unique_key] = item
+        return items.values()
 
-def get_item(item_node, test):
-    if (
-        item_node["severity"] == "Negligible"
-        or item_node["severity"] == "Unknown"
-    ):
-        severity = "Info"
-    else:
-        severity = item_node["severity"]
+    def get_item_clairklar(self, item_node, test):
+        if item_node["Severity"] == "Negligible":
+            severity = "Info"
+        elif item_node["Severity"] == "Unknown":
+            severity = "Critical"
+        elif item_node["Severity"] == "Defcon1":
+            severity = "Critical"
+        else:
+            severity = item_node["Severity"]
+        description = ""
+        if "Description" in item_node:
+            description += item_node["Description"] + "\n<br /> "
+        if "FeatureName" in item_node:
+            description += (
+                "Vulnerable feature: " + item_node["FeatureName"] + "\n<br />"
+            )
+        if "FeatureVersion" in item_node:
+            description += " Vulnerable Versions: " + str(
+                item_node["FeatureVersion"]
+            )
 
-    finding = Finding(
-        title=item_node["vulnerability"]
-        + " - "
-        + "("
-        + item_node["featurename"]
-        + ", "
-        + item_node["featureversion"]
-        + ")",
-        test=test,
-        severity=severity,
-        description=item_node["description"]
-        + "\n Vulnerable feature: "
-        + item_node["featurename"]
-        + "\n Vulnerable Versions: "
-        + str(item_node["featureversion"])
-        + "\n Fixed by: "
-        + str(item_node["fixedby"])
-        + "\n Namespace: "
-        + str(item_node["namespace"])
-        + "\n CVE: "
-        + str(item_node["vulnerability"]),
-        mitigation=item_node["fixedby"],
-        references=item_node["link"],
-        component_name=item_node["featurename"],
-        component_version=item_node["featureversion"],
-        false_p=False,
-        duplicate=False,
-        out_of_scope=False,
-        mitigated=None,
-        static_finding=True,
-        dynamic_finding=False,
-        impact="No impact provided",
-    )
+        mitigation = ""
+        if "FixedBy" in item_node:
+            description = description + "\n Fixed by: " + str(item_node["FixedBy"])
+            mitigation = (
+                "Please use version "
+                + item_node["FixedBy"]
+                + " of library "
+                + item_node["FeatureName"]
+            )
+        else:
+            mitigation = "A patch could not been found"
 
-    if item_node["vulnerability"]:
-        finding.unsaved_vulnerability_ids = [item_node["vulnerability"]]
+        link = ""
+        if "Link" in item_node:
+            link = item_node["Link"]
 
-    return finding
+        finding = Finding(
+            title=item_node["Name"]
+            + " - "
+            + "("
+            + item_node["FeatureName"]
+            + ", "
+            + item_node["FeatureVersion"]
+            + ")",
+            test=test,
+            severity=severity,
+            description=description,
+            mitigation=mitigation,
+            references=link,
+            false_p=False,
+            duplicate=False,
+            out_of_scope=False,
+            mitigated=None,
+            cwe=1035,  # Vulnerable Third Party Component
+            static_finding=True,
+            dynamic_finding=False,
+            impact="No impact provided",
+        )
+        return finding