fix conflict

README.md

@@ -1,6 +1,6 @@
 [![bbot_banner](https://github.com/user-attachments/assets/f02804ce-9478-4f1e-ac4d-9cf5620a3214)](https://github.com/blacklanternsecurity/bbot)
 
-[![Python Version](https://img.shields.io/badge/python-3.9+-FF8400)](https://www.python.org) [![License](https://img.shields.io/badge/license-GPLv3-FF8400.svg)](https://github.com/blacklanternsecurity/bbot/blob/dev/LICENSE) [![DEF CON Recon Village 2024](https://img.shields.io/badge/DEF%20CON%20Demo%20Labs-2023-FF8400.svg)](https://www.reconvillage.org/talks) [![PyPi Downloads](https://static.pepy.tech/personalized-badge/bbot?right_color=orange&left_color=grey)](https://pepy.tech/project/bbot) [![Black](https://img.shields.io/badge/code%20style-black-000000.svg)](https://github.com/psf/black) [![Tests](https://github.com/blacklanternsecurity/bbot/actions/workflows/tests.yml/badge.svg?branch=stable)](https://github.com/blacklanternsecurity/bbot/actions?query=workflow%3A"tests") [![Codecov](https://codecov.io/gh/blacklanternsecurity/bbot/branch/dev/graph/badge.svg?token=IR5AZBDM5K)](https://codecov.io/gh/blacklanternsecurity/bbot) [![Discord](https://img.shields.io/discord/859164869970362439)](https://discord.com/invite/PZqkgxu5SA)
+[![Python Version](https://img.shields.io/badge/python-3.9+-FF8400)](https://www.python.org) [![License](https://img.shields.io/badge/license-GPLv3-FF8400.svg)](https://github.com/blacklanternsecurity/bbot/blob/dev/LICENSE) [![DEF CON Recon Village 2024](https://img.shields.io/badge/DEF%20CON%20Demo%20Labs-2023-FF8400.svg)](https://www.reconvillage.org/talks) [![PyPi Downloads](https://static.pepy.tech/personalized-badge/bbot?right_color=orange&left_color=grey)](https://pepy.tech/project/bbot) [![Ruff](https://img.shields.io/endpoint?url=https://raw.githubusercontent.com/astral-sh/ruff/main/assets/badge/v2.json)](https://github.com/astral-sh/ruff) [![Tests](https://github.com/blacklanternsecurity/bbot/actions/workflows/tests.yml/badge.svg?branch=stable)](https://github.com/blacklanternsecurity/bbot/actions?query=workflow%3A"tests") [![Codecov](https://codecov.io/gh/blacklanternsecurity/bbot/branch/dev/graph/badge.svg?token=IR5AZBDM5K)](https://codecov.io/gh/blacklanternsecurity/bbot) [![Discord](https://img.shields.io/discord/859164869970362439)](https://discord.com/invite/PZqkgxu5SA)
 
 ### **BEE·bot** is a multipurpose scanner inspired by [Spiderfoot](https://github.com/smicallef/spiderfoot), built to automate your **Recon**, **Bug Bounties**, and **ASM**!
 

bbot/cli.py

@@ -130,8 +130,8 @@ async def _main():
         ]
         if deadly_modules and not options.allow_deadly:
             log.hugewarning(f"You enabled the following deadly modules: {','.join(deadly_modules)}")
-            log.hugewarning(f"Deadly modules are highly intrusive")
-            log.hugewarning(f"Please specify --allow-deadly to continue")
+            log.hugewarning("Deadly modules are highly intrusive")
+            log.hugewarning("Please specify --allow-deadly to continue")
             return False
 
         # --current-preset

bbot/core/config/logger.py

@@ -68,7 +68,7 @@ def __init__(self, core):
         self.listener = None
 
         # if we haven't set up logging yet, do it now
-        if not "_BBOT_LOGGING_SETUP" in os.environ:
+        if "_BBOT_LOGGING_SETUP" not in os.environ:
             os.environ["_BBOT_LOGGING_SETUP"] = "1"
             self.queue = multiprocessing.Queue()
             self.setup_queue_handler()

bbot/core/core.py

@@ -106,7 +106,7 @@ def default_config(self):
         if DEFAULT_CONFIG is None:
             self.default_config = self.files_config.get_default_config()
             # ensure bbot home dir
-            if not "home" in self.default_config:
+            if "home" not in self.default_config:
                 self.default_config["home"] = "~/.bbot"
         return DEFAULT_CONFIG
 

bbot/core/event/base.py

@@ -177,8 +177,8 @@ def __init__(
         self._scope_distance = None
         self._module_priority = None
         self._resolved_hosts = set()
-        self.dns_children = dict()
-        self.raw_dns_records = dict()
+        self.dns_children = {}
+        self.raw_dns_records = {}
         self._discovery_context = ""
         self._discovery_context_regex = re.compile(r"\{(?:event|module)[^}]*\}")
         self.web_spider_distance = 0
@@ -205,7 +205,7 @@ def __init__(
         # self.scan holds the instantiated scan object (for helpers, etc.)
         self.scan = scan
         if (not self.scan) and (not self._dummy):
-            raise ValidationError(f"Must specify scan")
+            raise ValidationError("Must specify scan")
         # self.scans holds a list of scan IDs from scans that encountered this event
         self.scans = []
         if scans is not None:
@@ -224,7 +224,7 @@ def __init__(
 
         self.parent = parent
         if (not self.parent) and (not self._dummy):
-            raise ValidationError(f"Must specify event parent")
+            raise ValidationError("Must specify event parent")
 
         if tags is not None:
             for tag in tags:
@@ -303,9 +303,9 @@ def internal(self, value):
         The purpose of internal events is to enable speculative/explorative discovery without cluttering
         the console with irrelevant or uninteresting events.
         """
-        if not value in (True, False):
+        if value not in (True, False):
             raise ValueError(f'"internal" must be boolean, not {type(value)}')
-        if value == True:
+        if value is True:
             self.add_tag("internal")
         else:
             self.remove_tag("internal")
@@ -770,7 +770,7 @@ def json(self, mode="json"):
         Returns:
             dict: JSON-serializable dictionary representation of the event object.
         """
-        j = dict()
+        j = {}
         # type, ID, scope description
         for i in ("type", "id", "uuid", "scope_description", "netloc"):
             v = getattr(self, i, "")
@@ -1015,12 +1015,12 @@ def __init__(self, *args, **kwargs):
         if not self.host:
             for parent in self.get_parents(include_self=True):
                 # inherit closest URL
-                if not "url" in self.data:
+                if "url" not in self.data:
                     parent_url = getattr(parent, "parsed_url", None)
                     if parent_url is not None:
                         self.data["url"] = parent_url.geturl()
                 # inherit closest path
-                if not "path" in self.data and isinstance(parent.data, dict) and not parent.type == "HTTP_RESPONSE":
+                if "path" not in self.data and isinstance(parent.data, dict) and not parent.type == "HTTP_RESPONSE":
                     parent_path = parent.data.get("path", None)
                     if parent_path is not None:
                         self.data["path"] = parent_path
@@ -1229,7 +1229,7 @@ def sanitize_data(self, data):
 
     def add_tag(self, tag):
         host_same_as_parent = self.parent and self.host == self.parent.host
-        if tag == "spider-danger" and host_same_as_parent and not "spider-danger" in self.tags:
+        if tag == "spider-danger" and host_same_as_parent and "spider-danger" not in self.tags:
             # increment the web spider distance
             if self.type == "URL_UNVERIFIED":
                 self.web_spider_distance += 1
@@ -1251,7 +1251,7 @@ def with_port(self):
 
     def _words(self):
         first_elem = self.parsed_url.path.lstrip("/").split("/")[0]
-        if not "." in first_elem:
+        if "." not in first_elem:
             return extract_words(first_elem)
         return set()
 
@@ -1279,7 +1279,7 @@ def __init__(self, *args, **kwargs):
     @property
     def resolved_hosts(self):
         # TODO: remove this when we rip out httpx
-        return set(".".join(i.split("-")[1:]) for i in self.tags if i.startswith("ip-"))
+        return {".".join(i.split("-")[1:]) for i in self.tags if i.startswith("ip-")}
 
     @property
     def pretty_string(self):
@@ -1667,7 +1667,7 @@ def make_event(
             event.parent = parent
         if context is not None:
             event.discovery_context = context
-        if internal == True:
+        if internal is True:
             event.internal = True
         if tags:
             event.tags = tags.union(event.tags)

bbot/core/helpers/command.py

@@ -269,11 +269,11 @@ def _prepare_command_kwargs(self, command, kwargs):
         (['sudo', '-E', '-A', 'LD_LIBRARY_PATH=...', 'PATH=...', 'ls', '-l'], {'limit': 104857600, 'stdout': -1, 'stderr': -1, 'env': environ(...)})
     """
     # limit = 100MB (this is needed for cases like httpx that are sending large JSON blobs over stdout)
-    if not "limit" in kwargs:
+    if "limit" not in kwargs:
         kwargs["limit"] = 1024 * 1024 * 100
-    if not "stdout" in kwargs:
+    if "stdout" not in kwargs:
         kwargs["stdout"] = asyncio.subprocess.PIPE
-    if not "stderr" in kwargs:
+    if "stderr" not in kwargs:
         kwargs["stderr"] = asyncio.subprocess.PIPE
     sudo = kwargs.pop("sudo", False)
 
@@ -286,7 +286,7 @@ def _prepare_command_kwargs(self, command, kwargs):
 
     # use full path of binary, if not already specified
     binary = command[0]
-    if not "/" in binary:
+    if "/" not in binary:
         binary_full_path = which(binary)
         if binary_full_path is None:
             raise SubprocessError(f'Command "{binary}" was not found')

bbot/core/helpers/depsinstaller/installer.py

@@ -97,11 +97,11 @@ async def install(self, *modules):
                         or self.deps_behavior == "force_install"
                     ):
                         if not notified:
-                            log.hugeinfo(f"Installing module dependencies. Please be patient, this may take a while.")
+                            log.hugeinfo("Installing module dependencies. Please be patient, this may take a while.")
                             notified = True
                         log.verbose(f'Installing dependencies for module "{m}"')
                         # get sudo access if we need it
-                        if preloaded.get("sudo", False) == True:
+                        if preloaded.get("sudo", False) is True:
                             self.ensure_root(f'Module "{m}" needs root privileges to install its dependencies.')
                         success = await self.install_module(m)
                         self.setup_status[module_hash] = success
@@ -159,7 +159,7 @@ async def install_module(self, module):
         deps_common = preloaded["deps"]["common"]
         if deps_common:
             for dep_common in deps_common:
-                if self.setup_status.get(dep_common, False) == True:
+                if self.setup_status.get(dep_common, False) is True:
                     log.debug(
                         f'Skipping installation of dependency "{dep_common}" for module "{module}" since it is already installed'
                     )
@@ -244,7 +244,7 @@ def shell(self, module, commands):
         if success:
             log.info(f"Successfully ran {len(commands):,} shell commands")
         else:
-            log.warning(f"Failed to run shell dependencies")
+            log.warning("Failed to run shell dependencies")
         return success
 
     def tasks(self, module, tasks):
@@ -310,7 +310,7 @@ def ansible_run(self, tasks=None, module=None, args=None, ansible_args=None):
         return success, err
 
     def read_setup_status(self):
-        setup_status = dict()
+        setup_status = {}
         if self.setup_status_cache.is_file():
             with open(self.setup_status_cache) as f:
                 with suppress(Exception):

bbot/core/helpers/diff.py

@@ -101,7 +101,7 @@ async def _baseline(self):
             ddiff = DeepDiff(baseline_1_json, baseline_2_json, ignore_order=True, view="tree")
             self.ddiff_filters = []
 
-            for k, v in ddiff.items():
+            for k in ddiff.keys():
                 for x in list(ddiff[k]):
                     log.debug(f"Added {k} filter for path: {x.path()}")
                     self.ddiff_filters.append(x.path())
@@ -140,7 +140,7 @@ def compare_headers(self, headers_1, headers_2):
 
         ddiff = DeepDiff(headers_1, headers_2, ignore_order=True, view="tree")
 
-        for k, v in ddiff.items():
+        for k in ddiff.keys():
             for x in list(ddiff[k]):
                 try:
                     header_value = str(x).split("'")[1]
@@ -183,7 +183,7 @@ async def compare(
 
         await self._baseline()
 
-        if timeout == None:
+        if timeout is None:
             timeout = self.timeout
 
         reflection = False
@@ -238,11 +238,11 @@ async def compare(
 
         different_headers = self.compare_headers(self.baseline.headers, subject_response.headers)
         if different_headers:
-            log.debug(f"headers were different, no match")
+            log.debug("headers were different, no match")
             diff_reasons.append("header")
 
-        if self.compare_body(self.baseline_json, subject_json) == False:
-            log.debug(f"difference in HTML body, no match")
+        if self.compare_body(self.baseline_json, subject_json) is False:
+            log.debug("difference in HTML body, no match")
 
             diff_reasons.append("body")
 
@@ -275,6 +275,6 @@ async def canary_check(self, url, mode, rounds=3):
             )
 
             # if a nonsense header "caused" a difference, we need to abort. We also need to abort if our canary was reflected
-            if match == False or reflection == True:
+            if match is False or reflection is True:
                 return False
         return True

bbot/core/helpers/dns/brute.py

@@ -164,7 +164,7 @@ def gen_random_subdomains(self, n=50):
         for i in range(0, max(0, n - 5)):
             d = delimiters[i % len(delimiters)]
             l = lengths[i % len(lengths)]
-            segments = list(random.choice(self.devops_mutations) for _ in range(l))
+            segments = [random.choice(self.devops_mutations) for _ in range(l)]
             segments.append(self.parent_helper.rand_string(length=8, digits=False))
             subdomain = d.join(segments)
             yield subdomain

bbot/core/helpers/dns/dns.py

@@ -178,7 +178,7 @@ def _wildcard_prevalidation(self, host):
 
         host = clean_dns_record(host)
         # skip check if it's an IP or a plain hostname
-        if is_ip(host) or not "." in host:
+        if is_ip(host) or "." not in host:
             return False
 
         # skip if query isn't a dns name

bbot/core/helpers/dns/engine.py

@@ -54,7 +54,7 @@ def __init__(self, socket_path, config={}, debug=False):
         dns_omit_queries = self.dns_config.get("omit_queries", None)
         if not dns_omit_queries:
             dns_omit_queries = []
-        self.dns_omit_queries = dict()
+        self.dns_omit_queries = {}
         for d in dns_omit_queries:
             d = d.split(":")
             if len(d) == 2:
@@ -72,7 +72,7 @@ def __init__(self, socket_path, config={}, debug=False):
             self.wildcard_ignore = []
         self.wildcard_ignore = tuple([str(d).strip().lower() for d in self.wildcard_ignore])
         self.wildcard_tests = self.dns_config.get("wildcard_tests", 5)
-        self._wildcard_cache = dict()
+        self._wildcard_cache = {}
         # since wildcard detection takes some time, This is to prevent multiple
         # modules from kicking off wildcard detection for the same domain at the same time
         self._wildcard_lock = NamedLock()
@@ -82,7 +82,7 @@ def __init__(self, socket_path, config={}, debug=False):
         self._last_connectivity_warning = time.time()
         # keeps track of warnings issued for wildcard detection to prevent duplicate warnings
         self._dns_warnings = set()
-        self._errors = dict()
+        self._errors = {}
         self._debug = self.dns_config.get("debug", False)
         self._dns_cache = LRUCache(maxsize=10000)
 
@@ -638,7 +638,7 @@ async def _connectivity_check(self, interval=5):
                     self._last_dns_success = time.time()
                     return True
         if time.time() - self._last_connectivity_warning > interval:
-            self.log.warning(f"DNS queries are failing, please check your internet connection")
+            self.log.warning("DNS queries are failing, please check your internet connection")
             self._last_connectivity_warning = time.time()
         self._errors.clear()
         return False

bbot/core/helpers/files.py

@@ -83,7 +83,7 @@ def _feed_pipe(self, pipe, content, text=True):
                     for c in content:
                         p.write(decode_fn(c) + newline)
         except BrokenPipeError:
-            log.debug(f"Broken pipe in _feed_pipe()")
+            log.debug("Broken pipe in _feed_pipe()")
         except ValueError:
             log.debug(f"Error _feed_pipe(): {traceback.format_exc()}")
     except KeyboardInterrupt:

bbot/core/helpers/helper.py

@@ -153,7 +153,9 @@ def temp_filename(self, extension=None):
         return self.temp_dir / filename
 
     def clean_old_scans(self):
-        _filter = lambda x: x.is_dir() and self.regexes.scan_name_regex.match(x.name)
+        def _filter(x):
+            return x.is_dir() and self.regexes.scan_name_regex.match(x.name)
+
         self.clean_old(self.scans_dir, keep=self.keep_old_scans, filter=_filter)
 
     def make_target(self, *targets, **kwargs):

bbot/core/helpers/interactsh.py

@@ -155,7 +155,7 @@ async def register(self, callback=None):
             break
 
         if not self.server:
-            raise InteractshError(f"Failed to register with an interactsh server")
+            raise InteractshError("Failed to register with an interactsh server")
 
         log.info(
             f"Successfully registered to interactsh server {self.server} with correlation_id {self.correlation_id} [{self.domain}]"
@@ -181,7 +181,7 @@ async def deregister(self):
             >>> await interactsh_client.deregister()
         """
         if not self.server or not self.correlation_id or not self.secret:
-            raise InteractshError(f"Missing required information to deregister")
+            raise InteractshError("Missing required information to deregister")
 
         headers = {}
         if self.token:
@@ -226,7 +226,7 @@ async def poll(self):
             ]
         """
         if not self.server or not self.correlation_id or not self.secret:
-            raise InteractshError(f"Missing required information to poll")
+            raise InteractshError("Missing required information to poll")
 
         headers = {}
         if self.token: