X-Git-Url: http://git.tremily.us/?a=blobdiff_plain;f=irkerd;h=b0bec84fbc9953d88b87c54f9fee6dc3db6a80c0;hb=dac73a4bbc8e25631ee6d53fa96dc1d596fadc03;hp=b9814c1f21d62125e5cb2ae55cf04ddfa4d4676f;hpb=897be7c8b3380935759502091568ff0a9b80a479;p=irker.git diff --git a/irkerd b/irkerd index b9814c1..b0bec84 100755 --- a/irkerd +++ b/irkerd @@ -13,21 +13,23 @@ 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 - -http://pypi.python.org/pypi/irc/ +Requires Python 2.6 or 2.5 with the simplejson library installed. """ + +from __future__ import with_statement + # These things might need tuning 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 @@ -35,36 +37,16 @@ CHANNEL_TTL = (3 * 60 * 60) # Time to live, seconds from last transmit DISCONNECT_TTL = (24 * 60 * 60) # Time to live, seconds from last connect 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_FLOOD_DELAY = 1.0 # 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 = "2.4" -# 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 threading, Queue, SocketServer -import irc.client, logging +import sys, getopt, urlparse, time, random, socket, signal, re +import threading, Queue, SocketServer, select try: import simplejson as json # Faster, also makes us Python-2.4-compatible except ImportError: @@ -87,26 +69,302 @@ 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 -# 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 +# This code uses only NICK, JOIN, PART, MODE, PRIVMSG, USER, and QUIT. +# 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 # draft-brocklesby-irc-isupport-03 at . # The ",isnick" feature is as described in # . +# Historical note: the IRCClient and IRCServerConnection classes +# (~270LOC) replace the overweight, overcomplicated 3KLOC mass of +# irclib code that irker formerly used as a service library. They +# still look similar to parts of irclib because I contributed to that +# code before giving up on it. + +class IRCError(Exception): + "An IRC exception" + pass + +class IRCClient(): + "An IRC client session to one or more servers." + 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." + conn = IRCServerConnection(self) + with self.mutex: + self.server_connections.append(conn) + return conn + + def spin(self, timeout=0.2): + "Spin processing data from connections forever." + # Outer loop should specifically *not* be mutex-locked. + # 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] + sockets = [x.socket for x in connected] + if sockets: + connmap = dict([(c.socket.fileno(), c) for c in connected]) + (insocks, _o, _e) = select.select(sockets, [], [], timeout) + for s in insocks: + connmap[s.fileno()].consume() + else: + nextsleep = timeout + time.sleep(nextsleep) + + def add_event_handler(self, event, handler): + "Set a handler to be called later." + with self.mutex: + event_handlers = self.event_handlers.setdefault(event, []) + event_handlers.append(handler) + + def handle_event(self, connection, event): + with self.mutex: + h = self.event_handlers + th = sorted(h.get("all_events", []) + h.get(event.type, [])) + for handler in th: + handler(connection, event) + + def drop_connection(self, connection): + 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') + + def __init__(self): + self.buffer = '' + + def append(self, newbytes): + self.buffer += newbytes + + def lines(self): + "Iterate over lines in the buffer." + lines = LineBufferedStream.crlf_re.split(self.buffer) + self.buffer = lines.pop() + return iter(lines) + + def __iter__(self): + return self.lines() + +class IRCServerConnectionError(IRCError): + pass + +class IRCServerConnection(): + command_re = re.compile("^(:(?P[^ ]+) +)?(?P[^ ]+)( *(?P .+))?") + # The full list of numeric-to-event mappings is in Perl's Net::IRC. + # We only need to ensure that if some ancient server throws numerics + # for the ones we actually want to catch, they're mapped. + codemap = { + "001": "welcome", + "005": "featurelist", + "432": "erroneusnickname", + "433": "nicknameinuse", + "436": "nickcollision", + "437": "unavailresource", + } + + def __init__(self, master): + self.master = master + self.socket = None + + def connect(self, server, port, nickname, + password=None, username=None, ircname=None): + self.master.debug(2, "connect(server=%r, port=%r, nickname=%r, ...)" % + (server, port, nickname)) + if self.socket is not None: + self.disconnect("Changing servers") + + self.buffer = LineBufferedStream() + self.event_handlers = {} + self.real_server_name = "" + self.server = server + self.port = port + self.server_address = (server, port) + self.nickname = nickname + self.username = username or nickname + self.ircname = ircname or nickname + self.password = password + try: + self.socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM) + self.socket.bind(('', 0)) + self.socket.connect(self.server_address) + except socket.error as err: + raise IRCServerConnectionError("Couldn't connect to socket: %s" % err) + + if self.password: + self.ship("PASS " + self.password) + self.nick(self.nickname) + self.user(self.username, self.ircname) + return self + + def close(self): + # Without this thread lock, there is a window during which + # select() can find a closed socket, leading to an EBADF error. + with self.master.mutex: + self.disconnect("Closing object") + self.master.drop_connection(self) + + def consume(self): + try: + incoming = self.socket.recv(16384) + except socket.error: + # Server hung up on us. + self.disconnect("Connection reset by peer") + return + if not incoming: + # Dead air also indicates a connection reset. + self.disconnect("Connection reset by peer") + return + + self.buffer.append(incoming) + + for line in self.buffer: + self.master.debug(2, "FROM: %s" % line) + + if not line: + continue + + prefix = None + command = None + arguments = None + self.handle_event(Event("every_raw_message", + self.real_server_name, + None, + [line])) + + m = IRCServerConnection.command_re.match(line) + if m.group("prefix"): + prefix = m.group("prefix") + if not self.real_server_name: + self.real_server_name = prefix + if m.group("command"): + command = m.group("command").lower() + if m.group("argument"): + a = m.group("argument").split(" :", 1) + arguments = a[0].split() + if len(a) == 2: + arguments.append(a[1]) + + command = IRCServerConnection.codemap.get(command, command) + if command in ["privmsg", "notice"]: + target = arguments.pop(0) + else: + target = None + + if command == "quit": + arguments = [arguments[0]] + elif command == "ping": + target = arguments[0] + else: + target = arguments[0] + arguments = arguments[1:] + + 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): + self.master.handle_event(self, event) + if event.type in self.event_handlers: + for fn in self.event_handlers[event.type]: + fn(self, event) + + def is_connected(self): + return self.socket is not None + + def disconnect(self, message=""): + if self.socket is None: + return + # Don't send a QUIT here - causes infinite loop! + try: + self.socket.shutdown(socket.SHUT_WR) + self.socket.close() + except socket.error: + pass + del self.socket + self.socket = None + self.handle_event(Event("disconnect", self.server, "", [message])) + + def join(self, channel, key=""): + self.ship("JOIN %s%s" % (channel, (key and (" " + key)))) + + def mode(self, target, command): + self.ship("MODE %s %s" % (target, command)) + + def nick(self, newnick): + self.ship("NICK " + newnick) + + def part(self, channel, message=""): + cmd_parts = ['PART', channel] + if message: + cmd_parts.append(message) + self.ship(' '.join(cmd_parts)) + + def privmsg(self, target, text): + self.ship("PRIVMSG %s :%s" % (target, text)) + + def quit(self, message=""): + self.ship("QUIT" + (message and (" :" + message))) + + def user(self, username, realname): + self.ship("USER %s 0 * :%s" % (username, realname)) + + def ship(self, string): + "Ship a command to the server, appending CR/LF" + try: + 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.") + +class Event(object): + def __init__(self, evtype, source, target, arguments=None): + self.type = evtype + self.source = source + self.target = target + if arguments is None: + arguments = [] + self.arguments = arguments + +def is_channel(string): + return string and string[0] in "#&+!" + class Connection: def __init__(self, irkerd, servername, port): self.irker = irkerd @@ -126,26 +384,36 @@ 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() 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 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()) + "The server says our nick is ill-formed or has a conflict." + 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 + # 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 - self.status = "disconnected" + if self.status != "expired": + self.status = "disconnected" def handle_kick(self, outof): "We've been kicked." self.status = "handshaking" @@ -156,20 +424,22 @@ class Connection: % (self.servername, outof)) qcopy = [] while not self.queue.empty(): - (channel, message) = self.queue.get() + (channel, message, key) = self.queue.get() if channel != outof: - qcopy.append((channel, message)) - for (channel, message) in qcopy: - self.queue.put((channel, message)) + qcopy.append((channel, message, key)) + for (channel, message, key) in qcopy: + self.queue.put((channel, message, key)) self.status = "ready" - def enqueue(self, channel, message): + 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" self.thread = threading.Thread(target=self.dequeue) self.thread.setDaemon(True) self.thread.start() - self.queue.put((channel, message)) + 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: @@ -187,12 +457,15 @@ 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": - 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)) - with self.irker.library_lock: + 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.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") - self.connection.close() self.connection = None self.status = "disconnected" else: @@ -202,27 +475,35 @@ class Connection: # reflex arc it is highly unlikely any human will ever # notice. time.sleep(ANTI_BUZZ_DELAY) - elif not self.connection: + 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 and self.status != "expired": # Queue is nonempty but server isn't connected. - with self.irker.library_lock: - self.connection = self.irker.irc.server() + with self.irker.irc.mutex: + self.connection = self.irker.irc.newserver() 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 + # IRCServerConnectionError on failure 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.irker.irc.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" + self.last_ping = time.time() + except IRCServerConnectionError: + self.status = "expired" elif self.status == "handshaking": if time.time() > self.last_xmit + HANDSHAKE_TTL: self.status = "expired" @@ -231,13 +512,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 @@ -248,25 +522,53 @@ class Connection: self.status = "expired" break elif self.status == "ready": - with self.irker.library_lock: - (channel, message) = self.queue.get() - if channel not in self.channels_joined: - self.connection.join(channel) - self.irker.debug(1, "joining %s on %s." % (channel, self.servername)) + (channel, message, key) = self.queue.get() + if channel not in self.channels_joined: + self.connection.join(channel, key=key) + 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. + elif message: for segment in message.split("\n"): - self.connection.privmsg(channel, segment) + # 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.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.queue.task_done() + self.last_xmit = self.channels_joined[channel] = time.time() + 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() 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" @@ -278,6 +580,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) @@ -287,6 +592,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: @@ -298,11 +605,18 @@ 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] if self.channel and not isnick and self.channel[0] not in "#&+": self.channel = "#" + self.channel + # support both channel?secret and channel?key=secret + self.key = "" + if parsed.query: + self.key = re.sub("^key=", "", parsed.query) self.port = int(ircport) def valid(self): "Both components must be present for a valid target." @@ -318,7 +632,7 @@ class Dispatcher: self.servername = servername self.port = port self.connections = [] - def dispatch(self, channel, message): + 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. @@ -326,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) + eligibles[0].enqueue(channel, message, key, quit_after) return # All connections are full up. Look for one old enough to be # scavenged. @@ -341,55 +655,49 @@ 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) + 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) + 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]) + return max(x.last_xmit for x in self.connections) class Irker: "Persistent IRC multiplexer." def __init__(self, debuglevel=0): self.debuglevel = debuglevel - self.irc = irc.client.IRC() - self.irc.add_global_handler("ping", self._handle_ping) - self.irc.add_global_handler("welcome", self._handle_welcome) - self.irc.add_global_handler("erroneusnickname", self._handle_badnick) - self.irc.add_global_handler("nicknameinuse", self._handle_badnick) - self.irc.add_global_handler("nickcollision", self._handle_badnick) - self.irc.add_global_handler("unavailresource", self._handle_badnick) - 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.library_lock = threading.Lock() - thread = threading.Thread(target=self._process_forever) + 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) + self.irc.add_event_handler("nicknameinuse", self._handle_badnick) + self.irc.add_event_handler("nickcollision", self._handle_badnick) + self.irc.add_event_handler("unavailresource", self._handle_badnick) + self.irc.add_event_handler("featurelist", self._handle_features) + 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 _process_forever(self): - "IRC library process_forever with mutex." - self.debug(1, "process_forever()") - while True: - with self.library_lock: - self.irc.process_once() def _handle_ping(self, connection, _event): "PING arrived, bump the last-received time for the connection." if connection.context: @@ -406,14 +714,16 @@ 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 + 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 "#&+": 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(",") @@ -423,22 +733,32 @@ 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." - self.debug(1, "irker has been kicked from %s on %s" % (event.target(), connection.server)) + target = event.target + self.irc.debug(1, "irker has been kicked from %s on %s" % (target, connection.server)) if connection.context: - connection.context.handle_kick(event.target()) - def handle(self, line): + connection.context.handle_kick(target) + def _handle_every_raw_message(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 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()) @@ -465,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) + 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: @@ -485,13 +805,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) @@ -512,32 +827,71 @@ class IrkerUDPHandler(SocketServer.BaseRequestHandler): #socket = self.request[1] irker.handle(data) +def usage(): + sys.stdout.write(""" +Usage: + 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__': debuglvl = 0 - (options, arguments) = getopt.getopt(sys.argv[1:], "d:V") + immediate = None + namestyle = "irker%03d" + password = None + logfile = None + try: + (options, arguments) = getopt.getopt(sys.argv[1:], "d:i:l:n:p:Vh") + except getopt.GetoptError as e: + sys.stderr.write("%s" % e) + usage() + sys.exit(1) 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 + 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) + elif opt == '-h': + usage() + 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: - while True: - time.sleep(10) - 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