bundlewrap/bundles/icinga2/files/scripts/icinga_notification_wrapper

#!/usr/bin/env python3

import email.mime.text
import smtplib
from argparse import ArgumentParser
from json import dumps, load
from subprocess import run
from sys import argv

from requests import post

with open('/etc/icinga2/notification_config.json') as f:
    CONFIG = load(f)

parser = ArgumentParser(
    prog='icinga_notification_wrapper',
    description='Icinga2 Notification Wrapper',
)
parser.add_argument(
    '--host_name',
    type=str,
    required=True,
)
parser.add_argument(
    '--service_name',
    type=str,
)
parser.add_argument(
    '--output',
    type=str,
    required=True,
)
parser.add_argument(
    '--state',
    type=str,
    required=True,
)
parser.add_argument(
    '--sms',
    type=str,
)
parser.add_argument(
    '--mail',
    type=str,
)
parser.add_argument(
    '--notification-type',
    type=str,
    required=True,
)

def log_to_syslog(message):
    try:
        msg = '{}/{}: {}'.format(args.host_name, args.service_name, message)
        run(['logger', '-t', 'icinga_notification_wrapper', msg])
    except:
        # We don't expect this to fail. However, *if* it fails, we don't
        # want it to get in the way of other notifications we may have
        # to send.
        pass


def notify_per_sms():
    if args.notification_type.lower() == 'recovery':
        output_text = ''
    else:
        output_text = '\n\n{}'.format(args.output)

    message_text = 'ICINGA: {host}{service} is {state}{output}'.format(
        host=args.host_name,
        service=('/'+args.service_name if args.service_name else ''),
        state=args.state.upper(),
        output=output_text,
    )

    message = {
        'message': message_text,
        'smsId': 's0', # XXX what does this mean? Documentation is unclear
        'recipient': args.sms
    }

    headers = {
        'Content-type': 'application/json',
        'Accept': 'application/json'
    }

    try:
        r = post(
            'https://api.sipgate.com/v2/sessions/sms',
            json=message,
            headers=headers,
            auth=(CONFIG['sipgate']['user'], CONFIG['sipgate']['password']),
        )

        if r.status_code == 204:
            log_to_syslog('Sent a SMS to "{}"'.format(args.sms))
        else:
            r.raise_for_status()
    except Exception as e:
        log_to_syslog('Sending a SMS to "{}" failed: {}'.format(args.sms, repr(e)))


def notify_per_ntfy():
    message_text = 'ICINGA: {host}{service} is {state}\n\n{output}'.format(
        host=args.host_name,
        service=('/'+args.service_name if args.service_name else ''),
        state=args.state.upper(),
        output=args.output,
    )

    if args.service_name:
        subject = '[ICINGA] {}/{}'.format(args.host_name, args.service_name)
    else:
        subject = '[ICINGA] {}'.format(args.host_name)

    if args.notification_type.lower() == 'recovery':
        priority = 'default'
    else:
        priority = 'urgent'

    headers = {
        'Title': subject,
        'Priority': priority,
    }

    try:
        r = post(
            CONFIG['ntfy']['url'],
            data=message_text,
            headers=headers,
            auth=(CONFIG['ntfy']['user'], CONFIG['ntfy']['password']),
            timeout=10,
        )

        r.raise_for_status()
    except Exception as e:
        log_to_syslog('Sending a Notification failed: {}'.format(repr(e)))
        return False
    return True


def notify_per_mail():
    if args.notification_type.lower() == 'recovery':
        # Do not send recovery emails.
        return

    # editorconfig-checker-disable
    text = """
    _      _
   (_)____(_)___  ____ _____ _
  / / ___/ / __ \/ __ `/ __ `/
 / / /__/ / / / / /_/ / /_/ /
/_/\___/_/_/ /_/\__, /\__,_/
               /____/

Host: {host}"""

    if args.service_name:
        text += """
Service: {service}"""

    text += """
State: {state}

{output}"""
    # editorconfig-checker-enable

    mail = email.mime.text.MIMEText(text.format(
            host=args.host_name,
            service=args.service_name,
            state=args.state.upper(),
            output=args.output
        ),
        'plain',
        'utf-8',
    )

    if args.service_name:
        mail['Subject'] = '[ICINGA] {}/{} is {}'.format(args.host_name, args.service_name, args.state)
    else:
        mail['Subject'] = '[ICINGA] {} is {}'.format(args.host_name, args.state)

    mail['To'] = args.mail
    mail['From'] = 'noreply+icinga@kunbox.net'

    try:
        s = smtplib.SMTP('localhost')
        s.sendmail(mail['From'], [args.mail], mail.as_string())
        s.quit()

        log_to_syslog('Sent mail to "{}"'.format(args.mail))
    except Exception as e:
        log_to_syslog('Sending mail to "{}" failed: {}'.format(args.mail, repr(e)))


if __name__ == '__main__':
    args = parser.parse_args()

    log_to_syslog(' '.join(argv))

    if args.mail:
        notify_per_mail()

    if args.sms:
        ntfy_worked = False
        if CONFIG['ntfy']['user']:
            ntfy_worked = notify_per_ntfy()
        if not args.service_name or not ntfy_worked:
            notify_per_sms()