"
+ },
+ {
+ "equals": "watched-changed",
+ "replacement": "
"
+ },
+ {
+ "equals": "new",
+ "replacement": "
"
+ },
+ {
+ "equals": "missing-in-last-scan",
+ "replacement": "
"
+ }
+ ],
+ "localized": [
+ "name"
+ ],
+ "name": [
+ {
+ "language_code": "en_us",
+ "string": "Status"
+ },
+ {
+ "language_code": "es_es",
+ "string": "Estado"
+ }
+ ]
+ }
+ ]
+}
diff --git a/front/plugins/nslookup_scan/nslookup.py b/front/plugins/nslookup_scan/nslookup.py
new file mode 100755
index 00000000..834e9c6a
--- /dev/null
+++ b/front/plugins/nslookup_scan/nslookup.py
@@ -0,0 +1,134 @@
+#!/usr/bin/env python
+# test script by running:
+# tbc
+
+import os
+import pathlib
+import argparse
+import subprocess
+import sys
+import hashlib
+import csv
+import sqlite3
+import re
+from io import StringIO
+from datetime import datetime
+
+sys.path.append("/home/pi/pialert/front/plugins")
+sys.path.append('/home/pi/pialert/pialert')
+
+from plugin_helper import Plugin_Object, Plugin_Objects, decodeBase64
+from logger import mylog, append_line_to_file
+from helper import timeNowTZ, get_setting_value
+from const import logPath, pialertPath, fullDbPath
+from database import DB
+from device import Device_obj
+
+
+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')
+
+pluginName = 'NSLOOKUP'
+
+def main():
+
+ mylog('verbose', [f'[{pluginName}] In script'])
+
+
+ timeout = get_setting_value('NSLOOKUP_RUN_TIMEOUT')
+
+ # Create a database connection
+ db = DB() # instance of class DB
+ db.open()
+
+ # Initialize the Plugin obj output file
+ plugin_objects = Plugin_Objects(RESULT_FILE)
+
+ # Create a Device_obj instance
+ device_handler = Device_obj(db)
+
+ # Retrieve devices
+ unknown_devices = device_handler.getUnknown()
+
+ for device in unknown_devices:
+ domain_name, dns_server = execute_nslookup(device['dev_LastIP'], timeout)
+
+ if domain_name != '':
+ plugin_objects.add_object(
+ # "MAC", "IP", "Server", "Name"
+ primaryId = device['dev_MAC'],
+ secondaryId = device['dev_LastIP'],
+ watched1 = dns_server,
+ watched2 = domain_name,
+ watched3 = '',
+ watched4 = '',
+ extra = '',
+ foreignKey = device['dev_MAC'])
+
+ plugin_objects.write_result_file()
+
+
+ mylog('verbose', [f'[{pluginName}] Script finished'])
+
+ return 0
+
+#===============================================================================
+# Execute scan
+#===============================================================================
+def execute_nslookup (ip, timeout):
+ """
+ Execute the NSLOOKUP command on IP.
+ """
+
+ nslookup_args = ['nslookup', ip]
+
+ # Execute command
+ output = ""
+
+ try:
+ # try runnning a subprocess with a forced (timeout) in case the subprocess hangs
+ output = subprocess.check_output (nslookup_args, universal_newlines=True, stderr=subprocess.STDOUT, timeout=(timeout), text=True)
+
+ domain_name = ''
+ dns_server = ''
+
+ # Parse output using regular expressions
+ domain_pattern = re.compile(r'Name:\s+(.+)')
+ server_pattern = re.compile(r'Server:\s+(.+)')
+
+ domain_match = domain_pattern.search(output.stdout)
+ server_match = server_pattern.search(output.stdout)
+
+ if domain_match:
+ domain_name = domain_match.group(1)
+ mylog('verbose', [f'[{pluginName}] Domain Name: {domain_name}'])
+
+ if server_match:
+ dns_server = server_match.group(1)
+ mylog('verbose', [f'[{pluginName}] DNS Server: {dns_server}'])
+
+ return domain_name, dns_server
+
+ except subprocess.CalledProcessError as e:
+ # An error occured, handle it
+ mylog('verbose', [f'[{pluginName}]', e.output])
+ mylog('verbose', [f'[{pluginName}] ⚠ ERROR - check logs'])
+ except subprocess.TimeoutExpired as timeErr:
+ mylog('verbose', [f'[{pluginName}] TIMEOUT - the process forcefully terminated as timeout reached'])
+
+ if output == "": # check if the subprocess failed
+ mylog('verbose', [f'[{pluginName}] Scan: FAIL - check logs'])
+ else:
+ mylog('verbose', [f'[{pluginName}] Scan: SUCCESS'])
+
+ return '', ''
+
+
+
+
+#===============================================================================
+# BEGIN
+#===============================================================================
+if __name__ == '__main__':
+ main()
\ No newline at end of file
diff --git a/front/plugins/pholus_scan/script.py b/front/plugins/pholus_scan/script.py
index 52bf89c8..60291570 100755
--- a/front/plugins/pholus_scan/script.py
+++ b/front/plugins/pholus_scan/script.py
@@ -23,6 +23,7 @@ LOG_FILE = os.path.join(CUR_PATH, 'script.log')
RESULT_FILE = os.path.join(CUR_PATH, 'last_result.log')
fullPholusPath = os.path.join(CUR_PATH, 'pholus/pholus3.py')
+pluginName = 'PHOLUS'
def main():
# sample
@@ -40,13 +41,13 @@ def main():
plugin_objects = Plugin_Objects(RESULT_FILE)
# Print a message to indicate that the script is starting.
- mylog('verbose',['[PHOLUS] In script'])
+ mylog('verbose',[f'[{pluginName}] 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('verbose',['[PHOLUS] Subnets: ', values.userSubnets])
- mylog('verbose',['[PHOLUS] len Subnets: ', len(values.userSubnets)])
+ mylog('verbose',[f'[{pluginName}] Subnets: ', values.userSubnets])
+ mylog('verbose',[f'[{pluginName}] len Subnets: ', len(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
'.
@@ -54,14 +55,14 @@ def main():
timeoutSec = values.timeoutSec[0].split('=')[1]
# Printing the extracted base64-encoded subnet information.
- mylog('verbose', [f'[PHOLUS] { userSubnetsParamBase64 }'])
- mylog('verbose', [f'[PHOLUS] { timeoutSec }'])
+ mylog('verbose', [f'[{pluginName}] { userSubnetsParamBase64 }'])
+ mylog('verbose', [f'[{pluginName}] { timeoutSec }'])
# 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('verbose', [f'[PHOLUS] 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.
@@ -99,16 +100,15 @@ def execute_pholus_scan(userSubnets, timeoutSec):
timeoutPerSubnet = float(timeoutSec) / len(userSubnets)
- mylog('verbose', [f'[PHOLUS] { timeoutPerSubnet } '])
+ mylog('verbose', [f'[{pluginName}] { timeoutPerSubnet } '])
- # scan each interface
-
+ # scan each interface
for interface in userSubnets:
temp = interface.split("--interface=")
if len(temp) != 2:
- mylog('none', ["[PHOLUS] Skip scan (need interface in format '192.168.1.0/24 --inteface=eth0'), got: ", interface])
+ mylog('verbose', [f'[{pluginName}] Skip scan (need interface in format "192.168.1.0/24 --inteface=eth0"), got: ', interface])
return
mask = temp[0].strip()
@@ -116,14 +116,14 @@ def execute_pholus_scan(userSubnets, timeoutSec):
pholus_output_list = execute_pholus_on_interface (interface, timeoutPerSubnet, mask)
- mylog('verbose', [f'[PHOLUS] { pholus_output_list } '])
+ mylog('verbose', [f'[{pluginName}] { pholus_output_list } '])
result_list += pholus_output_list
- mylog('verbose', ["[PHOLUS] Pholus output number of entries:", len(result_list)])
- mylog('verbose', ["[PHOLUS] List:", result_list])
+ mylog('verbose', [f'[{pluginName}] Pholus output number of entries:', len(result_list)])
+ mylog('verbose', [f'[{pluginName}] List:', result_list])
return result_list
@@ -132,8 +132,8 @@ def execute_pholus_on_interface(interface, timeoutSec, mask):
# logging & updating app state
- mylog('verbose', ['[PHOLUS] Scan: Pholus for ', str(timeoutSec), 's ('+ str(round(int(timeoutSec) / 60, 1)) +'min)'])
- mylog('verbose', ["[PHOLUS] Pholus scan on [interface] ", interface, " [mask] " , mask])
+ mylog('verbose', [f'[{pluginName}] Scan: Pholus for ', str(timeoutSec), 's ('+ str(round(int(timeoutSec) / 60, 1)) +'min)'])
+ mylog('verbose', [f'[{pluginName}] 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
adjustedTimeout = str(round(int(timeoutSec) / 2, 0))
@@ -149,15 +149,15 @@ def execute_pholus_on_interface(interface, timeoutSec, mask):
output = subprocess.check_output (pholus_args, universal_newlines=True, stderr=subprocess.STDOUT, timeout=(timeoutSec + 30))
except subprocess.CalledProcessError as e:
# An error occured, handle it
- mylog('none', ['[PHOLUS]', e.output])
- mylog('none', ["[PHOLUS] ⚠ ERROR - Pholus Scan - check logs"])
+ mylog('verbose', [f'[{pluginName}]', e.output])
+ mylog('verbose', [f'[{pluginName}] ⚠ ERROR - Pholus Scan - check logs'])
except subprocess.TimeoutExpired as timeErr:
- mylog('none', ['[PHOLUS] Pholus TIMEOUT - the process forcefully terminated as timeout reached'])
+ mylog('verbose', [f'[{pluginName}] Pholus TIMEOUT - the process forcefully terminated as timeout reached'])
if output == "": # check if the subprocess failed
- mylog('none', ['[PHOLUS] Scan: Pholus FAIL - check logs'])
+ mylog('verbose', [f'[{pluginName}] Scan: Pholus FAIL - check logs'])
else:
- mylog('verbose', ['[PHOLUS] Scan: Pholus SUCCESS'])
+ mylog('verbose', [f'[{pluginName}] Scan: Pholus SUCCESS'])
# check the last run output
f = open(logPath + '/pialert_pholus_lastrun.log', 'r+')
diff --git a/pialert/device.py b/pialert/device.py
index e47cb69a..0f02b344 100755
--- a/pialert/device.py
+++ b/pialert/device.py
@@ -8,8 +8,28 @@ from logger import mylog, print_log
from const import vendorsPath
#-------------------------------------------------------------------------------
+# Device object handling (WIP)
+#-------------------------------------------------------------------------------
+class Device_obj:
+ def __init__(self, db):
+ self.db = db
+
+ # Get all
+ def getAll(self):
+ self.db.sql.execute("""
+ SELECT * FROM Devices
+ """)
+ return self.db.sql.fetchall()
+
+ # Get all with unknown names
+ def getUnknown(self):
+ self.db.sql.execute("""
+ SELECT * FROM Devices WHERE dev_Name in ("(unknown)", "(name not found)", "" )
+ """)
+ return self.db.sql.fetchall()
+#-------------------------------------------------------------------------------
def save_scanned_devices (db):
sql = db.sql #TO-DO