Prototype RRDP unpacker.

svn path=/branches/tk705/; revision=5905

1 files changed, 57 insertions, 0 deletions

diff --git a/potpourri/rrdp-test-tool b/potpourri/rrdp-test-tool
new file mode 100755
index 00000000..0fe548b7
--- /dev/null
+++ b/potpourri/rrdp-test-tool
@@ -0,0 +1,57 @@
+#!/usr/bin/env python
+# $Id$
+#
+# Copyright (C) 2013  Dragon Research Labs ("DRL")
+#
+# 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 DRL DISCLAIMS ALL WARRANTIES WITH
+# REGARD TO THIS SOFTWARE INCLUDING ALL IMPLIED WARRANTIES OF MERCHANTABILITY
+# AND FITNESS.  IN NO EVENT SHALL DRL 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.
+
+"""
+Test tool for prototype RRDP implementation.  Eventually some of this
+code will likely be refactored into more user-friendly form, but for
+the moment this just does whatever insane thing I need to do this week
+for testing.
+"""
+
+import rpki.relaxng
+import lxml.etree
+import argparse
+import os
+
+parser = argparse.ArgumentParser(description = __doc__)
+parser.add_argument("rrdp_snapshot",
+                    help = "RRDP snapshot file")
+parser.add_argument("rcynic_tree", nargs = "?", default = "unauthenticated",
+                    help = "directory tree in which to write extracted RPKI objects")
+args = parser.parse_args()
+
+xml = lxml.etree.ElementTree(file = args.rrdp_snapshot).getroot()
+rpki.relaxng.rrdp.assertValid(xml)
+assert xml.tag == rpki.relaxng.rrdp.xmlns + "snapshot"
+
+if not os.path.isdir(args.rcynic_tree):
+  os.makedirs(args.rcynic_tree)
+
+print "Unpacking version %s session %s serial %s" % (
+  xml.get("version"), xml.get("session_id"), xml.get("serial"))
+
+for elt in xml:
+  assert elt.tag == rpki.relaxng.rrdp.xmlns + "publish"
+  uri = elt.get("uri")
+  print " ", uri
+  assert uri.startswith("rsync://")
+  fn = os.path.join(args.rcynic_tree, uri[len("rsync://"):])
+  dn = os.path.dirname(fn)
+  if not os.path.isdir(dn):
+    os.makedirs(dn)
+  with open(fn, "wb") as f:
+    f.write(elt.text.decode("base64"))