c00fff224c
Both zero-dependency (stdlib http.server + urllib), consistent with the project's no-dependency-tree stance for a security daemon. Web dashboard (enodia_sentinel/web.py + static/dashboard.html): - read-only JSON API over the log dir: /api/status, /api/alerts, /api/alerts/<id>, /api/events - bearer-token auth (constant-time; header or ?token=), required on non- loopback binds, auto-generated + persisted (0600) when unset - binds the host's Tailscale IP by default (auto-detected), reachable from the tailnet but not the LAN/internet - self-contained dark SPA: severity cards, live alert list, full snapshot viewer; 10s auto-refresh - path-traversal-safe alert lookup; `enodia-sentinel web` subcommand; daemon now writes a pidfile so the dashboard can show live status - hardened enodia-sentinel-web.service (read-only, no caps) Phone push (enodia_sentinel/notify/): - pluggable backends — ntfy, Pushover, generic webhook — each separating a pure build() (unit-tested, no network) from send() - a backend turns on when its config keys are set; pushes gated by notify_min_severity; severity → per-service priority/tags - fired from snapshot.capture on worker threads, errors swallowed - desktop notify-send retained Tests: +16 (9 web incl. a real-server 401/200 auth test, 7 notify request-build cases). 55/55 pass. Live end-to-end verified: daemon → alert → API → page. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
100 lines
3.2 KiB
Python
100 lines
3.2 KiB
Python
# SPDX-License-Identifier: GPL-3.0-or-later
|
|
"""Phone-push backends. Each separates a pure ``build()`` (testable, no network)
|
|
from ``send()`` (does the request), so request construction is unit-tested
|
|
without ever hitting the wire."""
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import urllib.parse
|
|
import urllib.request
|
|
|
|
from ..alert import Severity
|
|
from ..config import Config
|
|
|
|
_TIMEOUT = 6
|
|
|
|
# Severity -> per-service priority/tags.
|
|
_NTFY_PRIORITY = {Severity.MEDIUM: "3", Severity.HIGH: "4", Severity.CRITICAL: "5"}
|
|
_NTFY_TAGS = {Severity.MEDIUM: "mag", Severity.HIGH: "warning",
|
|
Severity.CRITICAL: "rotating_light"}
|
|
_PUSHOVER_PRIORITY = {Severity.MEDIUM: "-1", Severity.HIGH: "0", Severity.CRITICAL: "1"}
|
|
|
|
|
|
def _send(req: urllib.request.Request) -> bool:
|
|
with urllib.request.urlopen(req, timeout=_TIMEOUT) as resp:
|
|
return 200 <= resp.status < 300
|
|
|
|
|
|
class Ntfy:
|
|
name = "ntfy"
|
|
|
|
@staticmethod
|
|
def enabled(cfg: Config) -> bool:
|
|
return bool(cfg.notify_ntfy_url and cfg.notify_ntfy_topic)
|
|
|
|
@staticmethod
|
|
def build(cfg: Config, notif) -> urllib.request.Request:
|
|
url = f"{cfg.notify_ntfy_url.rstrip('/')}/{cfg.notify_ntfy_topic}"
|
|
headers = {
|
|
"Title": notif.title(),
|
|
"Priority": _NTFY_PRIORITY.get(notif.severity, "4"),
|
|
"Tags": _NTFY_TAGS.get(notif.severity, "warning"),
|
|
}
|
|
if cfg.notify_ntfy_token:
|
|
headers["Authorization"] = f"Bearer {cfg.notify_ntfy_token}"
|
|
return urllib.request.Request(
|
|
url, data=notif.message().encode("utf-8"),
|
|
headers=headers, method="POST")
|
|
|
|
@classmethod
|
|
def send(cls, cfg: Config, notif) -> bool:
|
|
return _send(cls.build(cfg, notif))
|
|
|
|
|
|
class Pushover:
|
|
name = "pushover"
|
|
|
|
@staticmethod
|
|
def enabled(cfg: Config) -> bool:
|
|
return bool(cfg.notify_pushover_token and cfg.notify_pushover_user)
|
|
|
|
@staticmethod
|
|
def build(cfg: Config, notif) -> urllib.request.Request:
|
|
form = urllib.parse.urlencode({
|
|
"token": cfg.notify_pushover_token,
|
|
"user": cfg.notify_pushover_user,
|
|
"title": notif.title(),
|
|
"message": notif.message(),
|
|
"priority": _PUSHOVER_PRIORITY.get(notif.severity, "0"),
|
|
}).encode("utf-8")
|
|
return urllib.request.Request(
|
|
"https://api.pushover.net/1/messages.json", data=form,
|
|
headers={"Content-Type": "application/x-www-form-urlencoded"},
|
|
method="POST")
|
|
|
|
@classmethod
|
|
def send(cls, cfg: Config, notif) -> bool:
|
|
return _send(cls.build(cfg, notif))
|
|
|
|
|
|
class Webhook:
|
|
name = "webhook"
|
|
|
|
@staticmethod
|
|
def enabled(cfg: Config) -> bool:
|
|
return bool(cfg.notify_webhook_url)
|
|
|
|
@staticmethod
|
|
def build(cfg: Config, notif) -> urllib.request.Request:
|
|
body = json.dumps({
|
|
"title": notif.title(),
|
|
"message": notif.message(),
|
|
**notif.to_dict(),
|
|
}).encode("utf-8")
|
|
return urllib.request.Request(
|
|
cfg.notify_webhook_url, data=body,
|
|
headers={"Content-Type": "application/json"}, method="POST")
|
|
|
|
@classmethod
|
|
def send(cls, cfg: Config, notif) -> bool:
|
|
return _send(cls.build(cfg, notif))
|