Ruff

.github/workflows/python-package.yml

@@ -37,7 +37,7 @@ jobs:
           poetry run pip install git+https://github.com/CAPESandbox/pyattck maco
 
       - name: Run Ruff
-        run: poetry run ruff check . --line-length 132 --ignore E501,E402
+        run: poetry run ruff check . --output-format=github .
 
       - name: Run unit tests
         run: poetry run python -m pytest --import-mode=append
@@ -63,20 +63,15 @@ jobs:
         with:
           python-version: ${{ matrix.python-version }}
 
-      - name: Format with black
-        run: poetry run black .
-
-      # to be replaced with ruff
-      - name: Format imports with isort
-        run: poetry run isort .
-
       - name: Commit changes if any
         # Skip this step if being run by nektos/act
         if: ${{ !env.ACT }}
         run: |
           git config user.name "GitHub Actions"
           git config user.email "action@github.com"
           if output=$(git status --porcelain) && [ ! -z "$output" ]; then
+            git pull
+            git add .
             git commit -m "style: Automatic code formatting" -a
             git push
           fi

agent/agent.py

@@ -227,7 +227,6 @@ def handle(self, obj):
             self.close_connection = True
 
     def shutdown(self):
-
         # BaseServer also features a .shutdown() method, but you can't use
         # that from the same thread as that will deadlock the whole thing.
         if hasattr(self, "s"):

analyzer/linux/analyzer.py

@@ -71,7 +71,7 @@ def monitor_new_processes(parent_pid, interval=0.25):
         new_processes = current_processes - known_processes
 
         for pid in new_processes:
-            log.info(f"New child process detected: {pid}")
+            log.info("New child process detected: %s", str(pid))
             dump_memory(pid)
             add_pids(pid)  # Add the new process to PROCESS_LIST
 
@@ -118,20 +118,20 @@ def dump_memory(pid):
                     chunk = mem_file.read(end - start)
                     output_file.write(chunk)
                 except (OSError, ValueError) as e:
-                    log.error(f"Could not read memory range {start:x}-{end:x}: {e}")
+                    log.error("Could not read memory range %s: {e}", f"{start:x}-{end:x}", str(e))
         maps_file.close()
         mem_file.close()
         output_file.close()
     except FileNotFoundError:
-        log.error(f"Process with PID {pid} not found.")
+        log.error("Process with PID %s not found.", str(pid))
     except PermissionError:
-        log.error(f"Permission denied to access process with PID {pid}.")
+        log.error("Permission denied to access process with PID %s.", str(pid))
 
     if os.path.exists(f"{MEM_PATH}/{pid}.dmp"):
         upload_to_host(f"{MEM_PATH}/{pid}.dmp", f"memory/{pid}.dmp")
         DUMPED_LIST.add(pid)
     else:
-        log.error(f"Memdump file not found in guest machine for PID {pid}")
+        log.error("Memdump file not found in guest machine for PID %s", str(pid))
 
 
 class Analyzer:

analyzer/linux/lib/api/screenshot.py

@@ -139,7 +139,7 @@ async def is_gnome(self):
                     log.info("Detected non-Gnome desktop environment.")
                 else:
                     self._is_gnome = True
-                    log.info(f"Detected Gnome version {version}")
+                    log.info("Detected Gnome version %s", str(version))
                     name = "org.gnome.Screenshot"
                     resp = await self.bus.request_name(name)
                     if resp not in (
@@ -205,8 +205,8 @@ async def take_screenshot_gnome(self):
                 "http://www.freedesktop.org/standards/dbus/1.0/introspect.dtd">
                 <node>
                     <interface name="org.gnome.Shell.Screenshot">
-                      <method name="Screenshot">
-                          <arg name="include_cursor" direction="in" type="b" />
+                        <method name="Screenshot">
+                            <arg name="include_cursor" direction="in" type="b" />
                             <arg name="flash" direction="in" type="b" />
                             <arg name="filename" direction="in" type="s" />
                             <arg name="success" direction="out" type="b" />
@@ -260,7 +260,7 @@ async def handler(response, results):
                 if response == 0:
                     await queue.put(urllib.parse.urlparse(results["uri"].value).path)
                 else:
-                    log.warning(f"Received non-zero response when taking screenshot: {response}")
+                    log.warning("Received non-zero response when taking screenshot: %s", str(response))
                     await queue.put(None)
 
             # Set up the signal handler

analyzer/linux/lib/common/results.py

@@ -38,7 +38,7 @@ def upload_to_host(file_path, dump_path, pids="", ppids="", metadata="", categor
                     nc.send(buf, retry=True)
                     buf = infd.read(BUFSIZE)
     except Exception as e:
-        log.error("Exception uploading file %s to host: %s", file_path, e, exc_info=True)
+        log.exception("Exception uploading file %s to host: %s", file_path, e)
     finally:
         if nc:
             nc.close()

analyzer/linux/modules/auxiliary/filecollector.py

@@ -51,7 +51,6 @@ def __init__(self, options, config):
                 self.thread.join(0.5)
 
     def run(self):
-
         if not HAVE_PYINOTIFY:
             log.info("Missed dependency: pip3 install pyinotify")
             return False

analyzer/linux/modules/auxiliary/screenshots.py

@@ -11,6 +11,7 @@
 
 if HAVE_PIL and HAVE_DBUS_NEXT:
     from PIL import Image
+
     from lib.api.screenshot import Screenshot, ScreenshotGrabber, ScreenshotsUnsupported
 
 from lib.common.abstracts import Auxiliary

analyzer/linux/modules/packages/zip.py

@@ -17,7 +17,6 @@
 
 
 class Zip(Package):
-
     real_package = None
 
     def prepare(self):

analyzer/windows/analyzer.py

@@ -691,7 +691,7 @@ def analysis_loop(self, aux_modules):
                                     try:
                                         Process(pid=pid).upload_memdump()
                                     except Exception as e:
-                                        log.error(e, exc_info=True)
+                                        log.exception(e)
                                 log.info("Process with pid %s appears to have terminated", pid)
                                 if pid in self.process_list.pids:
                                     self.process_list.remove_pid(pid)
@@ -915,7 +915,7 @@ def dump_file(self, filepath, metadata="", pids="", ppids="", category="files"):
         except (IOError, socket.error) as e:
             log.error('Unable to upload dropped file at path "%s": %s', filepath, e)
         except Exception as e:
-            log.error(e, exc_info=True)
+            log.exception(e)
 
     def delete_file(self, filepath, pid=None):
         """A file is about to removed and thus should be dumped right away."""
@@ -1508,8 +1508,7 @@ def dispatch(self, data):
                 try:
                     response = fn(arguments)
                 except Exception as e:
-                    log.error(e, exc_info=True)
-                    log.exception("Pipe command handler exception occurred (command %s args %s)", command, arguments)
+                    log.exception("Pipe command handler exception occurred (command %s args %s). %s", command, arguments, str(e))
 
         return response
 
@@ -1536,7 +1535,7 @@ def dispatch(self, data):
 
     # When user set wrong package, Example: Emotet package when submit doc, package only is for EXE!
     except CuckooError:
-        log.info("You probably submitted the job with wrong package", exc_info=True)
+        log.exception("You probably submitted the job with wrong package")
         data["status"] = "exception"
         data["description"] = "You probably submitted the job with wrong package"
         try:

analyzer/windows/lib/api/process.py

@@ -43,13 +43,13 @@
         CAPEMON64_NAME,
         LOADER32_NAME,
         LOADER64_NAME,
-        TTD32_NAME,
-        TTD64_NAME,
         LOGSERVER_PREFIX,
         PATHS,
         PIPE,
         SHUTDOWN_MUTEX,
         TERMINATE_EVENT,
+        TTD32_NAME,
+        TTD64_NAME,
     )
     from lib.common.defines import (
         KERNEL32,
@@ -601,7 +601,6 @@ def is_64bit(self):
         return False
 
     def write_monitor_config(self, interest=None, nosleepskip=False):
-
         config_path = os.path.join(Path.cwd(), "dll", f"{self.pid}.ini")
         log.info("Monitor config for %s: %s", self, config_path)
 
@@ -759,7 +758,7 @@ def upload_memdump(self):
         try:
             upload_to_host(file_path, os.path.join("memory", f"{self.pid}.dmp"), category="memory")
         except Exception as e:
-            log.error(e, exc_info=True)
+            log.exception(e)
             log.error(os.path.join("memory", f"{self.pid}.dmp"))
             log.error(file_path)
         log.info("Memory dump of %s uploaded", self)

analyzer/windows/lib/common/results.py

@@ -61,7 +61,7 @@ def upload_to_host(file_path, dump_path, pids="", ppids="", metadata="", categor
                         size -= read_size
                         buf = infd.read(BUFSIZE)
     except Exception as e:
-        log.error("Exception uploading file %s to host: %s", file_path, e, exc_info=True)
+        log.exception("Exception uploading file %s to host: %s", file_path, e)
 
 
 def upload_buffer_to_host(buffer, dump_path, filepath=False, pids="", ppids="", metadata="", category="", duplicated=False):

analyzer/windows/lib/common/zip_utils.py

@@ -50,7 +50,7 @@ def extract_archive(seven_zip_path, archive_path, extract_path, password="infect
     @param extract_path: where to extract
     @param password: archive password
     @param try_multiple_passwords: we will be splitting the password on the ':' symbol,
-           and trying each one to extract the archive
+        and trying each one to extract the archive
     """
     log.debug([seven_zip_path, "x", "-p", "-y", f"-o{extract_path}", archive_path])
     p = subprocess.run(
@@ -60,7 +60,7 @@ def extract_archive(seven_zip_path, archive_path, extract_path, password="infect
         stdout=subprocess.PIPE,
     )
     stdoutput, stderr = p.stdout, p.stderr
-    log.debug(f"{p.stdout} {p.stderr}")
+    log.debug("%s %s", p.stdout, p.stderr)
 
     if try_multiple_passwords:
         passwords = password.split(":")
@@ -84,9 +84,9 @@ def extract_archive(seven_zip_path, archive_path, extract_path, password="infect
                 stdout=subprocess.PIPE,
             )
             stdoutput, stderr = p.stdout, p.stderr
-            log.debug(f"{p.stdout} {p.stderr}")
+            log.debug("%s - %s", p.stdout, p.stderr)
             if b"Wrong password" in stderr:
-                log.debug(f"The provided password '{pword}' was incorrect")
+                log.debug("The provided password '%s' was incorrect", str(pword))
                 continue
             else:
                 # We did it!
@@ -151,7 +151,7 @@ def extract_zip(zip_path, extract_path, password=b"infected", recursion_depth=1,
     @param password: ZIP password
     @param recursion_depth: how deep we are in a nested archive
     @param try_multiple_passwords: we will be splitting the password on the ':' symbol,
-           and trying each one to extract the archive
+        and trying each one to extract the archive
     """
     # Test if zip file contains a file named as itself.
     if is_overwritten(zip_path):
@@ -195,15 +195,15 @@ def extract_zip(zip_path, extract_path, password=b"infected", recursion_depth=1,
                 raise CuckooPackageError("Invalid Zip file") from e
             except RuntimeError as e:
                 if "Bad password for file" in repr(e):
-                    log.debug(f"Password '{pword}' was unsuccessful in extracting the archive.")
+                    log.debug("Password '%s' was unsuccessful in extracting the archive.", str(pword))
                     password_fail = True
                     continue
                 else:
                     # Try twice, just for kicks
                     try:
                         archive.extractall(path=extract_path, pwd=pword)
                     except RuntimeError as e:
-                        raise CuckooPackageError(f"Unable to extract Zip file: {e}") from e
+                        raise CuckooPackageError("Unable to extract Zip file: %s", str(e)) from e
             finally:
                 if recursion_depth < 4:
                     # Extract nested archives.
@@ -227,7 +227,7 @@ def extract_zip(zip_path, extract_path, password=b"infected", recursion_depth=1,
                                 log.error("Error extracting nested Zip file %s with details: %s", name, run_err)
 
         if password_fail:
-            raise CuckooPackageError(f"Unable to extract password-protected Zip file with the password(s): {passwords}")
+            raise CuckooPackageError("Unable to extract password-protected Zip file with the password(s): %s", str(passwords))
 
 
 def is_overwritten(zip_path):
@@ -264,7 +264,7 @@ def winrar_extractor(winrar_binary, extract_path, archive_path):
         stdout=subprocess.PIPE,
     )
     # stdoutput, stderr = p.stdout, p.stderr
-    log.debug(p.stdout + p.stderr)
+    log.debug("%s - %s", p.stdout, p.stderr)
 
     return os.listdir(extract_path)
 
@@ -289,11 +289,11 @@ def upload_extracted_files(root, files_at_root):
     for entry in files_at_root:
         try:
             file_path = os.path.join(root, entry)
-            log.info("Uploading {0} to host".format(file_path))
+            log.info("Uploading %s to host", str(file_path))
             filename = f"files/{hash_file(hashlib.sha256, file_path)}"
             upload_to_host(file_path, filename, metadata=Path(entry).name, duplicated=False)
         except Exception as e:
-            log.warning(f"Couldn't upload file {Path(entry).name} to host {e}")
+            log.warning("Couldn't upload file %s to host %s", str(Path(entry).name), str(e))
 
 
 def attempt_multiple_passwords(options: dict, password: str) -> bool:

analyzer/windows/lib/core/pipe.py

@@ -224,7 +224,7 @@ def stop(self):
                 if h.is_alive():
                     h.stop()
             except Exception as e:
-                log.error(e, exc_info=True)
+                log.exception(e)
 
 
 def disconnect_pipes():

analyzer/windows/modules/auxiliary/amsi.py

@@ -30,7 +30,6 @@
 import logging
 import sys
 import threading
-import traceback
 import uuid
 
 logger = logging.getLogger(__name__)
@@ -945,7 +944,7 @@ def _unpackSimpleType(self, record, info, event_property):
 
         # if there is no data remaining then return
         if user_data_remaining <= 0:
-            logger.warning("No more user data left, returning none for field {:s}".format(name_field))
+            logger.warning("No more user data left, returning none for field %s", str(name_field))
             return {name_field: None}
 
         in_type = event_property.epi_u1.nonStructType.InType
@@ -986,7 +985,7 @@ def _unpackSimpleType(self, record, info, event_property):
 
         if status != ERROR_SUCCESS:
             # We can handle this error and still capture the data.
-            logger.warning("Failed to get data field data for {:s}, incrementing by reported size".format(name_field))
+            logger.warning("Failed to get data field data for %s, incrementing by reported size", str(name_field))
             self.index += property_length
             return {name_field: None}
 
@@ -1135,16 +1134,15 @@ def _processEvent(self, record):
             if record.contents.EventHeader.Flags & EVENT_HEADER_FLAG_EXTENDED_INFO:
                 parsed_data["EventExtendedData"] = self._parseExtendedData(record)
         except Exception as e:
-            logger.warning("Unable to parse event: {}".format(e))
+            logger.warning("Unable to parse event: %s", str(e))
 
         try:
             out.update(parsed_data)
             # Call the user's specified callback function
             if self.event_callback:
                 self.event_callback(out)
         except Exception as e:
-            logger.error("Exception during callback: {}".format(e))
-            logger.error(traceback.format_exc())
+            logger.exception("Exception during callback: %s", str(e))
 
 
 class TraceProperties:
@@ -1170,7 +1168,7 @@ def __init__(self, event_callback=None):
             raise OSError("AMSI not supported on this platform") from err
         self.provider = None
         self.properties = TraceProperties()
-        self.session_name = "{:s}".format(str(uuid.uuid4()))
+        self.session_name = str(uuid.uuid4())
         self.running = False
         self.event_callback = event_callback
         self.trace_logfile = None

analyzer/windows/modules/auxiliary/browsermonitor.py

@@ -35,7 +35,7 @@ def _find_browser_extension(self):
             for directory in temp_dir_list:
                 # TOR Browser saves directly to %temp%
                 if directory.startswith("bext_") and directory.endswith(".json"):
-                    log.debug(f"Found extension logs: {self.browser_logfile}")
+                    log.debug("Found extension logs: %s", self.browser_logfile)
                     self.browser_logfile = os.path.join(temp_dir, directory)
                     break
                 tmp_directory_path = os.path.join(temp_dir, directory)
@@ -47,7 +47,7 @@ def _find_browser_extension(self):
                 for file in tmp_dir_files:
                     if file.startswith("bext_") and file.endswith(".json"):
                         self.browser_logfile = os.path.join(temp_dir, directory, file)
-                        log.debug(f"Found extension logs: {self.browser_logfile}")
+                        log.debug("Found extension logs: %s", self.browser_logfile)
                         break
             time.sleep(1)
 

analyzer/windows/modules/auxiliary/disguise.py

@@ -248,7 +248,7 @@ def add_persistent_route(self, gateway: str):
 
     def start(self):
         if self.config.windows_static_route:
-            log.info(f"Config for route is: {str(self.config.windows_static_route)}")
+            log.info("Config for route is: %s", str(self.config.windows_static_route))
             self.add_persistent_route(self.config.windows_static_route_gateway)
         self.change_productid()
         self.set_office_mrus()