heartbeat/hbd/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 traceback

DEFAULT_PUSHPROVIDERS = ["all", "pushover", "mattermost", "signal"]

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


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)