adminsys/icingabot
8
0
Fork 0
Code Issues Pull Requests Releases Wiki Activity

reformattage à l aide de black et passage de flake8

Browse Source
This commit is contained in:
François Poulain 2020-04-26 11:58:57 +02:00
parent d218d69b97
commit 9fdd726de0
1 changed files with 236 additions and 169 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

251
icingabot.py
View File

@ -10,6 +10,17 @@
# Joel Rosdahl <joel@rosdahl.net>
# https://github.com/jaraco/irc/raw/master/scripts/testbot.py
import configparser
import json
import os.path
import re
import sys
import requests
import irc.bot
import irc.strings
"""A simple IRC Bot statusing icinga2.
It is based on TestBot example bot that uses the SingleServerIRCBot class from
@ -23,39 +34,31 @@ Requirements: python3-irc python3-requests
The known commands are:
"""
commands = {
"ack" : {
"help": "Acknowledge a given service.",
},
"ack": {"help": "Acknowledge a given service."},
"recheck": {
"help": "Recheck a given service or all services.",
"synonyms": [r"refresh"],
},
"list" : {
"help": "List all KO services.",
"synonyms" : [r"lsit", r"lits"],
},
"list": {"help": "List all KO services.", "synonyms": [r"lsit", r"lits"]},
"leave": {
"help": "Disconnect the bot. The bot will try to reconnect after 300 seconds.",
"help": "Disconnect the bot."
" The bot will try to reconnect after 300 seconds.",
},
"mute": {
"help": "Mute the bot (no more status report). The bot will unmute after 1 hour or after receiving any command.",
"synonyms" : [r"fuck", r"chut", r"couché", r"sieste", r"t(a|o)\s*g(ueu|o)le"],
},
"die" : {
"help": "Let the bot cease to exist.",
},
"help" : {
"help": "Print help.",
"help": "Mute the bot (no more status report)."
" The bot will unmute after 1 hour or after receiving any command.",
"synonyms": [
r"fuck",
r"chut",
r"couché",
r"sieste",
r"t(a|o)\s*g(ueu|o)le",
],
},
"die": {"help": "Let the bot cease to exist."},
"help": {"help": "Print help."},
}
import os.path
import irc.bot
import irc.strings
import re
import requests, json
import configparser
# Load configuration.
configurationFilename = "/etc/icingabot/icingabot.conf"
if os.path.isfile(configurationFilename):
@ -68,13 +71,12 @@ if os.path.isfile(configurationFilename):
"ircport": config.getint("irc", "irc.port"),
"ircnick": config.get("irc", "irc.nick"),
"irccmd": config.get("irc", "irc.cmd_prefix"),
# see /etc/icinga2/conf.d/api-users.conf
"icinga2user": config.get("icinga2", "icinga2.user"),
"icinga2pass": config.get("icinga2", "icinga2.password"),
"icinga2ca": config.get("icinga2", "icinga2.ca"),
"icinga2fqdn": config.get("icinga2", "icinga2.fqdn"),
"icinga2port" : config.getint("icinga2", "icinga2.port")
"icinga2port": config.getint("icinga2", "icinga2.port"),
}
else:
print("Missing configuration file [/etc/icingabot/icingabot.conf].")
@ -88,43 +90,64 @@ class Icinga2ServiceManager:
# Since icinga2 wants « URL-encoded strings » but requests
# seems to makes « application/x-www-form-urlencoded »,
# so we munge params and shortcut urlencode.
# See https://www.icinga.com/docs/icinga2/latest/doc/12-icinga2-api/#parameters
url = 'https://{}:{}{}'.format(settings["icinga2fqdn"], settings["icinga2port"], uri)
return requests.Request('GET', url, params=params).prepare().url.replace('+', '%20')
# See
# https://www.icinga.com/docs/icinga2/latest/doc/12-icinga2-api/#parameters
url = "https://{}:{}{}".format(
settings["icinga2fqdn"], settings["icinga2port"], uri
)
return (
requests.Request("GET", url, params=params)
.prepare()
.url.replace("+", "%20")
)
def fetch_notifications(self):
headers = {
'Accept': 'application/json',
'X-HTTP-Method-Override': 'GET'
"Accept": "application/json",
"X-HTTP-Method-Override": "GET",
}
data = {
"attrs": ["last_check_result"],
"filter": "service.state!=ServiceOK",
}
try:
r = requests.post(self.build_request_url ("/v1/objects/services"),
r = requests.post(
self.build_request_url("/v1/objects/services"),
headers=headers,
auth=(settings["icinga2user"], settings["icinga2pass"]),
data=json.dumps(data),
verify=settings["icinga2ca"])
if (r.status_code == 200):
new_notifications = [n for n in r.json()['results'] if n != None]
news= [n for n in new_notifications if n['name'] not in [nn['name'] for nn in self.notifications]]
lost= [n for n in self.notifications if n['name'] not in [nn['name'] for nn in new_notifications ]]
verify=settings["icinga2ca"],
)
if r.status_code == 200:
new_notifications = [
n for n in r.json()["results"] if n is not None
]
news = [
n
for n in new_notifications
if n["name"]
not in [nn["name"] for nn in self.notifications]
]
lost = [
n
for n in self.notifications
if n["name"]
not in [nn["name"] for nn in new_notifications]
]
self.notifications = new_notifications
return (lost, news)
except:
except Exception:
self.send("Unable to fetch from Icinga2")
return (False, False)
def ack_service(self, srv, comment, nick):
# weird but needed:
if comment == '':
comment = ' '
if comment == "":
comment = " "
# /weird
headers = {
'Accept': 'application/json',
'X-HTTP-Method-Override': 'POST'
"Accept": "application/json",
"X-HTTP-Method-Override": "POST",
}
data = {
"author": nick,
@ -132,57 +155,80 @@ class Icinga2ServiceManager:
}
params = {
"type": "Service",
"filter" : ('service.__name=="{}"'.format(srv) if srv != None else "service.state!=ServiceOK"),
"filter": (
'service.__name=="{}"'.format(srv)
if srv is not None
else "service.state!=ServiceOK"
),
}
try:
r = requests.post(self.build_request_url ("/v1/actions/acknowledge-problem", params=params),
r = requests.post(
self.build_request_url(
"/v1/actions/acknowledge-problem", params=params
),
headers=headers,
auth=(settings["icinga2user"], settings["icinga2pass"]),
data=json.dumps(data),
verify=settings["icinga2ca"])
verify=settings["icinga2ca"],
)
if r.status_code == 200:
for a in r.json()['results']:
for a in r.json()["results"]:
if a["code"] == 200.0:
self.send(a["status"])
if srv != None and not r.json()['results']:
if srv is not None and not r.json()["results"]:
self.send("No result for service name « {} »".format(srv))
else:
self.send("{} for service name « {} »".format(r.text, srv))
except:
except Exception:
self.send("Unable to post to Icinga2")
def recheck_service(self, srv):
headers = {
'Accept': 'application/json',
'X-HTTP-Method-Override': 'POST'
"Accept": "application/json",
"X-HTTP-Method-Override": "POST",
}
params = {
"type": "Service",
"filter" : ('service.__name=="{}"'.format(srv) if srv != None else "service.state!=ServiceOK"),
"filter": (
'service.__name=="{}"'.format(srv)
if srv is not None
else "service.state!=ServiceOK"
),
}
try:
r = requests.post(self.build_request_url ("/v1/actions/reschedule-check", params=params),
r = requests.post(
self.build_request_url(
"/v1/actions/reschedule-check", params=params
),
headers=headers,
auth=(settings["icinga2user"], settings["icinga2pass"]),
verify=settings["icinga2ca"])
verify=settings["icinga2ca"],
)
if r.status_code == 200:
for a in r.json()['results']:
for a in r.json()["results"]:
if a["code"] == 200.0:
self.send(a["status"])
if srv != None and not r.json()['results']:
if srv is not None and not r.json()["results"]:
self.send("No result for service name « {} »".format(srv))
else:
self.send("{} for service name « {} »".format(r.text, srv))
except:
except Exception:
self.send("Unable to post to Icinga2")
class IcingaBot(Icinga2ServiceManager, irc.bot.SingleServerIRCBot):
args= ''
args = ""
muted = False
nick_suffix= ''
nick_suffix = ""
def __init__(self, channel, nickname, server, port=6667):
irc.bot.SingleServerIRCBot.__init__(self, [(server, port)], nickname, nickname, reconnection_interval=300)
irc.bot.SingleServerIRCBot.__init__(
self,
[(server, port)],
nickname,
nickname,
reconnection_interval=300,
)
self.nick = nickname
self.channel = channel
self.connection.execute_every(30, self.refresh_notifications)
@ -191,36 +237,41 @@ class IcingaBot(Icinga2ServiceManager, irc.bot.SingleServerIRCBot):
def suffix_nick(self, suffix):
self.nick_suffix = suffix
if self.connection.is_connected():
self.connection.nick ('{}{}'.format(self.nick, suffix))
self.connection.nick("{}{}".format(self.nick, suffix))
def unmute(self):
self.muted = False
if self.notifications:
self.suffix_nick ('[{}]'.format(len(self.notifications)))
self.suffix_nick("[{}]".format(len(self.notifications)))
else:
self.suffix_nick ('')
self.suffix_nick("")
def send(self, msg):
if not self.muted and self.connection.is_connected():
for line in msg.split('\n'):
for line in msg.split("\n"):
self.connection.privmsg(self.channel, line)
def refresh_notifications(self):
lost, news = self.fetch_notifications()
if lost == False and news == False:
if lost is False and news is False:
return
if self.notifications:
self.suffix_nick ('[{}]'.format(len(self.notifications)))
self.suffix_nick("[{}]".format(len(self.notifications)))
else:
self.suffix_nick ('')
self.suffix_nick("")
for srv in lost:
if srv != None:
self.send("{} is OK".format(srv['name']))
if srv is not None:
self.send("{} is OK".format(srv["name"]))
for srv in news:
try:
self.send("{}: => {}".format(srv['name'], srv['attrs']['last_check_result']['output']))
except:
self.send("{}: => No check result.".format(srv['name']))
self.send(
"{}: => {}".format(
srv["name"],
srv["attrs"]["last_check_result"]["output"],
)
)
except Exception:
self.send("{}: => No check result.".format(srv["name"]))
def on_nicknameinuse(self, c, e):
c.nick(+"_")
@ -232,20 +283,23 @@ class IcingaBot(Icinga2ServiceManager, irc.bot.SingleServerIRCBot):
self.do_command(e, e.arguments[0])
def on_pubmsg(self, c, e):
if e.arguments[0].startswith (settings['irccmd']):
if e.arguments[0].startswith(settings["irccmd"]):
self.do_command(e, e.arguments[0][1:])
return
a = e.arguments[0].split(":", 1)
if len(a) > 1 and a[0].lower() == self.connection.get_nickname().lower():
if (
len(a) > 1
and a[0].lower() == self.connection.get_nickname().lower()
):
self.do_command(e, a[1].strip())
return
def do_cmd(self, s):
self.args = None
l = s.split(' ', 1)
cmd= l[0]
if len(l)>1:
self.args= l[1].strip()
tokens = s.split(" ", 1)
cmd = tokens[0]
if len(tokens) > 1:
self.args = tokens[1].strip()
for k, v in commands.items():
if cmd == k and hasattr(self, "do_" + cmd):
@ -263,7 +317,7 @@ class IcingaBot(Icinga2ServiceManager, irc.bot.SingleServerIRCBot):
self.send(" synonyms: {}".format(", ".join(v["synonyms"])))
def do_die(self, c, e):
print('Ok master, I die. Aaargh...')
print("Ok master, I die. Aaargh...")
self.die()
def do_leave(self, c, e):
@ -273,36 +327,42 @@ class IcingaBot(Icinga2ServiceManager, irc.bot.SingleServerIRCBot):
if self.notifications:
for srv in self.notifications:
try:
self.send("{}: => {}".format(srv['name'], srv['attrs']['last_check_result']['output']))
except:
self.send("{}: => No check result.".format(srv['name']))
self.send(
"{}: => {}".format(
srv["name"],
srv["attrs"]["last_check_result"]["output"],
)
)
except Exception:
self.send("{}: => No check result.".format(srv["name"]))
else:
self.send("Nothing particularly exciting.")
def do_ack(self, c, e):
if self.args == None:
self.send(e.source.nick + ": usage: !ack <service|all> [: comment]")
if self.args is None:
self.send(
e.source.nick + ": usage: !ack <service|all> [: comment]"
)
return
l = self.args.split(':', 1)
srv, comment= l[0].strip(), ''
if len(l)>1:
comment= l[1].strip()
if srv == 'all':
tokens = self.args.split(":", 1)
srv, comment = tokens[0].strip(), ""
if len(tokens) > 1:
comment = tokens[1].strip()
if srv == "all":
srv = None
self.ack_service(srv, comment, e.source.nick)
def do_recheck(self, c, e):
if self.args == 'all':
self.args == None
if self.args == "all":
self.args = None
self.recheck_service(self.args)
self.connection.execute_delayed(1, self.refresh_notifications)
def do_mute(self, c, e):
self.muted = True
self.suffix_nick ('[zZz]')
self.suffix_nick("[zZz]")
self.connection.execute_delayed(3600, self.refresh_notifications)
def do_command(self, e, cmd):
nick = e.source.nick
c = self.connection
@ -313,9 +373,16 @@ class IcingaBot(Icinga2ServiceManager, irc.bot.SingleServerIRCBot):
else:
self.send(nick + ": Not understood: " + cmd)
def main():
bot = IcingaBot(settings['ircchan'], settings['ircnick'], settings['ircsrv'], settings['ircport'])
bot = IcingaBot(
settings["ircchan"],
settings["ircnick"],
settings["ircsrv"],
settings["ircport"],
)
bot.start()
if __name__ == "__main__":
main()