Fixed Major Issue

Re-hauled event_log.py, now made it better and more useful, as well as fixed issues with dependencies

Also updated Logicytics.py to remove deprecated code

Updated exception logging to discourage it, making only one script now using it

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

Author: DefinetlyNotAI

CODE/Logicytics.py

@@ -124,14 +124,6 @@ def handle_special_actions():
         log.info("Opening developer menu...")
         special_execute("_dev.py")
 
-    # Deprecated, remove in v3.3.0
-    if ACTION == "extra":
-        print("\033[91mDeprecationWarning: The `extra` feature has been removed! 🚫\n"
-              "Why? It didn't match our code quality standards.\n"
-              "What to use instead? Check out our new features with --help\033[0m")
-        input("Press Enter to exit...")
-        exit(0)
-
     if ACTION == "update":
         log.info("Updating...")
         message, log_type = Health.update()
@@ -162,14 +154,6 @@ def handle_special_actions():
         input("Press Enter to exit...")
         exit(0)
 
-    # Deprecated, remove in v3.3.0
-    if ACTION == "unzip_extra":
-        print("\033[91mDeprecationWarning: The `unzip_extra` feature has been removed! 🚫\n"
-              "Why? It didn't match our code quality standards.\n"
-              "What to use instead? Check out our new features with --help\033[0m")
-        input("Press Enter to exit...")
-        exit(0)
-
 
 def check_privileges():
     """
@@ -204,6 +188,7 @@ def generate_execution_list() -> list | list[str] | list[str | Any]:
     execution_list.remove("dir_list.py")
     execution_list.remove("tree.ps1")
     execution_list.remove("vulnscan.py")
+    execution_list.remove("event_log.py")
 
     if ACTION == "minimal":
         execution_list = [
@@ -237,6 +222,7 @@ def generate_execution_list() -> list | list[str] | list[str | Any]:
         execution_list.append("sensitive_data_miner.py")
         execution_list.append("dir_list.py")
         execution_list.append("tree.ps1")
+        execution_list.append("event_log.py")
         log.warning("This flag will use threading!")
 
     if ACTION == "vulnscan_ai":

CODE/_dev.py

@@ -127,7 +127,7 @@ def dev_checks() -> None:
         print("\nGreat Job! Please tick the box in the GitHub PR request for completing steps in --dev")
     except Exception as e:
         # Log any exceptions that occur during the process
-        log_dev.exception(str(e))
+        log_dev.error(f"An error occurred: {e}")
 
 
 if __name__ == "__main__":

CODE/event_log.py

@@ -1,7 +1,9 @@
 import os
+import shutil
+import threading
 from os import mkdir
 
-import win32evtlog  # TODO - Remove this dependency, find a alternative
+import wmi  # Import the wmi library
 
 from logicytics import Log, DEBUG
 
@@ -12,49 +14,64 @@
 
 
 @log.function
-def parse_event_logs(log_type: str, output_file: str, server: str = 'localhost'):
+def parse_event_logs(log_type: str, output_file: str):
     """
-    Parses Windows event logs of a specified type and writes them to an output file.
+    Parses Windows event logs of a specified type and writes them to an output file using WMI.
 
     Args:
         log_type (str): The type of event log to parse (e.g., 'Security', 'Application').
         output_file (str): The file path where the parsed event logs will be written.
-        server (str): The name of the server to connect to. Default is 'localhost'.
 
     Raises:
         Exception: If there is an error opening or reading the event log, or writing to the output file.
     """
+    log.info(f"Parsing {log_type} events (Windows Events) and writing to {output_file}, this may take a while...")
     try:
-        hand = win32evtlog.OpenEventLog(server, log_type)
-        flags = win32evtlog.EVENTLOG_BACKWARDS_READ | win32evtlog.EVENTLOG_SEQUENTIAL_READ
-        total = win32evtlog.GetNumberOfEventLogRecords(hand)
+        # Initialize WMI connection
+        c = wmi.WMI()
 
+        # Query based on log_type ('Security', 'Application', or 'System')
+        query = f"SELECT * FROM Win32_NTLogEvent WHERE Logfile = '{log_type}'"
+        log.debug(f"Executing WMI query: {query}")
+
+        # Open the output file for writing
         with open(output_file, 'w') as f:
-            f.write(f"Total records: {total}\n\n")
-            events = win32evtlog.ReadEventLog(hand, flags, 0)
-            while events:
-                for event in events:
-                    event_data = {
-                        'Event Category': event.EventCategory,
-                        'Time Generated': event.TimeGenerated.Format(),
-                        'Source Name': event.SourceName,
-                        'Event ID': event.EventID,
-                        'Event Type': event.EventType,
-                        'Event Data': event.StringInserts
-                    }
-                    f.write(str(event_data) + '\n\n')
-                events = win32evtlog.ReadEventLog(hand, flags, 0)
-
-        win32evtlog.CloseEventLog(hand)
+            events = c.query(query)
+            f.write(f"Total records: {len(events)}\n\n")
+            log.debug(f"Number of events retrieved: {len(events)}")
+            for event in events:
+                event_data = {
+                    'Event Category': event.Category,
+                    'Time Generated': event.TimeGenerated,
+                    'Source Name': event.SourceName,
+                    'Event ID': event.EventCode,
+                    'Event Type': event.Type,
+                    'Event Data': event.InsertionStrings
+                }
+                f.write(str(event_data) + '\n\n')
+
         log.info(f"{log_type} events (Windows Events) have been written to {output_file}")
+    except Exception as err:
+        log.error(f"Fatal issue: {err}")
+
+
+if __name__ == "__main__":
+    try:
+        if os.path.exists('event_logs'):
+            shutil.rmtree('event_logs')
+        mkdir('event_logs')
     except Exception as e:
         log.error(f"Fatal issue: {e}")
+        exit(1)
 
+    threads = []
+    threads_items = [('Security', 'event_logs/Security_events.txt'),
+                     ('Application', 'event_logs/App_events.txt'),
+                     ('System', 'event_logs/System_events.txt')]
 
-if __name__ == "__main__":
-    if os.path.exists('event_logs'):
-        os.rmdir('event_logs')
-    mkdir('event_logs')
-    parse_event_logs('Security', 'event_logs/Security_events.txt')
-    parse_event_logs('Application', 'event_logs/App_events.txt')
-    parse_event_logs('System', 'event_logs/System_events.txt')
+    for log_type_main, output_file_main in threads_items:
+        thread = threading.Thread(target=parse_event_logs, args=(log_type_main, output_file_main))
+        threads.append(thread)
+        thread.start()
+    for thread in threads:
+        thread.join()

CODE/log_miner.py

@@ -35,7 +35,7 @@ def backup_windows_logs():
         stdout, stderr = process.communicate(input=cmd)
 
         if process.returncode != 0:
-            log.exception(f"Failed to backup logs: {stderr.strip()}", ChildProcessError)
+            log.error(f"Failed to backup logs: {stderr.strip()}")
 
         log.info(f"Windows logs backed up to {backup_file}")
     except Exception as e:

CODE/logicytics/Logger.py

@@ -233,6 +233,9 @@ def exception(self, message, exception_type: Type = Exception):
         """
         Logs an exception message.
 
+        This is not recommended for use in production code, as it raises the exception after logging it.
+        Use Log().error() instead.
+
         :param message: The exception message to be logged.
         :param exception_type: The type of exception to raise.
         """

requirements.txt

@@ -19,4 +19,5 @@ safetensors~=0.4.5
 prettytable~=3.12.0
 pandas~=2.2.2
 scapy~=2.5.0
-sentence-transformers~=3.3.1
+sentence-transformers~=3.3.1
+WMI~=1.5.1