diff --git a/src/bin/pubc.rs b/src/bin/pubc.rs index 3e76d3aa..609958b3 100644 --- a/src/bin/pubc.rs +++ b/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); diff --git a/src/provisioning/mod.rs b/src/provisioning/mod.rs index 84238498..55cc091c 100644 --- a/src/provisioning/mod.rs +++ b/src/provisioning/mod.rs @@ -1,3 +1 @@ -pub mod info; -pub mod publisher; -pub mod publisher_list; \ No newline at end of file +pub mod info; \ No newline at end of file diff --git a/src/pubc/client.rs b/src/pubc/client.rs index bc3bb61c..60a08bde 100644 --- a/src/pubc/client.rs +++ b/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 { + pub fn new(work_dir: &PathBuf) -> Result { 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(); diff --git a/src/pubd/config.rs b/src/pubd/config.rs index 460aae77..27971916 100644 --- a/src/pubd/config.rs +++ b/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() diff --git a/src/pubd/daemon.rs b/src/pubd/daemon.rs index d763e44e..98d18a95 100644 --- a/src/pubd/daemon.rs +++ b/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() diff --git a/src/pubd/server.rs b/src/pubd/server.rs index e4743272..bcef5a0a 100644 --- a/src/pubd/server.rs +++ b/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 { @@ -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 { - let mut publisher_list = PublisherList::new( - work_dir.clone(), + pub_xml_dir: &PathBuf, + base_uri: &uri::Rsync + ) -> Result { + 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 for Error { } } -impl From for Error { - fn from(e: publisher_list::Error) -> Self { +impl From 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(); diff --git a/src/repo/file.rs b/src/repo/file.rs new file mode 100644 index 00000000..e73fbe09 --- /dev/null +++ b/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 {} \ No newline at end of file diff --git a/src/repo/file_store.rs b/src/repo/file_store.rs new file mode 100644 index 00000000..4cddf6d6 --- /dev/null +++ b/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 { + 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, Error> { + let path = self.file_path(base_uri); + self.recurse_disk(&path) + } + + fn recurse_disk( + &self, + path: &PathBuf + ) -> Result, 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 { + 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, 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, 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 for Error { + fn from(e: io::Error) -> Self { + Error::IoError(e) + } +} + +impl From 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") } + } + }); + } + + +} \ No newline at end of file diff --git a/src/repo/mod.rs b/src/repo/mod.rs index 4e693837..07f45983 100644 --- a/src/repo/mod.rs +++ b/src/repo/mod.rs @@ -1 +1,5 @@ +pub mod file; +pub mod file_store; +pub mod publisher; +pub mod publisher_store; pub mod repository; \ No newline at end of file diff --git a/src/provisioning/publisher.rs b/src/repo/publisher.rs similarity index 100% rename from src/provisioning/publisher.rs rename to src/repo/publisher.rs diff --git a/src/provisioning/publisher_list.rs b/src/repo/publisher_store.rs similarity index 94% rename from src/provisioning/publisher_list.rs rename to src/repo/publisher_store.rs index a81675b7..325f5cbb 100644 --- a/src/provisioning/publisher_list.rs +++ b/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 { 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, 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()); }) diff --git a/src/repo/repository.rs b/src/repo/repository.rs index fb5dac63..f9a2e098 100644 --- a/src/repo/repository.rs +++ b/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 { - 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 { + 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>, 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, 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 -} - -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 { - &self.files - } - - fn process_update( - list: &mut Vec, - 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, - 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, - 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 { - - 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 for Error { - fn from(e: io::Error) -> Self { - Error::IoError(e) +impl From for Error { + fn from(e: publisher_store::Error) -> Self { + Error::PublisherStoreError(e) } } -impl From for Error { - fn from(e: keystore::Error) -> Self { - Error::KeyStoreError(e) +impl From 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()); - }); - } - -} \ No newline at end of file diff --git a/src/signing/softsigner.rs b/src/signing/softsigner.rs index 792023d9..b2bffd5a 100644 --- a/src/signing/softsigner.rs +++ b/src/signing/softsigner.rs @@ -38,7 +38,7 @@ pub struct OpenSslSigner { } impl OpenSslSigner { - pub fn new(work_dir: PathBuf) -> Result { + pub fn new(work_dir: &PathBuf) -> Result { 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(); diff --git a/tests/integration_test.rs b/tests/integration_test.rs index 079f6e85..a941e2c8 100644 --- a/tests/integration_test.rs +++ b/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