ASN Helper / Add ASN's as an available type type

bbot/cli.py

@@ -34,8 +34,13 @@ async def _main():
     import traceback
     from contextlib import suppress
 
-    # fix tee buffering
-    sys.stdout.reconfigure(line_buffering=True)
+    # fix tee buffering (only if on real TTY)
+    if hasattr(sys.stdout, "reconfigure"):
+        try:
+            if sys.stdout.isatty():
+                sys.stdout.reconfigure(line_buffering=True)
+        except Exception:
+            pass
 
     log = logging.getLogger("bbot.cli")
 
@@ -90,7 +95,7 @@ async def _main():
                 preset._default_output_modules = options.output_modules
                 preset._default_internal_modules = []
 
-            preset.bake()
+            await preset.bake()
 
             # --list-modules
             if options.list_modules:
@@ -150,6 +155,8 @@ async def _main():
             log.warning(str(e))
             return
 
+        await scan._prep()
+
         deadly_modules = [
             m for m in scan.preset.scan_modules if "deadly" in preset.preloaded_module(m).get("flags", [])
         ]
@@ -191,12 +198,18 @@ async def _main():
             log.verbose(row)
 
         scan.helpers.word_cloud.load()
-        await scan._prep()
 
         if not options.dry_run:
             log.trace(f"Command: {' '.join(sys.argv)}")
 
-            if sys.stdin.isatty():
+            try:
+                is_tty = (
+                    hasattr(sys.stdin, "isatty") and not getattr(sys.stdin, "closed", False) and sys.stdin.isatty()
+                )
+            except Exception:
+                is_tty = False
+
+            if 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/event/base.py

@@ -623,7 +623,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)
@@ -1135,6 +1135,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
@@ -1604,18 +1639,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)