Add isMalicious expansion and hover module

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

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"]