Merge pull request #1971 from blacklanternsecurity/nmap-output-module

New Module: Nmap XML Output

bbot-docker.sh

@@ -1,2 +1,3 @@
-# run the docker image
-docker run --rm -it -v "$HOME/.bbot:/root/.bbot" -v "$HOME/.config/bbot:/root/.config/bbot" blacklanternsecurity/bbot:stable "$@"
+# OUTPUTS SCAN DATA TO ~/.bbot/scans
+
+docker run --rm -it -v "$HOME/.bbot/scans:/root/.bbot/scans" -v "$HOME/.config/bbot:/root/.config/bbot" blacklanternsecurity/bbot:stable "$@"

bbot/cli.py

@@ -78,7 +78,7 @@ async def _main():
             return
 
         # if we're listing modules or their options
-        if options.list_modules or options.list_module_options:
+        if options.list_modules or options.list_output_modules or options.list_module_options:
             # if no modules or flags are specified, enable everything
             if not (options.modules or options.output_modules or options.flags):
                 for module, preloaded in preset.module_loader.preloaded().items():
@@ -96,7 +96,17 @@ async def _main():
                 print("")
                 print("### MODULES ###")
                 print("")
-                for row in preset.module_loader.modules_table(preset.modules).splitlines():
+                modules = sorted(set(preset.scan_modules + preset.internal_modules))
+                for row in preset.module_loader.modules_table(modules).splitlines():
+                    print(row)
+                return
+
+            # --list-output-modules
+            if options.list_output_modules:
+                print("")
+                print("### OUTPUT MODULES ###")
+                print("")
+                for row in preset.module_loader.modules_table(preset.output_modules).splitlines():
                     print(row)
                 return
 

bbot/core/helpers/names_generator.py

@@ -293,6 +293,7 @@
     "alyssa",
     "amanda",
     "amber",
+    "amir",
     "amy",
     "andrea",
     "andrew",

bbot/modules/base.py

@@ -51,7 +51,7 @@ class BaseModule:
 
         target_only (bool): Accept only the initial target event(s). Default is False.
 
-        in_scope_only (bool): Accept only explicitly in-scope events. Default is False.
+        in_scope_only (bool): Accept only explicitly in-scope events, regardless of the scan's search distance. Default is False.
 
         options (Dict): Customizable options for the module, e.g., {"api_key": ""}. Empty dict by default.
 

bbot/modules/internal/cloudcheck.py

@@ -5,7 +5,11 @@
 
 class CloudCheck(BaseInterceptModule):
     watched_events = ["*"]
-    meta = {"description": "Tag events by cloud provider, identify cloud resources like storage buckets"}
+    meta = {
+        "description": "Tag events by cloud provider, identify cloud resources like storage buckets",
+        "created_date": "2024-07-07",
+        "author": "@TheTechromancer",
+    }
     scope_distance_modifier = 1
     _priority = 3
 

bbot/modules/internal/dnsresolve.py

@@ -9,6 +9,8 @@
 
 class DNSResolve(BaseInterceptModule):
     watched_events = ["*"]
+    produced_events = ["DNS_NAME", "IP_ADDRESS", "RAW_DNS_RECORD"]
+    meta = {"description": "Perform DNS resolution", "created_date": "2022-04-08", "author": "@TheTechromancer"}
     _priority = 1
     scope_distance_modifier = None
 

bbot/modules/internal/speculate.py

@@ -104,7 +104,7 @@ async def handle_event(self, event):
             # don't act on unresolved DNS_NAMEs
             usable_dns = False
             if event.type == "DNS_NAME":
-                if self.dns_disable or ("a-record" in event.tags or "aaaa-record" in event.tags):
+                if self.dns_disable or event.resolved_hosts:
                     usable_dns = True
 
             if event.type == "IP_ADDRESS" or usable_dns:

bbot/modules/output/mysql.py

@@ -3,7 +3,7 @@
 
 class MySQL(SQLTemplate):
     watched_events = ["*"]
-    meta = {"description": "Output scan data to a MySQL database"}
+    meta = {"description": "Output scan data to a MySQL database", "created_date": "2024-11-13", "author": "@TheTechromancer"}
     options = {
         "username": "root",
         "password": "bbotislife",

bbot/modules/output/nmap_xml.py

@@ -0,0 +1,171 @@
+import sys
+from xml.dom import minidom
+from datetime import datetime
+from xml.etree.ElementTree import Element, SubElement, tostring
+
+from bbot import __version__
+from bbot.modules.output.base import BaseOutputModule
+
+
+class NmapHost:
+    __slots__ = ["hostnames", "open_ports"]
+
+    def __init__(self):
+        self.hostnames = set()
+        # a dict of {port: {protocol: banner}}
+        self.open_ports = dict()
+
+
+class Nmap_XML(BaseOutputModule):
+    watched_events = ["OPEN_TCP_PORT", "DNS_NAME", "IP_ADDRESS", "PROTOCOL", "HTTP_RESPONSE"]
+    meta = {"description": "Output to Nmap XML", "created_date": "2024-11-16", "author": "@TheTechromancer"}
+    output_filename = "output.nmap.xml"
+    in_scope_only = True
+
+    async def setup(self):
+        self.hosts = {}
+        self._prep_output_dir(self.output_filename)
+        return True
+
+    async def handle_event(self, event):
+        event_host = event.host
+
+        # we always record by IP
+        ips = []
+        for ip in event.resolved_hosts:
+            try:
+                ips.append(self.helpers.make_ip_type(ip))
+            except ValueError:
+                continue
+        if not ips and self.helpers.is_ip(event_host):
+            ips = [event_host]
+
+        for ip in ips:
+            try:
+                nmap_host = self.hosts[ip]
+            except KeyError:
+                nmap_host = NmapHost()
+                self.hosts[ip] = nmap_host
+
+            event_port = getattr(event, "port", None)
+            if event.type == "OPEN_TCP_PORT":
+                if event_port not in nmap_host.open_ports:
+                    nmap_host.open_ports[event.port] = {}
+            elif event.type in ("PROTOCOL", "HTTP_RESPONSE"):
+                if event_port is not None:
+                    try:
+                        existing_services = nmap_host.open_ports[event.port]
+                    except KeyError:
+                        existing_services = {}
+                        nmap_host.open_ports[event.port] = existing_services
+                    if event.type == "PROTOCOL":
+                        protocol = event.data["protocol"].lower()
+                        banner = event.data.get("banner", None)
+                    elif event.type == "HTTP_RESPONSE":
+                        protocol = event.parsed_url.scheme.lower()
+                        banner = event.http_title
+                    if protocol not in existing_services:
+                        existing_services[protocol] = banner
+
+            if self.helpers.is_ip(event_host):
+                if str(event.module) == "PTR":
+                    nmap_host.hostnames.add(event.parent.data)
+            else:
+                nmap_host.hostnames.add(event_host)
+
+    async def report(self):
+        scan_start_time = str(int(self.scan.start_time.timestamp()))
+        scan_start_time_str = self.scan.start_time.strftime("%a %b %d %H:%M:%S %Y")
+        scan_end_time = datetime.now()
+        scan_end_time_str = scan_end_time.strftime("%a %b %d %H:%M:%S %Y")
+        scan_end_time_timestamp = str(scan_end_time.timestamp())
+        scan_duration = scan_end_time - self.scan.start_time
+        num_hosts_up = len(self.hosts)
+
+        # Create the root element
+        nmaprun = Element(
+            "nmaprun",
+            {
+                "scanner": "bbot",
+                "args": " ".join(sys.argv),
+                "start": scan_start_time,
+                "startstr": scan_start_time_str,
+                "version": str(__version__),
+                "xmloutputversion": "1.05",
+            },
+        )
+
+        ports_scanned = []
+        speculate_module = self.scan.modules.get("speculate", None)
+        if speculate_module is not None:
+            ports_scanned = speculate_module.ports
+        portscan_module = self.scan.modules.get("portscan", None)
+        if portscan_module is not None:
+            ports_scanned = self.helpers.parse_port_string(str(portscan_module.ports))
+        num_ports_scanned = len(sorted(ports_scanned))
+        ports_scanned = ",".join(str(x) for x in sorted(ports_scanned))
+
+        # Add scaninfo
+        SubElement(
+            nmaprun,
+            "scaninfo",
+            {"type": "syn", "protocol": "tcp", "numservices": str(num_ports_scanned), "services": ports_scanned},
+        )
+
+        # Add host information
+        for ip, nmap_host in self.hosts.items():
+            hostnames = sorted(nmap_host.hostnames)
+            ports = sorted(nmap_host.open_ports)
+
+            host_elem = SubElement(nmaprun, "host")
+            SubElement(host_elem, "status", {"state": "up", "reason": "user-set", "reason_ttl": "0"})
+            SubElement(host_elem, "address", {"addr": str(ip), "addrtype": f"ipv{ip.version}"})
+
+            if hostnames:
+                hostnames_elem = SubElement(host_elem, "hostnames")
+                for hostname in hostnames:
+                    SubElement(hostnames_elem, "hostname", {"name": hostname, "type": "user"})
+
+            ports = SubElement(host_elem, "ports")
+            for port, protocols in nmap_host.open_ports.items():
+                port_elem = SubElement(ports, "port", {"protocol": "tcp", "portid": str(port)})
+                SubElement(port_elem, "state", {"state": "open", "reason": "syn-ack", "reason_ttl": "0"})
+                # <port protocol="tcp" portid="443"><state state="open" reason="syn-ack" reason_ttl="53"/><service name="http" product="AkamaiGHost" extrainfo="Akamai&apos;s HTTP Acceleration/Mirror service" tunnel="ssl" method="probed" conf="10"/></port>
+                for protocol, banner in protocols.items():
+                    attrs = {"name": protocol, "method": "probed", "conf": "10"}
+                    if banner is not None:
+                        attrs["product"] = banner
+                        attrs["extrainfo"] = banner
+                    SubElement(port_elem, "service", attrs)
+
+        # Add runstats
+        runstats = SubElement(nmaprun, "runstats")
+        SubElement(
+            runstats,
+            "finished",
+            {
+                "time": scan_end_time_timestamp,
+                "timestr": scan_end_time_str,
+                "summary": f"BBOT done at {scan_end_time_str}; {num_hosts_up} scanned in {scan_duration} seconds",
+                "elapsed": str(scan_duration.total_seconds()),
+                "exit": "success",
+            },
+        )
+        SubElement(runstats, "hosts", {"up": str(num_hosts_up), "down": "0", "total": str(num_hosts_up)})
+
+        # make backup of the file
+        self.helpers.backup_file(self.output_file)
+
+        # Pretty-format the XML
+        rough_string = tostring(nmaprun, encoding="utf-8")
+        reparsed = minidom.parseString(rough_string)
+
+        # Create a new document with the doctype
+        doctype = minidom.DocumentType("nmaprun")
+        reparsed.insertBefore(doctype, reparsed.documentElement)
+
+        pretty_xml = reparsed.toprettyxml(indent="  ")
+
+        with open(self.output_file, "w") as f:
+            f.write(pretty_xml)
+        self.info(f"Saved Nmap XML output to {self.output_file}")

bbot/modules/output/postgres.py

@@ -3,7 +3,11 @@
 
 class Postgres(SQLTemplate):
     watched_events = ["*"]
-    meta = {"description": "Output scan data to a SQLite database"}
+    meta = {
+        "description": "Output scan data to a SQLite database",
+        "created_date": "2024-11-08",
+        "author": "@TheTechromancer",
+    }
     options = {
         "username": "postgres",
         "password": "bbotislife",

bbot/modules/output/sqlite.py

@@ -5,7 +5,11 @@
 
 class SQLite(SQLTemplate):
     watched_events = ["*"]
-    meta = {"description": "Output scan data to a SQLite database"}
+    meta = {
+        "description": "Output scan data to a SQLite database",
+        "created_date": "2024-11-07",
+        "author": "@TheTechromancer",
+    }
     options = {
         "database": "",
     }

bbot/modules/output/stdout.py

@@ -6,7 +6,7 @@
 
 class Stdout(BaseOutputModule):
     watched_events = ["*"]
-    meta = {"description": "Output to text"}
+    meta = {"description": "Output to text", "created_date": "2024-04-03", "author": "@TheTechromancer"}
     options = {"format": "text", "event_types": [], "event_fields": [], "in_scope_only": False, "accept_dupes": True}
     options_desc = {
         "format": "Which text format to display, choices: text,json",

bbot/modules/output/txt.py

@@ -5,7 +5,7 @@
 
 class TXT(BaseOutputModule):
     watched_events = ["*"]
-    meta = {"description": "Output to text"}
+    meta = {"description": "Output to text", "created_date": "2024-04-03", "author": "@TheTechromancer"}
     options = {"output_file": ""}
     options_desc = {"output_file": "Output to file"}
 

bbot/scanner/preset/args.py

@@ -52,6 +52,11 @@ class BBOTArgs:
             "",
             "bbot -l",
         ),
+        (
+            "List output modules",
+            "",
+            "bbot -lo",
+        ),
         (
             "List presets",
             "",
@@ -290,12 +295,6 @@ def create_parser(self, *args, **kwargs):
         )
 
         output = p.add_argument_group(title="Output")
-        output.add_argument(
-            "-o",
-            "--output-dir",
-            help="Directory to output scan results",
-            metavar="DIR",
-        )
         output.add_argument(
             "-om",
             "--output-modules",
@@ -304,6 +303,13 @@ def create_parser(self, *args, **kwargs):
             help=f'Output module(s). Choices: {",".join(sorted(self.preset.module_loader.output_module_choices))}',
             metavar="MODULE",
         )
+        output.add_argument("-lo", "--list-output-modules", action="store_true", help="List available output modules")
+        output.add_argument(
+            "-o",
+            "--output-dir",
+            help="Directory to output scan results",
+            metavar="DIR",
+        )
         output.add_argument("--json", "-j", action="store_true", help="Output scan data in JSON format")
         output.add_argument("--brief", "-br", action="store_true", help="Output only the data itself")
         output.add_argument("--event-types", nargs="+", default=[], help="Choose which event types to display")

bbot/test/test_step_1/test_cli.py

@@ -150,11 +150,23 @@ async def test_cli_args(monkeypatch, caplog, capsys, clean_default_config):
     out, err = capsys.readouterr()
     # internal modules
     assert "| excavate " in out
-    # output modules
-    assert "| csv " in out
+    # no output modules
+    assert not "| csv " in out
     # scan modules
     assert "| wayback " in out
 
+    # list output modules
+    monkeypatch.setattr("sys.argv", ["bbot", "--list-output-modules"])
+    result = await cli._main()
+    assert result == None
+    out, err = capsys.readouterr()
+    # no internal modules
+    assert not "| excavate " in out
+    # output modules
+    assert "| csv " in out
+    # no scan modules
+    assert not "| wayback " in out
+
     # output dir and scan name
     output_dir = bbot_test_dir / "bbot_cli_args_output"
     scan_name = "bbot_cli_args_scan_name"

bbot/test/test_step_1/test_modules_basic.py

@@ -156,17 +156,15 @@ async def test_modules_basic_checks(events, httpx_mock):
             assert not (
                 "web-basic" in flags and "web-thorough" in flags
             ), f'module "{module_name}" should have either "web-basic" or "web-thorough" flags, not both'
-            meta = preloaded.get("meta", {})
-            # make sure every module has a description
-            assert meta.get("description", ""), f"{module_name} must have a description"
-            # make sure every module has an author
-            assert meta.get("author", ""), f"{module_name} must have an author"
-            # make sure every module has a created date
-            created_date = meta.get("created_date", "")
-            assert created_date, f"{module_name} must have a created date"
-            assert created_date_regex.match(
-                created_date
-            ), f"{module_name}'s created_date must match the format YYYY-MM-DD"
+        meta = preloaded.get("meta", {})
+        # make sure every module has a description
+        assert meta.get("description", ""), f"{module_name} must have a description"
+        # make sure every module has an author
+        assert meta.get("author", ""), f"{module_name} must have an author"
+        # make sure every module has a created date
+        created_date = meta.get("created_date", "")
+        assert created_date, f"{module_name} must have a created date"
+        assert created_date_regex.match(created_date), f"{module_name}'s created_date must match the format YYYY-MM-DD"
 
         # attribute checks
         watched_events = preloaded.get("watched_events")