Merge branch 'dev' into 3.0

Author: github-actions

bbot/cli.py

@@ -1,5 +1,6 @@
 #!/usr/bin/env python3
 
+import io
 import sys
 import logging
 import multiprocessing
@@ -240,7 +241,12 @@ def handle_keyboard_input(keyboard_input):
 
                 # set stdout and stderr to blocking mode
                 # this is needed to prevent BlockingIOErrors in logging etc.
-                fds = [sys.stdout.fileno(), sys.stderr.fileno()]
+                fds = []
+                for stream in [sys.stdout, sys.stderr]:
+                    try:
+                        fds.append(stream.fileno())
+                    except io.UnsupportedOperation:
+                        log.debug(f"Can't get fileno for {stream}")
                 for fd in fds:
                     flags = fcntl.fcntl(fd, fcntl.F_GETFL)
                     fcntl.fcntl(fd, fcntl.F_SETFL, flags & ~os.O_NONBLOCK)

bbot/core/config/logger.py

@@ -9,9 +9,8 @@
 from contextlib import suppress
 
 from ..helpers.misc import mkdir, error_and_exit
-from ...logger import colorize, loglevel_mapping
 from ..multiprocess import SHARED_INTERPRETER_STATE
-
+from ...logger import colorize, loglevel_mapping, GzipRotatingFileHandler
 
 debug_format = logging.Formatter("%(asctime)s [%(levelname)s] %(name)s %(filename)s:%(lineno)s %(message)s")
 
@@ -216,15 +215,11 @@ def log_handlers(self):
             if not mkdir(log_dir, raise_error=False):
                 error_and_exit(f"Failure creating or error writing to BBOT logs directory ({log_dir})")
 
-            # Main log file
-            main_handler = logging.handlers.TimedRotatingFileHandler(
-                f"{log_dir}/bbot.log", when="d", interval=1, backupCount=14
-            )
+            # Main log file (compressed)
+            main_handler = GzipRotatingFileHandler(f"{log_dir}/bbot.log", when="d", interval=1, backupCount=14)
 
-            # Separate log file for debugging
-            debug_handler = logging.handlers.TimedRotatingFileHandler(
-                f"{log_dir}/bbot.debug.log", when="d", interval=1, backupCount=14
-            )
+            # Separate log file for debugging (compressed)
+            debug_handler = GzipRotatingFileHandler(f"{log_dir}/bbot.debug.log", when="d", interval=1, backupCount=14)
 
             # Log to stderr
             stderr_handler = logging.StreamHandler(sys.stderr)

bbot/logger.py

@@ -1,4 +1,5 @@
 import sys
+import logging.handlers
 
 loglevel_mapping = {
     "DEBUG": "DBUG",
@@ -50,3 +51,11 @@ def log_to_stderr(msg, level="INFO", logname=True):
         if logname:
             msg = f"{levelshort} {msg}"
         print(msg, file=sys.stderr)
+
+
+# Create a compressed file handler for logs
+class GzipRotatingFileHandler(logging.handlers.TimedRotatingFileHandler):
+    def _open(self):
+        import gzip
+
+        return gzip.open(f"{self.baseFilename}.gz", mode="at", encoding=self.encoding)

bbot/modules/crt_db.py

@@ -29,11 +29,11 @@ async def setup(self):
     async def request_url(self, query):
         if not self.db_conn:
             self.db_conn = await asyncpg.connect(
-                host=self.db_host, 
-                port=self.db_port, 
-                user=self.db_user, 
+                host=self.db_host,
+                port=self.db_port,
+                user=self.db_user,
                 database=self.db_name,
-                statement_cache_size=0  # Disable automatic statement preparation
+                statement_cache_size=0,  # Disable automatic statement preparation
             )
 
         sql = """

bbot/scanner/scanner.py

@@ -15,6 +15,7 @@
 from .manager import ScanIngress, ScanEgress
 from bbot.core.helpers.misc import sha1, rand_string
 from bbot.core.helpers.names_generator import random_name
+from bbot.core.config.logger import GzipRotatingFileHandler
 from bbot.core.multiprocess import SHARED_INTERPRETER_STATE
 from bbot.core.helpers.async_helpers import async_to_sync_gen
 from bbot.errors import BBOTError, ScanError, ValidationError
@@ -1230,13 +1231,9 @@ def log_level(self):
     def _log_handlers(self):
         if self.__log_handlers is None:
             self.helpers.mkdir(self.home)
-            main_handler = logging.handlers.TimedRotatingFileHandler(
-                str(self.home / "scan.log"), when="d", interval=1, backupCount=14
-            )
+            main_handler = GzipRotatingFileHandler(str(self.home / "scan.log"), when="d", interval=1, backupCount=14)
             main_handler.addFilter(lambda x: x.levelno != logging.TRACE and x.levelno >= logging.VERBOSE)
-            debug_handler = logging.handlers.TimedRotatingFileHandler(
-                str(self.home / "debug.log"), when="d", interval=1, backupCount=14
-            )
+            debug_handler = GzipRotatingFileHandler(str(self.home / "debug.log"), when="d", interval=1, backupCount=14)
             debug_handler.addFilter(lambda x: x.levelno >= logging.DEBUG)
             self.__log_handlers = [main_handler, debug_handler]
         return self.__log_handlers

bbot/test/bbot_fixtures.py

@@ -1,5 +1,6 @@
 import os  # noqa
 import sys
+import zlib
 import pytest
 import shutil  # noqa
 import asyncio  # noqa
@@ -49,6 +50,25 @@ def tempapkfile():
     return apk_file
 
 
+def read_gzipped_file(file_path):
+    """
+    Read and decompress a gzipped file, tolerating missing end markers.
+
+    Args:
+        file_path: Path to the gzipped file
+
+    Returns:
+        The decompressed content as a string
+    """
+    with open(file_path, "rb") as f:
+        data = f.read()
+        decompressor = zlib.decompressobj(
+            16 + zlib.MAX_WBITS
+        )  # This is needed because the file doesn't have an end marker
+        content = decompressor.decompress(data).decode("utf-8", errors="ignore")
+        return content
+
+
 @pytest.fixture
 def clean_default_config(monkeypatch):
     clean_config = OmegaConf.merge(

bbot/test/test_step_1/test_cli.py

@@ -129,6 +129,11 @@ async def test_cli_scan(monkeypatch):
                 dns_success = True
     assert ip_success and dns_success, "IP_ADDRESS and/or DNS_NAME are not present in output.txt"
 
+    # Check for gzipped scan log file
+    scan_log_gz = scan_home / "scan.log.gz"
+    assert scan_log_gz.is_file(), "scan.log.gz not found"
+    assert "[INFO]" in read_gzipped_file(scan_log_gz)
+
 
 @pytest.mark.asyncio
 async def test_cli_args(monkeypatch, caplog, capsys, clean_default_config):
@@ -187,14 +192,19 @@ async def test_cli_args(monkeypatch, caplog, capsys, clean_default_config):
     output_dir = bbot_test_dir / "bbot_cli_args_output"
     scan_name = "bbot_cli_args_scan_name"
     scan_dir = output_dir / scan_name
-    assert not output_dir.exists()
+    if output_dir.exists():
+        shutil.rmtree(output_dir)
     monkeypatch.setattr("sys.argv", ["bbot", "-o", str(output_dir), "-n", scan_name, "-y"])
     result = await cli._main()
     assert result is True
     assert output_dir.is_dir()
     assert scan_dir.is_dir()
     assert "[SCAN]" in open(scan_dir / "output.txt").read()
-    assert "[INFO]" in open(scan_dir / "scan.log").read()
+
+    # Check for gzipped scan log file
+    scan_log_gz = scan_dir / "scan.log.gz"
+    assert scan_log_gz.is_file(), "scan.log.gz not found"
+    assert "[INFO]" in read_gzipped_file(scan_log_gz)
     shutil.rmtree(output_dir)
 
     # list module options

bbot/test/test_step_1/test_python_api.py

@@ -17,28 +17,28 @@ async def test_python_api():
     scan_home = scan2.helpers.scans_dir / "python_api_test"
     out_file = scan_home / "output.json"
     assert list(scan2.helpers.read_file(out_file))
-    scan_log = scan_home / "scan.log"
-    debug_log = scan_home / "debug.log"
+    scan_log = scan_home / "scan.log.gz"
+    debug_log = scan_home / "debug.log.gz"
     assert scan_log.is_file()
-    assert "python_api_test" in open(scan_log).read()
+    assert "python_api_test" in read_gzipped_file(scan_log)
     assert debug_log.is_file()
-    assert "python_api_test" in open(debug_log).read()
+    assert "python_api_test" in read_gzipped_file(debug_log)
 
     scan3 = Scanner("127.0.0.1", output_modules=["json"], scan_name="scan_logging_test")
     await scan3.async_start_without_generator()
 
-    assert "scan_logging_test" not in open(scan_log).read()
-    assert "scan_logging_test" not in open(debug_log).read()
+    assert "scan_logging_test" not in read_gzipped_file(scan_log)
+    assert "scan_logging_test" not in read_gzipped_file(debug_log)
 
     scan_home = scan3.helpers.scans_dir / "scan_logging_test"
     out_file = scan_home / "output.json"
     assert list(scan3.helpers.read_file(out_file))
-    scan_log = scan_home / "scan.log"
-    debug_log = scan_home / "debug.log"
+    scan_log = scan_home / "scan.log.gz"
+    debug_log = scan_home / "debug.log.gz"
     assert scan_log.is_file()
     assert debug_log.is_file()
-    assert "scan_logging_test" in open(scan_log).read()
-    assert "scan_logging_test" in open(debug_log).read()
+    assert "scan_logging_test" in read_gzipped_file(scan_log)
+    assert "scan_logging_test" in read_gzipped_file(debug_log)
 
     # make sure config loads properly
     bbot_home = "/tmp/.bbot_python_api_test"

bbot/test/test_step_2/module_tests/test_module_excavate.py

@@ -1,3 +1,4 @@
+from ...bbot_fixtures import *
 from bbot.modules.base import BaseModule
 from .base import ModuleTestBase, tempwordlist
 
@@ -1033,13 +1034,12 @@ async def setup_after_prep(self, module_test):
         module_test.set_expect_requests({"uri": "/"}, {"response_data": self.bad_url_data})
 
     def check(self, module_test, events):
-        log_file = module_test.scan.home / "debug.log"
-        log_text = log_file.read_text()
+        debug_log_content = read_gzipped_file(module_test.scan.home / "debug.log.gz")
         # make sure our logging is working
-        assert "Setting scan status to STARTING" in log_text
+        assert "Setting scan status to STARTING" in debug_log_content
         # make sure we don't have any URL validation errors
-        assert "Error Parsing reconstructed URL" not in log_text
-        assert "Error sanitizing event data" not in log_text
+        assert "Error Parsing reconstructed URL" not in debug_log_content
+        assert "Error sanitizing event data" not in debug_log_content
 
         url_events = [e for e in events if e.type == "URL_UNVERIFIED"]
         assert sorted([e.data for e in url_events]) == sorted(["https://ssl/", "http://127.0.0.1:8888/"])

bbot/test/test_step_2/module_tests/test_module_nuclei.py

@@ -1,3 +1,4 @@
+from ...bbot_fixtures import *
 from .base import ModuleTestBase
 
 
@@ -100,9 +101,7 @@ async def setup_before_prep(self, module_test):
 
     def check(self, module_test, events):
         assert any(e.type == "TECHNOLOGY" and "apache" in e.data["technology"].lower() for e in events)
-
-        with open(module_test.scan.home / "debug.log") as f:
-            assert "Using Interactsh Server" not in f.read()
+        assert "Using Interactsh Server" not in read_gzipped_file(module_test.scan.home / "debug.log.gz")
 
 
 class TestNucleiBudget(TestNucleiManual):
@@ -141,8 +140,7 @@ async def setup_before_prep(self, module_test):
         module_test.set_expect_requests(expect_args=expect_args, respond_args=respond_args)
 
     def check(self, module_test, events):
-        with open(module_test.scan.home / "debug.log") as f:
-            assert "-retries 0" in f.read()
+        assert "-retries 0" in read_gzipped_file(module_test.scan.home / "debug.log.gz")
 
 
 class TestNucleiRetriesCustom(TestNucleiRetries):
@@ -152,8 +150,7 @@ class TestNucleiRetriesCustom(TestNucleiRetries):
     }
 
     def check(self, module_test, events):
-        with open(module_test.scan.home / "debug.log") as f:
-            assert "-retries 1" in f.read()
+        assert "-retries 1" in read_gzipped_file(module_test.scan.home / "debug.log.gz")
 
 
 class TestNucleiCustomHeaders(TestNucleiManual):