Add ScanMalware expansion module

misp_modules/modules/expansion/scanmalware.py

@@ -0,0 +1,261 @@
+import json
+from urllib.parse import quote, urlparse
+
+import requests
+from pymisp import MISPEvent
+
+from . import check_input_attribute, standard_error_message
+
+misperrors = {"error": "Error"}
+mispattributes = {
+    "input": ["domain", "hostname", "url", "ip-src", "ip-dst", "domain|ip"],
+    "output": ["domain", "hostname", "ip-src", "text", "link"],
+    "format": "misp_standard",
+}
+moduleinfo = {
+    "version": "1.0",
+    "author": "Jonas Lejon",
+    "description": "Query ScanMalware for subdomains, browser-observed hosts and sandbox scan results.",
+    "module-type": ["expansion", "hover"],
+    "name": "ScanMalware Lookup",
+    "logo": "scanmalware.png",
+    "requirements": [],
+    "features": (
+        "The module takes a domain, hostname, URL or IP attribute and queries the public ScanMalware API."
+        " For a domain or hostname it returns subdomains seen in Certificate Transparency DNS records, and the"
+        " hosts a browser actually resolved and requested while rendering pages on that domain. For a URL it"
+        " returns the most recent sandboxed scan of that host, with its verdict and a link to the full report."
+        " For an IP it returns the URLs in the public scan archive that resolved to that address."
+        "\n\nNo API key is required: the ScanMalware API is anonymous, and a key only raises the rate limit."
+        " Only the queried indicator leaves the MISP instance."
+    ),
+    "references": ["https://scanmalware.com", "https://scanmalware.com/api-docs"],
+    "input": "A domain, hostname, URL or IP address.",
+    "output": "Domain attributes for the hosts found, plus text and link attributes describing the scan.",
+}
+# No API key: the public API is anonymous. api_url is exposed only so an operator can
+# point the module at a different deployment; max_results bounds how many attributes a
+# single enrichment may add.
+moduleconfig = ["api_url", "max_results"]
+
+DEFAULT_API_URL = "https://scanmalware.com"
+USER_AGENT = "scanmalware-misp/1.0 (+https://github.com/MISP/misp-modules)"
+
+# The API rejects a larger value rather than clamping it.
+SUBDOMAIN_LIMIT = 5000
+
+# Cold queries have been measured at several seconds; a short timeout drops real results.
+TIMEOUT = 30
+
+# A busy domain can have well over a thousand subdomains (cloudflare.com returns ~1200).
+# Adding that many attributes to one event is unusable in practice, so results are capped
+# and the cap is reported rather than applied silently.
+DEFAULT_MAX_RESULTS = 200
+
+
+def _query_value(attribute):
+    """Return the value to look up, and whether it is an IP."""
+    value = str(attribute.get("value", "")).strip()
+    attribute_type = attribute.get("type")
+
+    if attribute_type == "domain|ip" and "|" in value:
+        return value.split("|", 1)[0], False
+    if attribute_type in ("ip-src", "ip-dst"):
+        return value, True
+    if attribute_type == "url":
+        return (urlparse(value).hostname or value), False
+    return value, False
+
+
+def _get(api_url, path, params=None):
+    """GET a JSON document, or None when the endpoint has nothing to give."""
+    try:
+        response = requests.get(
+            f"{api_url}{path}",
+            params=params,
+            headers={"User-Agent": USER_AGENT, "Accept": "application/json"},
+            timeout=TIMEOUT,
+        )
+        if response.status_code == 404:
+            return None
+        response.raise_for_status()
+        return response.json()
+    except (requests.exceptions.RequestException, ValueError):
+        return None
+
+
+class ScanMalwareParser:
+    def __init__(self, api_url, max_results=DEFAULT_MAX_RESULTS):
+        self.api_url = api_url
+        self.max_results = max_results
+        self.misp_event = MISPEvent()
+        self.found = False
+        self.added = 0
+        self.capped = False
+
+    def _add(self, **kwargs):
+        """Add one attribute, respecting the result cap."""
+        if self.added >= self.max_results:
+            self.capped = True
+            return False
+        self.misp_event.add_attribute(**kwargs)
+        self.added += 1
+        self.found = True
+        return True
+
+    def _add_hosts(self, hosts, comment):
+        for host in sorted(set(hosts)):
+            # Wildcard certificates appear literally as "*.example.com" and are not
+            # resolvable hosts.
+            if not host or host.startswith("*."):
+                continue
+            if not self._add(type="domain", value=host, comment=comment):
+                return
+
+    def parse_domain(self, domain):
+        """Subdomains from Certificate Transparency, and hosts a browser contacted."""
+        ct = _get(
+            self.api_url,
+            f"/api/v1/ct/dns/{quote(domain, safe='')}",
+            {"subdomain_limit": SUBDOMAIN_LIMIT},
+        )
+        if ct:
+            self._add_hosts(ct.get("subdomains") or [], "ScanMalware: seen in Certificate Transparency DNS records")
+            if ct.get("subdomains_truncated"):
+                self._add(
+                    type="text",
+                    value=f"ScanMalware returned a truncated subdomain list for {domain}; more exist.",
+                    comment="ScanMalware: partial result",
+                    disable_correlation=True,
+                )
+
+        # A different population: hosts a browser resolved and requested. Dev and
+        # staging hosts holding no certificate appear only here.
+        hosts = _get(
+            self.api_url,
+            f"/api/v1/hosts/{quote(domain, safe='')}",
+            {"subdomains_only": "true"},
+        )
+        if hosts:
+            self._add_hosts(hosts.get("subdomains") or [], "ScanMalware: contacted by a browser during a scan")
+
+    def parse_url(self, hostname, url_value):
+        """The most recent sandboxed scan of the host, with its verdict."""
+        scans = _get(
+            self.api_url,
+            f"/api/v1/domains/{quote(hostname, safe='')}/scans",
+            {"limit": 5, "status": "completed"},
+        )
+        rows = scans if isinstance(scans, list) else (scans or {}).get("results") or []
+        if not rows:
+            return
+
+        scan = rows[0]
+        scan_id = scan.get("scan_id")
+        if not scan_id:
+            return
+
+        result = _get(self.api_url, f"/api/v1/result/{quote(scan_id, safe='')}")
+        verdict = ((result or {}).get("security_verdict") or {}).get("verdict")
+        summary = f"ScanMalware scan of {scan.get('url', url_value)}"
+        if verdict:
+            summary = f"{summary}: {verdict}"
+
+        self._add(type="text", value=summary, comment="ScanMalware: sandbox verdict", disable_correlation=True)
+        self._add(
+            type="link",
+            value=f"{self.api_url}/result/{scan_id}",
+            comment="ScanMalware: full sandboxed report",
+            disable_correlation=True,
+        )
+
+        for row in (result or {}).get("ip_table") or []:
+            if row.get("ip"):
+                if not self._add(
+                    type="ip-src", value=row["ip"], comment="ScanMalware: contacted while rendering the page"
+                ):
+                    break
+
+    def parse_ip(self, ip_address):
+        """URLs in the public scan archive that resolved to this address."""
+        search = _get(self.api_url, f"/api/v1/search/ip/{quote(ip_address, safe='')}", {"limit": 20})
+        results = (search or {}).get("results") or []
+        if not results:
+            return
+
+        self._add(
+            type="text",
+            value=f"ScanMalware: {search.get('total', len(results))} archived scan(s) resolved to {ip_address}",
+            comment="ScanMalware: scan archive",
+            disable_correlation=True,
+        )
+
+        hosts = []
+        for row in results:
+            hostname = urlparse(row.get("url") or "").hostname
+            if hostname:
+                hosts.append(hostname)
+        self._add_hosts(hosts, "ScanMalware: scanned and resolved to this address")
+
+    def get_results(self):
+        if self.capped:
+            # Report the cap instead of returning a silently shortened list.
+            self.misp_event.add_attribute(
+                type="text",
+                value=(
+                    f"ScanMalware returned more results than the {self.max_results}-attribute limit;"
+                    " raise max_results in the module config to see the rest."
+                ),
+                comment="ScanMalware: partial result",
+                disable_correlation=True,
+            )
+        if not self.found:
+            return {"error": "No ScanMalware results for this attribute."}
+        event = json.loads(self.misp_event.to_json())
+        results = {key: event[key] for key in ("Attribute", "Object") if event.get(key)}
+        if not results:
+            return {"error": "No ScanMalware results for this attribute."}
+        return {"results": results}
+
+
+def handler(q=False):
+    if q is False:
+        return False
+    request = json.loads(q)
+
+    if not request.get("attribute") or not check_input_attribute(request["attribute"]):
+        return {"error": f"{standard_error_message}, which should contain at least a type, a value and an UUID."}
+
+    attribute = request["attribute"]
+    if attribute.get("type") not in mispattributes["input"]:
+        return {"error": "Unsupported attribute type."}
+
+    query, is_ip = _query_value(attribute)
+    if not query:
+        return {"error": "The provided attribute value is empty."}
+
+    config = request.get("config") or {}
+    api_url = str(config.get("api_url") or DEFAULT_API_URL).rstrip("/")
+
+    try:
+        max_results = int(config.get("max_results") or DEFAULT_MAX_RESULTS)
+    except (TypeError, ValueError):
+        max_results = DEFAULT_MAX_RESULTS
+
+    parser = ScanMalwareParser(api_url, max_results)
+    if is_ip:
+        parser.parse_ip(query)
+    elif attribute["type"] == "url":
+        parser.parse_url(query, attribute.get("value", ""))
+    else:
+        parser.parse_domain(query)
+    return parser.get_results()
+
+
+def introspection():
+    return mispattributes
+
+
+def version():
+    moduleinfo["config"] = moduleconfig
+    return moduleinfo