fix tests for 3.0

Author: TheTechromancer

bbot/modules/shodan_enterprise.py

@@ -3,7 +3,7 @@
 
 class shodan_enterprise(BaseModule):
     watched_events = ["IP_ADDRESS"]
-    produced_events = ["OPEN_TCP_PORT", "TECHNOLOGY", "OPEN_UDP_PORT", "ASN", "VULNERABILITY"]
+    produced_events = ["OPEN_TCP_PORT", "TECHNOLOGY", "OPEN_UDP_PORT", "ASN", "FINDING"]
     flags = ["passive", "safe"]
     meta = {
         "created_date": "2026-01-27",
@@ -71,7 +71,7 @@ async def handle_event(self, event):
                 "ASN",
                 parent=event,
                 tags=host.get("tags") or [],
-                context=f"Shodan API {ip} request and find ASN",
+                context=f"{{module}} queried Shodan API for {ip} and found ASN",
             )
 
         if "data" not in host:
@@ -91,7 +91,7 @@ async def handle_event(self, event):
                     "TECHNOLOGY",
                     parent=event,
                     tags=data.get("tags") or [],
-                    context=f"Shodan API {ip} request and find TECHNOLOGY: {technology}",
+                    context=f"{{module}} queried Shodan API for {ip} and found TECHNOLOGY: {technology}",
                 )
 
             for technology in data.get("cpe23", []):
@@ -101,7 +101,7 @@ async def handle_event(self, event):
                     "TECHNOLOGY",
                     parent=event,
                     tags=data.get("tags") or [],
-                    context=f"Shodan API {ip} request and find TECHNOLOGY: {technology}",
+                    context=f"{{module}} queried Shodan API for {ip} and found TECHNOLOGY: {technology}",
                 )
 
             # TECHNOLOGY Additional Formats
@@ -116,7 +116,7 @@ async def handle_event(self, event):
                     "TECHNOLOGY",
                     parent=event,
                     tags=data.get("tags") or [],
-                    context=f"Shodan API {ip} request and find TECHNOLOGY: {data['product']}",
+                    context=f"{{module}} queried Shodan API for {ip} and found TECHNOLOGY: {data['product']}",
                 )
 
             if "http" in data:
@@ -130,7 +130,7 @@ async def handle_event(self, event):
                             "TECHNOLOGY",
                             parent=event,
                             tags=tags,
-                            context=f"Shodan API {ip} request and find TECHNOLOGY: {technology}",
+                            context=f"{{module}} queried Shodan API for {ip} and found TECHNOLOGY: {technology}",
                         )
 
             # OPEN_TCP_PORT, OPEN_UDP_PORT Extraction
@@ -141,20 +141,20 @@ async def handle_event(self, event):
                         "OPEN_TCP_PORT",
                         parent=event,
                         tags=data.get("tags") or [],
-                        context=f"Shodan API {ip} request and find OPEN_TCP_PORT: {data.get('port')}",
+                        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"Shodan API {ip} request and find OPEN_UDP_PORT: {data.get('port')}",
+                        context=f"{{module}} queried Shodan API for {ip} and found OPEN_UDP_PORT: {data.get('port')}",
                     )
                 else:
                     self.warning(f"Unknown transport {data['transport']}")
 
-            # VULNERABILITY Extraction
+            # FINDING Extraction
             if "vulns" in data:
                 for cve, vuln_data in data["vulns"].items():
                     cvss = vuln_data.get("cvss", 0)
@@ -163,14 +163,16 @@ async def handle_event(self, event):
                         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,
-                        "VULNERABILITY",
+                        "FINDING",
                         parent=event,
                         tags=[],
-                        context=f"Shodan API {ip} request and find VULNERABILITY {cve}",
+                        context=f"{{module}} queried Shodan API for {ip} and found FINDING {cve}",
                     )

bbot/test/test_step_2/module_tests/test_module_shodan_enterprise.py

@@ -62,15 +62,15 @@ def check(self, module_test, events):
         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"
-        vuln_events = [e for e in events if e.type == "VULNERABILITY"]
-        vuln_map = {e.data.get("description"): e.data.get("severity") for e in vuln_events}
-        assert "CVE-2021-12345" in vuln_map
-        assert vuln_map["CVE-2021-12345"] == "HIGH"
-        assert "CVE-2020-00001" in vuln_map
-        assert vuln_map["CVE-2020-00001"] == "LOW"
+        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 "google public dns" in tech_names
+        assert "openssl" in tech_names
         assert "nginx" in tech_names