Fixed major bugs that affect Logicytics (performance checker)

.gitignore

@@ -322,3 +322,9 @@ $RECYCLE.BIN/
 /CODE/VulnScan/tools/NN features/
 /CODE/logicytics/User_History.json.gz
 /CODE/logicytics/User_History.json
+/CODE/SysInternal_Suite/psfile.exe
+/CODE/SysInternal_Suite/PsGetsid.exe
+/CODE/SysInternal_Suite/PsInfo.exe
+/CODE/SysInternal_Suite/pslist.exe
+/CODE/SysInternal_Suite/PsLoggedon.exe
+/CODE/SysInternal_Suite/psloglist.exe

CODE/Logicytics.py

@@ -12,12 +12,12 @@
 import psutil
 from prettytable import PrettyTable
 
-from logicytics import Log, Execute, Check, Get, FileManagement, Flag, DEBUG, DELETE_LOGS, CONFIG
+from logicytics import Log, Execute, Check, Get, FileManagement, Flag, DEBUG, DELETE_LOGS, config
 
 # Initialization
 log = Log({"log_level": DEBUG, "delete_log": DELETE_LOGS})
 ACTION, SUB_ACTION = None, None
-MAX_WORKERS = CONFIG.getint("Settings", "max_workers", fallback=min(32, (os.cpu_count() or 1) + 4))
+MAX_WORKERS = config.getint("Settings", "max_workers", fallback=min(32, (os.cpu_count() or 1) + 4))
 log.debug(f"MAX_WORKERS: {MAX_WORKERS}")
 
 
@@ -67,7 +67,8 @@ def __generate_execution_list(self) -> list[str]:
             - Warns users about potential long execution times for certain actions
         """
         execution_list = Get.list_of_files(".", only_extensions=(".py", ".exe", ".ps1", ".bat"),
-                                           exclude_files=["Logicytics.py"])
+                                           exclude_files=["Logicytics.py"],
+                                           exclude_dirs=["logicytics", "SysInternal_Suite"])
         files_to_remove = {
             "sensitive_data_miner.py",
             "dir_list.py",
@@ -101,7 +102,8 @@ def __generate_execution_list(self) -> list[str]:
         elif ACTION == "modded":
             # Add all files in MODS to execution list
             execution_list = Get.list_of_files("../MODS", only_extensions=(".py", ".exe", ".ps1", ".bat"),
-                                               append_file_list=execution_list, exclude_files=["Logicytics.py"])
+                                               append_file_list=execution_list, exclude_files=["Logicytics.py"],
+                                               exclude_dirs=["logicytics", "SysInternal_Suite"])
 
         elif ACTION == "depth":
             log.warning(
@@ -128,6 +130,7 @@ def __generate_execution_list(self) -> list[str]:
             log.critical("Nothing is in the execution list.. This is due to faulty code or corrupted Logicytics files!")
             exit(1)
 
+        log.debug(f"Execution list length: {len(execution_list)}")
         log.debug(f"The following will be executed: {execution_list}")
         return execution_list
 
@@ -208,18 +211,32 @@ def __performance(self):
             end_time = datetime.now()
             end_memory = process.memory_full_info().uss / 1024 / 1024  # MB
             elapsed_time = end_time - start_time
-            memory_delta = end_memory - start_memory
-            memory_usage.append((self.execution_list[file], str(memory_delta)))
+            memory_delta = max(0, end_memory - start_memory)  # Clamps negative delta to 0
+            memory_usage.append((self.execution_list[file], f"{memory_delta}"))
             execution_times.append((self.execution_list[file], elapsed_time))
             log.info(f"{self.execution_list[file]} executed in {elapsed_time}")
-            log.info(f"{self.execution_list[file]} used {memory_delta:.2f}MB of memory")
-            log.debug(f"Started with {start_memory}MB of memory and ended with {end_memory}MB of memory")
+            try:
+                if (end_memory - start_memory) < 0:
+                    log.info(
+                        f"{self.execution_list[file]} used {memory_delta:.3f}MB of memory - \033[33mPossible Affected by outside processes\033[0m")
+                else:
+                    log.info(f"{self.execution_list[file]} used {memory_delta:.3f}MB of memory")
+            except Exception as e:
+                log.warning("Failed to log memory usage delta, reason: " + str(e))
+            log.debug(f"Started with {start_memory:.3f}MB of memory and ended with {end_memory:.3f}MB of memory")
 
         table = PrettyTable()
         table.field_names = ["Script", "Execution Time", "Memory Usage (MB)"]
         for script, elapsed_time in execution_times:
-            memory = next(m[1] for m in memory_usage if m[0] == script)
-            table.add_row([script, elapsed_time, f"{memory:.2f}"])
+            try:
+                memory = f"{float(next(m[1] for m in memory_usage if m[0] == script)):.3f}"
+            except StopIteration:
+                log.warning(f"No memory data found for {script}")
+                memory = "N/A"
+            except Exception as e:
+                log.warning(f"Failed to log memory usage for {script}, reason: " + str(e))
+                memory = "N/A"
+            table.add_row([script, elapsed_time, f"{memory}"])
 
         try:
             with open(
@@ -228,9 +245,9 @@ def __performance(self):
             ) as f:
                 f.write(table.get_string())
                 f.write(
-                    "\nSome values may be negative, Reason may be due to external resources playing with memory usage, "
-                    "close background tasks to get more accurate readings")
-                f.write("Note: This is not a low-level memory logger, data here isn't 100% accurate!")
+                    "\nSome values may be negative, Reason may be due to external resources playing with memory usage,\n"
+                    "Close background tasks to get more accurate readings\n\n")
+                f.write("Note: This is not a low-level memory logger, data here isn't 100% accurate!\n")
             log.info("Performance check complete! Performance log found in ACCESS/LOGS/PERFORMANCE")
         except Exception as e:
             log.error(f"Error writing performance log: {e}")
@@ -434,10 +451,16 @@ def check_privileges():
         log.warning("UAC is enabled, this may cause issues - Please disable UAC if possible")
 
 
-def zip_generated_files():
-    """Zips generated files based on the action."""
+class ZIP:
+    @classmethod
+    def files(cls):
+        """Zips generated files based on the action."""
+        if ACTION == "modded":
+            cls.__and_log("..\\MODS", "MODS")
+        cls.__and_log(".", "CODE")
 
-    def zip_and_log(directory: str, name: str):
+    @staticmethod
+    def __and_log(directory: str, name: str):
         log.debug(f"Zipping directory '{directory}' with name '{name}' under action '{ACTION}'")
         zip_values = FileManagement.Zip.and_hash(
             directory,
@@ -451,10 +474,6 @@ def zip_and_log(directory: str, name: str):
             log.info(zip_loc)
             log.debug(hash_loc)
 
-    if ACTION == "modded":
-        zip_and_log("..\\MODS", "MODS")
-    zip_and_log(".", "CODE")
-
 
 def handle_sub_action():
     """
@@ -472,8 +491,7 @@ def handle_sub_action():
     elif SUB_ACTION == "reboot":
         subprocess.call("shutdown /r /t 3", shell=False)
     # elif sub_action == "webhook":
-    # Implement this in future
-    # log.warning("This feature is not implemented yet! Sorry")
+    # TODO: Implement this in future v3.5
 
 
 @log.function
@@ -501,7 +519,7 @@ def Logicytics():
     # Execute scripts
     ExecuteScript().handler()
     # Zip generated files
-    zip_generated_files()
+    ZIP.files()
     # Finish with sub actions
     handle_sub_action()
     # Finish

CODE/config.ini

@@ -56,14 +56,45 @@ model_debug = false
 
 ###################################################
 
+[DumpMemory Settings]
+# If the file size generated exceeds this limit,
+# the file will be truncated with a message
+# Put 0 to disable the limit - Limit is in MiB - int
+file_size_limit = 0
+# Safety margin to check, it multiplies with the size limit
+# This makes sure that after the file is created, there is still
+# disk space left for other tasks,
+# Make sure its above 1 or else it will fail
+# Put 1 to disable the limit - Limit is in MiB - float
+file_size_safety = 1.5
+
+###################################################
+
+[NetWorkPsutil Settings]
+# Total time this will take will be `sample_count * interval`
+
+# Number of samples to take for feature `measure network bandwidth usage`
+# This is an integer, and should be 1 and above
+sample_count = 5
+# Time between samples in seconds for feature `measure network bandwidth usage`
+# This is a float, and should be above 0
+interval = 1.5
+
+###################################################
+
 [PacketSniffer Settings]
 # The interface to sniff packets on, keep it as WiFi for most cases
 # Autocorrects between WiFi and Wi-Fi
 interface = WiFi
 # The number of packets to sniff,
-packet_count = 10000
-# The time to timeout the sniffing process
+# Must be greater than or equal to 1 - int
+packet_count = 5000
+# The time to timeout the sniffing process only,
+# Must be greater than or equal to 5 - int
 timeout = 10
+# The maximum retry time for the whole process,
+# Must be greater than or equal to 10 and timeout - int
+max_retry_time = 30
 
 ###################################################
 

CODE/dump_memory.py

@@ -5,16 +5,17 @@
 
 import psutil
 
-from logicytics import log
+from logicytics import log, config
 
 # TODO v3.4.1
 #  psutil.virtual_memory(): used, free, percent, total
 #  psutil.swap_memory(): used, free, percent, total
 
-# If the file size exceeds this limit, the file will be truncated with a message
-# Put 0 to disable the limit
-# TODO v3.4.1: Make this take from config.ini
-LIMIT_FILE_SIZE = 20  # Always in MiB
+LIMIT_FILE_SIZE = config.getint("DumpMemory Settings", "file_size_limit")  # Always in MiB
+SAFETY_MARGIN = config.getfloat("DumpMemory Settings", "file_size_safety")  # Always in MiB
+if SAFETY_MARGIN < 1:
+    log.critical("Invalid Safety Margin Inputted - Cannot proceed with dump memory")
+    exit(1)
 
 
 # Capture RAM Snapshot
@@ -41,7 +42,7 @@ def capture_ram_snapshot():
     log.info("Capturing RAM Snapshot...")
     memory = psutil.virtual_memory()
     swap = psutil.swap_memory()
-    with open("Ram_Snapshot.txt", "w") as file:
+    with open("memory_dumps/Ram_Snapshot.txt", "w") as file:
         try:
             file.write(f"Total RAM: {memory.total / (1024 ** 3):.2f} GB\n")
             file.write(f"Used RAM: {memory.used / (1024 ** 3):.2f} GB\n")
@@ -91,13 +92,13 @@ def gather_system_info():
     except Exception as e:
         log.error(f"Error gathering system information: {e}")
         sys_info = {'Error': 'Failed to gather system information'}
-    with open("SystemRam_Info.txt", "w") as file:
+    with open("memory_dumps/SystemRam_Info.txt", "w") as file:
         for key, value in sys_info.items():
             file.write(f"{key}: {value}\n")
     log.info("System Information saved to SystemRam_Info.txt")
 
 
-# Memory Dump (Windows-specific, using psutil)
+# Memory Dump
 def memory_dump():
     """
     Perform a memory dump of the current process, capturing detailed metadata for each readable memory region.
@@ -127,12 +128,12 @@ def memory_dump():
 
     try:
         process = psutil.Process(pid)
-        with open("Ram_Dump.txt", "wb") as dump_file:
+        with open("memory_dumps/Ram_Dump.txt", "wb") as dump_file:
             total_size = 0
             for mem_region in process.memory_maps(grouped=False):
                 # Check available disk space
                 if os.path.exists("Ram_Dump.txt"):
-                    required_space = LIMIT_FILE_SIZE * 1024 * 1024 * 1.5  # 2x safety margin
+                    required_space = LIMIT_FILE_SIZE * 1024 * 1024 * SAFETY_MARGIN  # 2x safety margin
                     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")
@@ -206,6 +207,7 @@ def main():
     No parameters.
     No return value.
     """
+    os.makedirs("memory_dumps", exist_ok=True)
     log.info("Starting system memory collection tasks...")
     capture_ram_snapshot()
     gather_system_info()

CODE/logicytics/Config.py

@@ -15,7 +15,7 @@ def __config_data() -> tuple[str, str, list[str], bool, str]:
             - Version (str): System version from configuration
             - Files (list[str]): List of files specified in configuration
             - Delete old logs (bool): Flag indicating whether to delete old log files
-            - CONFIG itself
+            - config itself
 
     Raises:
         SystemExit: If the 'config.ini' file cannot be found in any of the attempted locations
@@ -30,21 +30,21 @@ def _config_path() -> str:
             print("The config.ini file is not found in the expected location.")
             exit(1)
 
-    config = configparser.ConfigParser()
+    config_local = configparser.ConfigParser()
     path = _config_path()
-    config.read(path)
+    config_local.read(path)
 
-    log_using_debug = config.getboolean("Settings", "log_using_debug")
-    delete_old_logs = config.getboolean("Settings", "delete_old_logs")
-    version = config.get("System Settings", "version")
-    files = config.get("System Settings", "files").split(", ")
+    log_using_debug = config_local.getboolean("Settings", "log_using_debug")
+    delete_old_logs = config_local.getboolean("Settings", "delete_old_logs")
+    version = config_local.get("System Settings", "version")
+    files = config_local.get("System Settings", "files").split(", ")
 
     log_using_debug = "DEBUG" if log_using_debug else "INFO"
 
-    return log_using_debug, version, files, delete_old_logs, config
+    return log_using_debug, version, files, delete_old_logs, config_local
 
 
 # Check if the script is being run directly, if not, set up the library
 if __name__ == '__main__':
     exit("This is a library, Please import rather than directly run.")
-DEBUG, VERSION, CURRENT_FILES, DELETE_LOGS, CONFIG = __config_data()
+DEBUG, VERSION, CURRENT_FILES, DELETE_LOGS, config = __config_data()

CODE/logicytics/Flag.py

@@ -8,23 +8,23 @@
 from collections import Counter
 from datetime import datetime
 
-from .Config import CONFIG
+from logicytics.Config import config
 
 # Check if the script is being run directly, if not, set up the library
 if __name__ == '__main__':
     exit("This is a library, Please import rather than directly run.")
 else:
     # Save user preferences?
-    SAVE_PREFERENCES = CONFIG.getboolean("Settings", "save_preferences")
+    SAVE_PREFERENCES = config.getboolean("Settings", "save_preferences")
     # Debug mode for Sentence Transformer
-    DEBUG_MODE = CONFIG.getboolean("Flag Settings", "model_debug")  # Debug mode for Sentence Transformer
+    DEBUG_MODE = config.getboolean("Flag Settings", "model_debug")  # Debug mode for Sentence Transformer
     # File for storing user history data
     HISTORY_FILE = os.path.join(os.path.dirname(os.path.abspath(__file__)), 'User_History.json.gz')  # User history file
     # Minimum accuracy threshold for flag suggestions
     MIN_ACCURACY_THRESHOLD = float(
-        CONFIG.get("Flag Settings", "accuracy_min"))  # Minimum accuracy threshold for flag suggestions
-    if not 0 <= MIN_ACCURACY_THRESHOLD <= 100:
-        raise ValueError("accuracy_min must be between 0 and 100")
+        config.get("Flag Settings", "accuracy_min"))  # Minimum accuracy threshold for flag suggestions
+    if not 1 <= MIN_ACCURACY_THRESHOLD <= 99:
+        raise ValueError("accuracy_min must be between 1 and 99")
 
 
 class _Match:
@@ -61,11 +61,11 @@ def __get_sim(user_input: str, all_descriptions: list[str]) -> list[float]:
             logging.getLogger("sentence_transformers").setLevel(logging.ERROR)
 
         try:
-            MODEL = SentenceTransformer(CONFIG.get("Flag Settings", "model_to_use"))
+            MODEL = SentenceTransformer(config.get("Flag Settings", "model_to_use"))
         except Exception as e:
             print(f"Error: {e}")
             print("Please check the model name in the config file.")
-            print(f"Model name {CONFIG.get('Flag Settings', 'model_to_use')} may not be valid.")
+            print(f"Model name {config.get('Flag Settings', 'model_to_use')} may not be valid.")
             exit(1)
 
         user_embedding = MODEL.encode(user_input, convert_to_tensor=True, show_progress_bar=DEBUG_MODE)