vh/NetAlertX
1
0
Fork 0
mirror of https://github.com/jokob-sk/NetAlertX.git synced 2025-12-06 17:15:38 -08:00
Code Issues Actions 3 Packages Projects Releases Wiki Activity

/data and /tmp standarization

Browse Source
This commit is contained in:
Adam Outler
2025-11-04 22:26:35 +00:00
parent 90a07c61eb
commit 5b871865db
250 changed files with 7462 additions and 4940 deletions
Download Patch File Download Diff File Expand all files Collapse all files

169
front/plugins/arp_scan/script.py
View File

@@ -1,100 +1,90 @@
#!/usr/bin/env python
import os
import time
import pathlib
import argparse
import sys
import re
import base64
import subprocess
from time import strftime
# Register NetAlertX directories
INSTALL_PATH="/app"
sys.path.extend([f"{INSTALL_PATH}/front/plugins", f"{INSTALL_PATH}/server"])
from database import DB
from plugin_helper import Plugin_Object, Plugin_Objects, handleEmpty
from logger import mylog, Logger, append_line_to_file
from helper import timeNowTZ, get_setting_value
from const import logPath, applicationPath
from plugin_helper import Plugin_Objects, handleEmpty
from logger import mylog, Logger
from helper import get_setting_value
from const import logPath
import conf
from pytz import timezone
import os
import time
import argparse
import re
import base64
import subprocess
# Make sure the TIMEZONE for logging is correct
conf.tz = timezone(get_setting_value('TIMEZONE'))
conf.tz = timezone(get_setting_value("TIMEZONE"))
# Make sure log level is initialized correctly
Logger(get_setting_value('LOG_LEVEL'))
Logger(get_setting_value("LOG_LEVEL"))
pluginName = 'ARPSCAN'
LOG_PATH = logPath + '/plugins'
LOG_FILE = os.path.join(LOG_PATH, f'script.{pluginName}.log')
RESULT_FILE = os.path.join(LOG_PATH, f'last_result.{pluginName}.log')
pluginName = "ARPSCAN"
LOG_PATH = logPath + "/plugins"
LOG_FILE = os.path.join(LOG_PATH, f"script.{pluginName}.log")
RESULT_FILE = os.path.join(LOG_PATH, f"last_result.{pluginName}.log")
def main():
parser = argparse.ArgumentParser(description='Import devices from settings')
parser.add_argument('userSubnets', nargs='+', help="list of subnets with options")
values = parser.parse_args()
parser = argparse.ArgumentParser(description="Import devices from settings")
parser.add_argument("userSubnets", nargs="+", help="list of subnets with options")
values = parser.parse_args()
# Assuming Plugin_Objects is a class or function that reads data from the RESULT_FILE
# and returns a list of objects called 'devices'.
plugin_objects = Plugin_Objects(RESULT_FILE)
# Print a message to indicate that the script is starting.
mylog('verbose', [f'[{pluginName}] In script '])
mylog("verbose", [f"[{pluginName}] In script "])
# holds a list of user-submitted subnets.
# mylog('verbose', ['[ARP Scan] values.userSubnets: ', values.userSubnets])
# holds a list of user-submitted subnets.
# mylog('verbose', ['[ARP Scan] values.userSubnets: ', values.userSubnets])
# Extract the base64-encoded subnet information from the first element of the userSubnets list.
# The format of the element is assumed to be like 'userSubnets=b<base64-encoded-data>'.
userSubnetsParamBase64 = values.userSubnets[0].split('userSubnets=b')[1]
# The format of the element is assumed to be like 'userSubnets=<base64-encoded-data>'.
userSubnetsParamBase64 = values.userSubnets[0].split("userSubnets=")[1]
# Printing the extracted base64-encoded subnet information.
# mylog('verbose', ['[ARP Scan] userSubnetsParamBase64: ', userSubnetsParamBase64])
# mylog('verbose', ['[ARP Scan] userSubnetsParamBase64: ', userSubnetsParamBase64])
# Decode the base64-encoded subnet information to get the actual subnet information in ASCII format.
userSubnetsParam = base64.b64decode(userSubnetsParamBase64).decode('ascii')
userSubnetsParam = base64.b64decode(userSubnetsParamBase64).decode("ascii")
# Print the decoded subnet information.
mylog('verbose', [f'[{pluginName}] userSubnetsParam: ', userSubnetsParam])
mylog("verbose", [f"[{pluginName}] userSubnetsParam: ", userSubnetsParam])
# Check if the decoded subnet information contains multiple subnets separated by commas.
# If it does, split the string into a list of individual subnets.
# Otherwise, create a list with a single element containing the subnet information.
if ',' in userSubnetsParam:
subnets_list = userSubnetsParam.split(',')
if "," in userSubnetsParam:
subnets_list = userSubnetsParam.split(",")
else:
subnets_list = [userSubnetsParam]
# Create a database connection
db = DB() # instance of class DB
db.open()
# Execute the ARP scanning process on the list of subnets (whether it's one or multiple subnets).
# The function 'execute_arpscan' is assumed to be defined elsewhere in the code.
unique_devices = execute_arpscan(subnets_list)
for device in unique_devices:
plugin_objects.add_object(
primaryId = handleEmpty(device['mac']), # MAC (Device Name)
secondaryId = handleEmpty(device['ip']), # IP Address
watched1 = handleEmpty(device['ip']), # Device Name
watched2 = handleEmpty(device.get('hw', '')), # Vendor (assuming it's in the 'hw' field)
watched3 = handleEmpty(device.get('interface', '')), # Add the interface
watched4 = '',
extra = pluginName,
foreignKey = "")
primaryId=handleEmpty(device["mac"]), # MAC (Device Name)
secondaryId=handleEmpty(device["ip"]), # IP Address
watched1=handleEmpty(device["ip"]), # Device Name
watched2=handleEmpty(
device.get("hw", "")
), # Vendor (assuming it's in the 'hw' field)
watched3=handleEmpty(device.get("interface", "")), # Add the interface
watched4="",
extra=pluginName,
foreignKey="",
)
plugin_objects.write_result_file()
@@ -107,17 +97,19 @@ def execute_arpscan(userSubnets):
devices_list = []
# scan each interface
for interface in userSubnets :
arpscan_output = execute_arpscan_on_interface (interface)
for interface in userSubnets:
arpscan_output = execute_arpscan_on_interface(interface)
mylog("verbose", [f"[{pluginName}] arpscan_output: ", arpscan_output])
mylog('verbose', [f'[{pluginName}] arpscan_output: ', arpscan_output])
# Search IP + MAC + Vendor as regular expresion
re_ip = r'(?P<ip>((2[0-5]|1[0-9]|[0-9])?[0-9]\.){3}((2[0-5]|1[0-9]|[0-9])?[0-9]))'
re_mac = r'(?P<mac>([0-9a-fA-F]{2}[:-]){5}([0-9a-fA-F]{2}))'
re_hw = r'(?P<hw>.*)'
re_ip = (
r"(?P<ip>((2[0-5]|1[0-9]|[0-9])?[0-9]\.){3}((2[0-5]|1[0-9]|[0-9])?[0-9]))"
)
re_mac = r"(?P<mac>([0-9a-fA-F]{2}[:-]){5}([0-9a-fA-F]{2}))"
re_hw = r"(?P<hw>.*)"
re_pattern = re.compile(rf"{re_ip}\s+{re_mac}\s{re_hw}")
devices_list_tmp = [
@@ -127,44 +119,61 @@ def execute_arpscan(userSubnets):
devices_list += devices_list_tmp
# mylog('debug', ['[ARP Scan] Found: Devices including duplicates ', len(devices_list) ])
# Delete duplicate MAC
unique_mac = []
unique_devices = []
# mylog('debug', ['[ARP Scan] Found: Devices including duplicates ', len(devices_list) ])
for device in devices_list :
if device['mac'] not in unique_mac:
unique_mac.append(device['mac'])
unique_devices.append(device)
# Delete duplicate MAC
unique_mac = []
unique_devices = []
for device in devices_list:
if device["mac"] not in unique_mac:
unique_mac.append(device["mac"])
unique_devices.append(device)
# return list
mylog('verbose', [f'[{pluginName}] All devices List len:', len(devices_list)])
mylog('verbose', [f'[{pluginName}] Devices List:', devices_list])
mylog("verbose", [f"[{pluginName}] All devices List len:", len(devices_list)])
mylog("verbose", [f"[{pluginName}] Devices List:", devices_list])
mylog('verbose', [f'[{pluginName}] Found: Devices without duplicates ', len(unique_devices) ])
mylog(
"verbose",
[f"[{pluginName}] Found: Devices without duplicates ", len(unique_devices)],
)
return unique_devices
def execute_arpscan_on_interface(interface):
# Prepare command arguments
arpscan_args = get_setting_value('ARPSCAN_ARGS').split() + interface.split()
arpscan_args = get_setting_value("ARPSCAN_ARGS").split() + interface.split()
# Optional duration in seconds (0 = run once)
try:
scan_duration = int(get_setting_value('ARPSCAN_DURATION'))
scan_duration = int(get_setting_value("ARPSCAN_DURATION"))
except Exception:
scan_duration = 0 # default: single run
# Get timeout from plugin settings (default 30 seconds if not set)
try:
timeout_seconds = int(get_setting_value("ARPSCAN_RUN_TIMEOUT"))
except Exception:
timeout_seconds = 30
results = []
start_time = time.time()
while True:
try:
result = subprocess.check_output(arpscan_args, universal_newlines=True)
result = subprocess.check_output(
arpscan_args, universal_newlines=True, timeout=timeout_seconds
)
results.append(result)
except subprocess.CalledProcessError as e:
except subprocess.CalledProcessError:
result = ""
except subprocess.TimeoutExpired:
mylog(
"warning",
[f"[{pluginName}] arp-scan timed out after {timeout_seconds}s"],
)
result = ""
# stop looping if duration not set or expired
if scan_duration == 0 or (time.time() - start_time) > scan_duration:
@@ -175,10 +184,10 @@ def execute_arpscan_on_interface(interface):
return "\n".join(results)
# ===============================================================================
# BEGIN
# ===============================================================================
#===============================================================================
# BEGIN
#===============================================================================
if __name__ == '__main__':
if __name__ == "__main__":
main()