Merge pull request #2810 from blacklanternsecurity/untangle-cli-args

untangle cli arg issues

Author: liquidsec

bbot/cli.py

@@ -7,7 +7,7 @@
 from bbot.errors import *
 from bbot import __version__
 from bbot.logger import log_to_stderr
-from bbot.core.helpers.misc import chain_lists, rm_rf
+from bbot.core.helpers.misc import chain_lists
 
 
 if multiprocessing.current_process().name == "MainProcess":
@@ -34,13 +34,8 @@ async def _main():
     import traceback
     from contextlib import suppress
 
-    # fix tee buffering (only if on real TTY)
-    if hasattr(sys.stdout, "reconfigure"):
-        try:
-            if sys.stdout.isatty():
-                sys.stdout.reconfigure(line_buffering=True)
-        except Exception:
-            pass
+    # fix tee buffering
+    sys.stdout.reconfigure(line_buffering=True)
 
     log = logging.getLogger("bbot.cli")
 
@@ -61,6 +56,10 @@ async def _main():
             return
         # ensure arguments (-c config options etc.) are valid
         options = preset.args.parsed
+        # apply CLI log level options (e.g. --debug/--verbose/--silent) to the
+        # global core logger even for CLI-only commands (like --install-all-deps)
+        # that don't construct a full Scanner.
+        preset.apply_log_level(apply_core=True)
 
         # print help if no arguments
         if len(sys.argv) == 1:
@@ -95,7 +94,8 @@ async def _main():
                 preset._default_output_modules = options.output_modules
                 preset._default_internal_modules = []
 
-            await preset.bake()
+            # Bake a temporary copy of the preset so that flags correctly enable their associated modules before listing them
+            preset = await preset.bake()
 
             # --list-modules
             if options.list_modules:
@@ -149,61 +149,68 @@ async def _main():
                 print(row)
             return
 
-        try:
-            scan = Scanner(preset=preset)
-        except (PresetAbortError, ValidationError) as e:
-            log.warning(str(e))
+        baked_preset = await preset.bake()
+
+        # --current-preset / --current-preset-full
+        if options.current_preset or options.current_preset_full:
+            # Ensure we always have a human-friendly description. Prefer an
+            # explicit scan_name if present, otherwise fall back to the
+            # preset name (e.g. "bbot_cli_main").
+            if not baked_preset.description:
+                if baked_preset.scan_name:
+                    baked_preset.description = str(baked_preset.scan_name)
+                elif baked_preset.name:
+                    baked_preset.description = str(baked_preset.name)
+            if options.current_preset_full:
+                print(baked_preset.to_yaml(full_config=True))
+            else:
+                print(baked_preset.to_yaml())
+            sys.exit(0)
             return
 
-        await scan._prep()
-
+        # deadly modules (no scan required yet)
         deadly_modules = [
-            m for m in scan.preset.scan_modules if "deadly" in preset.preloaded_module(m).get("flags", [])
+            m for m in baked_preset.scan_modules if "deadly" in baked_preset.preloaded_module(m).get("flags", [])
         ]
         if deadly_modules and not options.allow_deadly:
             log.hugewarning(f"You enabled the following deadly modules: {','.join(deadly_modules)}")
             log.hugewarning("Deadly modules are highly intrusive")
             log.hugewarning("Please specify --allow-deadly to continue")
             return False
 
-        # --current-preset
-        if options.current_preset:
-            print(scan.preset.to_yaml())
-            sys.exit(0)
-            return
-
-        # --current-preset-full
-        if options.current_preset_full:
-            print(scan.preset.to_yaml(full_config=True))
-            sys.exit(0)
+        try:
+            scan = Scanner(preset=baked_preset)
+        except (PresetAbortError, ValidationError) as e:
+            log.warning(str(e))
             return
 
         # --install-all-deps
         if options.install_all_deps:
+            # create a throwaway Scanner solely so that Preset.bake(scan) can perform find_and_replace() on all module configs so that placeholders like "#{BBOT_TOOLS}" are resolved before running Ansible tasks.
+            from bbot.scanner import Scanner as _ScannerForDeps
+
             preloaded_modules = preset.module_loader.preloaded()
-            scan_modules = [k for k, v in preloaded_modules.items() if str(v.get("type", "")) == "scan"]
-            output_modules = [k for k, v in preloaded_modules.items() if str(v.get("type", "")) == "output"]
-            log.verbose("Creating dummy scan with all modules + output modules for deps installation")
-            dummy_scan = Scanner(preset=preset, modules=scan_modules, output_modules=output_modules)
-            dummy_scan.helpers.depsinstaller.force_deps = True
+            modules_for_deps = [
+                k for k, v in preloaded_modules.items() if str(v.get("type", "")) in ("scan", "output")
+            ]
+
+            # dummy scan used only for environment preparation
+            dummy_scan = _ScannerForDeps(preset=preset)
+            await dummy_scan._unbaked_preset.bake(dummy_scan)
+
+            helper = dummy_scan.helpers
             log.info("Installing module dependencies")
-            await dummy_scan.load_modules()
-            log.verbose("Running module setups")
-            succeeded, hard_failed, soft_failed = await dummy_scan.setup_modules(deps_only=True)
-            # remove any leftovers from the dummy scan
-            rm_rf(dummy_scan.home, ignore_errors=True)
-            rm_rf(dummy_scan.temp_dir, ignore_errors=True)
+            succeeded, failed = await helper.depsinstaller.install(*modules_for_deps)
             if succeeded:
                 log.success(
                     f"Successfully installed dependencies for {len(succeeded):,} modules: {','.join(succeeded)}"
                 )
-            if soft_failed or hard_failed:
-                failed = soft_failed + hard_failed
+            if failed:
                 log.warning(f"Failed to install dependencies for {len(failed):,} modules: {', '.join(failed)}")
                 return False
             return True
 
-        scan_name = str(scan.name)
+        await scan._prep()
 
         log.verbose("")
         log.verbose("### MODULES ENABLED ###")
@@ -213,17 +220,18 @@ async def _main():
 
         scan.helpers.word_cloud.load()
 
+        scan_name = str(scan.name)
+
         if not options.dry_run:
             log.trace(f"Command: {' '.join(sys.argv)}")
 
+            # In some environments (e.g. tests) stdin may be closed or not support isatty(). Treat those cases as non-interactive.
             try:
-                is_tty = (
-                    hasattr(sys.stdin, "isatty") and not getattr(sys.stdin, "closed", False) and sys.stdin.isatty()
-                )
-            except Exception:
-                is_tty = False
+                stdin_is_tty = sys.stdin.isatty()
+            except (ValueError, io.UnsupportedOperation):
+                stdin_is_tty = False
 
-            if is_tty:
+            if stdin_is_tty:
                 # warn if any targets belong directly to a cloud provider
                 if not scan.preset.strict_scope:
                     for event in scan.target.seeds.event_seeds:

bbot/core/helpers/web/web.py

@@ -56,10 +56,7 @@ def __init__(self, parent_helper):
         self.target = self.preset.target
         self.ssl_verify = self.config.get("ssl_verify", False)
         engine_debug = self.config.get("engine", {}).get("debug", False)
-        super().__init__(
-            server_kwargs={"config": self.config, "target": self.parent_helper.preset.target},
-            debug=engine_debug,
-        )
+        super().__init__(server_kwargs={"config": self.config, "target": self.target}, debug=engine_debug)
 
     def AsyncClient(self, *args, **kwargs):
         # cache by retries to prevent unwanted accumulation of clients

bbot/scanner/preset/args.py

@@ -368,14 +368,15 @@ def create_parser(self, *args, **kwargs):
         deps = p.add_argument_group(
             title="Module dependencies", description="Control how modules install their dependencies"
         )
+        # Behavior flags are mutually exclusive with each other. But need to be able to be combined with --install-all-deps.
         g2 = deps.add_mutually_exclusive_group()
         g2.add_argument("--no-deps", action="store_true", help="Don't install module dependencies")
         g2.add_argument("--force-deps", action="store_true", help="Force install all module dependencies")
         g2.add_argument("--retry-deps", action="store_true", help="Try again to install failed module dependencies")
         g2.add_argument(
             "--ignore-failed-deps", action="store_true", help="Run modules even if they have failed dependencies"
         )
-        g2.add_argument("--install-all-deps", action="store_true", help="Install dependencies for all modules")
+        deps.add_argument("--install-all-deps", action="store_true", help="Install dependencies for all modules")
 
         misc = p.add_argument_group(title="Misc")
         misc.add_argument("--version", action="store_true", help="show BBOT version and exit")

bbot/scanner/preset/preset.py

@@ -600,11 +600,34 @@ def apply_log_level(self, apply_core=False):
     @property
     def helpers(self):
         if self._helpers is None:
+            # Ensure we have at least a minimal target object before any helper (especially web helpers) is constructed.
+
+            self._ensure_minimal_target()
             from bbot.core.helpers.helper import ConfigAwareHelper
 
             self._helpers = ConfigAwareHelper(preset=self)
         return self._helpers
 
+    def _ensure_minimal_target(self):
+        """
+        Lazily construct a minimal BBOTTarget from the current seeds / whitelist / blacklist if one does not already exist.
+
+        This is intentionally lighter-weight than the full async target
+        preparation performed in `bake()` (which also calls
+        `target.generate_children()`).
+        """
+        if self._target is not None:
+            return
+
+        from bbot.scanner.target import BBOTTarget
+
+        self._target = BBOTTarget(
+            *list(self._seeds),
+            whitelist=self._whitelist,  # modify this after scope rework branch is merged into dev
+            blacklist=self._blacklist,
+            strict_scope=self.strict_scope,
+        )
+
     @property
     def module_loader(self):
         self.environ

bbot/scanner/scanner.py

@@ -158,6 +158,9 @@ def __init__(
         self.modules = OrderedDict({})
         self.dummy_modules = {}
         self.preset = None
+        # initial status before `_prep()` runs
+        self._status = "NOT_STARTED"
+        self._status_code = self._status_codes[self._status]
 
     async def _prep(self):
         """
@@ -216,26 +219,26 @@ async def _prep(self):
         self.scope_report_distance = int(self.scope_config.get("report_distance", 1))
 
         # web config
-        self.web_config = self.config.get("web", {})
-        self.web_spider_distance = self.web_config.get("spider_distance", 0)
-        self.web_spider_depth = self.web_config.get("spider_depth", 1)
-        self.web_spider_links_per_page = self.web_config.get("spider_links_per_page", 20)
-        max_redirects = self.web_config.get("http_max_redirects", 5)
+        web_config = self.config.get("web", {})
+        self.web_spider_distance = web_config.get("spider_distance", 0)
+        self.web_spider_depth = web_config.get("spider_depth", 1)
+        self.web_spider_links_per_page = web_config.get("spider_links_per_page", 20)
+        max_redirects = web_config.get("http_max_redirects", 5)
         self.web_max_redirects = max(max_redirects, self.web_spider_distance)
-        self.http_proxy = self.web_config.get("http_proxy", "")
-        self.http_timeout = self.web_config.get("http_timeout", 10)
-        self.httpx_timeout = self.web_config.get("httpx_timeout", 5)
-        self.http_retries = self.web_config.get("http_retries", 1)
-        self.httpx_retries = self.web_config.get("httpx_retries", 1)
-        self.useragent = self.web_config.get("user_agent", "BBOT")
+        self.http_proxy = web_config.get("http_proxy", "")
+        self.http_timeout = web_config.get("http_timeout", 10)
+        self.httpx_timeout = web_config.get("httpx_timeout", 5)
+        self.http_retries = web_config.get("http_retries", 1)
+        self.httpx_retries = web_config.get("httpx_retries", 1)
+        self.useragent = web_config.get("user_agent", "BBOT")
         # custom HTTP headers warning
-        self.custom_http_headers = self.web_config.get("http_headers", {})
+        self.custom_http_headers = web_config.get("http_headers", {})
         if self.custom_http_headers:
             self.warning(
                 "You have enabled custom HTTP headers. These will be attached to all in-scope requests and all requests made by httpx."
             )
         # custom HTTP cookies warning
-        self.custom_http_cookies = self.web_config.get("http_cookies", {})
+        self.custom_http_cookies = web_config.get("http_cookies", {})
         if self.custom_http_cookies:
             self.warning(
                 "You have enabled custom HTTP cookies. These will be attached to all in-scope requests and all requests made by httpx."
@@ -562,8 +565,18 @@ async def load_modules(self):
             After all modules are loaded, they are sorted by `_priority` and stored in the `modules` dictionary.
         """
         if not self._modules_loaded:
+            # If the preset hasn't been baked yet but modules have been
+            # manually attached (e.g. in tests), skip the automatic loading
+            # pipeline and operate only on the existing modules.
+            if self.preset is None:
+                if not self.modules:
+                    self.warning("No modules to load")
+                self._modules_loaded = True
+                return
+
             if not self.preset.modules:
                 self.warning("No modules to load")
+                self._modules_loaded = True
                 return
 
             if not self.preset.scan_modules:
@@ -897,9 +910,15 @@ async def _cleanup(self):
             # clean up modules
             for mod in self.modules.values():
                 await mod._cleanup()
-            with contextlib.suppress(Exception):
-                self.home.rmdir()
-            self.helpers.rm_rf(self.temp_dir, ignore_errors=True)
+            # In some test paths, `_prep()` is never called, so `home` and
+            # `temp_dir` may not exist. Treat those as best-effort cleanups.
+            home = getattr(self, "home", None)
+            if home is not None:
+                with contextlib.suppress(Exception):
+                    home.rmdir()
+            temp_dir = getattr(self, "temp_dir", None)
+            if temp_dir is not None:
+                self.helpers.rm_rf(temp_dir, ignore_errors=True)
             self.helpers.clean_old_scans()
 
     def in_scope(self, *args, **kwargs):
@@ -913,11 +932,29 @@ def blacklisted(self, *args, **kwargs):
 
     @property
     def core(self):
-        return self.preset.core
+        # Before `_prep()` runs, fall back to the unbaked preset's core so that basic configuration is still available (during module construction in tests)
+        if self.preset is not None:
+            return self.preset.core
+        return self._unbaked_preset.core
 
     @property
     def config(self):
-        return self.preset.core.config
+        # Allow access to the scan config even before `_prep()` by falling back to the unbaked preset's core config.
+        if self.preset is not None:
+            return self.preset.core.config
+        return self._unbaked_preset.core.config
+
+    @property
+    def web_config(self):
+        """
+        Web-related configuration for the scan.
+
+        Exposed as a property so it is available even before `_prep()` runs,
+        falling back to the underlying config's `web` section. During `_prep()`
+        an instance attribute of the same name is assigned, which will then
+        override this property for the remainder of the scan lifetime.
+        """
+        return self.config.get("web", {})
 
     @property
     def target(self):
@@ -937,7 +974,13 @@ def blacklist(self):
 
     @property
     def helpers(self):
-        return self.preset.helpers
+        # Before `_prep()` runs, `self.preset` is None. In those cases,
+        # fall back to the unbaked preset's helpers so that CLI utilities
+        # (e.g. depsinstaller) and other lightweight helper functionality
+        # remain available without requiring a full scan prep.
+        if self.preset is not None:
+            return self.preset.helpers
+        return self._unbaked_preset.helpers
 
     @property
     def force_start(self):
@@ -986,12 +1029,15 @@ def status(self, status):
                 if status != self._status:
                     self._status = status
                     self._status_code = self._status_codes[status]
-                    self.dispatcher_tasks.append(
-                        asyncio.create_task(
-                            self.dispatcher.catch(self.dispatcher.on_status, self._status, self.id),
-                            name=f"{self.name}.dispatcher.on_status({status})",
+                    # During early initialization (or in certain tests),`dispatcher` may not be set yet. In that case we just update the status without scheduling dispatcher tasks
+                    dispatcher = getattr(self, "dispatcher", None)
+                    if dispatcher is not None:
+                        self.dispatcher_tasks.append(
+                            asyncio.create_task(
+                                dispatcher.catch(self.dispatcher.on_status, self._status, self.id),
+                                name=f"{self.name}.dispatcher.on_status({status})",
+                            )
                         )
-                    )
                 else:
                     self.debug(f'Scan status is already "{status}"')
         else: