add ASN numbers as scan targets with async expansion via new ASN helper

Author: liquidsec

bbot/cli.py

@@ -7,7 +7,7 @@
 from bbot.errors import *
 from bbot import __version__
 from bbot.logger import log_to_stderr
-from bbot.core.helpers.misc import chain_lists, rm_rf
+from bbot.core.helpers.misc import chain_lists
 
 
 if multiprocessing.current_process().name == "MainProcess":
@@ -56,6 +56,10 @@ async def _main():
             return
         # ensure arguments (-c config options etc.) are valid
         options = preset.args.parsed
+        # apply CLI log level options (e.g. --debug/--verbose/--silent) to the
+        # global core logger even for CLI-only commands (like --install-all-deps)
+        # that don't construct a full Scanner.
+        preset.apply_log_level(apply_core=True)
 
         # print help if no arguments
         if len(sys.argv) == 1:
@@ -90,7 +94,8 @@ async def _main():
                 preset._default_output_modules = options.output_modules
                 preset._default_internal_modules = []
 
-            preset.bake()
+            # Bake a temporary copy of the preset so that flags correctly enable their associated modules before listing them
+            preset = preset.bake()
 
             # --list-modules
             if options.list_modules:
@@ -144,59 +149,67 @@ async def _main():
                 print(row)
             return
 
-        try:
-            scan = Scanner(preset=preset)
-        except (PresetAbortError, ValidationError) as e:
-            log.warning(str(e))
+        baked_preset = preset.bake()
+
+        # --current-preset / --current-preset-full
+        if options.current_preset or options.current_preset_full:
+            # Ensure we always have a human-friendly description. Prefer an
+            # explicit scan_name if present, otherwise fall back to the
+            # preset name (e.g. "bbot_cli_main").
+            if not baked_preset.description:
+                if baked_preset.scan_name:
+                    baked_preset.description = str(baked_preset.scan_name)
+                elif baked_preset.name:
+                    baked_preset.description = str(baked_preset.name)
+            if options.current_preset_full:
+                print(baked_preset.to_yaml(full_config=True))
+            else:
+                print(baked_preset.to_yaml())
+            sys.exit(0)
             return
 
+        # deadly modules (no scan required yet)
         deadly_modules = [
-            m for m in scan.preset.scan_modules if "deadly" in preset.preloaded_module(m).get("flags", [])
+            m for m in baked_preset.scan_modules if "deadly" in baked_preset.preloaded_module(m).get("flags", [])
         ]
         if deadly_modules and not options.allow_deadly:
             log.hugewarning(f"You enabled the following deadly modules: {','.join(deadly_modules)}")
             log.hugewarning("Deadly modules are highly intrusive")
             log.hugewarning("Please specify --allow-deadly to continue")
             return False
 
-        # --current-preset
-        if options.current_preset:
-            print(scan.preset.to_yaml())
-            sys.exit(0)
-            return
-
-        # --current-preset-full
-        if options.current_preset_full:
-            print(scan.preset.to_yaml(full_config=True))
-            sys.exit(0)
+        try:
+            scan = Scanner(preset=baked_preset)
+        except (PresetAbortError, ValidationError) as e:
+            log.warning(str(e))
             return
 
         # --install-all-deps
         if options.install_all_deps:
+            # create a throwaway Scanner solely so that Preset.bake(scan) can perform find_and_replace() on all module configs so that placeholders like "#{BBOT_TOOLS}" are resolved before running Ansible tasks.
+            from bbot.scanner import Scanner as _ScannerForDeps
+
             preloaded_modules = preset.module_loader.preloaded()
-            scan_modules = [k for k, v in preloaded_modules.items() if str(v.get("type", "")) == "scan"]
-            output_modules = [k for k, v in preloaded_modules.items() if str(v.get("type", "")) == "output"]
-            log.verbose("Creating dummy scan with all modules + output modules for deps installation")
-            dummy_scan = Scanner(preset=preset, modules=scan_modules, output_modules=output_modules)
-            dummy_scan.helpers.depsinstaller.force_deps = True
+            modules_for_deps = [
+                k for k, v in preloaded_modules.items() if str(v.get("type", "")) in ("scan", "output")
+            ]
+
+            # dummy scan used only for environment preparation
+            dummy_scan = _ScannerForDeps(preset=preset)
+
+            helper = dummy_scan.helpers
             log.info("Installing module dependencies")
-            await dummy_scan.load_modules()
-            log.verbose("Running module setups")
-            succeeded, hard_failed, soft_failed = await dummy_scan.setup_modules(deps_only=True)
-            # remove any leftovers from the dummy scan
-            rm_rf(dummy_scan.home, ignore_errors=True)
-            rm_rf(dummy_scan.temp_dir, ignore_errors=True)
+            succeeded, failed = await helper.depsinstaller.install(*modules_for_deps)
             if succeeded:
                 log.success(
                     f"Successfully installed dependencies for {len(succeeded):,} modules: {','.join(succeeded)}"
                 )
-            if soft_failed or hard_failed:
-                failed = soft_failed + hard_failed
+            if failed:
                 log.warning(f"Failed to install dependencies for {len(failed):,} modules: {', '.join(failed)}")
                 return False
             return True
 
-        scan_name = str(scan.name)
+        await scan._prep()
 
         log.verbose("")
         log.verbose("### MODULES ENABLED ###")
@@ -205,12 +218,19 @@ async def _main():
             log.verbose(row)
 
         scan.helpers.word_cloud.load()
-        await scan._prep()
+
+        scan_name = str(scan.name)
 
         if not options.dry_run:
             log.trace(f"Command: {' '.join(sys.argv)}")
 
-            if sys.stdin.isatty():
+            # In some environments (e.g. tests) stdin may be closed or not support isatty(). Treat those cases as non-interactive.
+            try:
+                stdin_is_tty = sys.stdin.isatty()
+            except (ValueError, io.UnsupportedOperation):
+                stdin_is_tty = False
+
+            if stdin_is_tty:
                 # warn if any targets belong directly to a cloud provider
                 if not scan.preset.strict_scope:
                     for event in scan.target.seeds.event_seeds:

bbot/core/engine.py

@@ -343,6 +343,26 @@ async def shutdown(self):
                 self.context.term()
             except Exception:
                 print(traceback.format_exc(), file=sys.stderr)
+            # terminate the server process/thread
+            if self._server_process is not None:
+                try:
+                    self._server_process.join(timeout=5)
+                    if self._server_process.is_alive():
+                        # threads don't have terminate/kill, only processes do
+                        terminate = getattr(self._server_process, "terminate", None)
+                        if callable(terminate):
+                            terminate()
+                            self._server_process.join(timeout=3)
+                        if self._server_process.is_alive():
+                            kill = getattr(self._server_process, "kill", None)
+                            if callable(kill):
+                                kill()
+                except Exception:
+                    with suppress(Exception):
+                        kill = getattr(self._server_process, "kill", None)
+                        if callable(kill):
+                            kill()
+                self._server_process = None
             # delete socket file on exit
             self.socket_path.unlink(missing_ok=True)
 

bbot/core/event/base.py

@@ -605,7 +605,7 @@ def parent(self, parent):
                 self.web_spider_distance = getattr(parent, "web_spider_distance", 0)
                 event_has_url = getattr(self, "parsed_url", None) is not None
                 for t in parent.tags:
-                    if t in ("affiliate",):
+                    if t in ("affiliate"):
                         self.add_tag(t)
                     elif t.startswith("mutation-"):
                         self.add_tag(t)
@@ -1129,6 +1129,41 @@ class ASN(DictEvent):
     _always_emit = True
     _quick_emit = True
 
+    def sanitize_data(self, data):
+        if not isinstance(data, int):
+            raise ValidationError(f"ASN number must be an integer: {data}")
+        return data
+
+    def _data_human(self):
+        """Create a concise human-readable representation of ASN data."""
+        # Start with basic ASN info
+        display_data = {"asn": str(self.data)}
+
+        # Try to get additional ASN data from the helper if available
+        if hasattr(self, "scan") and self.scan and hasattr(self.scan, "helpers"):
+            try:
+                # Check if we can access the ASN helper synchronously
+                asn_helper = self.scan.helpers.asn
+                # Try to get cached data first (this should be synchronous)
+                cached_data = asn_helper._cache_lookup_asn(self.data)
+                if cached_data:
+                    display_data.update(
+                        {
+                            "name": cached_data.get("name", ""),
+                            "description": cached_data.get("description", ""),
+                            "country": cached_data.get("country", ""),
+                        }
+                    )
+                    # Replace subnets list with count for readability
+                    subnets = cached_data.get("subnets", [])
+                    if subnets and isinstance(subnets, list):
+                        display_data["subnet_count"] = len(subnets)
+            except Exception:
+                # If anything fails, just return basic ASN info
+                pass
+
+        return json.dumps(display_data, sort_keys=True)
+
 
 class CODE_REPOSITORY(DictHostEvent):
     _always_emit = True
@@ -1617,18 +1652,6 @@ def _pretty_string(self):
         return self.data["technology"]
 
 
-class VHOST(DictHostEvent):
-    class _data_validator(BaseModel):
-        host: str
-        vhost: str
-        url: Optional[str] = None
-        _validate_url = field_validator("url")(validators.validate_url)
-        _validate_host = field_validator("host")(validators.validate_host)
-
-    def _pretty_string(self):
-        return self.data["vhost"]
-
-
 class PROTOCOL(DictHostEvent):
     class _data_validator(BaseModel):
         host: str

bbot/core/event/helpers.py

@@ -9,6 +9,20 @@
 bbot_event_seeds = {}
 
 
+# Pre-compute sorted event classes for performance
+# This is computed once when the module is loaded instead of on every EventSeed() call
+def _get_sorted_event_classes():
+    """
+    Sort event classes by priority (higher priority first).
+    This ensures specific patterns like ASN:12345 are checked before broad patterns like hostname:port.
+    """
+    return sorted(bbot_event_seeds.items(), key=lambda x: getattr(x[1], "priority", 5), reverse=True)
+
+
+# This will be populated after all event seed classes are registered
+_sorted_event_classes = None
+
+
 """
 An "Event Seed" is a lightweight event containing only the minimum logic required to:
     - parse input to determine the event type + data
@@ -18,6 +32,19 @@
 It's useful for quickly parsing target lists without the cpu+memory overhead of creating full-fledged BBOT events
 
 Not every type of BBOT event needs to be represented here. Only ones that are meant to be targets.
+
+PRIORITY SYSTEM:
+Event seeds support a priority system to control the order in which regex patterns are checked.
+This prevents conflicts where one event type's regex might incorrectly match another type's input.
+
+Priority values:
+- Higher numbers = checked first
+- Default priority = 5
+- Range: 1-10
+
+To set priority on an event seed class:
+    class MyEventSeed(BaseEventSeed):
+        priority = 8  # Higher than default, will be checked before most others
 """
 
 
@@ -27,17 +54,25 @@ class EventSeedRegistry(type):
     """
 
     def __new__(mcs, name, bases, attrs):
-        global bbot_event_seeds
+        global bbot_event_seeds, _sorted_event_classes
         cls = super().__new__(mcs, name, bases, attrs)
         # Don't register the base EventSeed class
         if name != "BaseEventSeed":
             bbot_event_seeds[cls.__name__] = cls
+            # Recompute sorted classes whenever a new event seed is registered
+            _sorted_event_classes = _get_sorted_event_classes()
         return cls
 
 
 def EventSeed(input):
     input = smart_encode_punycode(smart_decode(input).strip())
-    for _, event_class in bbot_event_seeds.items():
+
+    # Use pre-computed sorted event classes for better performance
+    global _sorted_event_classes
+    if _sorted_event_classes is None:
+        _sorted_event_classes = _get_sorted_event_classes()
+
+    for _, event_class in _sorted_event_classes:
         if hasattr(event_class, "precheck"):
             if event_class.precheck(input):
                 return event_class(input)
@@ -53,6 +88,7 @@ def EventSeed(input):
 class BaseEventSeed(metaclass=EventSeedRegistry):
     regexes = []
     _target_type = "TARGET"
+    priority = 5  # Default priority for event seed matching (1-10, higher = checked first)
 
     __slots__ = ["data", "host", "port", "input"]
 
@@ -76,6 +112,9 @@ def _sanitize_and_extract_host(self, data):
         """
         return data, None, None
 
+    async def _generate_children(self, helpers):
+        return []
+
     def _override_input(self, input):
         return self.data
 
@@ -143,6 +182,7 @@ def _sanitize_and_extract_host(data):
 
 class OPEN_TCP_PORT(BaseEventSeed):
     regexes = regexes.event_type_regexes["OPEN_TCP_PORT"]
+    priority = 1  # Low priority: broad hostname:port pattern should be checked after specific patterns
 
     @staticmethod
     def _sanitize_and_extract_host(data):
@@ -236,3 +276,30 @@ def _override_input(self, input):
     @staticmethod
     def handle_match(match):
         return match.group(1)
+
+
+class ASN(BaseEventSeed):
+    regexes = (re.compile(r"^(?:ASN|AS):?(\d+)$", re.I),)  # adjust regex to match ASN:17178 AS17178
+    priority = 10  # High priority
+
+    def _override_input(self, input):
+        return f"ASN:{self.data}"
+
+    # ASNs are essentially just a superset of IP_RANGES.
+    # This method resolves the ASN to a list of IP_RANGES using the ASN API, and then adds the cidr string as a child event seed.
+    # These will later be automatically resolved to an IP_RANGE event seed and added to the target.
+    async def _generate_children(self, helpers):
+        asn_data = await helpers.asn.asn_to_subnets(int(self.data))
+        children = []
+        if asn_data:
+            subnets = asn_data.get("subnets")
+            if isinstance(subnets, str):
+                subnets = [subnets]
+            if subnets:
+                for cidr in subnets:
+                    children.append(cidr)
+        return children
+
+    @staticmethod
+    def handle_match(match):
+        return match.group(1)