X-Git-Url: http://git.tremily.us/?a=blobdiff_plain;f=irkerd;h=2b2a9dffd63634f094bdeb3dc73078de6b933c0f;hb=0c2b37e65bf3a13117604d2790f2b79b277b0d3e;hp=1fe96a119e68d61b287fd8649fa0db7312fadfb9;hpb=32721b014d4128f94bb51d27af856f0e8d69d60c;p=irker.git diff --git a/irkerd b/irkerd index 1fe96a1..2b2a9df 100755 --- a/irkerd +++ b/irkerd @@ -1,4 +1,5 @@ #!/usr/bin/env python +from __future__ import with_statement """ irkerd - a simple IRC multiplexer daemon @@ -13,12 +14,15 @@ all listed channels. Note that the channel portion of the URL need *not* have a leading '#' unless the channel name itself does. Options: -d sets the debug-message level (probably only of interest to -developers). The -V option prints the program version and exits. +developers). -l sets a logfile to capture message traffic from +channels. -n sets the nick and -p the nickserv password. The -V +option prints the program version and exits. Design and code by Eric S. Raymond . See the project resource page at . -Requires Python 2.6 and the irc client library at version >= 2.0.2: see +Requires Python 2.6 or 2.5 with the simplejson library installed, and +the irc client library at version >= 3.4 which requires 2.6: see http://pypi.python.org/pypi/irc/ """ @@ -27,7 +31,6 @@ http://pypi.python.org/pypi/irc/ HOST = "localhost" PORT = 6659 -NAMESTYLE = "irker%03d" # IRC nick template - must contain '%d' XMIT_TTL = (3 * 60 * 60) # Time to live, seconds from last transmit PING_TTL = (15 * 60) # Time to live, seconds from last PING HANDSHAKE_TTL = 60 # Time to live, seconds from nick transmit @@ -37,32 +40,13 @@ UNSEEN_TTL = 60 # Time to live, seconds since first request CHANNEL_MAX = 18 # Max channels open per socket (default) ANTI_FLOOD_DELAY = 0.5 # Anti-flood delay after transmissions, seconds ANTI_BUZZ_DELAY = 0.09 # Anti-buzz delay after queue-empty check +CONNECTION_MAX = 200 # To avoid hitting a thread limit # No user-serviceable parts below this line -version = "1.8" +version = "1.18" -# This black magic imports support for green threads (coroutines), -# then has kinky sex with the import library internals, replacing -# "threading" with a coroutine-using imposter. Threads then become -# ultra-light-weight and cooperatively scheduled. -try: - import eventlet - eventlet.monkey_patch() - green_threads = True - # With greenlets we don't worry about thread exhaustion, only the - # file descriptor limit (typically 1024 on modern Unixes). Thus we - # can handle a lot more concurrent sessions and generate less - # join/leave spam under heavy load. - CONNECTION_MAX = 1000 -except ImportError: - # Threads are more expensive if we have to use OS-level ones - # rather than greenlets. We need to avoid pushing thread limits - # as well as fd limits. See security.txt for discussion. - CONNECTION_MAX = 200 - green_threads = False - -import sys, getopt, urlparse, time, random, socket +import sys, getopt, urlparse, time, random, socket, signal, re import threading, Queue, SocketServer import irc.client, logging try: @@ -87,19 +71,23 @@ except ImportError: # even if the queue is nonempty but efforts to connect have failed for # a long time. # -# There are multiple threads. One accepts incoming traffic from all servers. -# Each Connection also has a consumer thread and a thread-safe message queue. -# The program main appends messages to queues as JSON requests are received; -# the consumer threads try to ship them to servers. When a socket write -# stalls, it only blocks an individual consumer thread; if it stalls long -# enough, the session will be timed out. +# There are multiple threads. One accepts incoming traffic from all +# servers. Each Connection also has a consumer thread and a +# thread-safe message queue. The program main appends messages to +# queues as JSON requests are received; the consumer threads try to +# ship them to servers. When a socket write stalls, it only blocks an +# individual consumer thread; if it stalls long enough, the session +# will be timed out. This solves the biggest problem with a +# single-threaded implementation, which is that you can't count on a +# single stalled write not hanging all other traffic - you're at the +# mercy of the length of the buffers in the TCP/IP layer. # # Message delivery is thus not reliable in the face of network stalls, # but this was considered acceptable because IRC (notoriously) has the # same problem - there is little point in reliable delivery to a relay # that is down or unreliable. # -# This code uses only NICK, JOIN, MODE, and PRIVMSG. It is strictly +# This code uses only NICK, JOIN, PART, MODE, and PRIVMSG. It is strictly # compliant to RFC1459, except for the interpretation and use of the # DEAF and CHANLIMIT and (obsolete) MAXCHANNELS features. CHANLIMIT # is as described in the Internet RFC draft @@ -126,7 +114,10 @@ class Connection: "Return a name for the nth server connection." if n is None: n = self.nick_trial - return (NAMESTYLE % n) + if fallback: + return (namestyle % n) + else: + return namestyle def handle_ping(self): "Register the fact that the server has pinged this connection." self.last_ping = time.time() @@ -134,14 +125,20 @@ class Connection: "The server says we're OK, with a non-conflicting nick." self.status = "ready" self.irker.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 has a conflict." + "The server says our nick is ill-formed or has a conflict." self.irker.debug(1, "nick %s rejected" % self.nickname()) - # Randomness prevents a malicious user or bot from antcipating the - # next trial name in order to block us from completing the handshake. - self.nick_trial += random.randint(1, 3) - self.last_xmit = time.time() - self.connection.nick(self.nickname()) + if fallback: + # Randomness prevents a malicious user or bot from + # anticipating the next trial name in order to block us + # from completing the handshake. + self.nick_trial += random.randint(1, 3) + self.last_xmit = time.time() + self.connection.nick(self.nickname()) + # Otherwise fall through, it might be possible to + # recover manually. def handle_disconnect(self): "Server disconnected us for flooding or some other reason." self.connection = None @@ -187,12 +184,16 @@ class Connection: now = time.time() xmit_timeout = now > self.last_xmit + XMIT_TTL ping_timeout = now > self.last_ping + PING_TTL - if (xmit_timeout or ping_timeout) and self.status != "disconnected": + if self.status == "disconnected": + # If the queue is empty, we can drop this 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.connection.context = None - self.connection.quit("transmission timeout") - self.connection.close() - self.connection = None + with self.irker.irc.mutex: + self.connection.context = None + self.connection.quit("transmission timeout") + self.connection = None self.status = "disconnected" else: # Prevent this thread from hogging the CPU by pausing @@ -201,25 +202,37 @@ class Connection: # reflex arc it is highly unlikely any human will ever # notice. time.sleep(ANTI_BUZZ_DELAY) + elif self.status == "disconnected" \ + and time.time() > self.last_xmit + DISCONNECT_TTL: + # Queue is nonempty, but the IRC server might be + # down. Letting failed connections retain queue + # space forever would be a memory leak. + self.status = "expired" + break elif not self.connection: # Queue is nonempty but server isn't connected. - self.connection = self.irker.irc.server() - self.connection.context = self - # Try to avoid colliding with other instances - self.nick_trial = random.randint(1, 990) - self.channels_joined = {} - # This will throw irc.client.ServerConnectionError on failure - try: - self.connection.connect(self.servername, - self.port, - nickname=self.nickname(), - username="irker", - ircname="irker relaying client") - self.status = "handshaking" - self.irker.debug(1, "XMIT_TTL bump (%s connection) at %s" % (self.servername, time.asctime())) - self.last_xmit = time.time() - except irc.client.ServerConnectionError: - self.status = "disconnected" + with self.irker.irc.mutex: + self.connection = self.irker.irc.server() + self.connection.context = self + # Try to avoid colliding with other instances + self.nick_trial = random.randint(1, 990) + self.channels_joined = {} + try: + # This will throw + # irc.client.ServerConnectionError on failure + self.connection.connect(self.servername, + self.port, + 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.last_xmit = time.time() + self.last_ping = time.time() + except irc.client.ServerConnectionError: + self.status = "disconnected" elif self.status == "handshaking": if time.time() > self.last_xmit + HANDSHAKE_TTL: self.status = "expired" @@ -228,13 +241,6 @@ class Connection: # Don't buzz on the empty-queue test while we're # handshaking time.sleep(ANTI_BUZZ_DELAY) - elif self.status == "disconnected" \ - and time.time() > self.last_xmit + DISCONNECT_TTL: - # Queue is nonempty, but the IRC server might be - # down. Letting failed connections retain queue - # space forever would be a memory leak. - self.status = "expired" - break elif self.status == "unseen" \ and time.time() > self.last_xmit + UNSEEN_TTL: # Nasty people could attempt a denial-of-service @@ -249,9 +255,23 @@ class Connection: if channel not in self.channels_joined: self.connection.join(channel) self.irker.debug(1, "joining %s on %s." % (channel, self.servername)) - for segment in message.split("\n"): - self.connection.privmsg(channel, segment) - time.sleep(ANTI_FLOOD_DELAY) + # 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: + for segment in message.split("\n"): + # Truncate the message if it's too long, + # but we're working with characters here, + # not bytes, so we could be off. + # 500 = 512 - CRLF - 'PRIVMSG ' - ' :' + maxlength = 500 - len(channel) + if len(segment) > maxlength: + segment = segment[:maxlength] + 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))) + 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.queue.task_done() @@ -259,10 +279,21 @@ class Connection: (exc_type, _exc_value, exc_traceback) = sys.exc_info() self.irker.logerr("exception %s in thread for %s" % \ (exc_type, self.servername)) + + # Maybe this should have its own status? + self.status = "expired" + # This is so we can see tracebacks for errors inside the thread # when we need to be able to for debugging purposes. if debuglvl > 0: raise exc_type, _exc_value, exc_traceback + finally: + try: + # Make sure we don't leave any zombies behind + self.connection.close() + except: + # Irclib has a habit of throwing fresh exceptions here. Ignore that + pass def live(self): "Should this connection not be scavenged?" return self.status != "expired" @@ -274,6 +305,9 @@ class Connection: if self.channel_limits: match_count = 0 for already in self.channels_joined: + # This obscure code is because the RFCs allow separate limits + # by channel type (indicated by the first character of the name) + # a feature that is almost never actually used. if already[0] == channel[0]: match_count += 1 return match_count < self.channel_limits.get(channel[0], CHANNEL_MAX) @@ -283,6 +317,8 @@ class Connection: class Target(): "Represent a transmission target." def __init__(self, url): + # Pre-2.6 Pythons don't recognize irc: as a valid URL prefix. + url = url.replace("irc://", "http://") parsed = urlparse.urlparse(url) irchost, _, ircport = parsed.netloc.partition(':') if not ircport: @@ -294,6 +330,9 @@ class Target(): # got kicked, and irkerd crashed because the server returned # "#channel" in the notification that our kick handler saw. self.channel = parsed.path.lstrip('/').lower() + # This deals with a tweak in recent versions of urlparse. + if parsed.fragment: + self.channel += "#" + parsed.fragment isnick = self.channel.endswith(",isnick") if isnick: self.channel = self.channel[:-7] @@ -351,7 +390,7 @@ class Dispatcher: return len(self.connections) > 0 def last_xmit(self): "Return the time of the most recent transmission." - return max([x.last_xmit for x in self.connections]) + return max(x.last_xmit for x in self.connections) class Irker: "Persistent IRC multiplexer." @@ -367,6 +406,7 @@ class Irker: self.irc.add_global_handler("featurelist", self._handle_features) self.irc.add_global_handler("disconnect", self._handle_disconnect) self.irc.add_global_handler("kick", self._handle_kick) + self.irc.add_global_handler("all_raw_messages", self._handle_all_raw_messages) thread = threading.Thread(target=self.irc.process_forever) thread.setDaemon(True) self.irc._thread = thread @@ -395,9 +435,14 @@ class Irker: "Determine if and how we can set deaf mode." if connection.context: cxt = connection.context - for lump in event.arguments(): + 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="): - connection.mode(cxt.nickname(), "+"+lump[5:]) + if not logfile: + connection.mode(cxt.nickname(), "+"+lump[5:]) elif lump.startswith("MAXCHANNELS="): m = int(lump[12:]) for pref in "#&+": @@ -419,13 +464,25 @@ class Irker: def _handle_disconnect(self, connection, _event): "Server hung up the connection." self.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." - self.debug(1, "irker has been kicked from %s on %s" % (event.target(), connection.server)) + 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)) if connection.context: - connection.context.handle_kick(event.target()) + connection.context.handle_kick(target) + def _handle_all_raw_messages(self, _connection, event): + "Log all messages when in watcher mode." + if logfile: + with open(logfile, "a") as logfp: + logfp.write("%03f|%s|%s\n" % \ + (time.time(), event.source, event.arguments[0])) def handle(self, line): "Perform a JSON relay request." try: @@ -473,13 +530,8 @@ class Irker: # choose the one longest idle on the # assumption that message activity is likely # to be clumpy. - oldest = None - oldtime = float("inf") if len(self.servers) >= CONNECTION_MAX: - for (name, server) in self.servers.items(): - if server.last_xmit() < oldtime: - oldest = name - oldtime = server.last_xmit() + oldest = min(self.servers.keys(), key=lambda name: self.servers[name].last_xmit()) del self.servers[oldest] except ValueError: self.logerr("can't recognize JSON on input: %r" % line) @@ -502,15 +554,25 @@ class IrkerUDPHandler(SocketServer.BaseRequestHandler): if __name__ == '__main__': debuglvl = 0 - (options, arguments) = getopt.getopt(sys.argv[1:], "d:V") + namestyle = "irker%03d" + password = None + logfile = None + (options, arguments) = getopt.getopt(sys.argv[1:], "d:l:n:p:V:") 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 == '-l': # Logfile mode - report traffic read in + logfile = val + elif opt == '-n': # Force the nick + namestyle = val + elif opt == '-p': # Set a nickserv password + password = val elif opt == '-V': # Emit version and exit sys.stdout.write("irkerd version %s\n" % version) sys.exit(0) + fallback = re.search("%.*d", namestyle) irker = Irker(debuglevel=debuglvl) irker.debug(1, "irkerd version %s" % version) try: @@ -521,8 +583,7 @@ if __name__ == '__main__': server.setDaemon(True) server.start() try: - while True: - time.sleep(10) + signal.pause() except KeyboardInterrupt: raise SystemExit(1) except socket.error, e: