gemini code base review

agent/agent.py

@@ -178,7 +178,7 @@ def run(
         self,
         host: ipaddress.IPv4Address = ipaddress.IPv4Address("0.0.0.0"),
         port: int = 8000,
-        event = None,
+        event=None,
     ):
         socketserver.ThreadingTCPServer.allow_reuse_address = True
         self.s = socketserver.ThreadingTCPServer((str(host), port), self.handler)

analyzer/windows/lib/api/process.py

@@ -72,7 +72,7 @@
 CSIDL_SYSTEM = 0x0025
 CSIDL_SYSTEMX86 = 0x0029
 CSIDL_PROGRAM_FILES = 0x0026
-CSIDL_PROGRAM_FILESX86 = 0x002a
+CSIDL_PROGRAM_FILESX86 = 0x002A
 
 IOCTL_PID = 0x222008
 IOCTL_CUCKOO_PATH = 0x22200C
@@ -117,6 +117,7 @@ def nt_path_to_dos_path_ansi(nt_path: str) -> str:
                 return converted.decode("utf-8", errors="ignore")
     return nt_path
 
+
 def NT_SUCCESS(val):
     return val >= 0
 
@@ -255,7 +256,7 @@ def get_folder_path(self, csidl):
         """Use SHGetFolderPathW to get the system folder path for a given CSIDL."""
         buf = create_string_buffer(MAX_PATH)
         windll.shell32.SHGetFolderPathA(None, csidl, None, 0, buf)
-        return buf.value.decode('utf-8', errors='ignore')
+        return buf.value.decode("utf-8", errors="ignore")
 
     def get_image_name(self):
         """Get the image name; returns an empty string on error."""
@@ -314,7 +315,7 @@ def detect_dll_sideloading(self, directory_path: str) -> bool:
             if not directory.is_dir():
                 return False
 
-            if (directory/"capemon.dll").exists():
+            if (directory / "capemon.dll").exists():
                 return False
 
             # Early exit if directory is a known system location
@@ -324,7 +325,7 @@ def detect_dll_sideloading(self, directory_path: str) -> bool:
                     Path(self.get_folder_path(CSIDL_SYSTEM)).resolve(),
                     Path(self.get_folder_path(CSIDL_SYSTEMX86)).resolve(),
                     Path(self.get_folder_path(CSIDL_PROGRAM_FILES)).resolve(),
-                    Path(self.get_folder_path(CSIDL_PROGRAM_FILESX86)).resolve()
+                    Path(self.get_folder_path(CSIDL_PROGRAM_FILESX86)).resolve(),
                 }
                 if directory.resolve() in system_dirs:
                     return False
@@ -860,11 +861,7 @@ def __str__(self):
     def has_msimg32(self, directory_path: str) -> bool:
         """Check if msimg32.dll exists in directory"""
         try:
-            return any(
-                f.name.lower() == "msimg32.dll"
-                for f in Path(directory_path).glob("*")
-                if f.is_file()
-            )
+            return any(f.name.lower() == "msimg32.dll" for f in Path(directory_path).glob("*") if f.is_file())
         except (OSError, PermissionError):
             return False
 
@@ -892,5 +889,10 @@ def deploy_version_proxy(self, directory_path: str):
         except OSError as e:
             log.error("Failed to copy DLL: %s", e)
             return
-        log.info("%s DLL to sideload is %s, sideloader %s", bit_str, os.path.join(directory_path, "capemon.dll"), os.path.join(directory_path, "version.dll"))
+        log.info(
+            "%s DLL to sideload is %s, sideloader %s",
+            bit_str,
+            os.path.join(directory_path, "capemon.dll"),
+            os.path.join(directory_path, "version.dll"),
+        )
         return

analyzer/windows/modules/auxiliary/dns_etw.py

@@ -23,7 +23,8 @@
 except ImportError as e:
     log.debug(
         "Could not load auxiliary module DNS_ETW due to '%s'\nIn order to use DNS_ETW functionality, it "
-        "is required to have pywintrace setup in python", str(e)
+        "is required to have pywintrace setup in python",
+        str(e),
     )
 
 __author__ = "[Canadian Centre for Cyber Security] @CybercentreCanada"

analyzer/windows/modules/auxiliary/wmi_etw.py

@@ -23,7 +23,8 @@
 except ImportError as e:
     log.debug(
         "Could not load auxiliary module WMI_ETW due to '%s'\nIn order to use WMI_ETW functionality, it "
-        "is required to have pywintrace setup in python", str(e)
+        "is required to have pywintrace setup in python",
+        str(e),
     )
 
 __author__ = "[Andrea Oliveri starting from code of Canadian Centre for Cyber Security]"

analyzer/windows/modules/packages/firefox_ext.py

@@ -11,6 +11,7 @@
 
 log = logging.getLogger(__name__)
 
+
 class Firefox_Ext(Package):
     """Firefox analysis package (with extension)."""
 
@@ -36,12 +37,12 @@ def start(self, url):
             log.error("Invalid base64 encoded user agent provided.")
             user_agent = None
         if user_agent and self.profile_path:
-            config = os.path.join(self.profile_path, 'prefs.js')
-            ua_decoded = base64.b64decode(user_agent).decode('utf-8')
+            config = os.path.join(self.profile_path, "prefs.js")
+            ua_decoded = base64.b64decode(user_agent).decode("utf-8")
             ua_config = f'user_pref("general.useragent.override", "{ua_decoded}");\n'
             try:
                 os.makedirs(os.path.dirname(config), exist_ok=True)
-                with open(config, 'a') as file:
+                with open(config, "a") as file:
                     file.write(ua_config)
                 log.info("Successfully appended user agent to prefs.js: %s", ua_decoded)
             except Exception as e:

analyzer/windows/modules/packages/msix.py

@@ -54,7 +54,7 @@ def start(self, path):
                     log.debug("msix file contains script %s", str(path))
 
         if not args:
-            args = fr"-NoProfile -ExecutionPolicy bypass {os.getcwd()}\data\msix.ps1 {path}"
+            args = rf"-NoProfile -ExecutionPolicy bypass {os.getcwd()}\data\msix.ps1 {path}"
             # now we need to get app id and launch it
 
         return self.execute(powershell, args, powershell)

analyzer/windows/prescripts/prescript_detection.py

@@ -690,7 +690,9 @@ def main(args):
                     overwrite=bool(params_dict[ACTIONS_PARAMETERS[parsed_action][2]]),
                 )
                 log.info(
-                    "Adding file from %s to %s", params_dict[ACTIONS_PARAMETERS[parsed_action][0]], params_dict[ACTIONS_PARAMETERS[parsed_action][1]]
+                    "Adding file from %s to %s",
+                    params_dict[ACTIONS_PARAMETERS[parsed_action][0]],
+                    params_dict[ACTIONS_PARAMETERS[parsed_action][1]],
                 )
                 # print(
                 #    f"Adding file from {params_dict[ACTIONS_PARAMETERS[parsed_action][0]]} to {params_dict[ACTIONS_PARAMETERS[parsed_action][1]]}"

cuckoo.py

@@ -20,15 +20,14 @@
 from lib.cuckoo.common.exceptions import CuckooCriticalError, CuckooDependencyError
 
 try:
-    import bson
-
     from lib.cuckoo.common.constants import CUCKOO_ROOT, CUCKOO_VERSION
     from lib.cuckoo.common.logo import logo
     from lib.cuckoo.core.resultserver import ResultServer
     from lib.cuckoo.core.scheduler import Scheduler
     from lib.cuckoo.core.startup import (
         check_configs,
         check_linux_dist,
+        check_network_settings,
         check_tcpdump_permissions,
         check_webgui_mongo,
         check_working_directory,
@@ -40,7 +39,6 @@
         init_tasks,
     )
 
-    bson  # Pretend like it's actually being used (for static checkers.)
 except (CuckooDependencyError, ImportError) as e:
     print(f"ERROR: Missing dependency: {e}")
     sys.exit()
@@ -57,6 +55,7 @@ def cuckoo_init(quiet=False, debug=False, artwork=False, test=False):
     logo()
     check_working_directory()
     check_configs()
+    check_network_settings()
     create_structure()
     init_database()
 
@@ -70,13 +69,7 @@ def cuckoo_init(quiet=False, debug=False, artwork=False, test=False):
         except KeyboardInterrupt:
             return
 
-    if quiet:
-        level = logging.WARN
-    elif debug:
-        level = logging.DEBUG
-    else:
-        level = logging.INFO
-    log.setLevel(level)
+    level = logging.WARN if quiet else logging.DEBUG if debug else logging.INFO
     init_logging(level)
 
     check_webgui_mongo()
@@ -87,7 +80,7 @@ def cuckoo_init(quiet=False, debug=False, artwork=False, test=False):
     init_routing()
     check_tcpdump_permissions()
 
-    # This is just a temporary hack, we need an actual test suite to integrate with Travis-CI.
+    # ToDo remove This is just a temporary hack, we need an actual test suite to integrate with Travis-CI.
     if test:
         return
 
@@ -98,18 +91,15 @@ def cuckoo_init(quiet=False, debug=False, artwork=False, test=False):
 def cuckoo_main(max_analysis_count=0):
     cur_path = Path.cwd()
     os.chdir(CUCKOO_ROOT)
-
     sched = Scheduler(max_analysis_count)
     try:
         sched.start()
     except KeyboardInterrupt:
         log.info("Received keyboard interrupt, stopping.")
     finally:
         sched.shutdown_machinery()
-
     os.chdir(cur_path)
 
-
 if __name__ == "__main__":
     parser = argparse.ArgumentParser()
     parser.add_argument("-q", "--quiet", help="Display only error messages", action="store_true", required=False)

dev_utils/mongo_hooks.py

@@ -197,5 +197,7 @@ def collect_file_dicts(report) -> itertools.chain:
     file_dicts.append(report.get("CAPE", {}).get("payloads", None) or [])
     file_dicts.append(report.get("procdump", None) or [])
     if report.get("suricata", {}).get("files", []):
-        file_dicts.append(list(filter(None, [file_info.get("file_info", []) for file_info in report.get("suricata", {}).get("files", [])])))
+        file_dicts.append(
+            list(filter(None, [file_info.get("file_info", []) for file_info in report.get("suricata", {}).get("files", [])]))
+        )
     return itertools.chain.from_iterable(file_dicts)