Add Open Pentest Format (OPF) parser

docs/content/supported_tools/parsers/file/opf.md

@@ -0,0 +1,23 @@
+---
+title: "Open Pentest Format (OPF)"
+toc_hide: true
+---
+
+Import an [Open Pentest Format](https://cairnsecurity.com/opf) (OPF) file. OPF is
+a JSON format for pentest findings.
+
+Upload the `.opf.json` export. Each OPF finding maps to a DefectDojo finding:
+
+- `severity` maps to DefectDojo severity (`informational` becomes `Info`).
+- `cvssScore` and `cvssVector` populate the CVSSv3 score and vector.
+- The first `cweIds` / `cweId` value populates the CWE.
+- `cveIds` populate the finding's vulnerability ids.
+- `recommendation` maps to Mitigation, `impact` to Impact, `stepsToReproduce`
+  to Steps to Reproduce, and `references` to References.
+- URL `affectedAssets` become endpoints; other assets (source paths, ARNs) are
+  listed in the description.
+- `testType`, `owaspCategory` and `mitreTechniques` are added as tags.
+
+### Sample Scan Data
+
+Sample OPF scans can be found [here](https://github.com/DefectDojo/django-DefectDojo/tree/master/unittests/scans/opf).

dojo/settings/settings.dist.py

@@ -1430,6 +1430,7 @@ def generate_url(scheme, double_slashes, user, password, host, port, path, param
     "PyRIT Scan": ["title", "vuln_id_from_tool"],
     "debsecan Scan": ["vulnerability_ids", "component_name"],
     "PMapper Scan": ["vuln_id_from_tool", "component_name"],
+    "OPF Scan": ["title", "cwe", "severity", "description"],
 }
 
 # Override the hardcoded settings here via the env var
@@ -1512,6 +1513,7 @@ def generate_url(scheme, double_slashes, user, password, host, port, path, param
     "Cyberwatch scan (Galeax)": True,
     "OpenVAS Parser v2": True,
     "OpenReports": True,
+    "OPF Scan": True,
 }
 
 # List of fields that are known to be usable in hash_code computation)
@@ -1877,6 +1879,7 @@ def generate_url(scheme, double_slashes, user, password, host, port, path, param
     "PyRIT Scan": DEDUPE_ALGO_HASH_CODE,
     "debsecan Scan": DEDUPE_ALGO_HASH_CODE,
     "PMapper Scan": DEDUPE_ALGO_HASH_CODE,
+    "OPF Scan": DEDUPE_ALGO_HASH_CODE,
 }
 
 # Override the hardcoded settings here via the env var

dojo/tools/opf/parser.py

@@ -0,0 +1,199 @@
+import datetime
+import html
+import json
+import re
+
+from dojo.models import Endpoint, Finding
+from dojo.utils import parse_cvss_data
+
+
+class OPFParser:
+
+    """
+    Parser for the Open Pentest Format (OPF), a JSON format for pentest findings.
+
+    Spec: https://cairnsecurity.com/opf
+    """
+
+    SEVERITY_MAP = {
+        "critical": "Critical",
+        "high": "High",
+        "medium": "Medium",
+        "low": "Low",
+        "informational": "Info",
+    }
+
+    def get_scan_types(self):
+        return ["OPF Scan"]
+
+    def get_label_for_scan_types(self, scan_type):
+        return "OPF Scan"
+
+    def get_description_for_scan_types(self, scan_type):
+        return (
+            "Import an Open Pentest Format (OPF) .opf.json file. "
+            "See https://cairnsecurity.com/opf."
+        )
+
+    def get_findings(self, file, test):
+        data = json.load(file)
+        if not isinstance(data, dict) or not isinstance(data.get("findings"), list):
+            msg = "Invalid OPF file: expected an object with a 'findings' array."
+            raise TypeError(msg)
+
+        report_date = self._parse_date((data.get("metadata") or {}).get("exportedAt"))
+
+        findings = []
+        for entry in data["findings"]:
+            if not isinstance(entry, dict) or not str(entry.get("title", "")).strip():
+                continue
+            findings.append(self._build_finding(entry, test, report_date))
+        return findings
+
+    def _build_finding(self, entry, test, report_date):
+        severity = self.SEVERITY_MAP.get(str(entry.get("severity", "")).lower(), "Info")
+
+        description = self._html_to_text(entry.get("description", "")) or entry["title"]
+
+        # Only URL assets become endpoints. Source paths, ARNs and the like go
+        # in the description instead.
+        url_assets, other_assets = self._split_assets(entry.get("affectedAssets"))
+        if other_assets:
+            description = description + "\n\nAffected assets:\n" + "\n".join(f"- {a}" for a in other_assets)
+
+        finding = Finding(
+            title=str(entry["title"])[:511],
+            test=test,
+            severity=severity,
+            description=description,
+            static_finding=False,
+            dynamic_finding=True,
+        )
+
+        if report_date:
+            finding.date = report_date
+
+        cwe = self._first_cwe(entry)
+        if cwe is not None:
+            finding.cwe = cwe
+
+        # OPF cvssVector is not pinned to a CVSS version, so let DefectDojo detect
+        # v4/v3/v2 and route each to its own field with an authoritative score
+        # rather than storing the vector raw in cvssv3.
+        vector = entry.get("cvssVector")
+        cvss = parse_cvss_data(vector) if isinstance(vector, str) and vector else {}
+        if cvss.get("cvssv3"):
+            finding.cvssv3 = cvss["cvssv3"]
+            if cvss.get("cvssv3_score") is not None:
+                finding.cvssv3_score = cvss["cvssv3_score"]
+        if cvss.get("cvssv4"):
+            finding.cvssv4 = cvss["cvssv4"]
+            if cvss.get("cvssv4_score") is not None:
+                finding.cvssv4_score = cvss["cvssv4_score"]
+        # Fall back to the tool-reported score only when the vector yielded none
+        # (no vector, an unparseable one, or a v2 vector with no v3/v4 field to hold it).
+        if finding.cvssv3_score is None and finding.cvssv4_score is None:
+            score = entry.get("cvssScore")
+            if isinstance(score, (int, float)):
+                finding.cvssv3_score = float(score)
+
+        impact = self._html_to_text(entry.get("impact", ""))
+        if impact:
+            finding.impact = impact
+        mitigation = self._html_to_text(entry.get("recommendation", ""))
+        if mitigation:
+            finding.mitigation = mitigation
+
+        steps = entry.get("stepsToReproduce")
+        if isinstance(steps, list) and steps:
+            finding.steps_to_reproduce = "\n".join(
+                f"{i}. {self._html_to_text(str(step))}" for i, step in enumerate(steps, start=1)
+            )
+
+        references = entry.get("references")
+        if isinstance(references, list) and references:
+            lines = []
+            for ref in references:
+                if isinstance(ref, dict) and ref.get("url"):
+                    title = ref.get("title")
+                    lines.append(f"{title}: {ref['url']}" if title else str(ref["url"]))
+            if lines:
+                finding.references = "\n".join(lines)
+
+        cve_ids = entry.get("cveIds")
+        if isinstance(cve_ids, list) and cve_ids:
+            finding.unsaved_vulnerability_ids = [str(cve) for cve in cve_ids]
+
+        opf_id = entry.get("id")
+        if opf_id:
+            finding.unique_id_from_tool = str(opf_id)
+            finding.vuln_id_from_tool = str(opf_id)
+
+        if url_assets:
+            finding.unsaved_endpoints = url_assets
+
+        tags = []
+        if entry.get("testType"):
+            tags.append(str(entry["testType"]))
+        if entry.get("owaspCategory"):
+            tags.append(str(entry["owaspCategory"]))
+        tags.extend(str(technique) for technique in entry.get("mitreTechniques") or [])
+        if tags:
+            finding.unsaved_tags = tags
+
+        return finding
+
+    @staticmethod
+    def _split_assets(assets):
+        if not isinstance(assets, list):
+            return [], []
+        endpoints, other = [], []
+        for asset in assets:
+            if not isinstance(asset, str) or not asset.strip():
+                continue
+            value = asset.strip()
+            if "://" not in value:
+                other.append(value)
+                continue
+            try:
+                endpoints.append(Endpoint.from_uri(value))
+            except Exception:
+                # Not a parseable URL after all; keep it in the description.
+                other.append(value)
+        return endpoints, other
+
+    @staticmethod
+    def _first_cwe(entry):
+        candidates = list(entry.get("cweIds") or [])
+        if entry.get("cweId"):
+            candidates.append(entry["cweId"])
+        for value in candidates:
+            match = re.search(r"(\d+)", str(value))
+            if match:
+                return int(match.group(1))
+        return None
+
+    @staticmethod
+    def _parse_date(value):
+        if not isinstance(value, str):
+            return None
+        match = re.match(r"(\d{4})-(\d{2})-(\d{2})", value)
+        if not match:
+            return None
+        try:
+            return datetime.date(int(match.group(1)), int(match.group(2)), int(match.group(3)))
+        except ValueError:
+            return None
+
+    @staticmethod
+    def _html_to_text(value):
+        if not value or not isinstance(value, str):
+            return ""
+        text = re.sub(r"<br\s*/?>", "\n", value, flags=re.IGNORECASE)
+        text = re.sub(r"</(p|div|li|h[1-6]|tr)>", "\n", text, flags=re.IGNORECASE)
+        text = re.sub(r"<li[^>]*>", "- ", text, flags=re.IGNORECASE)
+        text = re.sub(r"<[^>]+>", "", text)
+        # Decode the full named + numeric HTML entity set rather than a partial
+        # hand-rolled map, so entities like &apos; &#x27; &#8217; do not leak through.
+        text = html.unescape(text)
+        return re.sub(r"\n{3,}", "\n\n", text).strip()

unittests/scans/opf/many_findings.json

@@ -0,0 +1,21 @@
+{
+  "opfVersion": "1.1",
+  "textFormat": "html",
+  "metadata": { "source": "Cairn", "exportedAt": "2026-08-06T00:00:00Z", "findingCount": 4 },
+  "findings": [
+    { "id": "sqli-search", "title": "SQL injection in the report search endpoint", "severity": "critical",
+      "description": "<p>The <code>q</code> parameter is concatenated into the query without parameterisation.</p>",
+      "impact": "An attacker can read or modify arbitrary rows.", "recommendation": "Use parameterised queries.",
+      "cvssScore": 9.8, "cvssVector": "CVSS:3.1/AV:N/AC:L/PR:N/UI:N/S:U/C:H/I:H/A:H", "cvssVersion": "3.1",
+      "cweIds": ["CWE-89"], "owaspCategory": "A03:2021 Injection", "mitreTechniques": ["T1190"],
+      "affectedAssets": ["src/search/query.ts"],
+      "references": [{ "title": "CWE-89", "url": "https://cwe.mitre.org/data/definitions/89.html", "type": "cwe" }] },
+    { "id": "s3-public", "title": "Publicly readable S3 bucket", "severity": "high",
+      "description": "Bucket policy grants s3:GetObject to * (anonymous).", "recommendation": "Restrict the bucket policy; enable Block Public Access.",
+      "cvssScore": 7.5, "cweIds": ["CWE-200"], "affectedAssets": ["infra/s3.tf"] },
+    { "id": "verbose-errors", "title": "Verbose error messages leak stack traces", "severity": "medium",
+      "description": "Unhandled exceptions return full stack traces to the client.", "cweIds": ["CWE-209"], "affectedAssets": ["src/app/errors.ts"] },
+    { "id": "missing-hsts", "title": "HSTS header not set", "severity": "low",
+      "description": "Responses do not include Strict-Transport-Security.", "affectedAssets": ["src/server/headers.ts"] }
+  ]
+}

unittests/scans/opf/no_findings.json

@@ -0,0 +1,6 @@
+{
+  "opfVersion": "1.1",
+  "textFormat": "text",
+  "metadata": { "source": "Cairn", "exportedAt": "2026-08-06T00:00:00Z", "findingCount": 0 },
+  "findings": []
+}

unittests/tools/test_opf_parser.py

@@ -0,0 +1,93 @@
+import io
+import json
+
+from dojo.models import Test
+from dojo.tools.opf.parser import OPFParser
+from unittests.dojo_test_case import DojoTestCase, get_unit_tests_scans_path
+
+
+class TestOPFParser(DojoTestCase):
+
+    def test_opf_parser_no_findings(self):
+        with (get_unit_tests_scans_path("opf") / "no_findings.json").open(encoding="utf-8") as testfile:
+            parser = OPFParser()
+            findings = parser.get_findings(testfile, Test())
+            self.assertEqual(0, len(findings))
+
+    def test_opf_parser_many_findings(self):
+        with (get_unit_tests_scans_path("opf") / "many_findings.json").open(encoding="utf-8") as testfile:
+            parser = OPFParser()
+            findings = parser.get_findings(testfile, Test())
+            self.assertEqual(4, len(findings))
+
+            # Critical finding with full detail
+            critical = findings[0]
+            self.assertEqual("SQL injection in the report search endpoint", critical.title)
+            self.assertEqual("Critical", critical.severity)
+            self.assertEqual(89, critical.cwe)
+            self.assertEqual(9.8, critical.cvssv3_score)
+            self.assertIn("CVSS:3.1", critical.cvssv3)
+            self.assertEqual("Use parameterised queries.", critical.mitigation)
+            # HTML in the OPF description is flattened to text
+            self.assertNotIn("<p>", critical.description)
+            self.assertIn("concatenated", critical.description)
+
+            # High finding with a CWE but no CVSS vector
+            high = findings[1]
+            self.assertEqual("High", high.severity)
+            self.assertEqual(200, high.cwe)
+            self.assertEqual(7.5, high.cvssv3_score)
+
+            # Medium/Low findings carry no CVSS score at all
+            self.assertEqual("Medium", findings[2].severity)
+            self.assertEqual("Low", findings[3].severity)
+
+    def test_opf_parser_severity_and_tags(self):
+        with (get_unit_tests_scans_path("opf") / "many_findings.json").open(encoding="utf-8") as testfile:
+            parser = OPFParser()
+            findings = parser.get_findings(testfile, Test())
+            severities = {finding.severity for finding in findings}
+            self.assertEqual({"Critical", "High", "Medium", "Low"}, severities)
+            # OWASP / testType / MITRE ride along as tags on the critical finding
+            self.assertIn("A03:2021 Injection", findings[0].unsaved_tags)
+
+    def test_opf_parser_cvss_v4_vector_routes_to_cvssv4(self):
+        # A CVSS v4 vector must land in cvssv4 (with a derived score), not raw in cvssv3.
+        doc = {
+            "opfVersion": "1.1",
+            "findings": [
+                {
+                    "title": "SSRF in the webhook sender",
+                    "severity": "high",
+                    "cvssVector": "CVSS:4.0/AV:N/AC:L/AT:N/PR:N/UI:N/VC:H/VI:H/VA:H/SC:N/SI:N/SA:N",
+                },
+            ],
+        }
+        parser = OPFParser()
+        findings = parser.get_findings(io.StringIO(json.dumps(doc)), Test())
+        self.assertEqual(1, len(findings))
+        finding = findings[0]
+        self.assertIn("CVSS:4.0", finding.cvssv4)
+        self.assertIsNotNone(finding.cvssv4_score)
+        # v4 vector must not be mislabeled into the v3 field
+        self.assertFalse(finding.cvssv3)
+
+    def test_opf_parser_decodes_full_html_entity_set(self):
+        # Entities beyond the old hand-rolled map (numeric refs, &apos;) must decode.
+        doc = {
+            "opfVersion": "1.1",
+            "findings": [
+                {
+                    "title": "Entity handling",
+                    "severity": "low",
+                    "impact": "It&#8217;s the user&apos;s caf&#233; &#x27;else&#x27;.",
+                },
+            ],
+        }
+        parser = OPFParser()
+        findings = parser.get_findings(io.StringIO(json.dumps(doc)), Test())
+        impact = findings[0].impact
+        self.assertNotIn("&", impact)  # no entity codes leak through
+        self.assertIn("café", impact)
+        self.assertIn("user's", impact)
+        self.assertIn("'else'", impact)