feat: [criminal_ip] add IP and domain report enrichment

documentation/expansion/criminal_ip.md

@@ -0,0 +1,11 @@
+# Criminal IP Enrichment
+
+Retrieve IPv4 risk and infrastructure reports, existing domain scan listings, or a completed domain scan by numeric text ID. Provider JSON retains risk scores, flags and available vulnerability data. submit_unknown defaults to false; true submits a domain only when its report listing is empty. Submission returns a scan receipt without polling. Use its scan_id as text input to retrieve the full report.
+
+Input: ip-src, ip-dst, domain, hostname, text.
+
+Configuration: api_key, submit_unknown, max_results. Required: api_key.
+
+max_results defaults to 25 and accepts 1–100. New attributes use `to_ids=false`; the source UUID, detection flag and metadata are preserved. Objects reference the source attribute. Requests verify TLS, reject redirects and use connection/read timeouts of 5/30 seconds. API errors do not echo credentials. No automatic submission retries or background polling occur.
+
+Reference: https://www.criminalip.io/developer/api

misp_modules/modules/expansion/criminal_ip.py

@@ -0,0 +1,220 @@
+from urllib.parse import urlsplit
+import ipaddress
+import json
+import re
+import zipfile
+from uuid import UUID
+import requests
+from pymisp import MISPEvent, MISPObject
+from pymisp.exceptions import PyMISPError
+
+
+class ModuleError(ValueError):
+    """A diagnostic safe to show to the analyst."""
+
+
+def _integer(config, key, default, maximum=100):
+    value = config.get(key)
+    if value is None or value == "":
+        return default
+    if isinstance(value, bool) or not str(value).isdigit() or (not 1 <= int(value) <= maximum):
+        raise ModuleError(f"{key} must be an integer between 1 and {maximum}.")
+    return int(value)
+
+
+def _enabled(config, key, default=False):
+    value = config.get(key)
+    if value is None or value == "":
+        return default
+    if isinstance(value, bool):
+        return value
+    if str(value).lower() not in ("true", "false", "1", "0"):
+        raise ModuleError(f"{key} must be true or false.")
+    return str(value).lower() in ("true", "1")
+
+
+def _normalise(kind, value):
+    if not isinstance(value, str) or not value.strip() or any(ord(c) < 32 for c in value):
+        raise ModuleError("A nonempty string value is required.")
+    value = value.strip()
+    if kind in ("md5", "sha1", "sha256", "x509-fingerprint-sha256"):
+        length = {"md5": 32, "sha1": 40}.get(kind, 64)
+        if not re.fullmatch("[a-fA-F0-9]{%d}" % length, value):
+            raise ModuleError("Invalid hash.")
+        return value.lower()
+    if kind in ("ip-src", "ip-dst"):
+        return str(ipaddress.ip_address(value))
+    if kind in ("domain", "hostname"):
+        value = value.rstrip(".").encode("idna").decode("ascii").lower()
+        if len(value) > 253 or not all(
+            re.fullmatch("[a-z0-9](?:[a-z0-9-]{0,61}[a-z0-9])?", p) for p in value.split(".")
+        ):
+            raise ModuleError("Invalid DNS name.")
+    if kind == "vulnerability":
+        value = value.upper()
+        if not re.fullmatch("CVE-[0-9]{4}-[0-9]{4,}", value):
+            raise ModuleError("A CVE identifier is required.")
+    return value
+
+
+def _call(method, url, **kwargs):
+    try:
+        response = requests.request(method, url, timeout=(5, 30), verify=True, allow_redirects=False, **kwargs)
+    except requests.RequestException:
+        raise ModuleError(
+            "The service request failed or timed out; check the service before repeating a submission."
+        ) from None
+    if response.status_code in (204, 404) and method == "GET":
+        return None
+    if response.status_code not in (200, 201, 202):
+        raise ModuleError(f"The service returned HTTP {response.status_code}.")
+    try:
+        return response.json()
+    except ValueError:
+        raise ModuleError("The service returned invalid JSON.") from None
+
+
+def _object(event, attribute, name):
+    obj = MISPObject(name)
+    obj.add_reference(attribute["uuid"], "describes")
+    event.add_object(obj)
+    return obj
+
+
+def _annotation(event, attribute, title, data):
+    obj = _object(event, attribute, "annotation")
+    obj.comment = title
+    obj.add_attribute("text", value=json.dumps(data, sort_keys=True), to_ids=False, disable_correlation=True)
+    return obj
+
+
+def handler(q=False):
+    if q is False:
+        return False
+    try:
+        request = json.loads(q)
+        if not isinstance(request, dict) or not isinstance(request.get("attribute"), dict):
+            raise ModuleError("An attribute containing type, value and uuid is required.")
+        attribute = request["attribute"]
+        UUID(attribute["uuid"])
+        if attribute["type"] not in mispattributes["input"]:
+            raise ModuleError("Unsupported attribute type.")
+        value = _normalise(attribute["type"], attribute["value"])
+        config = request.get("config") or {}
+        if not isinstance(config, dict):
+            raise ModuleError("The config must be an object.")
+        for key in REQUIRED:
+            if not isinstance(config.get(key), str) or not config[key].strip():
+                raise ModuleError(f"Missing {key} configuration.")
+        limit = _integer(config, "max_results", 25)
+        event = MISPEvent()
+        event.add_attribute(**attribute)
+        _expand(event, attribute, value, config, limit)
+        result = json.loads(event.to_json())
+        return {"results": {key: result.get(key, []) for key in ("Attribute", "Object")}}
+    except ModuleError as exc:
+        return {"error": str(exc)}
+    except (
+        ValueError,
+        TypeError,
+        KeyError,
+        AttributeError,
+        IndexError,
+        OSError,
+        RuntimeError,
+        zipfile.BadZipFile,
+        PyMISPError,
+    ):
+        return {"error": "Invalid request or service response."}
+
+
+def introspection():
+    return mispattributes
+
+
+def version():
+    return dict(moduleinfo, config=moduleconfig)
+
+
+moduleinfo = {
+    "version": "1.0",
+    "author": "MISP Project",
+    "name": "Criminal IP Enrichment",
+    "description": (
+        "Retrieve IPv4 risk and infrastructure reports, existing domain scan listings, or a completed domain scan by"
+        " numeric text ID. Provider JSON retains risk scores, flags and available vulnerability data. submit_unknown"
+        " defaults to false; true submits a domain only when its report listing is empty. Submission returns a scan"
+        " receipt without polling. Use its scan_id as text input to retrieve the full report."
+    ),
+    "module-type": ["expansion"],
+    "data-source": "external",
+    "logo": "",
+    "requirements": ["requests", "pymisp"],
+    "features": (
+        "Retrieve IPv4 risk and infrastructure reports, existing domain scan listings, or a completed domain scan by"
+        " numeric text ID. Provider JSON retains risk scores, flags and available vulnerability data. submit_unknown"
+        " defaults to false; true submits a domain only when its report listing is empty. Submission returns a scan"
+        " receipt without polling. Use its scan_id as text input to retrieve the full report."
+    ),
+    "references": ["https://www.criminalip.io/developer/api"],
+    "input": "ip-src, ip-dst, domain, hostname, text",
+    "output": "MISP objects linked to the original attribute.",
+}
+mispattributes = {"input": ["ip-src", "ip-dst", "domain", "hostname", "text"], "format": "misp_standard"}
+moduleconfig = ["api_key", "submit_unknown", "max_results"]
+REQUIRED = ["api_key"]
+
+
+def _response(method, path, config, **kwargs):
+    data = _call(method, "https://api.criminalip.io" + path, headers={"x-api-key": config["api_key"]}, **kwargs)
+    if data is None:
+        return None
+    if not isinstance(data, dict) or data.get("status", 200) != 200:
+        raise ModuleError("Criminal IP rejected the request.")
+    return data
+
+
+def _expand(event, attribute, value, config, limit):
+    kind = attribute["type"]
+    if kind in ("ip-src", "ip-dst"):
+        if ipaddress.ip_address(value).version != 4:
+            raise ModuleError("Criminal IP IP reports require IPv4.")
+        data = _response("GET", "/v1/asset/ip/report", config, params={"ip": value, "full": "true"})
+    elif kind == "text":
+        if not re.fullmatch("[0-9]+", value):
+            raise ModuleError("text input must contain a domain scan ID.")
+        data = _response("GET", "/v2/domain/report/" + value, config)
+    else:
+        data = _response("GET", "/v1/domain/reports", config, params={"query": value, "offset": 0})
+        rows = (data or {}).get("data", {}).get("reports", [])
+
+        def exact(row):
+            candidate = row.get("url", "")
+            parsed = urlsplit(candidate if "://" in candidate else "https://" + candidate)
+            try:
+                return _normalise("hostname", parsed.hostname or "") == value
+            except (ValueError, TypeError):
+                return False
+
+        rows = [row for row in rows if isinstance(row, dict) and exact(row)]
+
+        if not rows and _enabled(config, "submit_unknown"):
+            receipt = _response("POST", "/v1/domain/scan", config, files={"query": (None, value)})
+            if not (receipt or {}).get("data", {}).get("scan_id"):
+                raise ModuleError("Criminal IP did not return a scan receipt; check the service before retrying.")
+            _annotation(event, attribute, "Domain scan submitted", receipt)
+            return
+        if rows:
+            scan_id = str(rows[0]["scan_id"])
+            if not re.fullmatch(r"[0-9]+", scan_id):
+                raise ModuleError("Criminal IP returned an invalid scan ID.")
+            data = _response("GET", "/v2/domain/report/" + scan_id, config)
+    if data and (kind in ("ip-src", "ip-dst") or data.get("data")):
+        _annotation(event, attribute, moduleinfo["name"], data)
+        if kind in ("ip-src", "ip-dst"):
+            for row in data.get("port", {}).get("data", [])[:limit]:
+                port = row.get("open_port_no")
+                if port is not None:
+                    obj = _object(event, attribute, "network-connection")
+                    obj.add_attribute(kind, value=value, to_ids=False)
+                    obj.add_attribute("src-port" if kind == "ip-src" else "dst-port", value=int(port), to_ids=False)

tests/test_criminal_ip.py

@@ -0,0 +1,146 @@
+import json
+from unittest.mock import Mock, patch
+
+import pytest
+import requests
+
+from misp_modules.modules.expansion import criminal_ip as module
+
+UUID = "32876e4a-916d-4935-9e51-7b69a76718f8"
+SECRET = "test-service-secret"
+
+
+def query(kind=None, value="example.com", config=None, **attribute_fields):
+    attribute = {
+        "type": kind or module.mispattributes["input"][0],
+        "value": value,
+        "uuid": UUID,
+        "to_ids": True,
+        "comment": "source context",
+        **attribute_fields,
+    }
+    return json.dumps(
+        {"attribute": attribute, "config": config if config is not None else {k: SECRET for k in module.REQUIRED}}
+    )
+
+
+def reply(payload=None, status=200):
+    return Mock(status_code=status, headers={}, json=Mock(return_value=payload))
+
+
+def assert_source(result):
+    assert "results" in result, result
+    source = next(a for a in result["results"]["Attribute"] if a["uuid"] == UUID)
+    assert source["to_ids"] is True
+    assert source["comment"] == "source context"
+    for obj in result["results"]["Object"]:
+        assert any(r["referenced_uuid"] == UUID for r in obj["ObjectReference"])
+        assert all(a["to_ids"] is False for a in obj["Attribute"])
+
+
+def objects(result, name):
+    return [obj for obj in result["results"]["Object"] if obj["name"] == name]
+
+
+def values(obj, relation):
+    return [a["value"] for a in obj["Attribute"] if a["object_relation"] == relation]
+
+
+def test_discovery():
+    assert module.handler(False) is False
+    assert module.introspection()["format"] == "misp_standard"
+    assert module.version()["config"] == module.moduleconfig
+
+
+@pytest.mark.parametrize("raw", ["{", "[]", "null", "{}", '{"attribute": []}'])
+def test_invalid_envelope(raw):
+    with patch.object(module.requests, "request") as request:
+        assert "error" in module.handler(raw)
+    request.assert_not_called()
+
+
+@pytest.mark.parametrize("status", [301, 401, 403, 429, 503])
+def test_http_errors_are_sanitized(status):
+    with patch.object(module.requests, "request", return_value=reply({"error": SECRET}, status)):
+        with pytest.raises(module.ModuleError) as failure:
+            module._call("GET", "https://service.example/report")
+    assert str(status) in str(failure.value)
+    assert SECRET not in str(failure.value)
+
+
+def test_timeout_is_bounded_and_sanitized():
+    with patch.object(module.requests, "request", side_effect=requests.Timeout(SECRET)) as request:
+        with pytest.raises(module.ModuleError) as failure:
+            module._call("POST", "https://service.example/submit")
+    assert SECRET not in str(failure.value)
+    assert request.call_count == 1
+    assert request.call_args.kwargs["verify"] is True
+    assert request.call_args.kwargs["allow_redirects"] is False
+    assert request.call_args.kwargs["timeout"] == (5, 30)
+
+
+@pytest.mark.parametrize("raw", [0, -1, True, "bad", 101])
+def test_invalid_result_limit(raw):
+    with pytest.raises(module.ModuleError):
+        module._integer({"max_results": raw}, "max_results", 25)
+
+
+def test_ip_report_preserves_provider_scores():
+    data = {"status": 200, "ip": "192.0.2.1", "score": {"inbound": 0}, "port": {"data": [{"open_port_no": 443}]}}
+    with patch.object(module.requests, "request", return_value=reply(data)) as request:
+        result = module.handler(query(kind="ip-src", value="192.0.2.1"))
+    assert_source(result)
+    assert request.call_args.kwargs["params"] == {"ip": "192.0.2.1", "full": "true"}
+    assert objects(result, "network-connection")
+
+
+def test_empty_domain_lookup_does_not_submit_by_default():
+    with patch.object(
+        module.requests, "request", return_value=reply({"status": 200, "data": {"reports": []}})
+    ) as request:
+        result = module.handler(query(kind="domain"))
+    assert_source(result)
+    assert request.call_count == 1
+    assert request.call_args.args[0] == "GET"
+
+
+def test_submission_is_explicit_and_returns_receipt():
+    with patch.object(
+        module.requests, "request", side_effect=[reply({"data": {"reports": []}}), reply({"data": {"scan_id": 123}})]
+    ) as request:
+        result = module.handler(query(kind="domain", config={"api_key": SECRET, "submit_unknown": "true"}))
+    assert_source(result)
+    assert request.call_count == 2
+    assert request.call_args.args[0] == "POST"
+
+
+def test_existing_domain_scan_is_retrieved_without_resubmission():
+    replies = [
+        reply({"status": 200, "data": {"reports": [{"scan_id": 123, "url": "https://example.com/path"}]}}),
+        reply({"status": 200, "data": {"summary": {"url_phishing_prob": 0}}}),
+    ]
+    with patch.object(module.requests, "request", side_effect=replies) as request:
+        result = module.handler(query(kind="domain"))
+    assert_source(result)
+    assert request.call_count == 2
+    assert request.call_args.args == ("GET", "https://api.criminalip.io/v2/domain/report/123")
+    assert json.loads(values(objects(result, "annotation")[0], "text")[0])["data"]["summary"]["url_phishing_prob"] == 0
+
+
+def test_domain_search_ignores_nonmatching_hosts():
+    replies = [
+        reply(
+            {
+                "data": {
+                    "reports": [
+                        {"scan_id": 1, "url": "https://other.example"},
+                        {"scan_id": 2, "url": "https://example.com/path"},
+                    ]
+                }
+            }
+        ),
+        reply({"data": {"summary": {}}}),
+    ]
+    with patch.object(module.requests, "request", side_effect=replies) as request:
+        assert_source(module.handler(query(kind="domain")))
+    assert request.call_args.args[1].endswith("/report/2")