fix(parser): support V3_FEATURE_LOCATIONS in Qualys VMDR parser

Replace direct Endpoint() instantiation with the new LocationData API when
V3_FEATURE_LOCATIONS is enabled, falling back to Endpoint for the legacy
code path. Tests pass under both flag states.

Authored by T. Walker - DefectDojo

Author: skywalke34

dojo/tools/qualys_vmdr/cve_parser.py

@@ -1,3 +1,5 @@
+from django.conf import settings
+
 from dojo.models import Finding
 from dojo.tools.qualys_vmdr.helpers import (
     build_description_cve,
@@ -6,6 +8,7 @@
     map_qualys_severity,
     parse_cvss_score,
     parse_endpoints,
+    parse_locations,
     parse_qualys_csv_content,
     parse_qualys_date,
     parse_tags,
@@ -54,10 +57,17 @@ def _create_finding(self, row):
         if cvss_score is not None:
             finding.cvssv3_score = cvss_score
 
-        finding.unsaved_endpoints = parse_endpoints(
-            row.get("Asset IPV4", ""),
-            row.get("Asset IPV6", ""),
-        )
+        if settings.V3_FEATURE_LOCATIONS:
+            finding.unsaved_locations = parse_locations(
+                row.get("Asset IPV4", ""),
+                row.get("Asset IPV6", ""),
+            )
+        else:
+            # TODO: Delete this after the move to Locations
+            finding.unsaved_endpoints = parse_endpoints(
+                row.get("Asset IPV4", ""),
+                row.get("Asset IPV6", ""),
+            )
         finding.unsaved_tags = parse_tags(row.get("Asset Tags", ""))
 
         if not is_qualys_null(cve):

dojo/tools/qualys_vmdr/helpers.py

@@ -5,6 +5,7 @@
 from dateutil import parser as dateutil_parser
 
 from dojo.models import Endpoint
+from dojo.tools.locations import LocationData
 
 SEVERITY_MAPPING = {
     "1": "Info",
@@ -143,16 +144,21 @@ def build_description_cve(row):
     return "\n\n".join(parts) if parts else "No details available."
 
 
+def _extract_hosts(ipv4_field, ipv6_field):
+    if ipv4_field and ipv4_field.strip():
+        return [ip.strip() for ip in ipv4_field.split(",") if ip.strip()]
+    if ipv6_field and ipv6_field.strip():
+        return [ipv6_field.strip()]
+    return []
+
+
+# TODO: Delete this after the move to Locations
 def parse_endpoints(ipv4_field, ipv6_field):
-    endpoints = []
+    return [Endpoint(host=host) for host in _extract_hosts(ipv4_field, ipv6_field)]
 
-    if ipv4_field and ipv4_field.strip():
-        ips = [ip.strip() for ip in ipv4_field.split(",") if ip.strip()]
-        endpoints.extend(Endpoint(host=ip) for ip in ips)
-    elif ipv6_field and ipv6_field.strip():
-        endpoints.append(Endpoint(host=ipv6_field.strip()))
 
-    return endpoints
+def parse_locations(ipv4_field, ipv6_field):
+    return [LocationData.url(host=host) for host in _extract_hosts(ipv4_field, ipv6_field)]
 
 
 def parse_tags(tags_field):

dojo/tools/qualys_vmdr/qid_parser.py

@@ -1,9 +1,12 @@
+from django.conf import settings
+
 from dojo.models import Finding
 from dojo.tools.qualys_vmdr.helpers import (
     build_description_qid,
     build_severity_justification,
     map_qualys_severity,
     parse_endpoints,
+    parse_locations,
     parse_qualys_csv_content,
     parse_qualys_date,
     parse_tags,
@@ -45,10 +48,17 @@ def _create_finding(self, row):
             dynamic_finding=False,
         )
 
-        finding.unsaved_endpoints = parse_endpoints(
-            row.get("Asset IPV4", ""),
-            row.get("Asset IPV6", ""),
-        )
+        if settings.V3_FEATURE_LOCATIONS:
+            finding.unsaved_locations = parse_locations(
+                row.get("Asset IPV4", ""),
+                row.get("Asset IPV6", ""),
+            )
+        else:
+            # TODO: Delete this after the move to Locations
+            finding.unsaved_endpoints = parse_endpoints(
+                row.get("Asset IPV4", ""),
+                row.get("Asset IPV6", ""),
+            )
         finding.unsaved_tags = parse_tags(row.get("Asset Tags", ""))
 
         return finding

unittests/tools/test_qualys_vmdr_parser.py

@@ -137,7 +137,7 @@ def test_qid_endpoints_single_ip(self):
         ) as testfile:
             parser = QualysVMDRParser()
             findings = parser.get_findings(testfile, Test())
-            endpoints = findings[0].unsaved_endpoints
+            endpoints = self.get_unsaved_locations(findings[0])
             self.assertEqual(1, len(endpoints))
             self.assertEqual("10.0.0.1", endpoints[0].host)
 
@@ -148,7 +148,7 @@ def test_qid_endpoints_multiple_ips(self):
             parser = QualysVMDRParser()
             findings = parser.get_findings(testfile, Test())
             multi_ip_finding = [f for f in findings if f.unique_id_from_tool == "100003"][0]
-            endpoints = multi_ip_finding.unsaved_endpoints
+            endpoints = self.get_unsaved_locations(multi_ip_finding)
             self.assertEqual(2, len(endpoints))
             hosts = [e.host for e in endpoints]
             self.assertIn("10.0.0.20", hosts)
@@ -161,7 +161,7 @@ def test_qid_endpoints_ipv6_fallback(self):
             parser = QualysVMDRParser()
             findings = parser.get_findings(testfile, Test())
             ipv6_finding = [f for f in findings if f.unique_id_from_tool == "100005"][0]
-            endpoints = ipv6_finding.unsaved_endpoints
+            endpoints = self.get_unsaved_locations(ipv6_finding)
             self.assertEqual(1, len(endpoints))
             self.assertEqual("2001:db8::1", endpoints[0].host)
 
@@ -288,5 +288,4 @@ def test_qid_endpoint_clean(self):
         ) as testfile:
             parser = QualysVMDRParser()
             findings = parser.get_findings(testfile, Test())
-            for endpoint in findings[0].unsaved_endpoints:
-                endpoint.clean()
+            self.validate_locations(findings)