Dev -> Stable 2.4.1

bbot/cli.py

@@ -78,7 +78,7 @@ async def _main():
             return
 
         # if we're listing modules or their options
-        if options.list_modules or options.list_output_modules or options.list_module_options:
+        if options.list_modules or options.list_output_modules or options.list_module_options or options.module_help:
             # if no modules or flags are specified, enable everything
             if not (options.modules or options.output_modules or options.flags):
                 for module, preloaded in preset.module_loader.preloaded().items():
@@ -119,6 +119,20 @@ async def _main():
                     print(row)
                 return
 
+            # --module-help
+            if options.module_help:
+                module_name = options.module_help
+                all_modules = list(preset.module_loader.preloaded())
+                if module_name not in all_modules:
+                    log.hugewarning(f'Module "{module_name}" not found')
+                    return
+
+                # Load the module class
+                loaded_modules = preset.module_loader.load_modules([module_name])
+                module_name, module_class = next(iter(loaded_modules.items()))
+                print(module_class.help_text())
+                return
+
         # --list-flags
         if options.list_flags:
             flags = preset.flags if preset.flags else None

bbot/modules/base.py

@@ -1595,6 +1595,48 @@ def critical(self, *args, trace=True, **kwargs):
         if trace:
             self.trace()
 
+    @classmethod
+    def help_text(self):
+        """
+        Returns a string containing help text for the module.
+        This includes the module's description, metadata, events, flags, and available options.
+        """
+        # Retrieve the module's metadata, options, events, and flags
+        meta = getattr(self, "meta", {})
+        options = getattr(self, "options", {})
+        options_desc = getattr(self, "options_desc", {})
+        watched_events = getattr(self, "watched_events", [])
+        produced_events = getattr(self, "produced_events", [])
+        flags = getattr(self, "flags", [])
+
+        help_text = "\n" + "=" * 40 + "\n"
+        help_text += f"Module Help: {self.__name__}\n"
+        help_text += "=" * 40 + "\n\n"
+
+        for key, value in meta.items():
+            help_text += f"{key.replace('_', ' ').title()}: {value}\n"
+
+        help_text += "\nWatched Events:\n"
+        help_text += "  " + ", ".join(watched_events) + "\n" if watched_events else "  None\n"
+
+        help_text += "\nProduced Events:\n"
+        help_text += "  " + ", ".join(produced_events) + "\n" if produced_events else "  None\n"
+
+        help_text += "\nFlags:\n"
+        help_text += "  " + ", ".join(flags) + "\n" if flags else "  None\n"
+
+        help_text += "\nOptions:\n"
+        if options:
+            for option, default_value in options.items():
+                option_description = options_desc.get(option, "No description available.")
+                help_text += f"  - {option}:\n"
+                help_text += f"      Description: {option_description}\n"
+                help_text += f"      Default: {default_value}\n"
+        else:
+            help_text += "  No options available."
+
+        return help_text
+
 
 class BaseInterceptModule(BaseModule):
     """

bbot/modules/deadly/nuclei.py

@@ -15,7 +15,7 @@ class nuclei(BaseModule):
     }
 
     options = {
-        "version": "3.3.9",
+        "version": "3.3.10",
         "tags": "",
         "templates": "",
         "severity": "",

bbot/modules/internal/excavate.py

@@ -262,6 +262,8 @@ async def report(
 
 
 class CustomExtractor(ExcavateRule):
+    description = "Enables custom, user-defined YARA rules."
+
     def __init__(self, excavate):
         super().__init__(excavate)
 
@@ -356,6 +358,7 @@ def url_unparse(self, param_type, parsed_url):
         )
 
     class ParameterExtractor(ExcavateRule):
+        description = "Extracts web parameters. Enabled if any modules are enabled that emit WEB_PARAMETER events."
         yara_rules = {}
 
         class ParameterExtractorRule:
@@ -531,6 +534,8 @@ async def process(self, yara_results, event, yara_rule_settings, discovery_conte
                                 self.excavate.debug(f"blocked parameter [{parameter_name}] due to validation failure")
 
     class CSPExtractor(ExcavateRule):
+        description = "Extracts domains from CSP headers."
+
         yara_rules = {
             "csp": r'rule csp { meta: tags = "affiliate" description = "contains CSP Header" strings: $csp = /Content-Security-Policy:[^\r\n]+/ nocase condition: $csp }',
         }
@@ -543,6 +548,8 @@ async def process(self, yara_results, event, yara_rule_settings, discovery_conte
                         await self.report(domain, event, yara_rule_settings, discovery_context, event_type="DNS_NAME")
 
     class EmailExtractor(ExcavateRule):
+        description = "Extract email addresses."
+
         yara_rules = {
             "email": 'rule email { meta: description = "contains email address" strings: $email = /[^\\W_][\\w\\-\\.\\+\']{0,100}@[a-zA-Z0-9\\-]{1,100}(\\.[a-zA-Z0-9\\-]{1,100})*\\.[a-zA-Z]{2,63}/ nocase fullword condition: $email }',
         }
@@ -556,11 +563,13 @@ async def process(self, yara_results, event, yara_rule_settings, discovery_conte
 
     # Future Work: Emit a JWT Object, and make a new Module to ingest it.
     class JWTExtractor(ExcavateRule):
+        description = "Extracts JSON Web Tokens."
         yara_rules = {
             "jwt": r'rule jwt { meta: emit_match = "True" description = "contains JSON Web Token (JWT)" strings: $jwt = /\beyJ[_a-zA-Z0-9\/+]*\.[_a-zA-Z0-9\/+]*\.[_a-zA-Z0-9\/+]*/ nocase condition: $jwt }',
         }
 
     class ErrorExtractor(ExcavateRule):
+        description = "Identifies error messages from various platforms."
         signatures = {
             "PHP_1": r"/\.php on line [0-9]+/",
             "PHP_2": r"/\.php<\/b> on line <b>[0-9]+/",
@@ -598,6 +607,7 @@ async def process(self, yara_results, event, yara_rule_settings, discovery_conte
                     await self.report(event_data, event, yara_rule_settings, discovery_context, event_type="FINDING")
 
     class SerializationExtractor(ExcavateRule):
+        description = "Identifies serialized objects from various platforms."
         regexes = {
             "Java": re.compile(r"[^a-zA-Z0-9\/+]rO0[a-zA-Z0-9+\/]+={0,2}"),
             "DOTNET": re.compile(r"[^a-zA-Z0-9\/+]AAEAAAD\/\/[a-zA-Z0-9\/+]+={0,2}"),
@@ -627,12 +637,14 @@ async def process(self, yara_results, event, yara_rule_settings, discovery_conte
                     await self.report(event_data, event, yara_rule_settings, discovery_context, event_type="FINDING")
 
     class FunctionalityExtractor(ExcavateRule):
+        description = "Detects potentially exploitable functionality and attack surface in web applications."
         yara_rules = {
             "File_Upload_Functionality": r'rule File_Upload_Functionality { meta: description = "contains file upload functionality" strings: $fileuploadfunc = /<input[^>]+type=["\']?file["\']?[^>]+>/ nocase condition: $fileuploadfunc }',
             "Web_Service_WSDL": r'rule Web_Service_WSDL { meta: emit_match = "True" description = "contains a web service WSDL URL" strings: $wsdl = /https?:\/\/[^\s]*\.(wsdl)/ nocase condition: $wsdl }',
         }
 
     class NonHttpSchemeExtractor(ExcavateRule):
+        description = "Detects URIs with non-HTTP schemes."
         yara_rules = {
             "Non_HTTP_Scheme": r'rule Non_HTTP_Scheme { meta: description = "contains non-http scheme URL" strings: $nonhttpscheme = /\b\w{2,35}:\/\/[\w.-]+(:\d+)?\b/ nocase fullword condition: $nonhttpscheme }'
         }
@@ -681,6 +693,7 @@ def abort_if(e):
                     )
 
     class URLExtractor(ExcavateRule):
+        description = "Extracts URLs."
         yara_rules = {
             "url_full": (
                 r"""
@@ -773,6 +786,8 @@ async def report_prep(self, event_data, event_type, event, tags, **kwargs):
             return event_draft
 
     class HostnameExtractor(ExcavateRule):
+        description = "DNS name discovery, based on the scan target."
+
         yara_rules = {}
 
         def __init__(self, excavate):
@@ -785,6 +800,7 @@ async def process(self, yara_results, event, yara_rule_settings, discovery_conte
                     await self.report(domain_str, event, yara_rule_settings, discovery_context, event_type="DNS_NAME")
 
     class LoginPageExtractor(ExcavateRule):
+        description = "Detects login pages with username and password fields."
         yara_rules = {
             "login_page": r"""
             rule login_page {
@@ -1084,3 +1100,39 @@ async def handle_event(self, event):
                 content_type="",
                 discovery_context="Parsed file content",
             )
+
+    @classmethod
+    def help_text(self):
+        # Call the base class help_text method
+        base_help_text = super().help_text()
+
+        # Import the current module to inspect its classes
+        import sys
+
+        current_module = sys.modules[self.__module__]
+
+        # Function to recursively find subclasses of ExcavateRule
+        def find_subclasses(cls):
+            subclasses = []
+            for name, obj in vars(cls).items():
+                if isinstance(obj, type) and issubclass(obj, ExcavateRule) and obj is not ExcavateRule:
+                    description = getattr(obj, "description", "No description available.")
+                    subclasses.append((name, description))
+                # Recursively check for nested classes
+                if isinstance(obj, type):
+                    subclasses.extend(find_subclasses(obj))
+            return subclasses
+
+        # Find all classes in the module that inherit from ExcavateRule
+        submodules = find_subclasses(current_module)
+
+        # Format submodules information
+        submodules_info = "\nSubmodules:\n"
+        if submodules:
+            for submodule, description in submodules:
+                submodules_info += f"  - {submodule}: {description}\n"
+        else:
+            submodules_info += "  No submodules available.\n"
+
+        # Combine the base help text with the submodules information
+        return base_help_text + submodules_info

bbot/modules/trufflehog.py

@@ -14,7 +14,7 @@ class trufflehog(BaseModule):
     }
 
     options = {
-        "version": "3.88.13",
+        "version": "3.88.15",
         "config": "",
         "only_verified": True,
         "concurrency": 8,

bbot/scanner/preset/args.py

@@ -67,6 +67,11 @@ class BBOTArgs:
             "",
             "bbot -lf",
         ),
+        (
+            "Show help for a specific module",
+            "",
+            "bbot -mh <module_name>",
+        ),
     ]
 
     epilog = "EXAMPLES\n"
@@ -314,6 +319,14 @@ def create_parser(self, *args, **kwargs):
             help="Show the current preset in its full form, including defaults",
         )
 
+        scan.add_argument(
+            "-mh",
+            "--module-help",
+            default=None,
+            help="Show help for a specific module",
+            metavar="MODULE",
+        )
+
         output = p.add_argument_group(title="Output")
         output.add_argument(
             "-o",

bbot/scanner/target.py

@@ -95,7 +95,7 @@ def add(self, targets):
             else:
                 event = self.make_event(target)
                 if event:
-                    self.inputs.add(target)
+                    self.inputs.add(str(target))
                     _events = [event]
             for event in _events:
                 events.add(event)

bbot/test/test_step_1/test_cli.py

@@ -449,6 +449,20 @@ async def test_cli_customheaders(monkeypatch, caplog, capsys):
     assert "Custom headers not formatted correctly (missing header name or value)" in caplog.text
 
 
+@pytest.mark.asyncio
+async def test_cli_module_help(monkeypatch, capsys):
+    monkeypatch.setattr(sys, "exit", lambda *args, **kwargs: True)
+    monkeypatch.setattr(os, "_exit", lambda *args, **kwargs: True)
+
+    monkeypatch.setattr("sys.argv", ["bbot", "--module-help", "excavate"])
+    success = await cli._main()
+    assert success is None, "module help failed to execute"
+    captured = capsys.readouterr()
+
+    assert "Extracts domains from CSP headers" in captured.out
+    assert "Module Help:" in captured.out
+
+
 def test_cli_config_validation(monkeypatch, caplog):
     monkeypatch.setattr(sys, "exit", lambda *args, **kwargs: True)
     monkeypatch.setattr(os, "_exit", lambda *args, **kwargs: True)

bbot/test/test_step_1/test_presets.py

@@ -1075,3 +1075,19 @@ async def test_preset_output_dir():
     assert output_file.is_file()
 
     shutil.rmtree(output_dir, ignore_errors=True)
+
+
+# regression test for https://github.com/blacklanternsecurity/bbot/issues/2337
+def test_preset_serialization():
+    from ipaddress import ip_address, ip_network
+
+    preset = Preset("192.168.1.1")
+    preset = preset.bake()
+
+    import orjson as json
+
+    preset_dict = preset.to_dict(include_target=True)
+    print(preset_dict)
+    preset_str = json.dumps(preset_dict)
+    preset_dict = json.loads(preset_str)
+    assert preset_dict == {"target": ["192.168.1.1"], "whitelist": ["192.168.1.1/32"]}