Skip to content

Commit 692cb85

Browse files
TheTechromancerTheTechromancer
committed
use API instead of python library, add in_scope_only option, warnings
1 parent 7e5d62e commit 692cb85

File tree

3 files changed

+188
-182
lines changed

3 files changed

+188
-182
lines changed

‎bbot/modules/censys_ip.py‎

Lines changed: 4 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -33,9 +33,10 @@ class censys_ip(censys):
3333

3434
async def setup(self):
3535
self.dns_names_limit = self.config.get("dns_names_limit", 100)
36-
self.warning(
37-
"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."
38-
)
36+
if not self.config.get("in_scope_only", True):
37+
self.warning(
38+
"in_scope_only is disabled. This module queries each IP individually and may consume a lot of API credits!"
39+
)
3940
return await super().setup()
4041

4142
async def filter_event(self, event):

‎bbot/modules/shodan_enterprise.py‎

Lines changed: 143 additions & 126 deletions
Original file line numberDiff line numberDiff line change
@@ -1,159 +1,176 @@
1-
import shodan
21
from bbot.modules.base import BaseModule
32

43

54
class shodan_enterprise(BaseModule):
65
watched_events = ["IP_ADDRESS"]
76
produced_events = ["OPEN_TCP_PORT", "TECHNOLOGY", "OPEN_UDP_PORT", "ASN", "VULNERABILITY"]
8-
flags = ["passive"]
7+
flags = ["passive", "safe"]
98
meta = {
109
"created_date": "2026-01-27",
1110
"author": "@Control-Punk-Delete",
1211
"description": "Shodan Enterprise API integration module.",
12+
"auth_required": True,
13+
}
14+
options = {"api_key": "", "in_scope_only": True}
15+
options_desc = {
16+
"api_key": "Shodan API Key",
17+
"in_scope_only": "Only query in-scope IPs. If False, will query up to distance 1.",
1318
}
14-
deps_pip = ["shodan"]
15-
options = {"api_key": None}
16-
options_desc = {"api_key": "Shodan API Key"}
17-
per_host_only = True
1819
scope_distance_modifier = 1
19-
target_only = True
20+
21+
base_url = "https://api.shodan.io"
2022

2123
async def setup(self):
22-
if not self.config.get("api_key"):
24+
self.api_key = self.config.get("api_key", "")
25+
if not self.api_key:
2326
return None, "No API key specified"
24-
self.api_key = self.config.get("api_key")
27+
if not self.config.get("in_scope_only", True):
28+
self.warning(
29+
"in_scope_only is disabled. This module queries each IP individually and may consume a lot of API credits!"
30+
)
31+
return True
32+
33+
async def filter_event(self, event):
34+
in_scope_only = self.config.get("in_scope_only", True)
35+
max_scope_distance = 0 if in_scope_only else (self.scan.scope_search_distance + 1)
36+
if event.scope_distance > max_scope_distance:
37+
return False, "event is not in scope"
2538
return True
2639

2740
async def handle_event(self, event):
41+
ip = event.data
42+
url = f"{self.base_url}/shodan/host/{self.helpers.quote(ip)}?key={{api_key}}"
43+
r = await self.api_request(url)
44+
if r is None:
45+
self.warning(f"No response from Shodan API for {ip}")
46+
return
47+
status_code = getattr(r, "status_code", 0)
48+
if status_code == 404:
49+
self.warning(f"No Shodan data about {ip}")
50+
return
51+
if not getattr(r, "is_success", False):
52+
self.warning(f"Shodan API error for {ip} (status {status_code})")
53+
return
2854
try:
29-
api = shodan.Shodan(self.api_key)
30-
host = api.host(ips=event.data, history=False, minify=False)
31-
32-
# ASN Extraction
55+
host = r.json()
56+
except Exception as e:
57+
self.warning(f"Failed to parse Shodan API response for {ip}: {e}")
58+
return
59+
60+
# ASN Extraction
61+
asn_raw = host.get("asn", "")
62+
if asn_raw:
3363
asn = {
34-
"asn": host["asn"][2:],
35-
"name": host["org"],
36-
"description": host["isp"],
37-
"country": host["country_code"],
64+
"asn": asn_raw[2:] if asn_raw.startswith("AS") else asn_raw,
65+
"name": host.get("org", ""),
66+
"description": host.get("isp", ""),
67+
"country": host.get("country_code", ""),
3868
}
39-
4069
await self.emit_event(
4170
asn,
4271
"ASN",
4372
parent=event,
44-
tags=host.get("tags"),
45-
context=f"Shodan API {event.data} request and find ASN",
73+
tags=host.get("tags") or [],
74+
context=f"Shodan API {ip} request and find ASN",
4675
)
4776

48-
if "data" in host:
49-
for data in host["data"]:
50-
# TECHNOLOGY Extraction
51-
## TECHNOLOGY CPE Formats
52-
if "cpe" in data:
53-
for technology in data["cpe"]:
54-
tech = {"technology": technology, "host": data.get("ip_str"), "port": data.get("port")}
55-
await self.emit_event(
56-
tech,
57-
"TECHNOLOGY",
58-
parent=event,
59-
tags=data.get("tags") or [],
60-
context=f"Shodan API {event.data} request and find TECHNOLOGY: {technology}",
61-
)
62-
63-
if "cpe23" in data:
64-
for technology in data["cpe23"]:
65-
tech = {"technology": technology, "host": data.get("ip_str"), "port": data.get("port")}
66-
await self.emit_event(
67-
tech,
68-
"TECHNOLOGY",
69-
parent=event,
70-
tags=data.get("tags") or [],
71-
context=f"Shodan API {event.data} request and find TECHNOLOGY: {technology}",
72-
)
73-
74-
# TECHNOLOGY Additional Formats
75-
if "product" in data:
76-
tech = {
77-
"technology": data.get("product"),
78-
"host": data.get("ip_str"),
79-
"port": data.get("port"),
80-
}
81-
77+
if "data" not in host:
78+
self.warning(f"No Shodan data about {ip}")
79+
return
80+
81+
# NIST cvss score severity mapping
82+
severity_map = {"NONE": 0.0, "LOW": 0.1, "MEDIUM": 4.0, "HIGH": 7.0, "CRITICAL": 9.0}
83+
84+
for data in host["data"]:
85+
# TECHNOLOGY Extraction
86+
## TECHNOLOGY CPE Formats
87+
for technology in data.get("cpe", []):
88+
tech = {"technology": technology, "host": data.get("ip_str"), "port": data.get("port")}
89+
await self.emit_event(
90+
tech,
91+
"TECHNOLOGY",
92+
parent=event,
93+
tags=data.get("tags") or [],
94+
context=f"Shodan API {ip} request and find TECHNOLOGY: {technology}",
95+
)
96+
97+
for technology in data.get("cpe23", []):
98+
tech = {"technology": technology, "host": data.get("ip_str"), "port": data.get("port")}
99+
await self.emit_event(
100+
tech,
101+
"TECHNOLOGY",
102+
parent=event,
103+
tags=data.get("tags") or [],
104+
context=f"Shodan API {ip} request and find TECHNOLOGY: {technology}",
105+
)
106+
107+
# TECHNOLOGY Additional Formats
108+
if "product" in data:
109+
tech = {
110+
"technology": data.get("product"),
111+
"host": data.get("ip_str"),
112+
"port": data.get("port"),
113+
}
114+
await self.emit_event(
115+
tech,
116+
"TECHNOLOGY",
117+
parent=event,
118+
tags=data.get("tags") or [],
119+
context=f"Shodan API {ip} request and find TECHNOLOGY: {data['product']}",
120+
)
121+
122+
if "http" in data:
123+
if "components" in data["http"]:
124+
for technology in data["http"]["components"]:
125+
tech = {"technology": technology, "host": data.get("ip_str"), "port": data.get("port")}
126+
tags = list(data["http"]["components"][technology].get("categories", []))
127+
tags.append("web-technology")
82128
await self.emit_event(
83129
tech,
84130
"TECHNOLOGY",
85131
parent=event,
86-
tags=data.get("tags") or [],
87-
context=f"Shodan API {event.data} request and find TECHNOLOGY: {data['product']}",
132+
tags=tags,
133+
context=f"Shodan API {ip} request and find TECHNOLOGY: {technology}",
88134
)
89135

90-
if "http" in data:
91-
if "components" in data["http"]:
92-
for technology in data["http"]["components"]:
93-
tech = {"technology": technology, "host": data.get("ip_str"), "port": data.get("port")}
94-
tags = data["http"]["components"][technology]["categories"]
95-
tags.append("web-technology")
96-
await self.emit_event(
97-
tech,
98-
"TECHNOLOGY",
99-
parent=event,
100-
tags=tags or [],
101-
context=f"Shodan API {event.data} request and find TECHNOLOGY: {technology}",
102-
)
103-
104-
# OPEN_TCP_PORT, OPEN_UDP_PORT Extraction
105-
if "port" in data and "transport" in data:
106-
if data["transport"] == "tcp":
107-
await self.emit_event(
108-
self.helpers.make_netloc(event.data, data.get("port")),
109-
"OPEN_TCP_PORT",
110-
parent=event,
111-
tags=data.get("tags") or [],
112-
context=f"Shodan API {event.data} request and find TECHNOLOGY: {data.get('port')}",
113-
)
114-
115-
elif data["transport"] == "udp":
116-
await self.emit_event(
117-
self.helpers.make_netloc(event.data, data.get("port")),
118-
"OPEN_UDP_PORT",
119-
parent=event,
120-
tags=data.get("tags") or [],
121-
context=f"Shodan API {event.data} request and find TECHNOLOGY: {data.get('port')}",
122-
)
123-
124-
else:
125-
self.warning(f"[WARNING] unknown transport {data['transport']}")
126-
127-
# VULNERABILITY Extraction
128-
# NIST cvss score severity mapping
129-
severity_map = {"NONE": 0.0, "LOW": 0.1, "MEDIUM": 4.0, "HIGH": 7.0, "CRITICAL": 9.0}
130-
131-
if "vulns" in data:
132-
for item in data["vulns"]:
133-
cve = item
134-
vuln = {
135-
"host": data.get("ip_str"),
136-
"severity": max(
137-
(
138-
level
139-
for level, threshold in severity_map.items()
140-
if data["vulns"][item].get("cvss") >= threshold
141-
),
142-
key=lambda x: severity_map[x],
143-
),
144-
"description": f"{cve}",
145-
}
146-
147-
await self.emit_event(
148-
vuln,
149-
"VULNERABILITY",
150-
parent=event,
151-
tags=[],
152-
context=f"Shodan API {event.data} request and find VULNERABILITY {cve}",
153-
)
154-
155-
else:
156-
self.warning(f"No Shodan data about {event.data}")
157-
158-
except shodan.APIError as e:
159-
self.error(f"Shodan API error: {e}")
136+
# OPEN_TCP_PORT, OPEN_UDP_PORT Extraction
137+
if "port" in data and "transport" in data:
138+
if data["transport"] == "tcp":
139+
await self.emit_event(
140+
self.helpers.make_netloc(ip, data.get("port")),
141+
"OPEN_TCP_PORT",
142+
parent=event,
143+
tags=data.get("tags") or [],
144+
context=f"Shodan API {ip} request and find OPEN_TCP_PORT: {data.get('port')}",
145+
)
146+
elif data["transport"] == "udp":
147+
await self.emit_event(
148+
self.helpers.make_netloc(ip, data.get("port")),
149+
"OPEN_UDP_PORT",
150+
parent=event,
151+
tags=data.get("tags") or [],
152+
context=f"Shodan API {ip} request and find OPEN_UDP_PORT: {data.get('port')}",
153+
)
154+
else:
155+
self.warning(f"Unknown transport {data['transport']}")
156+
157+
# VULNERABILITY Extraction
158+
if "vulns" in data:
159+
for cve, vuln_data in data["vulns"].items():
160+
cvss = vuln_data.get("cvss", 0)
161+
severity = max(
162+
(level for level, threshold in severity_map.items() if cvss >= threshold),
163+
key=lambda x: severity_map[x],
164+
)
165+
vuln = {
166+
"host": data.get("ip_str"),
167+
"severity": severity,
168+
"description": cve,
169+
}
170+
await self.emit_event(
171+
vuln,
172+
"VULNERABILITY",
173+
parent=event,
174+
tags=[],
175+
context=f"Shodan API {ip} request and find VULNERABILITY {cve}",
176+
)

0 commit comments

Comments
 (0)