diff --git a/misp_modules/modules/import_mod/darkmoon_import.py b/misp_modules/modules/import_mod/darkmoon_import.py new file mode 100644 index 00000000..bdac8e83 --- /dev/null +++ b/misp_modules/modules/import_mod/darkmoon_import.py @@ -0,0 +1,183 @@ +import base64 +import json + +from pymisp import MISPObject + +misperrors = {"error": "Error"} + +moduleinfo = { + "version": "0.1", + "author": "ASC-IT", + "description": "Module to import the JSON findings of the Darkmoon autonomous AI penetration testing platform.", + "module-type": ["import"], + "name": "Darkmoon Import", + "logo": "", + "requirements": ["PyMISP"], + "features": ( + "Takes the JSON findings file written by the Darkmoon open source CLI (either a list of findings or an object" + " with a `findings` list) and creates one `vulnerability` object per finding, with the CVE, the CVSS score and" + " vector, the description, the remediation and the evidence that proves it. The endpoint of the finding is" + " added as a `url` object that the vulnerability affects. The severity, the exploitation status (exploited," + " confirmed or unconfirmed) and the MITRE ATT&CK technique are added as tags." + ), + "references": ["https://github.com/ASCIT31/Dark-Moon"], + "input": "Darkmoon JSON findings file", + "output": "MISP vulnerability and url objects", +} + +moduleconfig = [] + +mispattributes = { + "inputSource": ["file"], + "output": ["MISP objects"], + "format": "misp_standard", +} + +userConfig = { + "include_unconfirmed": { + "type": "Boolean", + "message": "Include findings whose exploitation could not be confirmed", + "checked": "true", + }, +} + +SEVERITIES = ("critical", "high", "medium", "low", "info") +EXPLOITATION_STATUSES = { + "exploited": "proven with a working exploit", + "confirmed": "confirmed", + "unconfirmed": "not confirmed", +} + + +def _is_enabled(value, default=True): + if value is None: + return default + if isinstance(value, str): + return value.strip().lower() in ("1", "true", "yes", "on") + return bool(value) + + +def get_findings(data): + """Return the list of findings of a Darkmoon report, or None when the data is not a Darkmoon report.""" + if isinstance(data, dict): + data = data.get("findings") + if not isinstance(data, list) or not data: + return None + for finding in data: + # `discovered_by_agent` and the exploitation `status` are specific to Darkmoon + if not ( + isinstance(finding, dict) + and finding.get("title") + and finding.get("discovered_by_agent") + and str(finding.get("status", "")).lower() in EXPLOITATION_STATUSES + ): + return None + return data + + +def _description(finding): + parts = [str(finding.get("description") or "").strip()] + if finding.get("remediation"): + parts.append("Remediation: {}".format(str(finding["remediation"]).strip())) + if finding.get("evidence_explanation"): + parts.append("Evidence: {}".format(str(finding["evidence_explanation"]).strip())) + commands = finding.get("evidence_commands") + if isinstance(commands, list) and commands: + parts.append("Commands:\n{}".format("\n".join(str(command) for command in commands))) + elif commands: + parts.append("Commands:\n{}".format(commands)) + if finding.get("evidence_logs"): + parts.append("Logs:\n{}".format(finding["evidence_logs"])) + if finding.get("raw_request"): + parts.append("Request:\n{}".format(finding["raw_request"])) + if finding.get("raw_response"): + parts.append("Response:\n{}".format(finding["raw_response"])) + return "\n\n".join(part for part in parts if part) + + +def _tags(finding, status): + tags = ["darkmoon:exploitation-status={}".format(json.dumps(status))] + severity = str(finding.get("severity") or "").lower() + if severity in SEVERITIES: + tags.append("darkmoon:severity={}".format(json.dumps(severity))) + if finding.get("category"): + tags.append("darkmoon:category={}".format(json.dumps(str(finding["category"])))) + if finding.get("discovered_by_agent"): + tags.append("darkmoon:agent={}".format(json.dumps(str(finding["discovered_by_agent"])))) + technique = finding.get("mitre_attack_id") + if technique and finding.get("mitre_attack_name"): + tags.append( + "misp-galaxy:mitre-attack-pattern={}".format( + json.dumps("{} - {}".format(finding["mitre_attack_name"], technique)) + ) + ) + return [{"name": tag} for tag in tags] + + +def _references(finding): + references = [] + if finding.get("cve"): + references.append("https://nvd.nist.gov/vuln/detail/{}".format(finding["cve"])) + technique = finding.get("mitre_attack_id") + if technique: + references.append("https://attack.mitre.org/techniques/{}/".format(str(technique).replace(".", "/"))) + return references + + +def parse_finding(finding): + """Convert one Darkmoon finding to a vulnerability object and, with an endpoint, the url object it affects.""" + status = str(finding.get("status")).lower() + vulnerability = MISPObject("vulnerability", standalone=False, comment="created by darkmoon_import") + vulnerability.add_attribute("summary", value=str(finding["title"]), Tag=_tags(finding, status)) + vulnerability.add_attribute("description", value=_description(finding)) + if finding.get("cve"): + vulnerability.add_attribute("id", value=str(finding["cve"])) + if finding.get("cvss_score") is not None: + vulnerability.add_attribute("cvss-score", value=str(finding["cvss_score"])) + if finding.get("cvss_vector"): + vulnerability.add_attribute("cvss-string", value=str(finding["cvss_vector"])) + for reference in _references(finding): + vulnerability.add_attribute("references", value=reference) + + objects = [vulnerability] + if finding.get("endpoint"): + url = MISPObject("url", standalone=False, comment="endpoint of the Darkmoon finding") + url.add_attribute("url", value=str(finding["endpoint"])) + vulnerability.add_reference(url.uuid, "affects") + objects.append(url) + return objects + + +def handler(q=False): + if q is False: + return False + request = json.loads(q) + try: + report = json.loads(base64.b64decode(request["data"]).decode("utf-8")) + except (KeyError, ValueError): + misperrors["error"] = "The input is not a valid JSON file." + return misperrors + findings = get_findings(report) + if findings is None: + misperrors["error"] = "The input is not a Darkmoon JSON findings report." + return misperrors + + config = request.get("config") or {} + include_unconfirmed = _is_enabled(config.get("include_unconfirmed")) + + objects = [] + for finding in findings: + if not include_unconfirmed and str(finding.get("status")).lower() == "unconfirmed": + continue + objects.extend(parse_finding(finding)) + return {"results": {"Object": [json.loads(misp_object.to_json()) for misp_object in objects]}} + + +def introspection(): + mispattributes["userConfig"] = userConfig + return mispattributes + + +def version(): + moduleinfo["config"] = moduleconfig + return moduleinfo diff --git a/tests/darkmoon_findings.json b/tests/darkmoon_findings.json new file mode 100644 index 00000000..1e410fff --- /dev/null +++ b/tests/darkmoon_findings.json @@ -0,0 +1,61 @@ +{ + "findings": [ + { + "title": "Unauthenticated remote code execution in file upload handler", + "severity": "critical", + "cvss_score": 9.8, + "cvss_vector": "CVSS:3.1/AV:N/AC:L/PR:N/UI:N/S:U/C:H/I:H/A:H", + "category": "remote_code_execution", + "status": "exploited", + "description": "The /upload endpoint writes the client supplied filename to disk without sanitisation, which allows a PHP web shell to be dropped and executed.", + "endpoint": "https://app.example.test/upload", + "discovered_by_agent": "php", + "remediation": "Validate and normalise uploaded filenames, store uploads outside the web root and disable script execution in the upload directory.", + "evidence_commands": [ + "curl -s -F 'file=@shell.php;filename=shell.php' https://app.example.test/upload", + "curl -s 'https://app.example.test/uploads/shell.php?cmd=id'" + ], + "evidence_logs": "uid=33(www-data) gid=33(www-data) groups=33(www-data)", + "evidence_explanation": "A PHP web shell was uploaded and executed with the id command, confirming code execution as www-data.", + "cve": "CVE-2026-12345", + "mitre_attack_id": "T1190", + "mitre_attack_name": "Exploit Public-Facing Application", + "iso27001_control": "A.8.26", + "node_id": "finding-0001", + "plugin_or_component": "upload-handler", + "raw_request": "POST /upload HTTP/1.1\nHost: app.example.test\nContent-Type: multipart/form-data; boundary=x\n\n--x\nContent-Disposition: form-data; name=\"file\"; filename=\"shell.php\"\n\n\n--x--", + "raw_response": "HTTP/1.1 200 OK\nContent-Type: application/json\n\n{\"stored\":\"/uploads/shell.php\"}" + }, + { + "title": "Stored cross-site scripting in profile display name", + "severity": "high", + "cvss_score": 8.2, + "cvss_vector": "CVSS:3.1/AV:N/AC:L/PR:L/UI:R/S:C/C:H/I:L/A:N", + "category": "xss_stored", + "status": "confirmed", + "description": "The display name is rendered unescaped on the public profile page, so a script payload saved in the field executes for every visitor.", + "endpoint": "https://app.example.test/profile/42", + "discovered_by_agent": "nodejs", + "remediation": "HTML-encode user supplied values on output and add a restrictive Content-Security-Policy.", + "evidence_commands": [ + "curl -s -X POST https://app.example.test/api/profile -d 'name='" + ], + "evidence_explanation": "The payload was stored and reflected unescaped in the profile HTML.", + "mitre_attack_id": "T1059.007", + "mitre_attack_name": "JavaScript", + "node_id": "finding-0002" + }, + { + "title": "Server-side request forgery in URL preview feature", + "severity": "medium", + "cvss_score": 6.5, + "category": "ssrf", + "status": "unconfirmed", + "description": "The link preview fetcher follows user supplied URLs, including internal addresses. Reaching the cloud metadata service could not be confirmed.", + "endpoint": "https://api.example.test/v1/preview?url=http://example.org", + "discovered_by_agent": "nodejs", + "remediation": "Resolve and validate the target host against an allow list and block link-local and private address ranges.", + "node_id": "finding-0003" + } + ] +} diff --git a/tests/test_darkmoon_import.py b/tests/test_darkmoon_import.py new file mode 100644 index 00000000..7a2677df --- /dev/null +++ b/tests/test_darkmoon_import.py @@ -0,0 +1,82 @@ +import base64 +import json +from pathlib import Path + +from misp_modules.modules.import_mod import darkmoon_import + +FIXTURE = Path(__file__).parent / "darkmoon_findings.json" + + +def _request(data, **config): + return json.dumps({"data": base64.b64encode(data).decode(), "config": config}) + + +def _objects(result, name): + return [obj for obj in result["results"]["Object"] if obj["name"] == name] + + +def _values(obj, relation): + return [attribute["value"] for attribute in obj["Attribute"] if attribute["object_relation"] == relation] + + +def test_import_findings(): + result = darkmoon_import.handler(_request(FIXTURE.read_bytes())) + vulnerabilities = _objects(result, "vulnerability") + urls = _objects(result, "url") + assert len(vulnerabilities) == 3 + assert len(urls) == 3 + + rce = vulnerabilities[0] + assert _values(rce, "summary") == ["Unauthenticated remote code execution in file upload handler"] + assert _values(rce, "id") == ["CVE-2026-12345"] + assert _values(rce, "cvss-score") == ["9.8"] + assert _values(rce, "cvss-string") == ["CVSS:3.1/AV:N/AC:L/PR:N/UI:N/S:U/C:H/I:H/A:H"] + assert _values(rce, "references") == [ + "https://nvd.nist.gov/vuln/detail/CVE-2026-12345", + "https://attack.mitre.org/techniques/T1190/", + ] + description = _values(rce, "description")[0] + assert "Remediation: Validate and normalise uploaded filenames" in description + assert "Commands:\ncurl -s -F" in description + assert "Logs:\nuid=33(www-data)" in description + + summary = next(a for a in rce["Attribute"] if a["object_relation"] == "summary") + tags = {tag["name"] for tag in summary["Tag"]} + assert 'darkmoon:exploitation-status="exploited"' in tags + assert 'darkmoon:severity="critical"' in tags + assert 'misp-galaxy:mitre-attack-pattern="Exploit Public-Facing Application - T1190"' in tags + + assert _values(urls[0], "url") == ["https://app.example.test/upload"] + reference = rce["ObjectReference"][0] + assert reference["relationship_type"] == "affects" + assert reference["referenced_uuid"] == urls[0]["uuid"] + + ssrf = vulnerabilities[2] + assert _values(ssrf, "id") == [] + assert _values(ssrf, "cvss-string") == [] + assert _values(ssrf, "references") == [] + + +def test_accepts_bare_list_of_findings(): + findings = json.loads(FIXTURE.read_text())["findings"] + result = darkmoon_import.handler(_request(json.dumps(findings).encode())) + assert len(_objects(result, "vulnerability")) == 3 + + +def test_exclude_unconfirmed(): + result = darkmoon_import.handler(_request(FIXTURE.read_bytes(), include_unconfirmed=False)) + summaries = [_values(obj, "summary")[0] for obj in _objects(result, "vulnerability")] + assert len(summaries) == 2 + assert "Server-side request forgery in URL preview feature" not in summaries + + +def test_rejects_other_json(): + gitleaks_like = json.dumps({"findings": [{"title": "x", "severity": "high"}]}).encode() + assert "error" in darkmoon_import.handler(_request(gitleaks_like)) + assert "error" in darkmoon_import.handler(_request(b"{}")) + assert "error" in darkmoon_import.handler(_request(b"not json")) + + +def test_introspection_and_version(): + assert darkmoon_import.introspection()["format"] == "misp_standard" + assert darkmoon_import.version()["module-type"] == ["import"]