add: [expansion] Shodan InternetDB

Author: adulau

documentation/expansion/shodan_internetdb.md

@@ -0,0 +1,37 @@
+# Shodan InternetDB
+
+Look up exposed services, hostnames, CPEs and vulnerabilities in the free Shodan InternetDB API.
+
+No API key is required. Supports IPv4 and IPv6. A 404 means no information is available. Hostnames and vulnerabilities are pivotable attributes; ports are grouped in an ip-port object. Service tags are retained as context, without automatically tagging the source as malicious.
+
+Input: ip-src, ip-dst. Output: An ip-port object and hostname, CPE and vulnerability attributes.
+
+Configuration: none. Required: none.
+Requests use a 30 second timeout and verify TLS certificates. Errors distinguish
+missing configuration, invalid input, authentication failures, rate limits and
+invalid responses. No API credentials are included in error messages.
+
+All new enrichment attributes use `to_ids=false`. Source attributes retain their
+UUID and metadata. Returned objects reference the source attribute where possible.
+Empty lookups return the source attribute without fabricated enrichment.
+
+References:
+
+- https://internetdb.shodan.io/
+
+Example request:
+
+```json
+{
+  "module": "shodan_internetdb",
+  "attribute": {
+    "type": "ip-dst",
+    "value": "1.1.1.1",
+    "uuid": "32876e4a-916d-4935-9e51-7b69a76718f8"
+  },
+  "config": {}
+}
+```
+
+Validation: `pytest tests/test_shodan_internetdb.py` uses mocked service responses;
+no paid API credentials are needed.

misp_modules/modules/expansion/shodan_internetdb.py

@@ -0,0 +1,130 @@
+import json
+from ipaddress import ip_address
+from urllib.parse import quote
+
+import requests
+from pymisp import MISPEvent, MISPObject
+
+misperrors = {"error": "Error"}
+mispattributes = {"input": ["ip-src", "ip-dst"], "format": "misp_standard"}
+moduleconfig = []
+moduleinfo = {
+    "version": "1",
+    "author": "MISP Project",
+    "name": "Shodan InternetDB",
+    "description": "Look up exposed services, hostnames, CPEs and vulnerabilities in the free Shodan InternetDB API.",
+    "module-type": ["expansion"],
+    "logo": "",
+    "requirements": ["requests"],
+    "features": (
+        "No API key is required. Supports IPv4 and IPv6. A 404 means no information is available. Hostnames and"
+        " vulnerabilities are pivotable attributes; ports are grouped in an ip-port object. Service tags are retained"
+        " as context, without automatically tagging the source as malicious."
+    ),
+    "references": ["https://internetdb.shodan.io/"],
+    "input": "ip-src, ip-dst MISP attribute.",
+    "output": "An ip-port object and hostname, CPE and vulnerability attributes.",
+}
+TIMEOUT = 30
+
+
+class ModuleError(ValueError):
+    """An actionable module error safe to return to the caller."""
+
+
+def _get(url, **kwargs):
+    try:
+        response = requests.get(url, timeout=TIMEOUT, allow_redirects=False, **kwargs)
+    except requests.RequestException:
+        raise ModuleError("The service request failed or timed out.") from None
+    if response.status_code == 404:
+        return None
+    if response.status_code == 204:
+        return None
+    if response.status_code != 200:
+        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 _ip(value):
+    return str(ip_address(value))
+
+
+def _annotation(event, attribute, title, data):
+    obj = MISPObject("annotation")
+    obj.comment = title
+    obj.add_attribute("text", value=json.dumps(data, sort_keys=True, indent=2), to_ids=False, disable_correlation=True)
+    obj.add_reference(attribute["uuid"], "describes")
+    event.add_object(obj)
+
+
+def _results(event):
+    data = json.loads(event.to_json())
+    return {"results": {key: data.get(key, []) for key in ("Attribute", "Object")}}
+
+
+def handler(q=False):
+    if q is False:
+        return False
+    try:
+        request = json.loads(q)
+        if not isinstance(request, dict):
+            raise ModuleError("The request must be a JSON object.")
+        attribute = request.get("attribute")
+        if not isinstance(attribute, dict) or not all((attribute.get(k) for k in ("type", "value", "uuid"))):
+            raise ModuleError("An attribute containing type, value and uuid is required.")
+        if attribute["type"] not in mispattributes["input"]:
+            raise ModuleError("Unsupported attribute type.")
+        if not isinstance(attribute["value"], str) or not attribute["value"].strip():
+            raise ModuleError("The attribute value must be a nonempty string.")
+        config = request.get("config") or {}
+        if not isinstance(config, dict):
+            raise ModuleError("The config must be a JSON object.")
+        event = MISPEvent()
+        event.add_attribute(**attribute)
+        return _expand(event, attribute, config)
+    except ModuleError as exc:
+        return {"error": str(exc)}
+    except (ValueError, TypeError, KeyError, OverflowError, OSError):
+        return {"error": "Invalid request or service response."}
+
+
+def introspection():
+    return mispattributes
+
+
+def version():
+    return dict(moduleinfo, config=moduleconfig)
+
+
+def _expand(event, attribute, config):
+    address = _ip(attribute["value"].strip())
+    data = _get("https://internetdb.shodan.io/" + quote(address, safe=""))
+    if data is None:
+        return _results(event)
+    if not isinstance(data, dict) or _ip(data.get("ip", "")) != address:
+        raise ModuleError("InternetDB returned an unexpected IP address or response.")
+    for key in ("ports", "hostnames", "cpes", "vulns", "tags"):
+        if not isinstance(data.get(key, []), list):
+            raise ModuleError("InternetDB returned an unexpected response.")
+    if data.get("ports"):
+        obj = MISPObject("ip-port")
+        obj.add_attribute(attribute["type"], value=address, to_ids=False)
+        relation = attribute["type"].replace("ip-", "") + "-port"
+        for port in dict.fromkeys(data["ports"]):
+            if isinstance(port, bool) or not isinstance(port, int) or (not 0 <= port <= 65535):
+                raise ModuleError("InternetDB returned an invalid port.")
+            obj.add_attribute(relation, value=port, to_ids=False)
+        obj.add_reference(attribute["uuid"], "describes")
+        event.add_object(obj)
+    for key, kind in (("hostnames", "hostname"), ("cpes", "cpe"), ("vulns", "vulnerability")):
+        for value in dict.fromkeys(data.get(key, [])):
+            if not isinstance(value, str) or not value:
+                raise ModuleError("InternetDB returned an invalid value.")
+            event.add_attribute(kind, value, to_ids=False, comment="Shodan InternetDB")
+    if data.get("tags"):
+        _annotation(event, attribute, "Shodan InternetDB service tags", {"tags": data["tags"]})
+    return _results(event)

tests/test_shodan_internetdb.py

@@ -0,0 +1,152 @@
+import json
+from unittest.mock import Mock, patch
+
+import pytest
+import requests
+
+from misp_modules.modules.expansion import shodan_internetdb as module
+
+UUID = "32876e4a-916d-4935-9e51-7b69a76718f8"
+SECRET = "test-api-secret"
+
+
+def query(kind="ip-dst", value="1.1.1.1", config=None):
+    return json.dumps(
+        {
+            "attribute": {"type": kind, "value": value, "uuid": UUID, "to_ids": True, "comment": "source metadata"},
+            "config": {} if config is None else config,
+        }
+    )
+
+
+def response(payload=None, status=200):
+    result = Mock(status_code=status, headers={})
+    result.json.return_value = payload
+    result.iter_content.return_value = iter([b"sample bytes"])
+    return result
+
+
+def handle(payload, **kwargs):
+    with patch.object(module.requests, "get", return_value=response(payload)) as get:
+        result = module.handler(query(**kwargs))
+    return result, get
+
+
+def objects(result, name):
+    return [obj for obj in result["results"]["Object"] if obj["name"] == name]
+
+
+def values(obj):
+    return {item["object_relation"]: item["value"] for item in obj["Attribute"]}
+
+
+def assert_source(result):
+    source = next(item for item in result["results"]["Attribute"] if item["uuid"] == UUID)
+    assert source["to_ids"] is True
+    assert source["comment"] == "source metadata"
+    for obj in result["results"]["Object"]:
+        assert any(ref["referenced_uuid"] == UUID for ref in obj["ObjectReference"])
+        assert all(item["to_ids"] is False for item in obj["Attribute"])
+    assert all(item["to_ids"] is False for item in result["results"]["Attribute"] if item["uuid"] != UUID)
+
+
+def test_discovery_contract():
+    assert module.handler(False) is False
+    assert module.introspection()["format"] == "misp_standard"
+    assert module.version()["config"] == module.moduleconfig
+    assert "expansion" in module.version()["module-type"]
+
+
+@pytest.mark.parametrize(
+    "raw", ["{", "[]", "null", "{}", '{"attribute": []}', '{"attribute":{"type":"ip-dst","value":"1.1.1.1"}}']
+)
+def test_invalid_request_never_calls_service(raw):
+    with patch.object(module.requests, "get") as get:
+        assert "error" in module.handler(raw)
+    get.assert_not_called()
+
+
+def test_unsupported_type_never_calls_service():
+    with patch.object(module.requests, "get") as get:
+        assert "error" in module.handler(query(kind="comment"))
+    get.assert_not_called()
+
+
+@pytest.mark.parametrize("value", ["", "   ", 123, ["example.com"]])
+def test_invalid_values_never_call_service(value):
+    with patch.object(module.requests, "get") as get:
+        assert "error" in module.handler(query(value=value))
+    get.assert_not_called()
+
+
+@pytest.mark.parametrize("status", [401, 403, 429, 500, 302])
+def test_http_failures_are_reported(status):
+    with patch.object(module.requests, "get", return_value=response(status=status)):
+        result = module.handler(query())
+    assert str(status) in result["error"]
+    assert SECRET not in result["error"]
+
+
+@pytest.mark.parametrize("failure", [requests.Timeout, requests.ConnectionError])
+def test_network_failures_do_not_leak_credentials(failure):
+    with patch.object(module.requests, "get", side_effect=failure("url?apikey=" + SECRET)):
+        result = module.handler(query())
+    assert "error" in result
+    assert SECRET not in result["error"]
+
+
+@pytest.mark.parametrize("status", [404, 204])
+def test_empty_service_response_preserves_source(status):
+    with patch.object(module.requests, "get", return_value=response(status=status)):
+        result = module.handler(query())
+    assert_source(result)
+    assert result["results"]["Object"] == []
+
+
+def test_invalid_service_json_is_reported():
+    reply = response()
+    reply.json.side_effect = ValueError("invalid JSON " + SECRET)
+    with patch.object(module.requests, "get", return_value=reply):
+        result = module.handler(query())
+    assert "error" in result
+    assert SECRET not in result["error"]
+
+
+@pytest.mark.parametrize("payload", [[], "wrong", 1])
+def test_unexpected_service_shape_is_reported(payload):
+    result, _ = handle(payload)
+    assert "error" in result
+
+
+PAYLOAD = {
+    "ip": "1.1.1.1",
+    "ports": [53, 443, 443],
+    "hostnames": ["one.one.one.one"],
+    "cpes": ["cpe:/a:example:server:1"],
+    "vulns": ["CVE-2025-1234"],
+    "tags": ["cdn"],
+}
+
+
+def test_host_intelligence_is_pivotable_and_referenced():
+    result, get = handle(PAYLOAD)
+    assert_source(result)
+    obj = objects(result, "ip-port")[0]
+    assert values(obj)["ip-dst"] == "1.1.1.1"
+    assert [x["value"] for x in obj["Attribute"] if x["object_relation"] == "dst-port"] == ["53", "443"]
+    assert {x["type"] for x in result["results"]["Attribute"]} == {"ip-dst", "hostname", "cpe", "vulnerability"}
+    assert json.loads(values(objects(result, "annotation")[0])["text"])["tags"] == ["cdn"]
+    get.assert_called_once_with("https://internetdb.shodan.io/1.1.1.1", timeout=30, allow_redirects=False)
+
+
+def test_ipv6_source_port_direction():
+    payload = dict(PAYLOAD, ip="2001:4860:4860::8888")
+    result, get = handle(payload, kind="ip-src", value=payload["ip"])
+    assert "src-port" in values(objects(result, "ip-port")[0])
+    assert "%3A" in get.call_args.args[0]
+
+
+@pytest.mark.parametrize("changes", [{"ip": "8.8.8.8"}, {"ports": "443"}, {"ports": [65536]}, {"ports": [True]}])
+def test_mismatched_or_invalid_results(changes):
+    result, _ = handle(dict(PAYLOAD, **changes))
+    assert "error" in result