diff --git a/misp_modules/modules/expansion/ismalicious.py b/misp_modules/modules/expansion/ismalicious.py new file mode 100644 index 00000000..2a55410b --- /dev/null +++ b/misp_modules/modules/expansion/ismalicious.py @@ -0,0 +1,144 @@ +import json + +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", + "author": "isMalicious", + "description": "Query isMalicious for IP, domain, hostname, and URL reputation.", + "module-type": ["expansion", "hover"], + "name": "isMalicious Lookup", + "logo": "", + "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": "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" + + +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): + 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 _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 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."} + + 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: + 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, + ) + 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(): + return mispattributes + + +def version(): + moduleinfo["config"] = moduleconfig + return moduleinfo 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"]