Add Endpoint capability to Wazuh 4.8 exporter

Author: Emmanuel Iturbide

dojo/tools/wazuh/v4_8.py

@@ -1,5 +1,7 @@
-from dojo.models import Finding
+from django.conf import settings
 
+from dojo.models import Endpoint, Finding
+from dojo.tools.locations import LocationData
 
 class WazuhV4_8:
     def parse_findings(self, test, data):
@@ -17,10 +19,8 @@ def parse_findings(self, test, data):
                 continue  # Skip if this finding has already been processed
 
             description = vuln.get("description")
-            description += "\nAgent id:" + item.get("agent").get("id")
-            description += "\nAgent name:" + item.get("agent").get("name")
             severity = vuln.get("severity")
-            cvssv3_score = vuln.get("score").get("base")
+            cvssv3_score = vuln.get("score").get("base") if vuln.get("score") else None
             publish_date = vuln.get("published_at").split("T")[0]
             detection_time = vuln.get("detected_at").split("T")[0]
             references = vuln.get("reference")
@@ -42,6 +42,14 @@ def parse_findings(self, test, data):
                 cve + " affects (version: " + item.get("package").get("version") + ")"
             )
 
+            # Create endpoint from agent name
+            agent_name = item.get("agent").get("name")
+
+            # Prepare endpoints list (will be processed after Finding is saved)
+            endpoints = []
+            if agent_name:
+                endpoints = [Endpoint(host=agent_name)]
+
             find = Finding(
                 title=title,
                 test=test,
@@ -56,7 +64,15 @@ def parse_findings(self, test, data):
                 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 it exists.
+            if settings.V3_FEATURE_LOCATIONS:
+                find.unsaved_locations = [LocationData.url(host=agent_name)]
+            else:
+                find.unsaved_endpoints = [Endpoint(host=agent_name)]
+            
             find.unsaved_vulnerability_ids = [cve]
             dupes[dupe_key] = find
 
         return list(dupes.values())
+

unittests/tools/test_wazuh_parser.py

@@ -65,3 +65,19 @@ def test_parse_wazuh_abnormal_severity(self):
             findings = parser.get_findings(testfile, Test())
             for finding in findings:
                 self.assertEqual("Info", finding.severity)
+
+    def test_parse_v4_8_many_findings_with_location(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())
+            finding = findings[0]
+            self.assertEqual(10, len(findings))
+            self.validate_locations(findings)
+            self.assertEqual("CVE-2025-27558 affects (version: 6.8.0-60.63)", findings[0].title)
+            self.assertEqual("Critical", findings[0].severity)
+            self.assertEqual(9.1, findings[0].cvssv3_score)
+            location = self.get_unsaved_locations(finding)[0]
+            self.assertEqual("myhost0", location.host)
+            self.assertEqual("linux-image-6.8.0-60-generic", finding.component_name)
+            self.assertEqual("6.8.0-60.63", finding.component_version)
+            self.assertEqual("2025-06-30", finding.date)