diff options
Diffstat (limited to 'plugin.py')
| -rw-r--r-- | plugin.py | 258 |
1 files changed, 163 insertions, 95 deletions
@@ -28,11 +28,23 @@ ### -from supybot import utils, plugins, ircutils, callbacks, ircdb, conf, log, world, ircmsgs +from supybot import ( + utils, + plugins, + ircutils, + callbacks, + ircdb, + conf, + log, + world, + ircmsgs, +) from supybot.commands import * + try: from supybot.i18n import PluginInternationalization - _ = PluginInternationalization('SnoParser') + + _ = PluginInternationalization("SnoParser") except ImportError: # Placeholder that allows to run the plugin on a bot # without the i18n module @@ -49,19 +61,21 @@ from datetime import timedelta from ipwhois import IPWhois import ipwhois + class SnoParser(callbacks.Plugin): """Parses the Server Notices from ErgoIRCd""" + threaded = True def redis_connect_whois(self) -> redis.client.Redis: try: redis_client_whois = redis.Redis( - host = self.registryValue('redis.host'), - port = self.registryValue('redis.port'), - password = self.registryValue('redis.password'), - username = self.registryValue('redis.username'), - db = self.registryValue('whois.redis.db'), - socket_timeout = int(self.registryValue('redis.timeout')) + host=self.registryValue("redis.host"), + port=self.registryValue("redis.port"), + password=self.registryValue("redis.password"), + username=self.registryValue("redis.username"), + db=self.registryValue("whois.redis.db"), + socket_timeout=int(self.registryValue("redis.timeout")), ) ping = redis_client_whois.ping() if ping is True: @@ -72,12 +86,12 @@ class SnoParser(callbacks.Plugin): def redis_connect_nicks(self) -> redis.client.Redis: try: redis_client_nicks = redis.Redis( - host = self.registryValue('redis.host'), - port = self.registryValue('redis.port'), - password = self.registryValue('redis.password'), - username = self.registryValue('redis.username'), - db = self.registryValue('redis.db1'), - socket_timeout = int(self.registryValue('redis.timeout')) + host=self.registryValue("redis.host"), + port=self.registryValue("redis.port"), + password=self.registryValue("redis.password"), + username=self.registryValue("redis.username"), + db=self.registryValue("redis.db1"), + socket_timeout=int(self.registryValue("redis.timeout")), ) ping = redis_client_nicks.ping() if ping is True: @@ -88,12 +102,12 @@ class SnoParser(callbacks.Plugin): def redis_connect_ips(self) -> redis.client.Redis: try: redis_client_ips = redis.Redis( - host = self.registryValue('redis.host'), - port = self.registryValue('redis.port'), - password = self.registryValue('redis.password'), - username = self.registryValue('redis.username'), - db = self.registryValue('redis.db2'), - socket_timeout = int(self.registryValue('redis.timeout')) + host=self.registryValue("redis.host"), + port=self.registryValue("redis.port"), + password=self.registryValue("redis.password"), + username=self.registryValue("redis.username"), + db=self.registryValue("redis.db2"), + socket_timeout=int(self.registryValue("redis.timeout")), ) ping = redis_client_ips.ping() if ping is True: @@ -101,7 +115,6 @@ class SnoParser(callbacks.Plugin): except redis.AuthenticationError: print("Could not authenticate to Redis backend.") - def __init__(self, irc): super().__init__(irc) self.redis_client_whois = self.redis_connect_whois() @@ -111,19 +124,19 @@ class SnoParser(callbacks.Plugin): def whois_fresh(self, sourceip: str) -> dict: """Data from WHOIS backend (IANA or respective RIR).""" asn = 0 - subnet = '' + subnet = "" try: whois = IPWhois(sourceip) - whoisres = whois.lookup_rdap(depth=1,retry_count=0) + whoisres = whois.lookup_rdap(depth=1, retry_count=0) results = whoisres - if self.registryValue('whois.debug'): + if self.registryValue("whois.debug"): print(results) - asn = whoisres['asn_registry'] - country = whoisres['asn_country_code'] - description = whoisres['asn_description'] - whoisout = asn + ' ' + country + ' ' + description + asn = whoisres["asn_registry"] + country = whoisres["asn_country_code"] + description = whoisres["asn_description"] + whoisout = asn + " " + country + " " + description except ipwhois.exceptions.IPDefinedError: - whoisout = 'RFC 4291 (Local)' + whoisout = "RFC 4291 (Local)" response = whoisout return response @@ -133,16 +146,20 @@ class SnoParser(callbacks.Plugin): k = self.redis_client_whois.get(key) -# self = SnoParser() -# val = self.redis_client_whois.get(key) + # self = SnoParser() + # val = self.redis_client_whois.get(key) val = k return val def whois_set_cache(self, key: str, value: str) -> bool: """Data to Redis.""" - duration = int(self.registryValue('whois.ttl')) - state = self.redis_client_whois.setex(key, timedelta(seconds=duration), value=value,) + duration = int(self.registryValue("whois.ttl")) + state = self.redis_client_whois.setex( + key, + timedelta(seconds=duration), + value=value, + ) return state def whois_run(self, sourceip: str) -> dict: @@ -151,20 +168,22 @@ class SnoParser(callbacks.Plugin): data = self.whois_get_cache(key=sourceip) if data is not None: data = json.loads(data) - if self.registryValue('whois.debug'): + if self.registryValue("whois.debug"): print("SNOPARSER DEBUG - WHOIS_RUN WITH CACHE: TRUE") print(data) print(sourceip) return data else: data = self.whois_fresh(sourceip) - if self.registryValue('whois.debug'): + if self.registryValue("whois.debug"): print("SNOPARSER DEBUG - WHOIS_RUN WITH CACHE: FALSE") print(data) print(sourceip) if data.startswith: - if self.registryValue('whois.debug'): - print("SNOPARSER DEBUG - WHOIS_RUN WITH CACHE: FALSE AND CORRECT STARTING CHARACTER") + if self.registryValue("whois.debug"): + print( + "SNOPARSER DEBUG - WHOIS_RUN WITH CACHE: FALSE AND CORRECT STARTING CHARACTER" + ) print(data) data = json.dumps(data) state = self.whois_set_cache(key=sourceip, value=data) @@ -172,8 +191,10 @@ class SnoParser(callbacks.Plugin): if state is True: return json.loads(data) else: - if self.registryValue('whois.debug'): - print("SNOPARSER DEBUG _ WHOIS_RUN WITH CACHE: FALSE AND WRONG STARTING CHARACTER") + if self.registryValue("whois.debug"): + print( + "SNOPARSER DEBUG _ WHOIS_RUN WITH CACHE: FALSE AND WRONG STARTING CHARACTER" + ) print(data) return data @@ -183,24 +204,24 @@ class SnoParser(callbacks.Plugin): data = self.redis_client_nicks.get(nickname) if data is not None: - if self.registryValue('debug'): + if self.registryValue("debug"): print("SNOPARSER DEBUG - NICK_RUN, SEEN: TRUE") print(nickname) print(data) - self.redis_client_nicks.incrby(nickname,amount=1) + self.redis_client_nicks.incrby(nickname, amount=1) if data: - decoded = data.decode('utf-8') + decoded = data.decode("utf-8") return decoded else: return 0 else: - if self.registryValue('debug'): + if self.registryValue("debug"): print("SNOPARSER DEBUG _ NICK_RUN, SEEN: FALSE") print(nickname) print(data) - self.redis_client_nicks.set(nickname,value='1') + self.redis_client_nicks.set(nickname, value="1") if data: - decoded = data.decode('utf-8') + decoded = data.decode("utf-8") return decoded else: return 0 @@ -211,24 +232,24 @@ class SnoParser(callbacks.Plugin): data = self.redis_client_ips.get(ipaddress) if data is not None: - if self.registryValue('debug'): + if self.registryValue("debug"): print("SNOPARSER DEBUG - IP_RUN, SEEN: TRUE") print(ipaddress) print(data) - self.redis_client_ips.incrby(ipaddress,amount=1) + self.redis_client_ips.incrby(ipaddress, amount=1) if data: - decoded = data.decode('utf-8') + decoded = data.decode("utf-8") return decoded else: return 0 else: - if self.registryValue('debug'): + if self.registryValue("debug"): print("SNOPARSER DEBUG _ IP_RUN, SEEN: FALSE") print(ipaddress) print(data) - self.redis_client_ips.set(ipaddress,value='1') + self.redis_client_ips.set(ipaddress, value="1") if data: - decoded = data.decode('utf-8') + decoded = data.decode("utf-8") return decoded else: return 0 @@ -239,32 +260,33 @@ class SnoParser(callbacks.Plugin): """ data = self.whois_get_cache(key=ipaddress) - decoded_data = data.decode('utf-8') + decoded_data = data.decode("utf-8") ttl = self.redis_client_whois.ttl(ipaddress) count = self.redis_client_ips.get(ipaddress) - decoded_count = count.decode('utf-8') - print('SnoParser manual query: ', data, ' ', ttl) - irc.reply(f'{decoded_data} - Count: {decoded_count} - Remaining: {ttl}s') - - ipquery = wrap(ipquery, ['ip']) + decoded_count = count.decode("utf-8") + print("SnoParser manual query: ", data, " ", ttl) + irc.reply(f"{decoded_data} - Count: {decoded_count} - Remaining: {ttl}s") + ipquery = wrap(ipquery, ["ip"]) def doNotice(self, irc, msg): (target, text) = msg.args if target == irc.nick: - # server notices CONNECT, KILL, XLINE, NICK, ACCOUNT, OPER, QUIT, + # server notices CONNECT, KILL, XLINE, NICK, ACCOUNT, OPER, QUIT, text = ircutils.stripFormatting(text) # if 'CONNECT' in text: - RE_CLICONN = re.compile(r"^-CONNECT- Client connected \[(.+)\] \[u\:~(.+)\] \[h\:(.+)\] \[ip\:(.+)\] \[r\:(.+)\]$") + RE_CLICONN = re.compile( + r"^-CONNECT- Client connected \[(.+)\] \[u\:~(.+)\] \[h\:(.+)\] \[ip\:(.+)\] \[r\:(.+)\]$" + ) couple = RE_CLICONN.match(text) # check `if couple:` ie was there a match even? if yes proceed if couple: nickname = couple.group(1) username = couple.group(2) host = couple.group(3) - if self.registryValue('whois.sample'): - ip = self.registryValue('whois.sample') + if self.registryValue("whois.sample"): + ip = self.registryValue("whois.sample") else: ip = couple.group(4) realname = couple.group(5) @@ -273,12 +295,23 @@ class SnoParser(callbacks.Plugin): nick_seen = self.nick_run(nickname=nickname) whois = self.whois_run(sourceip=ip) - snote_dict = {'notice': 'connect', 'nickname': nickname, 'username': username, 'host': host, 'ip': ip, 'realname': realname, 'ipCount': ip_seen, 'nickCount': nick_seen} + snote_dict = { + "notice": "connect", + "nickname": nickname, + "username": username, + "host": host, + "ip": ip, + "realname": realname, + "ipCount": ip_seen, + "nickCount": nick_seen, + } repl = f"\x02\x1F{snote_dict['notice']} \x0F\x11\x0303==>>\x0F \x02Nick:\x0F {snote_dict['nickname']} \x02Username:\x0F {snote_dict['username']} \x02Hostname:\x0F {snote_dict['host']} \x02IP:\x0F {snote_dict['ip']} {whois} \x02Realname:\x0F {snote_dict['realname']} \x02IPcount:\x0F {snote_dict['ipCount']} \x02NickCount:\x0F {snote_dict['nickCount']}" self._sendSnotice(irc, msg, repl) # if 'XLINE' in text and 'temporary' in text: - RE_XLINE = re.compile(r"^-XLINE- (.+) \[(.+)\] added temporary \((.+)\) (K-Line|D-Line) for (.+)$") + RE_XLINE = re.compile( + r"^-XLINE- (.+) \[(.+)\] added temporary \((.+)\) (K-Line|D-Line) for (.+)$" + ) couple = RE_XLINE.match(text) if couple: who = couple.group(1) @@ -286,30 +319,50 @@ class SnoParser(callbacks.Plugin): duration = couple.group(3) which_line = couple.group(4) host_or_ip = couple.group(5) - snote_dict = {'notice': 'tempban', 'who': who, 'operator': who_operator, 'duration': duration, 'type': which_line, 'target': host_or_ip} + snote_dict = { + "notice": "tempban", + "who": who, + "operator": who_operator, + "duration": duration, + "type": which_line, + "target": host_or_ip, + } repl = f"\x02\x1FNOTICE: {snote_dict['notice']}\x0F \x11\x0303 X_X \x0F \x02BannedBy:\x0F {snote_dict['who']} \x02BannedByOper:\x0F {snote_dict['operator']} \x02Duration:\x0F {snote_dict['duration']} \x02XLINE Type:\x0F {snote_dict['type']} \x02Nick:\x0F {snote_dict['target']}" self._sendSnotice(irc, msg, repl) # WHY THE FUCK IS IT elif ?? - elif 'XLINE' in text and 'temporary' not in text and 'removed' not in text: - perm_xline_regex = "^-XLINE- (.+) \[(.+)\] added (D-Line|K-Line) for (.+)$" + elif "XLINE" in text and "temporary" not in text and "removed" not in text: + perm_xline_regex = ( + "^-XLINE- (.+) \[(.+)\] added (D-Line|K-Line) for (.+)$" + ) couple = re.match(perm_xline_regex, text) who = couple.group(1) who_operator = couple.group(2) which_line = couple.group(3) host_or_ip = couple.group(4) - snote_dict = {'notice': 'Permaban', 'who': who, 'operator': who_operator, 'type': which_line, 'target': host_or_ip} + snote_dict = { + "notice": "Permaban", + "who": who, + "operator": who_operator, + "type": which_line, + "target": host_or_ip, + } repl = f"\x02\x1FNOTICE: {snote_dict['notice']} \x0F \x11\x0303 X_X \x0F \x02BannedBy:\x0F {snote_dict['who']} \x02BannedByOper:\x0F {snote_dict['operator']} \x02XLINE Type:\x0F {snote_dict['type']} \x02Host/IP:\x0F {snote_dict['target']}" self._sendSnotice(irc, msg, repl) - elif 'XLINE' in text and 'removed' in text: + elif "XLINE" in text and "removed" in text: unxlineregex = "^-XLINE- (.+) removed (D-Line|K-Line) for (.+)$" couple = re.match(unxlineregex, text) who = couple.group(1) which_line = couple.group(2) host_or_ip = couple.group(3) - snote_dict = {'notice': 'unxline', 'who': who, 'type': which_line, 'target': host_or_ip} + snote_dict = { + "notice": "unxline", + "who": who, + "type": which_line, + "target": host_or_ip, + } repl = f"\x02\x1FNOTICE: {snote_dict['notice']} \x0F\x11\x0303 :=D\x0F \x02UnbannedBy:\x0F {snote_dict['who']} \x02XLINE type:\x0F {snote_dict['type']} \x02Host/IP:\x0F {snote_dict['target']}" self._sendSnotice(irc, msg, repl) - if 'KILL' in text: + if "KILL" in text: killregex = "^-KILL- (.+) \[(.+)\] killed (\d) clients with a (KLINE|DLINE) \[(.+)\]$" couple = re.match(killregex, text) who = couple.group(1) @@ -317,49 +370,69 @@ class SnoParser(callbacks.Plugin): clients = couple.group(3) which_line = couple.group(4) nick = couple.group(5) - snote_dict = {'notice': 'kill', 'who': who, 'operator': who_operator, "client": clients, 'type': which_line, 'nick': nick} + snote_dict = { + "notice": "kill", + "who": who, + "operator": who_operator, + "client": clients, + "type": which_line, + "nick": nick, + } repl = f"\x02\x1FNOTICE: {snote_dict['notice']} \x0F\x11\x0303☠\x0F \x02KilledBy:\x0F {snote_dict['who']} \x02KilledByOper:\x0F {snote_dict['operator']} \x02NumofClientsAffected:\x0F {snote_dict['client']} \x02XLINE Type:\x0F {snote_dict['type']} \x02Nick:\x0F {snote_dict['nick']}" self._sendSnotice(irc, msg, repl) - if 'NICK' in text and 'changed nickname to' in text: + if "NICK" in text and "changed nickname to" in text: nickregex = "^-NICK- (.+) changed nickname to (.+)$" couple = re.match(nickregex, text) old_nick = couple.group(1) new_nick = couple.group(2) - snote_dict = {'notice': 'nick change', 'old_nick': old_nick, 'new_nick': new_nick} + snote_dict = { + "notice": "nick change", + "old_nick": old_nick, + "new_nick": new_nick, + } repl = f"\x02\x1FNOTICE: {snote_dict['notice']} ==> {snote_dict['old_nick']} changed their nick to {snote_dict['new_nick']}" self._sendSnotice(irc, msg, repl) - if 'QUIT' in text and 'exited' in text: + if "QUIT" in text and "exited" in text: quitregex = "^-QUIT- (.+) exited the network$" couple = re.match(quitregex, text) nick = couple.group(1) - snote_dict = {'notice': 'quit', 'nick': nick} + snote_dict = {"notice": "quit", "nick": nick} repl = f"\x02\x1FNOTICE: quit nick: {nick} has exited the network" self._sendSnotice(irc, msg, repl) - if 'ACCOUNT' in text and 'registered account' in text: + if "ACCOUNT" in text and "registered account" in text: accregex = "^-ACCOUNT- Client \[(.*)\] registered account \[(.*)\] from IP (.*)$" couple = re.match(accregex, text) hostmask = couple.group(1) account = couple.group(2) ip = couple.group(3) - snote_dict = {'notice': 'accreg', 'hostmask': hostmask, 'account': account, 'ip': ip} + snote_dict = { + "notice": "accreg", + "hostmask": hostmask, + "account": account, + "ip": ip, + } repl = f"\x02\x1FNOTICE: accreg -> [{account}] was registered by hostmask [{hostmask}] from IP {ip}" # Trigger HS SET self._setvhost(irc, msg, account) self._sendSnotice(irc, msg, repl) - - if 'ACCOUNT' in text and 'registered account' in text and 'SAREGISTER' in text: + + if ( + "ACCOUNT" in text + and "registered account" in text + and "SAREGISTER" in text + ): accregex = "^-ACCOUNT- Operator \[(.*)\] registered account \[(.*)\] with SAREGISTER$" couple = re.match(accregex, text) oper = couple.group(1) account = couple.group(2) - snote_dict = {'notice': 'sareg', 'oper': oper, 'account': account} + snote_dict = {"notice": "sareg", "oper": oper, "account": account} repl = f"\x02\x1FNOTICE: sareg -> [{account}] was registered by operator [{oper}]" self._setvhost(irc, msg, account) self._sendSnotice(irc, msg, repl) - if 'OPER' in text and 'Client opered up' in text: + if "OPER" in text and "Client opered up" in text: operregex = "^-OPER- Client opered up \[(.*)\, \ (.*)\]$" couple = re.match(operregex, text) hostmask = couple.group(1) @@ -367,41 +440,36 @@ class SnoParser(callbacks.Plugin): print(couple) - snote_dict = {'notice': 'opered', 'hostmask': hostmask, 'oper': oper} + snote_dict = {"notice": "opered", "hostmask": hostmask, "oper": oper} repl = f"\x02\x1FNOTICE:\x0F [{hostmask}] opered up as [{oper}]." self._sendSnotice(irc, msg, repl) - if 'OPER' in text and 'Client deopered' in text: - + if "OPER" in text and "Client deopered" in text: operregex = "^-OPER- Client deopered \[(.*)\]" couple = re.match(operregex, text) account = couple.group(1) - snote_dict = {'notice': 'deopered', 'name': account} + snote_dict = {"notice": "deopered", "name": account} repl = f"\x02\x1FNOTICE:\x0F [{account}] opered down." self._sendSnotice(irc, msg, repl) - # Post Registration def _setvhost(self, irc, msg, account): - arg = ['SET'] + arg = ["SET"] arg.append(account) - vhost = self.registryValue('AutoVhost') - arg.append(f'{vhost}{account}') - irc.sendMsg(msg=ircmsgs.IrcMsg(command='HS', - args=arg)) - + vhost = self.registryValue("AutoVhost") + arg.append(f"{vhost}{account}") + irc.sendMsg(msg=ircmsgs.IrcMsg(command="HS", args=arg)) # Send formatted SNO to channel def _sendSnotice(self, irc, msg, repl): try: - channel = self.registryValue('targetChannel') + channel = self.registryValue("targetChannel") if irc.isChannel(channel): - irc.queueMsg(msg=ircmsgs.IrcMsg(command='NOTICE', - args=(channel, repl))) + irc.queueMsg(msg=ircmsgs.IrcMsg(command="NOTICE", args=(channel, repl))) # what sort of exception does one raise except: pass |