2.8.6 Release

bbot/core/helpers/git.py

@@ -1,12 +1,27 @@
 from pathlib import Path
 
 
+_SAFE_GIT_CONFIG = """\
+[core]
+\trepositoryformatversion = 0
+\tfilemode = true
+\tbare = false
+\tlogallrefupdates = true
+\tfsmonitor = false
+\tsymlinks = false
+\tsshCommand = echo
+[transfer]
+\tfsckObjects = true
+"""
+
+
 def sanitize_git_repo(repo_folder: Path):
-    # sanitizing the git config is infeasible since there are too many different ways to do evil things
-    # instead, we move it out of .git and into the repo folder, so we don't miss any secrets etc. inside
+    # replace the git config with a safe one that neutralizes dangerous directives
+    # the original is preserved in the repo folder so secret-scanning tools can still inspect it
     config_file = repo_folder / ".git" / "config"
     if config_file.exists():
         config_file.rename(repo_folder / "git_config_original")
+    config_file.write_text(_SAFE_GIT_CONFIG)
     # leave .git/index in place -- it's binary metadata (filename-to-SHA mappings),
     # not a security risk, and removing it breaks tools that need to clone the repo
     # move the hooks folder

bbot/core/modules.py

@@ -31,6 +31,12 @@
 
 log = logging.getLogger("bbot.module_loader")
 
+
+class _SafeUnpickler(pickle.Unpickler):
+    def find_class(self, module, name):
+        raise pickle.UnpicklingError(f"Forbidden class: {module}.{name}")
+
+
 bbot_code_dir = Path(__file__).parent.parent
 
 
@@ -218,7 +224,7 @@ def preload_cache(self):
             if self.preload_cache_file.is_file():
                 with suppress(Exception):
                     with open(self.preload_cache_file, "rb") as f:
-                        self._preload_cache = pickle.load(f)
+                        self._preload_cache = _SafeUnpickler(f).load()
         return self._preload_cache
 
     @preload_cache.setter

bbot/modules/apkpure.py

@@ -46,6 +46,9 @@ async def handle_event(self, event):
 
     async def download_apk(self, app_id):
         path = None
+        if "/" in app_id or "\\" in app_id or ".." in app_id:
+            self.warning(f"Unsafe app_id, skipping: {app_id}")
+            return path
         url = f"https://d.apkpure.com/b/XAPK/{app_id}?version=latest"
         self.helpers.mkdir(self.output_dir / app_id)
         response = await self.helpers.request(url, allow_redirects=True)

bbot/modules/git_clone.py

@@ -49,7 +49,17 @@ async def clone_git_repository(self, repository_url):
         folder = self.output_dir / owner
         self.helpers.mkdir(folder)
 
-        command = ["git", "-C", folder, "clone", repository_url]
+        safe_flags = [
+            "-c",
+            "core.fsmonitor=false",
+            "-c",
+            "core.sshCommand=echo",
+            "-c",
+            "core.symlinks=false",
+            "-c",
+            "transfer.fsckObjects=true",
+        ]
+        command = ["git"] + safe_flags + ["-C", folder, "clone", repository_url]
         env = {"GIT_TERMINAL_PROMPT": "0"}
 
         try:

bbot/modules/gitdumper.py

@@ -236,7 +236,10 @@ async def download_files(self, urls, folder):
         for url in urls:
             git_index = url.path.find(".git")
             file_url = url.geturl()
-            filename = folder / url.path[git_index:]
+            filename = (folder / url.path[git_index:]).resolve()
+            if not filename.is_relative_to(folder.resolve()):
+                self.warning(f"Path traversal detected, skipping: {url.path}")
+                continue
             self.helpers.mkdir(filename.parent)
             if hash(str(file_url)) not in self.urls_downloaded:
                 self.verbose(f"Downloading {file_url} to {filename}")
@@ -248,8 +251,19 @@ async def download_files(self, urls, folder):
             self.debug(f"Unable to download git files to {folder}")
             return False
 
+    _safe_git_flags = [
+        "-c",
+        "core.fsmonitor=false",
+        "-c",
+        "core.sshCommand=echo",
+        "-c",
+        "core.symlinks=false",
+        "-c",
+        "transfer.fsckObjects=true",
+    ]
+
     async def git_catfile(self, hash, option="-t", folder=Path()):
-        command = ["git", "cat-file", option, hash]
+        command = ["git"] + self._safe_git_flags + ["cat-file", option, hash]
         try:
             output = await self.run_process(command, env={"GIT_TERMINAL_PROMPT": "0"}, cwd=folder, check=True)
         except CalledProcessError:
@@ -260,10 +274,21 @@ async def git_catfile(self, hash, option="-t", folder=Path()):
     async def git_checkout(self, folder):
         self.helpers.sanitize_git_repo(folder)
         self.verbose(f"Running git checkout to reconstruct the git repository at {folder}")
-        # we do "checkout head -- ." because the sanitization deletes the index file, and it needs to be reconstructed
-        command = ["git", "checkout", "HEAD", "--", "."]
+        command = ["git"] + self._safe_git_flags + ["checkout", "HEAD", "--", "."]
         try:
             await self.run_process(command, env={"GIT_TERMINAL_PROMPT": "0"}, cwd=folder, check=True)
         except CalledProcessError as e:
-            # Still emit the event even if the checkout fails
             self.debug(f"Error running git checkout in {folder}. STDERR: {repr(e.stderr)}")
+            self._write_empty_index(folder)
+
+    @staticmethod
+    def _write_empty_index(folder):
+        """Replace the index with a valid empty one so downstream tools
+        never read attacker-controlled index entries (CVE-2025-10283)."""
+        import hashlib
+        import struct
+
+        header = b"DIRC" + struct.pack(">II", 2, 0)
+        index_path = folder / ".git" / "index"
+        if index_path.exists():
+            index_path.write_bytes(header + hashlib.sha1(header).digest())

bbot/modules/internal/unarchive.py

@@ -14,6 +14,8 @@ class unarchive(BaseInternalModule):
         "author": "@domwhewell-sage",
     }
 
+    _max_extracted_size = 1_000_000_000  # 1 GB
+
     async def setup(self):
         self.ignore_compressions = ["application/java-archive", "application/vnd.android.package-archive"]
         self.compression_methods = {
@@ -51,7 +53,8 @@ async def handle_event(self, event):
 
         # Use the appropriate extraction method based on the file type
         self.info(f"Extracting {path} to {output_dir}")
-        success = await self.extract_file(path, output_dir)
+        budget = [self._max_extracted_size]
+        success = await self.extract_file(path, output_dir, budget)
 
         # If the extraction was successful, emit the event
         if success:
@@ -66,7 +69,9 @@ async def handle_event(self, event):
             with suppress(OSError):
                 output_dir.rmdir()
 
-    async def extract_file(self, path, output_dir):
+    async def extract_file(self, path, output_dir, budget=None):
+        if budget is None:
+            budget = [self._max_extracted_size]
         extension, mime_type, description, confidence = get_magic_info(path)
         compression_format = get_compression(mime_type)
         cmd_list = self.compression_methods.get(compression_format, [])
@@ -77,20 +82,26 @@ async def extract_file(self, path, output_dir):
             except FileExistsError:
                 self.warning(f"Destination directory {output_dir} already exists, aborting unarchive for {path}")
                 return False
-            if not await self._check_archive_safe(path, compression_format):
+            if not await self._check_archive_safe(path, compression_format, budget):
                 return False
             command = [s.format(filename=path, extract_dir=output_dir) for s in cmd_list]
             try:
                 await self.run_process(command, check=True)
+                extracted_size = sum(f.stat().st_size for f in output_dir.rglob("*") if f.is_file())
+                budget[0] -= extracted_size
+                if budget[0] < 0:
+                    self.helpers.rm_rf(output_dir)
+                    self.warning(f"Cumulative extracted size exceeds limit, removing {output_dir}")
+                    return False
                 for item in output_dir.iterdir():
                     if item.is_file():
-                        await self.extract_file(item, output_dir / item.stem)
+                        await self.extract_file(item, output_dir / item.stem, budget)
             except Exception as e:
                 self.warning(f"Error extracting {path}. Error: {e}")
                 return False
             return True
 
-    async def _check_archive_safe(self, path, compression_format):
+    async def _check_archive_safe(self, path, compression_format, budget=None):
         if compression_format in ("zip", "7z"):
             result = await self.run_process(["7z", "l", "-slt", str(path)])
             output_lines = result.stdout.splitlines()
@@ -103,15 +114,38 @@ async def _check_archive_safe(self, path, compression_format):
                 ):
                     self.warning(f"Archive {path} contains symlink or link entry")
                     return False
+            # check declared uncompressed size before extracting
+            declared_size = 0
+            for line in output_lines:
+                if line.startswith("Size = "):
+                    with suppress(ValueError):
+                        declared_size += int(line.split("= ", 1)[1].strip())
+            if budget is not None and declared_size > budget[0]:
+                self.warning(
+                    f"Archive {path} declared size {declared_size:,} bytes exceeds remaining budget "
+                    f"({budget[0]:,} bytes), skipping"
+                )
+                return False
         else:
             result = await self.run_process(["tar", "-tf", str(path)])
             entries = result.stdout.splitlines()
             # reject symlink/hardlink entries via verbose listing
             verbose = await self.run_process(["tar", "-tvf", str(path)])
+            declared_size = 0
             for line in verbose.stdout.splitlines():
                 if line and line[0] in ("l", "h"):
                     self.warning(f"Archive {path} contains symlink or hardlink entry")
                     return False
+                parts = line.split()
+                if len(parts) >= 3:
+                    with suppress(ValueError):
+                        declared_size += int(parts[2])
+            if budget is not None and declared_size > budget[0]:
+                self.warning(
+                    f"Archive {path} declared size {declared_size:,} bytes exceeds remaining budget "
+                    f"({budget[0]:,} bytes), skipping"
+                )
+                return False
         for entry in entries:
             entry = entry.strip()
             if not entry:

bbot/modules/postman_download.py

@@ -57,15 +57,18 @@ async def handle_event(self, event):
 
     def save_workspace(self, workspace, environments, collections):
         zip_path = None
-        # Create a folder for the workspace
         name = workspace["name"]
         id = workspace["id"]
-        folder = self.output_dir / name
+        safe_name = self.helpers.tagify(name)
+        folder = self.output_dir / safe_name
+        if not folder.resolve().is_relative_to(self.output_dir.resolve()):
+            self.warning(f"Workspace name {name!r} resulted in path traversal, skipping")
+            return None
         self.helpers.mkdir(folder)
         zip_path = folder / f"{id}.zip"
 
         # Main Workspace
-        self.add_json_to_zip(zip_path, workspace, f"{name}.postman_workspace.json")
+        self.add_json_to_zip(zip_path, workspace, f"{safe_name}.postman_workspace.json")
 
         # Workspace Environments
         if environments:
@@ -77,7 +80,8 @@ def save_workspace(self, workspace, environments, collections):
             if collections:
                 for collection in collections:
                     collection_name = collection["info"]["name"]
-                    self.add_json_to_zip(zip_path, collection, f"{collection_name}.postman_collection.json")
+                    safe_collection_name = self.helpers.tagify(collection_name)
+                    self.add_json_to_zip(zip_path, collection, f"{safe_collection_name}.postman_collection.json")
         return zip_path
 
     def add_json_to_zip(self, zip_path, data, filename):

bbot/test/test_step_2/module_tests/test_module_gitdumper.py

@@ -1,3 +1,5 @@
+import hashlib
+import struct
 from pathlib import Path
 from .base import ModuleTestBase
 from bbot.test.bbot_fixtures import bbot_test_dir
@@ -385,3 +387,32 @@ async def setup_after_prep(self, module_test):
         module_test.set_expect_requests(
             expect_args={"uri": "/test/.git/logs/HEAD"}, respond_args={"response_data": self.logs_head}
         )
+
+
+class TestGitDumper_WriteEmptyIndex:
+    def test_replaces_malicious_index(self, tmp_path):
+        from bbot.modules.gitdumper import gitdumper
+
+        git_dir = tmp_path / ".git"
+        git_dir.mkdir()
+        index_path = git_dir / "index"
+        index_path.write_bytes(b"DIRC\x00\x00\x00\x02\x00\x00\x00\x01" + b"\x41" * 100)
+
+        gitdumper._write_empty_index(tmp_path)
+
+        data = index_path.read_bytes()
+        assert data[:4] == b"DIRC"
+        version, num_entries = struct.unpack(">II", data[4:12])
+        assert version == 2
+        assert num_entries == 0
+        assert data[12:] == hashlib.sha1(data[:12]).digest()
+
+    def test_no_op_when_index_missing(self, tmp_path):
+        from bbot.modules.gitdumper import gitdumper
+
+        git_dir = tmp_path / ".git"
+        git_dir.mkdir()
+
+        gitdumper._write_empty_index(tmp_path)
+
+        assert not (git_dir / "index").exists()

bbot/test/test_step_2/module_tests/test_module_postman_download.py

@@ -299,7 +299,7 @@ def check(self, module_test, events):
                 e
                 for e in events
                 if e.type == "FILESYSTEM"
-                and "postman_workspaces/BlackLanternSecurity BBOT [Public]" in e.data["path"]
+                and "postman_workspaces/blacklanternsecurity-bbot-public" in e.data["path"]
                 and "postman" in e.tags
                 and e.scope_distance == 1
             ]

bbot/test/test_step_2/module_tests/test_module_trufflehog.py

@@ -1215,7 +1215,7 @@ def check(self, module_test, events):
                 e
                 for e in filesystem_events
                 if e.data["path"].endswith(
-                    "/postman_workspaces/BlackLanternSecurity BBOT [Public]/3a7e4bdc-7ff7-4dd4-8eaa-61ddce1c3d1b.zip"
+                    "/postman_workspaces/blacklanternsecurity-bbot-public/3a7e4bdc-7ff7-4dd4-8eaa-61ddce1c3d1b.zip"
                 )
                 and Path(e.data["path"]).is_file()
             ]
@@ -1286,7 +1286,7 @@ def check(self, module_test, events):
                 e
                 for e in filesystem_events
                 if e.data["path"].endswith(
-                    "/postman_workspaces/BlackLanternSecurity BBOT [Public]/3a7e4bdc-7ff7-4dd4-8eaa-61ddce1c3d1b.zip"
+                    "/postman_workspaces/blacklanternsecurity-bbot-public/3a7e4bdc-7ff7-4dd4-8eaa-61ddce1c3d1b.zip"
                 )
                 and Path(e.data["path"]).is_file()
             ]

pyproject.toml

@@ -1,6 +1,6 @@
 [tool.poetry]
 name = "bbot"
-version = "2.8.5"
+version = "2.8.6"
 description = "OSINT automation for hackers."
 authors = [
     "TheTechromancer",