Fixed issues caused by coderabbit

Also improved some things,
Improved Logicytics.py threading
Upgraded Deprecation warning with stack tracing, and added the option to remove it.

Signed-off-by: Shahm Najeeb <[email protected]>

Author: DefinetlyNotAI

CODE/Logicytics.py

@@ -3,8 +3,8 @@
 import os
 import shutil
 import subprocess
-import threading
 import zipfile
+from concurrent.futures import ThreadPoolExecutor, as_completed
 from datetime import datetime
 from typing import Any
 
@@ -225,9 +225,6 @@ def generate_execution_list() -> list | list[str] | list[str | Any]:
     - 'depth': Comprehensive script execution with data mining and logging scripts
     - 'vulnscan_ai': Vulnerability scanning script only
     
-    Parameters:
-        None
-    
     Returns:
         list[str]: A list of script file paths to be executed, filtered and modified based on the current action.
     
@@ -293,34 +290,42 @@ def execute_scripts():
     """Executes the scripts in the execution list based on the action."""
     # Check weather to use threading or not, as well as execute code
     log.info("Starting Logicytics...")
+
     if ACTION == "threaded" or ACTION == "depth":
-        def threaded_execution(execution_list_thread, index_thread):
-            log.debug(f"Thread {index_thread} started")
+
+        def execute_single_script(script: str) -> tuple[str, Exception | None]:
+            """
+            Executes a single script and logs the result.
+
+            This function executes a single script and logs the result,
+            capturing any exceptions that occur during execution
+
+            Parameters:
+                script (str): The path to the script to be executed
+            """
+            log.debug(f"Executing {script}")
             try:
-                log.parse_execution(Execute.script(execution_list_thread[index_thread]))
-                log.info(f"{execution_list_thread[index_thread]} executed")
-            except UnicodeDecodeError as err:
-                log.error(f"Error in thread: {err}")
+                log.parse_execution(Execute.script(script))
+                log.info(f"{script} executed")
+                return script, None
             except Exception as err:
-                log.error(f"Error in thread: {err}")
-            log.debug(f"Thread {index_thread} finished")
+                log.error(f"Error executing {script}: {err}")
+                return script, err
 
         log.debug("Using threading")
-        threads = []
         execution_list = generate_execution_list()
-        for index, _ in enumerate(execution_list):
-            thread = threading.Thread(
-                target=threaded_execution,
-                args=(
-                    execution_list,
-                    index,
-                ),
-            )
-            threads.append(thread)
-            thread.start()
-
-        for thread in threads:
-            thread.join()
+        with ThreadPoolExecutor() as executor:
+            futures = {executor.submit(execute_single_script, script): script
+                       for script in execution_list}
+
+            for future in as_completed(futures):
+                script = futures[future]
+                result, error = future.result()
+                if error:
+                    log.error(f"Failed to execute {script}")
+                else:
+                    log.debug(f"Completed {script}")
+
     elif ACTION == "performance_check":
         execution_times = []
         execution_list = generate_execution_list()

CODE/_debug.py

@@ -185,9 +185,6 @@ def python_version():
     against predefined minimum and maximum version thresholds. It provides informative logging about
     the Python version status.
     
-    Parameters:
-        None
-    
     Logs:
         - Info: When Python version is within the recommended range (3.11.x to 3.12.x)
         - Warning: When Python version is below the minimum recommended version (< 3.11)

CODE/bluetooth_details.py

@@ -61,7 +61,7 @@ def _query_bluetooth_devices() -> bool | list[dict[str, str]]:
                                 capture_output=True, text=True, check=True)
         devices = json.loads(result.stdout)
     except subprocess.CalledProcessError as e:
-        log.error(f"Failed to query Bluetooth devices: {e}")
+        log.error(f"Failed to query Bluetooth devices with command '{command}': {e}")
         return False
     except json.JSONDecodeError as e:
         log.error(f"Failed to parse device information: {e}")
@@ -72,13 +72,14 @@ def _query_bluetooth_devices() -> bool | list[dict[str, str]]:
 
     device_info_list = []
     for device in devices:
+        FALLBACK_MSG = 'Unknown (Fallback due to failed Get request)'
         device_info = {
-            'Name': device.get('FriendlyName', 'Unknown (Fallback due to failed Get request)'),
-            'Device ID': device.get('DeviceID', 'Unknown (Fallback due to failed Get request)'),
-            'Description': device.get('Description', 'Unknown (Fallback due to failed Get request)'),
-            'Manufacturer': device.get('Manufacturer', 'Unknown (Fallback due to failed Get request)'),
-            'Status': device.get('Status', 'Unknown (Fallback due to failed Get request)'),
-            'PNP Device ID': device.get('PnpDeviceID', 'Unknown (Fallback due to failed Get request)')
+            'Name': device.get('FriendlyName', FALLBACK_MSG),
+            'Device ID': device.get('DeviceID', FALLBACK_MSG),
+            'Description': device.get('Description', FALLBACK_MSG),
+            'Manufacturer': device.get('Manufacturer', FALLBACK_MSG),
+            'Status': device.get('Status', FALLBACK_MSG),
+            'PNP Device ID': device.get('PnpDeviceID', FALLBACK_MSG)
         }
         log.debug(f"Retrieved device: {device_info['Name']}")
         device_info_list.append(device_info)

CODE/bluetooth_logger.py

@@ -33,8 +33,8 @@ def save_to_file(filename, section_title, data):
             file.write(f"\n{'=' * 50}\n{section_title}\n{'=' * 50}\n")
             file.write(f"{data}\n" if isinstance(data, str) else "\n".join(data) + "\n")
             file.write(f"{'=' * 50}\n")
-    except Exception as e:
-        log.error(f"Error writing to file {filename}: {e}")
+    except Exception as err:
+        log.error(f"Error writing to file {filename}: {err}")
 
 
 # Utility function to run PowerShell commands
@@ -64,8 +64,8 @@ def run_powershell_command(command):
             log.error(f"PowerShell command failed with return code {result.returncode}")
             return []
         return result.stdout.splitlines()
-    except Exception as e:
-        log.error(f"Error running PowerShell command: {e}")
+    except Exception as err:
+        log.error(f"Error running PowerShell command: {err}")
         return []
 
 
@@ -99,8 +99,8 @@ def parse_output(lines, regex, group_names):
             else:
                 log.debug(f"Skipping unrecognized line: {line}")
         return results
-    except Exception as e:
-        log.error(f"Parsing output failed: {e}")
+    except Exception as err:
+        log.error(f"Parsing output failed: {err}")
 
 
 # Function to get paired Bluetooth devices
@@ -177,7 +177,7 @@ def log_bluetooth():
     log.debug(f"{section_title}: {paired_devices}")
 
     # Collect and log event logs
-    def collect_logs(title, command):
+    def collect_logs(title: str, command: str):
         """
         Collects and logs event logs by executing a PowerShell command and saving the results.
         
@@ -215,4 +215,7 @@ def collect_logs(title, command):
 
 
 if __name__ == "__main__":
-    log_bluetooth()
+    try:
+        log_bluetooth()
+    except Exception as e:
+        log.error(f"Failed to log Bluetooth data: {e}")

CODE/dir_list.py

@@ -27,7 +27,8 @@ def run_command_threaded(directory: str, file: str, message: str, encoding: str
     """
     log.info(f"Executing {message} for {directory}")
     try:
-        command = f"powershell -Command Get-ChildItem {directory} -Recurse"
+        safe_directory = directory.replace('"', '`"')  # Escape quotes
+        command = f'powershell -NoProfile -Command "Get-ChildItem \\""{safe_directory}\\"" -Recurse"'
         output = Execute.command(command)
         open(file, "a", encoding=encoding).write(output)
         log.info(f"{message} Successful for {directory} - {file}")
@@ -56,7 +57,7 @@ def command_threaded(base_directory: str, file: str, message: str, encoding: str
         - Handles potential errors during thread execution
     """
     try:
-        with ThreadPoolExecutor() as executor:
+        with ThreadPoolExecutor(max_workers=min(32, os.cpu_count() * 4)) as executor:
             subdirectories = [os.path.join(base_directory, d) for d in os.listdir(base_directory) if
                               os.path.isdir(os.path.join(base_directory, d))]
             futures = [executor.submit(run_command_threaded, subdir, file, message, encoding) for subdir in

CODE/dump_memory.py

@@ -1,6 +1,7 @@
 import os
 import platform
 import struct
+from datetime import datetime
 
 import psutil
 
@@ -15,6 +16,7 @@
 
 # If the file size exceeds this limit, the file will be truncated with a message
 # Put 0 to disable the limit
+# TODO v3.3.1: Make this take from config.ini
 LIMIT_FILE_SIZE = 20  # Always in MiB
 
 
@@ -85,6 +87,9 @@ def gather_system_info():
             'Machine': platform.machine(),
             'Processor': platform.processor(),
             'Page Size (bytes)': struct.calcsize("P"),
+            'CPU Count': psutil.cpu_count(),
+            'CPU Frequency': psutil.cpu_freq().current if psutil.cpu_freq() else 'N/A',
+            'Boot Time': datetime.fromtimestamp(psutil.boot_time()).strftime('%Y-%m-%d %H:%M:%S'),
         }
     except Exception as e:
         log.error(f"Error gathering system information: {e}")
@@ -134,7 +139,7 @@ def memory_dump():
                     free_space = psutil.disk_usage(".").free
                     if free_space < required_space:
                         log.error(f"Not enough disk space. Need {required_space / 1024 / 1024:.2f}MB")
-                    return
+                        return
 
                 # Check if the memory region is readable ('r' permission)
                 if 'r' in mem_region.perms:
@@ -161,9 +166,7 @@ def memory_dump():
                         if mem_region.path:
                             # Try to get device and inode-like info
                             file_path = mem_region.path
-                            # file_device = win32api.GetFileAttributes(file_path) if os.path.exists(file_path) else 'N/A'
                             region_metadata['   File Path'] = file_path
-                            # region_metadata['File Device'] = file_device
 
                     except Exception as e:
                         log.error(f"Error adding extra file information: {str(e)}")

CODE/event_log.py

@@ -79,7 +79,10 @@ def parse_event_logs(log_type: str, output_file: str):
 
     for log_type_main, output_file_main in threads_items:
         thread = threading.Thread(target=parse_event_logs, args=(log_type_main, output_file_main))
+        thread.daemon = True  # Don't hang if main thread exits
         threads.append(thread)
         thread.start()
     for thread in threads:
-        thread.join()
+        thread.join(timeout=600)  # Wait max 10 minutes per thread
+        if thread.is_alive():
+            log.error(f"Thread for {thread.name} timed out (10 minutes)!")

CODE/log_miner.py

@@ -17,9 +17,6 @@ def backup_windows_logs():
     The function handles potential errors during log backup and logs the operation's outcome.
     If the backup fails, an error message is logged without raising an exception.
     
-    Args:
-        None
-    
     Returns:
         None
     

CODE/logicytics/Flag.py

@@ -376,9 +376,16 @@ def flag(cls, user_input: str, flags: list[str], flag_description: list[str]) ->
 class Flag:
     @classmethod
     def __colorify(cls, text: str, color: str) -> str:
-        The existing docstring is comprehensive and follows good documentation practices. It clearly explains the function's purpose, parameters, return value, and provides context about the color codes. Therefore:
-        
-        KEEP_EXISTING
+        """
+        Colorize text with ANSI color codes.
+
+        Args:
+            text (str): The text to colorize
+            color (str): The color code ('y' for yellow, 'r' for red, 'b' for blue)
+
+        Returns:
+            str: The colorized text with ANSI escape codes
+        """
         colors = {
             "y": "\033[93m",
             "r": "\033[91m",
@@ -395,10 +402,7 @@ def __available_arguments(cls) -> tuple[argparse.Namespace, argparse.ArgumentPar
         This method creates an ArgumentParser with a comprehensive set of flags for customizing the application's behavior. It supports various execution modes, debugging options, system management flags, and post-execution actions.
         
         The method handles argument parsing, provides helpful descriptions for each flag, and includes color-coded hints for user guidance. It also supports suggesting valid flags if an unknown flag is provided.
-        
-        Parameters:
-            cls (type): The class context in which the method is called.
-        
+
         Returns:
             tuple[argparse.Namespace, argparse.ArgumentParser]: A tuple containing:
                 - Parsed command-line arguments (Namespace)
@@ -701,10 +705,7 @@ def data(cls) -> tuple[str, str | None]:
         - Special flags are handled with specific logic
         - No invalid flag combinations are permitted
         - User history is optionally updated based on preferences
-        
-        Parameters:
-            cls (type): The class context for accessing class methods
-        
+
         Returns:
             tuple[str, str | None]: A tuple containing:
                 - The primary matched flag

CODE/logicytics/Get.py

@@ -7,29 +7,29 @@
 
 class Get:
     @staticmethod
-    def list_of_files(directory: str, extensions: tuple | bool = True, append_file_list: list = None,
-                      exclude_files: list = None) -> list:
+    def list_of_files(directory: str, extensions: tuple | bool = True, append_file_list: list[str] = None,
+                      exclude_files: list[str] = None) -> list:
         """
-                      Retrieves a list of files in the specified directory based on given extensions and exclusion criteria.
+        Retrieves a list of files in the specified directory based on given extensions and exclusion criteria.
                       
-                      Supports two modes of file retrieval:
-                      1. When `extensions` is `True`, retrieves all files recursively from the directory.
-                      2. When `extensions` is a tuple, retrieves files matching specific extensions while applying exclusion rules.
+        Supports two modes of file retrieval:
+            1. When `extensions` is `True`, retrieves all files recursively from the directory.
+            2. When `extensions` is a tuple, retrieves files matching specific extensions while applying exclusion rules.
                       
-                      Parameters:
-                          directory (str): Path of the directory to search for files.
-                          extensions (tuple | bool, optional): File extensions to filter or True to retrieve all files. Defaults to True.
-                          append_file_list (list, optional): Existing list to append found filenames to. Creates a new list if not provided. Defaults to None.
-                          exclude_files (list, optional): List of filenames to exclude from results. Defaults to None.
+        Parameters:
+            directory (str): Path of the directory to search for files.
+            extensions (tuple | bool, optional): File extensions to filter or True to retrieve all files. Defaults to True.
+            append_file_list (list, optional): Existing list to append found filenames to. Creates a new list if not provided. Defaults to None.
+            exclude_files (list, optional): List of filenames to exclude from results. Defaults to None.
                       
-                      Returns:
-                          list: A list of filenames matching the specified criteria.
+        Returns:
+            list: A list of filenames matching the specified criteria.
                       
-                      Exclusion rules:
-                          - Ignores files starting with an underscore (_)
-                          - Excludes "Logicytics.py"
-                          - Skips files specified in `exclude_files`
-                      """
+        Exclusion rules:
+            - Ignores files starting with an underscore (_)
+            - Excludes "Logicytics.py"
+            - Skips files specified in `exclude_files`
+        """
         append_file_list = [] if not append_file_list else append_file_list
 
         if isinstance(extensions, bool) and extensions:
@@ -44,7 +44,7 @@ def list_of_files(directory: str, extensions: tuple | bool = True, append_file_l
                     filename.endswith(extensions)
                     and not filename.startswith("_")
                     and filename != "Logicytics.py"
-                    and filename not in exclude_files
+                    and (exclude_files is None or filename not in exclude_files)
             ):
                 append_file_list.append(filename)
         return append_file_list