Convert to current IETF I-D version of publication protocol. See #705.

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

12 files changed, 608 insertions, 600 deletions

diff --git a/rpki/irdb/zookeeper.py b/rpki/irdb/zookeeper.py
index 6b762b0f..2c600ee5 100644
--- a/rpki/irdb/zookeeper.py
+++ b/rpki/irdb/zookeeper.py
@@ -35,6 +35,7 @@ import rpki.left_right
 import rpki.x509
 import rpki.async
 import rpki.irdb
+import rpki.publication_control
 import django.db.transaction
 
 from lxml.etree import (Element, SubElement, ElementTree,
@@ -536,12 +537,12 @@ class Zookeeper(object):
       updates = []
 
       updates.append(
-        rpki.publication.config_elt.make_pdu(
+        rpki.publication_control.config_elt.make_pdu(
           action = "set",
           bpki_crl = self.server_ca.latest_crl))
 
       updates.extend(
-        rpki.publication.client_elt.make_pdu(
+        rpki.publication_control.client_elt.make_pdu(
           action = "set",
           client_handle = client.handle,
           bpki_cert = client.certificate)
@@ -1143,9 +1144,9 @@ class Zookeeper(object):
                                                       clear_replay_protection = "yes")
                     for ca in rpki.irdb.ResourceHolderCA.objects.all())
     if self.run_pubd:
-      self.call_pubd(rpki.publication.client_elt.make_pdu(action = "set",
-                                                          client_handle = client.handle,
-                                                          clear_replay_protection = "yes")
+      self.call_pubd(rpki.publication_control.client_elt.make_pdu(action = "set",
+                                                                  client_handle = client.handle,
+                                                                  clear_replay_protection = "yes")
                      for client in self.server_ca.clients.all())
 
 
@@ -1172,7 +1173,7 @@ class Zookeeper(object):
       pdus = pdus[0]
 
     call_pubd = rpki.async.sync_wrapper(rpki.http.caller(
-      proto       = rpki.publication,
+      proto       = rpki.publication_control,
       client_key  = irbe.private_key,
       client_cert = irbe.certificate,
       server_ta   = self.server_ca.certificate,
@@ -1189,11 +1190,11 @@ class Zookeeper(object):
     throw exceptions as needed.
     """
 
-    if any(isinstance(pdu, (rpki.left_right.report_error_elt, rpki.publication.report_error_elt)) for pdu in pdus):
+    if any(isinstance(pdu, (rpki.left_right.report_error_elt, rpki.publication_control.report_error_elt)) for pdu in pdus):
       for pdu in pdus:
         if isinstance(pdu, rpki.left_right.report_error_elt):
           self.log("rpkid reported failure: %s" % pdu.error_code)
-        elif isinstance(pdu, rpki.publication.report_error_elt):
+        elif isinstance(pdu, rpki.publication_control.report_error_elt):
           self.log("pubd reported failure: %s" % pdu.error_code)
         else:
           continue
@@ -1531,14 +1532,14 @@ class Zookeeper(object):
 
     # Make sure that pubd's BPKI CRL is up to date.
 
-    self.call_pubd(rpki.publication.config_elt.make_pdu(
+    self.call_pubd(rpki.publication_control.config_elt.make_pdu(
       action = "set",
       bpki_crl = self.server_ca.latest_crl))
 
     # See what pubd already has on file
 
-    pubd_reply = self.call_pubd(rpki.publication.client_elt.make_pdu(action = "list"))
-    client_pdus = dict((x.client_handle, x) for x in pubd_reply if isinstance(x, rpki.publication.client_elt))
+    pubd_reply = self.call_pubd(rpki.publication_control.client_elt.make_pdu(action = "list"))
+    client_pdus = dict((x.client_handle, x) for x in pubd_reply if isinstance(x, rpki.publication_control.client_elt))
     pubd_query = []
 
     # Check all clients
@@ -1550,7 +1551,7 @@ class Zookeeper(object):
       if (client_pdu is None or
           client_pdu.base_uri != client.sia_base or
           client_pdu.bpki_cert != client.certificate):
-        pubd_query.append(rpki.publication.client_elt.make_pdu(
+        pubd_query.append(rpki.publication_control.client_elt.make_pdu(
           action = "create" if client_pdu is None else "set",
           client_handle = client.handle,
           bpki_cert = client.certificate,
@@ -1558,7 +1559,7 @@ class Zookeeper(object):
 
     # Delete any unknown clients
 
-    pubd_query.extend(rpki.publication.client_elt.make_pdu(
+    pubd_query.extend(rpki.publication_control.client_elt.make_pdu(
             action = "destroy", client_handle = p) for p in client_pdus)
 
     # If we changed anything, ship updates off to pubd
diff --git a/rpki/left_right.py b/rpki/left_right.py
index 12c69521..d05d0221 100644
--- a/rpki/left_right.py
+++ b/rpki/left_right.py
@@ -312,16 +312,18 @@ class self_elt(data_elt):
       for ca in parent.cas:
         ca_detail = ca.active_ca_detail
         if ca_detail is not None:
-          q_msg.append(rpki.publication.crl_elt.make_publish(
+          q_msg.append(rpki.publication.publish_elt.make(
             ca_detail.crl_uri, ca_detail.latest_crl))
-          q_msg.append(rpki.publication.manifest_elt.make_publish(
+          q_msg.append(rpki.publication.publish_elt.make(
             ca_detail.manifest_uri, ca_detail.latest_manifest))
-          q_msg.extend(rpki.publication.certificate_elt.make_publish(
+          q_msg.extend(rpki.publication.publish_elt.make(
             c.uri, c.cert) for c in ca_detail.child_certs)
-          q_msg.extend(rpki.publication.roa_elt.make_publish(
+          q_msg.extend(rpki.publication.publish_elt.make(
             r.uri, r.roa) for r in ca_detail.roas if r.roa is not None)
-          q_msg.extend(rpki.publication.ghostbuster_elt.make_publish(
+          q_msg.extend(rpki.publication.publish_elt.make(
             g.uri, g.ghostbuster) for g in ca_detail.ghostbusters)
+          q_msg.extend(rpki.publication.publish_elt.make(
+            c.uri, c.cert) for c in ca_detail.ee_certificates)
       parent.repository.call_pubd(iterator, eb, q_msg)
 
     rpki.async.iterator(self.parents, loop, cb)
@@ -544,7 +546,7 @@ class repository_elt(data_elt):
         handlers = {}
 
       for q_pdu in q_msg:
-        logger.info("Sending %s %s to pubd", q_pdu.action, q_pdu.uri)
+        logger.info("Sending %r to pubd", q_pdu)
 
       bsc = self.bsc
       q_der = rpki.publication.cms_msg().wrap(q_msg, bsc.private_key_id, bsc.signing_cert, bsc.signing_cert_crl)
diff --git a/rpki/pubd.py b/rpki/pubd.py
index 79315a78..e932f686 100644
--- a/rpki/pubd.py
+++ b/rpki/pubd.py
@@ -36,6 +36,7 @@ import rpki.exceptions
 import rpki.relaxng
 import rpki.log
 import rpki.publication
+import rpki.publication_control
 import rpki.daemonize
 
 logger = logging.getLogger(__name__)
@@ -110,39 +111,28 @@ class main(object):
       handlers = (("/control", self.control_handler),
                   ("/client/", self.client_handler)))
 
-  def handler_common(self, query, client, cb, certs, crl = None):
-    """
-    Common PDU handler code.
-    """
-
-    def done(r_msg):
-      reply = rpki.publication.cms_msg().wrap(r_msg, self.pubd_key, self.pubd_cert, crl)
-      self.sql.sweep()
-      cb(reply)
-
-    q_cms = rpki.publication.cms_msg(DER = query)
-    q_msg = q_cms.unwrap(certs)
-    if client is None:
-      self.irbe_cms_timestamp = q_cms.check_replay(self.irbe_cms_timestamp, "control")
-    else:
-      q_cms.check_replay_sql(client, client.client_handle)
-    q_msg.serve_top_level(self, client, done)
 
   def control_handler(self, query, path, cb):
     """
     Process one PDU from the IRBE.
     """
 
-    def done(body):
-      cb(200, body = body)
+    def done(r_msg):
+      self.sql.sweep()
+      cb(code = 200,
+         body = rpki.publication_control.cms_msg().wrap(r_msg, self.pubd_key, self.pubd_cert))
 
     try:
-      self.handler_common(query, None, done, (self.bpki_ta, self.irbe_cert))
+      q_cms = rpki.publication_control.cms_msg(DER = query)
+      q_msg = q_cms.unwrap((self.bpki_ta, self.irbe_cert))
+      self.irbe_cms_timestamp = q_cms.check_replay(self.irbe_cms_timestamp, "control")
+      q_msg.serve_top_level(self, done)
     except (rpki.async.ExitNow, SystemExit):
       raise
     except Exception, e:
       logger.exception("Unhandled exception processing control query, path %r", path)
-      cb(500, reason = "Unhandled exception %s: %s" % (e.__class__.__name__, e))
+      cb(code = 500, reason = "Unhandled exception %s: %s" % (e.__class__.__name__, e))
+
 
   client_url_regexp = re.compile("/client/([-A-Z0-9_/]+)$", re.I)
 
@@ -151,23 +141,29 @@ class main(object):
     Process one PDU from a client.
     """
 
-    def done(body):
-      cb(200, body = body)
+    def done(r_msg):
+      self.sql.sweep()
+      cb(code = 200,
+         body = rpki.publication.cms_msg().wrap(r_msg, self.pubd_key, self.pubd_cert, config.bpki_crl))
 
     try:
       match = self.client_url_regexp.search(path)
       if match is None:
         raise rpki.exceptions.BadContactURL("Bad path: %s" % path)
       client_handle = match.group(1)
-      client = rpki.publication.client_elt.sql_fetch_where1(self, "client_handle = %s", (client_handle,))
+      client = rpki.publication_control.client_elt.sql_fetch_where1(self, "client_handle = %s", (client_handle,))
       if client is None:
         raise rpki.exceptions.ClientNotFound("Could not find client %s" % client_handle)
-      config = rpki.publication.config_elt.fetch(self)
+      config = rpki.publication_control.config_elt.fetch(self)
       if config is None or config.bpki_crl is None:
         raise rpki.exceptions.CMSCRLNotSet
-      self.handler_common(query, client, done, (self.bpki_ta, client.bpki_cert, client.bpki_glue), config.bpki_crl)
+      q_cms = rpki.publication.cms_msg(DER = query)
+      q_msg = q_cms.unwrap((self.bpki_ta, client.bpki_cert, client.bpki_glue))
+      q_cms.check_replay_sql(client, client.client_handle)
+      q_msg.serve_top_level(self, client, done)
     except (rpki.async.ExitNow, SystemExit):
       raise
     except Exception, e:
       logger.exception("Unhandled exception processing client query, path %r", path)
-      cb(500, reason = "Could not process PDU: %s" % e)
+      cb(code = 500,
+         reason = "Could not process PDU: %s" % e)
diff --git a/rpki/publication.py b/rpki/publication.py
index 95f4f314..87a097c9 100644
--- a/rpki/publication.py
+++ b/rpki/publication.py
@@ -1,35 +1,24 @@
 # $Id$
 #
-# Copyright (C) 2009--2012  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.
-#
+# Copyright (C) 2013--2014  Dragon Research Labs ("DRL")
+# Portions copyright (C) 2009--2012  Internet Systems Consortium ("ISC")
 # Portions copyright (C) 2007--2008  American Registry for Internet Numbers ("ARIN")
 #
 # 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.
+# copyright notices and this permission notice appear in all copies.
 #
-# THE SOFTWARE IS PROVIDED "AS IS" AND ARIN DISCLAIMS ALL WARRANTIES WITH
-# REGARD TO THIS SOFTWARE INCLUDING ALL IMPLIED WARRANTIES OF MERCHANTABILITY
-# AND FITNESS.  IN NO EVENT SHALL ARIN 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.
+# THE SOFTWARE IS PROVIDED "AS IS" AND DRL, ISC, AND ARIN DISCLAIM ALL
+# WARRANTIES WITH REGARD TO THIS SOFTWARE INCLUDING ALL IMPLIED
+# WARRANTIES OF MERCHANTABILITY AND FITNESS.  IN NO EVENT SHALL DRL,
+# ISC, OR ARIN 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 "publication" protocol.
+RPKI publication protocol.
 """
 
 import os
@@ -48,208 +37,95 @@ import rpki.log
 
 logger = logging.getLogger(__name__)
 
+
 class publication_namespace(object):
-  """
-  XML namespace parameters for publication protocol.
-  """
 
   xmlns = "http://www.hactrn.net/uris/rpki/publication-spec/"
   nsmap = { None : xmlns }
 
-class control_elt(rpki.xml_utils.data_elt, rpki.sql.sql_persistent, publication_namespace):
-  """
-  Virtual class for control channel objects.
-  """
 
-  def serve_dispatch(self, r_msg, cb, eb):
-    """
-    Action dispatch handler.  This needs special handling because we
-    need to make sure that this PDU arrived via the control channel.
-    """
-    if self.client is not None:
-      raise rpki.exceptions.BadQuery("Control query received on client channel")
-    rpki.xml_utils.data_elt.serve_dispatch(self, r_msg, cb, eb)
-
-class config_elt(control_elt):
+class base_publication_elt(rpki.xml_utils.base_elt, publication_namespace):
   """
-  <config/> element.  This is a little weird because there should
-  never be more than one row in the SQL config table, but we have to
-  put the BPKI CRL somewhere and SQL is the least bad place available.
-
-  So we reuse a lot of the SQL machinery, but we nail config_id at 1,
-  we don't expose it in the XML protocol, and we only support the get
-  and set actions.
+  Base element for publication protocol.  Publish and withdraw PDUs subclass this.
   """
 
-  attributes = ("action", "tag")
-  element_name = "config"
-  elements = ("bpki_crl",)
-
-  sql_template = rpki.sql.template(
-    "config",
-    "config_id",
-    ("bpki_crl", rpki.x509.CRL))
-
-  wired_in_config_id = 1
-
-  def startElement(self, stack, name, attrs):
-    """
-    StartElement() handler for config object.  This requires special
-    handling because of the weird way we treat config_id.
-    """
-    control_elt.startElement(self, stack, name, attrs)
-    self.config_id = self.wired_in_config_id
-
-  @classmethod
-  def fetch(cls, gctx):
-    """
-    Fetch the config object from SQL.  This requires special handling
-    because of the weird way we treat config_id.
-    """
-    return cls.sql_fetch(gctx, cls.wired_in_config_id)
+  attributes = ("tag", "uri")
+  payload = None
 
-  def serve_set(self, r_msg, cb, eb):
-    """
-    Handle a set action.  This requires special handling because
-    config doesn't support the create method.
-    """
-    if self.sql_fetch(self.gctx, self.config_id) is None:
-      control_elt.serve_create(self, r_msg, cb, eb)
-    else:
-      control_elt.serve_set(self, r_msg, cb, eb)
+  def __repr__(self):
+    return rpki.log.log_repr(self, self.uri, self.payload)
 
-  def serve_fetch_one_maybe(self):
+  def serve_dispatch(self, r_msg, cb, eb):
     """
-    Find the config object on which a get or set method should
-    operate.
+    Action dispatch handler.
     """
-    return self.sql_fetch(self.gctx, self.config_id)
 
-class client_elt(control_elt):
-  """
-  <client/> element.
-  """
+    try:
+      self.client.check_allowed_uri(self.uri)
+      self.serve_action()
+      r_pdu = self.__class__()
+      r_pdu.tag = self.tag
+      r_pdu.uri = self.uri
+      r_msg.append(r_pdu)
+      cb()
+    except rpki.exceptions.NoObjectAtURI, e:
+      # This can happen when we're cleaning up from a prior mess, so
+      # we generate a <report_error/> PDU then carry on.
+      r_msg.append(report_error_elt.from_exception(e, self.tag))
+      cb()
 
-  element_name = "client"
-  attributes = ("action", "tag", "client_handle", "base_uri")
-  elements = ("bpki_cert", "bpki_glue")
-  booleans = ("clear_replay_protection",)
-
-  sql_template = rpki.sql.template(
-    "client",
-    "client_id",
-    "client_handle",
-    "base_uri",
-    ("bpki_cert", rpki.x509.X509),
-    ("bpki_glue", rpki.x509.X509),
-    ("last_cms_timestamp", rpki.sundial.datetime))
-
-  base_uri  = None
-  bpki_cert = None
-  bpki_glue = None
-  last_cms_timestamp = None
-
-  def serve_post_save_hook(self, q_pdu, r_pdu, cb, eb):
-    """
-    Extra server actions for client_elt.
-    """
-    actions = []
-    if q_pdu.clear_replay_protection:
-      actions.append(self.serve_clear_replay_protection)
-    def loop(iterator, action):
-      action(iterator, eb)
-    rpki.async.iterator(actions, loop, cb)
-
-  def serve_clear_replay_protection(self, cb, eb):
+  def uri_to_filename(self):
     """
-    Handle a clear_replay_protection action for this client.
+    Convert a URI to a local filename.
     """
-    self.last_cms_timestamp = None
-    self.sql_mark_dirty()
-    cb()
 
-  def serve_fetch_one_maybe(self):
-    """
-    Find the client object on which a get, set, or destroy method
-    should operate, or which would conflict with a create method.
-    """
-    return self.sql_fetch_where1(self.gctx, "client_handle = %s", (self.client_handle,))
+    if not self.uri.startswith("rsync://"):
+      raise rpki.exceptions.BadURISyntax(self.uri)
+    path = self.uri.split("/")[3:]
+    if not self.gctx.publication_multimodule:
+      del path[0]
+    path.insert(0, self.gctx.publication_base.rstrip("/"))
+    filename = "/".join(path)
+    if "/../" in filename or filename.endswith("/.."):
+      raise rpki.exceptions.BadURISyntax(filename)
+    return filename
 
-  def serve_fetch_all(self):
+  def raise_if_error(self):
     """
-    Find client objects on which a list method should operate.
+    No-op, since this is not a <report_error/> PDU.
     """
-    return self.sql_fetch_all(self.gctx)
+    pass
 
-  def check_allowed_uri(self, uri):
-    """
-    Make sure that a target URI is within this client's allowed URI space.
-    """
-    if not uri.startswith(self.base_uri):
-      raise rpki.exceptions.ForbiddenURI
 
-class publication_object_elt(rpki.xml_utils.base_elt, publication_namespace):
-  """
-  Virtual class for publishable objects.  These have very similar
-  syntax, differences lie in underlying datatype and methods.  XML
-  methods are a little different from the pattern used for objects
-  that support the create/set/get/list/destroy actions, but
-  publishable objects don't go in SQL either so these classes would be
-  different in any case.
-  """
+class publish_elt(base_publication_elt):
 
-  attributes = ("action", "tag", "client_handle", "uri")
-  payload_type = None
-  payload = None
+  element_name = "publish"
 
   def endElement(self, stack, name, text):
     """
-    Handle a publishable element element.
+    Handle reading of the object to be published
     """
+
     assert name == self.element_name, "Unexpected name %s, stack %s" % (name, stack)
     if text:
-      self.payload = self.payload_type(Base64 = text) # pylint: disable=E1102
+      self.payload = rpki.x509.uri_dispatch(self.uri)(Base64 = text)
     stack.pop()
 
   def toXML(self):
     """
     Generate XML element for publishable object.
     """
+
     elt = self.make_elt()
-    if self.payload:
+    if self.payload != None:
       elt.text = self.payload.get_Base64()
     return elt
 
-  def serve_dispatch(self, r_msg, cb, eb):
-    """
-    Action dispatch handler.
-    """
-    # pylint: disable=E0203
-    try:
-      if self.client is None:
-        raise rpki.exceptions.BadQuery("Client query received on control channel")
-      dispatch = { "publish"  : self.serve_publish,
-                   "withdraw" : self.serve_withdraw }
-      if self.action not in dispatch:
-        raise rpki.exceptions.BadQuery("Unexpected query: action %s" % self.action)
-      self.client.check_allowed_uri(self.uri)
-      dispatch[self.action]()
-      r_pdu = self.__class__()
-      r_pdu.action = self.action
-      r_pdu.tag = self.tag
-      r_pdu.uri = self.uri
-      r_msg.append(r_pdu)
-      cb()
-    except rpki.exceptions.NoObjectAtURI, e:
-      # This can happen when we're cleaning up from a prior mess, so
-      # we generate a <report_error/> PDU then carry on.
-      r_msg.append(report_error_elt.from_exception(e, self.tag))
-      cb()
-
-  def serve_publish(self):
+  def serve_action(self):
     """
     Publish an object.
     """
+
     logger.info("Publishing %s", self.payload.tracking_data(self.uri))
     filename = self.uri_to_filename()
     filename_tmp = filename + ".tmp"
@@ -261,10 +137,25 @@ class publication_object_elt(rpki.xml_utils.base_elt, publication_namespace):
     f.close()
     os.rename(filename_tmp, filename)
 
-  def serve_withdraw(self):
+  @classmethod
+  def make(cls, uri, obj, tag = None):
+    """
+    Construct a publication PDU.
+    """
+
+    assert isinstance(obj, rpki.x509.uri_dispatch(uri))
+    return cls.make_pdu(uri = uri, payload = obj, tag = tag)
+
+
+class withdraw_elt(base_publication_elt):
+
+  element_name = "withdraw"
+
+  def serve_action(self):
     """
     Withdraw an object, then recursively delete empty directories.
     """
+
     logger.info("Withdrawing %s", self.uri)
     filename = self.uri_to_filename()
     try:
@@ -284,86 +175,15 @@ class publication_object_elt(rpki.xml_utils.base_elt, publication_namespace):
       else:
         dirname = os.path.dirname(dirname)
 
-  def uri_to_filename(self):
-    """
-    Convert a URI to a local filename.
-    """
-    if not self.uri.startswith("rsync://"):
-      raise rpki.exceptions.BadURISyntax(self.uri)
-    path = self.uri.split("/")[3:]
-    if not self.gctx.publication_multimodule:
-      del path[0]
-    path.insert(0, self.gctx.publication_base.rstrip("/"))
-    filename = "/".join(path)
-    if "/../" in filename or filename.endswith("/.."):
-      raise rpki.exceptions.BadURISyntax(filename)
-    return filename
-
   @classmethod
-  def make_publish(cls, uri, obj, tag = None):
-    """
-    Construct a publication PDU.
-    """
-    assert cls.payload_type is not None and type(obj) is cls.payload_type
-    return cls.make_pdu(action = "publish", uri = uri, payload = obj, tag = tag)
-
-  @classmethod
-  def make_withdraw(cls, uri, obj, tag = None):
+  def make(cls, uri, obj, tag = None):
     """
     Construct a withdrawal PDU.
     """
-    assert cls.payload_type is not None and type(obj) is cls.payload_type
-    return cls.make_pdu(action = "withdraw", uri = uri, tag = tag)
-
-  def raise_if_error(self):
-    """
-    No-op, since this is not a <report_error/> PDU.
-    """
-    pass
-
-class certificate_elt(publication_object_elt):
-  """
-  <certificate/> element.
-  """
-
-  element_name = "certificate"
-  payload_type = rpki.x509.X509
-
-class crl_elt(publication_object_elt):
-  """
-  <crl/> element.
-  """
-
-  element_name = "crl"
-  payload_type = rpki.x509.CRL
-
-class manifest_elt(publication_object_elt):
-  """
-  <manifest/> element.
-  """
-
-  element_name = "manifest"
-  payload_type = rpki.x509.SignedManifest
-
-class roa_elt(publication_object_elt):
-  """
-  <roa/> element.
-  """
-
-  element_name = "roa"
-  payload_type = rpki.x509.ROA
-
-class ghostbuster_elt(publication_object_elt):
-  """
-  <ghostbuster/> element.
-  """
 
-  element_name = "ghostbuster"
-  payload_type = rpki.x509.Ghostbuster
+    assert isinstance(obj, rpki.x509.uri_dispatch(uri))
+    return cls.make_pdu(uri = uri, tag = tag)
 
-publication_object_elt.obj2elt = dict(
-  (e.payload_type, e) for e in
-  (certificate_elt, crl_elt, manifest_elt, roa_elt, ghostbuster_elt))
 
 class report_error_elt(rpki.xml_utils.text_elt, publication_namespace):
   """
@@ -376,6 +196,9 @@ class report_error_elt(rpki.xml_utils.text_elt, publication_namespace):
 
   error_text = None
 
+  def __repr__(self):
+    return rpki.log.log_repr(self)
+
   @classmethod
   def from_exception(cls, e, tag = None):
     """
@@ -406,6 +229,7 @@ class report_error_elt(rpki.xml_utils.text_elt, publication_namespace):
     else:
       raise rpki.exceptions.BadPublicationReply("Unexpected response from pubd: %s" % self)
 
+
 class msg(rpki.xml_utils.msg, publication_namespace):
   """
   Publication PDU.
@@ -413,12 +237,11 @@ class msg(rpki.xml_utils.msg, publication_namespace):
 
   ## @var version
   # Protocol version
-  version = 1
+  version = 3
 
   ## @var pdus
   # Dispatch table of PDUs for this protocol.
-  pdus = dict((x.element_name, x) for x in
-              (config_elt, client_elt, certificate_elt, crl_elt, manifest_elt, roa_elt, ghostbuster_elt, report_error_elt))
+  pdus = dict((x.element_name, x) for x in (publish_elt, withdraw_elt, report_error_elt))
 
   def serve_top_level(self, gctx, client, cb):
     """
@@ -450,6 +273,7 @@ class msg(rpki.xml_utils.msg, publication_namespace):
 
     rpki.async.iterator(self, loop, done)
 
+
 class sax_handler(rpki.xml_utils.sax_handler):
   """
   SAX handler for publication protocol.
@@ -457,7 +281,8 @@ class sax_handler(rpki.xml_utils.sax_handler):
 
   pdu = msg
   name = "msg"
-  version = "1"
+  version = "3"
+
 
 class cms_msg(rpki.x509.XML_CMS_object):
   """
diff --git a/rpki/publication_control.py b/rpki/publication_control.py
new file mode 100644
index 00000000..bd6a8db2
--- /dev/null
+++ b/rpki/publication_control.py
@@ -0,0 +1,292 @@
+# $Id$
+#
+# Copyright (C) 2009--2012  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.
+#
+# Portions copyright (C) 2007--2008  American Registry for Internet Numbers ("ARIN")
+#
+# 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 ARIN DISCLAIMS ALL WARRANTIES WITH
+# REGARD TO THIS SOFTWARE INCLUDING ALL IMPLIED WARRANTIES OF MERCHANTABILITY
+# AND FITNESS.  IN NO EVENT SHALL ARIN 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 publication control protocol.
+
+Per IETF SIDR WG discussion, this is now separate from the publication
+protocol itself.
+"""
+
+import logging
+import rpki.resource_set
+import rpki.x509
+import rpki.sql
+import rpki.exceptions
+import rpki.xml_utils
+import rpki.http
+import rpki.up_down
+import rpki.relaxng
+import rpki.sundial
+import rpki.log
+
+logger = logging.getLogger(__name__)
+
+
+class publication_control_namespace(object):
+  """
+  XML namespace parameters for publication control protocol.
+  """
+
+  xmlns = "http://www.hactrn.net/uris/rpki/publication-control/"
+  nsmap = { None : xmlns }
+
+
+class control_elt(rpki.xml_utils.data_elt, rpki.sql.sql_persistent, publication_control_namespace):
+  """
+  Virtual class for control channel objects.
+  """
+
+  pass
+
+
+class config_elt(control_elt):
+  """
+  <config/> element.  This is a little weird because there should
+  never be more than one row in the SQL config table, but we have to
+  put the BPKI CRL somewhere and SQL is the least bad place available.
+
+  So we reuse a lot of the SQL machinery, but we nail config_id at 1,
+  we don't expose it in the XML protocol, and we only support the get
+  and set actions.
+  """
+
+  attributes = ("action", "tag")
+  element_name = "config"
+  elements = ("bpki_crl",)
+
+  sql_template = rpki.sql.template(
+    "config",
+    "config_id",
+    ("bpki_crl", rpki.x509.CRL))
+
+  wired_in_config_id = 1
+
+  def startElement(self, stack, name, attrs):
+    """
+    StartElement() handler for config object.  This requires special
+    handling because of the weird way we treat config_id.
+    """
+    control_elt.startElement(self, stack, name, attrs)
+    self.config_id = self.wired_in_config_id
+
+  @classmethod
+  def fetch(cls, gctx):
+    """
+    Fetch the config object from SQL.  This requires special handling
+    because of the weird way we treat config_id.
+    """
+    return cls.sql_fetch(gctx, cls.wired_in_config_id)
+
+  def serve_set(self, r_msg, cb, eb):
+    """
+    Handle a set action.  This requires special handling because
+    config doesn't support the create method.
+    """
+    if self.sql_fetch(self.gctx, self.config_id) is None:
+      control_elt.serve_create(self, r_msg, cb, eb)
+    else:
+      control_elt.serve_set(self, r_msg, cb, eb)
+
+  def serve_fetch_one_maybe(self):
+    """
+    Find the config object on which a get or set method should
+    operate.
+    """
+    return self.sql_fetch(self.gctx, self.config_id)
+
+
+class client_elt(control_elt):
+  """
+  <client/> element.
+  """
+
+  element_name = "client"
+  attributes = ("action", "tag", "client_handle", "base_uri")
+  elements = ("bpki_cert", "bpki_glue")
+  booleans = ("clear_replay_protection",)
+
+  sql_template = rpki.sql.template(
+    "client",
+    "client_id",
+    "client_handle",
+    "base_uri",
+    ("bpki_cert", rpki.x509.X509),
+    ("bpki_glue", rpki.x509.X509),
+    ("last_cms_timestamp", rpki.sundial.datetime))
+
+  base_uri  = None
+  bpki_cert = None
+  bpki_glue = None
+  last_cms_timestamp = None
+
+  def serve_post_save_hook(self, q_pdu, r_pdu, cb, eb):
+    """
+    Extra server actions for client_elt.
+    """
+    actions = []
+    if q_pdu.clear_replay_protection:
+      actions.append(self.serve_clear_replay_protection)
+    def loop(iterator, action):
+      action(iterator, eb)
+    rpki.async.iterator(actions, loop, cb)
+
+  def serve_clear_replay_protection(self, cb, eb):
+    """
+    Handle a clear_replay_protection action for this client.
+    """
+    self.last_cms_timestamp = None
+    self.sql_mark_dirty()
+    cb()
+
+  def serve_fetch_one_maybe(self):
+    """
+    Find the client object on which a get, set, or destroy method
+    should operate, or which would conflict with a create method.
+    """
+    return self.sql_fetch_where1(self.gctx, "client_handle = %s", (self.client_handle,))
+
+  def serve_fetch_all(self):
+    """
+    Find client objects on which a list method should operate.
+    """
+    return self.sql_fetch_all(self.gctx)
+
+  def check_allowed_uri(self, uri):
+    """
+    Make sure that a target URI is within this client's allowed URI space.
+    """
+    if not uri.startswith(self.base_uri):
+      raise rpki.exceptions.ForbiddenURI
+
+
+class report_error_elt(rpki.xml_utils.text_elt, publication_control_namespace):
+  """
+  <report_error/> element.
+  """
+
+  element_name = "report_error"
+  attributes = ("tag", "error_code")
+  text_attribute = "error_text"
+
+  error_text = None
+
+  @classmethod
+  def from_exception(cls, e, tag = None):
+    """
+    Generate a <report_error/> element from an exception.
+    """
+    self = cls()
+    self.tag = tag
+    self.error_code = e.__class__.__name__
+    self.error_text = str(e)
+    return self
+
+  def __str__(self):
+    s = ""
+    if getattr(self, "tag", None) is not None:
+      s += "[%s] " % self.tag
+    s += self.error_code
+    if getattr(self, "error_text", None) is not None:
+      s += ": " + self.error_text
+    return s
+
+  def raise_if_error(self):
+    """
+    Raise exception associated with this <report_error/> PDU.
+    """
+    t = rpki.exceptions.__dict__.get(self.error_code)
+    if isinstance(t, type) and issubclass(t, rpki.exceptions.RPKI_Exception):
+      raise t(getattr(self, "text", None))
+    else:
+      raise rpki.exceptions.BadPublicationReply("Unexpected response from pubd: %s" % self)
+
+
+class msg(rpki.xml_utils.msg, publication_control_namespace):
+  """
+  Publication control PDU.
+  """
+
+  ## @var version
+  # Protocol version
+  version = 1
+
+  ## @var pdus
+  # Dispatch table of PDUs for this protocol.
+  pdus = dict((x.element_name, x) for x in (config_elt, client_elt, report_error_elt))
+
+  def serve_top_level(self, gctx, cb):
+    """
+    Serve one msg PDU.
+    """
+    if not self.is_query():
+      raise rpki.exceptions.BadQuery("Message type is not query")
+    r_msg = self.__class__.reply()
+
+    def loop(iterator, q_pdu):
+
+      def fail(e):
+        if not isinstance(e, rpki.exceptions.NotFound):
+          logger.exception("Exception processing PDU %r", q_pdu)
+        r_msg.append(report_error_elt.from_exception(e, q_pdu.tag))
+        cb(r_msg)
+
+      try:
+        q_pdu.gctx = gctx
+        q_pdu.serve_dispatch(r_msg, iterator, fail)
+      except (rpki.async.ExitNow, SystemExit):
+        raise
+      except Exception, e:
+        fail(e)
+
+    def done():
+      cb(r_msg)
+
+    rpki.async.iterator(self, loop, done)
+
+
+class sax_handler(rpki.xml_utils.sax_handler):
+  """
+  SAX handler for publication control protocol.
+  """
+
+  pdu = msg
+  name = "msg"
+  version = "1"
+
+
+class cms_msg(rpki.x509.XML_CMS_object):
+  """
+  Class to hold a CMS-signed publication control PDU.
+  """
+
+  encoding = "us-ascii"
+  schema = rpki.relaxng.publication_control
+  saxify = sax_handler.saxify
diff --git a/rpki/relaxng.py b/rpki/relaxng.py
index 07d7e05b..30a0824d 100644
--- a/rpki/relaxng.py
+++ b/rpki/relaxng.py
@@ -1478,9 +1478,9 @@ myrpki = lxml.etree.RelaxNG(lxml.etree.fromstring(r'''<?xml version="1.0" encodi
 -->
 '''))
 
-## @var publication
-## Parsed RelaxNG publication schema
-publication = lxml.etree.RelaxNG(lxml.etree.fromstring(r'''<?xml version="1.0" encoding="UTF-8"?>
+## @var publication_control
+## Parsed RelaxNG publication_control schema
+publication_control = lxml.etree.RelaxNG(lxml.etree.fromstring(r'''<?xml version="1.0" encoding="UTF-8"?>
 <!--
   $Id: publication-schema.rnc 5876 2014-06-26 19:00:12Z sra $
   
@@ -1503,7 +1503,7 @@ publication = lxml.etree.RelaxNG(lxml.etree.fromstring(r'''<?xml version="1.0" e
   NEGLIGENCE OR OTHER TORTIOUS ACTION, ARISING OUT OF OR IN CONNECTION
   WITH THE USE OR PERFORMANCE OF THIS SOFTWARE.
 -->
-<grammar ns="http://www.hactrn.net/uris/rpki/publication-spec/" xmlns="http://relaxng.org/ns/structure/1.0" datatypeLibrary="http://www.w3.org/2001/XMLSchema-datatypes">
+<grammar ns="http://www.hactrn.net/uris/rpki/publication-control/" xmlns="http://relaxng.org/ns/structure/1.0" datatypeLibrary="http://www.w3.org/2001/XMLSchema-datatypes">
   <!-- Top level PDU -->
   <start>
     <element name="msg">
@@ -1537,11 +1537,6 @@ publication = lxml.etree.RelaxNG(lxml.etree.fromstring(r'''<?xml version="1.0" e
     <choice>
       <ref name="config_query"/>
       <ref name="client_query"/>
-      <ref name="certificate_query"/>
-      <ref name="crl_query"/>
-      <ref name="manifest_query"/>
-      <ref name="roa_query"/>
-      <ref name="ghostbuster_query"/>
     </choice>
   </define>
   <!-- PDUs allowed in a reply -->
@@ -1549,11 +1544,6 @@ publication = lxml.etree.RelaxNG(lxml.etree.fromstring(r'''<?xml version="1.0" e
     <choice>
       <ref name="config_reply"/>
       <ref name="client_reply"/>
-      <ref name="certificate_reply"/>
-      <ref name="crl_reply"/>
-      <ref name="manifest_reply"/>
-      <ref name="roa_reply"/>
-      <ref name="ghostbuster_reply"/>
       <ref name="report_error_reply"/>
     </choice>
   </define>
@@ -1598,7 +1588,7 @@ publication = lxml.etree.RelaxNG(lxml.etree.fromstring(r'''<?xml version="1.0" e
     </data>
   </define>
   <!--
-    <config/> element (use restricted to repository operator)
+    <config/> element
     config_handle attribute, create, list, and destroy commands omitted deliberately, see code for details
   -->
   <define name="config_payload">
@@ -1650,7 +1640,7 @@ publication = lxml.etree.RelaxNG(lxml.etree.fromstring(r'''<?xml version="1.0" e
       <ref name="config_payload"/>
     </element>
   </define>
-  <!-- <client/> element (use restricted to repository operator) -->
+  <!-- <client/> element -->
   <define name="client_handle">
     <attribute name="client_handle">
       <ref name="object_handle"/>
@@ -1795,196 +1785,152 @@ publication = lxml.etree.RelaxNG(lxml.etree.fromstring(r'''<?xml version="1.0" e
       <ref name="client_handle"/>
     </element>
   </define>
-  <!-- <certificate/> element -->
-  <define name="certificate_query" combine="choice">
-    <element name="certificate">
-      <attribute name="action">
-        <value>publish</value>
-      </attribute>
-      <optional>
-        <ref name="tag"/>
-      </optional>
-      <ref name="uri"/>
-      <ref name="base64"/>
-    </element>
-  </define>
-  <define name="certificate_reply" combine="choice">
-    <element name="certificate">
-      <attribute name="action">
-        <value>publish</value>
-      </attribute>
-      <optional>
-        <ref name="tag"/>
-      </optional>
-      <ref name="uri"/>
-    </element>
-  </define>
-  <define name="certificate_query" combine="choice">
-    <element name="certificate">
-      <attribute name="action">
-        <value>withdraw</value>
-      </attribute>
-      <optional>
-        <ref name="tag"/>
-      </optional>
-      <ref name="uri"/>
-    </element>
-  </define>
-  <define name="certificate_reply" combine="choice">
-    <element name="certificate">
-      <attribute name="action">
-        <value>withdraw</value>
-      </attribute>
-      <optional>
-        <ref name="tag"/>
-      </optional>
-      <ref name="uri"/>
-    </element>
-  </define>
-  <!-- <crl/> element -->
-  <define name="crl_query" combine="choice">
-    <element name="crl">
-      <attribute name="action">
-        <value>publish</value>
-      </attribute>
-      <optional>
-        <ref name="tag"/>
-      </optional>
-      <ref name="uri"/>
-      <ref name="base64"/>
-    </element>
-  </define>
-  <define name="crl_reply" combine="choice">
-    <element name="crl">
-      <attribute name="action">
-        <value>publish</value>
-      </attribute>
-      <optional>
-        <ref name="tag"/>
-      </optional>
-      <ref name="uri"/>
-    </element>
-  </define>
-  <define name="crl_query" combine="choice">
-    <element name="crl">
-      <attribute name="action">
-        <value>withdraw</value>
-      </attribute>
-      <optional>
-        <ref name="tag"/>
-      </optional>
-      <ref name="uri"/>
-    </element>
+  <!-- <report_error/> element -->
+  <define name="error">
+    <data type="token">
+      <param name="maxLength">1024</param>
+    </data>
   </define>
-  <define name="crl_reply" combine="choice">
-    <element name="crl">
-      <attribute name="action">
-        <value>withdraw</value>
-      </attribute>
+  <define name="report_error_reply">
+    <element name="report_error">
       <optional>
         <ref name="tag"/>
       </optional>
-      <ref name="uri"/>
-    </element>
-  </define>
-  <!-- <manifest/> element -->
-  <define name="manifest_query" combine="choice">
-    <element name="manifest">
-      <attribute name="action">
-        <value>publish</value>
+      <attribute name="error_code">
+        <ref name="error"/>
       </attribute>
       <optional>
-        <ref name="tag"/>
+        <data type="string">
+          <param name="maxLength">512000</param>
+        </data>
       </optional>
-      <ref name="uri"/>
-      <ref name="base64"/>
     </element>
   </define>
-  <define name="manifest_reply" combine="choice">
-    <element name="manifest">
-      <attribute name="action">
-        <value>publish</value>
-      </attribute>
-      <optional>
-        <ref name="tag"/>
-      </optional>
-      <ref name="uri"/>
-    </element>
+</grammar>
+<!--
+  Local Variables:
+  indent-tabs-mode: nil
+  comment-start: "# "
+  comment-start-skip: "#[ \t]*"
+  End:
+-->
+'''))
+
+## @var publication
+## Parsed RelaxNG publication schema
+publication = lxml.etree.RelaxNG(lxml.etree.fromstring(r'''<?xml version="1.0" encoding="UTF-8"?>
+<!--
+  $Id: publication-schema.rnc 5876 2014-06-26 19:00:12Z sra $
+  
+  RelaxNG schema for RPKI publication protocol, from current I-D.
+  
+  Copyright (c) 2014 IETF Trust and the persons identified as authors
+  of the code.  All rights reserved.
+  
+  Redistribution and use in source and binary forms, with or without
+  modification, are permitted provided that the following conditions
+  are met:
+  
+  * Redistributions of source code must retain the above copyright
+    notice, this list of conditions and the following disclaimer.
+  
+  * Redistributions in binary form must reproduce the above copyright
+    notice, this list of conditions and the following disclaimer in
+    the documentation and/or other materials provided with the
+    distribution.
+  
+  * Neither the name of Internet Society, IETF or IETF Trust, nor the
+    names of specific contributors, may be used to endorse or promote
+    products derived from this software without specific prior written
+    permission.
+  
+  THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS
+  "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT
+  LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS
+  FOR A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE
+  COPYRIGHT OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT,
+  INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING,
+  BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES;
+  LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER
+  CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT
+  LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN
+  ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE
+  POSSIBILITY OF SUCH DAMAGE.
+-->
+<grammar ns="http://www.hactrn.net/uris/rpki/publication-spec/" xmlns="http://relaxng.org/ns/structure/1.0" datatypeLibrary="http://www.w3.org/2001/XMLSchema-datatypes">
+  <!-- This is version 3 of the protocol. -->
+  <define name="version">
+    <value>3</value>
   </define>
-  <define name="manifest_query" combine="choice">
-    <element name="manifest">
-      <attribute name="action">
-        <value>withdraw</value>
+  <!-- Top level PDU is either a query or a reply. -->
+  <start>
+    <element name="msg">
+      <attribute name="version">
+        <ref name="version"/>
       </attribute>
-      <optional>
-        <ref name="tag"/>
-      </optional>
-      <ref name="uri"/>
+      <choice>
+        <group>
+          <attribute name="type">
+            <value>query</value>
+          </attribute>
+          <zeroOrMore>
+            <ref name="query_elt"/>
+          </zeroOrMore>
+        </group>
+        <group>
+          <attribute name="type">
+            <value>reply</value>
+          </attribute>
+          <zeroOrMore>
+            <ref name="reply_elt"/>
+          </zeroOrMore>
+        </group>
+      </choice>
     </element>
+  </start>
+  <!-- PDUs allowed in  queries and replies. -->
+  <define name="query_elt">
+    <choice>
+      <ref name="publish_query"/>
+      <ref name="withdraw_query"/>
+    </choice>
   </define>
-  <define name="manifest_reply" combine="choice">
-    <element name="manifest">
-      <attribute name="action">
-        <value>withdraw</value>
-      </attribute>
-      <optional>
-        <ref name="tag"/>
-      </optional>
-      <ref name="uri"/>
-    </element>
+  <define name="reply_elt">
+    <choice>
+      <ref name="publish_reply"/>
+      <ref name="withdraw_reply"/>
+      <ref name="report_error_reply"/>
+    </choice>
   </define>
-  <!-- <roa/> element -->
-  <define name="roa_query" combine="choice">
-    <element name="roa">
-      <attribute name="action">
-        <value>publish</value>
-      </attribute>
-      <optional>
-        <ref name="tag"/>
-      </optional>
-      <ref name="uri"/>
-      <ref name="base64"/>
-    </element>
+  <!-- Tag attributes for bulk operations. -->
+  <define name="tag">
+    <attribute name="tag">
+      <data type="token">
+        <param name="maxLength">1024</param>
+      </data>
+    </attribute>
   </define>
-  <define name="roa_reply" combine="choice">
-    <element name="roa">
-      <attribute name="action">
-        <value>publish</value>
-      </attribute>
-      <optional>
-        <ref name="tag"/>
-      </optional>
-      <ref name="uri"/>
-    </element>
+  <!-- Base64 encoded DER stuff. -->
+  <define name="base64">
+    <data type="base64Binary"/>
   </define>
-  <define name="roa_query" combine="choice">
-    <element name="roa">
-      <attribute name="action">
-        <value>withdraw</value>
-      </attribute>
-      <optional>
-        <ref name="tag"/>
-      </optional>
-      <ref name="uri"/>
-    </element>
+  <!-- Publication URIs. -->
+  <define name="uri">
+    <attribute name="uri">
+      <data type="anyURI">
+        <param name="maxLength">4096</param>
+      </data>
+    </attribute>
   </define>
-  <define name="roa_reply" combine="choice">
-    <element name="roa">
-      <attribute name="action">
-        <value>withdraw</value>
-      </attribute>
-      <optional>
-        <ref name="tag"/>
-      </optional>
-      <ref name="uri"/>
-    </element>
+  <!-- Error codes. -->
+  <define name="error">
+    <data type="token">
+      <param name="maxLength">1024</param>
+    </data>
   </define>
-  <!-- <ghostbuster/> element -->
-  <define name="ghostbuster_query" combine="choice">
-    <element name="ghostbuster">
-      <attribute name="action">
-        <value>publish</value>
-      </attribute>
+  <!-- <publish/> element -->
+  <define name="publish_query" combine="choice">
+    <element name="publish">
       <optional>
         <ref name="tag"/>
       </optional>
@@ -1992,33 +1938,25 @@ publication = lxml.etree.RelaxNG(lxml.etree.fromstring(r'''<?xml version="1.0" e
       <ref name="base64"/>
     </element>
   </define>
-  <define name="ghostbuster_reply" combine="choice">
-    <element name="ghostbuster">
-      <attribute name="action">
-        <value>publish</value>
-      </attribute>
+  <define name="publish_reply" combine="choice">
+    <element name="publish">
       <optional>
         <ref name="tag"/>
       </optional>
       <ref name="uri"/>
     </element>
   </define>
-  <define name="ghostbuster_query" combine="choice">
-    <element name="ghostbuster">
-      <attribute name="action">
-        <value>withdraw</value>
-      </attribute>
+  <!-- <withdraw/> element -->
+  <define name="withdraw_query" combine="choice">
+    <element name="withdraw">
       <optional>
         <ref name="tag"/>
       </optional>
       <ref name="uri"/>
     </element>
   </define>
-  <define name="ghostbuster_reply" combine="choice">
-    <element name="ghostbuster">
-      <attribute name="action">
-        <value>withdraw</value>
-      </attribute>
+  <define name="withdraw_reply" combine="choice">
+    <element name="withdraw">
       <optional>
         <ref name="tag"/>
       </optional>
@@ -2026,11 +1964,6 @@ publication = lxml.etree.RelaxNG(lxml.etree.fromstring(r'''<?xml version="1.0" e
     </element>
   </define>
   <!-- <report_error/> element -->
-  <define name="error">
-    <data type="token">
-      <param name="maxLength">1024</param>
-    </data>
-  </define>
   <define name="report_error_reply">
     <element name="report_error">
       <optional>
diff --git a/rpki/rpkid.py b/rpki/rpkid.py
index 36ee2ea9..cb792572 100644
--- a/rpki/rpkid.py
+++ b/rpki/rpkid.py
@@ -933,8 +933,7 @@ class ca_detail_obj(rpki.sql.sql_persistent):
     repository = ca.parent.repository
     handler = False if allow_failure else None
     for child_cert in self.child_certs:
-      publisher.withdraw(cls = rpki.publication.certificate_elt,
-                         uri = child_cert.uri,
+      publisher.withdraw(uri = child_cert.uri,
                          obj = child_cert.cert,
                          repository = repository,
                          handler = handler)
@@ -948,8 +947,7 @@ class ca_detail_obj(rpki.sql.sql_persistent):
     except AttributeError:
       latest_manifest = None
     if latest_manifest is not None:
-      publisher.withdraw(cls = rpki.publication.manifest_elt,
-                         uri = self.manifest_uri,
+      publisher.withdraw(uri = self.manifest_uri,
                          obj = self.latest_manifest,
                          repository = repository,
                          handler = handler)
@@ -958,8 +956,7 @@ class ca_detail_obj(rpki.sql.sql_persistent):
     except AttributeError:
       latest_crl = None
     if latest_crl is not None:
-      publisher.withdraw(cls = rpki.publication.crl_elt,
-                         uri = self.crl_uri,
+      publisher.withdraw(uri = self.crl_uri,
                          obj = self.latest_crl,
                          repository = repository,
                          handler = handler)
@@ -1188,7 +1185,6 @@ class ca_detail_obj(rpki.sql.sql_persistent):
     child_cert.published = rpki.sundial.now()
     child_cert.sql_store()
     publisher.publish(
-      cls = rpki.publication.certificate_elt,
       uri = child_cert.uri,
       obj = child_cert.cert,
       repository = ca.parent.repository,
@@ -1232,7 +1228,6 @@ class ca_detail_obj(rpki.sql.sql_persistent):
     self.crl_published = rpki.sundial.now()
     self.sql_mark_dirty()
     publisher.publish(
-      cls        = rpki.publication.crl_elt,
       uri        = self.crl_uri,
       obj        = self.latest_crl,
       repository = parent.repository,
@@ -1290,8 +1285,7 @@ class ca_detail_obj(rpki.sql.sql_persistent):
 
     self.manifest_published = rpki.sundial.now()
     self.sql_mark_dirty()
-    publisher.publish(cls = rpki.publication.manifest_elt,
-                      uri = uri,
+    publisher.publish(uri = uri,
                       obj = self.latest_manifest,
                       repository = parent.repository,
                       handler = self.manifest_published_callback)
@@ -1361,8 +1355,7 @@ class ca_detail_obj(rpki.sql.sql_persistent):
          self.crl_published is not None and \
          self.crl_published < stale:
       logger.debug("Retrying publication for %s", self.crl_uri)
-      publisher.publish(cls = rpki.publication.crl_elt,
-                        uri = self.crl_uri,
+      publisher.publish(uri = self.crl_uri,
                         obj = self.latest_crl,
                         repository = repository,
                         handler = self.crl_published_callback)
@@ -1371,8 +1364,7 @@ class ca_detail_obj(rpki.sql.sql_persistent):
          self.manifest_published is not None and \
          self.manifest_published < stale:
       logger.debug("Retrying publication for %s", self.manifest_uri)
-      publisher.publish(cls = rpki.publication.manifest_elt,
-                        uri = self.manifest_uri,
+      publisher.publish(uri = self.manifest_uri,
                         obj = self.latest_manifest,
                         repository = repository,
                         handler = self.manifest_published_callback)
@@ -1386,7 +1378,6 @@ class ca_detail_obj(rpki.sql.sql_persistent):
     for child_cert in self.unpublished_child_certs(stale):
       logger.debug("Retrying publication for %s", child_cert)
       publisher.publish(
-        cls = rpki.publication.certificate_elt,
         uri = child_cert.uri,
         obj = child_cert.cert,
         repository = repository,
@@ -1395,7 +1386,6 @@ class ca_detail_obj(rpki.sql.sql_persistent):
     for roa in self.unpublished_roas(stale):
       logger.debug("Retrying publication for %s", roa)
       publisher.publish(
-        cls = rpki.publication.roa_elt,
         uri = roa.uri,
         obj = roa.roa,
         repository = repository,
@@ -1404,7 +1394,6 @@ class ca_detail_obj(rpki.sql.sql_persistent):
     for ghostbuster in self.unpublished_ghostbusters(stale):
       logger.debug("Retrying publication for %s", ghostbuster)
       publisher.publish(
-        cls = rpki.publication.ghostbuster_elt,
         uri = ghostbuster.uri,
         obj = ghostbuster.ghostbuster,
         repository = repository,
@@ -1492,7 +1481,6 @@ class child_cert_obj(rpki.sql.sql_persistent):
     logger.debug("Revoking %r %r", self, self.uri)
     revoked_cert_obj.revoke(cert = self.cert, ca_detail = ca_detail)
     publisher.withdraw(
-      cls        = rpki.publication.certificate_elt,
       uri        = self.uri,
       obj        = self.cert,
       repository = ca.parent.repository)
@@ -1895,7 +1883,6 @@ class roa_obj(rpki.sql.sql_persistent):
 
     logger.debug("Generating %r URI %s", self, self.uri)
     publisher.publish(
-      cls = rpki.publication.roa_elt,
       uri = self.uri,
       obj = self.roa,
       repository = ca.parent.repository,
@@ -1942,7 +1929,8 @@ class roa_obj(rpki.sql.sql_persistent):
 
     logger.debug("Withdrawing %r %s and revoking its EE cert", self, uri)
     rpki.rpkid.revoked_cert_obj.revoke(cert = cert, ca_detail = ca_detail)
-    publisher.withdraw(cls = rpki.publication.roa_elt, uri = uri, obj = roa,
+    publisher.withdraw(uri = uri,
+                       obj = roa,
                        repository = ca_detail.ca.parent.repository,
                        handler = False if allow_failure else None)
 
@@ -2098,7 +2086,6 @@ class ghostbuster_obj(rpki.sql.sql_persistent):
 
     logger.debug("Generating Ghostbuster record %r", self.uri)
     publisher.publish(
-      cls = rpki.publication.ghostbuster_elt,
       uri = self.uri,
       obj = self.ghostbuster,
       repository = ca.parent.repository,
@@ -2144,7 +2131,8 @@ class ghostbuster_obj(rpki.sql.sql_persistent):
 
     logger.debug("Withdrawing %r %s and revoking its EE cert", self, uri)
     rpki.rpkid.revoked_cert_obj.revoke(cert = cert, ca_detail = ca_detail)
-    publisher.withdraw(cls = rpki.publication.ghostbuster_elt, uri = uri, obj = ghostbuster,
+    publisher.withdraw(uri = uri,
+                       obj = ghostbuster,
                        repository = ca_detail.ca.parent.repository,
                        handler = False if allow_failure else None)
 
@@ -2293,7 +2281,6 @@ class ee_cert_obj(rpki.sql.sql_persistent):
       cert         = cert)
 
     publisher.publish(
-      cls        = rpki.publication.certificate_elt,
       uri        = self.uri,
       obj        = self.cert,
       repository = ca.parent.repository,
@@ -2316,8 +2303,7 @@ class ee_cert_obj(rpki.sql.sql_persistent):
     ca = ca_detail.ca
     logger.debug("Revoking %r %r", self, self.uri)
     revoked_cert_obj.revoke(cert = self.cert, ca_detail = ca_detail)
-    publisher.withdraw(cls = rpki.publication.certificate_elt,
-                       uri = self.uri,
+    publisher.withdraw(uri = self.uri,
                        obj = self.cert,
                        repository = ca.parent.repository)
     self.gctx.sql.sweep()
@@ -2402,7 +2388,6 @@ class ee_cert_obj(rpki.sql.sql_persistent):
     self.sql_mark_dirty()
 
     publisher.publish(
-      cls        = rpki.publication.certificate_elt,
       uri        = self.uri,
       obj        = self.cert,
       repository = ca_detail.ca.parent.repository,
@@ -2457,8 +2442,7 @@ class publication_queue(object):
       self.repositories[rid] = repository
       self.msgs[rid] = rpki.publication.msg.query()
     if self.replace and uri in self.uris:
-      logger.debug("Removing publication duplicate <%s %r %r>",
-                   self.uris[uri].action, self.uris[uri].uri, self.uris[uri].payload)
+      logger.debug("Removing publication duplicate %r", self.uris[uri])
       self.msgs[rid].remove(self.uris.pop(uri))
     pdu = make_pdu(uri = uri, obj = obj)
     if handler is not None:
@@ -2468,11 +2452,11 @@ class publication_queue(object):
     if self.replace:
       self.uris[uri] = pdu
 
-  def publish(self,  cls, uri, obj, repository, handler = None):
-    return self._add(     uri, obj, repository, handler, cls.make_publish)
+  def publish(self,  uri, obj, repository, handler = None):
+    return self._add(uri, obj, repository, handler, rpki.publication.publish_elt.make)
 
-  def withdraw(self, cls, uri, obj, repository, handler = None):
-    return self._add(     uri, obj, repository, handler, cls.make_withdraw)
+  def withdraw(self, uri, obj, repository, handler = None):
+    return self._add(uri, obj, repository, handler, rpki.publication.withdraw_elt.make)
 
   def call_pubd(self, cb, eb):
     def loop(iterator, rid):
diff --git a/rpki/rpkid_tasks.py b/rpki/rpkid_tasks.py
index e0bb6904..49b5b968 100644
--- a/rpki/rpkid_tasks.py
+++ b/rpki/rpkid_tasks.py
@@ -309,7 +309,6 @@ class UpdateChildrenTask(AbstractTask):
                          old_resources.valid_until, irdb_resources.valid_until)
             child_cert.sql_delete()
             self.publisher.withdraw(
-              cls = rpki.publication.certificate_elt,
               uri = child_cert.uri,
               obj = child_cert.cert,
               repository = ca.parent.repository)
diff --git a/rpki/rtr/bgpdump.py b/rpki/rtr/bgpdump.py
index fc3ae9df..5ffabc4d 100755
--- a/rpki/rtr/bgpdump.py
+++ b/rpki/rtr/bgpdump.py
@@ -295,7 +295,7 @@ def bgpdump_server_main(args):
   rpki.rtr.server.read_current = clock.read_current
 
   try:
-    server = rpki.rtr.server.ServerChannel(logger = logger)
+    server = rpki.rtr.server.ServerChannel(logger = logger, refresh = args.refresh, retry = args.retry, expire = args.expire)
     old_serial = server.get_serial()
     logger.debug("[Starting at serial %d (%s)]", old_serial, old_serial)
     while clock:
diff --git a/rpki/sql_schemas.py b/rpki/sql_schemas.py
index 07037970..2cd3bee7 100644
--- a/rpki/sql_schemas.py
+++ b/rpki/sql_schemas.py
@@ -4,33 +4,22 @@
 ## SQL schema rpkid
 rpkid = '''-- $Id: rpkid.sql 5845 2014-05-29 22:31:15Z sra $
 
--- Copyright (C) 2009--2011  Internet Systems Consortium ("ISC")
+-- Copyright (C) 2012--2014  Dragon Research Labs ("DRL")
+-- Portions copyright (C) 2009--2011  Internet Systems Consortium ("ISC")
+-- Portions copyright (C) 2007--2008  American Registry for Internet Numbers ("ARIN")
 --
 -- 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.
+-- copyright notices 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.
-
--- Copyright (C) 2007--2008  American Registry for Internet Numbers ("ARIN")
---
--- 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 ARIN DISCLAIMS ALL WARRANTIES WITH
--- REGARD TO THIS SOFTWARE INCLUDING ALL IMPLIED WARRANTIES OF MERCHANTABILITY
--- AND FITNESS.  IN NO EVENT SHALL ARIN 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.
+-- THE SOFTWARE IS PROVIDED "AS IS" AND DRL, ISC, AND ARIN DISCLAIM ALL
+-- WARRANTIES WITH REGARD TO THIS SOFTWARE INCLUDING ALL IMPLIED
+-- WARRANTIES OF MERCHANTABILITY AND FITNESS.  IN NO EVENT SHALL DRL,
+-- ISC, OR ARIN 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.
 
 -- SQL objects needed by the RPKI engine (rpkid.py).
 
@@ -258,39 +247,26 @@ CREATE TABLE ee_cert (
 ## SQL schema pubd
 pubd = '''-- $Id: pubd.sql 5757 2014-04-05 22:42:12Z sra $
 
--- 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.
-
--- Copyright (C) 2008  American Registry for Internet Numbers ("ARIN")
+-- Copyright (C) 2012--2014  Dragon Research Labs ("DRL")
+-- Portions copyright (C) 2009--2010  Internet Systems Consortium ("ISC")
+-- Portions copyright (C) 2008  American Registry for Internet Numbers ("ARIN")
 --
 -- 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.
+-- copyright notices and this permission notice appear in all copies.
 --
--- THE SOFTWARE IS PROVIDED "AS IS" AND ARIN DISCLAIMS ALL WARRANTIES WITH
--- REGARD TO THIS SOFTWARE INCLUDING ALL IMPLIED WARRANTIES OF MERCHANTABILITY
--- AND FITNESS.  IN NO EVENT SHALL ARIN 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.
+-- THE SOFTWARE IS PROVIDED "AS IS" AND DRL, ISC, AND ARIN DISCLAIM ALL
+-- WARRANTIES WITH REGARD TO THIS SOFTWARE INCLUDING ALL IMPLIED
+-- WARRANTIES OF MERCHANTABILITY AND FITNESS.  IN NO EVENT SHALL DRL,
+-- ISC, OR ARIN 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.
 
 -- SQL objects needed by pubd.py.
 
--- The config table is weird because we're really only using it
--- to store one BPKI CRL, but putting this here lets us use a lot of
--- existing machinery and the alternatives are whacky in other ways.
+-- The config table is weird because it only has one row.
 
 DROP TABLE IF EXISTS client;
 DROP TABLE IF EXISTS config;
diff --git a/rpki/up_down.py b/rpki/up_down.py
index 262003a2..4c2604bf 100644
--- a/rpki/up_down.py
+++ b/rpki/up_down.py
@@ -208,7 +208,7 @@ class class_elt(base_elt):
     elt = self.make_elt("class", "class_name", "cert_url", "resource_set_as",
                         "resource_set_ipv4", "resource_set_ipv6",
                         "resource_set_notafter", "suggested_sia_head")
-    elt.extend([i.toXML() for i in self.certs])
+    elt.extend(i.toXML() for i in self.certs)
     self.make_b64elt(elt, "issuer", self.issuer)
     return elt
 
diff --git a/rpki/xml_utils.py b/rpki/xml_utils.py
index e940d127..1574cd9e 100644
--- a/rpki/xml_utils.py
+++ b/rpki/xml_utils.py
@@ -460,7 +460,7 @@ class msg(list):
     Generate top-level PDU.
     """
     elt = lxml.etree.Element("{%s}msg" % (self.xmlns), nsmap = self.nsmap, version = str(self.version), type = self.type)
-    elt.extend([i.toXML() for i in self])
+    elt.extend(i.toXML() for i in self)
     return elt
 
   @classmethod