fix(parser): support V3_FEATURE_LOCATIONS in Alert Logic parser

The Endpoint model is deprecated and raises NotImplementedError when V3_FEATURE_LOCATIONS is enabled. Build LocationData URL locations in that mode and fall back to Endpoint otherwise, matching the established parser migration pattern (e.g. Qualys VMDR). Endpoint tests now read via the get_unsaved_locations helper so they pass under both settings.

Authored by T. Walker - DefectDojo

Author: skywalke34

dojo/tools/alertlogic/parser.py

@@ -1,7 +1,10 @@
 import csv
 import io
 
+from django.conf import settings
+
 from dojo.models import Endpoint, Finding
+from dojo.tools.locations import LocationData
 
 SEVERITY_MAPPING = {
     "Info": "Info",
@@ -67,12 +70,11 @@ def get_findings(self, filename, test):
             if cve:
                 finding.unsaved_vulnerability_ids = [cve]
 
-            endpoints = _build_endpoints(
+            _add_locations(
+                finding,
                 row.get("IP Address"),
                 row.get("Protocol/Port"),
             )
-            if endpoints:
-                finding.unsaved_endpoints = endpoints
 
             tags = _build_tags(row)
             if tags:
@@ -124,22 +126,26 @@ def _parse_cvss(value):
         return None
 
 
-def _build_endpoints(ip_field, protoport_field):
+def _add_locations(finding, ip_field, protoport_field):
     if not ip_field:
-        return []
+        return
     protocol, port = _parse_proto_port(protoport_field)
-    endpoints = []
     for raw_host in ip_field.split(","):
         host = raw_host.strip()
         if not host:
             continue
-        kwargs = {"host": host}
-        if protocol:
-            kwargs["protocol"] = protocol
-        if port:
-            kwargs["port"] = port
-        endpoints.append(Endpoint(**kwargs))
-    return endpoints
+        if settings.V3_FEATURE_LOCATIONS:
+            finding.unsaved_locations.append(
+                LocationData.url(host=host, protocol=protocol or "", port=port),
+            )
+        else:
+            # TODO: Delete this after the move to Locations
+            kwargs = {"host": host}
+            if protocol:
+                kwargs["protocol"] = protocol
+            if port:
+                kwargs["port"] = port
+            finding.unsaved_endpoints.append(Endpoint(**kwargs))
 
 
 def _parse_proto_port(value):

unittests/tools/test_alertlogic_parser.py

@@ -97,36 +97,39 @@ def test_unique_id_from_tool(self):
     def test_endpoint_single_ipv4(self):
         # Row 0 has a single IPv4 address
         finding = self._findings("many_vulns.csv")[0]
-        self.assertEqual(1, len(finding.unsaved_endpoints))
-        endpoint = finding.unsaved_endpoints[0]
+        endpoints = self.get_unsaved_locations(finding)
+        self.assertEqual(1, len(endpoints))
+        endpoint = endpoints[0]
         self.assertEqual("192.0.2.20", endpoint.host)
         self.assertEqual("tcp", endpoint.protocol)
         self.assertEqual(8080, endpoint.port)
 
     def test_endpoint_multi_ipv4_and_ipv6(self):
         # Row 1: "198.51.100.30, fe80::250:56ff:fe96:b97"
         finding = self._findings("many_vulns.csv")[1]
-        self.assertEqual(2, len(finding.unsaved_endpoints))
-        hosts = {ep.host for ep in finding.unsaved_endpoints}
+        endpoints = self.get_unsaved_locations(finding)
+        self.assertEqual(2, len(endpoints))
+        hosts = {ep.host for ep in endpoints}
         self.assertEqual({"198.51.100.30", "fe80::250:56ff:fe96:b97"}, hosts)
 
     def test_endpoint_ipv6_only(self):
         # Row 6 has IPv6-only address
         finding = self._findings("many_vulns.csv")[6]
-        self.assertEqual(1, len(finding.unsaved_endpoints))
-        self.assertEqual("2001:db8::1:80", finding.unsaved_endpoints[0].host)
+        endpoints = self.get_unsaved_locations(finding)
+        self.assertEqual(1, len(endpoints))
+        self.assertEqual("2001:db8::1:80", endpoints[0].host)
 
     def test_endpoint_port_zero_is_omitted(self):
         # Row 2 has Protocol/Port "TCP/0" — port should not be set
         finding = self._findings("many_vulns.csv")[2]
-        self.assertEqual(1, len(finding.unsaved_endpoints))
-        self.assertIsNone(finding.unsaved_endpoints[0].port)
+        endpoints = self.get_unsaved_locations(finding)
+        self.assertEqual(1, len(endpoints))
+        self.assertIsNone(endpoints[0].port)
 
     def test_endpoint_clean_succeeds(self):
-        # Hard guardrail: every endpoint must pass clean()
-        for finding in self._findings("many_vulns.csv"):
-            for endpoint in finding.unsaved_endpoints or []:
-                endpoint.clean()
+        # Hard guardrail: every endpoint/location must pass clean()
+        # (get_unsaved_locations cleans each entry internally)
+        self.validate_locations(self._findings("many_vulns.csv"))
 
     def test_cve_present(self):
         # Row 0 has CVE-2021-44228 (Log4Shell)