[packages/nagios-alert-pushover] --validate-user: ask Pushover what the recipients can actually receive
arekm
arekm at pld-linux.org
Mon Aug 17 11:11:47 CEST 2026
commit cd2f51eda21b267d3cddd30d31d6efec964f003e
Author: Arkadiusz Miśkiewicz <arekm at maven.pl>
Date: Mon Aug 17 11:11:01 2026 +0200
--validate-user: ask Pushover what the recipients can actually receive
An account with no active device and an unregistered device name are both silent at sending time. Release 2.
nagios-alert-pushover.spec | 2 +-
nagios-pushover.alert | 98 ++++++++++++++++++++++++++++++++++++++++++++
pushover.toml | 8 ++++
test_nagios_pushover.py | 100 ++++++++++++++++++++++++++++++++++++++++++++-
4 files changed, 206 insertions(+), 2 deletions(-)
---
diff --git a/nagios-alert-pushover.spec b/nagios-alert-pushover.spec
index 13790af..8281c14 100644
--- a/nagios-alert-pushover.spec
+++ b/nagios-alert-pushover.spec
@@ -6,7 +6,7 @@ Summary: Program to send (Nagios) alerts via Pushover
Summary(pl.UTF-8): Program do wysyłania alarmów (Nagiosa) przez Pushover
Name: nagios-alert-pushover
Version: 1.0
-Release: 1
+Release: 2
License: AGPL v3+
Group: Networking
Source0: nagios-pushover.alert
diff --git a/nagios-pushover.alert b/nagios-pushover.alert
index 66fc351..04a1c44 100644
--- a/nagios-pushover.alert
+++ b/nagios-pushover.alert
@@ -19,8 +19,14 @@ shell fallback in the nagios command can take over, for example:
nagios-notify notify-service-by-pushover | nagios-notify-pushover \
|| nagios-notify notify-service-by-sms | sms-gateway
+
+Run by hand, --validate-user asks Pushover what the configured recipients can receive.
+Both ways that can be wrong are silent at sending time: an account without an active
+device swallows the alert, and a device name nobody registered widens delivery to
+every device the person owns instead of failing.
"""
+import argparse
import base64
import gzip
import hashlib
@@ -40,6 +46,7 @@ from email.parser import Parser
CONFIG = "/etc/nagios/pushover.toml"
API = "https://api.pushover.net/1/messages.json"
RECEIPTS = "https://api.pushover.net/1/receipts"
+VALIDATE = "https://api.pushover.net/1/users/validate.json"
LOGFILE = "/var/log/nagios/nagios-pushover.log"
LOG_FORMAT = "%(asctime)s - %(levelname)s - %(message)s"
@@ -389,7 +396,98 @@ def read_config():
return config, recipients
+def check_user(config, recipient):
+ """Ask Pushover what this recipient's account can receive.
+
+ The whole device list is fetched rather than the configured name validated,
+ because a name Pushover does not know is only an error here: when a message
+ carries one, it is delivered to every device the person owns instead.
+ """
+ data = urllib.parse.urlencode({"token": config["token"],
+ "user": recipient["user_key"]}).encode()
+ try:
+ try:
+ with OPENER.open(urllib.request.Request(VALIDATE, data=data),
+ timeout=ATTEMPT_TIMEOUT) as response:
+ result = json.loads(response.read(MAX_RESPONSE))
+ except urllib.error.HTTPError as error:
+ # a key Pushover does not accept comes back as a 400 whose body says why
+ result = json.loads(error.read(MAX_RESPONSE))
+ except Exception as error:
+ return "FAILED", f"{type(error).__name__}: {error}"
+
+ if result.get("status") != 1:
+ return "FAILED", "; ".join(result.get("errors") or ["unexpected response"])
+
+ devices = result.get("devices") or []
+ licenses = result.get("licenses") or []
+ detail = "devices: " + (", ".join(devices) or "none listed")
+ if licenses:
+ detail += " licenses: " + ", ".join(licenses)
+
+ wanted = recipient.get("device", "")
+ if not devices:
+ # a delivery group key validates without naming devices, so there is nothing
+ # to hold a configured device name against
+ if wanted:
+ return "WARN", f'{detail}, so device "{wanted}" cannot be checked'
+ else:
+ missing = [name for name in wanted.split(",") if name and name not in devices]
+ if missing:
+ return "FAILED", (f'device "{",".join(missing)}" is not registered; '
+ f"the account has: {', '.join(devices)}")
+ if "encryption_key" in recipient:
+ detail += " encrypted"
+ # only the mobile apps from 5.0 decrypt; anywhere else the alert arrives as
+ # ciphertext, which Pushover reports as a delivery like any other
+ if licenses and not {"Android", "iOS"} & set(licenses):
+ return "WARN", detail + ", but no app on this account can decrypt it"
+ return "ok", detail
+
+
+def validate(phones):
+ """Check that the configured recipients can receive anything at all."""
+ try:
+ config, recipients = read_config()
+ except (OSError, tomllib.TOMLDecodeError) as error:
+ print(f"cannot read {CONFIG}: {error}", file=sys.stderr)
+ return 1
+ except ValueError as error:
+ print(f"{CONFIG}: {error}", file=sys.stderr)
+ return 1
+
+ if not recipients:
+ print(f"no recipients configured in {CONFIG}", file=sys.stderr)
+ return 1
+
+ failed = False
+ for phone in phones or recipients:
+ if phone in recipients:
+ state, detail = check_user(config, recipients[phone])
+ else:
+ state, detail = "FAILED", f"not configured in {CONFIG}"
+ print(f"{phone:<16}{state:<8}{detail}")
+ failed = failed or state == "FAILED"
+ return 1 if failed else 0
+
+
def main():
+ parser = argparse.ArgumentParser(
+ prog="nagios-notify-pushover",
+ description="Deliver the nagios notification waiting on stdin through Pushover.")
+ parser.add_argument(
+ "--validate-user", nargs="*", metavar="PHONE",
+ help="send nothing and ask Pushover what the configured recipients can "
+ "receive, all of them or only the phone numbers named")
+ arguments = parser.parse_args()
+
+ # nargs="*" tells the three cases apart: absent, given bare, given with numbers
+ if arguments.validate_user is not None:
+ # the log is created by whichever run writes to it first, and this mode is run
+ # by hand, usually by root: a root-owned logfile leaves nagios unable to write
+ logging.basicConfig(level=logging.INFO, format=LOG_FORMAT)
+ return validate(arguments.validate_user)
+
try:
logging.basicConfig(filename=LOGFILE, level=logging.INFO, format=LOG_FORMAT)
except OSError as error:
diff --git a/pushover.toml b/pushover.toml
index f65c9d7..e134687 100644
--- a/pushover.toml
+++ b/pushover.toml
@@ -33,6 +33,14 @@ token = "APPLICATIONTOKEN00000000000000"
# It matters most together with encryption: the key is stored per device, so either
# give every device the same key, or point this at the one that has it.
+# After editing this file, check who can actually receive:
+#
+# nagios-notify-pushover --validate-user
+#
+# It sends nothing, only asks Pushover which devices each account has - a key that
+# stopped working, or a device name with a typo in it, shows up there instead of at
+# the next alarm.
+
# The examples below are commented out on purpose: as shipped, this script delivers
# nothing and every notification falls straight through to the command's fallback.
diff --git a/test_nagios_pushover.py b/test_nagios_pushover.py
index 285d653..55716c0 100644
--- a/test_nagios_pushover.py
+++ b/test_nagios_pushover.py
@@ -106,6 +106,14 @@ ACCEPTED = json.dumps({"status": 1, "request": "req-1"}).encode()
RECEIPTED = json.dumps({"status": 1, "request": "req-2", "receipt": "rcpt-9"}).encode()
REFUSED = json.dumps({"status": 0, "request": "req-3",
"errors": ["user identifier is invalid"]}).encode()
+VALID = json.dumps({"status": 1, "request": "req-4", "devices": ["iphone", "work-ipad"],
+ "licenses": ["Android", "iOS", "Desktop"]}).encode()
+DESKTOP = json.dumps({"status": 1, "request": "req-5", "devices": ["workstation"],
+ "licenses": ["Desktop"]}).encode()
+GROUP = json.dumps({"status": 1, "request": "req-6", "devices": []}).encode()
+INVALID = json.dumps({"status": 0, "request": "req-7", "devices": [],
+ "errors": ["user identifier is not a valid user, group, or "
+ "subscribed user key"]}).encode()
def feed(text):
@@ -116,6 +124,7 @@ class Api(http.server.BaseHTTPRequestHandler):
"""Replies with the next entry of `script`, or accepts once the script runs out."""
script = []
cancel_script = []
+ validate_script = []
requests = []
quota = "5000"
@@ -124,6 +133,8 @@ class Api(http.server.BaseHTTPRequestHandler):
Api.requests.append((self.path, body))
if "cancel_by_tag" in self.path:
step = Api.cancel_script.pop(0) if Api.cancel_script else (200, ACCEPTED, {})
+ elif "users/validate" in self.path:
+ step = Api.validate_script.pop(0) if Api.validate_script else (200, VALID, {})
else:
step = Api.script.pop(0) if Api.script else (200, ACCEPTED, {})
if step == "garbage":
@@ -179,6 +190,7 @@ server = Server(("127.0.0.1", 0), Api)
threading.Thread(target=server.serve_forever, daemon=True).start()
g.API = f"http://127.0.0.1:{server.server_port}/1/messages.json"
g.RECEIPTS = f"http://127.0.0.1:{server.server_port}/1/receipts"
+g.VALIDATE = f"http://127.0.0.1:{server.server_port}/1/users/validate.json"
EVENTS = (f'[event.CRITICAL]\nicon = "{RED}"\npriority = 1\nsound = "gamelan"\n\n'
f'[event.DOWN]\nicon = "{RED}"\npriority = 2\nsound = "siren"\n'
@@ -200,15 +212,46 @@ def run(script=(), notification=NOTIF, config=None, cancel_script=()):
Api.cancel_script = list(cancel_script)
Api.requests = []
feed(notification)
+ # the notification mode takes no arguments, and under pytest sys.argv holds the
+ # runner's own, which the script's parser would rightly refuse
+ argv, sys.argv = sys.argv, ["nagios-notify-pushover"]
try:
return g.main()
finally:
+ sys.argv = argv
signal.alarm(0)
+def run_argv(args, config=None, validate_script=()):
+ """A command line mode, with both output streams captured."""
+ if config is not None:
+ with open(g.CONFIG, "w") as handle:
+ handle.write(config)
+ Api.validate_script = list(validate_script)
+ Api.requests = []
+ argv, streams = sys.argv, (sys.stdout, sys.stderr)
+ captured = io.StringIO()
+ sys.argv = ["nagios-notify-pushover", *args]
+ sys.stdout = sys.stderr = captured
+ try:
+ try:
+ code = g.main()
+ except SystemExit as error:
+ # argparse answers --help and a bad option by leaving on its own
+ code = error.code
+ return code, captured.getvalue()
+ finally:
+ sys.argv = argv
+ sys.stdout, sys.stderr = streams
+
+
+def run_validate(args=(), config=None, validate_script=()):
+ return run_argv(["--validate-user", *args], config, validate_script)
+
+
def sent():
"""The last message request as a plain dict."""
- body = [b for path, b in Api.requests if "cancel_by_tag" not in path][-1]
+ body = [b for path, b in Api.requests if path.endswith("/messages.json")][-1]
return {k: v[0] for k, v in urllib.parse.parse_qs(body.decode()).items()}
@@ -728,6 +771,61 @@ check("cleartext" in line and "device=iphone" in line,
f"the log says outright the alert went in the clear and to one device -> {line[-60:]!r}")
run(config=CONFIG)
+print("\n== --validate-user says who can receive, and sends nothing")
+code, out = run_validate(config=CONFIG)
+check(code == 0 and PHONE in out and " ok " in out,
+ "a recipient with an active device validates")
+check("iphone, work-ipad" in out and "encrypted" in out,
+ f"the line names the devices and says the alerts go encrypted -> {out.strip()!r}")
+check(len(Api.requests) == 1 and Api.requests[0][0].endswith("/users/validate.json"),
+ "one validation request, and not a single message sent")
+
+check(run_validate([PHONE])[0] == 0 and len(Api.requests) == 1,
+ "a phone number on the command line narrows the check to that recipient")
+
+code, out = run_validate(["+48999999999"])
+check(code == 1 and "not configured" in out and not Api.requests,
+ "a number nobody configured fails without asking Pushover about it")
+
+code, out = run_validate(config=CONFIG + 'device = "iphon"\n')
+check(code == 1 and "iphon" in out and "iphone, work-ipad" in out,
+ f"a device typo is named next to what the account really has -> {out.strip()!r}")
+check(run_validate(config=CONFIG + 'device = "work-ipad"\n')[0] == 0,
+ "a device the account has registered validates")
+
+code, out = run_validate(config=CONFIG, validate_script=[(400, INVALID, {})])
+check(code == 1 and "not a valid user" in out,
+ "a key Pushover rejects fails with the reason Pushover gave")
+
+code, out = run_validate(config=CONFIG, validate_script=[(200, DESKTOP, {})])
+check(code == 0 and "WARN" in out and "decrypt" in out,
+ f"encryption with no mobile app to read it warns, and is not a failure -> {out.strip()!r}")
+
+code, out = run_validate(config=NOKEY + 'device = "iphone"\n',
+ validate_script=[(200, GROUP, {})])
+check(code == 0 and "WARN" in out and "cannot be checked" in out,
+ "a key that names no devices, as a delivery group does, leaves the device unverified")
+
+code, out = run_validate(config=f'token = "{TOKEN}"\n')
+check(code == 1 and "no recipients" in out and not Api.requests,
+ "a configuration nobody is listed in is not a silent pass")
+
+code, out = run_validate(config=CONFIG, validate_script=[(500, b"nope", {})])
+check(code == 1 and "FAILED" in out,
+ "an unreadable answer is a failure with a name, not a traceback")
+
+check(run_argv(["--help"])[0] == 0, "--help is answered instead of read as a notification")
+code, out = run_argv(["--nonsense"])
+check(code == 2 and "usage" in out, "an unknown option leaves with the usage")
+
+logfile, g.LOGFILE = g.LOGFILE, f"{SCRATCH}/must-not-appear.log"
+unhook_logging()
+code, out = run_validate(config=CONFIG)
+check(code == 0 and not os.path.exists(g.LOGFILE),
+ "the validator writes no logfile: run by root it would leave one nagios cannot write")
+g.LOGFILE = logfile
+unhook_logging()
+
print("\n== logging must not cost the alert")
Api.quota = "12"
check(run(config=CONFIG) == 0, "a low monthly quota still sends (just a warning)")
================================================================
---- gitweb:
http://git.pld-linux.org/gitweb.cgi/packages/nagios-alert-pushover.git/commitdiff/cd2f51eda21b267d3cddd30d31d6efec964f003e
More information about the pld-cvs-commit
mailing list