180 lines
4.3 KiB
Python
180 lines
4.3 KiB
Python
#!/usr/bin/env python3
|
|
|
|
import email.mime.text
|
|
import smtplib
|
|
from argparse import ArgumentParser
|
|
from json import dumps
|
|
from subprocess import run
|
|
from sys import argv
|
|
|
|
from requests import post
|
|
|
|
SIPGATE_USER='${node.metadata['icinga2']['sipgate_user']}'
|
|
SIPGATE_PASS='${node.metadata['icinga2']['sipgate_pass']}'
|
|
|
|
STATUS_TO_EMOJI = {
|
|
'critical': '🔥',
|
|
'down': '🚨🚨🚨',
|
|
'ok': '🆗',
|
|
'up': '👌',
|
|
'warning': '⚡',
|
|
}
|
|
|
|
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)
|
|
if args.state.lower() in STATUS_TO_EMOJI:
|
|
message_text = '{emoji} {host}{service} {emoji}{output}'.format(
|
|
emoji=STATUS_TO_EMOJI[args.state.lower()],
|
|
host=args.host_name,
|
|
service=('/'+args.service_name if args.service_name else ''),
|
|
state=args.state.upper(),
|
|
output=output_text,
|
|
)
|
|
else:
|
|
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=(SIPGATE_USER, SIPGATE_PASS),
|
|
)
|
|
|
|
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_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:
|
|
notify_per_sms()
|