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

PLUGINS, NTFY, handleEmpty work⤵

Browse Source
This commit is contained in:
Jokob-sk
2023-10-19 08:08:24 +11:00
parent 1a3cf49c00
commit 5d64433be0
18 changed files with 594 additions and 128 deletions
Download Patch File Download Diff File Expand all files Collapse all files

5
dockerfiles/start.sh
View File

@@ -30,11 +30,6 @@ echo "[INSTALL] Run setup scripts"
"$INSTALL_DIR/pialert/install/install_dependencies.sh"
"$INSTALL_DIR/pialert/install/install_python.sh"
# # executes a new shell session with the user specified in the USER variable.
# if [ -n "$USER" ]; then
# exec su - "${USER}"
# fi
# Change port number if set
if [ -n "${PORT}" ]; then
sed -ie 's/listen 20211/listen '${PORT}'/g' /etc/nginx/sites-available/default

13
front/php/templates/language/en_us.json
View File

@@ -533,18 +533,7 @@
"WEBHOOK_SIZE_description" : "The maximum size of the webhook payload as number of characters in the passed string. If above limit, it will be truncated and a <code>(text was truncated)</code> message is appended.",
"WEBHOOK_SECRET_name": "HMAC Secret",
"WEBHOOK_SECRET_description": "When set, use this secret to generate the SHA256-HMAC hex digest value of the request body, which will be passed as the <code>X-Webhook-Signature</code> header to the request. You can find more informations <a target=\"_blank\" href=\"https://github.com/jokob-sk/Pi.Alert/blob/main/docs/WEBHOOK_SECRET.md\">here</a>.",
"NTFY_display_name" : "NTFY",
"NTFY_icon" : "<i class=\"fa fa-terminal\"></i>",
"REPORT_NTFY_name" : "Enable NTFY",
"REPORT_NTFY_description" : "Enable sending notifications via <a target=\"_blank\" href=\"https://ntfy.sh/\">NTFY</a>.",
"NTFY_HOST_name" : "NTFY host URL",
"NTFY_HOST_description" : "NTFY host URL starting with <code>http://</code> or <code>https://</code>. You can use the hosted instance on <a target=\"_blank\" href=\"https://ntfy.sh/\">https://ntfy.sh</a> by simply entering <code>https://ntfy.sh</code>.",
"NTFY_TOPIC_name" : "NTFY topic",
"NTFY_TOPIC_description" : "Your secret topic.",
"NTFY_USER_name" : "NTFY user",
"NTFY_USER_description" : "Enter user if you need (host) an instance with enabled authetication.",
"NTFY_PASSWORD_name" : "NTFY password",
"NTFY_PASSWORD_description" : "Enter password if you need (host) an instance with enabled authetication.",
"PUSHSAFER_display_name" : "Pushsafer",
"PUSHSAFER_icon" : "<i class=\"fa fa-bell\"></i>",
"REPORT_PUSHSAFER_name" : "Enable Pushsafer",

2
front/plugins/_publisher_apprise/apprise.py
View File

@@ -61,7 +61,7 @@ def main():
watched3 = 'null',
watched4 = 'null',
extra = 'null',
foreignKey = 'null'
foreignKey = notification["GUID"]
)
plugin_objects.write_result_file()

40
front/plugins/_publisher_email/email_smtp.py
View File

@@ -5,9 +5,12 @@ import argparse
import os
import pathlib
import sys
import re
from datetime import datetime
from email.mime.multipart import MIMEMultipart
from email.mime.text import MIMEText
from email.header import Header
from email.utils import parseaddr
import smtplib
import socket
import ssl
@@ -66,7 +69,7 @@ def main():
watched3 = 'null',
watched4 = 'null',
extra = 'null',
foreignKey = 'null'
foreignKey = notification["GUID"]
)
plugin_objects.write_result_file()
@@ -89,13 +92,16 @@ def send(pHTML, pText):
mylog('debug', [f'[{pluginName}] SMTP_REPORT_TO: {hide_email(str(get_setting_value("SMTP_REPORT_TO")))} SMTP_USER: {hide_email(str(get_setting_value("SMTP_USER")))}'])
subject, from_email, to_email, message_html, message_text = sanitize_email_content('Pi.Alert Report', get_setting_value("SMTP_REPORT_FROM"), get_setting_value("SMTP_REPORT_TO"), pHTML, pText)
# Compose email
msg = MIMEMultipart('alternative')
msg['Subject'] = 'Pi.Alert Report'
msg['From'] = get_setting_value("SMTP_REPORT_FROM")
msg['To'] = get_setting_value("SMTP_REPORT_TO")
msg.attach (MIMEText (pText, 'plain'))
msg.attach (MIMEText (pHTML, 'html'))
msg['Subject'] = subject
msg['From'] = from_email
msg['To'] = to_email
msg.attach (MIMEText (message_text, 'plain'))
msg.attach (MIMEText (message_html, 'html'))
# Set a timeout for the SMTP connection (in seconds)
smtp_timeout = 30
@@ -126,6 +132,7 @@ def send(pHTML, pText):
mylog('none', [' ERROR: Are you sure you need SMTP_FORCE_SSL enabled? Check your SMTP provider docs.'])
mylog('none', [' ERROR: ', str(e)])
# ----------------------------------------------------------------------------------
def send_email(msg):
# Send mail
if get_setting_value('SMTP_FORCE_SSL'):
@@ -168,5 +175,26 @@ def send_email(msg):
smtp_connection.sendmail (get_setting_value("SMTP_REPORT_FROM"), get_setting_value("SMTP_REPORT_TO"), msg.as_string())
smtp_connection.quit()
# ----------------------------------------------------------------------------------
def sanitize_email_content(subject, from_email, to_email, message_html, message_text):
# Validate and sanitize subject
subject = Header(subject, 'utf-8').encode()
# Validate and sanitize sender's email address
from_name, from_address = parseaddr(from_email)
from_email = Header(from_name, 'utf-8').encode() + ' <' + from_address + '>'
# Validate and sanitize recipient's email address
to_name, to_address = parseaddr(to_email)
to_email = Header(to_name, 'utf-8').encode() + ' <' + to_address + '>'
# Validate and sanitize message content
# Remove potentially problematic characters
message_html = re.sub(r'[^\x00-\x7F]+', ' ', message_html)
message_text = re.sub(r'[^\x00-\x7F]+', ' ', message_text)
return subject, from_email, to_email, message_html, message_text
# ----------------------------------------------------------------------------------
if __name__ == '__main__':
sys.exit(main())

2
front/plugins/_publisher_mqtt/mqtt.py
View File

@@ -125,7 +125,7 @@ class sensor_config:
watched3 = hash_value,
watched4 = mac,
extra = input_string,
foreignKey = deviceId
foreignKey = notification["GUID"]
)
#-------------------------------------------------------------------------------

8
front/plugins/_publisher_ntfy/README.md Executable file
View File

@@ -0,0 +1,8 @@
## Overview
- TBC
### Usage
- Go to settings and fill in relevant details.

385
front/plugins/_publisher_ntfy/config.json Executable file
View File

@@ -0,0 +1,385 @@
{
"code_name": "_publisher_ntfy",
"unique_prefix": "NTFY",
"plugin_type": "publisher",
"enabled": true,
"data_source": "script",
"show_ui": true,
"localized": ["display_name", "description", "icon"],
"display_name" : [
{
"language_code": "en_us",
"string" : "NTFY publisher"
},
{
"language_code": "es_es",
"string" : "Habilitar NTFY"
}
],
"icon":[{
"language_code": "en_us",
"string" : "<i class=\"fa-solid fa-terminal\"></i>"
}],
"description": [
{
"language_code": "en_us",
"string" : "A plugin to publish a notification via the NTFY gateway."
}
],
"params" : [
],
"database_column_definitions":
[
{
"column": "Index",
"css_classes": "col-sm-2",
"show": false,
"type": "label",
"default_value":"",
"options": [],
"localized": ["name"],
"name":[{
"language_code": "en_us",
"string" : "N/A"
},
{
"language_code": "es_es",
"string" : "N/A"
}]
},
{
"column": "Plugin",
"css_classes": "col-sm-2",
"show": false,
"type": "label",
"default_value":"",
"options": [],
"localized": ["name"],
"name":[{
"language_code": "en_us",
"string" : "N/A"
},
{
"language_code": "es_es",
"string" : "N/A"
}]
},
{
"column": "Object_PrimaryID",
"css_classes": "col-sm-2",
"show": false,
"type": "label",
"default_value":"",
"options": [],
"localized": ["name"],
"name":[{
"language_code": "en_us",
"string" : "N/A"
}]
},
{
"column": "Object_SecondaryID",
"css_classes": "col-sm-2",
"show": true,
"type": "label",
"default_value":"",
"options": [],
"localized": ["name"],
"name":[{
"language_code": "en_us",
"string" : "Sent when"
}]
},
{
"column": "Watched_Value1",
"css_classes": "col-sm-3",
"show": true,
"type": "label",
"default_value":"",
"options": [],
"localized": ["name"],
"name":[{
"language_code": "en_us",
"string" : "Notification GUID"
}]
},
{
"column": "Watched_Value2",
"css_classes": "col-sm-2",
"show": true,
"type": "textarea_readonly",
"default_value":"",
"options": [],
"localized": ["name"],
"name":[{
"language_code": "en_us",
"string" : "Response"
}]
},
{
"column": "Watched_Value3",
"css_classes": "col-sm-2",
"show": true,
"type": "label",
"default_value":"",
"options": [],
"localized": ["name"],
"name":[{
"language_code": "en_us",
"string" : "Response code"
}]
},
{
"column": "Watched_Value4",
"css_classes": "col-sm-2",
"show": false,
"type": "device_mac",
"default_value":"",
"options": [],
"localized": ["name"],
"name":[{
"language_code": "en_us",
"string" : "Device"
}
]
},
{
"column": "UserData",
"css_classes": "col-sm-2",
"show": false,
"type": "textbox_save",
"default_value":"",
"options": [],
"localized": ["name"],
"name":[{
"language_code": "en_us",
"string" : "Comments"
},
{
"language_code": "es_es",
"string" : "Comentarios"
}]
},
{
"column": "Status",
"css_classes": "col-sm-1",
"show": false,
"type": "replace",
"default_value":"",
"options": [
{
"equals": "watched-not-changed",
"replacement": "<div style='text-align:center'><i class='fa-solid fa-square-check'></i><div></div>"
},
{
"equals": "watched-changed",
"replacement": "<div style='text-align:center'><i class='fa-solid fa-triangle-exclamation'></i></div>"
},
{
"equals": "new",
"replacement": "<div style='text-align:center'><i class='fa-solid fa-circle-plus'></i></div>"
},
{
"equals": "missing-in-last-scan",
"replacement": "<div style='text-align:center'><i class='fa-solid fa-question'></i></div>"
}
],
"localized": ["name"],
"name":[{
"language_code": "en_us",
"string" : "Status"
},
{
"language_code": "es_es",
"string" : "Estado"
}]
},
{
"column": "Extra",
"css_classes": "col-sm-3",
"show": false,
"type": "label",
"default_value":"",
"options": [],
"localized": ["name"],
"name":[{
"language_code": "en_us",
"string" : "Extra"
},
{
"language_code": "es_es",
"string" : "Extra"
}]
}
],
"settings":[
{
"function": "RUN",
"events": ["test"],
"type": "text.select",
"default_value":"disabled",
"options": ["disabled", "on_notification" ],
"localized": ["name", "description"],
"name" :[{
"language_code": "en_us",
"string" : "When to run"
},
{
"language_code": "es_es",
"string" : "Cuando ejecuta"
}],
"description": [
{
"language_code": "en_us",
"string" : "Enable sending notifications via <a target=\"_blank\" href=\"https://ntfy.sh/\">NTFY</a>."
},
{
"language_code": "es_es",
"string" : "Habilitar el envío de notificaciones a través de <a target=\"_blank\" href=\"https://ntfy.sh/\">NTFY</a>."
}
]
},
{
"function": "CMD",
"type": "readonly",
"default_value":"python3 /home/pi/pialert/front/plugins/_publisher_ntfy/ntfy.py",
"options": [],
"localized": ["name", "description"],
"name" : [{
"language_code": "en_us",
"string" : "Command"
},
{
"language_code": "es_es",
"string" : "Comando"
}],
"description": [{
"language_code": "en_us",
"string" : "Command to run"
},
{
"language_code": "es_es",
"string" : "Comando a ejecutar"
}]
},
{
"function": "RUN_TIMEOUT",
"type": "integer",
"default_value": 10,
"options": [],
"localized": ["name", "description"],
"name" : [{
"language_code": "en_us",
"string" : "Run timeout"
},
{
"language_code": "es_es",
"string" : "Tiempo de espera de ejecución"
},
{
"language_code": "de_de",
"string" : "Wartezeit"
}],
"description": [{
"language_code": "en_us",
"string" : "Maximum time in seconds to wait for the script to finish. If this time is exceeded the script is aborted."
},
{
"language_code": "es_es",
"string" : "Tiempo máximo en segundos para esperar a que finalice el script. Si se supera este tiempo, el script se cancela."
}]
},
{
"function": "HOST",
"type": "text",
"default_value": "https://ntfy.sh",
"options": [],
"localized": ["name", "description"],
"name" : [{
"language_code": "en_us",
"string" : "NTFY host URL"
},
{
"language_code": "es_es",
"string" : "URL del host NTFY"
}],
"description": [{
"language_code": "en_us",
"string" : "NTFY host URL starting with <code>http://</code> or <code>https://</code>. You can use the hosted instance on <a target=\"_blank\" href=\"https://ntfy.sh/\">https://ntfy.sh</a> by simply entering <code>https://ntfy.sh</code>."
},
{
"language_code": "es_es",
"string" : "URL de host NTFY que comienza con <code>http://</code> o <code>https://</code>. Puede usar la instancia alojada en <a target=\"_blank\" href=\"https://ntfy.sh/\">https://ntfy.sh</a> simplemente ingresando <code>https://ntfy. sh</código>."
}]
},
{
"function": "TOPIC",
"type": "text",
"default_value": "",
"options": [],
"localized": ["name", "description"],
"name" : [{
"language_code": "en_us",
"string" : "NTFY topic"
},
{
"language_code": "es_es",
"string" : "Tema de NTFY"
}],
"description": [{
"language_code": "en_us",
"string" : "Your secret topic."
},
{
"language_code": "es_es",
"string" : "Tu tema secreto."
}]
},
{
"function": "USER",
"type": "text",
"default_value": "",
"options": [],
"localized": ["name", "description"],
"name" : [{
"language_code": "en_us",
"string" : "NTFY user"
},
{
"language_code": "es_es",
"string" : "Usuario de NTFY"
}],
"description": [{
"language_code": "en_us",
"string" : "Enter user if you need (host) an instance with enabled authetication."
},
{
"language_code": "es_es",
"string" : "Ingrese usuario si necesita (alojar) una instancia con autenticación habilitada."
}]
},
{
"function": "PASSWORD",
"type": "password",
"default_value": "",
"options": [],
"localized": ["name", "description"],
"name" : [{
"language_code": "en_us",
"string" : "NTFY password"
},
{
"language_code": "es_es",
"string" : "Contraseña de NTFY"
}],
"description": [{
"language_code": "en_us",
"string" : "Enter password if you need (host) an instance with enabled authetication."
},
{
"language_code": "es_es",
"string" : "Ingrese la contraseña si necesita (host) una instancia con autenticación habilitada."
}]
}
]
}

1
front/plugins/_publisher_ntfy/ignore_plugin
View File

@@ -1 +0,0 @@
This plugin will not be loaded

128
front/plugins/_publisher_ntfy/ntfy.py Executable file
View File

@@ -0,0 +1,128 @@
#!/usr/bin/env python
import json
import subprocess
import argparse
import os
import pathlib
import sys
import requests
from datetime import datetime
from base64 import b64encode
# Replace these paths with the actual paths to your Pi.Alert directories
sys.path.extend(["/home/pi/pialert/front/plugins", "/home/pi/pialert/pialert"])
import conf
from plugin_helper import Plugin_Objects
from logger import mylog, append_line_to_file
from helper import timeNowTZ, noti_obj, get_setting_value
from notification import Notification_obj
from database import DB
CUR_PATH = str(pathlib.Path(__file__).parent.resolve())
RESULT_FILE = os.path.join(CUR_PATH, 'last_result.log')
pluginName = 'NTFY'
def main():
mylog('verbose', [f'[{pluginName}](publisher) In script'])
# Check if basic config settings supplied
if check_config() == False:
mylog('none', [f'[{pluginName}] Error: Publisher notification gateway not set up correctly. Check your pialert.conf {pluginName}_* variables.'])
return
# 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 Notification_obj instance
notifications = Notification_obj(db)
# Retrieve new notifications
new_notifications = notifications.getNew()
# Process the new notifications
for notification in new_notifications:
# Send notification
response_text, response_status_code = send(notification["HTML"], notification["Text"])
# Log result
plugin_objects.add_object(
primaryId = pluginName,
secondaryId = timeNowTZ(),
watched1 = notification["GUID"],
watched2 = response_text,
watched3 = response_status_code,
watched4 = 'null',
extra = 'null',
foreignKey = notification["GUID"]
)
plugin_objects.write_result_file()
#-------------------------------------------------------------------------------
def check_config():
if get_setting_value('NTFY_HOST') == '' or get_setting_value('NTFY_TOPIC') == '':
return False
else:
return True
#-------------------------------------------------------------------------------
def send(html, text):
response_text = ''
response_status_code = ''
headers = {
"Title": "Pi.Alert Notification",
"Actions": "view, Open Dashboard, "+ get_setting_value('REPORT_DASHBOARD_URL'),
"Priority": "urgent",
"Tags": "warning"
}
# if username and password are set generate hash and update header
if get_setting_value('NTFY_USER') != "" and get_setting_value('NTFY_PASSWORD') != "":
# Generate hash for basic auth
# usernamepassword = "{}:{}".format(get_setting_value('NTFY_USER'),get_setting_value('NTFY_PASSWORD'))
basichash = b64encode(bytes(get_setting_value('NTFY_USER') + ':' + get_setting_value('NTFY_PASSWORD'), "utf-8")).decode("ascii")
# add authorization header with hash
headers["Authorization"] = "Basic {}".format(basichash)
try:
response = requests.post("{}/{}".format( get_setting_value('NTFY_HOST'),
get_setting_value('NTFY_TOPIC')),
data = text,
headers = headers)
response_status_code = response.status_code
# Check if the request was successful (status code 200)
if response_status_code == 200:
response_text = response.text # This captures the response body/message
except requests.exceptions.RequestException as e:
mylog('none', [f'[{pluginName}] Error: ', e])
response_text = e
return response_text, response_status_code
return response_text, response_status_code
if __name__ == '__main__':
sys.exit(main())

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

@@ -12,7 +12,7 @@ from time import strftime
sys.path.append("/home/pi/pialert/front/plugins")
sys.path.append('/home/pi/pialert/pialert')
from plugin_helper import Plugin_Object, Plugin_Objects
from plugin_helper import Plugin_Object, Plugin_Objects, handleEmpty
from logger import mylog, append_line_to_file
from helper import timeNowTZ
from const import logPath, pialertPath
@@ -70,11 +70,11 @@ def main():
for device in unique_devices:
plugin_objects.add_object(
primaryId=device['mac'], # MAC (Device Name)
secondaryId=device['ip'], # IP Address
watched1=device['ip'], # Device Name
watched2=device.get('hw', ''), # Vendor (assuming it's in the 'hw' field)
watched3=device.get('interface', ''), # Add the interface
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 = 'arp-scan',
foreignKey = "")

5
front/plugins/plugin_helper.py
View File

@@ -1,6 +1,7 @@
from time import strftime
import pytz
import sys
import re
import base64
from datetime import datetime
@@ -38,6 +39,10 @@ def handleEmpty(input):
if input == '' or None:
return 'null'
else:
# Validate and sanitize message content
# Remove potentially problematic characters if string
if isinstance(input, str):
input = re.sub(r'[^\x00-\x7F]+', ' ', input)
return input
# -------------------------------------------------------------------

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

@@ -76,14 +76,13 @@ def main():
mylog('verbose', [f'[SNMPDSC] IP: {ipAddress} MAC: {macAddress}'])
plugin_objects.add_object(
primaryId=macAddress,
secondaryId=ipAddress.strip(), # Remove leading/trailing spaces from IP
primaryId = handleEmpty(macAddress),
secondaryId = handleEmpty(ipAddress.strip()), # Remove leading/trailing spaces from IP
watched1 = '(unknown)',
watched2=snmpwalkArgs[6], # router IP
extra=line,
foreignKey=macAddress # Use the primary ID as the foreign key
watched2 = handleEmpty(snmpwalkArgs[6]), # router IP
extra = handleEmpty(line),
foreignKey = handleEmpty(macAddress) # Use the primary ID as the foreign key
)
mylog('verbose', ['[SNMPDSC] Entries found: ', len(plugin_objects)])

12
front/plugins/vendor_update/script.py
View File

@@ -16,7 +16,7 @@ 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 plugin_helper import Plugin_Object, Plugin_Objects, decodeBase64, handleEmpty
from logger import mylog, append_line_to_file
from helper import timeNowTZ
from const import logPath, pialertPath
@@ -106,14 +106,14 @@ def update_vendors (dbPath, plugin_objects):
ignored += 1
else :
plugin_objects.add_object(
primaryId = device[0], # MAC (Device Name)
secondaryId = device[1], # IP Address (always 0.0.0.0)
watched1 = vendor,
watched2 = device[2], # Device name
primaryId = handleEmpty(device[0]), # MAC (Device Name)
secondaryId = handleEmpty(device[1]), # IP Address (always 0.0.0.0)
watched1 = handleEmpty(vendor),
watched2 = handleEmpty(device[2]), # Device name
watched3 = "",
watched4 = "",
extra = "",
foreignKey = device[0]
foreignKey = handleEmpty(device[0])
)
# Print log

18
pialert/conf.py
View File

@@ -45,18 +45,6 @@ REPORT_DASHBOARD_URL = 'http://pi.alert/'
# Notification gateways
# -------------------------------------------
# Email
REPORT_MAIL = False
SMTP_SERVER = ''
SMTP_PORT = 587
REPORT_TO = 'user@gmail.com'
REPORT_FROM = 'Pi.Alert <user@gmail.com>'
SMTP_SKIP_LOGIN = False
SMTP_USER = ''
SMTP_PASS = ''
SMTP_SKIP_TLS = False
SMTP_FORCE_SSL = False
# Webhooks
REPORT_WEBHOOK = False
WEBHOOK_URL = ''
@@ -64,12 +52,6 @@ WEBHOOK_PAYLOAD = 'json'
WEBHOOK_REQUEST_METHOD = 'GET'
WEBHOOK_SECRET = ''
# NTFY
REPORT_NTFY = False
NTFY_HOST = 'https://ntfy.sh'
NTFY_TOPIC = ''
NTFY_USER = ''
NTFY_PASSWORD = ''
# PUSHSAFER
REPORT_PUSHSAFER = False

7
pialert/initialise.py
View File

@@ -128,13 +128,6 @@ def importConfigs (db):
conf.WEBHOOK_SIZE = ccd('WEBHOOK_SIZE', 1024 , c_d, 'Payload size', 'integer', '', 'Webhooks')
conf.WEBHOOK_SECRET = ccd('WEBHOOK_SECRET', '' , c_d, 'Secret', 'text', '', 'Webhooks')
# NTFY
conf.REPORT_NTFY = ccd('REPORT_NTFY', False , c_d, 'Enable NTFY', 'boolean', '', 'NTFY', ['test'])
conf.NTFY_HOST = ccd('NTFY_HOST', 'https://ntfy.sh' , c_d, 'NTFY host URL', 'text', '', 'NTFY')
conf.NTFY_TOPIC = ccd('NTFY_TOPIC', '' , c_d, 'NTFY topic', 'text', '', 'NTFY')
conf.NTFY_USER = ccd('NTFY_USER', '' , c_d, 'NTFY user', 'text', '', 'NTFY')
conf.NTFY_PASSWORD = ccd('NTFY_PASSWORD', '' , c_d, 'NTFY password', 'password', '', 'NTFY')
# PUSHSAFER
conf.REPORT_PUSHSAFER = ccd('REPORT_PUSHSAFER', False , c_d, 'Enable PUSHSAFER', 'boolean', '', 'PUSHSAFER', ['test'])
conf.PUSHSAFER_TOKEN = ccd('PUSHSAFER_TOKEN', 'ApiKey' , c_d, 'PUSHSAFER token', 'text', '', 'PUSHSAFER')

43
pialert/publishers/ntfy.py
View File

@@ -1,43 +0,0 @@
import conf
import requests
from base64 import b64encode
from logger import mylog
from helper import noti_obj
#-------------------------------------------------------------------------------
def check_config():
if conf.NTFY_HOST == '' or conf.NTFY_TOPIC == '':
mylog('none', ['[Check Config] Error: NTFY service not set up correctly. Check your pialert.conf NTFY_* variables.'])
return False
else:
return True
#-------------------------------------------------------------------------------
def send (msg: noti_obj):
headers = {
"Title": "Pi.Alert Notification",
"Actions": "view, Open Dashboard, "+ conf.REPORT_DASHBOARD_URL,
"Priority": "urgent",
"Tags": "warning"
}
# if username and password are set generate hash and update header
if conf.NTFY_USER != "" and conf.NTFY_PASSWORD != "":
# Generate hash for basic auth
# usernamepassword = "{}:{}".format(conf.NTFY_USER,conf.NTFY_PASSWORD)
basichash = b64encode(bytes(conf.NTFY_USER + ':' + conf.NTFY_PASSWORD, "utf-8")).decode("ascii")
# add authorization header with hash
headers["Authorization"] = "Basic {}".format(basichash)
try:
requests.post("{}/{}".format( conf.NTFY_HOST, conf.NTFY_TOPIC),
data=msg.text,
headers=headers)
except requests.exceptions.RequestException as e:
mylog('none', ['[NTFY] Error: ', e])
return -1
return 0

2
pialert/reporting.py
View File

@@ -24,8 +24,6 @@ from const import pialertPath, logPath, apiPath
from helper import noti_obj, generate_mac_links, removeDuplicateNewLines, timeNowTZ, hide_email, updateState, get_file_content, write_file
from logger import logResult, mylog, print_log
from publishers.ntfy import (check_config as ntfy_check_config,
send as send_ntfy )
from publishers.webhook import (check_config as webhook_check_config,
send as send_webhook)
from publishers.pushsafer import (check_config as pushsafer_check_config,