Replace massdns with blastdns 2.0

.github/workflows/tests.yml

@@ -17,7 +17,7 @@ jobs:
       # if one python version fails, let the others finish
       fail-fast: false
       matrix:
-        python-version: ["3.10", "3.11", "3.12", "3.13", "3.14"]
+        python-version: ["3.11", "3.12", "3.13", "3.14", "3.15"]
     steps:
       - uses: actions/checkout@v7
       - name: Set up Python

bbot/core/config/models.py

@@ -311,9 +311,15 @@ class DnsConfig(BaseModel):
 
     disable: Optional[bool] = None
     minimal: Optional[bool] = None
+    nameservers: Optional[list[str]] = None
     threads: Optional[int] = None
     cache_size: Optional[int] = None
     brute_threads: Optional[int] = None
+    brute_inflight_per_resolver: Optional[int] = None
+    brute_rate_limit: Optional[float] = None
+    brute_retries: Optional[int] = None
+    brute_timeout: Optional[float] = None
+    brute_persistent_socket: Optional[bool] = None
     brute_nameservers: Optional[str] = None
     search_distance: Optional[int] = None
     runaway_limit: Optional[int] = None

bbot/core/event/base.py

@@ -2024,8 +2024,11 @@ def _data_human(self):
         parts.append(f"Confidence: [{self.data['confidence']}]")
         parts.append(self.data["description"])
         url = self.data.get("url", "")
+        host = self.data.get("host", "")
         if url and url not in self.data["description"]:
             parts.append(f"({url})")
+        elif host and not url and host not in self.data["description"]:
+            parts.append(f"Host: [{host}]")
         cves = self.data.get("cves", [])
         if cves:
             parts.append(f"[{', '.join(cves)}]")

bbot/core/helpers/async_helpers.py

@@ -4,6 +4,7 @@
 import asyncio
 import logging
 import functools
+import itertools
 from contextlib import suppress
 from cachetools import keys, LRUCache
 from contextlib import asynccontextmanager
@@ -20,6 +21,25 @@ def _get(self):
         return self._queue.popleft()
 
 
+class LaneQueue(asyncio.PriorityQueue):
+    """
+    A queue divided into lanes, where items in a lower lane are always dequeued before items in a higher one.
+
+    `lane_fn` assigns each item its lane (an int). Items within the same lane are dequeued in FIFO order.
+    """
+
+    def __init__(self, lane_fn, maxsize=0):
+        self._lane_fn = lane_fn
+        self._sequence = itertools.count()
+        super().__init__(maxsize)
+
+    def _put(self, item):
+        super()._put((self._lane_fn(item), next(self._sequence), item))
+
+    def _get(self):
+        return super()._get()[-1]
+
+
 class _Lock(asyncio.Lock):
     def __init__(self, name):
         self.name = name

bbot/core/helpers/diff.py

@@ -3,7 +3,6 @@
 import xmltodict
 from time import monotonic
 from itertools import islice
-from deepdiff import DeepDiff
 from contextlib import suppress
 from xml.parsers.expat import ExpatError
 from bbot.errors import HttpCompareError
@@ -153,7 +152,7 @@ class _BaselineSnapshot:
 
     The blasthttp Headers object is kept by reference (it survives
     Response GC independently) to preserve case-insensitive lookups and
-    duplicate-header semantics that DeepDiff relies on.
+    duplicate-header semantics that compare_headers() relies on.
     """
 
     __slots__ = ("status_code", "headers", "_text", "_spill_key", "_spill_store")
@@ -358,15 +357,11 @@ def compare_headers(self, headers_1, headers_2):
                     with suppress(KeyError):
                         del headers[header]
 
-        ddiff = DeepDiff(headers_1, headers_2, ignore_order=True, view="tree", threshold_to_diff_deeper=0)
-
-        for k in ddiff.keys():
-            for x in list(ddiff[k]):
-                try:
-                    header_value = str(x).split("'")[1]
-                except (KeyError, IndexError):
-                    continue
-                differing_headers.append(header_value)
+        # Headers iterate as lower-cased unique names; indexing returns the last value of a repeated header
+        names_1, names_2 = set(headers_1), set(headers_2)
+        for name in sorted(names_1 | names_2):
+            if name not in names_1 or name not in names_2 or headers_1[name] != headers_2[name]:
+                differing_headers.append(name)
         return differing_headers
 
     @staticmethod

bbot/core/helpers/dns/brute.py

@@ -1,18 +1,22 @@
-import json
 import random
 import asyncio
 import logging
-import subprocess
+
+from blastdns import Client, ClientConfig, DNSError
 
 
 class DNSBrute:
     """
     Helper for DNS brute-forcing.
 
+    Uses its own blastdns client, separate from the one backing normal
+    resolution: brute-forcing runs against a large public resolver list rather
+    than the system resolvers, and needs its own concurrency and rate limits.
+
     Examples:
     >>> domain = "evilcorp.com"
     >>> subdomains = ["www", "mail"]
-    >>> results = await self.helpers.dns.brute(self, domain, subdomains)
+    >>> results = await self.helpers.dns.brute(domain, subdomains)
     """
 
     _nameservers_url = (
@@ -24,11 +28,18 @@ def __init__(self, parent_helper):
         self.log = logging.getLogger("bbot.helper.dns.brute")
         self.dns_config = self.parent_helper.config.get("dns", {})
         self.num_canaries = 100
-        self.max_resolvers = self.dns_config.get("brute_threads", 1000)
+        self.concurrency = self.dns_config.get("brute_threads", 1000)
+        self.inflight_per_resolver = self.dns_config.get("brute_inflight_per_resolver", 2)
+        self.rate_limit = self.dns_config.get("brute_rate_limit", 0)
+        self.retries = self.dns_config.get("brute_retries", 20)
+        self.timeout = self.dns_config.get("brute_timeout", 0.5)
+        self.persistent_socket = self.dns_config.get("brute_persistent_socket", True)
         self.nameservers_url = self.dns_config.get("brute_nameservers", self._nameservers_url)
         self.devops_mutations = list(self.parent_helper.word_cloud.devops_mutations)
         self.digit_regex = self.parent_helper.re.compile(r"\d+")
         self._resolver_file = None
+        self._client = None
+        self._client_lock = None
         self._dnsbrute_lock = None
 
     async def __call__(self, *args, **kwargs):
@@ -40,7 +51,43 @@ def dnsbrute_lock(self):
             self._dnsbrute_lock = asyncio.Lock()
         return self._dnsbrute_lock
 
-    async def dnsbrute(self, module, domain, subdomains, type=None):
+    @property
+    def client_lock(self):
+        if self._client_lock is None:
+            self._client_lock = asyncio.Lock()
+        return self._client_lock
+
+    async def client(self):
+        """The blastdns client used for brute-forcing, built on first use.
+
+        Construction is deferred because the resolver list has to be downloaded
+        first, and blastdns takes its resolvers at construction time.
+        """
+        async with self.client_lock:
+            if self._client is None:
+                resolver_file = await self.resolver_file()
+                resolvers = list(self.parent_helper.read_file(resolver_file))
+                self.log.verbose(f"Starting brute-force DNS client with {len(resolvers):,} resolvers")
+                self._client = Client(
+                    resolvers,
+                    ClientConfig(
+                        max_concurrency=self.concurrency,
+                        max_inflight_per_resolver=self.inflight_per_resolver,
+                        rate_limit=self.rate_limit or None,
+                        # A public resolver list carries a lot of dead entries;
+                        # drop them up front instead of paying for them all scan.
+                        resolver_probe=True,
+                        # Brute-force queries are unique by construction, so a
+                        # cache would only consume memory.
+                        cache_capacity=0,
+                        request_timeout_ms=int(self.timeout * 1000),
+                        max_retries=self.retries,
+                        persistent_socket=self.persistent_socket,
+                    ),
+                )
+            return self._client
+
+    async def dnsbrute(self, domain, subdomains, type=None):
         subdomains = list(subdomains)
 
         if type is None:
@@ -53,7 +100,7 @@ async def dnsbrute(self, module, domain, subdomains, type=None):
             wildcard_rdtypes.update(rdtypes)
         if wildcard_domains:
             self.log.hugewarning(
-                f"Aborting massdns on {domain} because it's a wildcard domain ({','.join(sorted(wildcard_rdtypes))})"
+                f"Aborting brute-force of {domain} because it's a wildcard domain ({','.join(sorted(wildcard_rdtypes))})"
             )
             return []
 
@@ -66,7 +113,7 @@ async def dnsbrute(self, module, domain, subdomains, type=None):
 
         results = []
         canaries_triggered = []
-        async for hostname, ip, rdtype in self._massdns(module, domain, subdomains, rdtype=type):
+        async for hostname in self._resolve_subdomains(domain, subdomains, rdtype=type):
             sub = hostname.split(domain)[0].rstrip(".")
             if sub in canary_set:
                 canaries_triggered.append(sub)
@@ -75,77 +122,84 @@ async def dnsbrute(self, module, domain, subdomains, type=None):
 
         if len(canaries_triggered) > 5:
             self.log.info(
-                f"Aborting massdns on {domain} due to false positive: ({len(canaries_triggered):,} canaries triggered - {','.join(canaries_triggered)})"
+                f"Aborting brute-force of {domain} due to false positive: ({len(canaries_triggered):,} canaries triggered - {','.join(canaries_triggered)})"
             )
             return []
 
         # everything checks out
         return results
 
-    async def _massdns(self, module, domain, subdomains, rdtype):
-        """
-        {
-            "name": "www.blacklanternsecurity.com.",
-            "type": "A",
-            "class": "IN",
-            "status": "NOERROR",
-            "data": {
-            "answers": [
-                {
-                "ttl": 3600,
-                "type": "CNAME",
-                "class": "IN",
-                "name": "www.blacklanternsecurity.com.",
-                "data": "blacklanternsecurity.github.io."
-                },
-                {
-                "ttl": 3600,
-                "type": "A",
-                "class": "IN",
-                "name": "blacklanternsecurity.github.io.",
-                "data": "185.199.108.153"
-                }
-            ]
-            },
-            "resolver": "168.215.165.186:53"
-        }
+    async def _resolve_subdomains(self, domain, subdomains, rdtype):
+        """Resolve each candidate under ``domain``, yielding the ones that exist.
+
+        Uses ``resolve_batch_full`` rather than ``resolve_batch`` so failures stay
+        visible: a query that ran out of retries is a silently missed subdomain,
+        which is the one error brute-forcing must not hide.
         """
-        resolver_file = await self.resolver_file()
-        command = (
-            "massdns",
-            "-r",
-            resolver_file,
-            "-s",
-            self.max_resolvers,
-            "-t",
-            rdtype,
-            "-o",
-            "J",
-            "-q",
-        )
-        subdomains = self.gen_subdomains(subdomains, domain)
+        client = await self.client()
+        queries = self.gen_subdomains(subdomains, domain)
+        suffix = f".{domain}"
         hosts_yielded = set()
+        answered = 0
+        empty = 0
+        failed = 0
+
+        # Runs are serialized by dnsbrute_lock, so a stats snapshot taken here
+        # describes this run and no other.
         async with self.dnsbrute_lock:
-            async for line in module.run_process_live(*command, stderr=subprocess.DEVNULL, input=subdomains):
-                try:
-                    j = json.loads(line)
-                except json.decoder.JSONDecodeError:
-                    self.log.debug(f"Failed to decode line: {line}")
+            before = {s.resolver: s for s in client.stats()}
+            async for host, result in client.resolve_batch_full(queries, rdtype):
+                if isinstance(result, DNSError):
+                    failed += 1
                     continue
-                answers = j.get("data", {}).get("answers", [])
-                if type(answers) == list and len(answers) > 0:
-                    answer = answers[0]
-                    hostname = answer.get("name", "").strip(".").lower()
-                    if hostname.endswith(f".{domain}"):
-                        data = answer.get("data", "")
-                        rdtype = answer.get("type", "").upper()
-                        if data and rdtype:
-                            hostname_hash = hash(hostname)
-                            if hostname_hash not in hosts_yielded:
-                                hosts_yielded.add(hostname_hash)
-                                yield hostname, data, rdtype
-
-    async def gen_subdomains(self, prefixes, domain):
+                if not result.response.answers:
+                    empty += 1
+                    continue
+                answered += 1
+
+                # A CNAME can point anywhere; only names under the domain we
+                # asked about are subdomains of it.
+                hostname = host.strip(".").lower()
+                if not hostname.endswith(suffix) or hostname in hosts_yielded:
+                    continue
+                hosts_yielded.add(hostname)
+                yield hostname
+            after = client.stats()
+
+        self._log_delivery(domain, answered, empty, failed, before, after)
+
+    def _log_delivery(self, domain, answered, empty, failed, before, after):
+        """Report how much of the wordlist actually got an answer.
+
+        A run that silently drops queries looks identical to one that found
+        nothing, so unanswered queries are worth surfacing. When a meaningful
+        share went unanswered, also report the resolver-side picture, since
+        otherwise there's no way to tell rate-limiting apart from backoff.
+        """
+        total = answered + empty + failed
+        if not total:
+            return
+        message = f"Brute-force of {domain}: {answered:,} resolved, {empty:,} no record, {failed:,} unanswered"
+        if failed / total <= 0.05:
+            self.log.verbose(message)
+            return
+
+        used = 0
+        timeouts = 0
+        paced = 0
+        for stats in after:
+            previous = before.get(stats.resolver)
+            if stats.attempted - (previous.attempted if previous else 0):
+                used += 1
+            timeouts += stats.timeout - (previous.timeout if previous else 0)
+            if stats.rate_qps is not None:
+                paced += 1
+        self.log.warning(
+            f"{message} ({failed / total:.0%} unanswered, results may be incomplete). "
+            f"{used:,} resolvers used, {paced:,} throttled by backoff, {timeouts:,} timeouts"
+        )
+
+    def gen_subdomains(self, prefixes, domain):
         for p in prefixes:
             if domain:
                 p = f"{p}.{domain}"
@@ -158,12 +212,20 @@ async def resolver_file(self):
                 cache_hrs=24 * 7,
             )
             nameservers = set(self.parent_helper.read_file(self._resolver_file_original))
-            nameservers.difference_update(self.parent_helper.dns.system_resolvers)
-            # exclude system nameservers from brute-force
+            # exclude whatever the main DNS path is using, custom or system
             # this helps prevent rate-limiting which might cause BBOT's main dns queries to fail
+            # compare by host, since our resolvers may carry a port (8.8.8.8:53) while the brute list is bare IPs
+            excluded_hosts = {self._resolver_host(r) for r in self.parent_helper.dns.resolvers}
+            nameservers = {n for n in nameservers if self._resolver_host(n) not in excluded_hosts}
             self._resolver_file = self.parent_helper.tempfile(nameservers, pipe=False)
         return self._resolver_file
 
+    def _resolver_host(self, resolver):
+        try:
+            return str(self.parent_helper.split_host_port(resolver)[0])
+        except ValueError:
+            return resolver
+
     def gen_random_subdomains(self, n=50):
         delimiters = (".", "-")
         lengths = list(range(3, 8))