From cabcae3b2129baf6d97427168fab50780582d01a Mon Sep 17 00:00:00 2001 From: JVQ Date: Thu, 13 Aug 2026 11:17:26 +0200 Subject: [PATCH 1/2] Add isMalicious expansion and hover module. Closes #797 Co-authored-by: Cursor --- misp_modules/modules/expansion/ismalicious.py | 140 ++++++++++++++++++ 1 file changed, 140 insertions(+) create mode 100644 misp_modules/modules/expansion/ismalicious.py diff --git a/misp_modules/modules/expansion/ismalicious.py b/misp_modules/modules/expansion/ismalicious.py new file mode 100644 index 00000000..1759ab0a --- /dev/null +++ b/misp_modules/modules/expansion/ismalicious.py @@ -0,0 +1,140 @@ +"""isMalicious MISP expansion and hover module. + +Copy to misp_modules/modules/expansion/ismalicious.py, then open a PR against +MISP/misp-modules. Modules in that directory are auto-discovered. +""" + +from __future__ import annotations + +import json +from typing import Any + +import requests + +misperrors = {"error": "Error"} +mispattributes = { + "input": ["ip-src", "ip-dst", "hostname", "domain", "url", "domain|ip"], + "output": ["text"], +} +moduleinfo = { + "version": "1.0", + "author": "isMalicious", + "description": "Query isMalicious for IP, domain, hostname, and URL reputation.", + "module-type": ["expansion", "hover"], + "name": "isMalicious Lookup", + "logo": "", + "requirements": ["requests"], + "features": "Returns malicious flag, risk score, categories, and source count.", + "references": ["https://ismalicious.com/integrations/misp"], + "input": "ip-src, ip-dst, hostname, domain, url, domain|ip", + "output": "text attributes with score, categories, and sources", +} +moduleconfig = ["api_key", "api_url"] + +DEFAULT_API_URL = "https://api.ismalicious.com" +checking_error = 'containing at least a "type" field and a "value" field' +standard_error_message = 'This module requires an "attribute" field as input' + + +def check_input_attribute(attribute, requirements=("type", "value")): + return isinstance(attribute, dict) and all( + feature in attribute for feature in requirements + ) + + +def check_indicator( + query: str, + api_key: str, + api_url: str = DEFAULT_API_URL, + timeout: int = 30, +) -> dict[str, Any]: + if not query or not str(query).strip(): + raise ValueError("query is required") + if not api_key: + raise ValueError("api_key is required") + + response = requests.get( + f"{api_url.rstrip('/')}/check", + params={"query": str(query).strip(), "enrichment": "standard"}, + headers={"X-API-KEY": api_key, "Accept": "application/json"}, + timeout=timeout, + ) + response.raise_for_status() + return response.json() + + +def _query_from_attribute(attribute: dict[str, Any]) -> str: + value = str(attribute.get("value", "")) + if attribute.get("type") == "domain|ip" and "|" in value: + return value.split("|", 1)[0] + return value + + +def _risk_score(payload: dict[str, Any]) -> int | None: + raw = payload.get("riskScore") + if isinstance(raw, dict): + score = raw.get("score") + return int(score) if score is not None else None + if isinstance(raw, (int, float)): + return int(raw) + return None + + +def handler(q: bool | str = False): + if q is False: + return False + request = json.loads(q) if isinstance(q, str) else q + + config = request.get("config") or {} + api_key = config.get("api_key") + if not api_key: + return {"error": "isMalicious API key is missing"} + + attribute = request.get("attribute") or {} + if not check_input_attribute(attribute): + return {"error": f"{standard_error_message}, {checking_error}."} + if attribute.get("type") not in mispattributes["input"]: + return {"error": "Unsupported attribute type."} + + query = _query_from_attribute(attribute) + try: + payload = check_indicator( + query, + api_key=api_key, + api_url=config.get("api_url") or DEFAULT_API_URL, + ) + except requests.RequestException as exc: + return {"error": f"isMalicious API request failed: {exc}"} + except ValueError as exc: + return {"error": str(exc)} + + malicious = bool(payload.get("malicious")) + score = _risk_score(payload) + categories = payload.get("categories") or payload.get("classification", {}).get( + "primary" + ) + sources = payload.get("sources") or [] + source_count = len(sources) if isinstance(sources, list) else sources + + summary = ( + f"malicious={malicious} score={score if score is not None else 'n/a'} " + f"categories={categories} sources={source_count}" + ) + return { + "results": [{"types": "text", "values": summary}], + "isMalicious": { + "malicious": malicious, + "riskScore": score, + "categories": categories, + "sourceCount": source_count, + }, + } + + +def introspection(): + return mispattributes + + +def version(): + moduleinfo["config"] = moduleconfig + return moduleinfo From 8c3a3452279fba5833c9917608f8971baa9c2d99 Mon Sep 17 00:00:00 2001 From: JVQ Date: Wed, 19 Aug 2026 15:12:33 +0200 Subject: [PATCH 2/2] Align isMalicious with expansion helpers and add unit tests. The previous CI run failed on the unrelated BTC scam check live query; this rebases onto main (where that test is skipped in LiveCI) and covers the new module with mocked requests like the other recent expansions. Co-authored-by: Cursor --- misp_modules/modules/expansion/ismalicious.py | 174 +++++++++--------- tests/test_ismalicious.py | 95 ++++++++++ 2 files changed, 184 insertions(+), 85 deletions(-) create mode 100644 tests/test_ismalicious.py diff --git a/misp_modules/modules/expansion/ismalicious.py b/misp_modules/modules/expansion/ismalicious.py index 1759ab0a..2a55410b 100644 --- a/misp_modules/modules/expansion/ismalicious.py +++ b/misp_modules/modules/expansion/ismalicious.py @@ -1,20 +1,15 @@ -"""isMalicious MISP expansion and hover module. - -Copy to misp_modules/modules/expansion/ismalicious.py, then open a PR against -MISP/misp-modules. Modules in that directory are auto-discovered. -""" - -from __future__ import annotations - import json -from typing import Any import requests +from pymisp import MISPEvent + +from . import check_input_attribute, standard_error_message misperrors = {"error": "Error"} mispattributes = { "input": ["ip-src", "ip-dst", "hostname", "domain", "url", "domain|ip"], "output": ["text"], + "format": "misp_standard", } moduleinfo = { "version": "1.0", @@ -23,54 +18,30 @@ "module-type": ["expansion", "hover"], "name": "isMalicious Lookup", "logo": "", - "requirements": ["requests"], - "features": "Returns malicious flag, risk score, categories, and source count.", + "requirements": ["An isMalicious API key."], + "features": ( + "The module takes an IP, domain, hostname or URL attribute and queries GET /check on the isMalicious API." + " It returns a text summary with the malicious flag, risk score, categories and source count. Hover and" + " expansion share the same handler. The queried indicator is sent over TLS; nothing else leaves the MISP" + " instance besides the configured API key." + ), "references": ["https://ismalicious.com/integrations/misp"], - "input": "ip-src, ip-dst, hostname, domain, url, domain|ip", - "output": "text attributes with score, categories, and sources", + "input": "An IP address, domain, hostname or URL.", + "output": "Text attributes with the isMalicious reputation summary.", } moduleconfig = ["api_key", "api_url"] DEFAULT_API_URL = "https://api.ismalicious.com" -checking_error = 'containing at least a "type" field and a "value" field' -standard_error_message = 'This module requires an "attribute" field as input' -def check_input_attribute(attribute, requirements=("type", "value")): - return isinstance(attribute, dict) and all( - feature in attribute for feature in requirements - ) - - -def check_indicator( - query: str, - api_key: str, - api_url: str = DEFAULT_API_URL, - timeout: int = 30, -) -> dict[str, Any]: - if not query or not str(query).strip(): - raise ValueError("query is required") - if not api_key: - raise ValueError("api_key is required") - - response = requests.get( - f"{api_url.rstrip('/')}/check", - params={"query": str(query).strip(), "enrichment": "standard"}, - headers={"X-API-KEY": api_key, "Accept": "application/json"}, - timeout=timeout, - ) - response.raise_for_status() - return response.json() - - -def _query_from_attribute(attribute: dict[str, Any]) -> str: - value = str(attribute.get("value", "")) +def _query_from_attribute(attribute): + value = str(attribute.get("value", "")).strip() if attribute.get("type") == "domain|ip" and "|" in value: return value.split("|", 1)[0] return value -def _risk_score(payload: dict[str, Any]) -> int | None: +def _risk_score(payload): raw = payload.get("riskScore") if isinstance(raw, dict): score = raw.get("score") @@ -80,55 +51,88 @@ def _risk_score(payload: dict[str, Any]) -> int | None: return None -def handler(q: bool | str = False): +def _categories(payload): + categories = payload.get("categories") + if categories: + return categories + classification = payload.get("classification") or {} + return classification.get("primary") + + +def _source_count(payload): + sources = payload.get("sources") or [] + return len(sources) if isinstance(sources, list) else sources + + +class IsMaliciousParser: + def __init__(self): + self.misp_event = MISPEvent() + + def parse(self, payload): + malicious = bool(payload.get("malicious")) + score = _risk_score(payload) + categories = _categories(payload) + source_count = _source_count(payload) + summary = ( + f"malicious={malicious} score={score if score is not None else 'n/a'} " + f"categories={categories} sources={source_count}" + ) + self.misp_event.add_attribute( + type="text", + value=summary, + comment="isMalicious reputation summary", + disable_correlation=True, + ) + + def get_results(self): + event = json.loads(self.misp_event.to_json()) + results = {key: event[key] for key in ("Attribute",) if event.get(key)} + if not results: + return {"error": "No results from isMalicious for this attribute."} + return {"results": results} + + +def handler(q=False): if q is False: return False - request = json.loads(q) if isinstance(q, str) else q - - config = request.get("config") or {} - api_key = config.get("api_key") - if not api_key: - return {"error": "isMalicious API key is missing"} + 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.get("attribute") or {} - if not check_input_attribute(attribute): - return {"error": f"{standard_error_message}, {checking_error}."} + attribute = request["attribute"] if attribute.get("type") not in mispattributes["input"]: return {"error": "Unsupported attribute type."} + config = request.get("config") or {} + api_key = str(config.get("api_key") or "").strip() + if not api_key: + return {"error": "An isMalicious API key is required (set api_key in the module config)."} + query = _query_from_attribute(attribute) + if not query: + return {"error": "The provided attribute value is empty."} + + api_url = str(config.get("api_url") or DEFAULT_API_URL).rstrip("/") try: - payload = check_indicator( - query, - api_key=api_key, - api_url=config.get("api_url") or DEFAULT_API_URL, + response = requests.get( + f"{api_url}/check", + params={"query": query, "enrichment": "standard"}, + headers={"User-Agent": "misp-modules", "X-API-KEY": api_key, "Accept": "application/json"}, + timeout=30, ) - except requests.RequestException as exc: - return {"error": f"isMalicious API request failed: {exc}"} - except ValueError as exc: - return {"error": str(exc)} - - malicious = bool(payload.get("malicious")) - score = _risk_score(payload) - categories = payload.get("categories") or payload.get("classification", {}).get( - "primary" - ) - sources = payload.get("sources") or [] - source_count = len(sources) if isinstance(sources, list) else sources - - summary = ( - f"malicious={malicious} score={score if score is not None else 'n/a'} " - f"categories={categories} sources={source_count}" - ) - return { - "results": [{"types": "text", "values": summary}], - "isMalicious": { - "malicious": malicious, - "riskScore": score, - "categories": categories, - "sourceCount": source_count, - }, - } + response.raise_for_status() + payload = response.json() + except requests.exceptions.HTTPError as http_error: + status = http_error.response.status_code if http_error.response is not None else "unknown" + return {"error": f"isMalicious API returned HTTP status {status}."} + except requests.exceptions.RequestException as request_error: + return {"error": f"isMalicious API request failed: {request_error}."} + except ValueError: + return {"error": "isMalicious API returned an invalid JSON response."} + + parser = IsMaliciousParser() + parser.parse(payload) + return parser.get_results() def introspection(): diff --git a/tests/test_ismalicious.py b/tests/test_ismalicious.py new file mode 100644 index 00000000..b6372090 --- /dev/null +++ b/tests/test_ismalicious.py @@ -0,0 +1,95 @@ +import json +from unittest.mock import Mock, patch + +from misp_modules.modules.expansion import ismalicious + +HIT_PAYLOAD = { + "malicious": True, + "riskScore": {"score": 91}, + "categories": ["c2"], + "sources": [{"name": "feed-a"}, {"name": "feed-b"}], +} + + +class MockResponse: + def __init__(self, payload, status_code=200): + self.payload = payload + self.status_code = status_code + + def json(self): + return self.payload + + def raise_for_status(self): + if self.status_code >= 400: + raise ismalicious.requests.exceptions.HTTPError(response=self) + + +def _query(value="1.2.3.4", type_="ip-src", config=None): + attribute = {"type": type_, "value": value, "uuid": "5b582d80-7a7e-4b6a-9f22-77656e72bb3b"} + if config is None: + config = {"api_key": "k"} + return {"module": "ismalicious", "attribute": attribute, "config": config} + + +def test_ismalicious_returns_summary_attribute(): + with patch.object(ismalicious.requests, "get", return_value=MockResponse(HIT_PAYLOAD)) as mocked_get: + result = ismalicious.handler(json.dumps(_query())) + + mocked_get.assert_called_once_with( + "https://api.ismalicious.com/check", + params={"query": "1.2.3.4", "enrichment": "standard"}, + headers={"User-Agent": "misp-modules", "X-API-KEY": "k", "Accept": "application/json"}, + timeout=30, + ) + values = [attribute["value"] for attribute in result["results"]["Attribute"]] + assert values == ["malicious=True score=91 categories=['c2'] sources=2"] + assert result["results"]["Attribute"][0]["comment"] == "isMalicious reputation summary" + + +def test_ismalicious_uses_domain_side_of_composite_and_custom_api_url(): + query = _query( + value="evil.example|1.2.3.4", + type_="domain|ip", + config={"api_key": "k", "api_url": "https://example.test/"}, + ) + with patch.object(ismalicious.requests, "get", return_value=MockResponse({"malicious": False})) as mocked_get: + ismalicious.handler(json.dumps(query)) + + mocked_get.assert_called_once_with( + "https://example.test/check", + params={"query": "evil.example", "enrichment": "standard"}, + headers={"User-Agent": "misp-modules", "X-API-KEY": "k", "Accept": "application/json"}, + timeout=30, + ) + + +def test_ismalicious_missing_api_key(): + with patch.object(ismalicious.requests, "get") as mocked_get: + result = ismalicious.handler(json.dumps(_query(config={}))) + mocked_get.assert_not_called() + assert result == {"error": "An isMalicious API key is required (set api_key in the module config)."} + + +def test_ismalicious_reports_http_error(): + response = Mock(payload=None) + response.status_code = 500 + response.raise_for_status.side_effect = ismalicious.requests.exceptions.HTTPError(response=response) + with patch.object(ismalicious.requests, "get", return_value=response): + result = ismalicious.handler(json.dumps(_query())) + assert result == {"error": "isMalicious API returned HTTP status 500."} + + +def test_ismalicious_rejects_invalid_input(): + assert ismalicious.handler(json.dumps({"module": "ismalicious"}))["error"].startswith( + 'This module requires an "attribute" field' + ) + assert ismalicious.handler(json.dumps(_query(value="abc", type_="md5"))) == {"error": "Unsupported attribute type."} + + +def test_ismalicious_introspection_and_version(): + assert ismalicious.introspection() == ismalicious.mispattributes + info = ismalicious.version() + assert info["name"] == "isMalicious Lookup" + assert "api_key" in info["config"] + assert "expansion" in info["module-type"] + assert "hover" in info["module-type"]