# 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 set_agora(self, agora): self.agora = agora def set_revolut(self, revolut): self.revolut = revolut def set_tx(self, tx): self.tx = tx def set_notify(self, notify): self.notify = notify 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.revolut, 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: if msg.split()[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 set_agora(self, agora): self.agora = agora def set_revolut(self, revolut): self.revolut = revolut def set_tx(self, tx): self.tx = tx def set_notify(self, notify): self.notify = notify 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 self.client.set_agora(self.agora) self.client.set_revolut(self.revolut) self.client.set_tx(self.tx) self.client.set_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