From b12316761091f41ab1b37aa93b7071c5ccf1ab95 Mon Sep 17 00:00:00 2001 From: liquidsec Date: Tue, 14 Apr 2026 16:44:50 -0400 Subject: [PATCH 1/4] Pause ingress on memory pressure to prevent OOM kills --- bbot/defaults.yml | 3 ++ bbot/scanner/manager.py | 6 ++++ bbot/scanner/scanner.py | 46 +++++++++++++++++++++---- bbot/test/test_step_1/test_scan.py | 54 ++++++++++++++++++++++++++++++ 4 files changed, 103 insertions(+), 6 deletions(-) diff --git a/bbot/defaults.yml b/bbot/defaults.yml index 3856a1f644..84545b2202 100644 --- a/bbot/defaults.yml +++ b/bbot/defaults.yml @@ -15,6 +15,9 @@ home: ~/.bbot keep_scans: 20 # Interval for displaying status messages status_frequency: 15 +# Pause event processing when system memory exceeds this percentage. +# Workers finish their current event then stop pulling new ones until memory drops. +max_mem_percent: 90 # Include the raw data of files (i.e. PDFs, web screenshots) as base64 in the event file_blobs: false # Include the raw data of directories (i.e. git repos) as tar.gz base64 in the event diff --git a/bbot/scanner/manager.py b/bbot/scanner/manager.py index 0ad7f59c85..f34a4c59e8 100644 --- a/bbot/scanner/manager.py +++ b/bbot/scanner/manager.py @@ -145,6 +145,12 @@ def module_priority_weights(self): return self._module_priority_weights async def get_incoming_event(self): + # memory-pressure backpressure: pause ingress while RAM is high. + # modules keep draining their queues (freeing memory via _minimize()), + # we just stop feeding new events into the pipeline. + if not self.scan._memory_ok.is_set(): + await self.scan._memory_ok.wait() + for q in self.helpers.weighted_shuffle(self.incoming_queues, self.module_priority_weights): try: return q.get_nowait() diff --git a/bbot/scanner/scanner.py b/bbot/scanner/scanner.py index edfe5ca8db..f1b3900ab7 100644 --- a/bbot/scanner/scanner.py +++ b/bbot/scanner/scanner.py @@ -1,4 +1,5 @@ import sys +import time import asyncio import logging import traceback @@ -264,6 +265,12 @@ def __init__( # how often to print scan status self.status_frequency = self.config.get("status_frequency", 15) + # memory-pressure backpressure: cleared when RAM is high, set when OK. + # module worker loops await this before pulling new events. + self._memory_ok = asyncio.Event() + self._memory_ok.set() + self._memory_paused_since = None + from .stats import ScanStats self.stats = ScanStats(self) @@ -728,14 +735,41 @@ def modules_status(self, _log=False, detailed=False): modules_errored = [m for m, s in status["modules"].items() if s["errored"]] - max_mem_percent = 90 mem_status = self.helpers.memory_status() - # abort if we don't have the memory mem_percent = mem_status.percent - if mem_percent > max_mem_percent: - free_memory = mem_status.available - free_memory_human = self.helpers.bytes_to_human(free_memory) - self.warning(f"System memory is at {mem_percent:.1f}% ({free_memory_human} remaining)") + high_watermark = self.config.get("max_mem_percent", 90) + # require memory to drop a few points before resuming to avoid flapping + low_watermark = max(0, high_watermark - 5) + # safety valve: force-resume after this many seconds to prevent stuck scans + max_pause_seconds = 120 + + if mem_percent > high_watermark: + free_memory_human = self.helpers.bytes_to_human(mem_status.available) + if self._memory_ok.is_set(): + self.warning( + f"System memory is at {mem_percent:.1f}% ({free_memory_human} remaining); " + f"pausing event processing until it drops below {low_watermark}%" + ) + self._memory_ok.clear() + self._memory_paused_since = time.time() + else: + paused_for = time.time() - self._memory_paused_since + if paused_for >= max_pause_seconds: + self.warning( + f"System memory still at {mem_percent:.1f}% after {paused_for:.0f}s paused; " + f"force-resuming to prevent stuck scan" + ) + self._memory_ok.set() + self._memory_paused_since = None + else: + self.warning( + f"System memory still at {mem_percent:.1f}% ({free_memory_human} remaining); " + f"event processing paused ({paused_for:.0f}s / {max_pause_seconds}s)" + ) + elif not self._memory_ok.is_set() and mem_percent <= low_watermark: + self.hugesuccess(f"System memory dropped to {mem_percent:.1f}%; resuming event processing") + self._memory_ok.set() + self._memory_paused_since = None if _log: modules_status = [] diff --git a/bbot/test/test_step_1/test_scan.py b/bbot/test/test_step_1/test_scan.py index a23b3c4240..c2867e934a 100644 --- a/bbot/test/test_step_1/test_scan.py +++ b/bbot/test/test_step_1/test_scan.py @@ -431,3 +431,57 @@ async def test_scan_name(bbot_scanner): await scan._prep() assert scan.name == "test_scan_name" assert scan.preset.scan_name == "test_scan_name" + + +@pytest.mark.asyncio +async def test_memory_backpressure(bbot_scanner, monkeypatch): + """Test that memory pressure pauses ingress and the scan still completes.""" + from types import SimpleNamespace + + mem_percent = [50.0] # mutable so the mock can be changed mid-scan + + def mock_memory_status(): + return SimpleNamespace(percent=mem_percent[0], available=1_000_000_000) + + scan = bbot_scanner("127.0.0.1", config={"max_mem_percent": 90}) + await scan._prep() + await scan.helpers.dns._mock_dns({"1.1.1.1.in-addr.arpa": {"PTR": ["one.one.one.one"]}}) + monkeypatch.setattr("bbot.core.helpers.misc.memory_status", mock_memory_status) + + # starts in the OK state + assert scan._memory_ok.is_set() + + # simulate high memory — should pause + mem_percent[0] = 95.0 + scan.modules_status(_log=False) + assert not scan._memory_ok.is_set(), "Ingress should be paused when memory exceeds high watermark" + assert scan._memory_paused_since is not None + + # memory drops but not below low watermark (85) — should stay paused + mem_percent[0] = 87.0 + scan.modules_status(_log=False) + assert not scan._memory_ok.is_set(), "Should stay paused until memory drops below low watermark" + + # memory drops below low watermark — should resume + mem_percent[0] = 80.0 + scan.modules_status(_log=False) + assert scan._memory_ok.is_set(), "Ingress should resume when memory drops below low watermark" + assert scan._memory_paused_since is None + + # test force-resume safety valve + import time + + mem_percent[0] = 95.0 + scan.modules_status(_log=False) + assert not scan._memory_ok.is_set() + # fake that we've been paused for a long time + scan._memory_paused_since = time.time() - 200 + scan.modules_status(_log=False) + assert scan._memory_ok.is_set(), "Should force-resume after max pause duration to prevent stuck scan" + + # verify the scan still completes with memory pressure active during the run + mem_percent[0] = 50.0 + scan._memory_ok.set() + scan._memory_paused_since = None + events = [e async for e in scan.async_start()] + assert any(e.type == "IP_ADDRESS" for e in events), "Scan should still produce events" From 731569a34da3d6de07e9df5d55649d50b5f6d00e Mon Sep 17 00:00:00 2001 From: liquidsec Date: Tue, 14 Apr 2026 16:45:34 -0400 Subject: [PATCH 2/4] Add max_mem_percent to config docs --- docs/scanning/configuration.md | 3 +++ 1 file changed, 3 insertions(+) diff --git a/docs/scanning/configuration.md b/docs/scanning/configuration.md index bbc5aa7a22..737307725c 100644 --- a/docs/scanning/configuration.md +++ b/docs/scanning/configuration.md @@ -74,6 +74,9 @@ home: ~/.bbot keep_scans: 20 # Interval for displaying status messages status_frequency: 15 +# Pause event processing when system memory exceeds this percentage. +# Workers finish their current event then stop pulling new ones until memory drops. +max_mem_percent: 90 # Include the raw data of files (i.e. PDFs, web screenshots) as base64 in the event file_blobs: false # Include the raw data of directories (i.e. git repos) as tar.gz base64 in the event From 790eab4e378f29662f7bd0ae56d70876b76548dc Mon Sep 17 00:00:00 2001 From: liquidsec Date: Tue, 14 Apr 2026 19:01:43 -0400 Subject: [PATCH 3/4] Add graduated throttle mode: scale ingress delay with memory overshoot --- bbot/scanner/manager.py | 2 ++ bbot/scanner/scanner.py | 48 ++++++++++++++++++++++-------- bbot/test/test_step_1/test_scan.py | 31 +++++++++++++------ 3 files changed, 59 insertions(+), 22 deletions(-) diff --git a/bbot/scanner/manager.py b/bbot/scanner/manager.py index f34a4c59e8..1d5c8fa556 100644 --- a/bbot/scanner/manager.py +++ b/bbot/scanner/manager.py @@ -150,6 +150,8 @@ async def get_incoming_event(self): # we just stop feeding new events into the pipeline. if not self.scan._memory_ok.is_set(): await self.scan._memory_ok.wait() + elif self.scan._memory_throttled: + await asyncio.sleep(self.scan._memory_throttle_delay) for q in self.helpers.weighted_shuffle(self.incoming_queues, self.module_priority_weights): try: diff --git a/bbot/scanner/scanner.py b/bbot/scanner/scanner.py index f1b3900ab7..73845e0923 100644 --- a/bbot/scanner/scanner.py +++ b/bbot/scanner/scanner.py @@ -266,10 +266,14 @@ def __init__( self.status_frequency = self.config.get("status_frequency", 15) # memory-pressure backpressure: cleared when RAM is high, set when OK. - # module worker loops await this before pulling new events. + # ingress awaits _memory_ok before pulling new events. + # _memory_throttled = slow drip mode after safety valve fires. + # _memory_throttle_delay = seconds between events, scales with overshoot. self._memory_ok = asyncio.Event() self._memory_ok.set() self._memory_paused_since = None + self._memory_throttled = False + self._memory_throttle_delay = 0.0 from .stats import ScanStats @@ -737,38 +741,49 @@ def modules_status(self, _log=False, detailed=False): mem_status = self.helpers.memory_status() mem_percent = mem_status.percent - high_watermark = self.config.get("max_mem_percent", 90) - # require memory to drop a few points before resuming to avoid flapping - low_watermark = max(0, high_watermark - 5) + max_mem_percent = self.config.get("max_mem_percent", 90) + resume_threshold = max(0, max_mem_percent - 2) # safety valve: force-resume after this many seconds to prevent stuck scans max_pause_seconds = 120 - if mem_percent > high_watermark: + if mem_percent > max_mem_percent: free_memory_human = self.helpers.bytes_to_human(mem_status.available) - if self._memory_ok.is_set(): + if self._memory_ok.is_set() and not self._memory_throttled: self.warning( f"System memory is at {mem_percent:.1f}% ({free_memory_human} remaining); " - f"pausing event processing until it drops below {low_watermark}%" + f"pausing event processing until it drops below {resume_threshold}%" ) self._memory_ok.clear() + self._memory_throttled = False self._memory_paused_since = time.time() - else: + elif not self._memory_ok.is_set(): paused_for = time.time() - self._memory_paused_since if paused_for >= max_pause_seconds: + self._memory_ok.set() + self._memory_throttled = True + self._memory_paused_since = None + self._update_throttle_delay(mem_percent, max_mem_percent) self.warning( f"System memory still at {mem_percent:.1f}% after {paused_for:.0f}s paused; " - f"force-resuming to prevent stuck scan" + f"switching to throttled ingress ({self._memory_throttle_delay:.1f}s/event)" ) - self._memory_ok.set() - self._memory_paused_since = None else: self.warning( f"System memory still at {mem_percent:.1f}% ({free_memory_human} remaining); " f"event processing paused ({paused_for:.0f}s / {max_pause_seconds}s)" ) - elif not self._memory_ok.is_set() and mem_percent <= low_watermark: - self.hugesuccess(f"System memory dropped to {mem_percent:.1f}%; resuming event processing") + else: + # throttled mode, memory still high — update delay and keep dripping + self._update_throttle_delay(mem_percent, max_mem_percent) + self.warning( + f"System memory still at {mem_percent:.1f}% ({free_memory_human} remaining); " + f"ingress throttled ({self._memory_throttle_delay:.1f}s/event)" + ) + elif (self._memory_throttled or not self._memory_ok.is_set()) and mem_percent <= resume_threshold: + self.hugesuccess(f"System memory dropped to {mem_percent:.1f}%; resuming full-speed event processing") self._memory_ok.set() + self._memory_throttled = False + self._memory_throttle_delay = 0.0 self._memory_paused_since = None if _log: @@ -842,6 +857,13 @@ def modules_status(self, _log=False, detailed=False): return status + def _update_throttle_delay(self, mem_percent, max_mem_percent): + """Scale throttle delay linearly: 0s at threshold, 5s at cap (threshold+5 or 95%, whichever is lower).""" + cap = min(max_mem_percent + 5, 95) + overshoot_range = max(cap - max_mem_percent, 1) + overshoot = min(mem_percent - max_mem_percent, overshoot_range) + self._memory_throttle_delay = max(0.0, (overshoot / overshoot_range) * 5.0) + async def async_stop(self): """Stops the in-progress scan and performs necessary cleanup. diff --git a/bbot/test/test_step_1/test_scan.py b/bbot/test/test_step_1/test_scan.py index c2867e934a..9229119935 100644 --- a/bbot/test/test_step_1/test_scan.py +++ b/bbot/test/test_step_1/test_scan.py @@ -454,34 +454,47 @@ def mock_memory_status(): # simulate high memory — should pause mem_percent[0] = 95.0 scan.modules_status(_log=False) - assert not scan._memory_ok.is_set(), "Ingress should be paused when memory exceeds high watermark" + assert not scan._memory_ok.is_set(), "Ingress should be paused when memory exceeds threshold" assert scan._memory_paused_since is not None - # memory drops but not below low watermark (85) — should stay paused - mem_percent[0] = 87.0 + # memory drops but not below resume threshold (88) — should stay paused + mem_percent[0] = 89.0 scan.modules_status(_log=False) - assert not scan._memory_ok.is_set(), "Should stay paused until memory drops below low watermark" + assert not scan._memory_ok.is_set(), "Should stay paused until memory drops below resume threshold" - # memory drops below low watermark — should resume - mem_percent[0] = 80.0 + # memory drops below resume threshold — should resume + mem_percent[0] = 87.0 scan.modules_status(_log=False) - assert scan._memory_ok.is_set(), "Ingress should resume when memory drops below low watermark" + assert scan._memory_ok.is_set(), "Ingress should resume when memory drops below resume threshold" assert scan._memory_paused_since is None - # test force-resume safety valve + # test safety valve → throttled mode import time mem_percent[0] = 95.0 scan.modules_status(_log=False) assert not scan._memory_ok.is_set() + assert not scan._memory_throttled # fake that we've been paused for a long time scan._memory_paused_since = time.time() - 200 scan.modules_status(_log=False) - assert scan._memory_ok.is_set(), "Should force-resume after max pause duration to prevent stuck scan" + # should switch to throttled mode: _memory_ok set but _memory_throttled on + assert scan._memory_ok.is_set(), "Should open ingress after max pause duration" + assert scan._memory_throttled, "Should be in throttled mode, not full speed" + # next tick: memory still high → stays throttled (doesn't re-pause) + scan.modules_status(_log=False) + assert scan._memory_ok.is_set(), "Should stay open in throttled mode" + assert scan._memory_throttled, "Should remain throttled while memory is high" + # memory drops → back to full speed + mem_percent[0] = 80.0 + scan.modules_status(_log=False) + assert scan._memory_ok.is_set(), "Should be fully resumed" + assert not scan._memory_throttled, "Throttle should be cleared when memory drops" # verify the scan still completes with memory pressure active during the run mem_percent[0] = 50.0 scan._memory_ok.set() + scan._memory_throttled = False scan._memory_paused_since = None events = [e async for e in scan.async_start()] assert any(e.type == "IP_ADDRESS" for e in events), "Scan should still produce events" From d4d0e7474e5a0b4710bff974cf5b5e1a464691da Mon Sep 17 00:00:00 2001 From: liquidsec Date: Wed, 15 Apr 2026 15:25:54 -0400 Subject: [PATCH 4/4] Fix httpx reference in iis_shortnames gateway error test --- .../test/test_step_2/module_tests/test_module_iis_shortnames.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/bbot/test/test_step_2/module_tests/test_module_iis_shortnames.py b/bbot/test/test_step_2/module_tests/test_module_iis_shortnames.py index abc2f14719..0821eba42e 100644 --- a/bbot/test/test_step_2/module_tests/test_module_iis_shortnames.py +++ b/bbot/test/test_step_2/module_tests/test_module_iis_shortnames.py @@ -110,7 +110,7 @@ class TestIIS_Shortnames_GatewayError(ModuleTestBase): """Negative test: server returns 502 gateway errors. Should NOT detect IIS shortnames.""" targets = ["http://127.0.0.1:8888"] - modules_overrides = ["httpx", "iis_shortnames"] + modules_overrides = ["http", "iis_shortnames"] async def setup_after_prep(self, module_test): module_test.httpserver.no_handler_status_code = 404