X-Git-Url: http://git.tremily.us/?a=blobdiff_plain;f=irkerd;h=b0bec84fbc9953d88b87c54f9fee6dc3db6a80c0;hb=dac73a4bbc8e25631ee6d53fa96dc1d596fadc03;hp=fd97424ff0a68b6f20f6cc933eba2d6832fcf4c3;hpb=2b3dfc7414743871ed7e50a13b90347358e1ed4b;p=irker.git diff --git a/irkerd b/irkerd index fd97424..b0bec84 100755 --- a/irkerd +++ b/irkerd @@ -43,11 +43,10 @@ CONNECTION_MAX = 200 # To avoid hitting a thread limit # No user-serviceable parts below this line -version = "1.20" +version = "2.4" import sys, getopt, urlparse, time, random, socket, signal, re import threading, Queue, SocketServer, select -import logging try: import simplejson as json # Faster, also makes us Python-2.4-compatible except ImportError: @@ -107,12 +106,13 @@ class IRCError(Exception): class IRCClient(): "An IRC client session to one or more servers." - def __init__(self): + def __init__(self, debuglevel): self.mutex = threading.RLock() self.server_connections = [] self.event_handlers = {} self.add_event_handler("ping", lambda c, e: c.ship("PONG %s" % e.target)) + self.debuglevel = debuglevel def newserver(self): "Initialize a new server-connection object." @@ -127,6 +127,7 @@ class IRCClient(): # Otherwise no other thread would ever be able to change # the shared state of an IRC object running this function. while True: + nextsleep = 0 with self.mutex: connected = [x for x in self.server_connections if x is not None and x.socket is not None] @@ -136,9 +137,9 @@ class IRCClient(): (insocks, _o, _e) = select.select(sockets, [], [], timeout) for s in insocks: connmap[s.fileno()].consume() - else: - time.sleep(timeout) + nextsleep = timeout + time.sleep(nextsleep) def add_event_handler(self, event, handler): "Set a handler to be called later." @@ -157,6 +158,11 @@ class IRCClient(): with self.mutex: self.server_connections.remove(connection) + def debug(self, level, errmsg): + "Debugging information." + if self.debuglevel >= level: + sys.stderr.write("irkerd: %s\n" % errmsg) + class LineBufferedStream(): "Line-buffer a read stream." crlf_re = re.compile(b'\r?\n') @@ -199,8 +205,8 @@ class IRCServerConnection(): def connect(self, server, port, nickname, password=None, username=None, ircname=None): - log.debug("connect(server=%r, port=%r, nickname=%r, ...)", - server, port, nickname) + self.master.debug(2, "connect(server=%r, port=%r, nickname=%r, ...)" % + (server, port, nickname)) if self.socket is not None: self.disconnect("Changing servers") @@ -249,7 +255,7 @@ class IRCServerConnection(): self.buffer.append(incoming) for line in self.buffer: - log.debug("FROM: %s", line) + self.master.debug(2, "FROM: %s" % line) if not line: continue @@ -289,8 +295,8 @@ class IRCServerConnection(): target = arguments[0] arguments = arguments[1:] - log.debug("command: %s, source: %s, target: %s, " - "arguments: %s", command, prefix, target, arguments) + self.master.debug(2, + "command: %s, source: %s, target: %s, arguments: %s" % (command, prefix, target, arguments)) self.handle_event(Event(command, prefix, target, arguments)) def handle_event(self, event): @@ -305,7 +311,7 @@ class IRCServerConnection(): def disconnect(self, message=""): if self.socket is None: return - self.quit(message) + # Don't send a QUIT here - causes infinite loop! try: self.socket.shutdown(socket.SHUT_WR) self.socket.close() @@ -334,7 +340,6 @@ class IRCServerConnection(): self.ship("PRIVMSG %s :%s" % (target, text)) def quit(self, message=""): - # Triggers an error that forces a disconnect. self.ship("QUIT" + (message and (" :" + message))) def user(self, username, realname): @@ -343,8 +348,8 @@ class IRCServerConnection(): def ship(self, string): "Ship a command to the server, appending CR/LF" try: - self.socket.send(string + b'\r\n') - log.debug("TO: %s", string) + self.socket.send(string.encode('utf-8') + b'\r\n') + self.master.debug(2, "TO: %s" % string) except socket.error: self.disconnect("Connection reset by peer.") @@ -389,12 +394,12 @@ class Connection: def handle_welcome(self): "The server says we're OK, with a non-conflicting nick." self.status = "ready" - self.irker.debug(1, "nick %s accepted" % self.nickname()) + self.irker.irc.debug(1, "nick %s accepted" % self.nickname()) if password: self.connection.privmsg("nickserv", "identify %s" % password) def handle_badnick(self): "The server says our nick is ill-formed or has a conflict." - self.irker.debug(1, "nick %s rejected" % self.nickname()) + self.irker.irc.debug(1, "nick %s rejected" % self.nickname()) if fallback: # Randomness prevents a malicious user or bot from # anticipating the next trial name in order to block us @@ -407,7 +412,8 @@ class Connection: def handle_disconnect(self): "Server disconnected us for flooding or some other reason." self.connection = None - self.status = "disconnected" + if self.status != "expired": + self.status = "disconnected" def handle_kick(self, outof): "We've been kicked." self.status = "handshaking" @@ -424,7 +430,7 @@ class Connection: for (channel, message, key) in qcopy: self.queue.put((channel, message, key)) self.status = "ready" - def enqueue(self, channel, message, key): + def enqueue(self, channel, message, key, quit_after=False): "Enque a message for transmission." if self.thread is None or not self.thread.is_alive(): self.status = "unseen" @@ -432,6 +438,8 @@ class Connection: self.thread.setDaemon(True) self.thread.start() self.queue.put((channel, message, key)) + if quit_after: + self.queue.put((channel, None, key)) def dequeue(self): "Try to ship pending messages from the queue." try: @@ -454,7 +462,7 @@ class Connection: self.status = "expired" break elif xmit_timeout or ping_timeout: - self.irker.debug(1, "timing out connection to %s at %s (ping_timeout=%s, xmit_timeout=%s)" % (self.servername, time.asctime(), ping_timeout, xmit_timeout)) + self.irker.irc.debug(1, "timing out connection to %s at %s (ping_timeout=%s, xmit_timeout=%s)" % (self.servername, time.asctime(), ping_timeout, xmit_timeout)) with self.irker.irc.mutex: self.connection.context = None self.connection.quit("transmission timeout") @@ -474,7 +482,7 @@ class Connection: # space forever would be a memory leak. self.status = "expired" break - elif not self.connection: + elif not self.connection and self.status != "expired": # Queue is nonempty but server isn't connected. with self.irker.irc.mutex: self.connection = self.irker.irc.newserver() @@ -490,14 +498,12 @@ class Connection: nickname=self.nickname(), username="irker", ircname="irker relaying client") - if hasattr(self.connection, "buffer"): - self.connection.buffer.errors = 'replace' self.status = "handshaking" - self.irker.debug(1, "XMIT_TTL bump (%s connection) at %s" % (self.servername, time.asctime())) + self.irker.irc.debug(1, "XMIT_TTL bump (%s connection) at %s" % (self.servername, time.asctime())) self.last_xmit = time.time() self.last_ping = time.time() except IRCServerConnectionError: - self.status = "disconnected" + self.status = "expired" elif self.status == "handshaking": if time.time() > self.last_xmit + HANDSHAKE_TTL: self.status = "expired" @@ -519,11 +525,14 @@ class Connection: (channel, message, key) = self.queue.get() if channel not in self.channels_joined: self.connection.join(channel, key=key) - self.irker.debug(1, "joining %s on %s." % (channel, self.servername)) + self.irker.irc.debug(1, "joining %s on %s." % (channel, self.servername)) + # None is magic - it's a request to quit the server + if message is None: + self.connection.quit() # An empty message might be used as a keepalive or # to join a channel for logging, so suppress the # privmsg send unless there is actual traffic. - if message: + elif message: for segment in message.split("\n"): # Truncate the message if it's too long, # but we're working with characters here, @@ -535,10 +544,11 @@ class Connection: try: self.connection.privmsg(channel, segment) except ValueError as err: - self.irker.debug(1, "irclib rejected a message to %s on %s because: %s" % (channel, self.servername, str(err))) + self.irker.irc.debug(1, "irclib rejected a message to %s on %s because: %s" % (channel, self.servername, str(err))) + self.irker.irc.debug(50, err.format_exc()) time.sleep(ANTI_FLOOD_DELAY) self.last_xmit = self.channels_joined[channel] = time.time() - self.irker.debug(1, "XMIT_TTL bump (%s transmission) at %s" % (self.servername, time.asctime())) + self.irker.irc.debug(1, "XMIT_TTL bump (%s transmission) at %s" % (self.servername, time.asctime())) self.queue.task_done() except: (exc_type, _exc_value, exc_traceback) = sys.exc_info() @@ -622,7 +632,7 @@ class Dispatcher: self.servername = servername self.port = port self.connections = [] - def dispatch(self, channel, message, key): + def dispatch(self, channel, message, key, quit_after=False): "Dispatch messages for our server-port combination." # First, check if there is room for another channel # on any of our existing connections. @@ -630,7 +640,7 @@ class Dispatcher: eligibles = [x for x in connections if x.joined_to(channel)] \ or [x for x in connections if x.accepting(channel)] if eligibles: - eligibles[0].enqueue(channel, message, key) + eligibles[0].enqueue(channel, message, key, quit_after) return # All connections are full up. Look for one old enough to be # scavenged. @@ -645,18 +655,21 @@ class Dispatcher: found_connection.part(drop_channel, "scavenged by irkerd") del found_connection.channels_joined[drop_channel] #time.sleep(ANTI_FLOOD_DELAY) - found_connection.enqueue(channel, message, key) + found_connection.enqueue(channel, message, key, quit_after) return # Didn't find any channels with no recent activity newconn = Connection(self.irker, self.servername, self.port) self.connections.append(newconn) - newconn.enqueue(channel, message, key) + newconn.enqueue(channel, message, key, quit_after) def live(self): "Does this server-port combination have any live connections?" self.connections = [x for x in self.connections if x.live()] return len(self.connections) > 0 + def pending(self): + "Return all connections with pending traffic." + return [x for x in self.connections if not x.queue.empty()] def last_xmit(self): "Return the time of the most recent transmission." return max(x.last_xmit for x in self.connections) @@ -665,7 +678,7 @@ class Irker: "Persistent IRC multiplexer." def __init__(self, debuglevel=0): self.debuglevel = debuglevel - self.irc = IRCClient() + self.irc = IRCClient(self.debuglevel) self.irc.add_event_handler("ping", self._handle_ping) self.irc.add_event_handler("welcome", self._handle_welcome) self.irc.add_event_handler("erroneusnickname", self._handle_badnick) @@ -676,18 +689,15 @@ class Irker: self.irc.add_event_handler("disconnect", self._handle_disconnect) self.irc.add_event_handler("kick", self._handle_kick) self.irc.add_event_handler("every_raw_message", self._handle_every_raw_message) + self.servers = {} + def thread_launch(self): thread = threading.Thread(target=self.irc.spin) thread.setDaemon(True) self.irc._thread = thread thread.start() - self.servers = {} def logerr(self, errmsg): "Log a processing error." sys.stderr.write("irkerd: " + errmsg + "\n") - def debug(self, level, errmsg): - "Debugging information." - if self.debuglevel >= level: - sys.stderr.write("irkerd: %s\n" % errmsg) def _handle_ping(self, connection, _event): "PING arrived, bump the last-received time for the connection." if connection.context: @@ -705,9 +715,6 @@ class Irker: if connection.context: cxt = connection.context arguments = event.arguments - # irclib 5.0 compatibility, because the maintainer is a fool - if callable(arguments): - arguments = arguments() for lump in arguments: if lump.startswith("DEAF="): if not logfile: @@ -716,7 +723,7 @@ class Irker: m = int(lump[12:]) for pref in "#&+": cxt.channel_limits[pref] = m - self.debug(1, "%s maxchannels is %d" + self.irc.debug(1, "%s maxchannels is %d" % (connection.server, m)) elif lump.startswith("CHANLIMIT=#:"): limits = lump[10:].split(",") @@ -726,24 +733,20 @@ class Irker: limit = int(limit) for c in prefixes: cxt.channel_limits[c] = limit - self.debug(1, "%s channel limit map is %s" + self.irc.debug(1, "%s channel limit map is %s" % (connection.server, cxt.channel_limits)) except ValueError: self.logerr("ill-formed CHANLIMIT property") def _handle_disconnect(self, connection, _event): "Server hung up the connection." - self.debug(1, "server %s disconnected" % connection.server) + self.irc.debug(1, "server %s disconnected" % connection.server) connection.close() if connection.context: connection.context.handle_disconnect() def _handle_kick(self, connection, event): "Server hung up the connection." target = event.target - # irclib 5.0 compatibility, because the maintainer continues - # to be a fool. - if callable(target): - target = target() - self.debug(1, "irker has been kicked from %s on %s" % (target, connection.server)) + self.irc.debug(1, "irker has been kicked from %s on %s" % (target, connection.server)) if connection.context: connection.context.handle_kick(target) def _handle_every_raw_message(self, _connection, event): @@ -752,7 +755,10 @@ class Irker: with open(logfile, "a") as logfp: logfp.write("%03f|%s|%s\n" % \ (time.time(), event.source, event.arguments[0])) - def handle(self, line): + def pending(self): + "Do we have any pending message traffic?" + return [k for (k, v) in self.servers.items() if v.pending()] + def handle(self, line, quit_after=False): "Perform a JSON relay request." try: request = json.loads(line.strip()) @@ -779,7 +785,7 @@ class Irker: return if target.server() not in self.servers: self.servers[target.server()] = Dispatcher(self, target.servername, target.port) - self.servers[target.server()].dispatch(target.channel, message, target.key) + self.servers[target.server()].dispatch(target.channel, message, target.key, quit_after=quit_after) # GC dispatchers with no active connections servernames = self.servers.keys() for servername in servernames: @@ -824,25 +830,26 @@ class IrkerUDPHandler(SocketServer.BaseRequestHandler): def usage(): sys.stdout.write(""" Usage: - irkerd [-d debuglevel] [-l logfile] [-n nick] [-p password] [-V] [-h] + irkerd [-d debuglevel] [-l logfile] [-n nick] [-p password] [-i channel message] [-V] [-h] Options -d set debug level -l set logfile -n set nick-style -p set nickserv password + -i immediate mode -V return irkerd version -h print this help dialog """) if __name__ == '__main__': - log = logging.getLogger(__name__) debuglvl = 0 + immediate = None namestyle = "irker%03d" password = None logfile = None try: - (options, arguments) = getopt.getopt(sys.argv[1:], "d:l:n:p:Vh") + (options, arguments) = getopt.getopt(sys.argv[1:], "d:i:l:n:p:Vh") except getopt.GetoptError as e: sys.stderr.write("%s" % e) usage() @@ -850,8 +857,8 @@ if __name__ == '__main__': for (opt, val) in options: if opt == '-d': # Enable debug/progress messages debuglvl = int(val) - if debuglvl > 1: - logging.basicConfig(level=logging.DEBUG) + elif opt == '-i': # Immediate mode - send one message, then exit. + immediate = val elif opt == '-l': # Logfile mode - report traffic read in logfile = val elif opt == '-n': # Force the nick @@ -866,19 +873,25 @@ if __name__ == '__main__': sys.exit(0) fallback = re.search("%.*d", namestyle) irker = Irker(debuglevel=debuglvl) - irker.debug(1, "irkerd version %s" % version) - try: - tcpserver = SocketServer.TCPServer((HOST, PORT), IrkerTCPHandler) - udpserver = SocketServer.UDPServer((HOST, PORT), IrkerUDPHandler) - for server in [tcpserver, udpserver]: - server = threading.Thread(target=server.serve_forever) - server.setDaemon(True) - server.start() + irker.irc.debug(1, "irkerd version %s" % version) + if immediate: + irker.irc.add_event_handler("quit", lambda _c, _e: sys.exit(0)) + irker.handle('{"to":"%s","privmsg":"%s"}' % (immediate, arguments[0]), quit_after=True) + irker.irc.spin() + else: + irker.thread_launch() try: - signal.pause() - except KeyboardInterrupt: - raise SystemExit(1) - except socket.error, e: - sys.stderr.write("irkerd: server launch failed: %r\n" % e) + tcpserver = SocketServer.TCPServer((HOST, PORT), IrkerTCPHandler) + udpserver = SocketServer.UDPServer((HOST, PORT), IrkerUDPHandler) + for server in [tcpserver, udpserver]: + server = threading.Thread(target=server.serve_forever) + server.setDaemon(True) + server.start() + try: + signal.pause() + except KeyboardInterrupt: + raise SystemExit(1) + except socket.error, e: + sys.stderr.write("irkerd: server launch failed: %r\n" % e) # end