diff --git a/docs/content/supported_tools/parsers/file/opf.md b/docs/content/supported_tools/parsers/file/opf.md new file mode 100644 index 00000000000..41db4a7040e --- /dev/null +++ b/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). diff --git a/dojo/settings/settings.dist.py b/dojo/settings/settings.dist.py index 8ae5492b905..c3422b30247 100644 --- a/dojo/settings/settings.dist.py +++ b/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 diff --git a/dojo/tools/opf/__init__.py b/dojo/tools/opf/__init__.py new file mode 100644 index 00000000000..e69de29bb2d diff --git a/dojo/tools/opf/parser.py b/dojo/tools/opf/parser.py new file mode 100644 index 00000000000..7fc23ddb044 --- /dev/null +++ b/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"", "\n", value, flags=re.IGNORECASE) + text = re.sub(r"", "\n", text, flags=re.IGNORECASE) + text = re.sub(r"]*>", "- ", 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 ' ' ’ do not leak through. + text = html.unescape(text) + return re.sub(r"\n{3,}", "\n\n", text).strip() diff --git a/unittests/scans/opf/many_findings.json b/unittests/scans/opf/many_findings.json new file mode 100644 index 00000000000..a4b9f58658f --- /dev/null +++ b/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": "

The q parameter is concatenated into the query without parameterisation.

", + "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"] } + ] +} diff --git a/unittests/scans/opf/no_findings.json b/unittests/scans/opf/no_findings.json new file mode 100644 index 00000000000..ac63fc0c850 --- /dev/null +++ b/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": [] +} diff --git a/unittests/tools/test_opf_parser.py b/unittests/tools/test_opf_parser.py new file mode 100644 index 00000000000..1328ba37e38 --- /dev/null +++ b/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("

", 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, ') must decode. + doc = { + "opfVersion": "1.1", + "findings": [ + { + "title": "Entity handling", + "severity": "low", + "impact": "It’s the user's café 'else'.", + }, + ], + } + 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)