remove dead code

bbot/core/engine.py

@@ -636,7 +636,7 @@ async def finished_tasks(self, tasks, timeout=None):
         """
         if tasks:
             try:
-                done, pending = await asyncio.wait(tasks, return_when=asyncio.FIRST_COMPLETED, timeout=timeout)
+                done, _ = await asyncio.wait(tasks, return_when=asyncio.FIRST_COMPLETED, timeout=timeout)
                 return done
             except BaseException as e:
                 if isinstance(e, (TimeoutError, asyncio.exceptions.TimeoutError)):

bbot/core/helpers/dns/dns.py

@@ -38,7 +38,6 @@ class DNSHelper(EngineClient):
         _wildcard_cache (dict): Cache for wildcard detection results.
         _dns_cache (LRUCache): Cache for DNS resolution results, limited in size.
         resolver_file (Path): File containing system's current resolver nameservers.
-        filter_bad_ptrs (bool): Whether to filter out DNS names that appear to be auto-generated PTR records. Defaults to True.
 
     Args:
         parent_helper: The parent helper object with configuration details and utilities.

bbot/core/helpers/dns/engine.py

@@ -86,8 +86,6 @@ def __init__(self, socket_path, config={}, debug=False):
         self._debug = self.dns_config.get("debug", False)
         self._dns_cache = LRUCache(maxsize=10000)
 
-        self.filter_bad_ptrs = self.dns_config.get("filter_ptrs", True)
-
     async def resolve(self, query, **kwargs):
         """Resolve DNS names and IP addresses to their corresponding results.
 

bbot/core/helpers/ntlm.py

@@ -17,11 +17,9 @@ def __init__(self, pos_tup, raw):
         self.alloc = alloc
         self.offset = offset
         self.raw = raw[offset : offset + length]
-        self.utf16 = False
 
         if len(self.raw) >= 2 and self.raw[1] == "\0":
             self.string = self.raw.decode("utf-16")
-            self.utf16 = True
         else:
             self.string = self.raw
 

bbot/core/helpers/regex.py

@@ -65,7 +65,7 @@ def new_task(regex_name, r):
 
         while tasks:  # While there are tasks pending
             # Wait for the first task to complete
-            done, pending = await asyncio.wait(tasks, return_when=asyncio.FIRST_COMPLETED)
+            done, _ = await asyncio.wait(tasks, return_when=asyncio.FIRST_COMPLETED)
 
             for task in done:
                 result = task.result()

bbot/core/modules.py

@@ -512,60 +512,6 @@ def load_module(self, module_name):
                         # then we have a module
                         return value
 
-    def recommend_dependencies(self, modules):
-        """
-        Returns a dictionary containing missing dependencies and their suggested resolutions
-
-        Needs work. For this we should probably be building a dependency graph
-        """
-        resolve_choices = {}
-        # step 1: build a dictionary containing event types and their associated modules
-        # {"IP_ADDRESS": set("masscan", "ipneighbor", ...)}
-        watched = {}
-        produced = {}
-        for modname in modules:
-            preloaded = self._preloaded.get(modname)
-            if preloaded:
-                for event_type in preloaded.get("watched_events", []):
-                    self.add_or_create(watched, event_type, modname)
-                for event_type in preloaded.get("produced_events", []):
-                    self.add_or_create(produced, event_type, modname)
-        watched_all = {}
-        produced_all = {}
-        for modname, preloaded in self.preloaded().items():
-            if preloaded:
-                for event_type in preloaded.get("watched_events", []):
-                    self.add_or_create(watched_all, event_type, modname)
-                for event_type in preloaded.get("produced_events", []):
-                    self.add_or_create(produced_all, event_type, modname)
-
-        # step 2: check to see if there are missing dependencies
-        for modname in modules:
-            preloaded = self._preloaded.get(modname)
-            module_type = preloaded.get("type", "unknown")
-            if module_type != "scan":
-                continue
-            watched_events = preloaded.get("watched_events", [])
-            missing_deps = {e: not self.check_dependency(e, modname, produced) for e in watched_events}
-            if all(missing_deps.values()):
-                for event_type in watched_events:
-                    if event_type == "SCAN":
-                        continue
-                    choices = produced_all.get(event_type, [])
-                    choices = set(choices)
-                    with suppress(KeyError):
-                        choices.remove(modname)
-                    if event_type not in resolve_choices:
-                        resolve_choices[event_type] = {}
-                    deps = resolve_choices[event_type]
-                    self.add_or_create(deps, "required_by", modname)
-                    for c in choices:
-                        choice_type = self._preloaded.get(c, {}).get("type", "unknown")
-                        if choice_type == "scan":
-                            self.add_or_create(deps, "recommended", c)
-
-        return resolve_choices
-
     def check_dependency(self, event_type, modname, produced):
         if event_type not in produced:
             return False

bbot/modules/dnstlsrpt.py

@@ -44,20 +44,17 @@ class dnstlsrpt(BaseModule):
         "emit_emails": True,
         "emit_raw_dns_records": False,
         "emit_urls": True,
-        "emit_vulnerabilities": True,
     }
     options_desc = {
         "emit_emails": "Emit EMAIL_ADDRESS events",
         "emit_raw_dns_records": "Emit RAW_DNS_RECORD events",
         "emit_urls": "Emit URL_UNVERIFIED events",
-        "emit_vulnerabilities": "Emit VULNERABILITY events",
     }
 
     async def setup(self):
         self.emit_emails = self.config.get("emit_emails", True)
         self.emit_raw_dns_records = self.config.get("emit_raw_dns_records", False)
         self.emit_urls = self.config.get("emit_urls", True)
-        self.emit_vulnerabilities = self.config.get("emit_vulnerabilities", True)
         return await super().setup()
 
     def _incoming_dedup_hash(self, event):
@@ -139,6 +136,3 @@ async def handle_event(self, event):
                                                     tags=tags.append(f"tlsrpt-record-{key}"),
                                                     parent=event,
                                                 )
-
-
-# EOF

bbot/modules/iis_shortnames.py

@@ -116,13 +116,6 @@ async def duplicate_check(self, target, method, url_hint, affirmative_status_cod
 
         return duplicates
 
-    async def threaded_request(self, method, url, affirmative_status_code, c):
-        r = await self.helpers.request(method=method, url=url, allow_redirects=False, retries=2, timeout=10)
-        if r is not None:
-            if r.status_code == affirmative_status_code:
-                return True, c
-        return None, c
-
     async def solve_valid_chars(self, method, target, affirmative_status_code):
         confirmed_chars = []
         confirmed_exts = []

bbot/modules/securitytxt.py

@@ -123,6 +123,3 @@ async def handle_event(self, event):
 
                         if found_url != url and self._urls is True:
                             await self.emit_event(found_url, "URL_UNVERIFIED", parent=event, tags=tags)
-
-
-# EOF

bbot/test/test_step_1/test_scan.py

@@ -111,7 +111,6 @@ async def handle_event(self, event):
     class LongBatchModule(BaseModule):
         watched_events = ["IP_ADDRESS"]
         handled_event = False
-        canceled = False
         _name = "long_batch"
         _batch_size = 2
 

bbot/test/test_step_2/module_tests/test_module_lightfuzz.py

@@ -1723,7 +1723,7 @@ def request_handler(self, request):
                     input_value = param.split("=")[1]
                     break
 
-            if input_value:
+            if input_value is not None:
                 # Simulate flawed escaping
                 sanitized_input = input_value.replace('"', '\\"').replace("'", "\\'")
                 sanitized_input = sanitized_input.replace("<", "%3C").replace(">", "%3E")
@@ -1786,7 +1786,7 @@ def request_handler(self, request):
                     input_value = param.split("=")[1]
                     break
 
-            if input_value:
+            if input_value is not None:
                 # Simulate flawed escaping with opposite quotes
                 sanitized_input = input_value.replace("'", "\\").replace("%22", '\\"')
                 sanitized_input = sanitized_input.replace("<", "%3C").replace(">", "%3E")