radixtarget overhaul

bbot/core/helpers/misc.py

@@ -591,12 +591,13 @@ def is_dns_name(d, include_local=True):
     return False
 
 
-def is_ip(d, version=None):
+def is_ip(d, version=None, include_network=False):
     """
     Checks if the given string or object represents a valid IP address.
 
     Args:
         d (str or ipaddress.IPvXAddress): The IP address to check.
+        include_network (bool, optional): Whether to include network types (IPv4Network or IPv6Network). Defaults to False.
         version (int, optional): The IP version to validate (4 or 6). Default is None.
 
     Returns:
@@ -612,12 +613,17 @@ def is_ip(d, version=None):
         >>> is_ip('evilcorp.com')
         False
     """
+    ip = None
     try:
         ip = ipaddress.ip_address(d)
-        if version is None or ip.version == version:
-            return True
     except Exception:
-        pass
+        if include_network:
+            try:
+                ip = ipaddress.ip_network(d, strict=False)
+            except Exception:
+                pass
+    if ip is not None and (version is None or ip.version == version):
+        return True
     return False
 
 

bbot/core/helpers/regexes.py

@@ -40,7 +40,7 @@
 
 # dns names with periods
 _dns_name_regex = r"(?:\w(?:[\w-]{0,100}\w)?\.)+(?:[xX][nN]--)?[^\W_]{1,63}\.?"
-dns_name_regex = re.compile(_dns_name_regex, re.I)
+dns_name_regex = re.compile(r"^" + _dns_name_regex + r"$", re.I)
 
 # dns names without periods
 _hostname_regex = r"(?!\w*\.\w+)\w(?:[\w-]{0,100}\w)?"

bbot/scanner/target.py

@@ -1,4 +1,3 @@
-import copy
 import logging
 import regex as re
 from hashlib import sha1
@@ -7,6 +6,8 @@
 
 from bbot.errors import *
 from bbot.core.event import make_event, is_event
+from bbot.core.helpers.misc import is_dns_name, is_ip
+
 
 log = logging.getLogger("bbot.core.target")
 
@@ -59,8 +60,20 @@ def __init__(self, *targets, scan=None, **kwargs):
                     self.special_target_types[func._regex] = func
 
     def get(self, event, single=True, **kwargs):
-        event = self.make_event(event)
-        results = super().get(event.host, **kwargs)
+        """
+        Override default .get() to accept events and optionally return multiple results
+        """
+        if is_event(event):
+            host = event.host
+        # save resources by checking if the event is an IP or DNS name
+        elif is_ip(event, include_network=True) or is_dns_name(event):
+            host = event
+        elif isinstance(event, str):
+            event = self.make_event(event)
+            host = event.host
+        else:
+            raise ValueError(f"Invalid host/event: {event} ({type(event)})")
+        results = super().get(host, **kwargs)
         if results and single:
             return next(iter(results))
         return results
@@ -146,18 +159,27 @@ def handle_username(self, match):
             return [username_event]
         return []
 
+    def _hash_value(self):
+        # seeds get hashed by event data
+        return sorted(str(e.data).encode() for e in self.events)
 
-class ScanWhitelist(BaseTarget):
-    """
-    A collection of BBOT events that represent a scan's whitelist.
-    """
 
+class ACLTarget(BaseTarget):
     def __init__(self, *args, **kwargs):
+        # ACL mode dedupes by host (and skips adding already-contained hosts) for efficiency
         kwargs["acl_mode"] = True
         super().__init__(*args, **kwargs)
 
 
-class ScanBlacklist(BaseTarget):
+class ScanWhitelist(ACLTarget):
+    """
+    A collection of BBOT events that represent a scan's whitelist.
+    """
+
+    pass
+
+
+class ScanBlacklist(ACLTarget):
     """
     A collection of BBOT events that represent a scan's blacklist.
     """
@@ -189,6 +211,12 @@ def get(self, event, **kwargs):
                 return event
         return None
 
+    def _hash_value(self):
+        # regexes are included in blacklist hash
+        regex_patterns = [str(r.pattern).encode() for r in self.blacklist_regexes]
+        hosts = [str(h).encode() for h in self.sorted_hosts]
+        return hosts + regex_patterns
+
 
 class BBOTTarget:
     """
@@ -240,13 +268,6 @@ def scope_hash(self):
             sha1_hash.update(target_hash)
         return sha1_hash.digest()
 
-    def copy(self):
-        self_copy = copy.copy(self)
-        self_copy.seeds = self.seeds.copy()
-        self_copy.whitelist = self.whitelist.copy()
-        self_copy.blacklist = self.blacklist.copy()
-        return self_copy
-
     def in_scope(self, host):
         """
         Check whether a hostname, url, IP, etc. is in scope.
@@ -311,3 +332,6 @@ def minimal(self):
             blacklist=self.blacklist.inputs,
             strict_scope=self.strict_scope,
         )
+
+    def __eq__(self, other):
+        return self.hash == other.hash

bbot/test/test_step_1/test_helpers.py

@@ -93,8 +93,23 @@ async def test_helpers_misc(helpers, scan, bbot_scanner, bbot_httpserver):
         ipaddress.ip_network("0.0.0.0/0"),
     ]
     assert helpers.is_ip("127.0.0.1")
+    assert helpers.is_ip("127.0.0.1", include_network=True)
+    assert helpers.is_ip("127.0.0.1", version=4)
+    assert not helpers.is_ip("127.0.0.1", version=6)
     assert not helpers.is_ip("127.0.0.0.1")
 
+    assert helpers.is_ip("dead::beef")
+    assert helpers.is_ip("dead::beef", include_network=True)
+    assert not helpers.is_ip("dead::beef", version=4)
+    assert helpers.is_ip("dead::beef", version=6)
+    assert not helpers.is_ip("dead:::beef")
+
+    assert not helpers.is_ip("1.2.3.4/24")
+    assert helpers.is_ip("1.2.3.4/24", include_network=True)
+    assert not helpers.is_ip("1.2.3.4/24", version=4)
+    assert helpers.is_ip("1.2.3.4/24", include_network=True, version=4)
+    assert not helpers.is_ip("1.2.3.4/24", include_network=True, version=6)
+
     assert not helpers.is_ip_type("127.0.0.1")
     assert helpers.is_ip_type(ipaddress.ip_address("127.0.0.1"))
     assert not helpers.is_ip_type(ipaddress.ip_address("127.0.0.1"), network=True)
@@ -104,6 +119,8 @@ async def test_helpers_misc(helpers, scan, bbot_scanner, bbot_httpserver):
     assert not helpers.is_ip_type(ipaddress.ip_network("127.0.0.0/8"), network=False)
 
     assert helpers.is_dns_name("evilcorp.com")
+    assert not helpers.is_dns_name("evilcorp.com:80")
+    assert not helpers.is_dns_name("http://evilcorp.com:80")
     assert helpers.is_dns_name("evilcorp")
     assert not helpers.is_dns_name("evilcorp", include_local=False)
     assert helpers.is_dns_name("ドメイン.テスト")