vh/NetAlertX
1
0
Fork 0
mirror of https://github.com/jokob-sk/NetAlertX.git synced 2025-12-07 09:36:05 -08:00
Code Issues Actions Packages Projects Releases Wiki Activity

BE: linting fixes

Browse Source 
Signed-off-by: jokob-sk <jokob.sk@gmail.com>
This commit is contained in:
jokob-sk
2025-11-22 13:14:06 +11:00
parent f0abd500d9
commit 5c14b34a8b
104 changed files with 2163 additions and 2199 deletions
Download Patch File Download Diff File Expand all files Collapse all files

7
scripts/checkmk/script.py
View File

@@ -1,5 +1,5 @@
#!/usr/bin/env python3
# !/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
NetAlertX-New-Devices-Checkmk-Script
@@ -19,11 +19,12 @@ import subprocess
import json
import os
def check_new_devices():
# Get API path from environment variable, fallback to /tmp/api
api_path = os.environ.get('NETALERTX_API', '/tmp/api')
table_devices_path = f'{api_path}/table_devices.json'
try:
# Rufe die JSON-Datei aus dem Docker-Container ab
result = subprocess.run(
@@ -73,6 +74,6 @@ def check_new_devices():
)
print(f"1 NetAlertX_New_Devices - WARNING - Found {len(new_devices)} new device(s): {device_list_str}")
if __name__ == "__main__":
check_new_devices()

32
scripts/db_cleanup/db_cleanup.py
View File

@@ -1,8 +1,8 @@
#!/usr/bin/env python3
# !/usr/bin/env python3
import subprocess
import sys
import os
def run_sqlite_command(command):
# Use environment variable with fallback
db_path = os.path.join(
@@ -19,18 +19,19 @@ def run_sqlite_command(command):
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}'",
@@ -40,14 +41,14 @@ def check_and_clean_device():
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':
@@ -60,16 +61,16 @@ def check_and_clean_device():
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}'",
@@ -79,14 +80,14 @@ def check_and_clean_device():
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':
@@ -99,19 +100,20 @@ def check_and_clean_device():
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()

45
scripts/opnsense_leases/opnsense_leases.py
View File

@@ -1,6 +1,5 @@
#!/usr/bin/env python3
import paramiko
import re
from datetime import datetime
import argparse
import sys
@@ -8,6 +7,9 @@ from pathlib import Path
import time
import logging
logger = None
def setup_logging(debug=False):
"""Configure logging based on debug flag."""
level = logging.DEBUG if debug else logging.INFO
@@ -18,6 +20,7 @@ def setup_logging(debug=False):
)
return logging.getLogger(__name__)
def parse_timestamp(date_str):
"""Convert OPNsense timestamp to Unix epoch time."""
try:
@@ -27,7 +30,7 @@ def parse_timestamp(date_str):
dt = datetime.strptime(clean_date, '%Y/%m/%d %H:%M:%S')
return int(dt.timestamp())
except Exception as e:
logger.error(f"Failed to parse timestamp: {date_str}")
logger.error(f"Failed to parse timestamp: {date_str} ({e})")
return None
@@ -39,8 +42,14 @@ def get_lease_file(hostname, username, password=None, key_filename=None, port=22
try:
logger.debug(f"Attempting to connect to {hostname}:{port} as {username}")
ssh.connect(hostname, port=port, username=username,
password=password, key_filename=key_filename)
ssh.connect(
hostname,
port=port,
username=username,
password=password,
key_filename=key_filename
)
# Get an interactive shell session
logger.debug("Opening interactive SSH channel")
@@ -75,10 +84,23 @@ def get_lease_file(hostname, username, password=None, key_filename=None, port=22
# Clean up the output by removing the command echo and shell prompts
lines = output.split('\n')
# Remove first line (command echo) and any lines containing shell prompts
cleaned_lines = [line for line in lines
if not line.strip().startswith(command.strip())
and not line.strip().endswith('> ')
and not line.strip().endswith('# ')]
# cleaned_lines = [line for line in lines
# if not line.strip().startswith(command.strip()) and not line.strip().endswith('> ') and not line.strip().endswith('# ')]
cmd = command.strip()
cleaned_lines = []
for line in lines:
stripped = line.strip()
if stripped.startswith(cmd):
continue
if stripped.endswith('> '):
continue
if stripped.endswith('# '):
continue
cleaned_lines.append(line)
cleaned_output = '\n'.join(cleaned_lines)
logger.debug(f"Final cleaned output length: {len(cleaned_output)} characters")
@@ -156,9 +178,7 @@ def parse_lease_file(lease_content):
# Filter only active leases
active_leases = [lease for lease in leases
if lease.get('state') == 'active'
and 'mac' in lease
and 'ip' in lease]
if lease.get('state') == 'active' and 'mac' in lease and 'ip' in lease]
logger.debug(f"Found {len(active_leases)} active leases out of {len(leases)} total leases")
logger.debug("Active leases:")
@@ -206,6 +226,7 @@ def convert_to_dnsmasq(leases):
return dnsmasq_lines
def main():
parser = argparse.ArgumentParser(description='Convert OPNsense DHCP leases to dnsmasq format')
parser.add_argument('--host', required=True, help='OPNsense hostname or IP')
@@ -219,6 +240,7 @@ def main():
args = parser.parse_args()
# Setup logging
global logger
logger = setup_logging(args.debug)
try:
@@ -255,5 +277,6 @@ def main():
logger.error(f"Error: {str(e)}")
sys.exit(1)
if __name__ == '__main__':
main()