From db918780b4299a38dfb4ea6943dcc9bd4ec8d782 Mon Sep 17 00:00:00 2001 From: Alexandre Dulaunoy Date: Thu, 29 Oct 2026 18:06:41 +0100 Subject: [PATCH 1/2] feat: [criminal_ip] add IP and domain report enrichment --- documentation/expansion/criminal_ip.md | 11 + misp_modules/modules/expansion/criminal_ip.py | 208 ++++++++++++++++++ tests/test_criminal_ip.py | 127 +++++++++++ 3 files changed, 346 insertions(+) create mode 100644 documentation/expansion/criminal_ip.md create mode 100644 misp_modules/modules/expansion/criminal_ip.py create mode 100644 tests/test_criminal_ip.py diff --git a/documentation/expansion/criminal_ip.md b/documentation/expansion/criminal_ip.md new file mode 100644 index 00000000..7eb9767e --- /dev/null +++ b/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 diff --git a/misp_modules/modules/expansion/criminal_ip.py b/misp_modules/modules/expansion/criminal_ip.py new file mode 100644 index 00000000..e03eff87 --- /dev/null +++ b/misp_modules/modules/expansion/criminal_ip.py @@ -0,0 +1,208 @@ +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}) + 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", []) + 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("ip-dst", value=value, to_ids=False) + obj.add_attribute("dst-port", value=int(port), to_ids=False) diff --git a/tests/test_criminal_ip.py b/tests/test_criminal_ip.py new file mode 100644 index 00000000..d98c82ea --- /dev/null +++ b/tests/test_criminal_ip.py @@ -0,0 +1,127 @@ +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"} + 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}]}}), + 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 From 6e881a4f620e7f0d7088bd5bab9945dd5a2ed539 Mon Sep 17 00:00:00 2001 From: Alexandre Dulaunoy Date: Thu, 29 Oct 2026 23:22:22 +0100 Subject: [PATCH 2/2] fix: [criminal_ip] address review findings --- misp_modules/modules/expansion/criminal_ip.py | 18 ++++++++++++--- tests/test_criminal_ip.py | 23 +++++++++++++++++-- 2 files changed, 36 insertions(+), 5 deletions(-) diff --git a/misp_modules/modules/expansion/criminal_ip.py b/misp_modules/modules/expansion/criminal_ip.py index e03eff87..335dfca7 100644 --- a/misp_modules/modules/expansion/criminal_ip.py +++ b/misp_modules/modules/expansion/criminal_ip.py @@ -1,3 +1,4 @@ +from urllib.parse import urlsplit import ipaddress import json import re @@ -178,7 +179,7 @@ def _expand(event, attribute, value, config, limit): 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}) + 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.") @@ -186,6 +187,17 @@ def _expand(event, attribute, value, config, limit): 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"): @@ -204,5 +216,5 @@ def _expand(event, attribute, value, config, limit): port = row.get("open_port_no") if port is not None: obj = _object(event, attribute, "network-connection") - obj.add_attribute("ip-dst", value=value, to_ids=False) - obj.add_attribute("dst-port", value=int(port), to_ids=False) + 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) diff --git a/tests/test_criminal_ip.py b/tests/test_criminal_ip.py index d98c82ea..978ac712 100644 --- a/tests/test_criminal_ip.py +++ b/tests/test_criminal_ip.py @@ -90,7 +90,7 @@ def test_ip_report_preserves_provider_scores(): 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"} + assert request.call_args.kwargs["params"] == {"ip": "192.0.2.1", "full": "true"} assert objects(result, "network-connection") @@ -116,7 +116,7 @@ def test_submission_is_explicit_and_returns_receipt(): def test_existing_domain_scan_is_retrieved_without_resubmission(): replies = [ - reply({"status": 200, "data": {"reports": [{"scan_id": 123}]}}), + 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: @@ -125,3 +125,22 @@ def test_existing_domain_scan_is_retrieved_without_resubmission(): 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")