path: root/rtr-origin/rtr-origin.py

"""
Router origin-authentication rpki-router protocol implementation.  See
draft-ietf-sidr-rpki-rtr in fine Internet-Draft repositories near you.

Run the program with the --help argument for usage information, or see
documentation for the *_main() functions.


$Id$

Copyright (C) 2009-2010  Internet Systems Consortium ("ISC")

Permission to use, copy, modify, and distribute this software for any
purpose with or without fee is hereby granted, provided that the above
copyright notice and this permission notice appear in all copies.

THE SOFTWARE IS PROVIDED "AS IS" AND ISC DISCLAIMS ALL WARRANTIES WITH
REGARD TO THIS SOFTWARE INCLUDING ALL IMPLIED WARRANTIES OF MERCHANTABILITY
AND FITNESS.  IN NO EVENT SHALL ISC BE LIABLE FOR ANY SPECIAL, DIRECT,
INDIRECT, OR CONSEQUENTIAL DAMAGES OR ANY DAMAGES WHATSOEVER RESULTING FROM
LOSS OF USE, DATA OR PROFITS, WHETHER IN AN ACTION OF CONTRACT, NEGLIGENCE
OR OTHER TORTIOUS ACTION, ARISING OUT OF OR IN CONNECTION WITH THE USE OR
PERFORMANCE OF THIS SOFTWARE.
"""

import sys, os, struct, time, glob, socket, fcntl, signal, syslog
import asyncore, asynchat, subprocess, traceback, getopt
import rpki.x509, rpki.ipaddrs, rpki.sundial, rpki.async

class read_buffer(object):
  """
  Wrapper around synchronous/asynchronous read state.
  """

  def __init__(self):
    self.buffer = ""

  def update(self, need, callback):
    """
    Update count of needed bytes and callback, then dispatch to callback.
    """
    self.need = need
    self.callback = callback
    return self.callback(self)

  def available(self):
    """
    How much data do we have available in this buffer?
    """
    return len(self.buffer)

  def needed(self):
    """
    How much more data does this buffer need to become ready?
    """
    return self.need - self.available()

  def ready(self):
    """
    Is this buffer ready to read yet?
    """
    return self.available() >= self.need

  def get(self, n):
    """
    Hand some data to the caller.
    """
    b = self.buffer[:n]
    self.buffer = self.buffer[n:]
    return b

  def put(self, b):
    """
    Accumulate some data.
    """
    self.buffer += b

  def retry(self):
    """
    Try dispatching to the callback again.
    """
    return self.callback(self)

class pdu(object):
  """
  Object representing a generic PDU in the rpki-router protocol.
  Real PDUs are subclasses of this class.
  """

  version = 0                           # Protocol version

  _pdu = None                           # Cached when first generated

  header_struct = struct.Struct("!BBHL")

  def __cmp__(self, other):
    return cmp(self.to_pdu(), other.to_pdu())

  def check(self):
    """
    Check attributes to make sure they're within range.
    """
    pass

  @classmethod
  def read_pdu(cls, reader):
    return reader.update(need = cls.header_struct.size, callback = cls.got_header)

  @classmethod
  def got_header(cls, reader):
    if not reader.ready():
      return None
    assert reader.available() >= cls.header_struct.size
    version, pdu_type, whatever, length = cls.header_struct.unpack(reader.buffer[:cls.header_struct.size])
    assert version == cls.version, "PDU version is %d, expected %d" % (version, cls.version)
    assert length >= 8
    self = cls.pdu_map[pdu_type]()
    return reader.update(need = length, callback = self.got_pdu)

  def consume(self, client):
    """
    Handle results in test client.  Default behavior is just to print
    out the PDU.
    """
    log(self)

  def send_file(self, server, filename):
    """
    Send a content of a file as a cache response.  Caller should catch IOError.
    """
    f = open(filename, "rb")
    server.push_pdu(cache_response())
    server.push_file(f)
    server.push_pdu(end_of_data(serial = server.current_serial))

  def send_nodata(self, server):
    """
    Send a nodata error.
    """
    server.push_pdu(error_report(errno = error_report.codes["No Data Available"], errpdu = self))

class pdu_with_serial(pdu):
  """
  Base class for PDUs consisting of just a serial number.
  """

  header_struct = struct.Struct("!BBHLL")

  def __init__(self, serial = None):
    if serial is not None:
      if isinstance(serial, str):
        serial = int(serial)
      assert isinstance(serial, int)
      self.serial = serial

  def __str__(self):
    return "[%s, serial #%s]" % (self.__class__.__name__, self.serial)

  def to_pdu(self):
    """
    Generate the wire format PDU for this prefix.
    """
    if self._pdu is None:
      self._pdu = self.header_struct.pack(self.version, self.pdu_type, 0, self.header_struct.size, self.serial)
    return self._pdu

  def got_pdu(self, reader):
    if not reader.ready():
      return None
    b = reader.get(self.header_struct.size)
    version, pdu_type, zero, length, self.serial = self.header_struct.unpack(b)
    assert zero == 0
    assert length == 12
    assert b == self.to_pdu()
    return self

class pdu_empty(pdu):
  """
  Base class for empty PDUs.
  """

  header_struct = struct.Struct("!BBHL")

  def __str__(self):
    return "[%s]" % self.__class__.__name__

  def to_pdu(self):
    """
    Generate the wire format PDU for this prefix.
    """
    if self._pdu is None:
      self._pdu = self.header_struct.pack(self.version, self.pdu_type, 0, self.header_struct.size)
    return self._pdu

  def got_pdu(self, reader):
    if not reader.ready():
      return None
    b = reader.get(self.header_struct.size)
    version, pdu_type, zero, length = self.header_struct.unpack(b)
    assert zero == 0
    assert length == 8
    assert b == self.to_pdu()
    return self

class serial_notify(pdu_with_serial):
  """
  Serial Notify PDU.
  """

  pdu_type = 0

  def consume(self, client):
    """
    Respond to a serial_notify message with either a serial_query or
    reset_query, depending on what we already know.
    """
    log(self)
    if client.current_serial is None:
      client.push_pdu(reset_query())
    elif self.serial != client.current_serial:
      client.push_pdu(serial_query(serial = client.current_serial))
    else:
      log("[Notify did not change serial number, ignoring]")

class serial_query(pdu_with_serial):
  """
  Serial Query PDU.
  """

  pdu_type = 1

  def serve(self, server):
    """
    Received a serial query, send incremental transfer in response.
    If client is already up to date, just send an empty incremental
    transfer.
    """
    log(self)
    if server.get_serial() is None:
      self.send_nodata(server)
    elif int(server.current_serial) == self.serial:
      log("[Client is already current, sending empty IXFR]")
      server.push_pdu(cache_response())
      server.push_pdu(end_of_data(serial = server.current_serial))
    else:
      try:
        self.send_file(server, "%s.ix.%s" % (server.current_serial, self.serial))
      except IOError:
        server.push_pdu(cache_reset())

class reset_query(pdu_empty):
  """
  Reset Query PDU.
  """

  pdu_type = 2

  def serve(self, server):
    """
    Received a reset query, send full current state in response.
    """
    log(self)
    if server.get_serial() is None:
      self.send_nodata(server)
    else:
      try:
        fn = "%s.ax" % server.current_serial
        self.send_file(server, fn)
      except IOError:
        server.push_pdu(error_report(errno = error_report.codes["Internal Error"], errpdu = self, errmsg = "Couldn't open %s" % fn))

class cache_response(pdu_empty):
  """
  Incremental Response PDU.
  """

  pdu_type = 3

class end_of_data(pdu_with_serial):
  """
  End of Data PDU.
  """

  pdu_type = 7

  def consume(self, client):
    """
    Handle end_of_data response.
    """
    log(self)
    client.current_serial = self.serial

class cache_reset(pdu_empty):
  """
  Cache reset PDU.
  """

  pdu_type = 8

  def consume(self, client):
    """
    Handle cache_reset response, by issuing a reset_query.
    """
    log(self)
    client.push_pdu(reset_query())

class prefix(pdu):
  """
  Object representing one prefix.  This corresponds closely to one PDU
  in the rpki-router protocol, so closely that we use lexical ordering
  of the wire format of the PDU as the ordering for this class.
  """

  header_struct = struct.Struct("!BB2xLBBBx")
  asnum_struct = struct.Struct("!L")

  @classmethod
  def from_asn1(cls, asnum, t):
    """
    Read a prefix from a ROA in the tuple format used by our ASN.1
    decoder.
    """
    assert len(t[0]) <= cls.addr_type.bits
    x = 0L
    for y in t[0]:
      x = (x << 1) | y
    x <<= (cls.addr_type.bits - len(t[0]))
    self = cls()
    self.asn = asnum
    self.prefix = cls.addr_type(x)
    self.prefixlen = len(t[0])
    self.max_prefixlen = self.prefixlen if t[1] is None else t[1]
    self.announce = 1
    self.check()
    return self

  def __str__(self):
    plm = "%s/%s-%s" % (self.prefix, self.prefixlen, self.max_prefixlen)
    return "%s %8s  %-32s %s" % ("+" if self.announce else "-", self.asn, plm, ":".join(("%02X" % ord(b) for b in self.to_pdu())))

  def show(self):
    print "# Class:       ", self.__class__.__name__
    print "# ASN:         ", self.asn
    print "# Prefix:      ", self.prefix
    print "# Prefixlen:   ", self.prefixlen
    print "# MaxPrefixlen:", self.max_prefixlen
    print "# Announce:    ", self.announce

  def check(self):
    """
    Check attributes to make sure they're within range.
    """
    assert self.announce in (0, 1)
    assert self.prefixlen >= 0 and self.prefixlen <= self.addr_type.bits
    assert self.max_prefixlen >= self.prefixlen and self.max_prefixlen <= self.addr_type.bits
    pdulen = self.header_struct.size + self.addr_type.bits / 8 + self.asnum_struct.size
    assert len(self.to_pdu()) == pdulen, "Expected %d byte PDU, got %d" % pd(pdulen, len(self.to_pdu()))

  def to_pdu(self, announce = None):
    """
    Generate the wire format PDU for this prefix.
    """
    if announce is not None:
      assert announce in (0, 1)
    elif self._pdu is not None:
      return self._pdu
    pdulen = self.header_struct.size + self.addr_type.bits / 8 + self.asnum_struct.size
    pdu = (self.header_struct.pack(self.version, self.pdu_type, pdulen,
                                   announce if announce is not None else self.announce,
                                   self.prefixlen, self.max_prefixlen) +
           self.prefix.to_bytes() +
           self.asnum_struct.pack(self.asn))
    if announce is None:
      assert self._pdu is None
      self._pdu = pdu
    return pdu

  def got_pdu(self, reader):
    if not reader.ready():
      return None
    b1 = reader.get(self.header_struct.size)
    b2 = reader.get(self.addr_type.bits / 8)
    b3 = reader.get(self.asnum_struct.size)
    version, pdu_type, length, self.announce, self.prefixlen, self.max_prefixlen = self.header_struct.unpack(b1)
    assert length == len(b1) + len(b2) + len(b3)
    self.prefix = self.addr_type.from_bytes(b2)
    self.asn = self.asnum_struct.unpack(b3)[0]
    assert b1 + b2 + b3 == self.to_pdu()
    return self

class ipv4_prefix(prefix):
  """
  IPv4 flavor of a prefix.
  """
  pdu_type = 4
  addr_type = rpki.ipaddrs.v4addr

class ipv6_prefix(prefix):
  """
  IPv6 flavor of a prefix.
  """
  pdu_type = 6
  addr_type = rpki.ipaddrs.v6addr

class error_report(pdu):
  """
  Error Report PDU.
  """

  pdu_type = 10

  header_struct = struct.Struct("!BBHL")
  string_struct = struct.Struct("!L")

  msgs = {
    1 : "Internal Error",
    2 : "No Data Available" }

  codes = dict((v, k) for k, v in msgs.items())

  def __init__(self, errno = None, errpdu = None, errmsg = None):
    assert errno is None or errno in self.msgs
    self.errno = errno
    self.errpdu = errpdu
    self.errmsg = errmsg if errmsg is not None or errno is None else self.msgs[errno]

  def __str__(self):
    return "Error #%s: %s" % (self.errno, self.errmsg)

  def to_counted_string(self, s):
    return self.string_struct.pack(len(s)) + s

  def read_counted_string(self, reader, remaining):
    assert remaining >= self.string_struct.size
    n = self.string_struct.unpack(reader.get(self.string_struct.size))[0]
    assert remaining >= self.string_struct.size + n
    return n, reader.get(n), (remaining - self.string_struct.size - n)

  def to_pdu(self):
    """
    Generate the wire format PDU for this prefix.
    """
    if self._pdu is None:
      assert isinstance(self.errno, int)
      assert not isinstance(self.errpdu, error_report)
      p = self.errpdu
      if p is None:
        p = ""
      elif isinstance(p, pdu):
        p = p.to_pdu()
      assert isinstance(p, str)
      pdulen = self.header_struct.size + self.string_struct.size * 2 + len(p) + len(self.errmsg)
      self._pdu = self.header_struct.pack(self.version, self.pdu_type, self.errno, pdulen)
      self._pdu += self.to_counted_string(p)
      self._pdu += self.to_counted_string(self.errmsg.encode("utf8"))
    return self._pdu

  def got_pdu(self, reader):
    if not reader.ready():
      return None
    header = reader.get(self.header_struct.size)
    version, pdu_type, self.errno, length = self.header_struct.unpack(header)
    remaining = length - self.header_struct.size
    self.pdulen, self.errpdu, remaining = self.read_counted_string(reader, remaining)
    self.errlen, self.errmsg, remaining = self.read_counted_string(reader, remaining)
    assert length == self.header_struct.size + self.string_struct.size * 2 + self.pdulen + self.errlen
    assert header + self.to_counted_string(self.errpdu) + self.to_counted_string(self.errmsg.encode("utf8")) == self.to_pdu()
    return self

prefix.afi_map = { "\x00\x01" : ipv4_prefix, "\x00\x02" : ipv6_prefix }

pdu.pdu_map = dict((p.pdu_type, p) for p in (ipv4_prefix, ipv6_prefix, serial_notify, serial_query, reset_query,
                                             cache_response, end_of_data, cache_reset, error_report))

class prefix_set(list):
  """
  Object representing a set of prefixes, that is, one versioned and
  (theoretically) consistant set of prefixes extracted from rcynic's
  output.
  """

  @classmethod
  def _load_file(cls, filename):
    """
    Low-level method to read prefix_set from a file.
    """
    self = cls()
    f = open(filename, "rb")
    r = read_buffer()
    while True:
      p = pdu.read_pdu(r)
      while p is None:
        b = f.read(r.needed())
        if b == "":
          assert r.available() == 0
          return self
        r.put(b)
        p = r.retry()
      self.append(p)

class axfr_set(prefix_set):
  """
  Object representing a complete set of prefixes, that is, one
  versioned and (theoretically) consistant set of prefixes extracted
  from rcynic's output, all with the announce field set.
  """

  @classmethod
  def parse_rcynic(cls, rcynic_dir):
    """
    Parse ROAS fetched (and validated!) by rcynic to create a new
    axfr_set.
    """
    self = cls()
    self.serial = rpki.sundial.now().totimestamp()
    for root, dirs, files in os.walk(rcynic_dir):
      for f in files:
        if f.endswith(".roa"):
          f = os.path.join(root, f)
          try:
            roa = rpki.x509.ROA(DER_file = f).extract().get()
          except:
            print "Could not parse purported ROA file %r" % f
            continue
          if roa[0] != 0:
            print "ROA %r version is %d, expected version 0" % (f, roa[0])
            continue
          asnum = roa[1]
          for afi, addrs in roa[2]:
            for addr in addrs:
              self.append(prefix.afi_map[afi].from_asn1(asnum, addr))
    self.sort()
    for i in xrange(len(self) - 2, -1, -1):
      if self[i] == self[i + 1]:
        del self[i + 1]
    return self

  @classmethod
  def load(cls, filename):
    """
    Load an axfr_set from a file, parse filename to obtain serial.
    """
    fn1, fn2 = os.path.basename(filename).split(".")
    assert fn1.isdigit() and fn2 == "ax"
    self = cls._load_file(filename)
    self.serial = int(fn1)
    return self

  def filename(self):
    """
    Generate filename for this axfr_set.
    """
    return "%d.ax" % self.serial

  def save_axfr(self):
    """
    Write axfr__set to file with magic filename.
    """
    f = open(self.filename(), "wb")
    for p in self:
      f.write(p.to_pdu())
    f.close()

  def mark_current(self):
    """
    Mark the current serial number as current.
    """
    tmpfn = "current.%d.tmp" % os.getpid()
    try:
      f = open(tmpfn, "w")
      f.write("%d\n" % self.serial)
      f.close()
      os.rename(tmpfn, "current")
    except:
      os.unlink(tmpfn)
      raise

  def save_ixfr(self, other):
    """
    Comparing this axfr_set with an older one and write the resulting
    ixfr_set to file with magic filename.  Since we store prefix_sets
    in sorted order, computing the difference is a trivial linear
    comparison.
    """
    f = open("%d.ix.%d" % (self.serial, other.serial), "wb")
    old = other[:]
    new = self[:]
    while old and new:
      if old[0] < new[0]:
        f.write(old.pop(0).to_pdu(announce = 0))
      elif old[0] > new[0]:
        f.write(new.pop(0).to_pdu(announce = 1))
      else:
        del old[0]
        del new[0]
    while old:
      f.write(old.pop(0).to_pdu(announce = 0))
    while new:
      f.write(new.pop(0).to_pdu(announce = 1))
    f.close()

  def show(self):
    """
    Print this axfr_set.
    """
    print "# AXFR %d (%s)" % (self.serial, rpki.sundial.datetime.utcfromtimestamp(self.serial))
    for p in self:
      print p

class ixfr_set(prefix_set):
  """
  Object representing an incremental set of prefixes, that is, the
  differences between one versioned and (theoretically) consistant set
  of prefixes extracted from rcynic's output and another, with the
  announce fields set or cleared as necessary to indicate the changes.
  """

  @classmethod
  def load(cls, filename):
    """
    Load an ixfr_set from a file, parse filename to obtain serials.
    """
    fn1, fn2, fn3 = os.path.basename(filename).split(".")
    assert fn1.isdigit() and fn2 == "ix" and fn3.isdigit()
    self = cls._load_file(filename)
    self.from_serial = int(fn3)
    self.to_serial = int(fn1)
    return self

  def filename(self):
    """
    Generate filename for this ixfr_set.
    """
    return "%d.ix.%d" % (self.to_serial, self.from_serial)

  def show(self):
    """
    Print this ixfr_set.
    """
    print "# IXFR %d (%s) -> %d (%s)" % (self.from_serial, rpki.sundial.datetime.utcfromtimestamp(self.from_serial),
                                         self.to_serial, rpki.sundial.datetime.utcfromtimestamp(self.to_serial))
    for p in self:
      print p

class file_producer(object):
  """
  File-based producer object for asynchat.
  """

  def __init__(self, handle, buffersize):
    self.handle = handle
    self.buffersize = buffersize

  def more(self):
    return self.handle.read(self.buffersize)

class pdu_channel(asynchat.async_chat):
  """
  asynchat subclass that understands our PDUs.  This just handles
  network I/O.  Specific engines (client, server) should be subclasses
  of this with methods that do something useful with the resulting
  PDUs.
  """

  def __init__(self, conn = None):
    asynchat.async_chat.__init__(self, conn)
    self.reader = read_buffer()

  def start_new_pdu(self):
    """
    Start read of a new PDU.
    """
    p = pdu.read_pdu(self.reader)
    while p is not None:
      self.deliver_pdu(p)
      p = pdu.read_pdu(self.reader)
    assert not self.reader.ready()
    self.set_terminator(self.reader.needed())

  def collect_incoming_data(self, data):
    """
    Collect data into the read buffer.
    """
    self.reader.put(data)
    
  def found_terminator(self):
    """
    Got requested data, see if we now have a PDU.  If so, pass it
    along, then restart cycle for a new PDU.
    """
    p = self.reader.retry()
    if p is None:
      self.set_terminator(self.reader.needed())
    else:
      self.deliver_pdu(p)
      self.start_new_pdu()

  def push_pdu(self, pdu):
    """
    Write PDU to stream.
    """
    self.push(pdu.to_pdu())

  def push_file(self, f):
    """
    Write content of a file to stream.
    """
    self.push_with_producer(file_producer(f, self.ac_out_buffer_size))

  def log(self, msg):
    """
    Intercept asyncore's logging.
    """
    log(msg)

  def log_info(self, msg, tag = "info"):
    """
    Intercept asynchat's logging.
    """
    log("asynchat: %s: %s" % (tag, msg))

  def handle_error(self):
    """
    Handle errors caught by asyncore main loop.
    """
    for line in traceback.format_exc().splitlines():
      log(line)
    log("Exiting after unhandled exception")
    asyncore.close_all()

  def init_file_dispatcher(self, fd):
    """
    Kludge to plug asyncore.file_dispatcher into asynchat.  Call from
    subclass's __init__() method, after calling
    pdu_channel.__init__(), and don't read this on a full stomach.
    """
    self.connected = True
    self._fileno = fd
    self.socket = asyncore.file_wrapper(fd)
    self.add_channel()
    flags = fcntl.fcntl(fd, fcntl.F_GETFL, 0)
    flags = flags | os.O_NONBLOCK
    fcntl.fcntl(fd, fcntl.F_SETFL, flags)

class server_write_channel(pdu_channel):
  """
  Kludge to deal with ssh's habit of sometimes (compile time option)
  invoking us with two unidirectional pipes instead of one
  bidirectional socketpair.  All the server logic is in the
  server_channel class, this class just deals with sending the
  server's output to a different file descriptor.
  """

  def __init__(self):
    """
    Set up stdout.
    """
    pdu_channel.__init__(self)
    self.init_file_dispatcher(sys.stdout.fileno())

  def readable(self):
    """
    This channel is never readable.
    """
    return False

class server_channel(pdu_channel):
  """
  Server protocol engine, handles upcalls from pdu_channel to
  implement protocol logic.
  """

  def __init__(self):
    """
    Set up stdin and stdout as connection and start listening for
    first PDU.
    """
    pdu_channel.__init__(self)
    self.init_file_dispatcher(sys.stdin.fileno())
    self.writer = server_write_channel()
    self.get_serial()
    self.start_new_pdu()

  def writable(self):
    """
    This channel is never writable.
    """
    return False

  def push(self, data):
    """
    Redirect to writer channel.
    """
    return self.writer.push(data)

  def push_with_producer(self, producer):
    """
    Redirect to writer channel.
    """
    return self.writer.push_with_producer(producer)

  def push_pdu(self, pdu):
    """
    Redirect to writer channel.
    """
    return self.writer.push_pdu(pdu)

  def push_file(self, f):
    """
    Redirect to writer channel.
    """
    return self.writer.push_file(f)

  def deliver_pdu(self, pdu):
    """
    Handle received PDU.
    """
    pdu.serve(self)

  def handle_close(self):
    """
    Intercept close event so we can shut down other sockets.
    """
    asynchat.async_chat.handle_close(self)
    asyncore.close_all()

  def get_serial(self):
    """
    Read, cache, and return current serial number, or None if we can't
    find the serial number file.  The latter condition should never
    happen, but maybe we got started in server mode while the cronjob
    mode instance is still building its database.
    """
    try:
      f = open("current", "r")
      self.current_serial = f.read().strip()
      assert self.current_serial.isdigit()
      f.close()
    except IOError:
      self.current_serial = None
    return self.current_serial

  def check_serial(self):
    """
    Check for a new serial number.
    """
    old_serial = self.current_serial
    return old_serial != self.get_serial()

  def notify(self, data = None):
    """
    Cronjob instance kicked us, send a notify message.
    """
    if self.check_serial():
      self.push_pdu(serial_notify(serial = self.current_serial))
    else:
      log("Cronjob kicked me without a valid current serial number")

class client_channel(pdu_channel):
  """
  Client protocol engine, handles upcalls from pdu_channel.
  """

  current_serial = None

  timer = None

  def __init__(self, sock, proc, killsig):
    self.killsig = killsig
    self.proc = proc
    pdu_channel.__init__(self, conn = sock)
    self.start_new_pdu()

  @classmethod
  def ssh(cls, host, port):
    """
    Set up ssh connection and start listening for first PDU.
    """
    args = ("ssh", "-p", port, "-s", host, "rpki-rtr")
    log("[Running ssh: %s]" % " ".join(args))
    s = socket.socketpair()
    return cls(sock = s[1],
               proc = subprocess.Popen(args, executable = "/usr/bin/ssh", stdin = s[0], stdout = s[0], close_fds = True),
               killsig = signal.SIGKILL)

  @classmethod
  def tcp(cls, host, port):
    """
    Set up TCP connection and start listening for first PDU.
    """
    log("[Starting raw TCP connection to %s:%s]" % (host, port))
    s = socket.socket()
    s.connect((host, int(port)))
    return cls(sock = s, proc = None, killsig = None)

  @classmethod
  def loopback(cls):
    """
    Set up loopback connection and start listening for first PDU.
    """
    s = socket.socketpair()
    log("[Using direct subprocess kludge for testing]")
    return cls(sock = s[1],
               proc = subprocess.Popen(("/usr/local/bin/python", "rtr-origin.py", "--server"), stdin = s[0], stdout = s[0], close_fds = True),
               killsig = signal.SIGINT)

  def deliver_pdu(self, pdu):
    """
    Handle received PDU.
    """
    pdu.consume(self)

  def push_pdu(self, pdu):
    """
    Log outbound PDU then write it to stream.
    """
    log(pdu)
    pdu_channel.push_pdu(self, pdu)

  def cleanup(self):
    """
    Force clean up this client's child process.  If everything goes
    well, child will have exited already before this method is called,
    but we may need to whack it with a stick if something breaks.
    """
    if self.timer is not None:
      self.timer.cancel()
    if self.proc is not None and self.proc.returncode is None:
      try:
        os.kill(self.proc.pid, self.killsig)
      except OSError:
        pass

  def handle_close(self):
    """
    Intercept close event so we can log it, then shut down.
    """
    log("Server closed channel")
    rpki.async.exit_event_loop()

class kickme_channel(asyncore.dispatcher):
  """
  asyncore dispatcher for the PF_UNIX socket that cronjob mode uses to
  kick servers when it's time to send notify PDUs to clients.
  """

  def __init__(self, server):
    asyncore.dispatcher.__init__(self)
    self.server = server
    self.sockname = "%s.%d" % (kickme_base, os.getpid())
    self.create_socket(socket.AF_UNIX, socket.SOCK_DGRAM)
    try:
      self.bind(self.sockname)
      os.chmod(self.sockname, 0660)
    except socket.error, e:
      log("Couldn't bind() kickme socket: %r" % e)
      self.close()
    except OSError, e:
      log("Couldn't chmod() kickme socket: %r" % e)

  def writable(self):
    """
    This socket is read-only, never writable.
    """
    return False

  def handle_connect(self):
    """
    Ignore connect events (not very useful on datagram socket).
    """
    pass

  def handle_read(self):
    """
    Handle receipt of a datagram.
    """
    data = self.recv(512)
    self.server.notify(data)

  def cleanup(self):
    """
    Clean up this dispatcher's socket.
    """
    self.close()
    try:
      os.unlink(self.sockname)
    except:
      pass

  def log(self, msg):
    """
    Intercept asyncore's logging.
    """
    log(msg)

  def log_info(self, msg, tag = "info"):
    """
    Intercept asyncore's logging.
    """
    log("asyncore: %s: %s" % (tag, msg))

  def handle_error(self):
    """
    Handle errors caught by asyncore main loop.
    """
    for line in traceback.format_exc().splitlines():
      log(line)
    log("Exiting after unhandled exception")
    asyncore.close_all()

def cronjob_main(argv):
  """
  Run this mode right after rcynic to do the real work of groveling
  through the ROAs that rcynic collects and translating that data into
  the form used in the rpki-router protocol.  This mode prepares both
  full dumps (AXFR) and incremental dumps against a specific prior
  version (IXFR).  [Terminology here borrowed from DNS, as is much of
  the protocol design.]  Finally, this mode kicks any active servers,
  so that they can notify their clients that a new version is
  available.

  Run this in the directory where you want to write its output files,
  which should also be the directory in which you run this program in
  --server mode.

  This mode takes one argument on the command line, which specifies
  the directory name of rcynic's authenticated output tree (normally
  $somewhere/rcynic-data/authenticated/).
  """

  if len(argv) != 1:
    usage("Expected one argument, got %r" % (argv,))

  old_ixfrs = glob.glob("*.ix.*")

  cutoff = rpki.sundial.now() - rpki.sundial.timedelta(days = 1)
  for f in glob.iglob("*.ax"):
    t = rpki.sundial.datetime.utcfromtimestamp(os.stat(f).st_mtime)
    if  t < cutoff:
      print "# Deleting old file %s, timestamp %s" % (f, t)
      os.unlink(f)
  
  pdus = axfr_set.parse_rcynic(argv[0])
  pdus.save_axfr()
  for axfr in glob.iglob("*.ax"):
    if axfr != pdus.filename():
      pdus.save_ixfr(axfr_set.load(axfr))
  pdus.mark_current()

  print "# New serial is %s" % pdus.serial

  try:
    os.stat(kickme_dir)
  except OSError:
    print '# Creating directory "%s"' % kickme_dir
    os.makedirs(kickme_dir)

  msg = "Good morning, serial %s is ready" % pdus.serial
  sock = socket.socket(socket.AF_UNIX, socket.SOCK_DGRAM)
  for name in glob.iglob("%s.*" % kickme_base):
    try:
      print "# Kicking %s" % name
      sock.sendto(msg, name)
    except:
      print "# Failed to kick %s" % name
  sock.close()

  old_ixfrs.sort()
  for ixfr in old_ixfrs:
    print "# Deleting old file %s" % ixfr
    os.unlink(ixfr)

def show_main(argv):
  """
  Display dumps created by --cronjob mode in textual form.
  Intended only for debugging.

  This mode takes no command line arguments.  Run it in the directory
  where you ran --cronjob mode.
  """

  if argv:
    usage("Unexpected arguments: %r" % (argv,))

  g = glob.glob("*.ax")
  g.sort()
  for f in g:
    axfr_set.load(f).show()

  g = glob.glob("*.ix.*")
  g.sort()
  for f in g:
    ixfr_set.load(f).show()

def server_main(argv):
  """
  Implement the server side of the rpkk-router protocol.  Other than
  one PF_UNIX socket inode, this doesn't write anything to disk, so it
  can be run with minimal privileges.  Most of the hard work has
  already been done in --cronjob mode, so all that this mode has to do
  is serve up the results.

  In production use this server is run under sshd.  The subsystem
  mechanism in sshd does not allow us to pass arguments on the command
  line, so either we need a wrapper or we need wired-in names.  sshd
  will have us running in whatever it thinks is our home directory on
  startup, so it may be that the easiest approach here is to let sshd
  put us in the right directory and just look for our files there.

  This mode takes no arguments.  Run it in the directory where you ran
  --cronjob mode.

  The server is event driven, so everything interesting happens in the
  channel classes.
  """

  log("[Starting]")
  if argv:
    usage("Unexpected arguments: %r" % (argv,))
  kickme = None
  try:
    server = server_channel()
    kickme = kickme_channel(server = server)
    rpki.async.event_loop()
  finally:
    if kickme is not None:
      kickme.cleanup()

class client_timer(rpki.async.timer):
  """
  Timer class for client mode, to handle the periodic serial queries.
  """

  def __init__(self, client, period):
    rpki.async.timer.__init__(self)
    self.client = client
    self.period = period
    self.set(period)

  def handler(self):
    if self.client.current_serial is None:
      self.client.push_pdu(reset_query())
    else:
      self.client.push_pdu(serial_query(serial = self.client.current_serial))
    self.set(self.period)

def client_main(argv):
  """
  Toy client, intended only for debugging.

  This program takes one or more arguments.  The first argument
  determines what kind of connection it should open to the server, the
  remaining arguments are connection details specific to this
  particular type of connection.

  If the first argument is "loopback", the client will run a copy of
  the server directly in a subprocess, and communicate with it via a
  PF_UNIX socket pair.  This sub-mode takes no further arguments.

  If the first argument is "ssh", the client will attempt to run ssh
  in as subprocess to connect to the server using the ssh subsystem
  mechanism as specified for this protocol.  The remaining arguments
  should be a hostname (or IP address in a form acceptable to ssh) and
  a TCP port number.

  If the first argument is "tcp", the client will attempt to open a
  direct (and completely insecure!) TCP connection to the server.
  The remaining arguments should be a hostname (or IP address) and
  a TCP port number.
  """

  log("[Startup]")
  client = None
  try:
    if not argv or (argv[0] == "loopback" and len(argv) == 1):
      client = client_channel.loopback()
    elif argv[0] == "ssh" and len(argv) == 3:
      client = client_channel.ssh(*argv[1:])
    elif argv[0] == "tcp" and len(argv) == 3:
      client = client_channel.tcp(*argv[1:])
    else:
      usage("Unexpected arguments: %r" % (argv,))
    client.push_pdu(reset_query())
    client.timer = client_timer(client, rpki.sundial.timedelta(minutes = 10))
    rpki.async.event_loop()
  finally:
    if client is not None:
      client.cleanup()

def log(msg):
  rpki.log.warn(str(msg))

os.environ["TZ"] = "UTC"
time.tzset()

mode = None

kickme_dir  = "sockets"
kickme_base = os.path.join(kickme_dir, "kickme")

main_dispatch = {
  "cronjob" : cronjob_main,
  "client"  : client_main,
  "server"  : server_main,
  "show"    : show_main }

def usage(error = None):
  print "Usage: %s --mode [arguments]" % sys.argv[0]
  print
  print "where --mode is one of:"
  print
  for name, func in main_dispatch.iteritems():
    print "--%s:" % name
    print func.__doc__
  if isinstance(error, str):
    sys.exit("Error: %s" % error)
  else:
    sys.exit(error)

opts, argv = getopt.getopt(sys.argv[1:], "h?", ["help"] + main_dispatch.keys())
for o, a in opts:
  if o in ("-h", "--help", "-?"):
    usage()
  if len(o) > 2 and o[2:] in main_dispatch:
    if mode is not None:
      usage("Conflicting modes specified")
    mode = o[2:]

if mode is None:
  usage("No mode specified")

tag = mode

rpki.log.use_syslog = mode in ("cronjob", "server")

if mode == "server":
  #
  # Try to figure out peer address when we're in server mode.
  try:
    tag += "/tcp/" + str(socket.fromfd(0, socket.AF_INET, socket.SOCK_STREAM).getpeername()[0])
  except (socket.error, IndexError):
    if os.getenv("SSH_CONNECTION"):
      tag += "/ssh/" + os.getenv("SSH_CONNECTION").split()[0]

rpki.log.init("rtr-origin/" + tag, syslog.LOG_PID)

main_dispatch[mode](argv)