DB cleanup script + docs

2025-12-07 09:36:05 -08:00 · 2024-12-23 11:28:46 +11:00
parent 7248e73e03
commit 466aa62a02
4 changed files with 152 additions and 3 deletions
--- a/scripts/db_cleanup/db_cleanup.py
+++ b/scripts/db_cleanup/db_cleanup.py
@@ -0,0 +1,111 @@
+#!/usr/bin/env python3
+import subprocess
+import sys
+
+def run_sqlite_command(command):
+    full_command = f"sudo docker exec -i netalertx sqlite3 /app/db/app.db \"{command}\""
+    try:
+        result = subprocess.run(full_command, shell=True, text=True, capture_output=True)
+        if result.stderr:
+            print(f"Error: {result.stderr}")
+        return result.stdout
+    except subprocess.CalledProcessError as e:
+        print(f"Error executing command: {e}")
+        return None
+
+def check_and_clean_device():
+    while True:
+        print("\nDevice Cleanup Tool")
+        print("1. Check/Clean by MAC address")
+        print("2. Check/Clean by IP address")
+        print("3. Exit")
+        
+        choice = input("\nSelect option (1-3): ")
+        
+        if choice == "1":
+            mac = input("Enter MAC address (format: xx:xx:xx:xx:xx:xx): ").lower()
+            
+            # Check all tables for MAC
+            tables_checks = [
+                f"SELECT 'Events' as source, * FROM Events WHERE eve_MAC='{mac}'",
+                f"SELECT 'Devices' as source, * FROM Devices WHERE dev_MAC='{mac}'",
+                f"SELECT 'CurrentScan' as source, * FROM CurrentScan WHERE cur_MAC='{mac}'",
+                f"SELECT 'Notifications' as source, * FROM Notifications WHERE JSON LIKE '%{mac}%'",
+                f"SELECT 'AppEvents' as source, * FROM AppEvents WHERE ObjectPrimaryID LIKE '%{mac}%' OR ObjectSecondaryID LIKE '%{mac}%'",
+                f"SELECT 'Plugins_Objects' as source, * FROM Plugins_Objects WHERE Object_PrimaryID LIKE '%{mac}%'"
+            ]
+            
+            found = False
+            for check in tables_checks:
+                result = run_sqlite_command(check)
+                if result and result.strip():
+                    found = True
+                    print(f"\nFound entries:\n{result}")
+            
+            if found:
+                confirm = input("\nWould you like to clean these entries? (y/n): ")
+                if confirm.lower() == 'y':
+                    # Delete from all tables
+                    deletes = [
+                        f"DELETE FROM Events WHERE eve_MAC='{mac}'",
+                        f"DELETE FROM Devices WHERE dev_MAC='{mac}'",
+                        f"DELETE FROM CurrentScan WHERE cur_MAC='{mac}'",
+                        f"DELETE FROM Notifications WHERE JSON LIKE '%{mac}%'",
+                        f"DELETE FROM AppEvents WHERE ObjectPrimaryID LIKE '%{mac}%' OR ObjectSecondaryID LIKE '%{mac}%'",
+                        f"DELETE FROM Plugins_Objects WHERE Object_PrimaryID LIKE '%{mac}%'"
+                    ]
+                    
+                    for delete in deletes:
+                        run_sqlite_command(delete)
+                    print("Cleanup completed!")
+            else:
+                print("\nNo entries found for this MAC address")
+                
+        elif choice == "2":
+            ip = input("Enter IP address (format: xxx.xxx.xxx.xxx): ")
+            
+            # Check all tables for IP
+            tables_checks = [
+                f"SELECT 'Events' as source, * FROM Events WHERE eve_IP='{ip}'",
+                f"SELECT 'Devices' as source, * FROM Devices WHERE dev_LastIP='{ip}'",
+                f"SELECT 'CurrentScan' as source, * FROM CurrentScan WHERE cur_IP='{ip}'",
+                f"SELECT 'Notifications' as source, * FROM Notifications WHERE JSON LIKE '%{ip}%'",
+                f"SELECT 'AppEvents' as source, * FROM AppEvents WHERE ObjectSecondaryID LIKE '%{ip}%'",
+                f"SELECT 'Plugins_Objects' as source, * FROM Plugins_Objects WHERE Object_SecondaryID LIKE '%{ip}%'"
+            ]
+            
+            found = False
+            for check in tables_checks:
+                result = run_sqlite_command(check)
+                if result and result.strip():
+                    found = True
+                    print(f"\nFound entries:\n{result}")
+            
+            if found:
+                confirm = input("\nWould you like to clean these entries? (y/n): ")
+                if confirm.lower() == 'y':
+                    # Delete from all tables
+                    deletes = [
+                        f"DELETE FROM Events WHERE eve_IP='{ip}'",
+                        f"DELETE FROM Devices WHERE dev_LastIP='{ip}'",
+                        f"DELETE FROM CurrentScan WHERE cur_IP='{ip}'",
+                        f"DELETE FROM Notifications WHERE JSON LIKE '%{ip}%'",
+                        f"DELETE FROM AppEvents WHERE ObjectSecondaryID LIKE '%{ip}%'",
+                        f"DELETE FROM Plugins_Objects WHERE Object_SecondaryID LIKE '%{ip}%'"
+                    ]
+                    
+                    for delete in deletes:
+                        run_sqlite_command(delete)
+                    print("Cleanup completed!")
+            else:
+                print("\nNo entries found for this IP address")
+                
+        elif choice == "3":
+            print("\nExiting...")
+            break
+        
+        else:
+            print("\nInvalid option, please try again")
+
+if __name__ == "__main__":
+    check_and_clean_device()