SnoParser/plugin.py

###
# Copyright (c) 2021, mogad0n and Georg Pfuetzenreuter <georg@lysergic.dev>
# All rights reserved.
#
# Redistribution and use in source and binary forms, with or without
# modification, are permitted provided that the following conditions are met:
#
#   * Redistributions of source code must retain the above copyright notice,
#     this list of conditions, and the following disclaimer.
#   * Redistributions in binary form must reproduce the above copyright notice,
#     this list of conditions, and the following disclaimer in the
#     documentation and/or other materials provided with the distribution.
#   * Neither the name of the author of this software nor the name of
#     contributors to this software may be used to endorse or promote products
#     derived from this software without specific prior written consent.
#
# THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS"
# AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE
# IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE
# ARE DISCLAIMED.  IN NO EVENT SHALL THE COPYRIGHT OWNER OR CONTRIBUTORS BE
# LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR
# CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF
# SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS
# INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN
# CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE)
# ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE
# POSSIBILITY OF SUCH DAMAGE.

###

from supybot import (
    ircutils,
    callbacks,
    ircmsgs,
)
from supybot.commands import *

try:
    from supybot.i18n import PluginInternationalization

    _ = PluginInternationalization("SnoParser")
except ImportError:
    # Placeholder that allows to run the plugin on a bot
    # without the i18n module
    _ = lambda x: x

import re
import redis
import json
from datetime import timedelta
from ipwhois import IPWhois
from ipwhois.exceptions import IPDefinedError


class SnoParser(callbacks.Plugin):
    """Parses the Server Notices from ErgoIRCd"""

    threaded = True

    def _redis_connect(self, db) -> redis.client.Redis:
        try:
            redis_client = 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(f"redis.db.{db}"),
                socket_timeout=int(self.registryValue("redis.timeout")),
            )
            ping = redis_client.ping()
            if ping is True:
                return redis_client
        except redis.AuthenticationError:
            print("Could not authenticate to Redis backend.")

    def __init__(self, irc):
        super().__init__(irc)
        self.redis_clients = {
            'ips': self._redis_connect("ips"),
            'nicks': self._redis_connect("nicks"),
            'whois': self._redis_connect("whois"),
        }

    def _get_from_cache(self, db, key):
        """ Get value from Redis cache """

        return self.redis_clients[db].get(key)

    def whois_fresh(self, sourceip: str) -> dict:
        """Data from WHOIS backend (IANA or respective RIR)."""

        asn = 0
        subnet = ""
        try:
            whois = IPWhois(sourceip)
            whoisres = whois.lookup_rdap(depth=1, retry_count=0)
            results = whoisres
            if self.registryValue("whois.debug"):
                print(results)
            asn = whoisres["asn_registry"]
            country = whoisres["asn_country_code"]
            description = whoisres["asn_description"]
            whoisout = asn + " " + country + " " + description
        except IPDefinedError:
            whoisout = "RFC 4291 (Local)"

        response = whoisout
        return response

    def whois_set_cache(self, key: str, value: str) -> bool:
        """Data to Redis."""

        return self.redis_clients['whois'].setex(
            key,
            timedelta(seconds=int(self.registryValue("whois.ttl"))),
            value=value,
        )

    def whois_run(self, sourceip: str) -> dict:
        """Whois query router."""

        data = self._get_from_cache("whois", sourceip)
        if data is not None:
            data = json.loads(data)
            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"):
                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"
                    )
                    print(data)
                data = json.dumps(data)
                state = self.whois_set_cache(key=sourceip, value=data)

                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"
                    )
                    print(data)
            return data

    def nick_run(self, nickname: str) -> dict:
        """Tracks nicknames"""

        data = self._get_from_cache("nicks", nickname)

        if data is not None:
            if self.registryValue("debug"):
                print("SNOPARSER DEBUG - NICK_RUN, SEEN: TRUE")
                print(nickname)
                print(data)
            self.redis_clients['nicks'].incrby(nickname, amount=1)
            if data:
                decoded = data.decode("utf-8")
                return decoded
            else:
                return 0
        else:
            if self.registryValue("debug"):
                print("SNOPARSER DEBUG _ NICK_RUN, SEEN: FALSE")
                print(nickname)
                print(data)
            self.redis_clients['nicks'].set(nickname, value="1")
            if data:
                decoded = data.decode("utf-8")
                return decoded
            else:
                return 0

    def ip_run(self, ipaddress: str) -> dict:
        """Tracks IP addresses"""

        data = self._get_from_cache("ips", ipaddress)
        if data is not None:
            if self.registryValue("debug"):
                print("SNOPARSER DEBUG - IP_RUN, SEEN: TRUE")
                print(ipaddress)
                print(data)
            self.redis_clients['ips'].incrby(ipaddress, amount=1)
            if data:
                decoded = data.decode("utf-8")
                return decoded
            else:
                return 0
        else:
            if self.registryValue("debug"):
                print("SNOPARSER DEBUG _ IP_RUN, SEEN: FALSE")
                print(ipaddress)
                print(data)
            self.redis_clients['ips'].set(ipaddress, value="1")
            if data:
                decoded = data.decode("utf-8")
                return decoded
            else:
                return 0

    def ipquery(self, irc, msg, args, ipaddress):
        """<IP address>
        Queries the cache for an address.
        """

        data = self.redis_clients['whois'].get(sourceip)
        decoded_data = data.decode("utf-8")
        ttl = self.redis_clients['whois'].ttl(ipaddress)
        count = self.redis_clients['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"])

    def doNotice(self, irc, msg):
        (target, text) = msg.args

        if target == irc.nick:
            # 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\:(.+)\]$"
            )
            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")
                else:
                    ip = couple.group(4)
                realname = couple.group(5)

                ip_seen = self.ip_run(ipaddress=ip)
                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,
                }
                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 (.+)$"
            )
            couple = RE_XLINE.match(text)
            if couple:
                who = couple.group(1)
                who_operator = couple.group(2)
                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,
                }
                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 (.+)$"
                )
                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,
                }
                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:
                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,
                }
                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:
                killregex = "^-KILL- (.+) \[(.+)\] killed (\d) clients with a (KLINE|DLINE) \[(.+)\]$"
                couple = re.match(killregex, text)
                who = couple.group(1)
                who_operator = couple.group(2)
                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,
                }
                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:
                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,
                }
                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:
                quitregex = "^-QUIT- (.+) exited the network$"
                couple = re.match(quitregex, text)
                nick = couple.group(1)
                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:
                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,
                }
                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
            ):
                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}
                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:
                operregex = "^-OPER- Client opered up \[(.*), (.*)\]$"
                couple = re.match(operregex, text)
                hostmask = couple.group(1)
                oper = couple.group(2)

                print(couple)

                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:
                operregex = "^-OPER- Client deopered \[(.*)\]"
                couple = re.match(operregex, text)
                account = couple.group(1)
                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.append(account)
        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")
            if irc.isChannel(channel):
                irc.queueMsg(msg=ircmsgs.IrcMsg(command="NOTICE", args=(channel, repl)))
        # what sort of exception does one raise
        except:
            pass


Class = SnoParser


# vim:set shiftwidth=4 softtabstop=4 expandtab textwidth=79: