Shodan Enterprise Module

.gitignore

@@ -2,3 +2,4 @@ __pycache__/
 .coverage*
 /data/
 /neo4j/
+.DS_Store

bbot/modules/censys_ip.py

@@ -33,9 +33,10 @@ class censys_ip(censys):
 
     async def setup(self):
         self.dns_names_limit = self.config.get("dns_names_limit", 100)
-        self.warning(
-            "This module may consume a lot of API queries. Unless you specifically want to query on each individual IP, we recommend using the censys_dns module instead."
-        )
+        if not self.config.get("in_scope_only", True):
+            self.warning(
+                "in_scope_only is disabled. This module queries each IP individually and may consume a lot of API credits!"
+            )
         return await super().setup()
 
     async def filter_event(self, event):

bbot/modules/shodan_enterprise.py

@@ -0,0 +1,178 @@
+from bbot.modules.base import BaseModule
+
+
+class shodan_enterprise(BaseModule):
+    watched_events = ["IP_ADDRESS"]
+    produced_events = ["OPEN_TCP_PORT", "TECHNOLOGY", "OPEN_UDP_PORT", "ASN", "FINDING"]
+    flags = ["passive", "safe"]
+    meta = {
+        "created_date": "2026-01-27",
+        "author": "@Control-Punk-Delete",
+        "description": "Shodan Enterprise API integration module.",
+        "auth_required": True,
+    }
+    options = {"api_key": "", "in_scope_only": True}
+    options_desc = {
+        "api_key": "Shodan API Key",
+        "in_scope_only": "Only query in-scope IPs. If False, will query up to distance 1.",
+    }
+    scope_distance_modifier = 1
+
+    base_url = "https://api.shodan.io"
+
+    async def setup(self):
+        self.api_key = self.config.get("api_key", "")
+        if not self.api_key:
+            return None, "No API key specified"
+        if not self.config.get("in_scope_only", True):
+            self.warning(
+                "in_scope_only is disabled. This module queries each IP individually and may consume a lot of API credits!"
+            )
+        return True
+
+    async def filter_event(self, event):
+        in_scope_only = self.config.get("in_scope_only", True)
+        max_scope_distance = 0 if in_scope_only else (self.scan.scope_search_distance + 1)
+        if event.scope_distance > max_scope_distance:
+            return False, "event is not in scope"
+        return True
+
+    async def handle_event(self, event):
+        ip = event.data
+        url = f"{self.base_url}/shodan/host/{self.helpers.quote(ip)}?key={{api_key}}"
+        r = await self.api_request(url)
+        if r is None:
+            self.warning(f"No response from Shodan API for {ip}")
+            return
+        status_code = getattr(r, "status_code", 0)
+        if status_code == 404:
+            self.warning(f"No Shodan data about {ip}")
+            return
+        if not getattr(r, "is_success", False):
+            self.warning(f"Shodan API error for {ip} (status {status_code})")
+            return
+        try:
+            host = r.json()
+        except Exception as e:
+            self.warning(f"Failed to parse Shodan API response for {ip}: {e}")
+            return
+
+        # ASN Extraction
+        asn_raw = host.get("asn", "")
+        if asn_raw:
+            asn = {
+                "asn": asn_raw[2:] if asn_raw.startswith("AS") else asn_raw,
+                "name": host.get("org", ""),
+                "description": host.get("isp", ""),
+                "country": host.get("country_code", ""),
+            }
+            await self.emit_event(
+                asn,
+                "ASN",
+                parent=event,
+                tags=host.get("tags") or [],
+                context=f"{{module}} queried Shodan API for {ip} and found ASN",
+            )
+
+        if "data" not in host:
+            self.warning(f"No Shodan data about {ip}")
+            return
+
+        # NIST cvss score severity mapping
+        severity_map = {"NONE": 0.0, "LOW": 0.1, "MEDIUM": 4.0, "HIGH": 7.0, "CRITICAL": 9.0}
+
+        for data in host["data"]:
+            # TECHNOLOGY Extraction
+            ## TECHNOLOGY CPE Formats
+            for technology in data.get("cpe", []):
+                tech = {"technology": technology, "host": data.get("ip_str"), "port": data.get("port")}
+                await self.emit_event(
+                    tech,
+                    "TECHNOLOGY",
+                    parent=event,
+                    tags=data.get("tags") or [],
+                    context=f"{{module}} queried Shodan API for {ip} and found TECHNOLOGY: {technology}",
+                )
+
+            for technology in data.get("cpe23", []):
+                tech = {"technology": technology, "host": data.get("ip_str"), "port": data.get("port")}
+                await self.emit_event(
+                    tech,
+                    "TECHNOLOGY",
+                    parent=event,
+                    tags=data.get("tags") or [],
+                    context=f"{{module}} queried Shodan API for {ip} and found TECHNOLOGY: {technology}",
+                )
+
+            # TECHNOLOGY Additional Formats
+            if "product" in data:
+                tech = {
+                    "technology": data.get("product"),
+                    "host": data.get("ip_str"),
+                    "port": data.get("port"),
+                }
+                await self.emit_event(
+                    tech,
+                    "TECHNOLOGY",
+                    parent=event,
+                    tags=data.get("tags") or [],
+                    context=f"{{module}} queried Shodan API for {ip} and found TECHNOLOGY: {data['product']}",
+                )
+
+            if "http" in data:
+                if "components" in data["http"]:
+                    for technology in data["http"]["components"]:
+                        tech = {"technology": technology, "host": data.get("ip_str"), "port": data.get("port")}
+                        tags = list(data["http"]["components"][technology].get("categories", []))
+                        tags.append("web-technology")
+                        await self.emit_event(
+                            tech,
+                            "TECHNOLOGY",
+                            parent=event,
+                            tags=tags,
+                            context=f"{{module}} queried Shodan API for {ip} and found TECHNOLOGY: {technology}",
+                        )
+
+            # OPEN_TCP_PORT, OPEN_UDP_PORT Extraction
+            if "port" in data and "transport" in data:
+                if data["transport"] == "tcp":
+                    await self.emit_event(
+                        self.helpers.make_netloc(ip, data.get("port")),
+                        "OPEN_TCP_PORT",
+                        parent=event,
+                        tags=data.get("tags") or [],
+                        context=f"{{module}} queried Shodan API for {ip} and found OPEN_TCP_PORT: {data.get('port')}",
+                    )
+                elif data["transport"] == "udp":
+                    await self.emit_event(
+                        self.helpers.make_netloc(ip, data.get("port")),
+                        "OPEN_UDP_PORT",
+                        parent=event,
+                        tags=data.get("tags") or [],
+                        context=f"{{module}} queried Shodan API for {ip} and found OPEN_UDP_PORT: {data.get('port')}",
+                    )
+                else:
+                    self.warning(f"Unknown transport {data['transport']}")
+
+            # FINDING Extraction
+            if "vulns" in data:
+                for cve, vuln_data in data["vulns"].items():
+                    cvss = vuln_data.get("cvss", 0)
+                    severity = max(
+                        (level for level, threshold in severity_map.items() if cvss >= threshold),
+                        key=lambda x: severity_map[x],
+                    )
+                    vuln = {
+                        "name": "Shodan - Possible Vulnerabilities",
+                        "host": data.get("ip_str"),
+                        "severity": severity,
+                        "description": cve,
+                        "confidence": "LOW",
+                    }
+                    await self.emit_event(
+                        vuln,
+                        "FINDING",
+                        parent=event,
+                        tags=[],
+                        context=f"{{module}} queried Shodan API for {ip} and found FINDING {cve}",
+                    )

bbot/test/test_step_2/module_tests/test_module_shodan_enterprise.py

@@ -0,0 +1,76 @@
+from .base import ModuleTestBase
+
+
+class TestShodan_Enterprise(ModuleTestBase):
+    targets = ["8.8.8.8"]
+    config_overrides = {"modules": {"shodan_enterprise": {"api_key": "deadbeef"}}}
+
+    async def setup_before_prep(self, module_test):
+        module_test.httpx_mock.add_response(
+            url="https://api.shodan.io/shodan/host/8.8.8.8?key=deadbeef",
+            json={
+                "asn": "AS15169",
+                "org": "Google LLC",
+                "isp": "Google LLC",
+                "country_code": "US",
+                "tags": ["cloud", "public-dns", "verified"],
+                "data": [
+                    {
+                        "ip_str": "8.8.8.8",
+                        "port": 53,
+                        "transport": "tcp",
+                        "product": "Google Public DNS",
+                        "tags": ["dns", "nameserver"],
+                        "cpe": ["cpe:/a:google:dns"],
+                        "cpe23": ["cpe:2.3:a:google:dns:1.0:*:*:*:*:*:*:*"],
+                        "http": {
+                            "components": {
+                                "OpenSSL": {"categories": ["web-crypto"]},
+                                "nginx": {"categories": ["web-servers"]},
+                            }
+                        },
+                        "vulns": {
+                            "CVE-2021-12345": {"cvss": 7.5},
+                            "CVE-2022-11111": {"cvss": 9.7},
+                            "CVE-2020-00001": {"cvss": 2.5},
+                        },
+                    },
+                    {
+                        "ip_str": "8.8.8.8",
+                        "port": 53,
+                        "transport": "udp",
+                        "product": "Google Public DNS",
+                        "tags": ["dns"],
+                        "cpe": [],
+                        "cpe23": [],
+                        "http": {},
+                        "vulns": {},
+                    },
+                ],
+            },
+        )
+
+    def check(self, module_test, events):
+        asn_events = [e for e in events if e.type == "ASN"]
+        assert asn_events, "No ASN event detected"
+        asn = asn_events[0].data
+        assert asn.get("asn") == "15169"
+        assert asn.get("name") == "Google LLC"
+        assert asn.get("country") == "US"
+        assert asn.get("description") == "Google LLC"
+        tcp_ports = [e.data for e in events if e.type == "OPEN_TCP_PORT"]
+        udp_ports = [e.data for e in events if e.type == "OPEN_UDP_PORT"]
+        assert any("8.8.8.8:53" in str(p) for p in tcp_ports), "TCP port 53 not detected"
+        assert any("8.8.8.8:53" in str(p) for p in udp_ports), "UDP port 53 not detected"
+        finding_events = [e for e in events if e.type == "FINDING"]
+        finding_map = {e.data.get("description"): e.data.get("severity") for e in finding_events}
+        assert "CVE-2021-12345" in finding_map
+        assert finding_map["CVE-2021-12345"] == "HIGH"
+        assert "CVE-2020-00001" in finding_map
+        assert finding_map["CVE-2020-00001"] == "LOW"
+        tech_events = [e for e in events if e.type == "TECHNOLOGY"]
+        tech_names = {e.data.get("technology") for e in tech_events}
+        assert "cpe:/a:google:dns" in tech_names
+        assert "google public dns" in tech_names
+        assert "openssl" in tech_names
+        assert "nginx" in tech_names

docs/scanning/configuration.md

@@ -433,6 +433,8 @@ In addition to the stated options for each module, the following universal optio
 | modules.lightfuzz.disable_post                      | bool     | Disable processing of POST parameters, avoiding form submissions.                                                                                                                                                                                                                                                              | False                                                                                                                                                                                                                                                                                                                                                                                                                                                   |
 | modules.lightfuzz.enabled_submodules                | list     | A list of submodules to enable. Empty list enabled all modules.                                                                                                                                                                                                                                                                | ['sqli', 'cmdi', 'xss', 'path', 'ssti', 'crypto', 'serial', 'esi']                                                                                                                                                                                                                                                                                                                                                                                      |
 | modules.lightfuzz.force_common_headers              | bool     | Force emit commonly exploitable parameters that may be difficult to detect                                                                                                                                                                                                                                                     | False                                                                                                                                                                                                                                                                                                                                                                                                                                                   |
+| modules.lightfuzz.try_get_as_post                   | bool     | For each GETPARAM, also fuzz it as a POSTPARAM (in addition to normal GET fuzzing).                                                                                                                                                                                                                                            | False                                                                                                                                                                                                                                                                                                                                                                                                                                                   |
+| modules.lightfuzz.try_post_as_get                   | bool     | For each POSTPARAM, also fuzz it as a GETPARAM (in addition to normal POST fuzzing).                                                                                                                                                                                                                                           | False                                                                                                                                                                                                                                                                                                                                                                                                                                                   |
 | modules.medusa.snmp_versions                        | list     | List of SNMP versions to attempt against the SNMP server (default ['1', '2C'])                                                                                                                                                                                                                                                 | ['1', '2C']                                                                                                                                                                                                                                                                                                                                                                                                                                             |
 | modules.medusa.snmp_wordlist                        | str      | Wordlist url for SNMP community strings, newline separated (default https://raw.githubusercontent.com/danielmiessler/SecLists/refs/heads/master/Discovery/SNMP/snmp.txt)                                                                                                                                                       | https://raw.githubusercontent.com/danielmiessler/SecLists/refs/heads/master/Discovery/SNMP/common-snmp-community-strings.txt                                                                                                                                                                                                                                                                                                                            |
 | modules.medusa.threads                              | int      | Number of communities to be tested concurrently (default 5)                                                                                                                                                                                                                                                                    | 5                                                                                                                                                                                                                                                                                                                                                                                                                                                       |

docs/scanning/presets_list.md

@@ -295,6 +295,8 @@ Discover web parameters and lightly fuzz them for vulnerabilities, with more int
         lightfuzz:
           enabled_submodules: [cmdi,crypto,path,serial,sqli,ssti,xss,esi]
           disable_post: False
+          try_post_as_get: True
+          try_get_as_post: True
     ```
 
 Category: web
@@ -354,6 +356,7 @@ Discover web parameters and lightly fuzz them for vulnerabilities. Uses all ligh
       modules:
         lightfuzz:
           enabled_submodules: [cmdi,crypto,path,serial,sqli,ssti,xss,esi]
+          try_post_as_get: True
     ```
 
 Category: web