#! /usr/bin/env python import sys, ssl, random import irc.bot#, irc.strings #https://python-irc.readthedocs.io/en/latest/irc.html# #from irc.client import ip_numstr_to_quad#, ip_quad_to_numstr from jaraco.stream import buffer from postgres import Postgres # https://postgres-py.readthedocs.io/en/latest/ import events.on_welcome, events.on_join, events.on_kick, events.on_mode, events.on_pubmsg, events.on_action, events.on_whoreply, events.on_nick, events.on_part, events.on_quit, events.on_topic, events.on_privmsg, events.on_pubnotice, events.on_nicknameinuse, events.on_error, events.on_keyset, events.on_privnotice from common import log, font from events.common import Inform class PyRot(irc.bot.SingleServerIRCBot): def __init__(self, network, db, webgui): self.network = network self.db = db self.webgui = webgui # Pick a random host from the network. hosts = db.all("SELECT * FROM rotbot_host WHERE network_id=%(network)s", network=network.id) host = random.choice(hosts) connect_string = (host.address, host.port) irc.client.ServerConnection.buffer_class = buffer.LenientDecodingLineBuffer # Set buffer class. # self.protectees = {} # self.channelkeys = {} # Log and record to database. log.info('Connecting to %s:%s/%s' % (connect_string[0], connect_string[1], network.home_channel)) db.run("UPDATE rotbot_host SET connection_attempts = connection_attempts + 1 WHERE id=%s", [network.id]) # Create connect factory with correct scheme. if host.ssl: factory = irc.connection.Factory(wrapper=ssl.wrap_socket) else: factory = irc.connection.Factory() try: irc.bot.SingleServerIRCBot.__init__(self, [connect_string], network.nickname, network.username, connect_factory=factory) # Connect. except irc.client.ServerConnectionError: # Connection failure. sys.exit(sys.exc_info()[1]) # Exit. ## Events. def on_nicknameinuse(self, connection, event): # Bots nickname is already taken. events.on_nicknameinuse.process_event(self, connection, event) def on_welcome(self, connection, event): # IRC server welcomes bot after connecting. events.on_welcome.process_event(self, connection, event) # def on_error(self, connection, event): events.on_error.process_event(self, connection, event) def on_nick(self, connection, event): # User changes nickname. events.on_nick.process_event(self, connection, event) def on_join(self, connection, event): # User joins a channel. events.on_join.process_event(self, connection, event) def on_mode(self, connection, event): # Mode change. events.on_mode.process_event(self, connection, event) def on_kick(self, connection, event): # User kicks user from channel. events.on_kick.process_event(self, connection, event) def on_part(self, connection, event): # User parts channel. events.on_part.process_event(self, connection, event) def on_quit(self, connection, event): # User quit irc server. events.on_quit.process_event(self, connection, event) def on_invite(self, connection, event): # Channel invite events.on_invite.process_event(self, connection, event) def on_topic(self, connection, event): # Channel topic change. events.on_topic.process_event(self,connection, event) def on_pubmsg(self, connection, event): # Channel message events.on_pubmsg.process_event(self, connection, event) def on_action(self, connection, event): # Is this both channel and private actions? events.on_action.process_event(self, connection, event) def on_privmsg(self, connection, event): # Private message. events.on_privmsg.process_event(self, connection, event) def on_pubnotice(self, connection, event): # Channel notice events.on_pubnotice.process_event(self, connection, event) def on_privnotice(self, connection, event): # Private notice. events.on_privnotice.process_event(self, connection, event) def on_whoreply(self, connection, event): # events.on_whoreply.process_event(self, connection, event) pass def on_keyset(self, connection, event): events.on_keyset.process_event(self, connection, event) def on_yourhost(self, connection, event): log.info(event) def on_yourebannedcreep(self, connection, event): Inform.operators(self, connection, "I am banned " + font.red + event.arguments[0] + reset.reset + " from " + font.red + event.source) log.warning(event) def on_youwillbebanned(self, connection, event): log.warning(event) Inform.operators(self, connection, "I will be banned " + font.red + event.arguments[0] + reset.reset + " from " + font.red + event.source) # DCC stuff from originalexample file. def on_dccmsg(self, c, e): log.info(c, e) # non-chat DCC messages are raw bytes; decode as text #text = e.arguments[0].decode('utf-8') #c.privmsg("You said: " + text) def on_dccchat(self, c, e): log.info(c, e) # if len(e.arguments) != 2: # return # args = e.arguments[1].split() # if len(args) == 4: # try: # address = ip_numstr_to_quad(args[2]) # port = int(args[3]) # except ValueError: # return # self.dcc_connect(address, port) def main(): log.info('Starting RotBot: pyRot version.') # Validate system arguments and warn if needed.. if len(sys.argv) != 2: # Not 2 arguments. sys.exit('To run type "python bot.py database_network_id", not: %s. Terminating program.' % (sys.argv)) # Terminate program. instance = sys.argv[1] # Instance is the database network id. ### SETTINGS webgui = { 'base_url': 'https://rotbot.h0v1n8.nl/', # The django rotbot app url. 'register_url': '', } # Database credentials username = 'pyrot' password = 'oGPnbiqh55QKLhmnKQgS92h74j0e9d6LE58cSsD1' ### End of SETTINGS db_connect_string = 'postgres://%s:%s@localhost/website' % (username, password) try: db = Postgres(db_connect_string) # Connect to database. except: sys.exit('Database connection %s failed: %s' % (db_connect_string, sys.exc_info())) # Exit. try: network = db.one("SELECT * FROM rotbot_network WHERE id=%(instance)s", instance=instance) # Get network from database. except: sys.exit('Could not retrieve network %s from database: %s' % (instance, sys.exc_info())) # Exit. if network == None: # No data returned. sys.exit('Network %s not found, terminating program.' % (instance)) # Exit. if network.enabled == False: # Network disabled in database. sys.exit('Network %s marked as disabled in database, use webgui to enable. Terminating program.' % (network.name)) # Exit bot = PyRot(network, db, webgui) bot.start() if __name__ == "__main__": try: main() # Run the program. except KeyboardInterrupt: # User presses CTRL+C sys.exit('Interrupted by keyboard.') # Exit.