Checkpoint. Merge changes from trunk. Add hash-based withdrawal

checks and <list/> command to publication protocol.

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

7 files changed, 88 insertions, 39 deletions

diff --git a/rpki/exceptions.py b/rpki/exceptions.py
index 504c6f28..86c7fa27 100644
--- a/rpki/exceptions.py
+++ b/rpki/exceptions.py
@@ -288,6 +288,16 @@ class NoObjectAtURI(RPKI_Exception):
   No object published at specified URI.
   """
 
+class ExistingObjectAtURI(RPKI_Exception):
+  """
+  An object has already been published at specified URI.
+  """
+
+class DifferentObjectAtURI(RPKI_Exception):
+  """
+  An object with a different hash exists at specified URI.
+  """
+
 class CMSContentNotSet(RPKI_Exception):
   """
   Inner content of a CMS_object has not been set.  If object is known
diff --git a/rpki/pubd.py b/rpki/pubd.py
index 14de1999..0ee4d38c 100644
--- a/rpki/pubd.py
+++ b/rpki/pubd.py
@@ -40,6 +40,8 @@ import rpki.publication
 import rpki.publication_control
 import rpki.daemonize
 
+from lxml.etree import Element, SubElement, ElementTree, Comment
+
 logger = logging.getLogger(__name__)
 
 class main(object):
@@ -107,6 +109,9 @@ class main(object):
 
     self.publication_multimodule = self.cfg.getboolean("publication-multimodule", False)
 
+    self.rrdp_expiration_interval = rpki.sundial.timedelta.parse(self.cfg.get("rrdp-expiration-interval", "6h"))
+    self.rrdp_publication_base = self.cfg.get("rrdp-publication-base", "rrdp-publication/")
+
     self.session = session_obj.fetch(self)
 
     rpki.http.server(
@@ -187,11 +192,6 @@ class session_obj(rpki.sql.sql_persistent):
     "session_id",
     "uuid")
 
-  ## @var expiration_interval
-  # How long to wait after retiring a snapshot before purging it from the database.
-
-  expiration_interval = rpki.sundial.timedelta(hours = 6)
-
   def __repr__(self):
     return rpki.log.log_repr(self, self.uuid, self.serial)
 
@@ -231,7 +231,7 @@ class session_obj(rpki.sql.sql_persistent):
     now = rpki.sundial.now()
     old_snapshot = self.current_snapshot
     if old_snapshot is not None:
-      old_snapshot.expires = now + self.expiration_interval
+      old_snapshot.expires = now + self.gctx.rrdp_expiration_interval
       old_snapshot.sql_store()
     new_snapshot.activated = now
     new_snapshot.sql_store()
@@ -280,40 +280,27 @@ class snapshot_obj(rpki.sql.sql_persistent):
 
     Well, OK, only almost the right properties.  auto-increment
     probably does not back up if we ROLLBACK, which could leave gaps
-    in the sequence.  So may need to rework this.  Ignore for now.
+    in the sequence.  So may need to rework this, eg, to use a serial
+    field in the session object.  Ignore the issue until we have the
+    rest of this working.
     """
 
     return self.snapshot_id
 
-  def publish(self, client, obj, uri):
-
-    # Still a bit confused as to what we should do here.  The
-    # overwrite <publish/> with another <publish/> model doens't
-    # really match the IXFR model.  Current proposal is an attribute
-    # on <publish/> to say that this is an overwrite, haven't
-    # implemented that yet.  Would need to push knowledge of when
-    # we're overwriting all the way from rpkid code that decides to
-    # write each kind of object.  In most cases it looks like we
-    # already know, a priori, might be a few corner cases.
-
-    # Temporary kludge
-    if True:
-      try:
-        self.withdraw(client, uri)
-      except rpki.exceptions.NoObjectAtURI:
-        logger.debug("Withdrew %s", uri)
-      else:
-        logger.debug("No prior %s", uri)
-
+  def publish(self, client, obj, uri, hash):
+    if hash is not None:
+      self.withdraw(client, uri, hash)
+    if object_obj.current_object_at_uri(client, self, uri) is not None:
+      raise rpki.exceptions.ExistingObjectAtURI("Object already published at %s" % uri)
     logger.debug("Publishing %s", uri)
     return object_obj.create(client, self, obj, uri)
 
-  def withdraw(self, client, uri):
-    obj = object_obj.sql_fetch_where1(self.gctx,
-                                      "session_id = %s AND client_id = %s AND withdrawn_snapshot_id IS NULL AND uri = %s",
-                                      (self.session_id, client.client_id, uri))
+  def withdraw(self, client, uri, hash):
+    obj = object_obj.current_object_at_uri(client, self, uri)
     if obj is None:
       raise rpki.exceptions.NoObjectAtURI("No object published at %s" % uri)
+    if obj.hash != hash:
+      raise rpki.exceptions.DifferentObjectAtURI("Found different object at %s (%s, %s)" % (uri, obj.hash, hash))
     logger.debug("Withdrawing %s", uri)
     obj.delete(self)
 
@@ -354,8 +341,8 @@ class object_obj(rpki.sql.sql_persistent):
     self.gctx = snapshot.gctx
     self.uri = uri
     self.payload = obj
-    self.hash = rpki.x509.sha256(obj.get_Base64())
-    logger.debug("Computed hash %s of %r", self.hash.encode("hex"), obj)
+    self.hash = rpki.x509.sha256(obj.get_Base64()).encode("hex")
+    logger.debug("Computed hash %s of %r", self.hash, obj)
     self.published_snapshot_id = snapshot.snapshot_id
     self.withdrawn_snapshot_id = None
     self.session_id = snapshot.session_id
@@ -367,3 +354,9 @@ class object_obj(rpki.sql.sql_persistent):
     self.withdrawn_snapshot_id = snapshot.snapshot_id
     #self.sql_mark_dirty()
     self.sql_store()
+
+  @classmethod
+  def current_object_at_uri(cls, client, snapshot, uri):
+    return cls.sql_fetch_where1(client.gctx,
+                                "session_id = %s AND client_id = %s AND withdrawn_snapshot_id IS NULL AND uri = %s",
+                                (snapshot.session_id, client.client_id, uri))
diff --git a/rpki/publication.py b/rpki/publication.py
index 7b5abaf9..ec088a46 100644
--- a/rpki/publication.py
+++ b/rpki/publication.py
@@ -39,7 +39,6 @@ logger = logging.getLogger(__name__)
 
 
 class publication_namespace(object):
-
   xmlns = "http://www.hactrn.net/uris/rpki/publication-spec/"
   nsmap = { None : xmlns }
 
@@ -103,6 +102,9 @@ class base_publication_elt(rpki.xml_utils.base_elt, publication_namespace):
 
 
 class publish_elt(base_publication_elt):
+  """
+  <publish/> element.
+  """
 
   element_name = "publish"
 
@@ -132,7 +134,7 @@ class publish_elt(base_publication_elt):
     """
 
     logger.info("Publishing %s", self.payload.tracking_data(self.uri))
-    snapshot.publish(self.client, self.payload, self.uri)
+    snapshot.publish(self.client, self.payload, self.uri, self.hash)
     filename = self.uri_to_filename()
     filename_tmp = filename + ".tmp"
     dirname = os.path.dirname(filename)
@@ -144,6 +146,9 @@ class publish_elt(base_publication_elt):
 
 
 class withdraw_elt(base_publication_elt):
+  """
+  <withdraw/> element.
+  """
 
   element_name = "withdraw"
 
@@ -153,7 +158,7 @@ class withdraw_elt(base_publication_elt):
     """
 
     logger.info("Withdrawing %s", self.uri)
-    snapshot.withdraw(self.client, self.uri)
+    snapshot.withdraw(self.client, self.uri, self.hash)
     filename = self.uri_to_filename()
     try:
       os.remove(filename)
@@ -173,6 +178,24 @@ class withdraw_elt(base_publication_elt):
         dirname = os.path.dirname(dirname)
 
 
+class list_elt(base_publication_elt):
+  """
+  <list/> element.
+  """
+
+  def serve_dispatch(self, r_msg, snapshot, cb, eb):
+    """
+    Action dispatch handler.
+    """
+
+    for obj in self.client.published_objects:
+      r_pdu = self.__class__()
+      r_pdu.tag = self.tag
+      r_pdu.uri = obj.uri
+      r_pdu.hash = obj.hash
+      r_msg.append(r_pdu)
+
+
 class report_error_elt(rpki.xml_utils.text_elt, publication_namespace):
   """
   <report_error/> element.
diff --git a/rpki/publication_control.py b/rpki/publication_control.py
index f65fa15d..f1cc5f2c 100644
--- a/rpki/publication_control.py
+++ b/rpki/publication_control.py
@@ -90,6 +90,10 @@ class client_elt(rpki.xml_utils.data_elt, rpki.sql.sql_persistent, publication_c
   def objects(self):
     return rpki.pubd.object_obj.sql_fetch_where(self.gctx, "client_id = %s", (self.client_id,))
 
+  @property
+  def published_object(self):
+    return rpki.pubd.object_obj.sql_fetch_where(self.gctx, "client_id = %s AND withdrawn_snapshot_id IS NULL", (self.client_id,))
+
   def serve_post_save_hook(self, q_pdu, r_pdu, cb, eb):
     """
     Extra server actions for client_elt.
diff --git a/rpki/relaxng.py b/rpki/relaxng.py
index 4e8e9242..93ac16fe 100644
--- a/rpki/relaxng.py
+++ b/rpki/relaxng.py
@@ -1837,12 +1837,14 @@ publication = lxml.etree.RelaxNG(lxml.etree.fromstring(r'''<?xml version="1.0" e
     <choice>
       <ref name="publish_query"/>
       <ref name="withdraw_query"/>
+      <ref name="list_query"/>
     </choice>
   </define>
   <define name="reply_elt">
     <choice>
       <ref name="publish_reply"/>
       <ref name="withdraw_reply"/>
+      <ref name="list_reply"/>
       <ref name="report_error_reply"/>
     </choice>
   </define>
@@ -1919,6 +1921,23 @@ publication = lxml.etree.RelaxNG(lxml.etree.fromstring(r'''<?xml version="1.0" e
       <ref name="uri"/>
     </element>
   </define>
+  <!-- <list/> element -->
+  <define name="list_query">
+    <element name="list">
+      <optional>
+        <ref name="tag"/>
+      </optional>
+    </element>
+  </define>
+  <define name="list_reply">
+    <element name="list">
+      <optional>
+        <ref name="tag"/>
+      </optional>
+      <ref name="uri"/>
+      <ref name="hash"/>
+    </element>
+  </define>
   <!-- <report_error/> element -->
   <define name="report_error_reply">
     <element name="report_error">
diff --git a/rpki/rtr/server.py b/rpki/rtr/server.py
index b3e4fd7c..1c7a5e78 100644
--- a/rpki/rtr/server.py
+++ b/rpki/rtr/server.py
@@ -324,7 +324,7 @@ class ServerChannel(rpki.rtr.channels.PDUChannel):
     old_serial = self.current_serial
     return old_serial != self.get_serial()
 
-  def notify(self, data = None):
+  def notify(self, data = None, force = False):
     """
     Cronjob instance kicked us: check whether our serial number has
     changed, and send a notify message if so.
@@ -335,7 +335,7 @@ class ServerChannel(rpki.rtr.channels.PDUChannel):
     whether we care about a particular change set or not.
     """
 
-    if self.check_serial():
+    if force or self.check_serial():
       self.push_pdu(SerialNotifyPDU(version = self.version,
                                     serial  = self.current_serial,
                                     nonce   = self.current_nonce))
diff --git a/rpki/sql_schemas.py b/rpki/sql_schemas.py
index 7c7079c0..b28c8231 100644
--- a/rpki/sql_schemas.py
+++ b/rpki/sql_schemas.py
@@ -309,7 +309,7 @@ CREATE TABLE snapshot (
 CREATE TABLE object (
         object_id               SERIAL NOT NULL,
         uri                     VARCHAR(255) NOT NULL,
-        hash                    BINARY(32) NOT NULL,
+        hash                    CHAR(64) NOT NULL,
         payload                 LONGBLOB NOT NULL,
         published_snapshot_id   BIGINT UNSIGNED,
         withdrawn_snapshot_id   BIGINT UNSIGNED,