heartbeat/hbd/server/notify.py

"""Notification helpers: email, pushover, mattermost, signal and dispatcher."""

import logging
from typing import Optional
import http.client
import urllib.parse
import subprocess
import smtplib
import time
import sys
from . import ws as ws_mod

DEFAULT_PUSHPROVIDERS = ["all", "pushover", "mattermost", "signal"]
msg_to_websockets = ws_mod.broadcast

# module-level configuration set via setup()
_config = {}
logger = logging.getLogger(__name__)

msgs = []
logf = None

def initlog(logfile):
    global logf
    try:
        logf = open(logfile, "a+")
        return logf
    except Exception as e:
        import sys

        print("cannot open logfile %s, using STDERR: %s" % (logfile, e))
        return sys.stderr

def closelog():
    global logf
    if logf and logf != sys.stderr:
        try:
            logf.close()
        except Exception:
            pass

def log(host, m, service=None):
    ts = time.time()
    s = f"{time.strftime('%Y-%m-%d %H:%M:%S', time.localtime(ts))} {host or ''} {m}"
    msgs.append(s)
    logger.info(s)
    if logf:
        try:
            logf.write(s + "\n")
            logf.flush()
        except Exception as e:
            logger.warning("failed to write to logfile: %s", e)
    msg_to_websockets("message", s)

def setup(cfg: dict):
    """Initialize notifier defaults from a configuration dict."""
    global _config
    _config = dict(cfg)


def send_email(toaddrs, smtpserver, sender, subject, body, debug=0):
    """Send a plain email via SMTP. Returns True on success."""
    try:
        smtpport = _config.get("smtpport", 587)
        server = smtplib.SMTP(smtpserver, smtpport)
        if debug > 0:
            server.set_debuglevel(1)
        if smtpport == 587:
            server.starttls()
            server.ehlo()
            smtpuser = _config.get("smtpuser", None)
            smtppassword = _config.get("smtppassword", None)
            if smtpuser and smtppassword:
                server.login(smtpuser, smtppassword)
        server.sendmail(sender, toaddrs, body)
    except Exception as e:
        logger.warning("email send failed: %s", e)
        try:
            server.quit()
        except Exception:
            pass
        return False
    try:
        server.quit()
    except Exception:
        pass
    return True


def email(subject: str, msg: str, debug: int = 0) -> bool:
    """Convenience wrapper exposed to the rest of the application.

    Uses module-level configuration to supply recipient list, smtp server
    and sender address.
    """
    toaddrs = _config.get("toemail")
    fromemail = _config.get("fromemail")
    smtpserver = _config.get("smtpserver")
    if not toaddrs or not fromemail or not smtpserver:
        logger.warning(
            "email config incomplete: toemail=%s, fromemail=%s, smtpserver=%s",
            toaddrs,
            fromemail,
            smtpserver,
        )
        return False
    date = time.strftime("%a, %d %b %Y %H:%M:%S %z", time.localtime())
    body = "To: %s\nFrom: %s\nSubject: %s\nDate: %s\n\n%s" % (
        toaddrs[0] if toaddrs else "",
        fromemail,
        subject,
        date,
        msg,
    )
    return send_email(toaddrs, smtpserver, fromemail, subject, body, debug=debug)


def pushover(token: str, user: str, msg: str, debug: int = 0) -> bool:
    """Send message via Pushover API."""
    conn = http.client.HTTPSConnection("api.pushover.net:443")
    try:
        conn.request(
            "POST",
            "/1/messages.json",
            urllib.parse.urlencode({"token": token, "user": user, "message": msg}),
            {"Content-type": "application/x-www-form-urlencoded"},
        )
        r = conn.getresponse()
        logger.debug("pushover response: %s %s", r.status, r.reason)
        return r.status == 200
    except Exception as e:
        logger.error("pushover error: %s", e)
        return False


def pushmattermost(
    host: str,
    token: str,
    channel: str,
    msg: str,
    username: str = "hbd",
    icon: Optional[str] = None,
    debug: int = 0,
) -> bool:
    """Send a message to Mattermost via simple webhook driver if available.

    This helper tries to import mattermostdriver.Driver and uses webhooks if present.
    If the import fails it returns False.
    """
    try:
        from mattermostdriver import Driver
    except Exception:
        return False
    ses = {"url": host, "scheme": "http", "basepath": "/api/v4", "port": 8065}
    mm = Driver(ses)
    payload = {"text": msg, "channel": channel, "username": username}
    if icon:
        payload["icon_url"] = icon
    try:
        rc = mm.webhooks.call_webhook(token, payload)
        logger.debug("mattermost rc: %s", rc)
        return bool(rc is None or rc == "")
    except Exception as e:
        logger.error("mattermost error: %s", e)
        return False


def pushsignal(
    signal_cli_bin: str, user: str, recipient: str, msg: str, debug: int = 0
) -> bool:
    """Send a message via signal-cli (requires local installation).

    Uses subprocess to call signal-cli. Returns True if the command succeeded.
    """
    CLI = [signal_cli_bin, "-u", user, "send", "-m", msg, recipient]
    logger.debug("signal cli: %s", CLI)
    try:
        res = subprocess.run(CLI, capture_output=True)
        if res.returncode != 0:
            logger.error("signal failed: %s".res.stderr.decode())
            return False
        logger.debug("signal sent: %s", res.stdout.decode())
        return True
    except Exception as e:
        logger.exception("signal exception: %s", e)
        return False


def pushmsg(cfg: dict, msg: str, debug: int = 0):
    """Dispatch push notifications according to `cfg['pushsrv']`.

    cfg is expected to contain keys for different services when needed, e.g.
    - cfg['pushsrv'] : one of 'all', 'pushover', 'mattermost', 'signal'
    - cfg['pushover_token'], cfg['pushover_user']
    - cfg['matter_host'], cfg['matter_token'], cfg['matter_channel']
    - cfg['signal_cli'], cfg['signal_user'], cfg['signal_recipient']

    Returns a dict of results per provider.
    """
    
    results = {}
    p = cfg.get("pushsrv", "pushover")
    if p in ("all", "pushover"):
        ok = pushover(
            cfg.get("pushover_token", ""),
            cfg.get("pushover_user", ""),
            msg,
            debug=debug,
        )
        results["pushover"] = ok
    if p in ("all", "mattermost"):
        ok = pushmattermost(
            cfg.get("matter_host", ""),
            cfg.get("matter_token", ""),
            cfg.get("matter_channel", ""),
            msg,
            username=cfg.get("matter_username", "hbd"),
            icon=cfg.get("matter_icon"),
            debug=debug,
        )
        results["mattermost"] = ok
    if p in ("all", "signal"):
        ok = pushsignal(
            cfg.get("signal_cli", "/usr/local/bin/signal-cli"),
            cfg.get("signal_user", ""),
            cfg.get("signal_recipient", ""),
            msg,
            debug=debug,
        )
        results["signal"] = ok
    if p in ("all", "email"):
        ok = email("Heartbeat notification", msg, debug=debug)
        results["email"] = ok
    logger.debug("push results: %s", results)
    return results


def pushmsg_from_config(msg: str, debug: int = 0) -> dict:
    """Use the module-level configuration dict to dispatch a push message."""
    return pushmsg(_config, msg, debug=debug)