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

API v0.1

Browse Source
This commit is contained in:
Jokob-sk
2023-01-29 00:07:46 +11:00
parent 2c27248aa1
commit 51e865c98d
4 changed files with 179 additions and 72 deletions
Download Patch File Download Diff File Expand all files Collapse all files

1
.gitignore vendored
View File

@@ -2,5 +2,6 @@
config/pialert.conf
db/*
front/log/*
front/api/*
**/%40eaDir/
**/@eaDir/

226
back/pialert.py
View File

@@ -39,6 +39,16 @@ from pathlib import Path
from cron_converter import Cron
from pytz import timezone
#===============================================================================
# SQL queries
#===============================================================================
sql_devices_all = "select dev_MAC, dev_Name, dev_DeviceType, dev_Vendor, dev_Group, dev_FirstConnection, dev_LastConnection, dev_LastIP, dev_StaticIP, dev_PresentLastScan, dev_LastNotification, dev_NewDevice, dev_Network_Node_MAC_ADDR, dev_Network_Node_port, dev_Icon from Devices"
sql_devices_stats = "SELECT Online_Devices as online, Down_Devices as down, All_Devices as 'all', Archived_Devices as archived, (select count(*) from Devices a where dev_NewDevice = 1 ) as new, (select count(*) from Devices a where dev_Name = '(unknown)' or dev_Name = '(name not found)' ) as unknown from Online_History order by Scan_Date desc limit 1"
sql_nmap_scan_all = "SELECT * FROM Nmap_Scan"
sql_pholus_scan_all = "SELECT * FROM Pholus_Scan"
sql_events_pending_alert = "SELECT * FROM Events where eve_PendingAlertEmail is not 0"
#===============================================================================
# PATHS
#===============================================================================
@@ -55,6 +65,8 @@ piholeDhcpleases = '/etc/pihole/dhcp.leases'
# Global variables
debug_force_notification = False
userSubnets = []
time_started = datetime.datetime.now()
cron_instance = Cron()
@@ -535,10 +547,10 @@ def main ():
AND eve_EventType = 'New Device'
ORDER BY eve_DateTime""")
rows = sql.fetchall()
newDevices = sql.fetchall()
commitDB()
performNmapScan(rows)
performNmapScan(newDevices)
# send all configured notifications
send_notifications()
@@ -1983,8 +1995,7 @@ def skip_repeated_notifications ():
json_final = []
def send_notifications ():
global mail_text
global mail_html
global mail_text, mail_html, json_final
deviceUrl = REPORT_DASHBOARD_URL + '/deviceDetails.php?mac='
@@ -2171,13 +2182,15 @@ def send_notifications ():
"events": json_events
}
# DEBUG - Write output emails for testing
#if True :
# Write output emails for testing
write_file (logPath + '/report_output.txt', mail_text)
write_file (logPath + '/report_output.html', mail_html)
# Send Mail
if json_internet != [] or json_new_devices != [] or json_down_devices != [] or json_events != []:
if json_internet != [] or json_new_devices != [] or json_down_devices != [] or json_events != [] or debug_force_notification:
update_api()
file_print(' Changes detected, sending reports')
if REPORT_MAIL and check_config('email'):
@@ -2279,42 +2292,7 @@ def check_config(service):
else:
return True
#-------------------------------------------------------------------------------
def send_ntfy (_Text):
headers = {
"Title": "Pi.Alert Notification",
"Actions": "view, Open Dashboard, "+ REPORT_DASHBOARD_URL,
"Priority": "urgent",
"Tags": "warning"
}
# if username and password are set generate hash and update header
if NTFY_USER != "" and NTFY_PASSWORD != "":
# Generate hash for basic auth
usernamepassword = "{}:{}".format(NTFY_USER,NTFY_PASSWORD)
basichash = b64encode(bytes(NTFY_USER + ':' + NTFY_PASSWORD, "utf-8")).decode("ascii")
# add authorization header with hash
headers["Authorization"] = "Basic {}".format(basichash)
requests.post("{}/{}".format( NTFY_HOST, NTFY_TOPIC),
data=_Text,
headers=headers)
def send_pushsafer (_Text):
url = 'https://www.pushsafer.com/api'
post_fields = {
"t" : 'Pi.Alert Message',
"m" : _Text,
"s" : 11,
"v" : 3,
"i" : 148,
"c" : '#ef7f7f',
"d" : 'a',
"u" : REPORT_DASHBOARD_URL,
"ut" : 'Open Pi.Alert',
"k" : PUSHSAFER_TOKEN,
}
requests.post(url, data=post_fields)
#-------------------------------------------------------------------------------
def format_report_section (pActive, pSection, pTable, pText, pHTML):
@@ -2349,30 +2327,9 @@ def remove_tag (pText, pTag):
# return text without the tag
return pText.replace ('<'+ pTag +'>','').replace ('</'+ pTag +'>','')
#-------------------------------------------------------------------------------
def write_file (pPath, pText):
# Write the text depending using the correct python version
if sys.version_info < (3, 0):
file = io.open (pPath , mode='w', encoding='utf-8')
file.write ( pText.decode('unicode_escape') )
file.close()
else:
file = open (pPath, 'w', encoding='utf-8')
file.write (pText)
file.close()
#-------------------------------------------------------------------------------
def append_line_to_file (pPath, pText):
# append the line depending using the correct python version
if sys.version_info < (3, 0):
file = io.open (pPath , mode='a', encoding='utf-8')
file.write ( pText.decode('unicode_escape') )
file.close()
else:
file = open (pPath, 'a', encoding='utf-8')
file.write (pText)
file.close()
# Reporting
#-------------------------------------------------------------------------------
def send_email (pText, pHTML):
@@ -2414,7 +2371,10 @@ def send_email (pText, pHTML):
smtp_connection = smtplib.SMTP (SMTP_SERVER, SMTP_PORT)
failedAt = print_log('Setting SMTP debug level')
smtp_connection.set_debuglevel(1)
# Verbose debug of the communication between SMTP server and client
if PRINT_LOG:
smtp_connection.set_debuglevel(1)
failedAt = print_log( 'Sending .ehlo()')
smtp_connection.ehlo()
@@ -2438,8 +2398,44 @@ def send_email (pText, pHTML):
file_print(' ERROR: Failed at - ', failedAt)
file_print(' ERROR: Couldn\'t connect to the SMTP server (SMTPServerDisconnected), skipping Email (enable PRINT_LOG for more logging)')
file_print(' DEBUG: Last executed - ', failedAt)
print_log(' DEBUG: Last executed - ' + str(failedAt))
#-------------------------------------------------------------------------------
def send_ntfy (_Text):
headers = {
"Title": "Pi.Alert Notification",
"Actions": "view, Open Dashboard, "+ REPORT_DASHBOARD_URL,
"Priority": "urgent",
"Tags": "warning"
}
# if username and password are set generate hash and update header
if NTFY_USER != "" and NTFY_PASSWORD != "":
# Generate hash for basic auth
usernamepassword = "{}:{}".format(NTFY_USER,NTFY_PASSWORD)
basichash = b64encode(bytes(NTFY_USER + ':' + NTFY_PASSWORD, "utf-8")).decode("ascii")
# add authorization header with hash
headers["Authorization"] = "Basic {}".format(basichash)
requests.post("{}/{}".format( NTFY_HOST, NTFY_TOPIC),
data=_Text,
headers=headers)
def send_pushsafer (_Text):
url = 'https://www.pushsafer.com/api'
post_fields = {
"t" : 'Pi.Alert Message',
"m" : _Text,
"s" : 11,
"v" : 3,
"i" : 148,
"c" : '#ef7f7f',
"d" : 'a',
"u" : REPORT_DASHBOARD_URL,
"ut" : 'Open Pi.Alert',
"k" : PUSHSAFER_TOKEN,
}
requests.post(url, data=post_fields)
#-------------------------------------------------------------------------------
def send_webhook (_json, _html):
@@ -2507,6 +2503,8 @@ def send_apprise (html):
# An error occured, handle it
file_print(e.output)
#-------------------------------------------------------------------------------
# MQTT
#-------------------------------------------------------------------------------
mqtt_connected_to_broker = False
mqtt_sensors = []
@@ -2916,9 +2914,89 @@ def to_binary_sensor(input):
#===============================================================================
# API
#===============================================================================
def update_api():
file_print(' [API] Updating files in /front/api')
folder = pialertPath + '/front/api/'
write_file(folder + 'notification_text.txt' , mail_text)
write_file(folder + 'notification_text.html' , mail_html)
write_file(folder + 'notification_json_final.json' , json.dumps(json_final))
dataSourcesSQLs = [
["devices", sql_devices_all],
["nmap_scan", sql_nmap_scan_all],
["pholus_scan", sql_pholus_scan_all],
["events_pending_alert", sql_events_pending_alert]
]
# Save selected database tables
for dsSQL in dataSourcesSQLs:
sql.execute(dsSQL[1])
columnNames = list(map(lambda x: x[0], sql.description))
rows = sql.fetchall()
json_string = get_table_as_json(rows, columnNames)
write_file(folder + 'table_' + dsSQL[0] + '.json' , json.dumps(json_string))
#-------------------------------------------------------------------------------
def get_table_as_json(rows, names):
result = {"data":[]}
for row in rows:
tmp = fill_row(names, row)
result["data"].append(tmp)
return result
#-------------------------------------------------------------------------------
def fill_row(names, row):
rowEntry = {}
index = 0
for name in names:
rowEntry[name]= if_byte_then_to_str(row[name])
index += 1
return rowEntry
#===============================================================================
# UTIL
#===============================================================================
#-------------------------------------------------------------------------------
def write_file (pPath, pText):
# Write the text depending using the correct python version
if sys.version_info < (3, 0):
file = io.open (pPath , mode='w', encoding='utf-8')
file.write ( pText.decode('unicode_escape') )
file.close()
else:
file = open (pPath, 'w', encoding='utf-8')
file.write (pText)
file.close()
#-------------------------------------------------------------------------------
def append_line_to_file (pPath, pText):
# append the line depending using the correct python version
if sys.version_info < (3, 0):
file = io.open (pPath , mode='a', encoding='utf-8')
file.write ( pText.decode('unicode_escape') )
file.close()
else:
file = open (pPath, 'a', encoding='utf-8')
file.write (pText)
file.close()
#-------------------------------------------------------------------------------
# Make a regular expression
# for validating an Ip-address
@@ -2970,6 +3048,14 @@ def sanitize_string(input):
#-------------------------------------------------------------------------------
def if_byte_then_to_str(input):
if isinstance(input, bytes):
input = input.decode('utf-8')
input = bytes_to_string(re.sub('[^a-zA-Z0-9-_\s]', '', str(input)))
return input
#-------------------------------------------------------------------------------
def bytes_to_string(value):
# if value is of type bytes, convert to string
if isinstance(value, bytes):
@@ -3025,9 +3111,7 @@ def to_text(_json):
def get_device_stats():
# columns = ["online","down","all","archived","new","unknown"]
sql.execute("""
SELECT Online_Devices as online, Down_Devices as down, All_Devices as 'all', Archived_Devices as archived, (select count(*) from Devices a where dev_NewDevice = 1 ) as new, (select count(*) from Devices a where dev_Name = '(unknown)' or dev_Name = '(name not found)' ) as unknown from Online_History order by Scan_Date desc limit 1
""")
sql.execute(sql_devices_stats)
row = sql.fetchone()
commitDB()
@@ -3036,9 +3120,7 @@ def get_device_stats():
#-------------------------------------------------------------------------------
def get_all_devices():
sql.execute("""
select dev_MAC, dev_Name, dev_DeviceType, dev_Vendor, dev_Group, dev_FirstConnection, dev_LastConnection, dev_LastIP, dev_StaticIP, dev_PresentLastScan, dev_LastNotification, dev_NewDevice, dev_Network_Node_MAC_ADDR from Devices
""")
sql.execute(sql_devices_all)
row = sql.fetchall()

1
docker-compose.yml
View File

@@ -19,6 +19,7 @@ services:
- ${DEV_LOCATION}/pholus:/home/pi/pialert/pholus
- ${DEV_LOCATION}/dockerfiles:/home/pi/pialert/dockerfiles
- ${APP_DATA_LOCATION}/pialert/php.ini:/etc/php/7.4/fpm/php.ini
- ${DEV_LOCATION}/front/api:/home/pi/pialert/front/api
- ${DEV_LOCATION}/front/css:/home/pi/pialert/front/css
- ${DEV_LOCATION}/front/lib/AdminLTE:/home/pi/pialert/front/lib/AdminLTE
- ${DEV_LOCATION}/front/js:/home/pi/pialert/front/js

23
docs/API.md Executable file
View File

@@ -0,0 +1,23 @@
## Where are API endpoints located
PiAlert comes with a simple API. These API endpoints are static files, which are updated during:
1) A notification event
2) TBD
In the container, these files are located under the `/home/pi/pialert/front/api/` folder and thus on the `<pialert_url>/api/<File name>` url.
You can access the following files:
| File name | Description |
|----------------------|----------------------|
| `notification_text.txt` | The plain text version of the last notification. |
| `notification_text.html` | The full HTML of the last email notification. |
| `notification_json_final.json` | The json version of the last notification (e.g. used for webhooks - [sample JSON](https://github.com/jokob-sk/Pi.Alert/blob/main/back/webhook_json_sample.json)). |
| `table_devices.json` | The current (at the time of the last update as mentioned above on this page) state of all of the available Devices detected by the app. |
| `table_nmap_scan.json` | The current state of the discovered ports by the regular NMAP scans. |
| `pholus_scan.json` | The latest state of the [pholus](https://github.com/jokob-sk/Pi.Alert/tree/main/pholus) (A multicast DNS and DNS Service Discovery Security Assessment Tool) scan results. |
| `table_events_pending_alert.json` | The list of the unprocessed (pending) notification events. |
Current/latest state of the aforementioned files depends on your settings.