bouncer.py

import znc

VERSION = '2'
COMMAND = "BOUNCER"

class bouncer(znc.Module):

    module_types = [znc.CModInfo.GlobalModule]
    description = "Provides KiwiIRC BOUNCER support"

    def OnLoad(self, args, message):
        return True

    def OnClientLogin(self):
        client = self.GetClient()
        self.send_isupport(client)

    def OnIRCConnected(self):
        net = self.GetNetwork()
        clients = net.GetClients()
        for client in clients:
            client.PutClient("{} state {} connected".format(COMMAND, net.GetName()))

    def OnUserRaw(self, line):
        line_split = str(line).split()
        client = self.GetClient()
        cmd = line_split[0].upper()
        subcmd = line_split[1].lower()
        params = line_split[2:]

        reply = ""

        if cmd == COMMAND:
            if subcmd == "connect":
                reply = self.subcmd_connect(params)
            elif subcmd == "disconnect":
                reply = self.subcmd_disconnect(params)
            elif subcmd == "listnetworks":
                reply = self.subcmd_listnetworks(params)
            elif subcmd == "listbuffers":
                reply = self.subcmd_listbuffers(params)
            elif subcmd == "addnetwork":
                reply = self.subcmd_addnetwork(params)
            elif subcmd == "changenetwork":
                reply = self.subcmd_changenetwork(params)
            elif subcmd == "delnetwork":
                reply = self.subcmd_delnetwork(params)

            if reply:
                self.reply(client, reply)

            return znc.HALTCORE

        elif line_split[0].upper() == "VERSION":
            self.send_isupport(client)

    def subcmd_connect(self, params):
        try:
            net = params[0]
        except IndexError:
            return "connect * ERR_INVALIDARGS"

        net = self.GetUser().FindNetwork(net)

        if not net:
            return "connect * ERR_NETNOTFOUND"

        net.SetIRCConnectEnabled(True)

        znc.CZNC.Get().WriteConfig()

        return "state {} connecting".format(net.GetName())

    def subcmd_disconnect(self, params):
        try:
            net = params[0]
        except IndexError:
            return "connect * ERR_INVALIDARGS"

        net = self.GetUser().FindNetwork(net)

        if not net:
            return "connect * ERR_NETNOTFOUND"

        net.SetIRCConnectEnabled(False)

        znc.CZNC.Get().WriteConfig()

        return "state {} disconnected".format(net.GetName())

    def subcmd_listnetworks(self, params):
        user = self.GetUser()
        nets = user.GetNetworks()
        net_list = []
        for net in nets:
            server = net.GetCurrentServer()
            state = "connected" if net.GetIRCConnectEnabled() else "disconnected"
            formatted_net = "network={};host={};port={};nick={};user={};state={}".format(net.GetName(), server.GetName(), server.GetPort(), net.GetNick(), net.GetIdent(), state)
            if server.IsSSL():
                formatted_net = formatted_net + ";tls=1"
            net_list.append("listnetworks {}".format(formatted_net))

        net_list.append("listnetworks RPL_OK")

        return net_list

    def subcmd_listbuffers(self, params):
        net = params[0]
        net = self.GetUser().FindNetwork(net)

        buff_list = []
        queries = net.GetQueries()
        for query in queries:
            if query.GetBufferCount() > 0:
                formatted_buff = "listbuffers network={};buffer={}".format(net.GetName(), query.GetName())
                buff_list.append(formatted_buff)
        chans = net.GetChans()
        for chan in chans:
            if chan.GetBufferCount() > 0:
                formatted_buff = "listbuffers {} network={};buffer={}".format(net.GetName(), net.GetName(), chan.GetName())
                if not chan.IsDisabled():
                    formatted_buff = formatted_buff + ";joined=1"
                topic = (chan.GetTopic()).replace(" ", "\s")
                topic = topic.encode('utf-8', 'ignore').decode('utf-8')
                formatted_buff = formatted_buff + ";topic={}".format(topic)
            buff_list.append(formatted_buff)

        buff_list.append("listbuffers {} RPL_OK".format(net.GetName()))

        print(buff_list)

        return buff_list

    def subcmd_addnetwork(self, params):
        znc_user = self.GetUser()
        try:
            params = dict(item.split("=") for item in params[0].split(";"))
        except ValueError:
            return "addnetwork * ERR_INVALIDFORMAT"

        try:
            network = params['network']
        except KeyError:
            return "addnetwork * ERR_NONETNAMEGIVEN"

        try:
            host = params['host']
        except:
            host = None

        try:
            port = int(params['port'])
        except TypeError:
            return "addnetwork * ERR_PORTINVALID"
        except KeyError:
            port = None

        try:
            nick = params['nick']
        except KeyError:
            nick = znc_user.GetNick()

        try:
            user = params['user']
        except KeyError:
            user = znc_user.GetIdent()

        net = znc_user.FindNetwork(network)
        if net:
            return "addnetwork {} ERR_NAMEINUSE".format(network)

        try:
            ssl = True if params['tls'] == "1" else False
        except KeyError:
            ssl = False

        net = znc.CIRCNetwork(znc_user, network)
        net.SetNick(nick)
        net.SetAltNick("{}_".format(nick))
        net.SetIdent(user)
        znc_user.AddNetwork(net)

        net.thisown = 0

        if host and port:
            server = net.AddServer(host, port, "", ssl)

        net.SetIRCConnectEnabled(True)

        znc.CZNC.Get().WriteConfig()

        return "addnetwork {} RPL_OK".format(network)

    def subcmd_changenetwork(self, params):
        try:
            net_name = params[0]
            params = dict(item.split("=") for item in params[1].split(";"))
            net = self.GetUser().FindNetwork(net_name)
        except (IndexError, ValueError):
            return "changenetwork * ERR_INVALIDARGS"

        if not net:
            return "changenetwork {} ERR_NETNOTFOUND".format(net_name)

        curr_server = net.GetCurrentServer()
        curr_host = curr_server.GetName()
        curr_port = curr_server.GetPort()
        curr_ssl = curr_server.IsSSL()

        try:
            host = params['host']
        except KeyError:
            host = curr_host

        try:
            port = int(params['port'])
        except KeyError:
            port = curr_port

        try:
            ssl = params['tls']
            ssl = True if ssl == '1' else False
        except KeyError:
            ssl = curr_ssl

        net.DelServers()
        net.AddServer(host, port, "", ssl)

        try:
            nick = params['nick']
            net.SetNick(nick)
        except KeyError:
            pass

        try:
            user = params['user']
            net.SetIdent(user)
        except KeyError:
            pass

        znc.CZNC.Get().WriteConfig()

        return "changenetwork {} RPL_OK".format(net_name)

    def subcmd_delnetwork(self, params):
        try:
            net = params[0]
        except IndexError:
            return "delnetwork * ERR_NEEDSNAME"

        user = self.GetUser()
        net = user.FindNetwork(net)

        if not net:
            return "delnetwork * ERR_NETNOTFOUND"

        net = net.GetName()

        user.DeleteNetwork(net)

        znc.CZNC.Get().WriteConfig()

        return "delnetwork {} RPL_OK".format(net)

    def reply(self, client, reply):
        if type(reply) is str:
            client.PutClient("{} {}".format(COMMAND, reply))
        elif type(reply) is list:
            for item in reply:
                client.PutClient("{} {}".format(COMMAND, item))

    def send_isupport(self, client):
        client.PutClient(':irc.host 005 {} {}={} :are supported by this server'.format(client.GetNick(), COMMAND, VERSION))