Combining PublisherStore and FileStore in a single Repository. Also passing config options by reference as much as possible.

src/bin/pubc.rs

@@ -20,7 +20,7 @@ fn main() {
         }
     };
 
-    let mut client = match PubClient::new(config.state_dir().clone()) {
+    let mut client = match PubClient::new(config.state_dir()) {
         Ok(client) => client,
         Err(e) => {
             eprintln!("{}", e);

src/provisioning/mod.rs

@@ -1,3 +1 @@
-pub mod info;
-pub mod publisher;
-pub mod publisher_list;
+pub mod info;

src/pubc/client.rs

@@ -71,7 +71,7 @@ pub struct PubClient {
 
 impl PubClient {
     /// Creates a new publication client
-    pub fn new(work_dir: PathBuf) -> Result<Self, Error> {
+    pub fn new(work_dir: &PathBuf) -> Result<Self, Error> {
         let store = CachingDiskKeyStore::new(work_dir.clone())?;
         let signer = OpenSslSigner::new(work_dir)?;
         Ok(
@@ -229,9 +229,9 @@ mod tests {
         let service = test::http_uri("http://host/publish");
         let notify = test::http_uri("http://host/notify.xml");
         PubServer::new(
-            work_dir.clone(),
-            xml_dir.clone(),
-            uri,
+            work_dir,
+            xml_dir,
+            &uri,
             service,
             notify
         ).unwrap()
@@ -241,13 +241,13 @@ mod tests {
     fn should_initialise_keep_state_and_reinitialise() {
         test::test_with_tmp_dir(|d| {
             // Set up a new client and initialise
-            let mut client_1 = PubClient::new(d.clone()).unwrap();
+            let mut client_1 = PubClient::new(&d).unwrap();
             client_1.init("client".to_string()).unwrap();
             let pr_1 = client_1.publisher_request().unwrap();
 
             // Prove that a client starting from an initialised dir
             // comes up with the same state.
-            let mut client_2 = PubClient::new(d.clone()).unwrap();
+            let mut client_2 = PubClient::new(&d).unwrap();
             let pr_2 = client_2.publisher_request().unwrap();
             assert_eq!(pr_1.handle(), pr_2.handle());
             assert_eq!(pr_1.id_cert().to_bytes(), pr_2.id_cert().to_bytes());
@@ -268,7 +268,7 @@ mod tests {
             let xml_dir = test::create_sub_dir(&d);
 
             let alice_dir = test::create_sub_dir(&d);
-            let mut alice = PubClient::new(alice_dir).unwrap();
+            let mut alice = PubClient::new(&alice_dir).unwrap();
             alice.init("alice".to_string()).unwrap();
             let pr_alice = alice.publisher_request().unwrap();
 

src/pubd/config.rs

@@ -47,9 +47,7 @@ impl Config {
         &self.pub_xml_dir
     }
 
-    pub fn rsync_base(&self) -> uri::Rsync {
-        self.rsync_base.clone()
-    }
+    pub fn rsync_base(&self) -> &uri::Rsync { &self.rsync_base }
 
     pub fn service_uri(&self) -> uri::Http {
         self.service_uri.clone()

src/pubd/daemon.rs

@@ -17,8 +17,8 @@ const HTML_404: &'static [u8] = include_bytes!("../../static/html/404.html");
 pub fn serve(config: &Config) {
 
     let mut pub_server = match PubServer::new(
-        config.data_dir().clone(),
-        config.pub_xml_dir().clone(),
+        config.data_dir(),
+        config.pub_xml_dir(),
         config.rsync_base(),
         config.service_uri(),
         config.notify_sia()

src/pubd/server.rs

@@ -3,18 +3,17 @@
 use std::path::PathBuf;
 use std::sync::Arc;
 use provisioning::info::MyIdentity;
-use provisioning::publisher_list;
-use provisioning::publisher_list::PublisherList;
+use repo::publisher::Publisher;
+use repo::publisher_store;
+use repo::publisher_store::PublisherStore;
 use rpki::uri;
 use rpki::signing::PublicKeyAlgorithm;
 use rpki::signing::builder::IdCertBuilder;
 use rpki::signing::signer::{CreateKeyError, KeyUseError, Signer};
-use signing::softsigner;
-use signing::softsigner::OpenSslSigner;
+use rpki::oob::exchange::RepositoryResponse;
 use storage::caching_ks::CachingDiskKeyStore;
 use storage::keystore::{self, Info, Key, KeyStore};
-use provisioning::publisher::Publisher;
-use rpki::oob::exchange::RepositoryResponse;
+use signing::softsigner::{self, OpenSslSigner};
 
 
 /// # Naming things in the keystore.
@@ -43,7 +42,7 @@ pub struct PubServer {
     store: CachingDiskKeyStore,
     //   my_id -> MyIdentity
 
-    publisher_list: PublisherList,
+    publisher_list: PublisherStore,
 
     service_uri: uri::Http,
     notify_sia: uri::Http
@@ -54,9 +53,9 @@ impl PubServer {
     /// Creates a new publication server. Note that state is preserved
     /// on disk in the work_dir provided.
     pub fn new(
-        work_dir: PathBuf,
-        pub_xml_dir: PathBuf,
-        base_uri: uri::Rsync,
+        work_dir: &PathBuf,
+        pub_xml_dir: &PathBuf,
+        base_uri: &uri::Rsync,
         service_uri: uri::Http,
         rrdp_notification_uri: uri::Http
     ) -> Result<Self, Error> {
@@ -109,14 +108,14 @@ impl PubServer {
     /// Synchronize publishers from disk
     fn init_publishers(
         work_dir: &PathBuf,
-        pub_xml_dir: PathBuf,
-        base_uri: uri::Rsync
-    ) -> Result<PublisherList, Error> {
-        let mut publisher_list = PublisherList::new(
-            work_dir.clone(),
+        pub_xml_dir: &PathBuf,
+        base_uri: &uri::Rsync
+    ) -> Result<PublisherStore, Error> {
+        let mut publisher_store = PublisherStore::new(
+            work_dir,
             base_uri)?;
-        publisher_list.sync_from_dir(pub_xml_dir, actor())?;
-        Ok(publisher_list)
+        publisher_store.sync_from_dir(pub_xml_dir, actor())?;
+        Ok(publisher_store)
     }
 
     /// Returns all currently configured publishers.
@@ -178,7 +177,7 @@ pub enum Error {
     KeyStoreError(keystore::Error),
 
     #[fail(display="{}", _0)]
-    PublisherListError(publisher_list::Error),
+    PublisherListError(publisher_store::Error),
 
     #[fail(display="{}", _0)]
     SoftSignerError(softsigner::Error),
@@ -203,8 +202,8 @@ impl From<keystore::Error> for Error {
     }
 }
 
-impl From<publisher_list::Error> for Error {
-    fn from(e: publisher_list::Error) -> Self {
+impl From<publisher_store::Error> for Error {
+    fn from(e: publisher_store::Error) -> Self {
         Error::PublisherListError(e)
     }
 }
@@ -241,9 +240,9 @@ mod tests {
         let service = test::http_uri("http://host/publish");
         let notify = test::http_uri("http://host/notify.xml");
         PubServer::new(
-            work_dir.clone(),
-            xml_dir.clone(),
-            uri,
+            work_dir,
+            xml_dir,
+            &uri,
             service,
             notify
         ).unwrap()
@@ -278,12 +277,12 @@ mod tests {
             let xml_dir = test::create_sub_dir(&d);
 
             let alice_dir = test::create_sub_dir(&d);
-            let mut alice = PubClient::new(alice_dir).unwrap();
+            let mut alice = PubClient::new(&alice_dir).unwrap();
             alice.init("alice".to_string()).unwrap();
             let pr_alice = alice.publisher_request().unwrap();
 
             let bob_dir = test::create_sub_dir(&d);
-            let mut bob = PubClient::new(bob_dir).unwrap();
+            let mut bob = PubClient::new(&bob_dir).unwrap();
             bob.init("bob".to_string()).unwrap();
             let pr_bob = bob.publisher_request().unwrap();
 
@@ -324,7 +323,7 @@ mod tests {
             let pr_alice = alice.publisher_request().unwrap();
 
             let carol_dir = test::create_sub_dir(&d);
-            let mut carol = PubClient::new(carol_dir).unwrap();
+            let mut carol = PubClient::new(&carol_dir).unwrap();
             carol.init("carol".to_string()).unwrap();
             let pr_carol = carol.publisher_request().unwrap();
 
@@ -365,9 +364,9 @@ mod tests {
 
             assert!(
                 PubServer::new(
-                    d.clone(),
-                    xml_dir.clone(),
-                    uri,
+                    &d,
+                    &xml_dir,
+                    &uri,
                     service,
                     notify
                 ).is_err()
@@ -381,12 +380,12 @@ mod tests {
             let xml_dir = test::create_sub_dir(&d);
 
             let alice_dir = test::create_sub_dir(&d);
-            let mut alice = PubClient::new(alice_dir).unwrap();
+            let mut alice = PubClient::new(&alice_dir).unwrap();
             alice.init("alice".to_string()).unwrap();
             let pr_alice = alice.publisher_request().unwrap();
 
             let bob_dir = test::create_sub_dir(&d);
-            let mut bob = PubClient::new(bob_dir).unwrap();
+            let mut bob = PubClient::new(&bob_dir).unwrap();
             bob.init("bob".to_string()).unwrap();
             let pr_bob = bob.publisher_request().unwrap();
 

src/repo/file.rs

@@ -0,0 +1,70 @@
+use ext_serde;
+use rpki::uri;
+use bytes::Bytes;
+use rpki::publication;
+use rpki::publication::query::{ Publish, PublishElement, Update, Withdraw };
+
+#[derive(Clone, Debug, Deserialize, Serialize)]
+pub struct CurrentFile {
+    #[serde(
+    deserialize_with = "ext_serde::de_rsync_uri",
+    serialize_with = "ext_serde::ser_rsync_uri")]
+    uri:     uri::Rsync,
+
+    #[serde(
+    deserialize_with = "ext_serde::de_bytes",
+    serialize_with = "ext_serde::ser_bytes")]
+    /// The actual file content. Note that we may want to store this
+    /// only on disk in future (look up by sha256 hash), to save memory.
+    content: Bytes,
+
+    #[serde(
+    deserialize_with = "ext_serde::de_bytes",
+    serialize_with = "ext_serde::ser_bytes")]
+    /// The sha-256 hash of the file (as is used on the RPKI manifests and
+    /// in the publication protocol for list, update and withdraw). Saving
+    /// this rather than calculating on demand seems a small price for some
+    /// performance gain.
+    hash:    Bytes
+}
+
+impl CurrentFile {
+    pub fn new(uri: uri::Rsync, content: Bytes) -> Self {
+        let hash = publication::hash(&content);
+        CurrentFile {uri, content, hash}
+    }
+
+    pub fn uri(&self) -> &uri::Rsync {
+        &self.uri
+    }
+
+    pub fn content(&self) -> &Bytes {
+        &self.content
+    }
+
+    pub fn hash(&self) -> &Bytes {
+        &self.hash
+    }
+
+    pub fn as_publish(&self) -> PublishElement {
+        Publish::publish(&self.content, self.uri.clone())
+    }
+
+    pub fn as_update(&self, old_content: &Bytes) -> PublishElement {
+        Update::publish(old_content, &self.content, self.uri.clone())
+    }
+
+    pub fn as_withdraw(&self) -> PublishElement {
+        Withdraw::publish(&self.content, self.uri.clone())
+    }
+}
+
+impl PartialEq for CurrentFile {
+    fn eq(&self, other: &CurrentFile) -> bool {
+        self.uri == other.uri &&
+            self.hash == other.hash &&
+            self.content == other.content
+    }
+}
+
+impl Eq for CurrentFile {}

src/repo/file_store.rs

@@ -0,0 +1,363 @@
+use std::io;
+use std::io::{Read, Write};
+use std::fs;
+use std::fs::File;
+use std::path::PathBuf;
+use bytes::Bytes;
+use repo::file::CurrentFile;
+use rpki::publication::query::PublishElement;
+use rpki::publication::query::PublishQuery;
+use rpki::uri;
+
+#[derive(Clone, Debug)]
+pub struct FileStore {
+    base_dir: PathBuf
+}
+
+/// # Construct
+///
+impl FileStore {
+    pub fn new(work_dir: &PathBuf) -> Result<Self, Error> {
+        let mut rsync_dir = PathBuf::from(work_dir);
+        rsync_dir.push("rsync");
+        if ! rsync_dir.is_dir() {
+            fs::create_dir_all(&rsync_dir)?;
+        }
+        Ok ( FileStore { base_dir: rsync_dir } )
+    }
+}
+
+/// # Publishing
+///
+impl FileStore {
+    /// Process a PublishQuery update
+    pub fn update(
+        &self,
+        update: &PublishQuery,
+        base_uri: &uri::Rsync
+    ) -> Result<(), Error> {
+        self.verify_query(update, base_uri)?;
+        self.update_files(update)?;
+
+        Ok(())
+    }
+
+    pub fn list(
+        &self,
+        base_uri: &uri::Rsync
+    ) -> Result<Vec<CurrentFile>, Error> {
+        let path = self.file_path(base_uri);
+        self.recurse_disk(&path)
+    }
+
+    fn recurse_disk(
+        &self,
+        path: &PathBuf
+    ) -> Result<Vec<CurrentFile>, Error> {
+        let mut res = Vec::new();
+        for entry in fs::read_dir(path)? {
+            let entry = entry?;
+            let path = entry.path();
+            if path.is_dir() {
+                let mut other = self.recurse_disk(&path)?;
+                res.append(& mut other);
+            } else {
+                if let Some(file) = self.read_file(&path)? {
+                    res.push(file);
+                }
+            }
+        }
+
+        Ok(res)
+    }
+
+
+
+    /// Assert that all updates are confined to the given base_uri; i.e. do
+    /// not allow publishers to update things outside of their own jail.
+    fn verify_query(
+        &self,
+        update: &PublishQuery,
+        base_uri: &uri::Rsync
+    ) -> Result<(), Error> {
+        for q in update.elements() {
+            match q {
+                PublishElement::Publish(p) => {
+                    Self::assert_uri(base_uri, p.uri())?;
+                    if self.get_current_file_opt(p.uri())?.is_some() {
+                        return Err(Error::PublishWrongUri(p.uri().clone()))
+                    }
+                },
+                PublishElement::Update(u) => {
+                    Self::assert_uri(base_uri, u.uri())?;
+                    if let Some(cur) = self.get_current_file_opt(u.uri())? {
+                        if cur.hash() != u.hash() {
+                            return Err(Error::UpdateWrongHash)
+                        }
+                    } else {
+                        return Err(Error::UpdateWrongUri(u.uri().clone()))
+                    }
+                },
+                PublishElement::Withdraw(w) => {
+                    Self::assert_uri(base_uri, w.uri())?;
+                    if let Some(cur) = self.get_current_file_opt(w.uri())? {
+                        if cur.hash() != w.hash() {
+                            return Err(Error::WithdrawWrongHash)
+                        }
+                    } else {
+                        return Err(Error::UpdateWrongUri(w.uri().clone()))
+                    }
+                },
+            }
+        }
+
+        Ok(())
+    }
+
+    /// Perform the actual updates on disk. This assumes that the updates
+    /// have been verified.
+    fn update_files(
+        &self,
+        update: &PublishQuery
+    ) -> Result<(), Error> {
+        for q in update.elements() {
+            match q {
+                PublishElement::Publish(p) => {
+                    self.save_file(p.uri(), p.object())?;
+                },
+                PublishElement::Update(u) => {
+                    self.save_file(u.uri(), u.object())?;
+                },
+                PublishElement::Withdraw(w) => {
+                    self.delete_file(w.uri())?;
+                },
+            }
+        }
+        Ok(())
+    }
+
+    fn assert_uri(base: &uri::Rsync, file: &uri::Rsync) -> Result<(), Error> {
+        if base.module() == file.module() &&
+           file.path().starts_with(base.path()) {
+            Ok(())
+        } else {
+            Err(Error::OutsideBaseUri)
+        }
+    }
+
+    fn file_path(&self, file_uri: &uri::Rsync) -> PathBuf {
+        let mut path = self.base_dir.clone();
+        let module = file_uri.module();
+        path.push(PathBuf::from(module.authority()));
+        path.push(PathBuf::from(module.module()));
+        path.push(PathBuf::from(file_uri.path()));
+        path
+    }
+
+    /// Resolves a path on disk to an rsync uri (i.e. relative to base)
+    fn file_uri(&self, path: &PathBuf) -> Result<uri::Rsync, Error> {
+        let base_string = self.base_dir.to_string_lossy().to_string();
+        let mut path_string = path.to_string_lossy().to_string();
+
+        if path_string.as_str().starts_with(base_string.as_str()) {
+            let rel = path_string.split_off(base_string.len());
+            let uri = format!("rsync:/{}", rel);
+            let uri = uri::Rsync::from_string(uri)?;
+            Ok(uri)
+        } else {
+            panic!("This is a bug: we are looking for a file outside of our\
+             base directory")
+        }
+    }
+
+    fn save_file(
+        &self,
+        file_uri: &uri::Rsync,
+        content: &Bytes
+    ) -> Result<(), Error> {
+        let path = self.file_path(file_uri);
+
+        if let Some(parent) = path.parent() {
+            fs::create_dir_all(parent)?;
+        }
+
+        let mut f = File::create(path)?;
+        f.write(content)?;
+        Ok(())
+    }
+
+    fn delete_file(
+        &self,
+        file_uri: &uri::Rsync
+    ) -> Result<(), Error> {
+        let path = self.file_path(file_uri);
+        fs::remove_file(path)?;
+        Ok(())
+    }
+
+    fn read_file(
+        &self,
+        path: &PathBuf
+    ) -> Result<Option<CurrentFile>, Error> {
+        match File::open(path) {
+            Err(_)  => Ok(None),
+            Ok(mut f) => {
+                let mut bytes = Vec::new();
+                f.read_to_end(&mut bytes)?;
+                let content = Bytes::from(bytes);
+                Ok(Some(
+                    CurrentFile::new(
+                        self.file_uri(path)?,
+                        content
+                )))
+            }
+        }
+    }
+
+    fn get_current_file_opt(
+        &self,
+        file_uri: &uri::Rsync
+    ) -> Result<Option<CurrentFile>, Error> {
+        let path = self.file_path(file_uri);
+        self.read_file(&path)
+    }
+}
+
+
+//------------ Error ---------------------------------------------------------
+
+#[derive(Debug, Fail)]
+pub enum Error {
+    #[fail(display="{}", _0)]
+    IoError(io::Error),
+
+    #[fail(display="{}", _0)]
+    UriError(uri::Error),
+
+    #[fail(display="File already exists for uri (use update!): {}", _0)]
+    PublishWrongUri(uri::Rsync),
+
+    #[fail(display="File sent for update has no entry for uri: {}", _0)]
+    UpdateWrongUri(uri::Rsync),
+
+    #[fail(display="File for update exists, but hash does not match")]
+    UpdateWrongHash,
+
+    #[fail(display="The withdraw URI is not known: {}", _0)]
+    WithdrawWrongUri(uri::Rsync),
+
+    #[fail(display="File for withdraw exists, but hash does not match")]
+    WithdrawWrongHash,
+
+    #[fail(display="Publishing outside of base URI is not allowed.")]
+    OutsideBaseUri,
+}
+
+impl From<io::Error> for Error {
+    fn from(e: io::Error) -> Self {
+        Error::IoError(e)
+    }
+}
+
+impl From<uri::Error> for Error {
+    fn from(e: uri::Error) -> Self {
+        Error::UriError(e)
+    }
+}
+
+//------------ Tests ---------------------------------------------------------
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use bytes::Bytes;
+    use test;
+
+    #[test]
+    fn should_store_list_withdraw_files() {
+        test::test_with_tmp_dir(|d| {
+            let file_store = FileStore { base_dir: d };
+
+            // Using a port here to make sure that it works in mapping
+            // the rsync URI to and from disk.
+            let base_uri = test::rsync_uri
+                ("rsync://host:10873/module/alice/");
+
+            // Publish a file
+            let file = CurrentFile::new(
+                test::rsync_uri("rsync://host:10873/module/alice/file.txt"),
+                Bytes::from("example content")
+            );
+
+            let mut builder = PublishQuery::build();
+            builder.add(file.clone().as_publish());
+            let message = builder.build_message();
+            let publish = message.as_query().unwrap().as_publish().unwrap();
+
+            file_store.update(&publish, &base_uri).unwrap();
+
+            // See that it's the only one listed
+            let files = file_store.list(&base_uri).unwrap();
+            assert_eq!(1, files.len());
+            assert!(files.contains(&file));
+
+            // Update a file
+            let file_update = CurrentFile::new(
+                file.uri().clone(),
+                Bytes::from("example updated content")
+            );
+
+            let mut builder = PublishQuery::build();
+            builder.add(file_update.clone().as_update(file.content()));
+            let message = builder.build_message();
+            let publish = message.as_query().unwrap().as_publish().unwrap();
+            file_store.update(&publish, &base_uri).unwrap();
+
+            // See that it's the only one listed
+            let files = file_store.list(&base_uri).unwrap();
+            assert_eq!(1, files.len());
+            assert!(files.contains(&file_update));
+
+            // Withdraw a file
+            let mut builder = PublishQuery::build();
+            builder.add(file_update.as_withdraw());
+            let message = builder.build_message();
+            let publish = message.as_query().unwrap().as_publish().unwrap();
+            file_store.update(&publish, &base_uri).unwrap();
+
+            // See that there are no files listed
+            let files = file_store.list(&base_uri).unwrap();
+            assert_eq!(0, files.len());
+        });
+    }
+
+    #[test]
+    fn should_not_allow_publishing_or_withdrawing_outside_of_base() {
+        test::test_with_tmp_dir(|d| {
+            let file_store = FileStore { base_dir: d };
+
+            // Using a port here to make sure that it works in mapping
+            // the rsync URI to and from disk.
+            let base_uri = test::rsync_uri
+                ("rsync://host:10873/module/alice/");
+
+            // Publish a file
+            let file = CurrentFile::new(
+                test::rsync_uri("rsync://host:10873/module/bob/file.txt"),
+                Bytes::from("example content")
+            );
+
+            let mut builder = PublishQuery::build();
+            builder.add(file.clone().as_publish());
+            let message = builder.build_message();
+            let publish = message.as_query().unwrap().as_publish().unwrap();
+
+            match file_store.update(&publish, &base_uri) {
+                Err(Error::OutsideBaseUri) => {},
+                _ => { panic!("Expected Error::OutsideBaseUri") }
+            }
+        });
+    }
+
+
+}

src/repo/mod.rs

@@ -1 +1,5 @@
+pub mod file;
+pub mod file_store;
+pub mod publisher;
+pub mod publisher_store;
 pub mod repository;

src/repo/publisher_store.rs

@@ -1,18 +1,17 @@
 //! Responsible for storing and retrieving Publisher information.
+use std::collections::HashMap;
 use std::fs;
+use std::fs::File;
 use std::io;
+use std::io::BufReader;
 use std::path::PathBuf;
-use provisioning::publisher::Publisher;
+use std::sync::Arc;
+use repo::publisher::Publisher;
 use rpki::remote::idcert::IdCert;
 use rpki::uri;
-use rpki::oob::exchange::PublisherRequest;
+use rpki::oob::exchange::{PublisherRequest, PublisherRequestError};
 use storage::keystore::{self, Info, Key, KeyStore};
 use storage::caching_ks::CachingDiskKeyStore;
-use std::sync::Arc;
-use std::fs::File;
-use std::io::BufReader;
-use std::collections::HashMap;
-use rpki::oob::exchange::PublisherRequestError;
 
 
 //------------ PublisherList -------------------------------------------------
@@ -22,16 +21,16 @@ use rpki::oob::exchange::PublisherRequestError;
 /// contains all current publishers, and keeps a full audit trail of changes
 /// to this.
 #[derive(Clone, Debug)]
-pub struct PublisherList {
+pub struct PublisherStore {
     store: CachingDiskKeyStore,
     base_uri: uri::Rsync,
 }
 
 
-impl PublisherList {
+impl PublisherStore {
     pub fn new(
-        work_dir: PathBuf,
-        base_uri: uri::Rsync
+        work_dir: &PathBuf,
+        base_uri: &uri::Rsync
     ) -> Result<Self, Error> {
         let mut publisher_dir = PathBuf::from(work_dir);
         publisher_dir.push("publishers");
@@ -40,9 +39,9 @@ impl PublisherList {
         }
 
         Ok(
-            PublisherList {
+            PublisherStore {
                 store: CachingDiskKeyStore::new(publisher_dir)?,
-                base_uri
+                base_uri: base_uri.clone()
             }
         )
     }
@@ -175,14 +174,14 @@ impl PublisherList {
 
 
 /// # Initialise from disk
-impl PublisherList {
+impl PublisherStore {
     /// Synchronizes the list of Publisher based on request XML files on disk.
     /// Will add new publishers, remove removed publisher, and update the
     /// id_cert in case it was updated. Returns an error in case duplicate
     /// handler names are found in XML files in the directory.
     pub fn sync_from_dir(
         &mut self,
-        dir: PathBuf,
+        dir: &PathBuf,
         actor: String
     ) -> Result<(), Error> {
         // Find all the publisher requests on disk
@@ -226,7 +225,7 @@ impl PublisherList {
 
     fn prs_on_disk(
         &self,
-        dir: PathBuf
+        dir: &PathBuf
     ) -> Result<HashMap<String, PublisherRequest>, Error> {
         let mut prs_on_disk = HashMap::new();
         for e in dir.read_dir()? {
@@ -317,9 +316,9 @@ mod tests {
     use super::*;
     use test;
 
-    fn new_pl(dir: PathBuf) -> PublisherList {
+    fn new_pl(dir: &PathBuf) -> PublisherStore {
         let uri = test::rsync_uri("rsync://host/module/");
-        PublisherList::new(dir, uri).unwrap()
+        PublisherStore::new(dir, &uri).unwrap()
     }
 
     fn find_in_list(
@@ -332,7 +331,7 @@ mod tests {
     #[test]
     fn should_refuse_slash_in_publisher_handle() {
         test::test_with_tmp_dir(|d| {
-            let mut pl = new_pl(d);
+            let mut pl = new_pl(&d);
             let pr = test::new_publisher_request("test/below");
 
             match pl.add_publisher(pr, "test".to_string()) {
@@ -345,7 +344,7 @@ mod tests {
     #[test]
     fn should_add_publisher() {
         test::test_with_tmp_dir(|d| {
-            let mut pl = new_pl(d);
+            let mut pl = new_pl(&d);
             let name = "alice";
             let pr = test::new_publisher_request(name);
             let id_cert = pr.id_cert().clone();
@@ -372,7 +371,7 @@ mod tests {
     #[test]
     fn should_update_id_cert_publisher() {
         test::test_with_tmp_dir(|d| {
-            let mut pl = new_pl(d);
+            let mut pl = new_pl(&d);
             let name = "alice";
             let pr = test::new_publisher_request(name);
             let actor = "test".to_string();
@@ -406,7 +405,7 @@ mod tests {
     #[test]
     fn should_remove_publisher() {
         test::test_with_tmp_dir(|d| {
-            let mut pl = new_pl(d);
+            let mut pl = new_pl(&d);
 
             let name = "alice";
             let actor = "test".to_string();
@@ -426,7 +425,7 @@ mod tests {
         test::test_with_tmp_dir(|d|{
 
             let pl_dir = test::create_sub_dir(&d);
-            let mut pl = new_pl(pl_dir);
+            let mut pl = new_pl(&pl_dir);
 
             let actor = "test".to_string();
 
@@ -447,7 +446,7 @@ mod tests {
             );
 
             pl.sync_from_dir(
-                PathBuf::from(start_sync_dir),
+                &PathBuf::from(start_sync_dir),
                 actor.clone()
             ).unwrap();
 
@@ -476,7 +475,7 @@ mod tests {
                 &pr_carol.encode_vec()
             );
             pl.sync_from_dir(
-                PathBuf::from(updated_sync_dir),
+                &PathBuf::from(updated_sync_dir),
                 actor.clone()
             ).unwrap();
 
@@ -506,7 +505,7 @@ mod tests {
                 &pr_bob_2.encode_vec()
             );
             assert!(pl.sync_from_dir(
-                PathBuf::from(duplicates_sync_dir),
+                &PathBuf::from(duplicates_sync_dir),
                 actor.clone()
             ).is_err());
         })

src/repo/repository.rs

@@ -1,493 +1,58 @@
-use std::fs;
-use std::io;
 use std::path::PathBuf;
-use std::sync::Arc;
-use bytes::Bytes;
-use ext_serde;
 use rpki::uri;
-use rpki::publication;
-use storage::caching_ks::CachingDiskKeyStore;
-use storage::keystore::{ self, Info, Key, KeyStore };
-use rpki::publication::query::PublishQuery;
-use rpki::publication::query::PublishElement;
-use rpki::publication::query::Update;
-use rpki::publication::query::Publish;
-use rpki::publication::query::Withdraw;
+use repo::publisher_store::PublisherStore;
+use repo::file_store::FileStore;
+use repo::publisher_store;
+use repo::file_store;
 
 
-/// # Naming things in the keystore.
-fn actor() -> String {
-    "publication server".to_string()
-}
-
 //------------ Repository ----------------------------------------------------
 
-/// This type stores all files for each configured publisher, and makes them
-/// available to relying parties through RRDP, and by storing the files on
-/// disk in a folder that may be exposed by an rsync daemon.
+/// This type orchestrates the management of publishers that are allowed to
+/// publish, as well as making the published content available (1) on disk
+/// in a format that lends itself to being exposed by rsyncd, and (2)
+/// include it in notification, snapshot and delta filed for RRDP.
+#[derive(Clone, Debug)]
 pub struct Repository {
-    store: CachingDiskKeyStore
+    // publisher_store
+    ps: PublisherStore,
+    // file_store
+    fs: FileStore
+
+    // XXX TODO: rrdp..
 }
 
 /// # Construct
 ///
 impl Repository {
-    pub fn new(work_dir: &PathBuf) -> Result<Self, Error> {
-        let mut repo_data_dir = PathBuf::from(work_dir);
-        repo_data_dir.push("repo_data");
-        if ! repo_data_dir.is_dir() {
-            fs::create_dir_all(&repo_data_dir)?;
-        }
+    pub fn new(work_dir: &PathBuf, base_uri: &uri::Rsync) -> Result<Self, Error> {
+        let ps = PublisherStore::new(work_dir, base_uri)?;
+        let fs = FileStore::new(work_dir)?;
 
-        let store = CachingDiskKeyStore::new(repo_data_dir)?;
-        Ok( Repository { store } )
+        Ok( Repository { ps, fs } )
     }
 }
 
-/// # Publishers
-///
-impl Repository {
-    pub fn add_publisher(
-        &mut self,
-        repo_publisher: RepoPublisher
-    ) -> Result<(), Error> {
-        let key = Key::from_str(repo_publisher.publisher_handle.as_str());
-        let inf = Info::now(
-            actor(),
-            format!("Added publisher: {}", repo_publisher.publisher_handle.as_str())
-        );
-        self.store.store(key, repo_publisher, inf)?;
-        Ok(())
-    }
-
-    pub fn remove_publisher(
-        &mut self,
-        publisher_handle: &str
-    ) -> Result<(), Error> {
-        let key = Key::from_str(publisher_handle);
-        let inf = Info::now(
-            actor(),
-            format!("Removed publisher: {}", publisher_handle)
-        );
-        self.store.archive(&key, inf)?;
-        Ok(())
-    }
-
-    pub fn publishers(&self) -> Result<Vec<Arc<RepoPublisher>>, Error> {
-        let mut res = Vec::new();
-
-        for ref key in self.store.keys() {
-            if let Some(arc) = self.store.get(key)? {
-                res.push(arc);
-            }
-        }
-
-        Ok(res)
-    }
-
-    pub fn get_publisher(
-        &self,
-        publisher_handle: &str
-    ) -> Result<Arc<RepoPublisher>, Error> {
-        let key = Key::from_str(publisher_handle);
-        if let Some(repo_publisher) = self.store.get(&key)? {
-            Ok(repo_publisher)
-        } else {
-            Err(Error::UnknownPublisher(publisher_handle.to_string()))
-        }
-    }
-}
-
-/// # Publish / Withdraw
-///
-impl Repository {
-    pub fn publish(
-        &mut self,
-        publisher_handle: &str,
-        query: PublishQuery
-    ) -> Result<(), Error> {
-        let rp = self.get_publisher(publisher_handle)?;
-        let updated_rp = rp.publish(query)?;
-        let key = Key::from_str(&rp.publisher_handle);
-        let inf = Info::now(
-            actor(),
-            "Updated files".to_string()
-        );
-        self.store.store(key, updated_rp, inf)?;
-        Ok(())
-    }
-
-
-}
-
-//------------ RepoPublisher -------------------------------------------------
-
-#[derive(Clone, Debug, Deserialize, Serialize)]
-pub struct RepoPublisher {
-    publisher_handle:       String,
-
-    #[serde(
-    deserialize_with = "ext_serde::de_rsync_uri",
-    serialize_with = "ext_serde::ser_rsync_uri")]
-    base_uri:   uri::Rsync,
-
-    files:      Vec<CurrentFile>
-}
-
-impl RepoPublisher {
-    pub fn new(name: String, base_uri: uri::Rsync) -> Self {
-        let files = Vec::new();
-        RepoPublisher {
-            publisher_handle: name,
-            base_uri,
-            files
-        }
-    }
-
-    pub fn current_files(&self) -> &Vec<CurrentFile> {
-        &self.files
-    }
-
-    fn process_update(
-        list: &mut Vec<CurrentFile>,
-        update: &Update
-    ) -> Result<(), Error> {
-        if let Some(entry) = list.iter().find(
-            |f| {&f.uri == update.uri() }
-        ) {
-            if entry.hash != update.hash() {
-                return Err(Error::UpdateWrongHash)
-            }
-        } else {
-            return Err(Error::UpdateWrongUri(update.uri().clone()))
-        }
-
-        list.retain(|e| { &e.uri != update.uri() });
-        let new_entry = CurrentFile::new(
-            update.uri().clone(),
-            update.object().clone()
-        );
-        list.push(new_entry);
-        Ok(())
-
-    }
-
-    fn process_publish(
-        list: &mut Vec<CurrentFile>,
-        publish: &Publish
-    ) -> Result<(), Error> {
-        if let Some(entry) = list.iter().find(
-            |f| {&f.uri == publish.uri() }
-        ) {
-            return Err(Error::PublishHasExistingUri(entry.uri.clone()))
-        }
-        let new_entry = CurrentFile::new(
-            publish.uri().clone(),
-            publish.object().clone()
-        );
-        list.push(new_entry);
-        Ok(())
-    }
-
-    fn process_withdraw(
-        list: &mut Vec<CurrentFile>,
-        withdraw: &Withdraw
-    ) -> Result<(), Error> {
-        if let Some(entry) = list.iter().find(
-            |f| {&f.uri == withdraw.uri() }
-        ) {
-            if entry.hash != withdraw.hash() {
-                return Err(Error::WithdrawWrongHash)
-            }
-        } else {
-            return Err(Error::WithdrawWrongUri(withdraw.uri().clone()))
-        }
-
-        list.retain(|e| { &e.uri != withdraw.uri() });
-        Ok(())
-    }
-
-    fn assert_uri(base: &uri::Rsync, file: &uri::Rsync) -> Result<(), Error> {
-        if base.module() == file.module() &&
-           file.path().starts_with(base.path())
-        {
-            Ok(())
-        } else {
-            Err(Error::OutsideBaseUri)
-        }
-
-    }
-
-    /// Returns a new RepoPublisher with an updated list of files.
-    ///
-    /// Note that RepoPublishers are stored as versioned immutable (Arc)
-    /// in the keystore, and therefore a new instance is needed whenever
-    /// there is a change.
-    pub fn publish(&self, query: PublishQuery) -> Result<Self, Error> {
-
-        let mut new_list = self.files.clone();
-
-        for el in query.elements() {
-            match el {
-                PublishElement::Publish(p) => {
-                    Self::assert_uri(&self.base_uri, p.uri())?;
-                    Self::process_publish(&mut new_list, p)?;
-                },
-                PublishElement::Update(u) => {
-                    Self::assert_uri(&self.base_uri, u.uri())?;
-                    Self::process_update(&mut new_list, u)?;
-                },
-                PublishElement::Withdraw(w) => {
-                    Self::assert_uri(&self.base_uri, w.uri())?;
-                    Self::process_withdraw(&mut new_list, w)?;
-                },
-            }
-        }
-
-        Ok(RepoPublisher {
-            publisher_handle: self.publisher_handle.clone(),
-            base_uri: self.base_uri.clone(),
-            files: new_list
-        })
-    }
-}
-
-#[derive(Clone, Debug, Deserialize, Serialize)]
-pub struct CurrentFile {
-    #[serde(
-    deserialize_with = "ext_serde::de_rsync_uri",
-    serialize_with = "ext_serde::ser_rsync_uri")]
-    uri:     uri::Rsync,
-
-    #[serde(
-    deserialize_with = "ext_serde::de_bytes",
-    serialize_with = "ext_serde::ser_bytes")]
-    /// The actual file content. Note that we may want to store this
-    /// only on disk in future (look up by sha256 hash), to save memory.
-    content: Bytes,
-
-    #[serde(
-    deserialize_with = "ext_serde::de_bytes",
-    serialize_with = "ext_serde::ser_bytes")]
-    /// The sha-256 hash of the file (as is used on the RPKI manifests and
-    /// in the publication protocol for list, update and withdraw). Saving
-    /// this rather than calculating on demand seems a small price for some
-    /// performance gain.
-    hash:    Bytes
-}
-
-impl CurrentFile {
-    pub fn new(uri: uri::Rsync, content: Bytes) -> Self {
-        let hash = publication::hash(&content);
-        CurrentFile {uri, content, hash}
-    }
-
-    pub fn uri(&self) -> &uri::Rsync {
-        &self.uri
-    }
-
-    pub fn content(&self) -> &Bytes {
-        &self.content
-    }
-
-    pub fn hash(&self) -> &Bytes {
-        &self.hash
-    }
-
-    pub fn as_publish(&self) -> PublishElement {
-        Publish::publish(&self.content, self.uri.clone())
-    }
-
-    pub fn as_update(&self, old_content: &Bytes) -> PublishElement {
-        Update::publish(old_content, &self.content, self.uri.clone())
-    }
-
-    pub fn as_withdraw(&self) -> PublishElement {
-        Withdraw::publish(&self.content, self.uri.clone())
-    }
-}
-
-impl PartialEq for CurrentFile {
-    fn eq(&self, other: &CurrentFile) -> bool {
-        self.uri == other.uri &&
-        self.hash == other.hash &&
-        self.content == other.content
-    }
-}
-
-impl Eq for CurrentFile {}
-
 
 //------------ Error ---------------------------------------------------------
 
 #[derive(Debug, Fail)]
 pub enum Error {
     #[fail(display="{}", _0)]
-    IoError(io::Error),
+    PublisherStoreError(publisher_store::Error),
 
     #[fail(display="{}", _0)]
-    KeyStoreError(keystore::Error),
-
-    #[fail(display="Unknown publisher: {}", _0)]
-    UnknownPublisher(String),
-
-    #[fail(display="File already exists for uri (use update!): {}", _0)]
-    PublishHasExistingUri(uri::Rsync),
-
-    #[fail(display="File sent for update has no entry for uri: {}", _0)]
-    UpdateWrongUri(uri::Rsync),
-
-    #[fail(display="File for update exists, but hash does not match")]
-    UpdateWrongHash,
-
-    #[fail(display="The withdraw URI is not known: {}", _0)]
-    WithdrawWrongUri(uri::Rsync),
-
-    #[fail(display="File for withdraw exists, but hash does not match")]
-    WithdrawWrongHash,
-
-    #[fail(display="Publishing outside of base URI is not allowed.")]
-    OutsideBaseUri,
+    FileStoreError(file_store::Error),
 }
 
-impl From<io::Error> for Error {
-    fn from(e: io::Error) -> Self {
-        Error::IoError(e)
+impl From<publisher_store::Error> for Error {
+    fn from(e: publisher_store::Error) -> Self {
+        Error::PublisherStoreError(e)
     }
 }
 
-impl From<keystore::Error> for Error {
-    fn from(e: keystore::Error) -> Self {
-        Error::KeyStoreError(e)
+impl From<file_store::Error> for Error {
+    fn from(e: file_store::Error) -> Self {
+        Error::FileStoreError(e)
     }
 }
-
-
-
-//------------ Tests ---------------------------------------------------------
-
-#[cfg(test)]
-mod tests {
-    use super::*;
-    use test;
-
-    #[test]
-    fn should_add_and_remove_publisher() {
-        test::test_with_tmp_dir(|d| {
-            let mut repo = Repository::new(&d).unwrap();
-            let alice = RepoPublisher::new(
-                "alice".to_string(),
-                test::rsync_uri("rsync://host/module/alice/")
-            );
-
-            repo.add_publisher(alice).unwrap();
-
-            let publishers = repo.publishers().unwrap();
-            assert_eq!(1, publishers.len());
-
-            repo.remove_publisher("alice").unwrap();
-
-            let publishers = repo.publishers().unwrap();
-            assert_eq!(0, publishers.len());
-        })
-    }
-
-    #[test]
-    fn should_add_list_remove_files() {
-        test::test_with_tmp_dir(|d| {
-
-            let mut repo = Repository::new(&d).unwrap();
-            let alice = RepoPublisher::new(
-                "alice".to_string(),
-                test::rsync_uri("rsync://host/module/alice/")
-            );
-
-            repo.add_publisher(alice).unwrap();
-
-            let alice = repo.get_publisher("alice").unwrap();
-            let files = alice.current_files();
-            assert_eq!(0, files.len());
-
-            let file = CurrentFile::new(
-                test::rsync_uri("rsync://host/module/alice/file.txt"),
-                Bytes::from("example content")
-            );
-
-            //--------- Add a single file
-            let mut builder = PublishQuery::build();
-            builder.add(file.clone().as_publish());
-            let message = builder.build_message();
-            let publish = message.as_query().unwrap().as_publish().unwrap();
-
-            repo.publish("alice", publish.clone()).unwrap();
-
-            let alice = repo.get_publisher("alice").unwrap();
-            let files = alice.current_files();
-            assert_eq!(1, files.len());
-
-            // Can't publish the same thing again, should then update!
-            assert!(repo.publish("alice", publish).is_err());
-
-            //--------- Update a single file
-            let file_update = CurrentFile::new(
-                file.uri().clone(),
-                Bytes::from("example updated content")
-            );
-
-            let mut builder = PublishQuery::build();
-            builder.add(file_update.clone().as_update(file.content()));
-            let message = builder.build_message();
-            let publish = message.as_query().unwrap().as_publish().unwrap();
-
-            repo.publish("alice", publish.clone()).unwrap();
-
-            let alice = repo.get_publisher("alice").unwrap();
-            let files = alice.current_files();
-            assert_eq!(1, files.len());
-
-            //--------- Withdraw a single file
-            let mut builder = PublishQuery::build();
-            builder.add(file_update.as_withdraw());
-            let message = builder.build_message();
-            let publish = message.as_query().unwrap().as_publish().unwrap();
-
-            repo.publish("alice", publish.clone()).unwrap();
-
-            let alice = repo.get_publisher("alice").unwrap();
-            let files = alice.current_files();
-            assert_eq!(0, files.len());
-        });
-    }
-
-    #[test]
-    fn should_not_allow_publishing_or_withdrawing_outside_of_base() {
-        test::test_with_tmp_dir(|d| {
-
-            let mut repo = Repository::new(&d).unwrap();
-            let alice = RepoPublisher::new(
-                "alice".to_string(),
-                test::rsync_uri("rsync://host/module/alice/")
-            );
-
-            repo.add_publisher(alice).unwrap();
-
-            let alice = repo.get_publisher("alice").unwrap();
-            let files = alice.current_files();
-            assert_eq!(0, files.len());
-
-            let file = CurrentFile::new(
-                test::rsync_uri("rsync://host/module/bob/file.txt"),
-                Bytes::from("example content")
-            );
-
-            //--------- Add a single file
-            let mut builder = PublishQuery::build();
-            builder.add(file.clone().as_publish());
-            let message = builder.build_message();
-            let publish = message.as_query().unwrap().as_publish().unwrap();
-
-            assert!(repo.publish("alice", publish.clone()).is_err());
-        });
-    }
-
-}

src/signing/softsigner.rs

@@ -38,7 +38,7 @@ pub struct OpenSslSigner {
 }
 
 impl OpenSslSigner {
-    pub fn new(work_dir: PathBuf) -> Result<Self, Error> {
+    pub fn new(work_dir: &PathBuf) -> Result<Self, Error> {
         let meta_data = fs::metadata(&work_dir)?;
         if meta_data.is_dir() {
 
@@ -54,7 +54,7 @@ impl OpenSslSigner {
                 }
             )
         } else {
-            Err(Error::InvalidWorkDir(work_dir))
+            Err(Error::InvalidWorkDir(work_dir.clone()))
         }
     }
 }
@@ -274,7 +274,7 @@ pub mod tests {
     #[test]
     fn should_return_subject_public_key_info() {
         test::test_with_tmp_dir(|d| {
-            let mut s = OpenSslSigner::new(d).unwrap();
+            let mut s = OpenSslSigner::new(&d).unwrap();
             let ki = s.create_key(&PublicKeyAlgorithm::RsaEncryption).unwrap();
             s.get_key_info(&ki).unwrap();
             s.destroy_key(&ki).unwrap();

tests/integration_test.rs

@@ -8,18 +8,17 @@ extern crate tokio;
 use std::fs::File;
 use std::io::Write;
 use std::path::PathBuf;
+use std::str;
+use std::{thread, time};
 use hyper::Client;
 use rpki::oob::exchange::PublisherRequest;
 use rpubd::test;
+use rpubd::pubc::client::PubClient;
 use rpubd::pubd::config::Config;
 use rpubd::pubd::daemon;
-use rpubd::provisioning::publisher::Publisher;
-
-use std::str;
-use std::{thread, time};
+use rpubd::repo::publisher::Publisher;
 use tokio::prelude::*;
 use tokio::runtime::Runtime;
-use rpubd::pubc::client::PubClient;
 
 fn save_pr(base_dir: &PathBuf, file_name: &str, pr: &PublisherRequest) {
     let mut full_name = base_dir.clone();
@@ -40,7 +39,7 @@ fn testing() {
 
         // Set up a client
         let client_dir = test::create_sub_dir(&d);
-        let mut client = PubClient::new(client_dir).unwrap();
+        let mut client = PubClient::new(&client_dir).unwrap();
         client.init("client".to_string()).unwrap();
         let pr = client.publisher_request().unwrap();
 
@@ -58,7 +57,7 @@ fn testing() {
         );
 
         // XXX TODO: Find a better way to know the server is ready!
-        thread::sleep(time::Duration::from_millis(100));
+        thread::sleep(time::Duration::from_millis(150));
 
         // XXX TODO: Use a helper to create the futures to check the
         // XXX TODO: responses.. the compiler insists this crosses threads