pluto/handler/irc.py

# Twisted/Klein imports
from twisted.logger import Logger
from twisted.words.protocols import irc
from twisted.internet import protocol, reactor, ssl
from twisted.internet.task import deferLater

# Project imports
from settings import settings
from commands import IRCCommands


class IRCBot(irc.IRCClient):
    def __init__(self, log):
        """
        Initialise IRC bot.
        :param log: logger instance
        :type log: Logger
        """
        self.log = log
        self.cmd = IRCCommands()
        # Parse the commands into "commandname": "commandclass"
        self.cmdhash = {getattr(self.cmd, x).name: x for x in dir(self.cmd) if not x.startswith("_")}
        self.nickname = settings.IRC.Nick
        self.password = settings.IRC.Pass
        self.realname = self.nickname
        self.username = self.nickname

        # Don't give away information about our client
        self.userinfo = None
        self.fingerReply = None
        self.versionName = None
        self.sourceURL = None
        self.lineRate = None  # Don't throttle messages, we may need to send a lot

        self.prefix = settings.IRC.Prefix
        self.admins = (settings.IRC.Admins).split("\n")
        self.highlight = (settings.IRC.Highlight).split("\n")

        self.channel = settings.IRC.Channel

    def parse(self, user, host, channel, msg):
        """
        Simple handler for IRC commands.
        :param user: full user string with host
        :param host: user's hostname
        :param channel: channel the message was received on
        :param msg: the message
        :type user: string
        :type host: string
        :type channel: string
        :type msg: string
        """
        spl = msg.split()
        # nick = user.split("!")[0]

        cmd = spl[0]
        length = len(spl)

        # Check if user is authenticated
        authed = host in self.admins
        if cmd == "help" and length == 2 and authed:
            if spl[1] in self.cmdhash:
                cmdname = self.cmdhash[spl[1]]
                obj = getattr(self.cmd, cmdname)
                helptext = getattr(obj, "helptext")
                self.msg(channel, helptext)
                return
            else:
                self.msg(channel, f"No such command: {spl[1]}")
                return
        if cmd == "helpall" and authed:
            for command in self.cmdhash:
                cmdname = self.cmdhash[command]
                obj = getattr(self.cmd, cmdname)
                helptext = getattr(obj, "helptext")
                self.msg(channel, f"{cmdname}: {helptext}")
            return

        if cmd in self.cmdhash:
            # Get the class name of the referenced command
            cmdname = self.cmdhash[cmd]
            # Get the class name
            obj = getattr(self.cmd, cmdname)

            def msgl(x):
                self.msg(channel, x)

            # Check if the command required authentication
            if obj.authed:
                if host in self.admins:
                    obj.run(cmd, spl, length, authed, msgl, self.agora, self.tx, self.notify)
                else:
                    # Handle authentication here instead of in the command module for security
                    self.msg(channel, "Access denied.")
            else:
                # Run an unauthenticated command, without passing through secure library calls
                obj.run(cmd, spl, len(spl), authed, msgl)
            return
        self.msg(channel, "Command not found.")
        if authed:
            # Give user command hints if they are authenticated
            self.msg(channel, f"Commands loaded: {', '.join(self.cmdhash.keys())}")

    def signedOn(self):
        """
        Called when we have signed on to IRC.
        Join our channel.
        """
        self.log.info("Signed on as %s" % (self.nickname))
        deferLater(reactor, 2, self.join, self.channel)

    def joined(self, channel):
        """
        Called when we have joined a channel.
        Setup the Agora LoopingCall to get trades.
        This is here to ensure the IRC client is initialised enough to send the trades.
        :param channel: channel we joined
        :type channel: string
        """
        self.agora.setup_loop()
        self.log.info("Joined channel %s" % (channel))

    def privmsg(self, user, channel, msg):
        """
        Called on received PRIVMSGs.
        Pass through identified commands to the parse function.
        :param user: full user string with host
        :param channel: channel the message was received on
        :param msg: the message
        :type user: string
        :type channel: string
        :type msg: string
        """
        nick = user.split("!")[0]

        if channel == self.nickname:
            channel = nick

        host = user.split("!")[1]
        host = host.split("@")[1]

        ident = user.split("!")[1]
        ident = ident.split("@")[0]
        self.log.info("(%s) %s: %s" % (channel, user, msg))
        if msg[0] == self.prefix:
            if len(msg) > 1:
                if msg.split()[0] != "!":
                    self.parse(user, host, channel, msg[1:])
        elif host in self.admins and channel == nick:
            if len(msg) > 0:
                spl = msg.split()
                if len(spl) > 0:
                    if spl[0] != "!":
                        self.parse(user, host, channel, msg)

    def noticed(self, user, channel, msg):
        """
        Called on received NOTICEs.
        :param user: full user string with host
        :param channel: channel the notice was received on
        :param msg: the message
        :type user: string
        :type channel: string
        :type msg: string
        """
        nick = user.split("!")[0]
        if channel == self.nickname:
            channel = nick
        # self.log.info("[%s] %s: %s" % (channel, user, msg))


class IRCBotFactory(protocol.ClientFactory):
    def __init__(self):
        self.log = Logger("irc")

    def sendmsg(self, msg):
        """
        Passthrough function to send a message to the channel.
        """
        if self.client:
            self.client.msg(self.client.channel, msg)
        else:
            self.log.error("Trying to send a message without connected client: {msg}", msg=msg)
            return

    def buildProtocol(self, addr):
        """
        Custom override for the Twisted buildProtocol so we can access the Protocol instance.
        Passes through the Agora instance to IRC.
        :return: IRCBot Protocol instance
        """
        prcol = IRCBot(self.log)
        self.client = prcol
        setattr(self.client, "agora", self.agora)
        setattr(self.client, "truelayer", self.truelayer)
        setattr(self.client, "nordigen", self.nordigen)
        setattr(self.client, "tx", self.tx)
        setattr(self.client, "notify", self.notify)
        return prcol

    def clientConnectionLost(self, connector, reason):
        """
        Called when connection to IRC server lost. Reconnect.
        :param connector: connector object
        :param reason: reason connection lost
        :type connector: object
        :type reason: string
        """
        self.log.error("Lost connection: {reason}, reconnecting", reason=reason)
        connector.connect()

    def clientConnectionFailed(self, connector, reason):
        """
        Called when connection to IRC server failed. Reconnect.
        :param connector: connector object
        :param reason: reason connection failed
        :type connector: object
        :type reason: string
        """
        self.log.error("Could not connect: {reason}", reason=reason)
        connector.connect()


def bot():
    """
    Load the certificates, start the Bot Factory and connect it to the IRC server.
    :return: Factory instance
    :rtype: Factory
    """
    # Load the certificates
    context = ssl.DefaultOpenSSLContextFactory(settings.IRC.Cert, settings.IRC.Cert)
    # Define the factory instance
    factory = IRCBotFactory()
    reactor.connectSSL(settings.IRC.Host, int(settings.IRC.Port), factory, context)
    return factory