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 data
from . import ws as ws_mod
from . import main as main_mod

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

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

logf = None

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

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

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

def eventlog(host, lvl, m, service=None):
    ts = time.time()
    s = f"{time.strftime('%Y-%m-%d %H:%M:%S', time.localtime(ts))} {lvl} "
    if host:
        s += f"{host} "
    s += m
    data.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 _dispatch_to_channel(channel_name: str, channel_config: dict, msg: str, debug: int = 0) -> bool:
    """Dispatch a message to a specific notification channel.
    
    Args:
        channel_name: Name of the channel (for logging)
        channel_config: Channel configuration dictionary with 'type' and type-specific fields
        msg: Message to send
        debug: Debug level
        
    Returns:
        True if notification sent successfully, False otherwise
    """
    channel_type = channel_config.get("type")
    
    if channel_type == "pushover":
        return pushover(
            channel_config.get("token", ""),
            channel_config.get("user", ""),
            msg,
            debug=debug
        )
    
    elif channel_type == "email":
        # Build email from channel config
        recipients = channel_config.get("recipients", [])
        sender = channel_config.get("sender", "")
        smtp_server = channel_config.get("smtp_server", "")
        smtp_port = channel_config.get("smtp_port", 587)
        smtp_user = channel_config.get("smtp_user")
        smtp_password = channel_config.get("smtp_password")
        
        if not recipients or not sender or not smtp_server:
            logger.warning(
                "Email channel '%s' missing required fields: recipients=%s, sender=%s, smtp_server=%s",
                channel_name, recipients, sender, smtp_server
            )
            return False
        
        # Temporarily update _config for email() function
        old_config = dict(_config)
        _config["toemail"] = recipients
        _config["fromemail"] = sender
        _config["smtpserver"] = smtp_server
        _config["smtpport"] = smtp_port
        if smtp_user:
            _config["smtpuser"] = smtp_user
        if smtp_password:
            _config["smtppassword"] = smtp_password
        
        result = email("Heartbeat notification", msg, debug=debug)
        
        # Restore config
        _config.clear()
        _config.update(old_config)
        
        return result
    
    elif channel_type == "signal":
        return pushsignal(
            channel_config.get("cli_path", "/usr/local/bin/signal-cli"),
            channel_config.get("user", ""),
            channel_config.get("recipient", ""),
            msg,
            debug=debug
        )
    
    elif channel_type == "mattermost":
        return pushmattermost(
            channel_config.get("host", ""),
            channel_config.get("token", ""),
            channel_config.get("channel", ""),
            msg,
            username=channel_config.get("username", "hbd"),
            icon=channel_config.get("icon"),
            debug=debug
        )
    
    else:
        logger.warning("Unknown channel type '%s' for channel '%s'", channel_type, channel_name)
        return False


def pushmsg_for_host(hostname: str, msg: str, debug: int = 0) -> dict:
    """Send notification for a specific host using its configured channels.
    
    This function looks up the host's notification channels from the config
    and sends the message to those channels.
    
    Args:
        hostname: Name of the host to send notification for
        msg: Message to send
        debug: Debug level
        
    Returns:
        Dictionary of results per channel: {"channel_name": True/False}
    """
    from . import config as config_mod
    
    # Get notification channels for this host
    channels = config_mod.get_notification_channels_config(_config, hostname)
    
    if not channels:
        logger.warning("No notification channels configured for host '%s'", hostname)
        return {}
    
    # Dispatch to each channel
    results = {}
    for channel_name, channel_config in channels:
        try:
            success = _dispatch_to_channel(channel_name, channel_config, msg, debug=debug)
            results[channel_name] = success
            if success:
                logger.info("Notification sent to channel '%s': %s", channel_name, msg)
            else:
                logger.warning("Failed to send notification to channel '%s'", channel_name)
        except Exception as e:
            logger.error("Error sending to channel '%s': %s", channel_name, e)
            results[channel_name] = False
    
    return results