Salvage bgpdump-driven test code from rtr-origin, as separate module

rpki.rtr.bgpdump.  The normal rpki-rtr program intentionally does not
even import this module, because all of the bgpdump-* code is just a
confusing distraction to users who want the normal rpki-rtr code.

The refactored bgpdump-* code has passed pylint but otherwise has not
yet been tested, and may not work at all.  Since, as far as we know,
nobody but the author has ever run any of the code in this module,
this should not be a problem.

svn path=/trunk/; revision=5824

2 files changed, 339 insertions, 0 deletions

diff --git a/rpki/rtr/bgpdump.py b/rpki/rtr/bgpdump.py
new file mode 100755
index 00000000..fc3ae9df
--- /dev/null
+++ b/rpki/rtr/bgpdump.py
@@ -0,0 +1,330 @@
+# $Id$
+#
+# Copyright (C) 2014  Dragon Research Labs ("DRL")
+# Portions copyright (C) 2009-2013  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 notices and this permission notice appear in all copies.
+#
+# THE SOFTWARE IS PROVIDED "AS IS" AND DRL AND ISC DISCLAIM ALL
+# WARRANTIES WITH REGARD TO THIS SOFTWARE INCLUDING ALL IMPLIED
+# WARRANTIES OF MERCHANTABILITY AND FITNESS.  IN NO EVENT SHALL DRL OR
+# 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.
+
+"""
+rpki-rtr simulation code using bgpdump as an input source.  Test
+purposes only, not included in the normal rpki-rtr program.
+"""
+
+import sys
+import os
+import time
+import glob
+import logging
+import asyncore
+import subprocess
+import bisect
+import rpki.POW
+import rpki.oids
+import rpki.rtr.pdus
+import rpki.rtr.server
+import rpki.rtr.generator
+
+from rpki.rtr.channels import Timestamp
+
+
+class IgnoreThisRecord(Exception):
+  pass
+
+
+class PrefixPDU(rpki.rtr.generator.PrefixPDU):
+
+  @staticmethod
+  def from_bgpdump(line, rib_dump):
+    try:
+      assert isinstance(rib_dump, bool)
+      fields = line.split("|")
+
+      # Parse prefix, including figuring out IP protocol version
+      cls = rpki.rtr.generator.IPv6PrefixPDU if ":" in fields[5] else rpki.rtr.generator.IPv4PrefixPDU
+      self = cls()
+      self.timestamp = Timestamp(fields[1])
+      p, l = fields[5].split("/")
+      self.prefix = rpki.POW.IPAddress(p)
+      self.prefixlen = self.max_prefixlen = int(l)
+
+      # Withdrawals don't have AS paths, so be careful
+      assert fields[2] == "B" if rib_dump else fields[2] in ("A", "W")
+      if fields[2] == "W":
+        self.asn = 0
+        self.announce = 0
+      else:
+        self.announce = 1
+        if not fields[6] or "{" in fields[6] or "(" in fields[6]:
+          raise IgnoreThisRecord
+        a  = fields[6].split()[-1]
+        if "." in a:
+          a = [int(s) for s in a.split(".")]
+          if len(a) != 2 or a[0] < 0 or a[0] > 65535 or a[1] < 0 or a[1] > 65535:
+            logging.warn("Bad dotted ASNum %r, ignoring record", fields[6])
+            raise IgnoreThisRecord
+          a = (a[0] << 16) | a[1]
+        else:
+          a = int(a)
+        self.asn = a
+
+      self.check()
+      return self
+
+    except IgnoreThisRecord:
+      raise
+
+    except Exception, e:
+      logging.warn("Ignoring line %r: %s", line, e)
+      raise IgnoreThisRecord
+
+
+class AXFRSet(rpki.rtr.generator.AXFRSet):
+
+  @staticmethod
+  def read_bgpdump(filename):
+    assert filename.endswith(".bz2")
+    logging.debug("Reading %s", filename)
+    bunzip2 = subprocess.Popen(("bzip2", "-c", "-d", filename), stdout = subprocess.PIPE)
+    bgpdump = subprocess.Popen(("bgpdump", "-m", "-"), stdin = bunzip2.stdout, stdout = subprocess.PIPE)
+    return bgpdump.stdout
+
+  @classmethod
+  def parse_bgpdump_rib_dump(cls, filename):
+    assert os.path.basename(filename).startswith("ribs.")
+    self = cls()
+    self.serial = None
+    for line in cls.read_bgpdump(filename):
+      try:
+        pfx = PrefixPDU.from_bgpdump(line, rib_dump = True)
+      except IgnoreThisRecord:
+        continue
+      self.append(pfx)
+      self.serial = pfx.timestamp
+    if self.serial is None:
+      sys.exit("Failed to parse anything useful from %s" % filename)
+    self.sort()
+    for i in xrange(len(self) - 2, -1, -1):
+      if self[i] == self[i + 1]:
+        del self[i + 1]
+    return self
+
+  def parse_bgpdump_update(self, filename):
+    assert os.path.basename(filename).startswith("updates.")
+    for line in self.read_bgpdump(filename):
+      try:
+        pfx = PrefixPDU.from_bgpdump(line, rib_dump = False)
+      except IgnoreThisRecord:
+        continue
+      announce = pfx.announce
+      pfx.announce = 1
+      i = bisect.bisect_left(self, pfx)
+      if announce:
+        if i >= len(self) or pfx != self[i]:
+          self.insert(i, pfx)
+      else:
+        while i < len(self) and pfx.prefix == self[i].prefix and pfx.prefixlen == self[i].prefixlen:
+          del self[i]
+      self.serial = pfx.timestamp
+
+
+def bgpdump_convert_main(args):
+  """
+  * DANGER WILL ROBINSON! * DEBUGGING AND TEST USE ONLY! *
+  Simulate route origin data from a set of BGP dump files.
+  argv is an ordered list of filenames.  Each file must be a BGP RIB
+  dumps, a BGP UPDATE dumps, or an AXFR dump in the format written by
+  this program's --cronjob command.  The first file must be a RIB dump
+  or AXFR dump, it cannot be an UPDATE dump.  Output will be a set of
+  AXFR and IXFR files with timestamps derived from the BGP dumps,
+  which can be used as input to this program's --server command for
+  test purposes.  SUCH DATA PROVIDE NO SECURITY AT ALL.
+  * DANGER WILL ROBINSON! * DEBUGGING AND TEST USE ONLY! *
+  """
+
+  first = True
+  db = None
+  axfrs = []
+  version = max(rpki.rtr.pdus.PDU.version_map.iterkeys())
+
+  for filename in args.files:
+
+    if ".ax.v" in filename:
+      logging.debug("Reading %s", filename)
+      db = AXFRSet.load(filename)
+
+    elif os.path.basename(filename).startswith("ribs."):
+      db = AXFRSet.parse_bgpdump_rib_dump(filename)
+      db.save_axfr()
+
+    elif not first:
+      assert db is not None
+      db.parse_bgpdump_update(filename)
+      db.save_axfr()
+
+    else:
+      sys.exit("First argument must be a RIB dump or .ax file, don't know what to do with %s" % filename)
+
+    logging.debug("DB serial now %d (%s)", db.serial, db.serial)
+    if first and rpki.rtr.server.read_current(version) == (None, None):
+      db.mark_current()
+    first = False
+
+    for axfr in axfrs:
+      logging.debug("Loading %s", axfr)
+      ax = AXFRSet.load(axfr)
+      logging.debug("Computing changes from %d (%s) to %d (%s)", ax.serial, ax.serial, db.serial, db.serial)
+      db.save_ixfr(ax)
+      del ax
+
+    axfrs.append(db.filename())
+
+
+def bgpdump_select_main(args):
+  """
+  * DANGER WILL ROBINSON! * DEBUGGING AND TEST USE ONLY! *
+  Simulate route origin data from a set of BGP dump files.
+  Set current serial number to correspond to an .ax file created by
+  converting BGP dump files.  SUCH DATA PROVIDE NO SECURITY AT ALL.
+  * DANGER WILL ROBINSON! * DEBUGGING AND TEST USE ONLY! *
+  """
+
+
+  head, sep, tail = os.path.basename(args.ax_file).partition(".")
+  if not head.isdigit() or sep != "." or not tail.startswith("ax.v") or not tail[4:].isdigit():
+    sys.exit("Argument must be name of a .ax file")
+
+  serial = Timestamp(head)
+  version = int(tail[4:])
+
+  if version not in rpki.rtr.pdus.PDU.version_map:
+    sys.exit("Unknown protocol version %d" % version)
+
+  nonce = rpki.rtr.server.read_current(version)[1]
+  if nonce is None:
+    nonce = rpki.rtr.generator.new_nonce()
+
+  rpki.rtr.server.write_current(serial, nonce, version)
+  rpki.rtr.generator.kick_all(serial)
+
+
+class BGPDumpReplayClock(object):
+  """
+  Internal clock for replaying BGP dump files.
+
+                      * DANGER WILL ROBINSON! *
+                   * DEBUGGING AND TEST USE ONLY! *
+
+  This class replaces the normal on-disk serial number mechanism with
+  an in-memory version based on pre-computed data.
+
+  bgpdump_server_main() uses this hack to replay historical data for
+  testing purposes.  DO NOT USE THIS IN PRODUCTION.
+
+  You have been warned.
+  """
+
+  def __init__(self):
+    self.timestamps = [Timestamp(int(f.split(".")[0])) for f in glob.iglob("*.ax.v*")]
+    self.timestamps.sort()
+    self.offset = self.timestamps[0] - int(time.time())
+    self.nonce = rpki.rtr.generator.new_nonce()
+
+  def __nonzero__(self):
+    return len(self.timestamps) > 0
+
+  def now(self):
+    return Timestamp.now(self.offset)
+
+  def read_current(self, version):
+    now = self.now()
+    while len(self.timestamps) > 1 and now >= self.timestamps[1]:
+      del self.timestamps[0]
+    return self.timestamps[0], self.nonce
+
+  def siesta(self):
+    now = self.now()
+    if len(self.timestamps) <= 1:
+      return None
+    elif now < self.timestamps[1]:
+      return self.timestamps[1] - now
+    else:
+      return 1
+
+
+def bgpdump_server_main(args):
+  """
+  Simulate route origin data from a set of BGP dump files.
+
+                      * DANGER WILL ROBINSON! *
+                   * DEBUGGING AND TEST USE ONLY! *
+
+  This is a clone of server_main() which replaces the external serial
+  number updates triggered via the kickme channel by cronjob_main with
+  an internal clocking mechanism to replay historical test data.
+
+  DO NOT USE THIS IN PRODUCTION.
+
+  You have been warned.
+  """
+
+  logger = logging.LoggerAdapter(logging.root, dict(connection = rpki.rtr.server._hostport_tag()))
+
+  logger.debug("[Starting]")
+
+  if args.rpki_rtr_dir:
+    try:
+      os.chdir(args.rpki_rtr_dir)
+    except OSError, e:
+      sys.exit(e)
+
+  # Yes, this really does replace a global function defined in another
+  # module with a bound method to our clock object.  Fun stuff, huh?
+  #
+  clock = BGPDumpReplayClock()
+  rpki.rtr.server.read_current = clock.read_current
+
+  try:
+    server = rpki.rtr.server.ServerChannel(logger = logger)
+    old_serial = server.get_serial()
+    logger.debug("[Starting at serial %d (%s)]", old_serial, old_serial)
+    while clock:
+      new_serial = server.get_serial()
+      if old_serial != new_serial:
+        logger.debug("[Serial bumped from %d (%s) to %d (%s)]", old_serial, old_serial, new_serial, new_serial)
+        server.notify()
+        old_serial = new_serial
+      asyncore.loop(timeout = clock.siesta(), count = 1)
+  except KeyboardInterrupt:
+    sys.exit(0)
+
+
+def argparse_setup(subparsers):
+  """
+  Set up argparse stuff for commands in this module.
+  """
+
+  subparser = subparsers.add_parser("bgpdump-convert", description = bgpdump_convert_main.__doc__,
+                                    help = "Convert bgpdump to fake ROAs")
+  subparser.set_defaults(func = bgpdump_convert_main, default_log_to = "syslog")
+  subparser.add_argument("files", nargs = "+", help = "input files")
+
+  subparser = subparsers.add_parser("bgpdump-select", description = bgpdump_select_main.__doc__,
+                                    help = "Set current serial number for fake ROA data")
+  subparser.set_defaults(func = bgpdump_select_main, default_log_to = "syslog")
+  subparser.add_argument("ax_file", help = "name of the .ax to select")
+
+  subparser = subparsers.add_parser("bgpdump-server", description = bgpdump_server_main.__doc__,
+                                    help = "Replay fake ROAs generated from historical data")
+  subparser.set_defaults(func = bgpdump_server_main, default_log_to = "syslog")
+  subparser.add_argument("rpki_rtr_dir", nargs = "?", help = "directory containing RPKI-RTR database")
diff --git a/rpki/rtr/main.py b/rpki/rtr/main.py
index e5b50ef5..6add407d 100644
--- a/rpki/rtr/main.py
+++ b/rpki/rtr/main.py
@@ -22,6 +22,7 @@ RFC and Internet-Draft repositories near you.
 """
 
 import os
+import sys
 import time
 import logging
 import logging.handlers
@@ -31,6 +32,7 @@ from rpki.rtr.server    import argparse_setup as argparse_setup_server
 from rpki.rtr.client    import argparse_setup as argparse_setup_client
 from rpki.rtr.generator import argparse_setup as argparse_setup_generator
 
+
 class Formatter(logging.Formatter):
 
   converter = time.gmtime
@@ -55,6 +57,12 @@ def main():
   os.environ["TZ"] = "UTC"
   time.tzset()
 
+  if "rpki.rtr.bgpdump" in sys.modules:
+    from rpki.rtr.bgpdump import argparse_setup as argparse_setup_bgpdump
+  else:
+    def argparse_setup_bgpdump(ignored):
+      pass
+
   argparser = argparse.ArgumentParser(description = __doc__)
   argparser.add_argument("--debug", action = "store_true", help = "debugging mode")
   argparser.add_argument("--log-level", default = logging.DEBUG,
@@ -66,6 +74,7 @@ def main():
   argparse_setup_server(subparsers)
   argparse_setup_client(subparsers)
   argparse_setup_generator(subparsers)
+  argparse_setup_bgpdump(subparsers)
   args = argparser.parse_args()
 
   fmt = "rpki-rtr/" + args.mode + "%(connection)s[%(process)d] %(message)s"