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

NMAP plugin conversion v0.3

Browse Source
This commit is contained in:
Jokob-sk
2023-08-30 08:03:02 +10:00
parent 61df3ec581
commit f9baa603f4
11 changed files with 51 additions and 41 deletions
Download Patch File Download Diff File Expand all files Collapse all files

1
back/report_template.html
View File

@@ -42,7 +42,6 @@
<NEW_DEVICES_TABLE>
<DOWN_DEVICES_TABLE>
<EVENTS_TABLE>
<PORTS_TABLE>
<PLUGINS_TABLE>
</td>

1
back/report_template.txt
View File

@@ -5,5 +5,4 @@ Server: <SERVER_NAME>
<SECTION_DEVICES_DOWN>
<SECTION_EVENTS>
<SECTION_INTERNET>
<PORTS_TABLE>
<PLUGINS_TABLE>

1
back/report_template_new_version.html
View File

@@ -42,7 +42,6 @@
<NEW_DEVICES_TABLE>
<DOWN_DEVICES_TABLE>
<EVENTS_TABLE>
<PORTS_TABLE>
<PLUGINS_TABLE>
</td>

1
front/deviceDetails.php
View File

@@ -113,7 +113,6 @@
<li> <a id="tabSessions" href="#panSessions" data-toggle="tab"> <?= lang('DevDetail_Tab_Sessions');?> </a></li>
<li> <a id="tabPresence" href="#panPresence" data-toggle="tab"> <?= lang('DevDetail_Tab_Presence');?> </a></li>
<li> <a id="tabEvents" href="#panEvents" data-toggle="tab"> <?= lang('DevDetail_Tab_Events');?> </a></li>
<li> <a id="tabPholus" href="#panPholus" data-toggle="tab"> <?= lang('DevDetail_Tab_Pholus');?> </a></li>
<li> <a id="tabPlugins" href="#panPlugins" data-toggle="tab"> <?= lang('DevDetail_Tab_Plugins');?> </a></li>
<div class="btn-group pull-right">

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

@@ -13,7 +13,7 @@ sys.path.append("/home/pi/pialert/front/plugins")
sys.path.append('/home/pi/pialert/pialert')
from plugin_helper import Plugin_Object, Plugin_Objects
from logger import mylog
from logger import mylog, append_line_to_file
from helper import timeNowTZ
from const import logPath, pialertPath
@@ -35,12 +35,12 @@ def main():
devices = Plugin_Objects(RESULT_FILE)
# Print a message to indicate that the script is starting.
mylog('debug', ['[ARP Scan] In script '])
mylog('verbose', ['[ARP Scan] In script '])
# Assuming 'values' is a dictionary or object that contains a key 'userSubnets'
# which holds a list of user-submitted subnets.
# Printing the userSubnets list to check its content.
mylog('debug', ['[ARP Scan] values.userSubnets: ', values.userSubnets])
mylog('verbose', ['[ARP Scan] values.userSubnets: ', values.userSubnets])
# Extract the base64-encoded subnet information from the first element of the userSubnets list.
@@ -48,14 +48,14 @@ def main():
userSubnetsParamBase64 = values.userSubnets[0].split('userSubnets=b')[1]
# Printing the extracted base64-encoded subnet information.
mylog('debug', ['[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')
# Print the decoded subnet information.
mylog('debug', ['[ARP Scan] userSubnetsParam: ', userSubnetsParam])
mylog('verbose', ['[ARP Scan] 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.
@@ -97,7 +97,7 @@ def execute_arpscan(userSubnets):
arpscan_output = execute_arpscan_on_interface (interface)
mylog('debug', ['[ARP Scan] arpscan_output: ', arpscan_output])
mylog('verbose', ['[ARP Scan] 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]))'
@@ -124,10 +124,10 @@ def execute_arpscan(userSubnets):
unique_devices.append(device)
# return list
mylog('debug', ['[ARP Scan] Found: Devices without duplicates ', len(unique_devices) ])
mylog('verbose', ['[ARP Scan] Found: Devices without duplicates ', len(unique_devices) ])
mylog('debug', ["Devices List len:", len(devices_list)]) # Add this line to print devices_list
mylog('debug',["Devices List:", devices_list]) # Add this line to print devices_list
mylog('verbose', ["Devices List len:", len(devices_list)]) # Add this line to print devices_list
mylog('verbose',["Devices List:", devices_list]) # Add this line to print devices_list
return devices_list

24
front/plugins/nmap_scan/script.py
View File

@@ -1,4 +1,3 @@
#!/usr/bin/env python
import os
@@ -52,6 +51,8 @@ def main():
entries = performNmapScan(values.ips[0].split('=')[1].split(','), values.macs[0].split('=')[1].split(',') , values.timeout[0].split('=')[1], argsDecoded)
mylog('verbose', ['[NMAP Scan] Total number of ports found by NMAP: ', len(entries)])
for entry in entries:
results.add_object(
@@ -61,11 +62,12 @@ def main():
watched2 = entry.service,
watched3 = entry.ip + ":" + entry.port,
watched4 = "",
extra = "",
foreignKey = entry.extra
extra = entry.extra,
foreignKey = entry.mac
)
entries.write_result_file()
# generate last_result.log file
results.write_result_file()
#-------------------------------------------------------------------------------
@@ -88,6 +90,11 @@ def performNmapScan(deviceIPs, deviceMACs, timeoutSec, args):
run nmap scan on a list of devices
discovers open ports and keeps track existing and new open ports
"""
# collect ports / new Nmap Entries
newEntriesTmp = []
if len(deviceIPs) > 0:
devTotal = len(deviceIPs)
@@ -96,8 +103,6 @@ def performNmapScan(deviceIPs, deviceMACs, timeoutSec, args):
mylog('verbose', ['[NMAP Scan] Scan: Nmap for max ', str(timeoutSec), 's ('+ str(round(int(timeoutSec) / 60, 1)) +'min) per device'])
mylog('verbose', ["[NMAP Scan] Estimated max delay: ", (devTotal * int(timeoutSec)), 's ', '(', round((devTotal * int(timeoutSec))/60,1) , 'min)' ])
# collect ports / new Nmap Entries
newEntriesTmp = []
devIndex = 0
for ip in deviceIPs:
@@ -133,10 +138,10 @@ def performNmapScan(deviceIPs, deviceMACs, timeoutSec, args):
append_line_to_file (logPath + '/pialert_nmap.log', line +'\n')
index = 0
startCollecting = False
duration = ""
newPortsPerDevice = 0
for line in newLines:
if 'Starting Nmap' in line:
if len(newLines) > index+1 and 'Note: Host seems down' in newLines[index+1]:
@@ -147,13 +152,16 @@ def performNmapScan(deviceIPs, deviceMACs, timeoutSec, args):
startCollecting = False # end reached
elif startCollecting and len(line.split()) == 3:
newEntriesTmp.append(nmap_entry(ip, deviceMACs[devIndex], timeNowTZ(), line.split()[0], line.split()[1], line.split()[2]))
newPortsPerDevice += 1
elif 'Nmap done' in line:
duration = line.split('scanned in ')[1]
mylog('verbose', [f'[NMAP Scan] {newPortsPerDevice} ports found on {deviceMACs[devIndex]}'])
index += 1
devIndex += 1
mylog('verbose', ['[NMAP Scan] Ports found by NMAP: ', len(newEntriesTmp)])
#end for loop

2
front/plugins/pholus_scan/config.json
View File

@@ -22,7 +22,7 @@
"string": "Pholus-Scan (Name discovery)"
},
{
"language_code": "en_us",
"language_code": "es_es",
"string": "Pholus-Scan (Descubrimiento de nombre)"
}
],

10
front/plugins/pholus_scan/script.py
View File

@@ -40,12 +40,12 @@ def main():
plug_objects = Plugin_Objects(RESULT_FILE)
# Print a message to indicate that the script is starting.
print('In script:')
mylog('verbose',['[Pholus] In script'])
# Assuming 'values' is a dictionary or object that contains a key 'userSubnets'
# which holds a list of user-submitted subnets.
# Printing the userSubnets list to check its content.
print(values.userSubnets)
mylog('verbose',['[Pholus] Subnets: ', 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>'.
@@ -123,8 +123,8 @@ def execute_pholus_scan(userSubnets, timeoutSec):
result_list += pholus_output_list
print("List len:", len(result_list))
print("List:", result_list)
mylog('verbose', ["List len:", len(result_list)])
mylog('verbose',["List:", result_list])
return result_list
@@ -133,7 +133,7 @@ def execute_pholus_on_interface(interface, timeoutSec, mask):
# logging & updating app state
mylog('none', ['[PholusScan] Scan: Pholus for ', str(timeoutSec), 's ('+ str(round(int(timeoutSec) / 60, 1)) +'min)'])
mylog('verbose', ['[PholusScan] Scan: Pholus for ', str(timeoutSec), 's ('+ str(round(int(timeoutSec) / 60, 1)) +'min)'])
mylog('verbose', ["[PholusScan] Pholus scan on [interface] ", interface, " [mask] " , mask])
# the scan always lasts 2x as long, so the desired user time from settings needs to be halved

3
pialert/logger.py
View File

@@ -5,6 +5,7 @@ import datetime
import conf
from const import *
# from helper import get_setting_value
#-------------------------------------------------------------------------------
# duplication from helper to avoid circle
@@ -16,6 +17,8 @@ def timeNowTZ():
return datetime.datetime.now().replace(microsecond=0)
# conf.LOG_LEVEL = get_setting_value("LOG_LEVEL")
#-------------------------------------------------------------------------------
debugLevels = [
('none', 0), ('minimal', 1), ('verbose', 2), ('debug', 3)

9
pialert/plugin.py
View File

@@ -37,17 +37,18 @@ class plugin_param:
mylog('debug', f'[Plugins] setTyp: {setTyp}')
if '.select' in setTyp or setTyp in arrayConversion:
# store number of returned values
paramValuesCount = len(setVal)
if setTyp in noConversion:
resolved = setVal
elif setTyp in arrayConversion:
# make them safely passable to a python or linux script
resolved = flatten_array(setVal)
elif setTyp in arrayConversionBase64:
# make them safely passable to a python or linux script by converting them to a base64 string if necessary (if the arg contains spaces)
resolved = flatten_array(setVal)
else:
for item in jsonConversion:
@@ -61,6 +62,10 @@ class plugin_param:
if param["type"] == "sql":
inputValue = db.get_sql_array(param["value"])
# store number of returned values
paramValuesCount = len(inputValue)
# make them safely passable to a python or linux script
resolved = flatten_array(inputValue)

4
pialert/reporting.py
View File

@@ -467,13 +467,11 @@ def skip_repeated_notifications (db):
#-------------------------------------------------------------------------------
def check_and_run_event(db, pluginsState):
mylog('debug', [f'[MAIN] processScan1: {pluginsState.processScan}'])
sql = db.sql # TO-DO
sql.execute(""" select * from Parameters where par_ID = "Front_Event" """)
rows = sql.fetchall()
mylog('debug', [f'[MAIN] processScan2: {pluginsState.processScan}'])
event, param = ['','']
if len(rows) > 0 and rows[0]['par_Value'] != 'finished':
keyValue = rows[0]['par_Value'].split('|')