diff --git a/Cargo.toml b/Cargo.toml index f6b8c606..4366c21b 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -1,7 +1,6 @@ [workspace] -members = [ "client", "cms_proxy", "daemon", "pubc", "pubd" ] +members = [ "client", "cms_proxy", "commons", "daemon", "pubc", "pubd" ] [patch.crates-io] -krill_commons = { git="https://github.com/NLnetLabs/krill-commons.git" } rpki = { git="https://github.com/NLnetLabs/rpki-rs.git" } bcder = { git="https://github.com/NLnetLabs/bcder.git" } diff --git a/client/Cargo.toml b/client/Cargo.toml index 6330a61a..a121f20d 100644 --- a/client/Cargo.toml +++ b/client/Cargo.toml @@ -6,7 +6,6 @@ authors = ["Tim Bruijnzeels ", "Martin Hoffmann ", "Martin Hoffmann "] + +[dependencies] +actix = "^0.7" +actix-web = "^0.7" +base64 = "^0.9" +bytes = "^0.4" +chrono = { version = "^0.4", features = ["serde"] } +derive_more = "^0.13" +futures = "0.1" +hex = "^0.3" +log = "^0.4" +openssl = { version = "^0.10", features = ["v110"] } +rand = "^0.5" +reqwest = "^0.9" +rpki = "^0.3" +serde = "^1.0" +serde_derive = "^1.0" +serde_json = "^1.0" +syslog = "^4.0" +xml-rs = "0.8.0" \ No newline at end of file diff --git a/commons/README.md b/commons/README.md new file mode 100644 index 00000000..22d1731a --- /dev/null +++ b/commons/README.md @@ -0,0 +1,7 @@ +# Krill Commons + +This project defines common types used by the different Krill components. + +## License + +This software is distributed under the Mozilla Public License 2.0. See the LICENSE file included. diff --git a/commons/src/api/admin.rs b/commons/src/api/admin.rs new file mode 100644 index 00000000..ae1f953c --- /dev/null +++ b/commons/src/api/admin.rs @@ -0,0 +1,227 @@ +//! Support for admin tasks, such as managing publishers and RFC8181 clients + +use rpki::uri; +use crate::api::Link; +use crate::eventsourcing::AggregateId; +use crate::util::ext_serde; +use std::fmt; +use std::fmt::Display; + + +//------------ PublisherHandle ----------------------------------------------- + +/// A type for referring to publishers, both in the api as well as to the +/// aggregates. +#[derive(Clone, Debug, Deserialize, Eq, Hash, PartialEq, Serialize)] +pub struct PublisherHandle(AggregateId); + +impl PublisherHandle { + pub fn name(&self) -> &str { + self.0.as_str() + } +} + +impl From<&str> for PublisherHandle { + fn from(s: &str) -> Self { + PublisherHandle::from(AggregateId::from(s)) + } +} + +impl From for PublisherHandle { + fn from(s: String) -> Self { PublisherHandle::from(AggregateId::from(s))} +} + +impl From for PublisherHandle { + fn from(id: AggregateId) -> Self { + PublisherHandle(id) + } +} + +impl From<&AggregateId> for PublisherHandle { + fn from(id: &AggregateId) -> Self { + PublisherHandle(id.clone()) + } +} + +impl AsRef for PublisherHandle { + fn as_ref(&self) -> &str { + self.name() + } +} + +impl AsRef for PublisherHandle { + fn as_ref(&self) -> &AggregateId { + &self.0 + } +} + +impl Display for PublisherHandle { + fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result { + write!(f, "{}", self.name()) + } +} + + +//------------ PublisherRequest ---------------------------------------------- + +/// This type defines request for a new Publisher (CA that is allowed to +/// publish). +#[derive(Clone, Debug, Deserialize, Serialize)] +pub struct PublisherRequest { + handle: String, + + /// The token used by the API + token: String, + + #[serde( + deserialize_with = "ext_serde::de_rsync_uri", + serialize_with = "ext_serde::ser_rsync_uri")] + base_uri: uri::Rsync, +} + +impl PublisherRequest { + pub fn new( + handle: String, + token: String, + base_uri: uri::Rsync, + ) -> Self { + PublisherRequest { + handle, + token, + base_uri, + } + } +} + +impl PublisherRequest { + pub fn handle(&self) -> &String { + &self.handle + } + + pub fn token(&self) -> &String { + &self.token + } + + pub fn base_uri(&self) -> &uri::Rsync { + &self.base_uri + } + + /// Return all the values (handle, token, base_uri). + pub fn unwrap(self) -> (String, String, uri::Rsync) { + (self.handle, self.token, self.base_uri) + } +} + +impl PartialEq for PublisherRequest { + fn eq(&self, other: &PublisherRequest) -> bool { + self.handle == other.handle && + self.base_uri == other.base_uri + } +} + +impl Eq for PublisherRequest {} + + +//------------ PublisherSummaryInfo ------------------------------------------ + +/// Defines a summary of publisher information to be used in the publisher +/// list. +#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)] +pub struct PublisherSummary { + id: String, + links: Vec +} + +impl PublisherSummary { + pub fn from( + handle: &PublisherHandle, + path_publishers: &str + ) -> PublisherSummary { + let mut links = Vec::new(); + let self_link = Link { + rel: "self".to_string(), + link: format!("{}/{}", path_publishers, handle) + }; + links.push(self_link); + + PublisherSummary { + id: handle.to_string(), + links + } + } + + pub fn id(&self) -> &str { &self.id } +} + + +//------------ PublisherList ------------------------------------------------- + +/// This type represents a list of (all) current publishers to show in the API +#[derive(Clone, Eq, Debug, Deserialize, PartialEq, Serialize)] +pub struct PublisherList { + publishers: Vec +} + +impl PublisherList { + pub fn build( + publishers: &[PublisherHandle], + path_publishers: &str + ) -> PublisherList { + let publishers: Vec = publishers.iter().map(|p| + PublisherSummary::from(&p, path_publishers) + ).collect(); + + PublisherList { + publishers + } + } + + pub fn publishers(&self) -> &Vec { + &self.publishers + } +} + + +//------------ PublisherDetails ---------------------------------------------- + +/// This type defines the publisher details for: +/// /api/v1/publishers/{handle} +#[derive(Clone, Debug, Deserialize, Serialize)] +pub struct PublisherDetails { + handle: String, + + deactivated: bool, + + #[serde( + deserialize_with = "ext_serde::de_rsync_uri", + serialize_with = "ext_serde::ser_rsync_uri" + )] + base_uri: uri::Rsync, +} + +impl PublisherDetails { + pub fn new(handle: &str, deactivated: bool, base_uri: &uri::Rsync) -> Self { + PublisherDetails { + handle: handle.to_string(), + deactivated, + base_uri: base_uri.clone() + } + } + + pub fn handle(&self) -> &str { &self.handle } + pub fn deactivated(&self) -> bool { self.deactivated } + pub fn base_uri(&self) -> &uri::Rsync { &self.base_uri } +} + +impl PartialEq for PublisherDetails { + fn eq(&self, other: &PublisherDetails) -> bool { + match (serde_json::to_string(self), serde_json::to_string(other)) { + (Ok(ser_self), Ok(ser_other)) => ser_self == ser_other, + _ => false + } + } +} + +impl Eq for PublisherDetails {} + + diff --git a/commons/src/api/mod.rs b/commons/src/api/mod.rs new file mode 100644 index 00000000..36859365 --- /dev/null +++ b/commons/src/api/mod.rs @@ -0,0 +1,318 @@ +//! Data structures for the API, shared between client and server. +pub mod admin; +pub mod publication; +pub mod rrdp; + +use bytes::Bytes; +use serde::{Deserialize, Deserializer, Serialize, Serializer}; +use crate::util::sha256; + + +//------------ Base64 -------------------------------------------------------- + +/// This type contains a base64 encoded structure. The publication protocol +/// deals with objects in their base64 encoded form. +/// +/// Note that we store this in a Bytes to make it cheap to clone this. +#[derive(Clone, Debug, Eq, PartialEq)] +pub struct Base64(Bytes); + +impl Base64 { + pub fn from_content(content: &[u8]) -> Self { + Base64::from(base64::encode(content)) + } + + /// Decodes into bytes (e.g. for saving to disk for rcync) + pub fn to_bytes(&self) -> Bytes { + Bytes::from(base64::decode(&self.0).unwrap()) + } + + pub fn to_hex_hash(&self) -> String { + hex::encode(sha256(&self.to_bytes())) + } + + pub fn to_encoded_hash(&self) -> EncodedHash { + EncodedHash::from(self.to_hex_hash()) + } +} + +impl AsRef for Base64 { + fn as_ref(&self) -> &str { + use std::str; + str::from_utf8(&self.0).unwrap() + } +} + +impl From for Base64 { + fn from(s: String) -> Self { + Base64(Bytes::from(s)) + } +} + + +impl ToString for Base64 { + fn to_string(&self) -> String { + unsafe { + String::from_utf8_unchecked(self.0.to_vec()) + } + } +} + +impl Serialize for Base64 { + fn serialize( + &self, serializer: S + ) -> Result where S: Serializer { + self.to_string().serialize(serializer) + } +} + +impl<'de> Deserialize<'de> for Base64 { + fn deserialize( + deserializer: D + ) -> Result where D: Deserializer<'de> { + let string = String::deserialize(deserializer)?; + Ok(Base64::from(string)) + } +} + + +//------------ EncodedHash --------------------------------------------------- + +/// This type contains a hex encoded sha256 hash. +/// +/// Note that we store this in a Bytes for cheap cloning. +#[derive(Clone, Debug, Eq, Hash, PartialEq)] +pub struct EncodedHash(Bytes); + +impl EncodedHash { + pub fn from_content(content: &[u8]) -> Self { + let sha256 = sha256(content); + let hex = hex::encode(sha256); + EncodedHash::from(hex) + } +} + +impl AsRef for EncodedHash { + fn as_ref(&self) -> &str { + use std::str; + str::from_utf8(&self.0).unwrap() + } +} + +impl From for EncodedHash { + fn from(s: String) -> Self { + EncodedHash(Bytes::from(s.to_lowercase())) + } +} + +impl ToString for EncodedHash { + fn to_string(&self) -> String { + unsafe { + String::from_utf8_unchecked(self.0.to_vec()) + } + } +} + +impl Serialize for EncodedHash { + fn serialize( + &self, serializer: S + ) -> Result where S: Serializer { + self.to_string().serialize(serializer) + } +} + +impl<'de> Deserialize<'de> for EncodedHash { + fn deserialize( + deserializer: D + ) -> Result where D: Deserializer<'de> { + let string = String::deserialize(deserializer)?; + Ok(EncodedHash::from(string)) + } +} + + + +//------------ Link ---------------------------------------------------------- + +/// Defines a link element to include as part of a links array in a Json +/// response. +#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)] +pub struct Link { + rel: String, + link: String +} + + +//------------ ErrorResponse -------------------------------------------------- + +/// Defines an error response. Codes are unique and documented here: +/// https://rpki.readthedocs.io/en/latest/krill/pub/api.html#error-responses +#[derive(Debug, Deserialize, Serialize)] +pub struct ErrorResponse { + code: usize, + msg: String +} + +impl ErrorResponse { + pub fn new(code: usize, msg: String) -> Self { ErrorResponse { code, msg }} + pub fn code(&self) -> usize { self.code } + pub fn msg(&self) -> &str { &self.msg } +} + +impl Into for ErrorResponse { + fn into(self) -> ErrorCode { + ErrorCode::from(self.code) + } +} + +/// This type defines externally visible errors that the API may return. +#[derive(Clone, Debug, Display, Eq, PartialEq)] +pub enum ErrorCode { + // 1000s (User Input Errors) + #[display(fmt="Submitted Json cannot be parsed")] + InvalidJson, + + #[display(fmt="Invalid RFC8183 Publisher Request")] + InvalidPublisherRequest, + + #[display(fmt="Issue with submitted publication XML")] + InvalidPublicationXml, + + #[display(fmt="Invalid handle name")] + InvalidHandle, + + #[display(fmt="Handle already in use")] + DuplicateHandle, + + // 2000s (Authorisation and Consistency issues) + #[display(fmt="Unknown publisher")] + UnknownPublisher, + + #[display(fmt="Submitted protocol CMS does not validate")] + CmsValidation, + + #[display(fmt="Base URI for publisher is outside of publisher base URI")] + InvalidBaseUri, + + #[display(fmt="Out of sync with server, please send requests for instances sequentially")] + ConcurrentModification, + + #[display(fmt="Publisher has been deactivated")] + PublisherDeactivated, + + #[display(fmt="Not allowed to publish outside of publisher jail")] + UriOutsideJail, + + #[display(fmt="File already exists for uri (use update!)")] + ObjectAlreadyPresent, + + #[display(fmt="No file found for hash at uri")] + NoObjectForHashAndOrUri, + + // 3000s (Server Errors) + #[display(fmt="Cannot update internal state, issue with work_dir?")] + Persistence, + + #[display(fmt="Cannot update repository, issue with repo_dir?")] + RepositoryUpdate, + + #[display(fmt="Signing error, issue with openssl version or work_dir?")] + SigningError, + + #[display(fmt="Proxy server error.")] + ProxyError, + + #[display(fmt="Unrecognised error (this is a bug)")] + Unknown +} + +impl From for ErrorCode { + fn from(n: usize) -> Self { + match n { + 1001 => ErrorCode::InvalidJson, + 1002 => ErrorCode::InvalidPublisherRequest, + 1003 => ErrorCode::InvalidPublicationXml, + 1004 => ErrorCode::InvalidHandle, + + 2001 => ErrorCode::UnknownPublisher, + 2002 => ErrorCode::CmsValidation, + 2003 => ErrorCode::InvalidBaseUri, + 2004 => ErrorCode::ConcurrentModification, + 2005 => ErrorCode::PublisherDeactivated, + 2006 => ErrorCode::UriOutsideJail, + 2007 => ErrorCode::ObjectAlreadyPresent, + 2008 => ErrorCode::NoObjectForHashAndOrUri, + 2009 => ErrorCode::DuplicateHandle, + + 3001 => ErrorCode::Persistence, + 3002 => ErrorCode::RepositoryUpdate, + 3003 => ErrorCode::SigningError, + 3004 => ErrorCode::ProxyError, + + _ => ErrorCode::Unknown + } + } +} + +impl Into for ErrorCode { + fn into(self) -> ErrorResponse { + let code = match self { + ErrorCode::InvalidJson => 1001, + ErrorCode::InvalidPublisherRequest => 1002, + ErrorCode::InvalidPublicationXml => 1003, + ErrorCode::InvalidHandle => 1004, + + ErrorCode::UnknownPublisher => 2001, + ErrorCode::CmsValidation => 2002, + ErrorCode::InvalidBaseUri => 2003, + ErrorCode::ConcurrentModification => 2004, + ErrorCode::PublisherDeactivated => 2005, + ErrorCode::UriOutsideJail => 2006, + ErrorCode::ObjectAlreadyPresent => 2007, + ErrorCode::NoObjectForHashAndOrUri => 2008, + ErrorCode::DuplicateHandle => 2009, + + ErrorCode::Persistence => 3001, + ErrorCode::RepositoryUpdate => 3002, + ErrorCode::SigningError => 3003, + ErrorCode::ProxyError => 3004, + + ErrorCode::Unknown => 65535 + }; + let msg = format!("{}", self); + + ErrorResponse { code, msg } + } +} + +//------------ Tests --------------------------------------------------------- + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn should_convert_code_to_number_and_back() { + + fn test_code(number_to_test: usize) { + let code = ErrorCode::from(number_to_test); + let response: ErrorResponse = code.into(); + assert_eq!(number_to_test, response.code()); + } + + for n in 1001..1005 { + test_code(n) + } + + for n in 2001..2010 { + test_code(n) + } + + for n in 3001..3005 { + test_code(n) + } + + + } +} + diff --git a/commons/src/api/publication.rs b/commons/src/api/publication.rs new file mode 100644 index 00000000..57b35c0b --- /dev/null +++ b/commons/src/api/publication.rs @@ -0,0 +1,298 @@ +//! Support for requests sent to the Json API +use rpki::uri; +use crate::api::{ Base64, EncodedHash }; +use crate::util::ext_serde; +use crate::util::file::CurrentFile; + + +//------------ PublishRequest ------------------------------------------------ + +/// This type provides a convenience wrapper to contain the request found +/// inside of a validated RFC8181 request. +#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)] +pub enum PublishRequest { + List, // See https://tools.ietf.org/html/rfc8181#section-2.3 + Delta(PublishDelta) +} + + +//------------ PublishDelta ------------------------------------------------ + +/// This type represents a multi element query as described in +/// https://tools.ietf.org/html/rfc8181#section-3.7 +#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)] +pub struct PublishDelta { + publishes: Vec, + updates: Vec, + withdraws: Vec +} + +impl PublishDelta { + pub fn new( + publishes: Vec, + updates: Vec, + withdraws: Vec + ) -> Self { + PublishDelta { publishes, updates, withdraws } + } + + pub fn publishes(&self) -> &Vec { + &self.publishes + } + pub fn updates(&self) -> &Vec { + &self.updates + } + pub fn withdraws(&self) -> &Vec { + &self.withdraws + } + + pub fn len(&self) -> usize { + self.publishes.len() + self.updates.len() + self.withdraws.len() + } + + pub fn is_empty(&self) -> bool { self.len() == 0 } + + pub fn unwrap(self) -> (Vec, Vec, Vec) { + (self.publishes, self.updates, self.withdraws) + } +} + + +//------------ PublishDeltaBuilder ------------------------------------------- + +#[derive(Default)] +pub struct PublishDeltaBuilder { + publishes: Vec, + updates: Vec, + withdraws: Vec +} + +impl PublishDeltaBuilder { + pub fn new() -> Self { + Self::default() + } + + pub fn add_publish(&mut self, publish: Publish) { + self.publishes.push(publish); + } + + pub fn add_update(&mut self, update: Update) { + self.updates.push(update); + } + + pub fn add_withdraw(&mut self, withdraw: Withdraw) { + self.withdraws.push(withdraw); + } + + pub fn finish(self) -> PublishDelta { + PublishDelta { + publishes: self.publishes, + updates: self.updates, + withdraws: self.withdraws + } + } +} + + +//------------ Publish ------------------------------------------------------ + +/// Type representing a json equivalent to the publish element, that does not +/// update any existing object, defined in: +/// https://tools.ietf.org/html/rfc8181#section-3.1 +#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)] +pub struct Publish { + tag: Option, + + #[serde( + deserialize_with = "ext_serde::de_rsync_uri", + serialize_with = "ext_serde::ser_rsync_uri")] + uri: uri::Rsync, + + content: Base64 +} + +impl Publish { + pub fn new(tag: Option, uri: uri::Rsync, content: Base64) -> Self { + Publish { tag, uri, content } + } + pub fn with_hash_tag(uri: uri::Rsync, content: Base64) -> Self { + let tag = Some(content.to_hex_hash()); + Publish { tag, uri, content } + } + + pub fn tag(&self) -> &Option { &self.tag } + pub fn tag_for_xml(&self) -> String { + match &self.tag { + None => "".to_string(), + Some(t) => t.clone() + } + } + pub fn uri(&self) -> &uri::Rsync{ &self.uri} + pub fn content(&self) -> &Base64{ &self.content } + + pub fn unwrap(self) -> (Option, uri::Rsync, Base64) { + (self.tag, self.uri, self.content) + } +} + + +//------------ Update -------------------------------------------------------- + +/// Type representing a json equivalent to the publish element, that updates +/// an existing object: +/// https://tools.ietf.org/html/rfc8181#section-3.2 +#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)] +pub struct Update { + tag: Option, + + #[serde( + deserialize_with = "ext_serde::de_rsync_uri", + serialize_with = "ext_serde::ser_rsync_uri")] + uri: uri::Rsync, + + content: Base64, + + hash: EncodedHash, +} + +impl Update { + pub fn new( + tag: Option, + uri: uri::Rsync, + content: Base64, + old_hash: EncodedHash + ) -> Self { + Update { tag, uri, content, hash: old_hash } + } + pub fn with_hash_tag( + uri: uri::Rsync, + content: Base64, + old_hash: EncodedHash + ) -> Self { + let tag = Some(content.to_hex_hash()); + Update { tag, uri, content, hash: old_hash } + } + + pub fn tag(&self) -> &Option { &self.tag } + pub fn tag_for_xml(&self) -> String { + match &self.tag { + Some(t) => t.clone(), + None => "".to_string() + } + } + pub fn uri(&self) -> &uri::Rsync { &self.uri} + pub fn content(&self) -> &Base64 { &self.content } + pub fn hash(&self) -> &EncodedHash { &self.hash } + + pub fn unwrap(self) -> (Option, uri::Rsync, Base64, EncodedHash) { + (self.tag, self.uri, self.content, self.hash) + } +} + + +//------------ Withdraw ------------------------------------------------------ + +/// Type representing a json equivalent to a withdraw element that removes an +/// object from the repository: +/// https://tools.ietf.org/html/rfc8181#section-3.3 +#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)] +pub struct Withdraw { + tag: Option, + + #[serde( + deserialize_with = "ext_serde::de_rsync_uri", + serialize_with = "ext_serde::ser_rsync_uri")] + uri: uri::Rsync, + + hash: EncodedHash, +} + +impl Withdraw { + pub fn new(tag: Option, uri: uri::Rsync, hash: EncodedHash) -> Self { + Withdraw { tag, uri, hash } + } + + pub fn with_hash_tag(uri: uri::Rsync, hash: EncodedHash) -> Self { + let tag = Some(hash.to_string()); + Withdraw { tag, uri, hash } + } + + pub fn from_list_element(el: &ListElement) -> Self { + Withdraw { + tag: None, + uri: el.uri().clone(), + hash: el.hash().clone() + } + } + + pub fn tag(&self) -> &Option { &self.tag } + pub fn tag_for_xml(&self) -> String { + match &self.tag { + Some(t) => t.clone(), + None => "".to_string() + } + } + pub fn uri(&self) -> &uri::Rsync { &self.uri} + pub fn hash(&self) -> &EncodedHash { &self.hash } + + pub fn unwrap(self) -> (Option, uri::Rsync, EncodedHash) { + (self.tag, self.uri, self.hash) + } +} + +//------------ PublishReply -------------------------------------------------- + +/// This type is used to wrap API responses for publication requests. +pub enum PublishReply { + Success, // See https://tools.ietf.org/html/rfc8181#section-3.4 + List(ListReply) +} + + +//------------ ListReply ----------------------------------------------------- + +/// This type represents the list reply as described in +/// https://tools.ietf.org/html/rfc8181#section-2.3 +#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)] +pub struct ListReply { + elements: Vec +} + +impl ListReply { + pub fn new(elements: Vec) -> Self { + ListReply { elements } + } + + pub fn from_files(files: Vec) -> Self { + let elements = files.into_iter().map(|f| f.into_list_element()).collect(); + ListReply { elements } + } + + pub fn elements(&self) -> &Vec { + &self.elements + } +} + + +//------------ ListElement --------------------------------------------------- + +/// This type represents a single object that is published at a publication +/// server. +#[derive(Clone, Debug, Deserialize, Eq, Hash, PartialEq, Serialize)] +pub struct ListElement { + #[serde( + deserialize_with = "ext_serde::de_rsync_uri", + serialize_with = "ext_serde::ser_rsync_uri")] + uri: uri::Rsync, + + hash: EncodedHash +} + +impl ListElement { + pub fn new(uri: uri::Rsync, hash: EncodedHash) -> Self { + ListElement { uri, hash } + } + + pub fn uri(&self) -> &uri::Rsync { &self.uri } + pub fn hash(&self) -> &EncodedHash { &self.hash } +} diff --git a/commons/src/api/rrdp.rs b/commons/src/api/rrdp.rs new file mode 100644 index 00000000..ff5a5b9b --- /dev/null +++ b/commons/src/api/rrdp.rs @@ -0,0 +1,671 @@ +//! Data objects used in the (RRDP) repository. I.e. the publish, update, and +//! withdraw elements, as well as the notification, snapshot and delta file +//! definitions. +use std::collections::HashMap; +use std::io; +use std::path::PathBuf; +use bytes::Bytes; +use rpki::uri; +use crate::api::publication; +use crate::api::Base64; +use crate::api::EncodedHash; +use crate::util::ext_serde; +use crate::util::file; +use crate::util::Time; +use crate::util::xml::XmlWriter; + + +const VERSION: &str = "1"; +const NS: &str = "http://www.ripe.net/rpki/rrdp"; + +//------------ PublishElement ------------------------------------------------ + +/// The publishes as used in the RRDP protocol. +/// +/// Note that the difference with the publication protocol is the absence of +/// the tag. +#[derive(Clone, Debug, Deserialize, Serialize)] +pub struct PublishElement { + base64: Base64, + + #[serde( + deserialize_with = "ext_serde::de_rsync_uri", + serialize_with = "ext_serde::ser_rsync_uri")] + uri: uri::Rsync +} + +impl PublishElement { + pub fn new(base64: Base64, uri: uri::Rsync) -> Self { + PublishElement { base64, uri } + } + + pub fn base64(&self) -> &Base64 { &self.base64 } + pub fn uri(&self) -> &uri::Rsync { &self.uri } +} + +impl From for PublishElement { + fn from(p: publication::Publish) -> Self { + let (_tag, uri, base64) = p.unwrap(); + PublishElement { uri, base64 } + } +} + + +//------------ UpdateElement ------------------------------------------------- + +/// The updates as used in the RRDP protocol. +/// +/// Note that the difference with the publication protocol is the absence of +/// the tag. +#[derive(Clone, Debug, Deserialize, Serialize)] +pub struct UpdateElement { + #[serde( + deserialize_with = "ext_serde::de_rsync_uri", + serialize_with = "ext_serde::ser_rsync_uri")] + uri: uri::Rsync, + hash: EncodedHash, + base64: Base64 +} + +impl UpdateElement { + pub fn uri(&self) -> &uri::Rsync { &self.uri } + pub fn hash(&self) -> &EncodedHash { &self.hash } + pub fn base64(&self) -> &Base64 { &self.base64 } +} + +impl From for UpdateElement { + fn from(u: publication::Update) -> Self { + let (_tag, uri, base64, hash) = u.unwrap(); + UpdateElement { uri, base64, hash } + } +} + +impl Into for UpdateElement { + fn into(self) -> PublishElement { + PublishElement { uri: self.uri, base64: self.base64 } + } +} + + +//------------ WithdrawElement ----------------------------------------------- + +/// The withdraws as used in the RRDP protocol. +/// +/// Note that the difference with the publication protocol is the absence of +/// the tag. +#[derive(Clone, Debug, Deserialize, Serialize)] +pub struct WithdrawElement { + #[serde( + deserialize_with = "ext_serde::de_rsync_uri", + serialize_with = "ext_serde::ser_rsync_uri")] + uri: uri::Rsync, + hash: EncodedHash +} + +impl WithdrawElement { + pub fn uri(&self) -> &uri::Rsync { &self.uri } + pub fn hash(&self) -> &EncodedHash { &self.hash } +} + +impl From for WithdrawElement { + fn from(w: publication::Withdraw) -> Self { + let (_tag, uri, hash) = w.unwrap(); + WithdrawElement { uri, hash } + } +} + + + +#[derive(Clone, Debug, Deserialize, Serialize)] +pub struct Notification { + session: String, + serial: u64, + time: Time, + snapshot: SnapshotRef, + deltas: Vec, + old_refs: Vec<(Time, FileRef)> +} + +#[derive(Clone, Debug, Deserialize, Serialize)] +pub struct NotificationUpdate { + time: Time, + session: Option, + snapshot: SnapshotRef, + delta: DeltaRef, + last_delta: u64 +} + +impl NotificationUpdate { + pub fn new( + time: Time, + session: Option, + snapshot: SnapshotRef, + delta: DeltaRef, + last_delta: u64 + ) -> Self { + NotificationUpdate { time, session, snapshot, delta, last_delta } + } +} + +#[derive(Clone, Debug, Deserialize, Serialize)] +pub struct NotificationCreate { + session: String, + snapshot: SnapshotRef +} + +impl NotificationUpdate { + pub fn unwrap(self) -> (Time, Option, SnapshotRef, DeltaRef, u64) { + (self.time, self.session, self.snapshot, self.delta, self.last_delta) + } +} + +impl Notification { + pub fn old_refs(&self) -> &Vec<(Time, FileRef)> { + &self.old_refs + } + + pub fn update(&mut self, update: NotificationUpdate) { + let (time, session_opt, snapshot, delta, last_delta) = update.unwrap(); + if let Some(session) = session_opt { + self.session = session; + } + + self.serial += 1; + self.time = time; + + let mut refs_to_retire = vec![]; + + refs_to_retire.push((Time::now(), self.snapshot.clone())); + self.snapshot = snapshot; + + for d in &self.deltas { + if d.serial < last_delta { + refs_to_retire.push((Time::now(), d.file_ref.clone())); + } + } + + self.deltas.insert(0, delta); + self.deltas.retain(|delta| delta.serial >= last_delta); + self.old_refs.append(&mut refs_to_retire); + } + + /// Cleans up all old references from before the given time. + pub fn clean_up(&mut self, t: Time) { + self.old_refs.retain(|old_ref| {! old_ref.0.on_or_before(&t)}) + } + + pub fn create(session: String, snapshot: SnapshotRef) -> Self { + Notification { + session, + serial: 0, + time: Time::now(), + snapshot, + deltas: vec![], + old_refs: vec![] + } + } + + pub fn write_xml(&self, path: &PathBuf) -> Result<(), io::Error> { + debug!("Writing notification file: {}", path.to_string_lossy()); + let mut file = file::create_file_with_path(&path)?; + + XmlWriter::encode_to_file(& mut file, |w| { + + let a = [ + ("xmlns", NS), + ("version", VERSION), + ("session_id", self.session.as_ref()), + ("serial", &format!("{}", self.serial)), + ]; + + w.put_element( + "notification", + Some(&a), + |w| { + { + // snapshot ref + let uri = self.snapshot.uri.to_string(); + let a = [ + ("uri", uri.as_str()), + ("hash", self.snapshot.hash.as_ref()) + ]; + w.put_element( + "snapshot", + Some(&a), + |w| { w.empty() } + )?; + } + + { + // delta refs + for delta in &self.deltas { + let serial = format!("{}", delta.serial); + let uri = delta.file_ref.uri.to_string(); + let a = [ + ("serial", serial.as_ref()), + ("uri", uri.as_str()), + ("hash", delta.file_ref.hash.as_ref()) + ]; + w.put_element( + "delta", + Some(&a), + |w| { w.empty() } + )?; + } + } + + Ok(()) + } + ) + })?; + + Ok(()) + + } + +} + +#[derive(Clone, Debug, Deserialize, Serialize)] +pub struct FileRef { + #[serde( + deserialize_with = "ext_serde::de_http_uri", + serialize_with = "ext_serde::ser_http_uri")] + uri: uri::Http, + path: PathBuf, + hash: EncodedHash, +} + +impl FileRef { + pub fn new(uri: uri::Http, path: PathBuf, hash: EncodedHash) -> Self { + FileRef { uri, path, hash } + } + pub fn uri(&self) -> &uri::Http { &self.uri } + pub fn path(&self) -> &PathBuf { &self.path } + pub fn hash(&self) -> &EncodedHash { &self.hash } +} + +pub type SnapshotRef = FileRef; + +#[derive(Clone, Debug, Deserialize, Serialize)] +pub struct DeltaRef { + serial: u64, + file_ref: FileRef +} + +impl DeltaRef { + pub fn new(serial: u64, file_ref: FileRef) -> Self { + DeltaRef { serial, file_ref } + } + + pub fn serial(&self) -> u64 { self.serial } +} + +impl AsRef for DeltaRef { + fn as_ref(&self) -> &FileRef { + &self.file_ref + } +} + + +//------------ CurrentObjects ------------------------------------------------ + +/// Defines a current set of published elements. +/// +// Note this is mapped internally for speedy access, by hash, rather than uri +// for two reasons: +// a) URIs in RPKI may change in future +// b) The publish element as it appears in an RFC8182 snapshot.xml includes +// the uri and the base64, but not the hash. So keeping the actual elements +// around means we can be more efficient in producing that output. +#[derive(Clone, Debug, Deserialize, Serialize)] +pub struct CurrentObjects(HashMap); + +impl Default for CurrentObjects { + fn default() -> Self { + CurrentObjects(HashMap::new()) + } +} + +impl CurrentObjects { + fn elements(&self) -> Vec<&PublishElement> { + let mut res = vec![]; + for el in self.0.values() { + res.push(el) + } + res + } +} + + +//------------ VerificationError --------------------------------------------- + +/// Issues with relation to verifying deltas. +#[derive(Clone, Debug, Display)] +pub enum VerificationError { + #[display(fmt="Publishing ({}) outside of jail URI ({}) is not allowed.", _0, _1)] + UriOutsideJail(uri::Rsync, uri::Rsync), + + #[display(fmt="File already exists for uri (use update!): {}", _0)] + ObjectAlreadyPresent(uri::Rsync), + + #[display(fmt="File does not match hash at uri: {}", _0)] + NoObjectForHashAndOrUri(uri::Rsync), +} + +impl VerificationError { + fn outside(jail: &uri::Rsync, uri: &uri::Rsync) -> Self { + VerificationError::UriOutsideJail(uri.clone(), jail.clone()) + } + + fn present(uri: &uri::Rsync) -> Self { + VerificationError::ObjectAlreadyPresent(uri.clone()) + } + + fn no_match(uri: &uri::Rsync) -> Self { + VerificationError::NoObjectForHashAndOrUri(uri.clone()) + } +} + +impl CurrentObjects { + + fn has_match( + &self, + hash: &EncodedHash, + uri: &uri::Rsync + ) -> bool { + match self.0.get(hash) { + Some(el) => el.uri() == uri, + None => false + } + } + + pub fn verify_delta( + &self, + delta: &DeltaElements, + jail: &uri::Rsync + ) -> Result<(), VerificationError> { + + for p in delta.publishes() { + if ! jail.is_parent_of(p.uri()) { + return Err(VerificationError::outside(jail, p.uri())) + } + let hash = p.base64().to_encoded_hash(); + if self.0.contains_key(&hash) { + return Err(VerificationError::present(p.uri())) + } + } + + for u in delta.updates() { + if ! self.has_match(u.hash(), u.uri()) { + return Err(VerificationError::no_match(u.uri())); + } + } + + for w in delta.withdraws() { + if ! self.has_match(w.hash(), w.uri()) { + return Err(VerificationError::no_match(w.uri())); + } + } + + Ok(()) + } + + /// Applies a delta to CurrentObjects. This will asume that the delta + /// contains only valid updates for this delta. + pub fn apply_delta(&mut self, delta: DeltaElements) { + let (publishes, updates, withdraws) = delta.unwrap(); + + for p in publishes { + let hash = p.base64().to_encoded_hash(); + self.0.insert(hash, p); + } + + for u in updates { + self.0.remove(u.hash()); + let p: PublishElement = u.into(); + let hash = p.base64().to_encoded_hash(); + self.0.insert(hash, p); + } + + for w in withdraws { + self.0.remove(w.hash()); + } + } + + pub fn len(&self) -> usize { self.0.len() } + + pub fn is_empty(&self) -> bool { self.0.is_empty() } + + pub fn to_list_reply(&self) -> publication::ListReply { + let elements = self.0.iter().map(|el| { + let hash = el.0.clone(); + let uri = el.1.uri().clone(); + publication::ListElement::new(uri, hash) + }).collect(); + + publication::ListReply::new(elements) + } +} + + +//------------ Snapshot ------------------------------------------------------ + +/// A structure to contain the RRDP snapshot data. +#[derive(Clone, Debug, Deserialize, Serialize)] +pub struct Snapshot { + session: String, + serial: u64, + current_objects: CurrentObjects +} + +impl Snapshot { + pub fn new(session: String) -> Self { + let current_objects = CurrentObjects::default(); + Snapshot { session, serial: 0, current_objects } + } + + pub fn apply_delta(&mut self, delta: Delta) { + let (session, serial, elements) = delta.unwrap(); + self.session = session; + self.serial = serial; + self.current_objects.apply_delta(elements) + } + + pub fn len(&self) -> usize { self.current_objects.len() } + + pub fn is_empty(&self) -> bool { self.current_objects.is_empty() } + + pub fn write_xml(&self, path: &PathBuf) -> Result { + let vec = XmlWriter::encode_vec(|w| { + let a = [ + ("xmlns", NS), + ("version", VERSION), + ("session_id", self.session.as_ref()), + ("serial", &format!("{}", self.serial)), + ]; + + w.put_element( + "snapshot", + Some(&a), + |w| { + for el in self.current_objects.elements() { + let uri = el.uri.to_string(); + let atr = [ ("uri", uri.as_ref())]; + w.put_element( + "publish", + Some(&atr), + |w| { + w.put_text(el.base64.as_ref()) + } + )?; + } + Ok(()) + } + ) + }); + let bytes = Bytes::from(vec); + + file::save(&bytes, path)?; + let hash = EncodedHash::from_content(&bytes); + + Ok(hash) + } +} + + +//------------ DeltaElements ------------------------------------------------- + +/// Defines the elements for an RRDP delta. +#[derive(Clone, Debug, Deserialize, Serialize)] +pub struct DeltaElements { + publishes: Vec, + updates: Vec, + withdraws: Vec +} + +impl From for DeltaElements { + fn from(d: publication::PublishDelta) -> Self { + let (pbls, upds, wdrs) = d.unwrap(); + + let publishes = pbls.into_iter().map(PublishElement::from).collect(); + let updates = upds.into_iter().map(UpdateElement::from).collect(); + let withdraws = wdrs.into_iter().map(WithdrawElement::from).collect(); + + DeltaElements { publishes, updates, withdraws } + } +} + +impl DeltaElements { + pub fn unwrap( + self + ) -> (Vec, Vec, Vec) { + (self.publishes, self.updates, self.withdraws) + } + + pub fn len(&self) -> usize { + self.publishes.len() + self.updates.len() + self.withdraws.len() + } + + pub fn is_empty(&self) -> bool { + self.len() == 0 + } + + pub fn publishes(&self) -> &Vec { + &self.publishes + } + + pub fn updates(&self) -> &Vec { + &self.updates + } + + pub fn withdraws(&self) -> &Vec { + &self.withdraws + } +} + + +//------------ Delta --------------------------------------------------------- + +/// Defines an RRDP delta. +#[derive(Clone, Debug, Deserialize, Serialize)] +pub struct Delta { + session: String, + serial: u64, + time: Time, + elements: DeltaElements +} + +impl Delta { + pub fn new( + session: String, + serial: u64, + elements: DeltaElements + ) -> Self { + Delta { session, time: Time::now(), serial, elements } + } + + pub fn session(&self) -> &str { &self.session } + pub fn serial(&self) -> u64 { self.serial } + pub fn time(&self) -> &Time { &self.time } + pub fn elements(&self) -> &DeltaElements { &self.elements } + + /// Total number of elements + /// + /// This is a cheap approximation of the size of the delta that can help + /// in determining the choice of how many deltas to include in a + /// notification file. + pub fn len(&self) -> usize { self.elements.len() } + + pub fn is_empty(&self) -> bool { self.elements.is_empty() } + + pub fn unwrap(self) -> (String, u64, DeltaElements) { + (self.session, self.serial, self.elements) + } + + pub fn write_xml(&self, path: &PathBuf) -> Result { + + let vec = XmlWriter::encode_vec(|w| { + let a = [ + ("xmlns", NS), + ("version", VERSION), + ("session_id", self.session.as_ref()), + ("serial", &format!("{}", self.serial)), + ]; + + w.put_element( + "delta", + Some(&a), + |w| { + for el in &self.elements.publishes { + let uri = el.uri.to_string(); + let atr = [ ("uri", uri.as_ref())]; + w.put_element( + "publish", + Some(&atr), + |w| { + w.put_text(el.base64.as_ref()) + } + )?; + } + + for el in &self.elements.updates { + let uri = el.uri.to_string(); + let atr = [ + ("uri", uri.as_ref()), + ("hash", el.hash.as_ref()) + ]; + w.put_element( + "publish", + Some(&atr), + |w| { + w.put_text(el.base64.as_ref()) + } + )?; + } + + for el in &self.elements.withdraws { + let uri = el.uri.to_string(); + let atr = [ + ("uri", uri.as_ref()), + ("hash", el.hash.as_ref()) + ]; + w.put_element( + "withdraw", + Some(&atr), + |w| { w.empty() } + )?; + } + + Ok(()) + } + ) + }); + + let bytes = Bytes::from(vec); + file::save(&bytes, &path)?; + let hash = EncodedHash::from_content(&bytes); + + Ok(hash) + } + +} diff --git a/commons/src/eventsourcing/es_example.rs b/commons/src/eventsourcing/es_example.rs new file mode 100644 index 00000000..2fe06f2e --- /dev/null +++ b/commons/src/eventsourcing/es_example.rs @@ -0,0 +1,431 @@ +//! Example implementation using the eventsourcing module. +//! +//! Goal is two-fold: document using a simple domain, and test the module. +//! +use super::*; + + +//------------ InitPersonEvent ----------------------------------------------- + +/// Every aggregate defines their own initialisation event. This is the first +/// event stored for an instance. +/// +/// Here we define a type wrapping around the generic StoredEvent, so we only +/// need to define the unique initialisation details. +type InitPersonEvent = StoredEvent; + +impl InitPersonEvent { + + pub fn init(id: &AggregateId, name: &str) -> Self { + StoredEvent::new(id, 0, InitPersonDetails { name: name.to_string()}) + } +} + +#[derive(Clone, Deserialize, Serialize)] +struct InitPersonDetails { + pub name: String +} + + +//------------ InitPersonEvent ----------------------------------------------- + +/// Every aggregate defines their own set of events - i.e. state changes. The +/// state of an aggregate can only change when events are applied. And events +/// cannot have side effects. If they did, then replaying events would become +/// problematic. +/// +/// Here we make a type alias wrapped around the generic StoredEvent and +/// include an enum with event details specific for Persons. Furthermore we +/// provide an implementation for this type alias so that we can have some +/// convenience functions for creating these events. +type PersonEvent = StoredEvent; + +#[derive(Clone, Deserialize, Serialize)] +enum PersonEventDetails { + NameChanged(String), + HadBirthday +} + +impl PersonEvent { + pub fn had_birthday(p: &Person) -> Self { + StoredEvent::new(p.id(), p.version, PersonEventDetails::HadBirthday) + } + + pub fn name_changed(p: &Person, name: String) -> Self { + StoredEvent::new( + p.id(), + p.version, + PersonEventDetails::NameChanged(name)) + } +} + + +//------------ PersonCommand ------------------------------------------------- + +/// In order to change an aggregate a command is sent to it. The aggregate +/// will then validate the command and if there are no issues, it will return +/// a list (vec) of events that may be applied. This process in itself does +/// not change any state, the state of the aggregate is only changed when +/// those events are applied. +/// +/// Commands are not recorded. Only the resulting events are. For this reason +/// commands may have side-effects: e.g. write something to disk, send an +/// email, etc. +/// +/// Here we define a type wrapping around the generic SentCommand, so we only +/// need to provide an enum with specific command details. We also have an +/// implementation for this type alias providing some convenience methods. +type PersonCommand = SentCommand; + +#[derive(Clone, Deserialize, Serialize)] +enum PersonCommandDetails { + ChangeName(String), + GoAroundTheSun +} + +impl CommandDetails for PersonCommandDetails { + type Event = PersonEvent; +} + +impl PersonCommand { + + pub fn go_around_sun(id: &AggregateId, version: Option) -> Self { + Self::new(id, version, PersonCommandDetails::GoAroundTheSun) + } + + + pub fn change_name(id: &AggregateId, version: Option, s: &str) -> Self { + let details = PersonCommandDetails::ChangeName(s.to_string()); + Self::new(id, version, details) + } +} + +//------------ PersonError --------------------------------------------------- + +/// Errors specific to the Person aggregate, should only ever be returned when +/// applying a command that does not validate. +#[derive(Clone, Debug, Display)] +enum PersonError { + #[display(fmt = "No person can live longer than 255 years")] + TooOld +} + +impl std::error::Error for PersonError {} + + +//------------ PersonResult -------------------------------------------------- + +/// A shorthand for the result type returned by the process_command function +/// of the Person aggregate. +type PersonResult = Result, PersonError>; + + +//------------ Person ------------------------------------------------------ + +/// Defines a person object. Persons have a name and an age. +/// +#[derive(Clone, Deserialize, Serialize)] +struct Person { + /// The id is needed when generating events. + id: AggregateId, + + /// The version of for this particular Person. Versions + /// are incremented whenever events are applied. They are + /// used to store those and apply events in the correct + /// sequence, as well as to detect concurrency issues when + /// a command is sent. + version: u64, + + name: String, + age: u8 +} + +impl Person { + pub fn id(&self) -> &AggregateId { &self.id } + pub fn version(&self) -> u64 { self.version } + pub fn name(&self) -> &String { &self.name } + pub fn age(&self) -> u8 { self.age } +} + +impl Aggregate for Person { + type Command = PersonCommand; + type Event = PersonEvent; + type InitEvent = InitPersonEvent; + type Error = PersonError; + + fn init(event: InitPersonEvent) -> Result { + let (id, _version, init) = event.unwrap(); + Ok(Person { + id, version: 1, name: init.name, age: 0 + }) + } + + fn version(&self) -> u64 { + self.version + } + + fn apply(&mut self, event: PersonEvent) { + match event.into_details() { + PersonEventDetails::NameChanged(name) => { self.name = name }, + PersonEventDetails::HadBirthday => { self.age += 1 } + } + self.version += 1; + } + + fn process_command(&self, command: Self::Command) -> PersonResult { + match command.into_details() { + PersonCommandDetails::ChangeName(name) => { + let event = PersonEvent::name_changed(&self, name); + Ok(vec![event]) + }, + PersonCommandDetails::GoAroundTheSun => { + if self.age == 255 { + Err(PersonError::TooOld) + } else { + let event = PersonEvent::had_birthday(&self); + Ok(vec![event]) + } + } + } + } +} + + +/// This type is responsible for managing all persons. I.e. creating new +/// instances, returning a reference for reading, dispatching commands, and +/// storing them. +/// +/// It is generic over the keystore used. +/// +// Compiler does not see this is used in test +#[allow(dead_code)] +struct PersonManager { + /// Here we use a cache to make matters complicated^H interesting.. + /// Of course this may not always be the best idea.. + cache: RwLock>>, + + /// The keystore where snapshots and events may be retrieved and stored. + store: S +} + +impl PersonManager { + + // Compiler does not see this is used in test + #[allow(dead_code)] + pub fn new(store: S) -> Self { + let values = RwLock::new(HashMap::new()); + PersonManager { + cache: values, store + } + } + + // Compiler does not see this is used in test + #[allow(dead_code)] + + fn update_cache( + &self, + id: &AggregateId, + mut force: bool + ) -> Result<(), PersonManagerError> { + + let mut cache = self.cache.write().unwrap(); + + let mut has_key = cache.contains_key(id); + + if ! has_key { + let init_key = S::key_for_event(0); + if let Some(init) = self.store.get(id, &init_key)? { + let mut agg = Person::init(init)?; + + cache.insert(id.clone(), Arc::new(agg)); + force = true; + has_key = true; + } + } + + if has_key && force { + // We MUST have an entry now + let arc = cache.get_mut(id).unwrap(); + let agg = Arc::make_mut(arc); + + loop { + let ver = agg.version(); + let key = S::key_for_event(ver); + if let Some(event) = self.store.get(id, &key)? { + agg.apply(event) + } else { + break + } + } + } + + Ok(()) + } + + + /// Get a reference to the latest version of the aggregate. + // Compiler does not see this is used in test + #[allow(dead_code)] + + fn get_latest( + &self, + id: &AggregateId + ) -> Result>, PersonManagerError> { + self.update_cache(id, false)?; + Ok(self.cache.read().unwrap().get(id).cloned()) + } + + // Compiler does not see this is used in test + #[allow(dead_code)] + + fn create( + &self, + id: &AggregateId, + event: InitPersonEvent + ) -> Result<(), PersonManagerError> { + self.update_cache(id, true)?; + + let mut cache = self.cache.write().unwrap(); + if cache.contains_key(id) { + Err(PersonManagerError::AggregateAlreadyExists) + } else { + let key = S::key_for_event(0); + self.store.store(id, &key, &event)?; + + let agg = Person::init(event)?; + + cache.insert(id.clone(), Arc::new(agg)); + Ok(()) + } + } + + /// Apply a command to the latest aggregate, save the events and return + /// the updated aggregate. + // Compiler does not see this is used in test + #[allow(dead_code)] + fn apply( + &self, + command: PersonCommand + ) -> Result<(), PersonManagerError> { + let id = command.id().clone(); + self.update_cache(&id, true)?; + + let mut cache = self.cache.write().unwrap(); + + match cache.get_mut(&id) { + None => Err(PersonManagerError::AggregateDoesNotExist), + Some(agg) => { + + let agg = Arc::make_mut(agg); + + if let Some(version) = command.version() { + if version != agg.version() { + // TODO check conflicts + return Err(PersonManagerError::ConcurrentModification) + } + } + + let events = agg.process_command(command)?; + + for e in events { + let key = S::key_for_event(e.version()); + self.store.store(&id, &key, &e)?; + + agg.apply(e); + } + + Ok(()) + } + } + } +} + + +//------------ PersonManagerError -------------------------------------------- + +// Compiler does not see this is used in test +#[allow(dead_code)] +#[derive(Debug, Display)] +enum PersonManagerError { + #[display(fmt = "Aggregate does not exist")] + AggregateDoesNotExist, + + #[display(fmt = "Aggregate already exists")] + AggregateAlreadyExists, + + #[display(fmt = "Concurrent modification. Command rejected")] + ConcurrentModification, + + #[display(fmt = "{}", _0)] + PersonError(PersonError), + + #[display(fmt = "{}", _0)] + KeyStoreError(KeyStoreError), +} + +impl From for PersonManagerError { + fn from(e: PersonError) -> Self { PersonManagerError::PersonError(e) } +} + +impl From for PersonManagerError { + fn from(e: KeyStoreError) -> Self { PersonManagerError::KeyStoreError(e) } +} + +//------------ Tests --------------------------------------------------------- + +#[cfg(test)] +mod tests { + + use super::*; + use crate::util::test; + + #[test] + fn test() { + test::test_with_tmp_dir(|d| { + + let storage = DiskKeyStore::under_work_dir(&d, "person").unwrap(); + let manager = PersonManager::new(storage); + + let id_alice = AggregateId::from("alice"); + let alice_init = InitPersonEvent::init(&id_alice, "alice smith"); + + manager.create(&id_alice, alice_init).unwrap(); + + let alice = manager.get_latest(&id_alice).unwrap().unwrap(); + assert_eq!(alice.name(), "alice smith"); + assert_eq!(alice.age(), 0); + + let mut age = 0; + loop { + manager.apply( + PersonCommand::go_around_sun(&id_alice, None) + ).unwrap(); + age = age + 1; + if age == 21 { + break + } + } + + + let alice = manager.get_latest(&id_alice).unwrap().unwrap(); + assert_eq!(alice.name(), "alice smith"); + assert_eq!(alice.age(), 21); + + manager.apply( + PersonCommand::change_name(&id_alice, Some(22), "alice smith-doe") + ).unwrap(); + + let alice = manager.get_latest(&id_alice).unwrap().unwrap(); + assert_eq!(alice.name(), "alice smith-doe"); + assert_eq!(alice.age(), 21); + + // Should read state from disk + let storage = DiskKeyStore::under_work_dir(&d, "person").unwrap(); + let manager = PersonManager::new(storage); + + let alice = manager.get_latest(&id_alice).unwrap().unwrap(); + assert_eq!(alice.name(), "alice smith-doe"); + assert_eq!(alice.age(), 21); + }) + } +} \ No newline at end of file diff --git a/commons/src/eventsourcing/mod.rs b/commons/src/eventsourcing/mod.rs new file mode 100644 index 00000000..eb7a90ac --- /dev/null +++ b/commons/src/eventsourcing/mod.rs @@ -0,0 +1,722 @@ +//! Event sourcing support for Krill + +mod es_example; // Example implementation and tests. + +use std::any::Any; +use std::collections::HashMap; +use std::fs; +use std::fmt; +use std::fs::File; +use std::io; +use std::io::Write; +use std::path::Path; +use std::path::PathBuf; +use std::sync::Arc; +use std::sync::RwLock; +use serde::Serialize; +use serde::de::DeserializeOwned; +use serde_json; +use crate::util::file; + +const SNAPSHOT_FREQ: u64 = 5; + +//------------ Storable ------------------------------------------------------ + +pub trait Storable: Clone + Serialize + DeserializeOwned + Sized {} +impl Storable for T { } + + +//------------ AggregateId --------------------------------------------------- + +#[derive(Clone, Debug, Deserialize, Eq, Hash, PartialEq, Serialize)] +pub struct AggregateId(String); + + +impl AggregateId { + pub fn as_str(&self) -> &str { + &self.0.as_str() + } +} + +impl From<&str> for AggregateId { + fn from(s: &str) -> Self { + AggregateId(s.to_string()) + } +} + +impl From for AggregateId { + fn from(s: String) -> Self { + AggregateId(s) + } +} + +impl AsRef for AggregateId { + fn as_ref(&self) -> &str { + self.as_str() + } +} + +impl AsRef for AggregateId { + fn as_ref(&self) -> &String { + &self.0 + } +} + +impl AsRef for AggregateId { + fn as_ref(&self) -> &Path { + self.0.as_ref() + } +} + +impl fmt::Display for AggregateId { + fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result { + self.0.fmt(f) + } +} + +//------------ Aggregate ----------------------------------------------------- + +pub trait Aggregate: Storable + Send + Sync + 'static { + + type Command: Command; + type Event: Event; + type InitEvent: Event; + type Error: std::error::Error; + + /// Creates a new instance. Expects an event with data needed to + /// initialise the instance. Typically this means that a specific + /// 'create' event is passed, with all the needed data, or just an empty + /// marker if no data is needed. Implementations must return an error in + /// case the instance cannot be created. + fn init(event: Self::InitEvent) -> Result; + + /// Returns the current version of the aggregate. + fn version(&self) -> u64; + + /// Applies the event to this. This MUST not result in any errors, and + /// this MUST be side-effect free. Applying the event just updates the + /// internal data of the aggregate. + /// + /// Note the event is moved. This is done because we want to avoid + /// doing additional allocations where we can. + fn apply(&mut self, event: Self::Event); + + /// Applies all events. Assumes that the list ordered, starting with the + /// oldest event, applicable, self.version matches the oldest event, and + /// contiguous, i.e. there are no missing events. + fn apply_all(&mut self, events: Vec) { + for event in events { + self.apply(event); + } + } + + /// Processes a command. I.e. validate the command, and return a list of + /// events that will result in the desired new state, but do not apply + /// these event here. + /// + /// The command is moved, because we want to enable moving its data + /// without reallocating. + fn process_command(&self, command: Self::Command) -> Result, Self::Error>; +} + + +//------------ Event -------------------------------------------------------- + +pub trait Event: Storable + 'static { + /// Identifies the aggregate, useful when storing and retrieving the event. + fn id(&self) -> &AggregateId; + + /// The version of the aggregate that this event updates. An aggregate that + /// is currently at version x, will get version x + 1, when the event for + /// version x is applied. + fn version(&self) -> u64; +} + +#[derive(Clone, Deserialize, Serialize)] +pub struct StoredEvent { + id: AggregateId, + version: u64, + #[serde(deserialize_with = "E::deserialize")] + details: E +} + +impl StoredEvent { + + pub fn new(id: &AggregateId, version: u64, event: E) -> Self { + StoredEvent { id: id.clone(), version, details: event } + } + + pub fn details(&self) -> &E { & self.details } + + pub fn into_details(self) -> E { self.details } + + /// Return the parts of this event. + pub fn unwrap(self) -> (AggregateId, u64, E) { + (self.id, self.version, self.details) + } +} + +impl Event for StoredEvent { + fn id(&self) -> &AggregateId { + &self.id + } + + fn version(&self) -> u64 { + self.version + } +} + +//------------ Command ------------------------------------------------------- + +/// Commands are used to send an intent to change an aggregate. +/// +/// Think of this as the data container for your update API, plus some +/// meta-data to ensure that the command is sent to the right instance of an +/// Aggregate, and that concurrency issues are handled. +pub trait Command: Storable { + /// Identify the type of event returned by the aggregate that uses this + /// command. This is needed because we may need to check whether a + /// command conflicts with recent events. + type Event: Event; + + /// Identifies the aggregate, useful when storing and retrieving the event. + fn id(&self) -> &AggregateId; + + /// The version of the aggregate that this command updates. If this + /// command should update whatever the latest version happens to be, then + /// use None here. + fn version(&self) -> Option; + + /// In case of concurrent processing of commands, the aggregate may be + /// outdated when a command is applied. In such cases this method expects + /// the list of events that happened since the ['affected_version'] and + /// will return whether there is a conflict. If there is no conflict that + /// the command may be applied again. + /// + /// Note that this defaults to true, which is the safe choice when in + /// doubt. If you choose to implement this, then you will also need to + /// implement the ['set_affected_version'] function. + fn conflicts(&self, _events: &[Self::Event]) -> bool { true } +} + + +//------------ SentCommand --------------------------------------------------- + +/// Convenience wrapper so that implementations can just implement +/// ['CommandDetails'] and leave the id and version boilerplate. +#[derive(Clone, Deserialize, Serialize)] +pub struct SentCommand { + id: AggregateId, + version: Option, + #[serde(deserialize_with = "C::deserialize")] + details: C +} + +impl Command for SentCommand { + type Event = C::Event; + + fn id(&self) -> &AggregateId { + &self.id + } + + fn version(&self) -> Option { + self.version + } +} + +impl SentCommand { + + pub fn new(id: &AggregateId, version: Option, details: C) -> Self { + SentCommand { id: id.clone(), version, details } + } + + pub fn into_details(self) -> C { self.details } +} + + +//------------ CommandDetails ------------------------------------------------ + +/// Implement this for an enum with CommandDetails, so you you can reuse the +/// id and version boilerplate from ['SentCommand']. +pub trait CommandDetails: Storable + 'static { + type Event: Event; +} + + +//------------ KeyStore ------------------------------------------------------ + +/// Generic KeyStore for AggregateManager +pub trait KeyStore { + + type Key; + + fn key_for_snapshot() -> Self::Key; + fn key_for_event(version: u64) -> Self::Key; + + /// Returns whether a key already exists. + + fn has_key(&self, id: &AggregateId, key: &Self::Key) -> bool; + + + fn has_aggregate(&self, id: &AggregateId) -> bool; + + fn aggregates(&self) -> Vec; // Use Iterator? + + /// Throws an error if the key already exists. + + fn store( + &self, + id: &AggregateId, + key: &Self::Key, + value: &V + ) -> Result<(), KeyStoreError>; + + /// Get the value for this key, if any exists. + + fn get( + &self, + id: &AggregateId, + key: &Self::Key + ) -> Result, KeyStoreError>; + + /// Get the value for this key, if any exists. + + fn get_event( + &self, + id: &AggregateId, + version: u64 + ) -> Result, KeyStoreError>; + + fn store_event( + &self, + event: &V + ) -> Result<(), KeyStoreError>; + + /// Get the latest aggregate + + fn get_aggregate( + &self, + id: &AggregateId + ) -> Result, KeyStoreError>; + + /// Saves the latest snapshot - overwrites any previous snapshot. + + fn store_aggregate( + &self, + id: &AggregateId, + aggregate: &V + ) -> Result<(), KeyStoreError>; +} + + +//------------ KeyStoreError ------------------------------------------------- + +/// This type defines possible Errors for KeyStore +#[derive(Debug, Display)] +pub enum KeyStoreError { + #[display(fmt = "{}", _0)] + IoError(io::Error), + + #[display(fmt = "{}", _0)] + JsonError(serde_json::Error), + + #[display(fmt = "Key already exists: {}", _0)] + KeyExists(String), + + #[display(fmt = "Aggregate init event exists, but cannot be applied")] + InitError +} + +impl From for KeyStoreError { + fn from(e: io::Error) -> Self { KeyStoreError::IoError(e) } +} + +impl From for KeyStoreError { + fn from(e: serde_json::Error) -> Self { KeyStoreError::JsonError(e) } +} + +impl std::error::Error for KeyStoreError { } + + +//------------ DiskKeyStore -------------------------------------------------- + +/// This type can store and retrieve values to/from disk, using json +/// serialization. +pub struct DiskKeyStore { + dir: PathBuf, +} + +impl KeyStore for DiskKeyStore { + type Key = PathBuf; + + fn key_for_snapshot() -> Self::Key { + PathBuf::from("snapshot.json") + } + + fn key_for_event(version: u64) -> Self::Key { + PathBuf::from(format!("delta-{}.json", version)) + } + + fn has_key(&self, id: &AggregateId, key: &Self::Key) -> bool { + self.file_path(id, key).exists() + } + + fn has_aggregate(&self, id: &AggregateId) -> bool { + self.dir_for_aggregate(id).exists() + } + + fn aggregates(&self) -> Vec { + let mut res: Vec = Vec::new(); + + if let Ok(dir) = fs::read_dir(&self.dir) { + for d in dir { + let full_path = d.unwrap().path(); + let path = full_path.file_name().unwrap(); + + let id = AggregateId::from(path.to_string_lossy().as_ref()); + res.push(id); + } + } + + res + } + + fn store( + &self, + id: &AggregateId, + key: &Self::Key, + value: &V + ) -> Result<(), KeyStoreError> { + let mut f = file::create_file_with_path(&self.file_path(id, key))?; + let json = serde_json::to_string(value)?; + f.write_all(json.as_ref())?; + Ok(()) + } + + fn get( + &self, + id: &AggregateId, + key: &Self::Key + ) -> Result, KeyStoreError> { + if self.has_key(id, key) { + let f = File::open(self.file_path(id, key))?; + let v: V = serde_json::from_reader(f)?; + Ok(Some(v)) + } else { + Ok(None) + } + } + + /// Get the value for this key, if any exists. + fn get_event( + &self, + id: &AggregateId, + version: u64 + ) -> Result, KeyStoreError> { + let path = self.path_for_event(id, version); + if path.exists() { + let f = File::open(path)?; + let v: V = serde_json::from_reader(f)?; + Ok(Some(v)) + } else { + Ok(None) + } + } + + fn store_event( + &self, + event: &V + ) -> Result<(), KeyStoreError> { + let id = event.id(); + let key = Self::key_for_event(event.version()); + if self.has_key(id, &key) { + Err(KeyStoreError::KeyExists(key.to_string_lossy().to_string())) + } else { + self.store(id, &key, event) + } + } + + fn get_aggregate( + &self, + id: &AggregateId + ) -> Result, KeyStoreError> { + // try to get a snapshot. + // If that fails, try to get the init event. + // Then replay all newer events that can be found. + let key = Self::key_for_snapshot(); + let aggregate_opt = match self.get::(id, &key)? { + Some(aggregate) => Some(aggregate), + None => { + match self.get_event::(id, 0)? { + Some(e) => Some(V::init(e).map_err(|_|KeyStoreError::InitError)?), + None => None + } + } + }; + + match aggregate_opt { + None => Ok(None), + Some(mut aggregate) => { + self.update_aggregate(id, &mut aggregate)?; + Ok(Some(aggregate)) + } + } + } + + fn store_aggregate( + &self, + id: &AggregateId, + aggregate: &V + ) -> Result<(), KeyStoreError> { + let key = Self::key_for_snapshot(); + self.store(id, &key, aggregate) + } +} + +impl DiskKeyStore { + pub fn new(work_dir: &PathBuf, name_space: &str) -> Self { + let mut dir = work_dir.clone(); + dir.push(name_space); + DiskKeyStore { dir } + } + + /// Creates a directory for the name_space under the work_dir. + pub fn under_work_dir( + work_dir: &PathBuf, + name_space: &str + ) -> Result { + let mut path = work_dir.clone(); + path.push(name_space); + if ! path.is_dir() { + fs::create_dir_all(&path)?; + } + Ok(Self::new(work_dir, name_space)) + } + + fn file_path(&self, id: &AggregateId, key: &::Key) -> PathBuf { + let mut file_path = self.dir_for_aggregate(id); + file_path.push(key); + file_path + } + + fn dir_for_aggregate(&self, id: &AggregateId) -> PathBuf { + let mut dir_path = self.dir.clone(); + dir_path.push(id); + dir_path + } + + fn path_for_event( + &self, + id: &AggregateId, + version: u64 + ) -> PathBuf { + let mut file_path = self.dir_for_aggregate(id); + file_path.push(format!("delta-{}.json", version)); + file_path + } + + fn update_aggregate( + &self, + id: &AggregateId, + aggregate: &mut A + ) -> Result<(), KeyStoreError> { + while let Some(e) = self.get_event(id, aggregate.version())? { + aggregate.apply(e); + } + Ok(()) + } + +} + + +pub type StoreResult = Result; + +pub trait AggregateStore: Send + Sync { + /// Gets the latest version for the given aggregate. Returns + /// an AggregateStoreError::UnknownAggregate in case the aggregate + /// does not exist. + fn get_latest(&self, id: &AggregateId) -> StoreResult>; + + /// Adds a new aggregate instance based on the init event. + fn add(&self, id: &AggregateId, init: A::InitEvent) -> StoreResult<()>; + + /// Updates the aggregate instance in the store. Expects that the + /// Arc retrieved using 'get_latest' is moved here, so clone on + /// writes can be avoided, and a verification can be done that there + /// is no concurrent modification. Returns the updated instance if all + /// is well, or an AggregateStoreError::ConcurrentModification if you + /// try to update an outdated instance. + fn update(&self, id: &AggregateId, agg: Arc, events: Vec) -> StoreResult>; + + /// Returns true if an instance exists for the id + fn has(&self, id: &AggregateId) -> bool; + + /// Lists all known ids. + fn list(&self) -> Vec; +} + + +/// This type defines possible Errors for the AggregateStore +#[derive(Debug, Display)] +pub enum AggregateStoreError { + #[display(fmt = "{}", _0)] + KeyStoreError(KeyStoreError), + + #[display(fmt = "Unknown aggregate: {}", _0)] + UnknownAggregate(AggregateId), + + #[display(fmt = "Aggregate init event exists, but cannot be applied")] + InitError, + + #[display(fmt = "Event not applicable to aggregate, id or version is off")] + WrongEventForAggregate, + + #[display(fmt = "Trying to update outdated aggregate")] + ConcurrentModification, +} + +impl From for AggregateStoreError { + fn from(e: KeyStoreError) -> Self { AggregateStoreError::KeyStoreError(e) } +} + + +pub struct DiskAggregateStore { + store: DiskKeyStore, + cache: RwLock>>, + use_cache: bool +} + +impl DiskAggregateStore { + pub fn new(work_dir: &PathBuf, name_space: &str) -> Result { + let store = DiskKeyStore::under_work_dir(work_dir, name_space)?; + let cache = RwLock::new(HashMap::new()); + let use_cache = true; + Ok(DiskAggregateStore { store, cache, use_cache }) + } +} + +impl DiskAggregateStore { + fn has_updates( + &self, + id: &AggregateId, + aggregate: &A + ) -> StoreResult { + Ok(self.store.get_event::(id, aggregate.version())?.is_some()) + } + + fn cache_get(&self, id: &AggregateId) -> Option> { + if self.use_cache { + self.cache.read().unwrap().get(id).cloned() + } else { + None + } + } + + fn cache_update(&self, id: &AggregateId, arc: Arc) { + if self.use_cache { + self.cache.write().unwrap().insert(id.clone(), arc); + } + } +} + +impl AggregateStore for DiskAggregateStore { + fn get_latest(&self, id: &AggregateId) -> StoreResult> { + match self.cache_get(id) { + None => { + match self.store.get_aggregate(id)? { + None => Err(AggregateStoreError::UnknownAggregate(id.clone())), + Some(agg) => { + let arc: Arc = Arc::new(agg); + self.cache_update(id, arc.clone()); + Ok(arc) + } + } + }, + Some(mut arc) => { + if self.has_updates(id, &arc)? { + let agg = Arc::make_mut(&mut arc); + self.store.update_aggregate(id, agg)?; + } + Ok(arc) + } + } + } + + fn add(&self, id: &AggregateId, init: A::InitEvent) -> StoreResult<()> { + self.store.store_event(&init)?; + + let aggregate = A::init(init).map_err(|_| AggregateStoreError::InitError)?; + self.store.store_aggregate(id, &aggregate)?; + + let arc = Arc::new(aggregate); + self.cache_update(id, arc); + + Ok(()) + } + + + fn update(&self, id: &AggregateId, prev: Arc, events: Vec) -> StoreResult> { + // Get the latest arc. + let mut latest = self.get_latest(id)?; + + { + // Verify whether there is a concurrency issue + if prev.version() != latest.version() { + return Err(AggregateStoreError::ConcurrentModification) + } + + // forget the previous version + std::mem::forget(prev); + + // make the arc mutable, hopefully forgetting prev will avoid the clone + let agg = Arc::make_mut(&mut latest); + + // Using a lock on the hashmap here to ensure that all updates happen sequentially. + // It would be better to get a lock only for this specific aggregate. So it may be + // worth rethinking the structure. + // + // That said.. saving and applying events is really quick, so this should not hurt + // performance much. + // + // Also note that we don't need the lock to update the inner arc in the cache. We + // just need it to be in scope until we are done updating. + let _write_lock = self.cache.write().unwrap(); + + // There is a possible race condition. We may only have obtained the lock + if self.has_updates(id, &agg)? { + self.store.update_aggregate(id, agg)?; + } + + let version_before = agg.version(); + let nr_events = events.len() as u64; + + for i in 0..nr_events { + let event = &events[i as usize]; + if event.version() != version_before + i || event.id() != id { + return Err(AggregateStoreError::WrongEventForAggregate); + } + } + + for event in events { + self.store.store_event(&event)?; + agg.apply(event); + if agg.version() % SNAPSHOT_FREQ == 0 { + self.store.store_aggregate(id, agg)?; + } + } + } + + Ok(latest) + } + + fn has(&self, id: &AggregateId) -> bool { + self.store.has_aggregate(id) + } + + fn list(&self) -> Vec { + self.store.aggregates() + } +} + + + diff --git a/commons/src/lib.rs b/commons/src/lib.rs new file mode 100644 index 00000000..13063f12 --- /dev/null +++ b/commons/src/lib.rs @@ -0,0 +1,24 @@ +//! Common types used by the various Krill components. + +extern crate actix; +extern crate actix_web; +extern crate base64; +extern crate bytes; +extern crate chrono; +#[macro_use] extern crate derive_more; +extern crate futures; +extern crate hex; +#[macro_use] extern crate log; +extern crate openssl; +extern crate rand; +extern crate reqwest; +extern crate rpki; +#[macro_use] extern crate serde_derive; +extern crate serde; +extern crate serde_json; +extern crate syslog; +extern crate xml as xmlrs; + +pub mod api; +pub mod eventsourcing; +pub mod util; diff --git a/commons/src/util/actix.rs b/commons/src/util/actix.rs new file mode 100644 index 00000000..17df8b64 --- /dev/null +++ b/commons/src/util/actix.rs @@ -0,0 +1,100 @@ +//! Support conversions from actix requests to data types in this crate. + +use actix_web::FromRequest; +use actix_web::HttpResponse; +use actix_web::dev::MessageBody; +use actix_web::http::StatusCode; +use futures::Future; +use crate::api::admin::PublisherRequest; +use crate::api::admin::PublisherHandle; +use crate::api::publication::PublishDelta; + + +//------------ PublisherRequest ---------------------------------------------- + +/// Converts the body sent to 'add publisher' end-points to a +/// PublisherRequestChoice, which contains either an +/// rfc8183::PublisherRequest, or an API publisher request (no ID certs and +/// CMS etc). +impl FromRequest for PublisherRequest { + type Config = (); + type Result = Box>; + + fn from_request( + req: &actix_web::HttpRequest, + _cfg: &Self::Config + ) -> Self::Result { + Box::new(MessageBody::new(req) + .from_err() + .and_then(|bytes| { + let p: PublisherRequest = + serde_json::from_reader(bytes.as_ref()) + .map_err(Error::JsonError)?; + Ok(p) + }) + ) + } +} + + +//------------ PublisherHandle ----------------------------------------------- + +impl FromRequest for PublisherHandle { + type Config = (); + type Result = Result; + + fn from_request( + req: &actix_web::HttpRequest, + _cfg: &Self::Config + ) -> Self::Result { + if let Some(handle) = req.match_info().get("handle") { + Ok(PublisherHandle::from(handle)) + } else { + Err(Error::InvalidHandle.into()) + } + } +} + + +//------------ PublishDelta -------------------------------------------------- + +/// Support converting request body into PublishDelta +impl FromRequest for PublishDelta { + type Config = (); + type Result = Box>; + + fn from_request( + req: &actix_web::HttpRequest, + _cfg: &Self::Config + ) -> Self::Result { + Box::new(MessageBody::new(req).limit(255 * 1024 * 1024) // up to 256MB + .from_err() + .and_then(|bytes| { + let delta: PublishDelta = + serde_json::from_reader(bytes.as_ref())?; + Ok(delta) + }) + ) + } +} + +//------------ Error --------------------------------------------------------- + +#[derive(Debug, Display)] +#[allow(clippy::large_enum_variant)] +pub enum Error { + #[display(fmt = "{}", _0)] + JsonError(serde_json::Error), + + #[display(fmt = "Invalid handle")] + InvalidHandle, +} + +impl std::error::Error for Error {} + +impl actix_web::ResponseError for Error { + fn error_response(&self) -> HttpResponse { + HttpResponse::build(StatusCode::INTERNAL_SERVER_ERROR) + .body(format!("{}", self)) + } +} diff --git a/commons/src/util/ext_serde.rs b/commons/src/util/ext_serde.rs new file mode 100644 index 00000000..d1a391a2 --- /dev/null +++ b/commons/src/util/ext_serde.rs @@ -0,0 +1,81 @@ +//! Defines helper methods for Serializing and Deserializing external types. +use base64; +use bytes::Bytes; +use log::LevelFilter; +use rpki::uri; +use serde::{Deserialize, Deserializer, Serialize, Serializer}; +use serde::de; +use syslog::Facility; + + +//------------ Bytes --------------------------------------------------------- + +pub fn de_bytes<'de, D>(d: D) -> Result +where D: Deserializer<'de> +{ + let some = String::deserialize(d)?; + let dec = base64::decode(&some).map_err(de::Error::custom)?; + Ok(Bytes::from(dec)) +} + +pub fn ser_bytes(b: &Bytes, s: S) -> Result +where S: Serializer +{ + base64::encode(b).serialize(s) +} + + +//------------ uri::Rsync ---------------------------------------------------- + +pub fn de_rsync_uri<'de, D>(d: D) -> Result +where D: Deserializer<'de> +{ + let some = String::deserialize(d)?; + uri::Rsync::from_string(some).map_err(de::Error::custom) +} + +pub fn ser_rsync_uri(uri: &uri::Rsync, s: S) -> Result +where S: Serializer +{ + uri.to_string().serialize(s) +} + + +//------------ uri::Http ----------------------------------------------------- + +pub fn de_http_uri<'de, D>(d: D) -> Result +where D: Deserializer<'de> +{ + let some = String::deserialize(d)?; + uri::Http::from_string(some).map_err(de::Error::custom) +} + +pub fn ser_http_uri(uri: &uri::Http, s: S) -> Result +where S: Serializer +{ + uri.to_string().serialize(s) +} + + +//------------ LevelFilter --------------------------------------------------- + +pub fn de_level_filter<'de, D>(d: D) -> Result +where D: Deserializer<'de> +{ + use std::str::FromStr; + let string = String::deserialize(d)?; + LevelFilter::from_str(&string).map_err(de::Error::custom) +} + + +//------------ Facility ------------------------------------------------------ + +pub fn de_facility<'de, D>(d: D) -> Result + where D: Deserializer<'de> +{ + use std::str::FromStr; + let string = String::deserialize(d)?; + Facility::from_str(&string).map_err( + |_| { de::Error::custom( + format!("Unsupported syslog_facility: \"{}\"", string))}) +} diff --git a/commons/src/util/file.rs b/commons/src/util/file.rs new file mode 100644 index 00000000..17cbed33 --- /dev/null +++ b/commons/src/util/file.rs @@ -0,0 +1,390 @@ +use std::fs; +use std::fs::File; +use std::io::{self, Read, Write}; +use std::path::PathBuf; +use std::str::FromStr; +use bytes::Bytes; +use rpki::uri; +use crate::api::{ Base64, EncodedHash }; +use crate::api::publication; +use crate::util::ext_serde; +use serde::Serialize; +use serde::de::DeserializeOwned; + + +/// Creates a sub dir if needed, return full path to it +pub fn sub_dir(base: &PathBuf, name: &str) -> Result { + let mut full_path = base.clone(); + full_path.push(name); + create_dir(&full_path)?; + Ok(full_path) +} + +pub fn create_dir(dir: &PathBuf) -> Result<(), io::Error> { + if ! dir.is_dir() { + fs::create_dir(dir)?; + } + Ok(()) +} + +pub fn create_file_with_path(path: &PathBuf) -> Result { + if ! path.exists() { + if let Some(parent) = path.parent() { + trace!("Creating path: {}", parent.to_string_lossy()); + fs::create_dir_all(parent)?; + } + } + File::create(path) +} + +/// Derive the path for this file. +pub fn file_path(base_path: &PathBuf, file_name: &str) -> PathBuf { + let mut path = base_path.clone(); + path.push(file_name); + path +} + + +/// Saves a file, creating parent dirs as needed +pub fn save(content: &Bytes, full_path: &PathBuf) -> Result<(), io::Error> { + let mut f = create_file_with_path(full_path)?; + f.write_all(content)?; + + trace!("Saved file: {}", full_path.to_string_lossy()); + Ok(()) +} + +/// Saves an object to json - unwraps any json errors! +pub fn save_json(object: &O, full_path: &PathBuf) -> Result<(), io::Error> { + let json = serde_json::to_string(object).unwrap(); + save(&Bytes::from(json), full_path) +} + +/// Loads a files and deserialzes as json for the expected type. Maps json +/// errors to io::Error +pub fn load_json(full_path: &PathBuf) -> Result { + let bytes = read(full_path)?; + serde_json::from_slice(&bytes) + .map_err(|_| + io::Error::new(io::ErrorKind::Other, "could not deserialize json")) +} + +/// Saves a file, creating parent dirs as needed +pub fn save_in_dir( + content: &Bytes, + base_path: &PathBuf, + name: &str) -> Result<(), io::Error> { + let mut full_path = base_path.clone(); + full_path.push(name); + save(content, &full_path) +} + +/// Saves a file under a base directory, using the rsync uri to create +/// sub-directories preserving the rsync authority and module in dir names. +pub fn save_with_rsync_uri( + content: &Bytes, + base_path: &PathBuf, + uri: &uri::Rsync +) -> Result<(), io::Error> { + let path = path_with_rsync(base_path, uri); + save(content, &path) +} + +/// Reads a file to Bytes +pub fn read(path: &PathBuf) -> Result { + let mut f = File::open(path).map_err(|_| Error::cannot_read(path))?; + let mut bytes = Vec::new(); + f.read_to_end(&mut bytes)?; + Ok(Bytes::from(bytes)) +} + +pub fn read_with_rsync_uri( + base_path: &PathBuf, + uri: &uri::Rsync +) -> Result { + let path = path_with_rsync(base_path, uri); + read(&path) +} + +pub fn delete_with_rsync_uri( + base_path: &PathBuf, + uri: &uri::Rsync +) -> Result<(), io::Error> { + delete(&path_with_rsync(base_path, uri)) +} + +pub fn delete_in_dir( + base_path: &PathBuf, + name: &str +) -> Result<(), io::Error> { + let mut full_path = base_path.clone(); + full_path.push(name); + delete(&full_path) +} + +pub fn delete(full_path: &PathBuf) -> Result<(), io::Error> { + trace!("Removing file: {}", full_path.to_string_lossy()); + fs::remove_file(full_path)?; + Ok(()) +} + +pub fn clean_file_and_path(path: &PathBuf) -> Result<(), io::Error> { + if path.exists() { + fs::remove_file(&path)?; + + let mut parent_opt = path.parent(); + + while parent_opt.is_some() { + let parent = parent_opt.unwrap(); + if parent.read_dir()?.count() == 0 { + debug!("Will delete {}", parent.to_string_lossy().to_string()); + fs::remove_dir(parent)?; + } + + parent_opt = parent.parent(); + } + } + Ok(()) +} + +fn path_with_rsync(base_path: &PathBuf, uri: &uri::Rsync) -> PathBuf { + let mut path = base_path.clone(); + path.push(uri.module().authority()); + path.push(uri.module().module()); + path.push(uri.path()); + path +} + + +/// Recurses a path on disk and returns all files found as ['CurrentFile'], +/// using the provided rsync_base URI as the rsync prefix. +/// Allows a publication client to publish the contents below some base +/// dir, in their own designated rsync URI name space. +pub fn crawl_incl_rsync_base( + base_path: &PathBuf, + rsync_base: &uri::Rsync +) -> Result, Error> { + crawl_disk(base_path, base_path, Some(rsync_base)) +} + +/// Recurses a path on disk and returns all files found as ['CurrentFile'], +/// deriving the rsync_base URI from the directory structure. This is +/// useful when reading ['CurrentFile'] instances that were saved in some +/// base directory as is done by the ['FileStore']. +pub fn crawl_derive_rsync_uri( + base_path: &PathBuf +) -> Result, Error> { + crawl_disk(base_path, base_path, None) +} + +fn crawl_disk( + base_path: &PathBuf, + path: &PathBuf, + rsync_base: Option<&uri::Rsync> +) -> Result, Error> { + let mut res = Vec::new(); + + for entry in fs::read_dir(path).map_err(|_| Error::cannot_read(path))? { + let entry = entry.map_err(|_| Error::cannot_read(path))?; + let path = entry.path(); + if path.is_dir() { + let mut other = crawl_disk(base_path, &path, rsync_base)?; + res.append(&mut other); + } else { + let uri = derive_uri(base_path, &path, rsync_base)?; + let content = read(&path).map_err(|_| Error::cannot_read(&path))?; + let current_file = CurrentFile::new(uri, &content); + + res.push(current_file); + } + } + + Ok(res) +} + +fn derive_uri( + base_path: &PathBuf, + path: &PathBuf, + rsync_base: Option<&uri::Rsync> +) -> Result { + let rel = path + .strip_prefix(base_path) + .map_err(|_| Error::PathOutsideBasePath)?; + + let rel_string = rel.to_string_lossy().to_string(); + + let uri_string = match rsync_base { + Some(rsync_base) => + format!("{}{}", rsync_base.to_string(), rel_string), + None => + format!("rsync://{}", rel_string) + }; + + let uri = uri::Rsync::from_str(&uri_string) + .map_err(|_| Error::UnsupportedFileName(uri_string))?; + Ok(uri) +} + + + +//------------ CurrentFile --------------------------------------------------- + +#[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, + + /// 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: Base64, + + /// 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: EncodedHash +} + + +impl CurrentFile { + pub fn new(uri: uri::Rsync, content: &Bytes) -> Self { + let content = Base64::from_content(&content); + let hash = content.to_encoded_hash(); + CurrentFile {uri, content, hash} + } + + /// Saves this file under a base directory, based on the (rsync) uri of + /// this file. + pub fn save(&self, base_path: &PathBuf) -> Result<(), io::Error> { + save_with_rsync_uri(&self.content.to_bytes(), &base_path, &self.uri) + } + + pub fn uri(&self) -> &uri::Rsync { + &self.uri + } + + pub fn content(&self) -> &Base64 { + &self.content + } + + pub fn to_bytes(&self) -> Bytes { + self.content.to_bytes() + } + + pub fn hash(&self) -> &EncodedHash { + &self.hash + } + + pub fn as_publish(&self) -> publication::Publish { + let tag = Some(self.hash.to_string()); + let uri = self.uri.clone(); + let content = self.content.clone(); + publication::Publish::new(tag, uri, content) + } + + pub fn as_update(&self, old_hash: &EncodedHash) -> publication::Update { + let tag = None; + let uri = self.uri.clone(); + let content = self.content.clone(); + let hash = old_hash.clone(); + publication::Update::new(tag, uri, content, hash) + } + + pub fn as_withdraw(&self) -> publication::Withdraw { + let tag = None; + let uri = self.uri.clone(); + let hash = self.hash.clone(); + publication::Withdraw::new(tag, uri, hash) + } + + pub fn into_list_element(self) -> publication::ListElement { + publication::ListElement::new(self.uri, self.hash) + } + +} + +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, Display)] +pub enum Error { + + #[display(fmt="Cannot read: {}", _0)] + CannotRead(String), + + #[display(fmt="Unsupported characters: {}", _0)] + UnsupportedFileName(String), + + #[display(fmt = "Cannot use path outside of rsync jail")] + PathOutsideBasePath, +} + +impl Error { + pub fn cannot_read(path: &PathBuf) -> Error { + let str = path.to_string_lossy().to_string(); + Error::CannotRead(str) + } +} + +impl std::error::Error for Error {} + +impl From for io::Error { + fn from(e: Error) -> Self { + io::Error::new(io::ErrorKind::Other, e) + } +} + +//------------ Tests --------------------------------------------------------- + +#[cfg(test)] +mod tests { + use super::*; + use crate::util::test; + + #[test] + fn should_scan_disk() { + test::test_with_tmp_dir(|base_dir| { + + let file_1 = CurrentFile::new( + test::rsync_uri("rsync://host:10873/module/alice/file1.txt"), + &Bytes::from("content 1") + ); + let file_2 = CurrentFile::new( + test::rsync_uri("rsync://host:10873/module/alice/file2.txt"), + &Bytes::from("content 2") + ); + let file_3 = CurrentFile::new( + test::rsync_uri("rsync://host:10873/module/alice/sub/file1.txt"), + &Bytes::from("content sub file") + ); + let file_4 = CurrentFile::new( + test::rsync_uri("rsync://host:10873/module/bob/file.txt"), + &Bytes::from("content") + ); + + file_1.save(&base_dir).unwrap(); + file_2.save(&base_dir).unwrap(); + file_3.save(&base_dir).unwrap(); + file_4.save(&base_dir).unwrap(); + + let files = crawl_derive_rsync_uri(&base_dir).unwrap(); + + assert!(files.contains(&file_1)); + assert!(files.contains(&file_2)); + assert!(files.contains(&file_3)); + assert!(files.contains(&file_4)); + }); + } + +} diff --git a/commons/src/util/httpclient.rs b/commons/src/util/httpclient.rs new file mode 100644 index 00000000..4e176cd5 --- /dev/null +++ b/commons/src/util/httpclient.rs @@ -0,0 +1,249 @@ +//! Some helper functions for HTTP calls +use std::io::Read; +use std::time::Duration; +use bytes::Bytes; +use reqwest::{Client, Response, StatusCode}; +use reqwest::header::{ + HeaderMap, + HeaderValue, + InvalidHeaderValue, + USER_AGENT, + CONTENT_TYPE}; +use serde::Serialize; +use serde::de::DeserializeOwned; + +const JSON_CONTENT: &str = "application/json"; + + +/// Performs a GET request that expects a json response that can be +/// deserialized into the an owned value of the expected type. Returns an error +/// if nothing is returned. +pub fn get_json( + uri: &str, + token: Option<&str> +) -> Result { + let headers = headers(Some(JSON_CONTENT), token)?; + let res = client()?.get(uri).headers(headers).send()?; + process_json_response(res) +} + +/// Performs a get request and expects a response that can be turned +/// into a string (in particular, not a binary response). +pub fn get_text( + uri: &str, + content_type: &str, + token: Option<&str> +) -> Result { + let headers = headers(Some(content_type), token)?; + let res = client()?.get(uri).headers(headers).send()?; + match opt_text_response(res)? { + Some(res) => Ok(res), + None => Err(Error::EmptyResponse) + } +} + +/// Checks that there is a 200 OK response at the given URI. Discards the +/// response body. +pub fn get_ok(uri: &str, token: Option<&str>) -> Result<(), Error> { + let headers = headers(None, token)?; + let res = client()?.get(uri).headers(headers).send()?; + opt_text_response(res)?; // Will return nice errors with possible body. + Ok(()) +} + + +/// Performs a POST of data that can be serialized into json, and expects +/// a 200 OK response, without a body. +pub fn post_json( + uri: &str, + data: impl Serialize, + token: Option<&str> +) -> Result<(), Error> { + let headers = headers(Some(JSON_CONTENT), token)?; + let body = serde_json::to_string(&data)?; + let res = client()?.post(uri).headers(headers).body(body).send()?; + if let Some(res) = opt_text_response(res)? { + Err(Error::UnexpectedResponse(res)) + } else { + Ok(()) + } +} + + +/// Performs a POST of data that can be serialized into json, and expects +/// a json response that can be deserialized into the an owned value of the +/// expected type. +pub fn post_json_with_response( + uri: &str, + data: impl Serialize, + token: Option<&str> +) -> Result { + let headers = headers(Some(JSON_CONTENT), token)?; + let body = serde_json::to_string(&data)?; + let res = client()?.post(uri).headers(headers).body(body).send()?; + process_json_response(res) +} + + +/// Posts binary data, and expects a binary response. +/// +/// Note: Bytes may be empty if the post was successful, but the response was +/// empty. +pub fn post_binary( + uri: &str, + data: &Bytes, + content_type: &str +) -> Result { + let headers = headers(Some(content_type), None)?; + let body = data.to_vec(); + + let mut res = client()?.post(uri).headers(headers).body(body).send()?; + + match res.status() { + StatusCode::OK => { + let mut bytes: Vec = vec![]; + res.read_to_end(&mut bytes).unwrap(); + let bytes = bytes::Bytes::from(bytes); + Ok(bytes) + }, + status => { + match res.text() { + Ok(body) => { + if body.is_empty() { + Err(Error::BadStatus(status)) + } else { + Err(Error::ErrorWithBody(status, body)) + } + }, + _ => Err(Error::BadStatus(status)) + } + } + } +} + +/// Sends a delete request to the specified url. +pub fn delete( + uri: &str, + token: Option<&str> +) -> Result<(), Error> { + let headers = headers(None, token)?; + client()?.delete(uri).headers(headers).send()?; + Ok(()) +} + + +fn client() -> Result { + Client::builder() + .gzip(true) + .timeout(Duration::from_secs(300)) + .build() + .map_err(Error::RequestError) +} + +fn headers( + content_type: Option<&str>, + token: Option<&str> +) -> Result { + let mut headers = HeaderMap::new(); + headers.insert( + USER_AGENT, + HeaderValue::from_str("krill")? + ); + if let Some(content_type) = content_type { + headers.insert( + CONTENT_TYPE, + HeaderValue::from_str(content_type)? + ); + } + if let Some(token) = token { + headers.insert( + "Authorization", + HeaderValue::from_str(&format!("Bearer {}", token))? + ); + } + Ok(headers) +} + +fn process_json_response( + res: Response +) -> Result { + match opt_text_response(res) { + Err(e) => Err(e), + Ok(None) => Err(Error::EmptyResponse), + Ok(Some(s)) => { + let res: T = serde_json::from_str(&s)?; + Ok(res) + } + } +} + +fn opt_text_response(mut res: Response) -> Result, Error> { + match res.status() { + StatusCode::OK => { + match res.text().ok() { + None => Ok(None), + Some(s) => { + if s.is_empty() { + Ok(None) + } else { + Ok(Some(s)) + } + } + } + }, + StatusCode::FORBIDDEN => Err(Error::Forbidden), + status => { + match res.text() { + Ok(body) => { + if body.is_empty() { + Err(Error::BadStatus(status)) + } else { + Err(Error::ErrorWithBody(status, body)) + } + }, + _ => Err(Error::BadStatus(status)) + } + } + } +} + +//------------ Error --------------------------------------------------------- + +#[derive(Debug, Display)] +pub enum Error { + #[display(fmt="Request Error: {}", _0)] + RequestError(reqwest::Error), + + #[display(fmt="Access Forbidden")] + Forbidden, + + #[display(fmt="Received bad status: {}", _0)] + BadStatus(StatusCode), + + #[display(fmt="Status: {}, Error: {}", _0, _1)] + ErrorWithBody(StatusCode, String), + + #[display(fmt="{}", _0)] + JsonError(serde_json::Error), + + #[display(fmt="{}", _0)] + InvalidHeader(InvalidHeaderValue), + + #[display(fmt="Empty response received from server")] + EmptyResponse, + + #[display(fmt="Unexpected response: {}", _0)] + UnexpectedResponse(String) +} + +impl From for Error { + fn from(e: reqwest::Error) -> Self { Error::RequestError(e) } +} + +impl From for Error { + fn from(e: serde_json::Error) -> Self { Error::JsonError(e) } +} + +impl From for Error { + fn from(v: InvalidHeaderValue) -> Self { Error::InvalidHeader(v) } +} diff --git a/commons/src/util/mod.rs b/commons/src/util/mod.rs new file mode 100644 index 00000000..ea9735fb --- /dev/null +++ b/commons/src/util/mod.rs @@ -0,0 +1,62 @@ +//! General utility modules for use all over the code base +use std::time::Duration; +use bytes::Bytes; +use chrono::DateTime; +use chrono::Utc; +use chrono::offset::TimeZone; +use rpki::crypto::DigestAlgorithm; +use serde::Serialize; +use serde::Serializer; +use serde::Deserializer; +use serde::Deserialize; + +pub mod actix; +pub mod ext_serde; +pub mod file; +pub mod httpclient; +pub mod softsigner; +pub mod test; +pub mod xml; + +pub fn sha256(object: &[u8]) -> Bytes { + Bytes::from(DigestAlgorithm.digest(object).as_ref()) +} + +#[derive(Clone, Copy, Debug, Eq, Hash, Ord, PartialEq, PartialOrd)] +pub struct Time(DateTime); + +impl Time { + pub fn now() -> Self { + Time(Utc::now()) + } + + pub fn before_now(dur: Duration) -> Self { + let mut millis = Utc::now().timestamp_millis(); + millis -= dur.as_secs() as i64 * 1000; + + Time(Utc.timestamp_millis(millis)) + } + + pub fn on_or_before(&self, other: &Time) -> bool { + self.0.timestamp_millis() <= other.0.timestamp_millis() + } +} + +impl Serialize for Time { + fn serialize( + &self, serializer: S + ) -> Result where S: Serializer { + serializer.serialize_i64(self.0.timestamp_millis()) + } +} + +impl<'de> Deserialize<'de> for Time { + fn deserialize( + deserializer: D + ) -> Result where D: Deserializer<'de> { + + let timestamp: i64 = i64::deserialize(deserializer)?; + Ok(Time(Utc.timestamp_millis(timestamp))) + } +} + diff --git a/commons/src/util/softsigner.rs b/commons/src/util/softsigner.rs new file mode 100644 index 00000000..2477c3ea --- /dev/null +++ b/commons/src/util/softsigner.rs @@ -0,0 +1,333 @@ +//! Support for signing things using software keys (through openssl) and +//! storing them unencrypted on disk. +use std::{fs, io}; +use std::path::PathBuf; +use bytes::Bytes; +use openssl::rsa::Rsa; +use openssl::hash::MessageDigest; +use openssl::error::ErrorStack; +use openssl::pkey::{PKey, PKeyRef, Private}; +use rpki::crypto::{ + Signature, + SignatureAlgorithm, + Signer, + SigningError, + PublicKey, + PublicKeyFormat +}; +use rpki::crypto::signer::KeyError; +use serde::{de, ser}; +use serde::{Deserialize, Deserializer, Serialize, Serializer}; +use std::fs::File; +use std::io::Write; + + +//------------ SignerKeyId --------------------------------------------------- + +#[derive(Clone, Debug, Eq, PartialEq)] +pub struct SignerKeyId(String); + +impl SignerKeyId { + pub fn new(s: &str) -> Self { + SignerKeyId(s.to_string()) + } +} + +impl AsRef for SignerKeyId { + fn as_ref(&self) -> &str { + &self.0 + } +} + +impl Serialize for SignerKeyId { + fn serialize( + &self, + serializer: S + ) -> Result where S: Serializer { + self.as_ref().serialize(serializer) + } +} + +impl<'de> Deserialize<'de> for SignerKeyId { + fn deserialize( + deserializer: D + ) -> Result where D: Deserializer<'de> { + let s = String::deserialize(deserializer)?; + Ok(SignerKeyId::new(&s)) + } +} + + +//------------ OpenSslSigner ------------------------------------------------- + +/// An openssl based signer. +/// +/// Keeps the keys in memory (for now). +#[derive(Clone, Debug)] +pub struct OpenSslSigner { + keys_dir: PathBuf +} + +impl OpenSslSigner { + pub fn build(work_dir: &PathBuf) -> Result { + let meta_data = fs::metadata(&work_dir)?; + if meta_data.is_dir() { + + let mut keys_dir = PathBuf::from(work_dir); + keys_dir.push("keys"); + if ! keys_dir.is_dir() { + fs::create_dir_all(&keys_dir)?; + } + + Ok(OpenSslSigner { keys_dir } ) + } else { + Err(SignerError::InvalidWorkDir(work_dir.clone())) + } + } +} + +impl OpenSslSigner { + fn sign_with_key + ?Sized>( + pkey: &PKeyRef, + data: &D + ) -> Result + { + let mut signer = ::openssl::sign::Signer::new( + MessageDigest::sha256(), + pkey + )?; + signer.update(data.as_ref())?; + + let signature = Signature::new( + SignatureAlgorithm, + Bytes::from(signer.sign_to_vec()?)); + + Ok(signature) + } + + fn load_key(&self, id: &SignerKeyId) -> Result { + let path = self.key_path(id); + if path.exists() { + let f = File::open(path)?; + let kp: OpenSslKeyPair = serde_json::from_reader(f)?; + Ok(kp) + } else { + Err(SignerError::KeyNotFound) + } + + } + + fn key_path(&self, key_id: &SignerKeyId) -> PathBuf { + let mut path = self.keys_dir.clone(); + path.push(key_id.as_ref()); + path + } +} + +impl Signer for OpenSslSigner { + + type KeyId = SignerKeyId; + type Error = SignerError; + + fn create_key( + &mut self, + _algorithm: PublicKeyFormat + ) -> Result { + let kp = OpenSslKeyPair::build()?; + + let pk = &kp.subject_public_key_info()?; + let hex_hash = hex::encode(pk.key_identifier().as_ref()); + let key_id = SignerKeyId(hex_hash); + + let path = self.key_path(&key_id); + let json = serde_json::to_string(&kp)?; + + let mut f = File::create(path)?; + f.write_all(json.as_ref())?; + + Ok(key_id) + } + + fn get_key_info( + &self, + key_id: &Self::KeyId + ) -> Result> { + let key_pair = self.load_key(key_id)?; + Ok(key_pair.subject_public_key_info()?) + } + + fn destroy_key( + &mut self, + key_id: &Self::KeyId + ) -> Result<(), KeyError> { + let path = self.key_path(key_id); + if path.exists() { + fs::remove_file(path).map_err(SignerError::IoError)?; + } + Ok(()) + } + + fn sign + ?Sized>( + &self, + key_id: &Self::KeyId, + _algorithm: SignatureAlgorithm, + data: &D + ) -> Result> { + let key_pair = self.load_key(key_id)?; + Self::sign_with_key(key_pair.pkey.as_ref(), data) + .map_err(|e| { SigningError::Signer(e)}) + } + + fn sign_one_off + ?Sized>( + &self, + _algorithm: SignatureAlgorithm, + data: &D + ) -> Result<(Signature, PublicKey), SignerError> { + let kp = OpenSslKeyPair::build()?; + + let signature = Self::sign_with_key( + kp.pkey.as_ref(), + data + )?; + + let key = kp.subject_public_key_info()?; + + Ok((signature, key)) + } +} + + +//------------ OpenSslKeyPair ------------------------------------------------ + +/// An openssl based RSA key pair +pub struct OpenSslKeyPair { + pkey: PKey +} + +impl Serialize for OpenSslKeyPair { + fn serialize( + &self, + s: S + ) -> Result where + S: Serializer { + let bytes: Vec = self.pkey.as_ref().private_key_to_der() + .map_err(ser::Error::custom)?; + + base64::encode(&bytes).serialize(s) + } +} + +impl<'de> Deserialize<'de> for OpenSslKeyPair { + fn deserialize( + d: D + ) -> Result where + D: Deserializer<'de> { + match String::deserialize(d) { + Ok(base64) => { + let bytes = base64::decode(&base64) + .map_err(de::Error::custom)?; + + let pkey = PKey::private_key_from_der(&bytes) + .map_err(de::Error::custom)?; + + Ok( + OpenSslKeyPair { + pkey + } + ) + }, + Err(err) => Err(err) + } + } +} + +impl OpenSslKeyPair { + fn build() -> Result { + // Issues unwrapping this indicate a bug in the openssl library. + // So, there is no way to recover. + let rsa = Rsa::generate(2048)?; + let pkey = PKey::from_rsa(rsa)?; + Ok(OpenSslKeyPair{ pkey }) + } + + fn subject_public_key_info(&self) -> Result { + // Issues unwrapping this indicate a bug in the openssl library. + // So, there is no way to recover. + let mut b = Bytes::from(self.pkey.rsa().unwrap().public_key_to_der()?); + Ok(PublicKey::decode(&mut b).map_err(|_| SignerError::DecodeError)?) + } +} + + +//------------ OpenSslKeyError ----------------------------------------------- + +#[derive(Debug, Display)] +pub enum SignerError { + #[display(fmt = "OpenSsl Error: {}", _0)] + OpenSslError(ErrorStack), + + #[display(fmt = "Could not decode public key info: {}", _0)] + JsonError(serde_json::Error), + + #[display(fmt = "Invalid base path: {:?}", _0)] + InvalidWorkDir(PathBuf), + + #[display(fmt = "{}", _0)] + IoError(io::Error), + + #[display(fmt = "Could not find key")] + KeyNotFound, + + #[display(fmt = "Could not decode key")] + DecodeError, +} + +impl From for SignerError { + fn from(e: ErrorStack) -> Self { + SignerError::OpenSslError(e) + } +} + +impl From for SignerError { + fn from(e: serde_json::Error) -> Self { + SignerError::JsonError(e) + } +} + +impl From for SignerError { + fn from(e: io::Error) -> Self { + SignerError::IoError(e) + } +} + +//------------ Tests --------------------------------------------------------- + +#[cfg(test)] +pub mod tests { + + use super::*; + use crate::util::test; + + #[test] + fn should_return_subject_public_key_info() { + test::test_with_tmp_dir(|d| { + let mut s = OpenSslSigner::build(&d).unwrap(); + let ki = s.create_key(PublicKeyFormat).unwrap(); + s.get_key_info(&ki).unwrap(); + s.destroy_key(&ki).unwrap(); + }) + } + + #[test] + fn should_serialize_and_deserialize_key() { + + let key = OpenSslKeyPair::build().unwrap(); + let json = serde_json::to_string(&key).unwrap(); + let key_des: OpenSslKeyPair = serde_json::from_str(json.as_str()).unwrap(); + let json_from_des = serde_json::to_string(&key_des).unwrap(); + + // comparing json, because OpenSslKeyPair and its internal friends do + // not implement Eq and PartialEq. + assert_eq!(json, json_from_des); + } +} diff --git a/commons/src/util/test.rs b/commons/src/util/test.rs new file mode 100644 index 00000000..5dc370e8 --- /dev/null +++ b/commons/src/util/test.rs @@ -0,0 +1,60 @@ +use std::fs::File; +use std::io::Write; +use std::path::PathBuf; +use std::str::FromStr; +use bytes::Bytes; +use rpki::uri; + +/// This method sets up a test directory with a random name (a number) +/// under 'work', relative to where cargo is running. It then runs the +/// test provided in the closure, and finally it cleans up the test +/// directory. +/// +/// Note that if your test fails the directory is not cleaned up. +pub fn test_with_tmp_dir(op: F) where F: FnOnce(PathBuf) -> () { + use std::fs; + use std::path::PathBuf; + + let dir = create_sub_dir(&PathBuf::from("work")); + let path = PathBuf::from(&dir); + + op(dir); + + fs::remove_dir_all(path).unwrap(); +} + +/// This method sets up a random subdirectory and returns it. It is +/// assumed that the caller will clean this directory themselves. +pub fn create_sub_dir(base_dir: &PathBuf) -> PathBuf { + use std::fs; + use std::path::PathBuf; + use rand::{thread_rng, Rng}; + + let mut rng = thread_rng(); + let rnd: u32 = rng.gen(); + + let mut dir = base_dir.clone(); + dir.push(PathBuf::from(format!("{}", rnd))); + + let full_path = PathBuf::from(&dir); + fs::create_dir_all(&full_path).unwrap(); + + full_path +} + +pub fn rsync_uri(s: &str) -> uri::Rsync { + uri::Rsync::from_str(s).unwrap() +} + +pub fn http_uri(s: &str) -> uri::Http { + uri::Http::from_str(s).unwrap() +} + +pub fn as_bytes(s: &str) -> Bytes { Bytes::from(s) } + +pub fn save_file(base_dir: &PathBuf, file_name: &str, content: &[u8]) { + let mut full_name = base_dir.clone(); + full_name.push(PathBuf::from(file_name)); + let mut f = File::create(full_name).unwrap(); + f.write_all(content).unwrap(); +} \ No newline at end of file diff --git a/commons/src/util/xml.rs b/commons/src/util/xml.rs new file mode 100644 index 00000000..c4a96e69 --- /dev/null +++ b/commons/src/util/xml.rs @@ -0,0 +1,548 @@ +//! Support for RPKI XML structures. +use std::{fs, io}; +use std::fs::File; +use std::path::Path; +use base64; +use base64::DecodeError; +use bytes::Bytes; +use hex; +use hex::FromHexError; +use xmlrs::{reader, writer}; +use xmlrs::{EmitterConfig, EventReader, EventWriter, ParserConfig}; +use xmlrs::attribute::OwnedAttribute; +use xmlrs::reader::XmlEvent; + + +//------------ XmlReader ----------------------------------------------------- + +/// A convenience wrapper for RPKI XML parsing +/// +/// This type only exposes things we need for the RPKI XML structures. +pub struct XmlReader { + /// The underlying xml-rs reader + reader: EventReader, + + /// Placeholder for an event so that 'peak' can be supported, as + /// well as temporarily caching a close event in case a list of + /// inner elements is processed. + cached_event: Option, + + /// Name of the next start element, if any + next_start_name: Option +} + + +/// Reader methods +impl XmlReader { + + /// Gets the next XmlEvent + /// + /// Will take cached event if there is one + fn next(&mut self) -> Result { + match self.cached_event.take() { + Some(e) => Ok(e), + None => Ok(self.reader.next()?) + } + } + + /// Puts an XmlEvent back so that it can be retrieved by 'next' + fn cache(&mut self, e: XmlEvent) { + self.cached_event = Some(e); + } +} + + +/// Basic operations to parse the XML. +/// +/// These methods are private because they are used by the higher level +/// closure based methods, defined below, that one should use to parse +/// XML safely. +impl XmlReader { + /// Takes the next element and expects a start of document. + fn start_document(&mut self) -> Result<(), XmlReaderErr> { + match self.next() { + Ok(reader::XmlEvent::StartDocument {..}) => Ok(()), + _ => Err(XmlReaderErr::ExpectedStartDocument) + } + } + + /// Takes the next element and expects a start element with the given name. + fn expect_element(&mut self) -> Result<(Tag, Attributes), XmlReaderErr> { + match self.next() { + Ok(reader::XmlEvent::StartElement { name, attributes, ..}) => { + Ok((Tag{name: name.local_name}, Attributes{attributes})) + }, + _ => Err(XmlReaderErr::ExpectedStart) + } + } + + /// Takes the next element and expects a close element with the given name. + fn expect_close(&mut self, tag: Tag) -> Result<(), XmlReaderErr> { + match self.next() { + Ok(reader::XmlEvent::EndElement { name, ..}) => { + if name.local_name == tag.name { + Ok(()) + } else { + Err(XmlReaderErr::ExpectedClose(tag.name)) + } + } + _ => Err(XmlReaderErr::ExpectedClose(tag.name)) + } + } + + /// Takes the next element and expects the end of document. + /// + /// Returns Ok(true) if the element is the end of document, or + /// an error otherwise. + fn end_document(&mut self) -> Result<(), XmlReaderErr> { + match self.next() { + Ok(reader::XmlEvent::EndDocument) => Ok(()), + _ => Err(XmlReaderErr::ExpectedEnd) + } + } +} + +/// Closure based parsing of XML. +/// +/// This approach ensures that the consumer can only get opening tags, or +/// content (such as Characters), and process the enclosed content. In +/// particular it ensures that the consumer cannot accidentally get close +/// tags - so it forces that execution returns. +impl XmlReader { + /// Decodes an XML structure + /// + /// This method checks that the document starts, then passes a reader + /// instance to the provided closure, and will return the result from + /// that after checking that the XML document is fully processed. + pub fn decode(source: R, op: F) -> Result + where F: FnOnce(&mut Self) -> Result, + E: From { + let mut config = ParserConfig::new(); + config.trim_whitespace = true; + config.ignore_comments = true; + + let mut xml = XmlReader{ + reader: config.create_reader(source), + cached_event: None, + next_start_name: None + }; + + xml.start_document()?; + let res = op(&mut xml)?; + xml.end_document()?; + + Ok(res) + } + + /// Takes an element and process it in a closure + /// + /// This method checks that the next element is indeed a Start Element, + /// and passes the Tag and Attributes and this reader to a closure. After + /// the closure completes it will verify that the next element is the + /// Close Element for this Tag, and returns the result from the closure. + pub fn take_element(&mut self, op: F) -> Result + where F: FnOnce(&Tag, Attributes, &mut Self) -> Result, + E: From { + let (tag, attr) = self.expect_element()?; + let res = op(&tag, attr, self)?; + self.expect_close(tag)?; + Ok(res) + } + + /// Takes a named element and process it in a closure + /// + /// Checks that the element has the expected name and passed the closure + /// to the generic take_element method. + pub fn take_named_element( + &mut self, + name: &str, + op: F + ) -> Result + where + F: FnOnce(Attributes, &mut Self) -> Result, + E: From + { + self.take_element(|t, a, r| { + if t.name != name { + Err(XmlReaderErr::ExpectedNamedStart(name.to_string()).into()) + } + else { + op(a, r) + } + }) + } + + /// Takes the next element that is part of a list of elements under the + /// current element, and processes it using a closure. When the end of the + /// list is encountered, i.e. the next element is not a start element, then + /// the closure is not executed and Ok(None) is returned. The element is + /// put back on the cache for processing by the parent structure. + /// + /// Note: This will break if we encounter a parent XML element that has + /// both a list of children XML elements *and* some (character) content. + /// However, this is not used by the RPKI XML structures. Also, provided + /// that a 'take_*' method with a closure was used for the parent element, + /// then we will get a clear error there (expect end element). + pub fn take_opt_element(&mut self, op: F) -> Result, E> + where F: FnOnce(&Tag, Attributes, &mut Self) -> Result, E>, + E: From { + + let n = self.next()?; + match n { + XmlEvent::StartElement { name, attributes, ..} => { + let tag = Tag{name: name.local_name}; + let res = op( + &tag, + Attributes{attributes}, + self + ); + self.expect_close(tag)?; + res + }, + _ => { + self.cache(n); + Ok(None) + } + } + } + + /// Takes characters + pub fn take_chars(&mut self) -> Result { + match self.next() { + Ok(reader::XmlEvent::Characters(chars)) => { + Ok(chars) + } + _ => Err(XmlReaderErr::ExpectedCharacters) + } + } + + /// Takes base64 encoded bytes from the next 'characters' event. + pub fn take_bytes_characters(&mut self) -> Result { + let b64 = base64::decode_config(&self.take_chars()?, base64::MIME)?; + Ok(Bytes::from(b64)) + } + + pub fn take_empty(&mut self) -> Result<(), XmlReaderErr> { + Ok(()) + } + + /// Returns the name of the next start element or None if the next + /// element is not a start element. Also ensures that the next element + /// is kept in the cache for normal subsequent processing. + pub fn next_start_name(&mut self) -> Option<&str> { + match self.next() { + Err(_) => None, + Ok(e) => { + if let XmlEvent::StartElement { ref name, ..} = e { + // XXX not the most efficient.. but need a different + // underlying XML parser to get around ownership + // issues. + self.next_start_name = Some(name.local_name.clone()) + } else { + self.next_start_name = None; + } + self.cache(e); + self.next_start_name.as_ref().map(|s| s.as_ref()) + } + } + } +} + +impl XmlReader { + + /// Opens a file and decodes it as an XML file. + pub fn open(path: P, op: F) -> Result + where F: FnOnce(&mut Self) -> Result, + P: AsRef, + E: From + From { + Self::decode(fs::File::open(path)?, op) + } +} + +//------------ XmlReaderErr -------------------------------------------------- + +#[derive(Debug, Display)] +pub enum XmlReaderErr { + #[display(fmt = "Expected Start of Document")] + ExpectedStartDocument, + + #[display(fmt = "Expected Start Element")] + ExpectedStart, + + #[display(fmt = "Expected Start Element with name: {}", _0)] + ExpectedNamedStart(String), + + #[display(fmt = "Expected Characters Element")] + ExpectedCharacters, + + #[display(fmt = "Expected Close Element with name: {}", _0)] + ExpectedClose(String), + + #[display(fmt = "Expected End of Document")] + ExpectedEnd, + + #[display(fmt = "Error reading file: {}", _0)] + IoError(io::Error), + + #[display(fmt = "Attributes Error: {}", _0)] + AttributesError(AttributesError), + + #[display(fmt = "XML Reader Error: {}", _0)] + ReaderError(reader::Error), + + #[display(fmt = "Base64 decoding issue: {}", _0)] + Base64Error(DecodeError) +} + +impl From for XmlReaderErr { + fn from(e: io::Error) -> XmlReaderErr{ + XmlReaderErr::IoError(e) + } +} + +impl From for XmlReaderErr { + fn from(e: AttributesError) -> XmlReaderErr { + XmlReaderErr::AttributesError(e) + } +} + +impl From for XmlReaderErr { + fn from(e: reader::Error) -> XmlReaderErr { + XmlReaderErr::ReaderError(e) + } +} + +impl From for XmlReaderErr { + fn from(e: DecodeError) -> XmlReaderErr { + XmlReaderErr::Base64Error(e) + } +} + + +//------------ Attributes ---------------------------------------------------- + +/// A convenient wrapper for XML tag attributes +pub struct Attributes { + /// The underlying xml-rs structure + attributes: Vec +} + +impl Attributes { + + /// Takes an optional attribute by name + pub fn take_opt(&mut self, name: &str) -> Option { + let i = self.attributes.iter().position(|a| a.name.local_name == name); + match i { + Some(i) => { + let a = self.attributes.swap_remove(i); + Some(a.value) + } + None => None + } + } + + /// Takes an optional hexencoded attribute and converts it to Bytes + pub fn take_opt_hex(&mut self, name: &str) -> Option { + self.take_req_hex(name).ok() + } + + /// Takes a required attribute by name + pub fn take_req(&mut self, name: &str) -> Result { + self.take_opt(name) + .ok_or_else(|| AttributesError::MissingAttribute(name.to_string())) + } + + /// Takes a required hexencoded attribute and converts it to Bytes + pub fn take_req_hex(&mut self, name: &str) + -> Result { + + match hex::decode(self.take_req(name)?) { + Err(e) => Err(AttributesError::HexError(e)), + Ok(b) => Ok(Bytes::from(b)) + } + } + + /// Verifies that there are no more attributes + pub fn exhausted(&self) -> Result<(), AttributesError> { + if self.attributes.is_empty() { + Ok(()) + } else { + Err(AttributesError::ExtraAttributes) + } + } +} + + +//------------ AttributesError ----------------------------------------------- + +#[derive(Debug, Display)] +pub enum AttributesError { + #[display(fmt = "Required attribute missing: {}", _0)] + MissingAttribute(String), + + #[display(fmt = "Extra attributes found")] + ExtraAttributes, + + #[display(fmt = "Wrong hex encoding: {}", _0)] + HexError(FromHexError) +} + + +//------------ Tag ----------------------------------------------------------- + +pub struct Tag { + pub name: String +} + + +//------------ XmlWriter ----------------------------------------------------- + +/// A convenience wrapper for RPKI XML generation +/// +/// This type only exposes things we need for the RPKI XML structures. +pub struct XmlWriter { + /// The underlying xml-rs writer + writer: EventWriter +} + + +/// Generate the XML. +impl XmlWriter { + + + fn unwrap_emitter_error(r: Result) -> Result { + match r { + Ok(t) => Ok(t), + Err(e) => { + match e { + writer::Error::Io(io) => Err(io), + _ => { + // The other errors can only happen for stuff like + // not closing tags, starting a doc twice etc. But + // the XmlWriter lib already ensures that these things + // do not happen. They are not dependent on input. + panic!("XmlWriter library error: {:?}", e) + } + } + } + } + } + + /// Adds an element + pub fn put_element( + &mut self, + name: &str, + attr: Option<&[(&str, &str)]>, + op: F) -> Result<(), io::Error> + where F: FnOnce(&mut Self) -> Result<(), io::Error> { + let mut start = writer::XmlEvent::start_element(name); + + if let Some(v) = attr { + for a in v { + start = start.attr(a.0, a.1); + } + } + + Self::unwrap_emitter_error(self.writer.write(start))?; + op(self)?; + Self::unwrap_emitter_error( + self.writer.write(writer::XmlEvent::end_element()) + )?; + + Ok(()) + } + + /// Puts some String in a characters element + pub fn put_text(&mut self, text: &str) -> Result<(), io::Error> { + Self::unwrap_emitter_error( + self.writer.write(writer::XmlEvent::Characters(text)) + )?; + Ok(()) + } + + /// Converts bytes to base64 encoded Characters as the content. Note + /// that you cannot have both Characters and other included elements. + /// This would be valid XML, but it's not used by any of the RPKI XML + /// structures. + pub fn put_blob(&mut self, bytes: &Bytes) -> Result<(), io::Error> { + let b64 = base64::encode(bytes); + self.put_text(b64.as_ref()) + } + + /// Use this for convenience where empty content is required + pub fn empty(&mut self) -> Result<(), io::Error> { + Ok(()) + } + + /// Sets up the writer config and returns a closure that is expected + /// to add the actual content of the XML. + /// + /// This method is private because one should use the pub encode_vec + /// method, and in future others like it, to set up the writer for a + /// specific type (Vec, File, etc.). + fn encode(w: W, op: F) -> Result<(), io::Error> + where F: FnOnce(&mut Self) -> Result<(), io::Error> { + + let writer = EmitterConfig::new() + .write_document_declaration(false) + .normalize_empty_elements(true) + .perform_indent(true) + .create_writer(w); + + let mut x = XmlWriter { writer }; + + op(&mut x) + } +} + +impl XmlWriter<()> { + + /// Call this to encode XML into a Vec + pub fn encode_vec(op: F) -> Vec + where F: FnOnce(&mut XmlWriter<&mut Vec>) + -> Result<(), io::Error> + { + let mut b = Vec::new(); + XmlWriter::encode(&mut b, op).unwrap(); // IO error impossible for vec + b + } + + pub fn encode_to_file(file: &mut File, op: F) -> Result<(), io::Error> + where F: FnOnce(&mut XmlWriter<&mut File>) -> Result<(), io::Error> { + XmlWriter::encode(file, op) + } +} + + +//------------ Tests --------------------------------------------------------- + +#[cfg(test)] +mod tests { + + use super::*; + use std::str; + + #[test] + fn should_write_xml() { + + let xml = XmlWriter::encode_vec(|w| { + w.put_element( + "a", + Some(&[ + ("xmlns", "http://ns/"), + ("c", "d") + ]), + |w| { + w.put_element("b", None, |w| { + w.put_blob(&Bytes::from("X")) + }) + } + ) + }); + + assert_eq!( + str::from_utf8(&xml).unwrap(), + "\n WA==\n" + ); + } +} diff --git a/daemon/Cargo.toml b/daemon/Cargo.toml index c17d9bef..db943929 100644 --- a/daemon/Cargo.toml +++ b/daemon/Cargo.toml @@ -16,7 +16,6 @@ derive_more = "^0.13" fern = "^0.5" futures = "0.1" hex = "^0.3" -krill_commons = "^0.2" lazy_static = "^1.1" log = "^0.4" openssl = { version = "^0.10", features = ["v110"] } @@ -41,6 +40,10 @@ version = "0.2.0" path = "../cms_proxy" version = "0.2.0" +[dependencies.krill_commons] +path = "../commons" +version = "0.2.0" + [dependencies.krill_pubc] path = "../pubc" version = "0.2.0" diff --git a/pubc/Cargo.toml b/pubc/Cargo.toml index d8332624..05eed80d 100644 --- a/pubc/Cargo.toml +++ b/pubc/Cargo.toml @@ -6,7 +6,6 @@ authors = ["Tim Bruijnzeels ", "Martin Hoffmann ", "Martin Hoffmann