cphipps/NetAlertX
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Plugins code cleanup + refactoring 0.2

Browse Source
This commit is contained in:
Jokob-sk
2023-09-01 22:03:32 +10:00
parent b5e933ba12
commit a379054f5b
11 changed files with 191 additions and 460 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -32,7 +32,7 @@ def main():
# Assuming Plugin_Objects is a class or function that reads data from the RESULT_FILE # Assuming Plugin_Objects is a class or function that reads data from the RESULT_FILE
# and returns a list of objects called 'devices'. # and returns a list of objects called 'devices'.
devices = Plugin_Objects(RESULT_FILE) plugin_objects = Plugin_Objects(RESULT_FILE)
# Print a message to indicate that the script is starting. # Print a message to indicate that the script is starting.
mylog('verbose', ['[ARP Scan] In script ']) mylog('verbose', ['[ARP Scan] In script '])
@@ -69,7 +69,7 @@ def main():
for device in unique_devices: for device in unique_devices:
devices.add_object( plugin_objects.add_object(
primaryId=device['mac'], # MAC (Device Name) primaryId=device['mac'], # MAC (Device Name)
secondaryId=device['ip'], # IP Address secondaryId=device['ip'], # IP Address
watched1=device['ip'], # Device Name watched1=device['ip'], # Device Name
@@ -79,7 +79,7 @@ def main():
extra='arp-scan', extra='arp-scan',
foreignKey="") foreignKey="")
devices.write_result_file() plugin_objects.write_result_file()
return 0 return 0

99
front/plugins/dhcp_leases/script.py
View File

@@ -1,113 +1,74 @@
#!/usr/bin/env python #!/usr/bin/env python
# Based on the work of https://github.com/leiweibau/Pi.Alert
from __future__ import unicode_literals from __future__ import unicode_literals
from time import sleep, time, strftime
import requests
import pathlib import pathlib
import threading
import subprocess import subprocess
import socket
import argparse import argparse
import io
import sys
from requests.packages.urllib3.exceptions import InsecureRequestWarning
import pwd
import os import os
from dhcp_leases import DhcpLeases
sys.path.append("/home/pi/pialert/front/plugins") sys.path.append("/home/pi/pialert/front/plugins")
sys.path.append('/home/pi/pialert/pialert') sys.path.append('/home/pi/pialert/pialert')
from plugin_helper import Plugin_Object, Plugin_Objects from plugin_helper import Plugin_Object, Plugin_Objects
from logger import mylog, append_line_to_file from logger import mylog
from helper import timeNowTZ from dhcp_leases import DhcpLeases
from const import logPath, pialertPath
CUR_PATH = str(pathlib.Path(__file__).parent.resolve()) CUR_PATH = str(pathlib.Path(__file__).parent.resolve())
LOG_FILE = os.path.join(CUR_PATH, 'script.log') LOG_FILE = os.path.join(CUR_PATH, 'script.log')
RESULT_FILE = os.path.join(CUR_PATH, 'last_result.log') RESULT_FILE = os.path.join(CUR_PATH, 'last_result.log')
def main(): def main():
mylog('verbose', ['[DHCPLSS] In script'])
mylog('verbose',['[DHCPLSS] In script'])
last_run_logfile = open(RESULT_FILE, 'a') last_run_logfile = open(RESULT_FILE, 'a')
# empty file
last_run_logfile.write("") last_run_logfile.write("")
parser = argparse.ArgumentParser(description='Import devices from dhcp.leases files') parser = argparse.ArgumentParser(description='Import devices from dhcp.leases files')
parser.add_argument('paths', action="store", help="absolute dhcp.leases file paths to check separated by ','") parser.add_argument('paths', action="store", help="absolute dhcp.leases file paths to check separated by ','")
values = parser.parse_args() values = parser.parse_args()
plugin_objects = Plugin_Objects(RESULT_FILE)
# Init the file
plug_objects = Plugin_Objects( RESULT_FILE )
# parse output
if values.paths: if values.paths:
for path in values.paths.split('=')[1].split(','): for path in values.paths.split('=')[1].split(','):
plug_objects_tmp = get_entries(path, plugin_objects)
mylog('verbose', [f'[DHCPLSS] {len(plug_objects_tmp)} Entries found in "{path}"'])
plugin_objects = plugin_objects + plug_objects_tmp
plug_objects_tmp = get_entries(path, plug_objects) plugin_objects.write_result_file()
mylog('verbose',[f'[DHCPLSS] {len(plug_objects_tmp)} Entries found in "{path}"']) def get_entries(path, plugin_objects):
plug_objects = plug_objects + plug_objects_tmp
plug_objects.write_result_file()
# -----------------------------------------------------------------------------
def get_entries(path, plug_objects):
# PiHole dhcp.leases format
if 'pihole' in path: if 'pihole' in path:
data = [] data = []
reporting = False reporting = False
with open(piholeDhcpleases, 'r') as f: with open(piholeDhcpleases, 'r') as f:
for line in f: for line in f:
row = line.rstrip().split() row = line.rstrip().split()
# rows: DHCP_DateTime, DHCP_MAC, DHCP_IP, DHCP_Name, DHCP_MAC2 if len(row) == 5:
if len(row) == 5 : plugin_objects.add_object(
plug_objects.add_object( primaryId=row[1],
primaryId = row[1], secondaryId=row[2],
secondaryId = row[2], watched1='True',
watched1 = 'True', watched2=row[3],
watched2 = row[3], watched3=row[4],
watched3 = row[4], watched4='True',
watched4 = 'True', extra=path,
extra = path, foreignKey=row[1]
foreignKey = row[1] )
)
# Generic dhcp.leases format
else: else:
leases = DhcpLeases(path) leases = DhcpLeases(path)
leasesList = leases.get() leasesList = leases.get()
for lease in leasesList: for lease in leasesList:
plug_objects.add_object( plugin_objects.add_object(
primaryId = lease.ethernet, primaryId=lease.ethernet,
secondaryId = lease.ip, secondaryId=lease.ip,
watched1 = lease.active, watched1=lease.active,
watched2 = lease.hostname, watched2=lease.hostname,
watched3 = lease.hardware, watched3=lease.hardware,
watched4 = lease.binding_state, watched4=lease.binding_state,
extra = path, extra=path,
foreignKey = lease.ethernet foreignKey=lease.ethernet
) )
return plugin_objects
return plug_objects
#===============================================================================
# BEGIN
#===============================================================================
if __name__ == '__main__': if __name__ == '__main__':
main() main()

166
front/plugins/dhcp_servers/script.py
View File

@@ -1,133 +1,71 @@
#!/usr/bin/env python #!/usr/bin/env python
# Based on the work of https://github.com/leiweibau/Pi.Alert # Based on the work of https://github.com/leiweibau/Pi.Alert
from __future__ import unicode_literals
from time import sleep, time, strftime
import requests
import pathlib
import threading
import subprocess import subprocess
import socket from datetime import datetime
import argparse from plugin_helper import Plugin_Objects, Plugin_Object
import io from logger import mylog
import sys
from requests.packages.urllib3.exceptions import InsecureRequestWarning
import pwd
import os
curPath = str(pathlib.Path(__file__).parent.resolve()) def main():
log_file = curPath + '/script.log'
last_run = curPath + '/last_result.log'
print(last_run) mylog('verbose', ['[DHCPSRVS] In script'])
# Workflow
def main():
last_run_logfile = open(last_run, 'a')
RESULT_FILE = 'last_result.log'
plugin_objects = Plugin_Objects(RESULT_FILE)
timeoutSec = 10 timeoutSec = 10
nmapArgs = ['sudo', 'nmap', '--script', 'broadcast-dhcp-discover'] nmapArgs = ['sudo', 'nmap', '--script', 'broadcast-dhcp-discover']
# Execute N probes and insert in list try:
dhcp_probes = 1 # N probes dhcp_probes = 1
newLines = [] newLines = [datetime.now().strftime("%Y-%m-%d %H:%M:%S")]
newLines.append(strftime("%Y-%m-%d %H:%M:%S"))
#dhcp_server_list_time = []
for _ in range(dhcp_probes):
output = subprocess.check_output (nmapArgs, universal_newlines=True, stderr=subprocess.STDOUT, timeout=(timeoutSec ))
newLines = newLines + output.split("\n")
# parse output
newEntries = []
duration = ""
for line in newLines:
if newEntries is None: for _ in range(dhcp_probes):
index = 0 output = subprocess.check_output(nmapArgs, universal_newlines=True, stderr=subprocess.STDOUT, timeout=timeoutSec)
else: newLines += output.split("\n")
index = len(newEntries) - 1
if 'Response ' in line and ' of ' in line: newEntries = []
newEntries.append(plugin_object_class())
elif 'Server Identifier' in line :
newEntries[index].primaryId = line.split(':')[1].strip()
elif 'Domain Name' in line :
newEntries[index].secondaryId = line.split(':')[1].strip()
elif 'Domain Name Server' in line :
newEntries[index].watched1 = line.split(':')[1].strip()
elif 'IP Offered' in line :
newEntries[index].watched2 = line.split(':')[1].strip()
elif 'Interface' in line :
newEntries[index].watched3 = line.split(':')[1].strip()
elif 'Router' in line :
newEntries[index].watched4 = line.split(':')[1].strip()
newEntries[index].foreignKey = line.split(':')[1].strip()
elif ('IP Address Lease Time' in line or 'Subnet Mask' in line or 'Broadcast Address' in line) :
newVal = line.split(':')[1].strip() for line in newLines:
if 'Response ' in line and ' of ' in line:
newEntries.append(Plugin_Object())
elif 'Server Identifier' in line:
newEntries[-1].primaryId = line.split(':')[1].strip()
elif 'Domain Name' in line:
newEntries[-1].secondaryId = line.split(':')[1].strip()
elif 'Domain Name Server' in line:
newEntries[-1].watched1 = line.split(':')[1].strip()
elif 'IP Offered' in line:
newEntries[-1].watched2 = line.split(':')[1].strip()
elif 'Interface' in line:
newEntries[-1].watched3 = line.split(':')[1].strip()
elif 'Router' in line:
value = line.split(':')[1].strip()
newEntries[-1].watched4 = value
newEntries[-1].foreignKey = value
if newEntries[index].extra == '': if 'IP Address Lease Time' in line or 'Subnet Mask' in line or 'Broadcast Address' in line:
newEntries[index].extra = newVal newVal = line.split(':')[1].strip()
else: if newEntries[-1].extra == '':
newEntries[index].extra = newEntries[index].extra + ',' + newVal newEntries[-1].extra = newVal
else:
newEntries[-1].extra += ',' + newVal
for e in newEntries:
plugin_objects.add_object(
primaryId=e.primaryId,
secondaryId=e.secondaryId,
watched1=e.watched1,
watched2=e.watched2,
watched3=e.watched3,
watched4=e.watched4,
extra=e.extra,
foreignKey=e.foreignKey
)
for e in newEntries: plugin_objects.write_result_file()
# Insert list into the log except Exception as e:
service_monitoring_log(e.primaryId, e.secondaryId, e.created, e.watched1, e.watched2, e.watched3, e.watched4, e.extra, e.foreignKey ) mylog('none', ['Error in main:', str(e)])
# -----------------------------------------------------------------------------
def service_monitoring_log(primaryId, secondaryId, created, watched1, watched2 = '', watched3 = '', watched4 = '', extra ='', foreignKey ='' ):
if watched1 == '':
watched1 = 'null'
if watched2 == '':
watched2 = 'null'
if watched3 == '':
watched3 = 'null'
if watched4 == '':
watched4 = 'null'
with open(last_run, 'a') as last_run_logfile:
# https://www.duckduckgo.com|192.168.0.1|2023-01-02 15:56:30|200|0.9898|null|null|Best search engine|null
last_run_logfile.write("{}|{}|{}|{}|{}|{}|{}|{}|{}\n".format(
primaryId,
secondaryId,
created,
watched1,
watched2,
watched3,
watched4,
extra,
foreignKey
)
)
# -------------------------------------------------------------------
class plugin_object_class:
def __init__(self, primaryId = '',secondaryId = '', watched1 = '',watched2 = '',watched3 = '',watched4 = '',extra = '',foreignKey = ''):
self.pluginPref = ''
self.primaryId = primaryId
self.secondaryId = secondaryId
self.created = strftime("%Y-%m-%d %H:%M:%S")
self.changed = ''
self.watched1 = watched1
self.watched2 = watched2
self.watched3 = watched3
self.watched4 = watched4
self.status = ''
self.extra = extra
self.userData = ''
self.foreignKey = foreignKey
#===============================================================================
# BEGIN
#===============================================================================
if __name__ == '__main__': if __name__ == '__main__':
sys.exit(main()) main()

6
front/plugins/nmap_scan/config.json
View File

@@ -227,8 +227,8 @@
} , } ,
{ {
"column": "Extra", "column": "Extra",
"css_classes": "col-sm-3", "css_classes": "col-sm-1",
"show": true, "show": false,
"type": "label", "type": "label",
"default_value":"", "default_value":"",
"options": [], "options": [],
@@ -262,7 +262,7 @@
{ {
"column": "ForeignKey", "column": "ForeignKey",
"css_classes": "col-sm-2", "css_classes": "col-sm-2",
"show": true, "show": false,
"type": "device_mac", "type": "device_mac",
"default_value":"", "default_value":"",
"options": [], "options": [],

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

@@ -34,7 +34,7 @@ def main():
# Plugin_Objects is a class that reads data from the RESULT_FILE # Plugin_Objects is a class that reads data from the RESULT_FILE
# and returns a list of results. # and returns a list of results.
results = Plugin_Objects(RESULT_FILE) plugin_objects = Plugin_Objects(RESULT_FILE)
# Print a message to indicate that the script is starting. # Print a message to indicate that the script is starting.
mylog('debug', ['[NMAP Scan] In script ']) mylog('debug', ['[NMAP Scan] In script '])
@@ -55,7 +55,7 @@ def main():
for entry in entries: for entry in entries:
results.add_object( plugin_objects.add_object(
primaryId = entry.mac, # MAC (Device Name) primaryId = entry.mac, # MAC (Device Name)
secondaryId = entry.port, # IP Address (always 0.0.0.0) secondaryId = entry.port, # IP Address (always 0.0.0.0)
watched1 = entry.state, # Device Name watched1 = entry.state, # Device Name
@@ -67,7 +67,7 @@ def main():
) )
# generate last_result.log file # generate last_result.log file
results.write_result_file() plugin_objects.write_result_file()
#------------------------------------------------------------------------------- #-------------------------------------------------------------------------------

4
front/plugins/pholus_scan/pholus/pholus3.py
View File

@@ -18,9 +18,7 @@ from scapy.utils import PcapWriter
sys.setrecursionlimit(30000) sys.setrecursionlimit(30000)
logging.getLogger("scapy.runtime").setLevel(logging.ERROR)#supress Scapy warnings` logging.getLogger("scapy.runtime").setLevel(logging.ERROR)#supress Scapy warnings`
runPath = os.path.dirname(os.path.abspath(__file__)) logPath = '/home/pi/pialert/front/log'
runPathTmp = runPath + "/.."
logPath = runPathTmp + '/front/log'
# DEBUG # DEBUG
isDebug = False isDebug = False

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

@@ -37,7 +37,7 @@ def main():
# Assuming Plugin_Objects is a class or function that reads data from the RESULT_FILE # Assuming Plugin_Objects is a class or function that reads data from the RESULT_FILE
# and returns a list of objects called 'devices'. # and returns a list of objects called 'devices'.
plug_objects = Plugin_Objects(RESULT_FILE) plugin_objects = Plugin_Objects(RESULT_FILE)
# Print a message to indicate that the script is starting. # Print a message to indicate that the script is starting.
mylog('verbose',['[PHOLUS] In script']) mylog('verbose',['[PHOLUS] In script'])
@@ -78,7 +78,7 @@ def main():
for entry in all_entries: for entry in all_entries:
plug_objects.add_object( plugin_objects.add_object(
# "Info", "Time", "MAC", "IP_v4_or_v6", "Record_Type", "Value" # "Info", "Time", "MAC", "IP_v4_or_v6", "Record_Type", "Value"
primaryId = entry[2], primaryId = entry[2],
secondaryId = entry[3], secondaryId = entry[3],
@@ -89,7 +89,7 @@ def main():
extra = entry[0], extra = entry[0],
foreignKey = entry[2]) foreignKey = entry[2])
plug_objects.write_result_file() plugin_objects.write_result_file()
return 0 return 0
@@ -124,8 +124,8 @@ def execute_pholus_scan(userSubnets, timeoutSec):
result_list += pholus_output_list result_list += pholus_output_list
mylog('verbose', ["List len:", len(result_list)]) mylog('verbose', ["[PHOLUS] Pholus output number of entries:", len(result_list)])
mylog('verbose',["List:", result_list]) mylog('verbose', ["[PHOLUS] List:", result_list])
return result_list return result_list

190
front/plugins/snmp_discovery/script.py
View File

@@ -4,178 +4,94 @@
# python3 /home/pi/pialert/front/plugins/snmp_discovery/script.py routers='snmpwalk -v 2c -c public -OXsq 192.168.1.1 .1.3.6.1.2.1.3.1.1.2' # python3 /home/pi/pialert/front/plugins/snmp_discovery/script.py routers='snmpwalk -v 2c -c public -OXsq 192.168.1.1 .1.3.6.1.2.1.3.1.1.2'
from __future__ import unicode_literals from __future__ import unicode_literals
from time import sleep, time, strftime
import requests
from requests import Request, Session, packages
import pathlib import pathlib
import threading
import subprocess import subprocess
import socket
import json
import argparse import argparse
import io
import sys
from requests.packages.urllib3.exceptions import InsecureRequestWarning
import pwd
import os import os
import sys
sys.path.append("/home/pi/pialert/front/plugins") sys.path.append("/home/pi/pialert/front/plugins")
sys.path.append('/home/pi/pialert/pialert') sys.path.append('/home/pi/pialert/pialert')
from plugin_helper import Plugin_Object, Plugin_Objects, decodeBase64 from plugin_helper import Plugin_Object, Plugin_Objects, decodeBase64
from logger import mylog, append_line_to_file from logger import mylog
from helper import timeNowTZ from helper import timeNowTZ
from const import logPath, pialertPath from const import logPath, pialertPath
CUR_PATH = str(pathlib.Path(__file__).parent.resolve()) CUR_PATH = str(pathlib.Path(__file__).parent.resolve())
LOG_FILE = os.path.join(CUR_PATH, 'script.log')
RESULT_FILE = os.path.join(CUR_PATH, 'last_result.log') RESULT_FILE = os.path.join(CUR_PATH, 'last_result.log')
# Workflow # Workflow
def main(): def main():
mylog('verbose', ['[SNMPDSC] In script ']) mylog('verbose', ['[SNMPDSC] In script '])
# init global variables # init global variables
global ROUTERS global ROUTERS
# empty file
open(RESULT_FILE , 'w').close()
last_run_logfile = open(RESULT_FILE, 'a')
parser = argparse.ArgumentParser(description='This plugin is used to discover devices via the arp table(s) of a RFC1213 compliant router or switch.') parser = argparse.ArgumentParser(description='This plugin is used to discover devices via the arp table(s) of a RFC1213 compliant router or switch.')
parser.add_argument('routers', action="store", help="IP(s) of routers, separated by comma (,) if passing multiple") parser.add_argument('routers', action="store", help="IP(s) of routers, separated by comma (,) if passing multiple")
values = parser.parse_args() values = parser.parse_args()
# parse output plugin_objects = Plugin_Objects(RESULT_FILE)
newEntries = []
if values.routers: if values.routers:
ROUTERS = values.routers.split('=')[1].replace('\'','') ROUTERS = values.routers.split('=')[1].replace('\'','')
newEntries = get_entries(newEntries)
mylog('verbose', ['[SNMPDSC] Entries found: ', len(newEntries)])
for e in newEntries:
# Insert list into the log
service_monitoring_log(e.primaryId, e.secondaryId, e.created, e.watched1, e.watched2, e.watched3, e.watched4, e.extra, e.foreignKey )
# -----------------------------------------------------------------------------
def get_entries(newEntries):
routers = []
if ',' in ROUTERS:
# multiple
routers = ROUTERS.split(',')
else:
# only one
routers.append(ROUTERS)
for router in routers:
# snmpwalk -v 2c -c public -OXsq 192.168.1.1 .1.3.6.1.2.1.3.1.1.2
mylog('verbose', ['[SNMPDSC] Router snmpwalk command: ', router])
timeoutSec = 10
snmpwalkArgs = router.split(' ')
# Execute N probes and insert in list
probes = 1 # N probes
newLines = []
for _ in range(probes):
output = subprocess.check_output (snmpwalkArgs, universal_newlines=True, stderr=subprocess.STDOUT, timeout=(timeoutSec ))
newLines = newLines + output.split("\n")
# Process outputs
# Sample: iso.3.6.1.2.1.3.1.1.2.3.1.192.168.1.2 "6C 6C 6C 6C 6C 6C "
with open(LOG_FILE, 'a') as run_logfile:
for line in newLines:
# debug
run_logfile.write(line)
tmpSplt = line.split('"')
if len(tmpSplt) == 3:
ipStr = tmpSplt[0].split('.')[-4:] # Get the last 4 elements to extract the IP
macStr = tmpSplt[1].strip().split(' ') # Remove leading/trailing spaces from MAC
if 'iso.' in line and len(ipStr) == 4:
macAddress = ':'.join(macStr)
ipAddress = '.'.join(ipStr)
tmpEntry = plugin_object_class(
macAddress,
ipAddress,
watched1='(unknown)',
watched2=snmpwalkArgs[6], # router IP
extra=line
)
newEntries.append(tmpEntry)
return newEntries
if ',' in ROUTERS:
# ------------------------------------------------------------------- routers = ROUTERS.split(',')
class plugin_object_class: else:
def __init__(self, primaryId = '',secondaryId = '', watched1 = '',watched2 = '',watched3 = '',watched4 = '',extra = '',foreignKey = ''): routers = [ROUTERS]
self.pluginPref = ''
self.primaryId = primaryId
self.secondaryId = secondaryId
self.created = strftime("%Y-%m-%d %H:%M:%S")
self.changed = ''
self.watched1 = watched1
self.watched2 = watched2
self.watched3 = watched3
self.watched4 = watched4
self.status = ''
self.extra = extra
self.userData = ''
self.foreignKey = foreignKey
# -----------------------------------------------------------------------------
def service_monitoring_log(primaryId, secondaryId, created, watched1, watched2 = 'null', watched3 = 'null', watched4 = 'null', extra ='null', foreignKey ='null' ):
if watched1 == '': for router in routers:
watched1 = 'null' mylog('verbose', ['[SNMPDSC] Router snmpwalk command: ', router])
if watched2 == '': timeoutSec = 10
watched2 = 'null' snmpwalkArgs = router.split(' ')
if watched3 == '':
watched3 = 'null'
if watched4 == '':
watched4 = 'null'
if extra == '':
extra = 'null'
if foreignKey == '':
foreignKey = 'null'
with open(RESULT_FILE, 'a') as last_run_logfile: # Execute N probes and insert in list
last_run_logfile.write("{}|{}|{}|{}|{}|{}|{}|{}|{}\n".format( probes = 1 # N probes
primaryId,
secondaryId, for _ in range(probes):
created, output = subprocess.check_output (snmpwalkArgs, universal_newlines=True, stderr=subprocess.STDOUT, timeout=(timeoutSec ))
watched1,
watched2,
watched3,
watched4,
extra,
foreignKey
)
)
mylog('verbose', ['[SNMPDSC] output: ', output])
lines = output.split('\n')
for line in lines:
tmpSplt = line.split('"')
if len(tmpSplt) == 3:
ipStr = tmpSplt[0].split('.')[-4:] # Get the last 4 elements to extract the IP
macStr = tmpSplt[1].strip().split(' ') # Remove leading/trailing spaces from MAC
if 'iso.' in output and len(ipStr) == 4:
macAddress = ':'.join(macStr)
ipAddress = '.'.join(ipStr)
mylog('verbose', [f'[SNMPDSC] IP: {ipAddress} MAC: {macAddress}'])
plugin_objects.add_object(
primaryId=macAddress,
secondaryId=ipAddress,
watched1='(unknown)',
watched2=snmpwalkArgs[6], # router IP
extra=line,
foreignKey=macAddress # Use the primary ID as the foreign key
)
mylog('verbose', ['[SNMPDSC] Entries found: ', len(plugin_objects)])
plugin_objects.write_result_file()
#===============================================================================
# BEGIN # BEGIN
#===============================================================================
if __name__ == '__main__': if __name__ == '__main__':
main() main()

6
front/plugins/undiscoverables/script.py
View File

@@ -27,11 +27,11 @@ def main():
mylog('verbose', ['[UNDIS] In script']) mylog('verbose', ['[UNDIS] In script'])
UNDIS_devices = Plugin_Objects( RESULT_FILE ) plugin_objects = Plugin_Objects( RESULT_FILE )
if values.devices: if values.devices:
for fake_dev in values.devices.split('=')[1].split(','): for fake_dev in values.devices.split('=')[1].split(','):
UNDIS_devices.add_object( plugin_objects.add_object(
primaryId=fake_dev, # MAC (Device Name) primaryId=fake_dev, # MAC (Device Name)
secondaryId="0.0.0.0", # IP Address (always 0.0.0.0) secondaryId="0.0.0.0", # IP Address (always 0.0.0.0)
watched1=fake_dev, # Device Name watched1=fake_dev, # Device Name
@@ -41,7 +41,7 @@ def main():
extra="", extra="",
foreignKey="") foreignKey="")
UNDIS_devices.write_result_file() plugin_objects.write_result_file()
return 0 return 0

156
front/plugins/website_monitor/script.py
View File

@@ -1,148 +1,66 @@
#!/usr/bin/env python #!/usr/bin/env python
# Based on the work of https://github.com/leiweibau/Pi.Alert # Based on the work of https://github.com/leiweibau/Pi.Alert
# Example call
# python3 /home/pi/pialert/front/plugins/website_monitor/script.py urls=http://google.com,http://bing.com # python3 /home/pi/pialert/front/plugins/website_monitor/script.py urls=http://google.com,http://bing.com
from __future__ import unicode_literals import argparse
from time import sleep, time, strftime
import requests import requests
import pathlib import pathlib
import argparse
import io
#import smtplib
import sys import sys
#from smtp_config import sender, password, receivers, host, port import os
from requests.packages.urllib3.exceptions import InsecureRequestWarning from requests.packages.urllib3.exceptions import InsecureRequestWarning
import pwd sys.path.extend(["/home/pi/pialert/front/plugins", "/home/pi/pialert/pialert"])
import os
curPath = str(pathlib.Path(__file__).parent.resolve()) from plugin_helper import Plugin_Objects
log_file = curPath + '/script.log' from datetime import datetime
last_run = curPath + '/last_result.log' from const import logPath
print(last_run) CUR_PATH = str(pathlib.Path(__file__).parent.resolve())
RESULT_FILE = os.path.join(CUR_PATH, 'last_result.log')
# Workflow def main():
def main():
parser = argparse.ArgumentParser(description='Simple URL monitoring tool') parser = argparse.ArgumentParser(description='Simple URL monitoring tool')
parser.add_argument('urls', action="store", help="urls to check separated by ','") parser.add_argument('urls', action="store", help="URLs to check separated by ','")
values = parser.parse_args() values = parser.parse_args()
if values.urls: if values.urls:
with open(last_run, 'w') as last_run_logfile: plugin_objects = Plugin_Objects(RESULT_FILE)
# empty file plugin_objects = service_monitoring(values.urls.split('=')[1].split(','), plugin_objects)
last_run_logfile.write("") plugin_objects.write_result_file()
service_monitoring(values.urls.split('=')[1].split(','))
else: else:
return return
# -----------------------------------------------------------------------------
def service_monitoring_log(site, status, latency):
# global monitor_logfile
# Log status message to log file
with open(log_file, 'a') as monitor_logfile:
monitor_logfile.write("{} | {} | {} | {}\n".format(strftime("%Y-%m-%d %H:%M:%S"),
site,
status,
latency,
)
)
with open(last_run, 'a') as last_run_logfile:
# https://www.duckduckgo.com|192.168.0.1|2023-01-02 15:56:30|200|0.9898|null|null|Best search engine|null
last_run_logfile.write("{}|{}|{}|{}|{}|{}|{}|{}|{}\n".format(
site,
'null',
strftime("%Y-%m-%d %H:%M:%S"),
status,
latency,
'null',
'null',
'null',
'null',
)
)
# -----------------------------------------------------------------------------
def check_services_health(site): def check_services_health(site):
# Enable self signed SSL
requests.packages.urllib3.disable_warnings(InsecureRequestWarning) requests.packages.urllib3.disable_warnings(InsecureRequestWarning)
"""Send GET request to input site and return status code"""
try: try:
resp = requests.get(site, verify=False, timeout=10) resp = requests.get(site, verify=False, timeout=10)
latency = resp.elapsed latency = resp.elapsed.total_seconds()
latency_str = str(latency) status = resp.status_code
latency_str_seconds = latency_str.split(":")
format_latency_str = latency_str_seconds[2]
if format_latency_str[0] == "0" and format_latency_str[1] != "." :
format_latency_str = format_latency_str[1:]
return resp.status_code, format_latency_str
except requests.exceptions.SSLError: except requests.exceptions.SSLError:
pass status = 503
latency = 99999
except: except:
latency = "99999" status = 503
return 503, latency latency = 99999
return status, latency
# ----------------------------------------------------------------------------- def service_monitoring(urls, plugin_objects):
def get_username(): for site in urls:
status, latency = check_services_health(site)
plugin_objects.add_object(
primaryId=site,
secondaryId='null',
watched1=status,
watched2=latency,
watched3='null',
watched4='null',
extra='null',
foreignKey='null'
)
return plugin_objects
return pwd.getpwuid(os.getuid())[0]
# -----------------------------------------------------------------------------
def service_monitoring(urls):
# Empty Log and write new header
print("Prepare Services Monitoring")
print("... Prepare Logfile")
with open(log_file, 'w') as monitor_logfile:
monitor_logfile.write("Pi.Alert [Prototype]:\n---------------------------------------------------------\n")
monitor_logfile.write("Current User: %s \n\n" % get_username())
monitor_logfile.write("Monitor Web-Services\n")
monitor_logfile.write("Timestamp: " + strftime("%Y-%m-%d %H:%M:%S") + "\n")
monitor_logfile.close()
print("... Get Services List")
sites = urls
print("Start Services Monitoring")
with open(log_file, 'a') as monitor_logfile:
monitor_logfile.write("\nStart Services Monitoring\n\n| Timestamp | URL | StatusCode | ResponseTime |\n-----------------------------------------------\n")
monitor_logfile.close()
while sites:
for site in sites:
status,latency = check_services_health(site)
scantime = strftime("%Y-%m-%d %H:%M:%S")
# Debugging
# print("{} - {} STATUS: {} ResponseTime: {}".format(strftime("%Y-%m-%d %H:%M:%S"),
# site,
# status,
# latency)
# )
# Write Logfile
service_monitoring_log(site, status, latency)
sys.stdout.flush()
break
else:
with open(log_file, 'a') as monitor_logfile:
monitor_logfile.write("\n\nNo site(s) to monitor!")
monitor_logfile.close()
#===============================================================================
# BEGIN
#===============================================================================
if __name__ == '__main__': if __name__ == '__main__':
sys.exit(main()) sys.exit(main())

2
pialert/reporting.py
View File

@@ -100,7 +100,7 @@ def construct_notifications(db, sqlQuery, tableTitle, skipText = False, supplied
notiStruc = noti_struc(jsn, text, html) notiStruc = noti_struc(jsn, text, html)
mylog('debug', ['[Notification] Ports: notiStruc:', json.dumps(notiStruc.__dict__, indent=4) ]) mylog('debug', ['[Notification] notiStruc:', json.dumps(notiStruc.__dict__, indent=4) ])
return notiStruc return notiStruc