🎉 Implement json part for Cloudflare insights parser (#14096)

manuel-sommer · web-flow · commit deb198d7eafd · 2026-01-16T20:27:37.000-06:00
diff --git a/docs/content/supported_tools/parsers/file/cloudflare_insights.md b/docs/content/supported_tools/parsers/file/cloudflare_insights.md
@@ -3,7 +3,7 @@ title: "Cloudflare Insights"
 toc_hide: true
 ---
 
-Import Cloudflare Insights findings using the **CSV export** provided by Cloudflare.
+Import Cloudflare Insights findings using the **CSV export** or via api the **JSON output** provided by Cloudflare.
 
 ### Sample Scan Data
 Sample Cloudflare Insights files can be found [here](https://github.com/DefectDojo/django-DefectDojo/tree/master/unittests/scans/cloudflare_insights).
diff --git a/dojo/tools/cloudflare_insights/parser.py b/dojo/tools/cloudflare_insights/parser.py
@@ -1,5 +1,6 @@
 import csv
 import io
+import json
 from urllib.parse import urlparse
 
 from dojo.models import Endpoint, Finding
@@ -8,19 +9,27 @@
 class CloudflareInsightsParser:
 
     """
-    DefectDojo parser for Cloudflare Insights CSV exports.
+    DefectDojo parser for Cloudflare Insights CSV or JSON exports.
 
-    Expected columns:
+    CSV expected columns:
       - severity
       - issue_class
-      - subject            (used as Endpoint host; not repeated in description)
+      - subject
       - issue_type
-      - scan_performed_on  (ignored)
       - status
-      - insight            (optional)
-      - detection_method   (optional)
-      - risk               (optional)
-      - recommended_action (used as mitigation if present)
+      - insight (optional)
+      - detection_method (optional)
+      - risk (optional)
+      - recommended_action (optional)
+
+    JSON expected fields:
+      - severity
+      - issue_class
+      - subject
+      - issue_type
+      - dismissed (maps to status)
+      - resolve_text (optional mitigation)
+      - risk (optional)
     """
 
     def get_scan_types(self):
@@ -30,15 +39,15 @@ def get_label_for_scan_types(self, scan_type):
         return scan_type
 
     def get_description_for_scan_types(self, scan_type):
-        return "Import Cloudflare Insights (CSV export)."
+        return "Import Cloudflare Insights (CSV or JSON export)."
 
     def _map_severity(self, value):
         normalized = value.strip().lower()
         mapping = {
             "low": "Low",
             "moderate": "Medium",
             "critical": "Critical",
-            "high": "High",  # optional: Cloudflare occasionally uses this
+            "high": "High",
         }
         return mapping.get(normalized, "Info")
 
@@ -56,25 +65,14 @@ def _extract_host_from_subject(self, subject: str) -> str | None:
         host = netloc or s
         if ":" in host:
             host = host.split(":", 1)[0]
-        host = host.strip().strip("/").strip()
-
-        return host or None
+        return host.strip().strip("/") or None
 
     def _is_inactive_status(self, status: str) -> bool:
         inactive_markers = {"resolved", "mitigated", "closed", "fixed"}
         return bool(status) and status.strip().lower() in inactive_markers
 
-    def get_findings(self, filename, test):
-        content = filename.read()
-        if isinstance(content, bytes):
-            content = content.decode("utf-8", errors="replace")
-
-        reader = csv.DictReader(
-            io.StringIO(content),
-            delimiter=",",
-            quotechar='"',
-            skipinitialspace=True,
-        )
+    def _parse_csv(self, content: str, test):
+        reader = csv.DictReader(io.StringIO(content), delimiter=",", quotechar='"', skipinitialspace=True)
         findings = []
         for row in reader:
             severity_raw = (row.get("severity") or "").strip()
@@ -86,15 +84,10 @@ def get_findings(self, filename, test):
             detection_method = (row.get("detection_method") or "").strip()
             risk = (row.get("risk") or "").strip()
             recommended_action = (row.get("recommended_action") or "").strip()
+
             mapped_severity = self._map_severity(severity_raw)
-            if issue_type and subject:
-                title = f"{issue_type}: {subject}"
-            elif issue_type:
-                title = issue_type
-            elif subject:
-                title = subject
-            else:
-                title = "Cloudflare Insight"
+            title = f"{issue_type}: {subject}" if issue_type and subject else issue_type or subject or "Cloudflare Insight"
+
             description_parts = []
             if issue_class:
                 description_parts.append(f"**Issue class**: {issue_class}")
@@ -109,6 +102,7 @@ def get_findings(self, filename, test):
             if risk:
                 description_parts.append(f"**Risk**: {risk}")
             description = "\n\n".join(description_parts)
+
             finding = Finding(
                 test=test,
                 title=title,
@@ -124,5 +118,54 @@ def get_findings(self, filename, test):
             if host:
                 finding.unsaved_endpoints = [Endpoint(host=host, port=None)]
             findings.append(finding)
+        return findings
+
+    def _parse_json(self, content: str, test):
+        data = json.loads(content)
+        findings = []
+        for item in data:
+            severity_raw = (item.get("severity") or "").strip()
+            issue_class = (item.get("issue_class") or "").strip()
+            subject = (item.get("subject") or "").strip()
+            issue_type = (item.get("issue_type") or "").strip()
+            dismissed = item.get("dismissed", False)
+            risk = (item.get("risk") or "").strip()
+            recommended_action = (item.get("resolve_text") or "").strip()
+
+            mapped_severity = self._map_severity(severity_raw)
+            title = f"{issue_type}: {subject}" if issue_type and subject else issue_type or subject or "Cloudflare Insight"
+
+            description_parts = []
+            if issue_class:
+                description_parts.append(f"**Issue class**: {issue_class}")
+            if issue_type:
+                description_parts.append(f"**Issue type**: {issue_type}")
+            if risk:
+                description_parts.append(f"**Risk**: {risk}")
+            description = "\n\n".join(description_parts)
 
+            finding = Finding(
+                test=test,
+                title=title,
+                severity=mapped_severity,
+                description=description,
+                mitigation=recommended_action,
+                references="Not provided!",
+                static_finding=False,
+                dynamic_finding=True,
+            )
+            finding.active = not dismissed
+            host = self._extract_host_from_subject(subject)
+            if host:
+                finding.unsaved_endpoints = [Endpoint(host=host, port=None)]
+            findings.append(finding)
         return findings
+
+    def get_findings(self, filename, test):
+        content = filename.read()
+        if isinstance(content, bytes):
+            content = content.decode("utf-8", errors="replace")
+        content_strip = content.strip()
+        if content_strip.startswith("["):
+            return self._parse_json(content_strip, test)
+        return self._parse_csv(content_strip, test)
diff --git a/unittests/scans/cloudflare_insights/many_findings.json b/unittests/scans/cloudflare_insights/many_findings.json
@@ -0,0 +1,47 @@
+[
+  {
+    "id": "asdf-security_txt_not_enabled",
+    "issue_class": "security_txt_not_enabled",
+    "issue_type": "configuration_suggestion",
+    "payload": {
+      "zone_tag": "asdffwe"
+    },
+    "severity": "Low",
+    "subject": "test.de",
+    "timestamp": "2024-12-01T05:43:45.712676Z",
+    "dismissed": false,
+    "since": "2024-10-03T07:46:55.613682Z",
+    "resolve_link": "",
+    "resolve_text": ""
+  },
+  {
+    "id": "fwefwef-security_txt_not_enabled",
+    "issue_class": "security_txt_not_enabled",
+    "issue_type": "configuration_suggestion",
+    "payload": {
+      "zone_tag": "klklkl"
+    },
+    "severity": "Low",
+    "subject": "nannana.com",
+    "timestamp": "2024-12-01T05:43:44.175313Z",
+    "dismissed": false,
+    "since": "2024-10-03T08:48:44.581718Z",
+    "resolve_link": "",
+    "resolve_text": ""
+  },
+  {
+    "id": "yxcv-security_txt_not_enabled",
+    "issue_class": "security_txt_not_enabled",
+    "issue_type": "configuration_suggestion",
+    "payload": {
+      "zone_tag": "yxcv"
+    },
+    "severity": "Low",
+    "subject": "asdf.de",
+    "timestamp": "2024-12-01T05:43:44.252529Z",
+    "dismissed": false,
+    "since": "2024-10-03T08:48:43.923198Z",
+    "resolve_link": "",
+    "resolve_text": ""
+  }
+]
diff --git a/unittests/scans/cloudflare_insights/one_finding.json b/unittests/scans/cloudflare_insights/one_finding.json
@@ -0,0 +1,17 @@
+[
+  {
+    "id": "123-security_txt_not_enabled",
+    "issue_class": "security_txt_not_enabled",
+    "issue_type": "configuration_suggestion",
+    "payload": {
+      "zone_tag": "12"
+    },
+    "severity": "Low",
+    "subject": "domain.com",
+    "timestamp": "2024-12-01T05:43:45.712676Z",
+    "dismissed": false,
+    "since": "2024-10-03T07:46:55.613682Z",
+    "resolve_link": "",
+    "resolve_text": ""
+  }
+]
diff --git a/unittests/tools/test_cloudflare_insights_parser.py b/unittests/tools/test_cloudflare_insights_parser.py
@@ -22,3 +22,21 @@ def test_cloudflare_insights_parser_with_many_findings(self):
             finding = findings[0]
             self.assertEqual("Exposed infrastructure: domain1.com", finding.title)
             self.assertEqual("Medium", finding.severity)
+
+    def test_cloudflare_insights_parser_with_one_finding_json(self):
+        with (get_unit_tests_scans_path("cloudflare_insights") / "one_finding.json").open(encoding="utf-8") as testfile:
+            parser = CloudflareInsightsParser()
+            findings = parser.get_findings(testfile, Test())
+            self.assertEqual(1, len(findings))
+            finding = findings[0]
+            self.assertEqual("configuration_suggestion: domain.com", finding.title)
+            self.assertEqual("Low", finding.severity)
+
+    def test_cloudflare_insights_parser_with_many_findings_json(self):
+        with (get_unit_tests_scans_path("cloudflare_insights") / "many_findings.json").open(encoding="utf-8") as testfile:
+            parser = CloudflareInsightsParser()
+            findings = parser.get_findings(testfile, Test())
+            self.assertEqual(3, len(findings))
+            finding = findings[0]
+            self.assertEqual("configuration_suggestion: test.de", finding.title)
+            self.assertEqual("Low", finding.severity)
