Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
16 changes: 14 additions & 2 deletions bbot/modules/fingerprintx.py
Original file line number Diff line number Diff line change
Expand Up @@ -5,7 +5,7 @@

class fingerprintx(BaseModule):
watched_events = ["OPEN_TCP_PORT"]
produced_events = ["PROTOCOL"]
produced_events = ["PROTOCOL", "URL_UNVERIFIED"]
flags = ["safe", "active", "service-enum", "slow"]
meta = {
"description": "Fingerprint exposed services like RDP, SSH, MySQL, etc.",
Expand Down Expand Up @@ -78,7 +78,7 @@ async def handle_batch(self, *events):
if not host and port and protocol:
continue
banner = j.get("metadata", {}).get("banner", "").strip()
port_data = f"{host}:{port}"
port_data = self.helpers.make_netloc(host, port)
tags = set()
parent_event = _input.get(port_data)
protocol_data = {"host": host, "protocol": protocol}
Expand All @@ -93,3 +93,15 @@ async def handle_batch(self, *events):
tags=tags,
context=f"{{module}} probed {port_data} and detected {{event.type}}: {protocol}",
)
if protocol in ("HTTP", "HTTPS"):
port_int = int(port) if port else None
is_default_port = (protocol == "HTTP" and port_int == 80) or (protocol == "HTTPS" and port_int == 443)
netloc = self.helpers.make_netloc(host, None if is_default_port else port_int)
url = f"{protocol.lower()}://{netloc}"
await self.emit_event(
url,
"URL_UNVERIFIED",
parent=parent_event,
tags=tags,
context=f"{{module}} probed {port_data} and detected a {protocol} web service",
)
54 changes: 54 additions & 0 deletions bbot/test/test_step_2/module_tests/test_module_fingerprintx.py
Original file line number Diff line number Diff line change
@@ -1,3 +1,5 @@
import json

from .base import ModuleTestBase


Expand All @@ -12,3 +14,55 @@ def check(self, module_test, events):
and event.data["protocol"] == "HTTP"
for event in events
), "HTTP protocol not detected"


class TestFingerprintxURLs(ModuleTestBase):
"""Mocks fingerprintx output to verify URL_UNVERIFIED construction across IPv4/IPv6 and default/non-default ports."""

module_name = "fingerprintx"
targets = [
"127.0.0.1:80",
"127.0.0.1:8443",
"[::1]:443",
"[::1]:8080",
]
config_overrides = {"modules": {"fingerprintx": {"skip_common_web": False}}}

# (host, port, protocol) -> what fingerprintx pretends to find
fake_results = [
("127.0.0.1", 80, "HTTP"),
("127.0.0.1", 8443, "HTTPS"),
("::1", 443, "HTTPS"),
("::1", 8080, "HTTP"),
]

async def setup_after_prep(self, module_test):
results = self.fake_results

async def fake_run_process_live(self, command, **kwargs):
for host, port, protocol in results:
yield json.dumps({"ip": host, "host": host, "port": port, "protocol": protocol})

module_test.monkeypatch.setattr(module_test.module.__class__, "run_process_live", fake_run_process_live)

def check(self, module_test, events):
urls = {e.data["url"] for e in events if e.type == "URL_UNVERIFIED"}
expected = {
"http://127.0.0.1/",
"https://127.0.0.1:8443/",
"https://[::1]/",
"http://[::1]:8080/",
}
assert expected.issubset(urls), f"missing URLs; got {urls}"

protocol_events = [e for e in events if e.type == "PROTOCOL"]
assert any(
e.host == module_test.scan.helpers.make_ip_type("::1") and e.port == 443 and e.data["protocol"] == "HTTPS"
for e in protocol_events
), "IPv6 PROTOCOL event missing — parent lookup likely broken"
assert any(
e.host == module_test.scan.helpers.make_ip_type("127.0.0.1")
and e.port == 8443
and e.data["protocol"] == "HTTPS"
for e in protocol_events
), "IPv4 PROTOCOL event missing"
Loading