update

Author: manuel-sommer

dojo/tools/wazuh/v4_8.py

@@ -1,67 +1,48 @@
+import hashlib
+
+from dojo.models import Finding
 
 
 class WazuhV4_8:
     def parse_findings(self, test, data):
-        # dupes = {}
-        # vulnerabilities = data.get("data", {}).get("affected_items", [])
-        # for item in vulnerabilities:
-        #     if (
-        #         item["condition"] != "Package unfixed"
-        #         and item["severity"] != "Untriaged"
-        #     ):
-        #         cve = item.get("cve")
-        #         package_name = item.get("name")
-        #         package_version = item.get("version")
-        #         description = item.get("condition")
-        #         severity = item.get("severity").capitalize()
-        #         agent_ip = item.get("agent_ip")
-        #         links = item.get("external_references")
-        #         cvssv3_score = item.get("cvss3_score")
-        #         publish_date = item.get("published")
-        #         agent_name = item.get("agent_name")
-        #         agent_ip = item.get("agent_ip")
-        #         detection_time = item.get("detection_time").split("T")[0]
-
-        #         references = "\n".join(links) if links else None
-
-        #         title = (
-        #             item.get("title") + " (version: " + package_version + ")"
-        #         )
-
-        #         if agent_name:
-        #             dupe_key = title + cve + agent_name + package_name + package_version
-        #         else:
-        #             dupe_key = title + cve + package_name + package_version
-        #         dupe_key = hashlib.sha256(dupe_key.encode("utf-8")).hexdigest()
-
-        #         if dupe_key in dupes:
-        #             find = dupes[dupe_key]
-        #         else:
-        #             dupes[dupe_key] = True
-
-        #             find = Finding(
-        #                 title=title,
-        #                 test=test,
-        #                 description=description,
-        #                 severity=severity,
-        #                 references=references,
-        #                 static_finding=True,
-        #                 component_name=package_name,
-        #                 component_version=package_version,
-        #                 cvssv3_score=cvssv3_score,
-        #                 publish_date=publish_date,
-        #                 unique_id_from_tool=dupe_key,
-        #                 date=detection_time,
-        #             )
-
-        #             # in some cases the agent_ip is not the perfect way on how to identify a host. Thus prefer the agent_name, if existant.
-        #             if agent_name:
-        #                 find.unsaved_endpoints = [Endpoint(host=agent_name)]
-        #             elif agent_ip:
-        #                 find.unsaved_endpoints = [Endpoint(host=agent_ip)]
-
-        #             if id:
-        #                 find.unsaved_vulnerability_ids = cve
-
-        #             dupes[dupe_key] = find
-        return []
+        dupes = {}
+        vulnerabilities = data.get("hits", {}).get("hits", [])
+        for item in vulnerabilities:
+            vuln = item.get("vulnerability")
+            cve = vuln.get("id")
+            description = vuln.get("description")
+            severity = vuln.get("severity")
+            cvssv3_score = vuln.get("score").get("base")
+            publish_date = vuln.get("published_at").split("T")[0]
+            agent_name = item.get("agent").get("name")
+            agent_id = item.get("agent").get("id")
+            detection_time = vuln.get("detected_at").split("T")[0]
+
+            references = vuln.get("reference")
+
+            title = (
+                cve + " (agent_id: " + agent_id + ")"
+            )
+
+            dupe_key = title + agent_name + description
+            dupe_key = hashlib.sha256(dupe_key.encode("utf-8")).hexdigest()
+
+            if dupe_key in dupes:
+                find = dupes[dupe_key]
+            else:
+                dupes[dupe_key] = True
+
+            find = Finding(
+                title=title,
+                test=test,
+                description=description,
+                severity=severity,
+                references=references,
+                static_finding=True,
+                cvssv3_score=cvssv3_score,
+                publish_date=publish_date,
+                unique_id_from_tool=dupe_key,
+                date=detection_time,
+            )
+        dupes[dupe_key] = find
+        return list(dupes.values())

unittests/tools/test_wazuh_parser.py

@@ -56,4 +56,4 @@ def test_parse_v4_8_many_findings(self):
         with (get_unit_tests_scans_path("wazuh") / "v4-8_many_findings.json").open(encoding="utf-8") as testfile:
             parser = WazuhParser()
             findings = parser.get_findings(testfile, Test())
-            self.assertEqual(0, len(findings))
+            self.assertEqual(10, len(findings))