From 8f03afef5574e7f9f87383097e39f7deceb331af Mon Sep 17 00:00:00 2001 From: liquidsec Date: Tue, 19 May 2026 09:58:17 -0400 Subject: [PATCH] cloudcheck: skip storage-bucket regex pass for IP-only events MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Bucket-hostname regexes are anchored to provider-specific suffixes (`.amazonaws.com`, `.r2.dev`, etc.) and can never match a bare IP literal. Skip the regex loop when every host in `hosts_to_check` is an IP — relevant for scans where `speculate` expands large IP_RANGE events into per-IP IP_ADDRESS events. Adds a generic intercept-pipeline throughput benchmark covering the single-threaded intercept chain (dnsresolve, cloudcheck, speculate, etc.) under an IP-heavy workload. --- bbot/modules/internal/cloudcheck.py | 5 ++ .../test_intercept_throughput_benchmarks.py | 57 +++++++++++++++++++ 2 files changed, 62 insertions(+) create mode 100644 bbot/test/benchmarks/test_intercept_throughput_benchmarks.py diff --git a/bbot/modules/internal/cloudcheck.py b/bbot/modules/internal/cloudcheck.py index 461804ce18..5859aa9d34 100644 --- a/bbot/modules/internal/cloudcheck.py +++ b/bbot/modules/internal/cloudcheck.py @@ -67,6 +67,11 @@ async def handle_event(self, event, **kwargs): if event.scope_distance >= self.max_scope_distance: return + # storage-bucket hostnames are anchored to provider-specific suffixes + # (e.g. .amazonaws.com), so they can never match a bare IP literal + if all(self.helpers.is_ip(h) for h in hosts_to_check): + return + # see if any of our hosts are storage buckets, etc. regexes = await self.cloud_hostname_regexes() regexes = regexes.get("STORAGE_BUCKET_HOSTNAME", []) diff --git a/bbot/test/benchmarks/test_intercept_throughput_benchmarks.py b/bbot/test/benchmarks/test_intercept_throughput_benchmarks.py new file mode 100644 index 0000000000..9418437361 --- /dev/null +++ b/bbot/test/benchmarks/test_intercept_throughput_benchmarks.py @@ -0,0 +1,57 @@ +""" +Intercept-pipeline throughput benchmark. + +Measures how fast events flow through BBOT's intercept chain (dnsresolve, +cloudcheck, speculate, etc.) under a workload dominated by IP_ADDRESS events. +Useful for detecting regressions or improvements anywhere in the per-event +intercept path — single-threaded modules in this chain bound overall scan +throughput, so changes there are most easily observed here. + +The scenario uses a small private CIDR as the target. Speculate expands it +into IP_ADDRESS events, which then traverse the intercept chain. No external +network calls are made. + +Run with: + pytest bbot/test/benchmarks/test_intercept_throughput_benchmarks.py -v --benchmark-only +""" + +import asyncio + +import pytest + +from bbot.scanner import Scanner + + +def _run_scan(cidr): + config = { + "scope": {"search_distance": 0, "report_distance": 0}, + "dns": {"disable": True}, + "modules": {"speculate": {"ports": "80,443"}}, + "omit_event_types": [], + } + + scan = Scanner(cidr, config=config) + event_counts = {} + + async def _inner(): + async for event in scan.async_start(): + event_counts[event.type] = event_counts.get(event.type, 0) + 1 + + asyncio.run(_inner()) + return event_counts + + +class TestInterceptThroughputBenchmarks: + """Throughput of the intercept pipeline under IP-heavy workloads.""" + + @pytest.mark.benchmark(group="intercept_throughput_small") + def test_intercept_throughput_small(self, benchmark): + """/22 (1024 IPs) — modest fan-out, dominated by per-event intercept cost.""" + counts = benchmark.pedantic(_run_scan, args=("10.0.0.0/22",), rounds=3, warmup_rounds=1) + assert counts.get("IP_RANGE", 0) >= 1, f"expected ≥1 IP_RANGE event, got {counts}" + + @pytest.mark.benchmark(group="intercept_throughput_medium") + def test_intercept_throughput_medium(self, benchmark): + """/20 (4096 IPs) — larger fan-out exercising the intercept queue under load.""" + counts = benchmark.pedantic(_run_scan, args=("10.0.0.0/20",), rounds=3, warmup_rounds=1) + assert counts.get("IP_RANGE", 0) >= 1, f"expected ≥1 IP_RANGE event, got {counts}"