3.0.1 Release

.github/workflows/docs_updater.yml

@@ -32,3 +32,5 @@ jobs:
           base: dev
           title: "Automated Docs Update"
           body: "This is an automated pull request to update the documentation."
+          committer: blsaccess <info@blacklanternsecurity.com>
+          author: blsaccess <info@blacklanternsecurity.com>

README.md

@@ -343,7 +343,7 @@ For more information, see [Targets](https://www.blacklanternsecurity.com/bbot/St
 
 Similar to Amass or Subfinder, BBOT supports API keys for various third-party services such as SecurityTrails, etc.
 
-The standard way to do this is to enter your API keys in **`~/.config/bbot/bbot.yml`**. Note that multiple API keys are allowed:
+The standard way to do this is to enter your API keys in **`~/.config/bbot/secrets.yml`**. Note that multiple API keys are allowed:
 ```yaml
 modules:
   shodan_dns:

bbot/core/config/logger.py

@@ -8,7 +8,7 @@
 from pathlib import Path
 from contextlib import suppress
 
-from ..helpers.misc import mkdir, error_and_exit
+from ..helpers.misc import mkdir, error_and_exit, make_printable
 from ..multiprocess import SHARED_INTERPRETER_STATE
 from ...logger import colorize, loglevel_mapping, GzipRotatingFileHandler
 
@@ -25,6 +25,15 @@ class ColoredFormatter(logging.Formatter):
 
     def format(self, record):
         colored_record = copy(record)
+        # escape control characters in scan-derived text so they can't corrupt the terminal
+        colored_record.msg = make_printable(colored_record.getMessage())
+        colored_record.args = None
+        if colored_record.exc_info and not colored_record.exc_text:
+            colored_record.exc_text = self.formatException(colored_record.exc_info)
+        if colored_record.exc_text:
+            colored_record.exc_text = make_printable(colored_record.exc_text)
+        if colored_record.stack_info:
+            colored_record.stack_info = make_printable(colored_record.stack_info)
         levelname = colored_record.levelname
         levelshort = loglevel_mapping.get(levelname, "INFO")
         colored_record.levelname = colorize(f"[{levelshort}]", level=levelname)

bbot/core/event/base.py

@@ -790,7 +790,6 @@ def _minimize(self):
             # release container slots; lazy-init pattern means None == empty
             self._dns_children = None
             self._raw_dns_records = None
-            self._resolved_hosts = None
 
     def clone(self):
         # Create a shallow copy of the event first

bbot/core/helpers/depsinstaller/installer.py

@@ -453,14 +453,16 @@ def _core_dep_satisfied(self, command):
         For special entries like openssl_dev_headers, use a custom check.
         """
         if command == "openssl_dev_headers":
-            # check for openssl headers by looking for the pkg-config file or header
+            # look for the actual header. the openssl binary is NOT a proxy: many minimal
+            # images (e.g. python:3.11-slim) ship openssl but not the -dev headers, and
+            # anything that compiles against libssl needs the header.
             return any(
                 Path(p).exists()
                 for p in [
                     "/usr/include/openssl/ssl.h",
                     "/usr/local/include/openssl/ssl.h",
                 ]
-            ) or bool(self.parent_helper.which("openssl"))
+            )
         return bool(self.parent_helper.which(command))
 
     async def install_core_deps(self):

bbot/core/helpers/misc.py

@@ -2999,3 +2999,18 @@ def is_printable(s):
     # Exclude control characters that break display/printing
     s = set(s)
     return all(ord(c) >= 32 or c in "\t\n\r" for c in s)
+
+
+_control_char_re = re.compile(r"[\x00-\x08\x0b\x0c\x0e-\x1f]")
+
+
+def make_printable(s):
+    """
+    Escape non-printable control characters so a string is safe to write to a terminal.
+
+    Keeps tab, newline, and carriage return; renders other control bytes (e.g. 0x0e Shift Out) as \\xNN.
+    Inverse of is_printable().
+    """
+    if not isinstance(s, str):
+        s = smart_decode(s)
+    return _control_char_re.sub(lambda m: f"\\x{ord(m.group()):02x}", s)

bbot/logger.py

@@ -46,8 +46,11 @@ def log_to_stderr(msg, level="INFO", logname=True):
     """
     Print to stderr with BBOT logger colors
     """
+    from bbot.core.helpers.misc import make_printable
+
     levelname = level.upper()
     if not any(x in sys.argv for x in ("-s", "--silent")):
+        msg = make_printable(msg)
         levelshort = f"[{loglevel_mapping.get(level, 'INFO')}]"
         levelshort = f"{colorize(levelshort, level=levelname)}"
         if levelname == "CRITICAL" or levelname.startswith("HUGE"):

bbot/modules/nuclei.py

@@ -21,7 +21,7 @@ class nuclei(BaseModule):
     }
 
     class Config(BaseModuleConfig):
-        version: str = Field("3.9.0", description="nuclei version")
+        version: str = Field("3.11.0", description="nuclei version")
         tags: str = Field("", description="execute a subset of templates that contain the provided tags")
         templates: str = Field("", description="template or template directory paths to include in the scan")
         severity: str = Field("", description="Filter based on severity field available in the template.")

bbot/modules/output/stdout.py

@@ -70,6 +70,9 @@ async def handle_text(self, event, event_json):
         else:
             event_str = self.human_event_str(event)
 
+        # escape control characters so they can't corrupt the terminal
+        event_str = self.helpers.make_printable(event_str)
+
         # color findings: severity picks the hue, confidence dims the brightness
         if event.type == "FINDING" and isinstance(event.data, dict) and self.use_color:
             event_str = self._colorize_finding(event_str, event)

bbot/modules/trufflehog.py

@@ -15,7 +15,7 @@ class trufflehog(BaseModule):
     }
 
     class Config(BaseModuleConfig):
-        version: str = Field("3.95.5", description="trufflehog version")
+        version: str = Field("3.95.9", description="trufflehog version")
         config: str = Field("", description="File path or URL to YAML trufflehog config")
         only_verified: bool = Field(True, description="Only report credentials that have been verified")
         concurrency: int = Field(8, description="Number of concurrent workers")

bbot/modules/virtualhost.py

@@ -990,6 +990,10 @@ async def finish(self):
 
                 self.verbose(f"FINISH METHOD: Starting wildcard check for {host}")
                 baseline_response = await self._get_baseline_response(event, host, host_ip)
+                if not baseline_response:
+                    self.debug(f"FINISH METHOD: Failed to get baseline response for {host}, skipping wordcloud check")
+                    self.wordcloud_tried_hosts.add(host)
+                    continue
                 if not await self._wildcard_canary_check(
                     host_parsed_url.scheme, host_parsed_url.netloc, event, host_ip, baseline_response
                 ):
@@ -1015,10 +1019,10 @@ async def finish(self):
 
     async def filter_event(self, event):
         if (
-            "cdn-cloudflare" in event.tags
-            or "cdn-imperva" in event.tags
-            or "cdn-akamai" in event.tags
-            or "cdn-cloudfront" in event.tags
+            "cloudflare" in event.tags
+            or "imperva" in event.tags
+            or "akamai" in event.tags
+            or "cloudfront" in event.tags
         ):
             self.debug(f"Not processing URL {event.url} because it's behind a WAF or CDN, and that's pointless")
             return False

bbot/modules/waf_bypass.py

@@ -26,11 +26,16 @@ class Config(BaseModuleConfig):
         similarity_threshold: float = Field(0.90, description="Similarity threshold for content matching")
         search_ip_neighbors: bool = Field(True, description="Also check IP neighbors of the target domain")
         neighbor_cidr: int = Field(
-            24,
+            28,
             ge=24,
             le=31,
             description="CIDR mask (24-31) used for neighbor enumeration when search_ip_neighbors is true",
         )
+        max_concurrent_checks: int = Field(
+            100,
+            ge=1,
+            description="Maximum number of concurrent bypass-attempt HTTP checks in finish()",
+        )
 
     meta = {
         "description": "Detects potential WAF bypasses",
@@ -77,15 +82,15 @@ async def handle_event(self, event):
                         self.cloud_ips.add(ip_str)
                         self.debug(f"Added cloud-ip {ip_str} to cloud_ips")
                 else:
-                    self.warning(f"DNS resolution for {domain} returned non-IP result: {ip_str}")
+                    self.verbose(f"DNS resolution for {domain} returned non-IP result: {ip_str}")
         else:
             self.warning(f"DNS resolution failed for {domain}")
 
         # Detect WAF/CDN protection based on tags
         provider_name = None
-        if "cdn-cloudflare" in event.tags or "waf-cloudflare" in event.tags:
+        if "cloudflare" in event.tags:
             provider_name = "CloudFlare"
-        elif "cdn-imperva" in event.tags:
+        elif "imperva" in event.tags:
             provider_name = "Imperva"
 
         is_protected = provider_name is not None
@@ -189,8 +194,10 @@ async def finish(self):
                         continue
 
                     if ip not in waf_ips:  # And IP isn't a known WAF IP
-                        ip_bypass_candidates[ip] = domain
-                        self.debug(f"Added potential bypass IP {ip} from domain {domain}")
+                        # first attribution wins so logs are stable across scans
+                        if ip not in ip_bypass_candidates:
+                            ip_bypass_candidates[ip] = f"{domain} (direct)"
+                            self.debug(f"Added potential bypass IP {ip} from domain {domain}")
 
                         # if we have IP neighbors searching enabled, and the IP isn't a cloud IP, we can add the IP neighbors to our list of potential bypasses
                         if self.search_ip_neighbors and ip not in self.cloud_ips:
@@ -221,7 +228,9 @@ async def finish(self):
                                         self.debug(
                                             f"Added Neighbor IP ({ip} -> {neighbor_ip_str}) as potential bypass IP derived from {domain}"
                                         )
-                                        ip_bypass_candidates[neighbor_ip_str] = domain
+                                        ip_bypass_candidates[neighbor_ip_str] = (
+                                            f"{domain} (neighbor of {ip}/{self.neighbor_cidr})"
+                                        )
                     else:
                         self.debug(f"IP {ip} is in WAF IPS so we don't check as potential bypass")
 
@@ -245,7 +254,9 @@ async def finish(self):
         )
 
         self.debug(f"about to start {len(coros)} coroutines")
-        async for completed in self.helpers.as_completed(coros):
+        async for completed in self.helpers.as_completed(
+            coros, max_concurrent=int(self.config.get("max_concurrent_checks") or 100)
+        ):
             result = await completed
             if result:
                 confirmed_bypasses.append(result)

bbot/presets/kitchen-sink.yml

@@ -14,8 +14,6 @@ include:
 
 config:
   modules:
-    baddns:
-      enable_references: True
     dnsbrute:
       recursive_mutations: true
     dnscommonsrv:

bbot/presets/waf-bypass.yml

@@ -16,4 +16,4 @@ config:
     waf_bypass:
       similarity_threshold: 0.90
       search_ip_neighbors: true
-      neighbor_cidr: 24 
+      neighbor_cidr: 28

bbot/test/test_step_1/test_cli.py

@@ -922,6 +922,75 @@ async def test_cli_no_color(monkeypatch):
     assert os.environ.get("NO_COLOR") == "1"
 
 
+@pytest.mark.asyncio
+async def test_cli_console_control_chars(monkeypatch, capsys):
+    import logging
+    from bbot.logger import log_to_stderr
+    from bbot.core.config.logger import ColoredFormatter
+
+    monkeypatch.delenv("NO_COLOR", raising=False)
+
+    # scan-derived text carrying the bytes that flip a terminal into its line-drawing charset:
+    # 0x0e (Shift Out) and an "ESC ( 0" designate-line-drawing sequence
+    dirty = "matched banner: \x1b(0\x0edeadbeef"
+
+    # the stderr log formatter escapes control chars before adding its own colors
+    formatter = ColoredFormatter("%(levelname)s %(name)s: %(message)s")
+    record = logging.LogRecord(
+        name="bbot.modules.nuclei",
+        level=logging.INFO,
+        pathname=__file__,
+        lineno=1,
+        msg="%s",
+        args=(dirty,),
+        exc_info=None,
+    )
+    formatted = formatter.format(record)
+    assert "\x0e" not in formatted
+    assert "\x1b(0" not in formatted
+    assert "\\x0e" in formatted and "\\x1b(0" in formatted
+    # BBOT's own ANSI color codes survive (escaping happens on the message, before colorizing)
+    assert "\x1b[" in formatted
+
+    # same protection on the early bootstrap path
+    log_to_stderr(dirty, level="INFO")
+    out, err = capsys.readouterr()
+    assert "\x0e" not in err and "\x1b(0" not in err
+    assert "\\x0e" in err
+
+    # tracebacks and stack info are also sanitized (an exception carrying scan-derived
+    # bytes must not reach the terminal unescaped via logger.exception / exc_info=True)
+    try:
+        raise ValueError(dirty)
+    except ValueError:
+        exc_record = logging.LogRecord(
+            name="bbot.modules.nuclei",
+            level=logging.ERROR,
+            pathname=__file__,
+            lineno=1,
+            msg="boom",
+            args=None,
+            exc_info=sys.exc_info(),
+        )
+    formatted = formatter.format(exc_record)
+    assert "\x0e" not in formatted and "\x1b(0" not in formatted
+    assert "\\x0e" in formatted and "ValueError" in formatted
+
+    stack_record = logging.LogRecord(
+        name="bbot.modules.nuclei",
+        level=logging.ERROR,
+        pathname=__file__,
+        lineno=1,
+        msg="boom",
+        args=None,
+        exc_info=None,
+    )
+    stack_record.stack_info = f"Stack: {dirty}"
+    formatted = formatter.format(stack_record)
+    assert "\x0e" not in formatted and "\x1b(0" not in formatted
+    assert "\\x0e" in formatted
+
+
 @pytest.mark.asyncio
 async def test_cli_reset_config(monkeypatch, caplog, tmp_path):
     from bbot.core import CORE

bbot/test/test_step_1/test_helpers.py

@@ -472,6 +472,17 @@ async def test_helpers_misc(helpers, scan, bbot_scanner, bbot_httpserver):
     assert helpers.is_printable("4") is True
     assert helpers.is_printable("asdf\x00") is False
 
+    # make_printable
+    assert helpers.make_printable("asdf") == "asdf"
+    assert helpers.make_printable("ドメイン.テスト") == "ドメイン.テスト"
+    # 0x0e (Shift Out) is the byte that flips a terminal into its line-drawing charset
+    assert helpers.make_printable("a\x0eb") == "a\\x0eb"
+    assert helpers.make_printable("\x00\x1b\x7f") == "\\x00\\x1b\x7f"
+    # tab, newline, and carriage return are preserved
+    assert helpers.make_printable("tab\tnl\ncr\r") == "tab\tnl\ncr\r"
+    # bytes are decoded before escaping
+    assert helpers.make_printable(b"a\x0eb") == "a\\x0eb"
+
     # punycode
     assert helpers.smart_encode_punycode("ドメイン.テスト") == "xn--eckwd4c7c.xn--zckzah"
     assert helpers.smart_decode_punycode("xn--eckwd4c7c.xn--zckzah") == "ドメイン.テスト"

bbot/test/test_step_1/test_validate_preset.py

@@ -212,6 +212,23 @@ def test_validate_preset_interactsh_server_rejects_invalid():
         assert repr(v) in errs[0].message
 
 
+def test_bundled_presets_validate_clean():
+    """Every YAML preset shipped under bbot/presets/ must pass validation."""
+    from pathlib import Path
+    import bbot
+    from bbot.scanner import validate_preset_file
+
+    presets_dir = Path(bbot.__file__).parent / "presets"
+    failures = {}
+    for preset_path in sorted(presets_dir.rglob("*.yml")):
+        errs = validate_preset_file(preset_path)
+        if errs:
+            failures[str(preset_path.relative_to(presets_dir))] = [str(e) for e in errs]
+    assert not failures, "bundled presets failed validation:\n" + "\n".join(
+        f"  {name}:\n    " + "\n    ".join(msgs) for name, msgs in failures.items()
+    )
+
+
 def test_defaults_yml_validates_against_schema():
     """BBOT's own defaults.yml must pass its own validator. Guards against the pydantic
     schema (core/config/models.py) silently drifting from defaults.yml -- if a key is