Moving krill_commons into workspace.

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" }

client/Cargo.toml

@@ -6,7 +6,6 @@ authors = ["Tim Bruijnzeels <tim@nlnetlabs.nl>", "Martin Hoffmann <martin@nlnetl
 [dependencies]
 clap            = "^2.32"
 derive_more     = "^0.13"
-krill_commons   = "^0.2"
 rpki            = "^0.3"
 serde           = { version = "^1.0", features = ["rc"] }
 serde_derive    = "^1.0"
@@ -16,3 +15,6 @@ serde_json      = "^1.0"
 path = "../cms_proxy"
 version = "0.2.0"
 
+[dependencies.krill_commons]
+path = "../commons"
+version = "0.2.0"

cms_proxy/Cargo.toml

@@ -10,7 +10,6 @@ bytes         = "^0.4"
 chrono        = { version = "^0.4", features = ["serde"] }
 derive_more   = "^0.13"
 hex           = "^0.3"
-krill_commons = "^0.2"
 openssl       = { version = "^0.10", features = ["v110"] }
 rpki          = "^0.3"
 serde         = { version = "^1.0", features = ["rc"] }
@@ -31,3 +30,6 @@ futures       = "0.1"
 ring = "^0.13"
 untrusted = "^0.6"
 
+[dependencies.krill_commons]
+path = "../commons"
+version = "0.2.0"

commons/Cargo.toml

@@ -0,0 +1,24 @@
+[package]
+name = "krill_commons"
+version = "0.2.0"
+authors = ["Tim Bruijnzeels <tim@nlnetlabs.nl>", "Martin Hoffmann <martin@nlnetlabs.nl>"]
+
+[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"

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.

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<String> for PublisherHandle {
+    fn from(s: String) -> Self { PublisherHandle::from(AggregateId::from(s))}
+}
+
+impl From<AggregateId> for PublisherHandle {
+    fn from(id: AggregateId) -> Self {
+        PublisherHandle(id)
+    }
+}
+
+impl From<&AggregateId> for PublisherHandle {
+    fn from(id: &AggregateId) -> Self {
+        PublisherHandle(id.clone())
+    }
+}
+
+impl AsRef<str> for PublisherHandle {
+    fn as_ref(&self) -> &str {
+        self.name()
+    }
+}
+
+impl AsRef<AggregateId> 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<Link>
+}
+
+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<PublisherSummary>
+}
+
+impl PublisherList {
+    pub fn build(
+        publishers: &[PublisherHandle],
+        path_publishers: &str
+    ) -> PublisherList {
+        let publishers: Vec<PublisherSummary> = publishers.iter().map(|p|
+            PublisherSummary::from(&p, path_publishers)
+        ).collect();
+
+        PublisherList {
+            publishers
+        }
+    }
+
+    pub fn publishers(&self) -> &Vec<PublisherSummary> {
+        &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 {}
+
+

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<str> for Base64 {
+    fn as_ref(&self) -> &str {
+        use std::str;
+        str::from_utf8(&self.0).unwrap()
+    }
+}
+
+impl From<String> 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<S>(
+        &self, serializer: S
+    ) -> Result<S::Ok, S::Error> where S: Serializer {
+        self.to_string().serialize(serializer)
+    }
+}
+
+impl<'de> Deserialize<'de> for Base64 {
+    fn deserialize<D>(
+        deserializer: D
+    ) -> Result<Base64, D::Error> 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<str> for EncodedHash {
+    fn as_ref(&self) -> &str {
+        use std::str;
+        str::from_utf8(&self.0).unwrap()
+    }
+}
+
+impl From<String> 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<S>(
+        &self, serializer: S
+    ) -> Result<S::Ok, S::Error> where S: Serializer {
+        self.to_string().serialize(serializer)
+    }
+}
+
+impl<'de> Deserialize<'de> for EncodedHash {
+    fn deserialize<D>(
+        deserializer: D
+    ) -> Result<EncodedHash, D::Error> 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<ErrorCode> 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<usize> 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<ErrorResponse> 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)
+        }
+
+
+    }
+}
+

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<Publish>,
+    updates: Vec<Update>,
+    withdraws: Vec<Withdraw>
+}
+
+impl PublishDelta {
+    pub fn new(
+        publishes: Vec<Publish>,
+        updates: Vec<Update>,
+        withdraws: Vec<Withdraw>
+    ) -> Self {
+        PublishDelta { publishes, updates, withdraws }
+    }
+
+    pub fn publishes(&self) -> &Vec<Publish> {
+        &self.publishes
+    }
+    pub fn updates(&self) -> &Vec<Update> {
+        &self.updates
+    }
+    pub fn withdraws(&self) -> &Vec<Withdraw> {
+        &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<Publish>, Vec<Update>, Vec<Withdraw>) {
+        (self.publishes, self.updates, self.withdraws)
+    }
+}
+
+
+//------------ PublishDeltaBuilder -------------------------------------------
+
+#[derive(Default)]
+pub struct PublishDeltaBuilder {
+    publishes: Vec<Publish>,
+    updates: Vec<Update>,
+    withdraws: Vec<Withdraw>
+}
+
+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<String>,
+
+    #[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<String>, 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<String> { &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<String>, 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<String>,
+
+    #[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<String>,
+        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<String> { &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<String>, 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<String>,
+
+    #[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<String>, 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<String> { &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<String>, 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<ListElement>
+}
+
+impl ListReply {
+    pub fn new(elements: Vec<ListElement>) -> Self {
+        ListReply { elements }
+    }
+
+    pub fn from_files(files: Vec<CurrentFile>) -> Self {
+        let elements = files.into_iter().map(|f| f.into_list_element()).collect();
+        ListReply { elements }
+    }
+
+    pub fn elements(&self) -> &Vec<ListElement> {
+        &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 }
+}

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<publication::Publish> 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<publication::Update> for UpdateElement {
+    fn from(u: publication::Update) -> Self {
+        let (_tag, uri, base64, hash) = u.unwrap();
+        UpdateElement { uri, base64, hash }
+    }
+}
+
+impl Into<PublishElement> 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<publication::Withdraw> 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<DeltaRef>,
+    old_refs:   Vec<(Time, FileRef)>
+}
+
+#[derive(Clone, Debug, Deserialize, Serialize)]
+pub struct NotificationUpdate {
+    time:       Time,
+    session:    Option<String>,
+    snapshot:   SnapshotRef,
+    delta:      DeltaRef,
+    last_delta: u64
+}
+
+impl NotificationUpdate {
+    pub fn new(
+        time: Time,
+        session: Option<String>,
+        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<String>, 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<FileRef> 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<EncodedHash, PublishElement>);
+
+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<EncodedHash, io::Error> {
+        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<PublishElement>,
+    updates: Vec<UpdateElement>,
+    withdraws: Vec<WithdrawElement>
+}
+
+impl From<publication::PublishDelta> 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<PublishElement>, Vec<UpdateElement>, Vec<WithdrawElement>) {
+        (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<PublishElement> {
+        &self.publishes
+    }
+
+    pub fn updates(&self) -> &Vec<UpdateElement> {
+        &self.updates
+    }
+
+    pub fn withdraws(&self) -> &Vec<WithdrawElement> {
+        &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<EncodedHash, io::Error> {
+
+        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)
+    }
+
+}

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<InitPersonDetails>;
+
+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<PersonEventDetails>;
+
+#[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<PersonCommandDetails>;
+
+#[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<u64>) -> Self {
+        Self::new(id, version, PersonCommandDetails::GoAroundTheSun)
+    }
+
+
+    pub fn change_name(id: &AggregateId, version: Option<u64>, 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<Vec<PersonEvent>, 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<Self, PersonError> {
+        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<S: KeyStore> {
+    /// Here we use a cache to make matters complicated^H interesting..
+    /// Of course this may not always be the best idea..
+    cache: RwLock<HashMap<AggregateId, Arc<Person>>>,
+
+    /// The keystore where snapshots and events may be retrieved and stored.
+    store: S
+}
+
+impl<S: KeyStore> PersonManager<S> {
+
+    // 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<Option<Arc<Person>>, 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<PersonError> for PersonManagerError {
+    fn from(e: PersonError) -> Self { PersonManagerError::PersonError(e) }
+}
+
+impl From<KeyStoreError> 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);
+        })
+    }
+}

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<T: Clone + Serialize + DeserializeOwned + Sized> 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<String> for AggregateId {
+    fn from(s: String) -> Self {
+        AggregateId(s)
+    }
+}
+
+impl AsRef<str> for AggregateId {
+    fn as_ref(&self) -> &str {
+        self.as_str()
+    }
+}
+
+impl AsRef<String> for AggregateId {
+    fn as_ref(&self) -> &String {
+        &self.0
+    }
+}
+
+impl AsRef<Path> 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<Event = Self::Event>;
+    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<Self, Self::Error>;
+
+    /// 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<Self::Event>) {
+        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<Vec<Self::Event>, 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<E: Storable + 'static> {
+    id: AggregateId,
+    version: u64,
+    #[serde(deserialize_with = "E::deserialize")]
+    details: E
+}
+
+impl<E: Storable + 'static> StoredEvent<E> {
+
+    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<E: Storable + 'static> Event for StoredEvent<E> {
+    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<u64>;
+
+    /// 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<C: CommandDetails> {
+    id: AggregateId,
+    version: Option<u64>,
+    #[serde(deserialize_with = "C::deserialize")]
+    details: C
+}
+
+impl<C: CommandDetails> Command for SentCommand<C> {
+    type Event = C::Event;
+
+    fn id(&self) -> &AggregateId {
+        &self.id
+    }
+
+    fn version(&self) -> Option<u64> {
+        self.version
+    }
+}
+
+impl<C: CommandDetails> SentCommand<C> {
+
+    pub fn new(id: &AggregateId, version: Option<u64>, 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<AggregateId>; // Use Iterator?
+
+    /// Throws an error if the key already exists.
+
+    fn store<V: Any + Serialize>(
+        &self,
+        id: &AggregateId,
+        key: &Self::Key,
+        value: &V
+    ) -> Result<(), KeyStoreError>;
+
+    /// Get the value for this key, if any exists.
+
+    fn get<V: Any + Storable>(
+        &self,
+        id: &AggregateId,
+        key: &Self::Key
+    ) -> Result<Option<V>, KeyStoreError>;
+
+    /// Get the value for this key, if any exists.
+
+    fn get_event<V: Event>(
+        &self,
+        id: &AggregateId,
+        version: u64
+    ) -> Result<Option<V>, KeyStoreError>;
+
+    fn store_event<V: Event>(
+        &self,
+        event: &V
+    ) -> Result<(), KeyStoreError>;
+
+    /// Get the latest aggregate
+
+    fn get_aggregate<V: Aggregate>(
+        &self,
+        id: &AggregateId
+    ) -> Result<Option<V>, KeyStoreError>;
+
+    /// Saves the latest snapshot - overwrites any previous snapshot.
+
+    fn store_aggregate<V: 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<io::Error> for KeyStoreError {
+    fn from(e: io::Error) -> Self { KeyStoreError::IoError(e) }
+}
+
+impl From<serde_json::Error> 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<AggregateId> {
+        let mut res: Vec<AggregateId> = 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<V: Any + Serialize>(
+        &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<V: Any + Storable>(
+        &self,
+        id: &AggregateId,
+        key: &Self::Key
+    ) -> Result<Option<V>, 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<V: Event>(
+        &self,
+        id: &AggregateId,
+        version: u64
+    ) -> Result<Option<V>, 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<V: 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<V: Aggregate>(
+        &self,
+        id: &AggregateId
+    ) -> Result<Option<V>, 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::<V>(id, &key)? {
+            Some(aggregate) => Some(aggregate),
+            None => {
+                match self.get_event::<V::InitEvent>(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<V: 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<Self, io::Error> {
+        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: &<Self as KeyStore>::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<A: 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<T> = Result<T, AggregateStoreError>;
+
+pub trait AggregateStore<A: Aggregate>: 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<Arc<A>>;
+
+    /// 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<A> 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<A>, events: Vec<A::Event>) -> StoreResult<Arc<A>>;
+
+    /// Returns true if an instance exists for the id
+    fn has(&self, id: &AggregateId) -> bool;
+
+    /// Lists all known ids.
+    fn list(&self) -> Vec<AggregateId>;
+}
+
+
+/// 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<KeyStoreError> for AggregateStoreError {
+    fn from(e: KeyStoreError) -> Self { AggregateStoreError::KeyStoreError(e) }
+}
+
+
+pub struct DiskAggregateStore<A: Aggregate> {
+    store: DiskKeyStore,
+    cache: RwLock<HashMap<AggregateId, Arc<A>>>,
+    use_cache: bool
+}
+
+impl<A: Aggregate> DiskAggregateStore<A> {
+    pub fn new(work_dir: &PathBuf, name_space: &str) -> Result<Self, io::Error> {
+        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<A: Aggregate> DiskAggregateStore<A> {
+    fn has_updates(
+        &self,
+        id: &AggregateId,
+        aggregate: &A
+    ) -> StoreResult<bool> {
+        Ok(self.store.get_event::<A::Event>(id, aggregate.version())?.is_some())
+    }
+
+    fn cache_get(&self, id: &AggregateId) -> Option<Arc<A>> {
+        if self.use_cache {
+            self.cache.read().unwrap().get(id).cloned()
+        } else {
+            None
+        }
+    }
+
+    fn cache_update(&self, id: &AggregateId, arc: Arc<A>) {
+        if self.use_cache {
+            self.cache.write().unwrap().insert(id.clone(), arc);
+        }
+    }
+}
+
+impl<A: Aggregate> AggregateStore<A> for DiskAggregateStore<A> {
+    fn get_latest(&self, id: &AggregateId) -> StoreResult<Arc<A>> {
+        match self.cache_get(id) {
+            None => {
+                match self.store.get_aggregate(id)? {
+                    None => Err(AggregateStoreError::UnknownAggregate(id.clone())),
+                    Some(agg) => {
+                        let arc: Arc<A> = 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<A>, events: Vec<A::Event>) -> StoreResult<Arc<A>> {
+        // 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<AggregateId> {
+        self.store.aggregates()
+    }
+}
+
+
+

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;

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<S: 'static> FromRequest<S> for PublisherRequest {
+    type Config = ();
+    type Result = Box<Future<Item=Self, Error=actix_web::Error>>;
+
+    fn from_request(
+        req: &actix_web::HttpRequest<S>,
+        _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<S> FromRequest<S> for PublisherHandle {
+    type Config = ();
+    type Result = Result<Self, actix_web::Error>;
+
+    fn from_request(
+        req: &actix_web::HttpRequest<S>,
+        _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<S: 'static> FromRequest<S> for PublishDelta {
+    type Config = ();
+    type Result = Box<Future<Item=Self, Error=actix_web::Error>>;
+
+    fn from_request(
+        req: &actix_web::HttpRequest<S>,
+        _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))
+    }
+}

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<Bytes, D::Error>
+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<S>(b: &Bytes, s: S) -> Result<S::Ok, S::Error>
+where S: Serializer
+{
+    base64::encode(b).serialize(s)
+}
+
+
+//------------ uri::Rsync ----------------------------------------------------
+
+pub fn de_rsync_uri<'de, D>(d: D) -> Result<uri::Rsync, D::Error>
+where D: Deserializer<'de>
+{
+    let some = String::deserialize(d)?;
+    uri::Rsync::from_string(some).map_err(de::Error::custom)
+}
+
+pub fn ser_rsync_uri<S>(uri: &uri::Rsync, s: S) -> Result<S::Ok, S::Error>
+where S: Serializer
+{
+    uri.to_string().serialize(s)
+}
+
+
+//------------ uri::Http -----------------------------------------------------
+
+pub fn de_http_uri<'de, D>(d: D) -> Result<uri::Http, D::Error>
+where D: Deserializer<'de>
+{
+    let some = String::deserialize(d)?;
+    uri::Http::from_string(some).map_err(de::Error::custom)
+}
+
+pub fn ser_http_uri<S>(uri: &uri::Http, s: S) -> Result<S::Ok, S::Error>
+where S: Serializer
+{
+    uri.to_string().serialize(s)
+}
+
+
+//------------ LevelFilter ---------------------------------------------------
+
+pub fn de_level_filter<'de, D>(d: D) -> Result<LevelFilter, D::Error>
+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<Facility, D::Error>
+    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))})
+}

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<PathBuf, io::Error> {
+    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<File, io::Error> {
+    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<O: Serialize>(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<O: DeserializeOwned>(full_path: &PathBuf) -> Result<O, io::Error> {
+    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<Bytes, io::Error> {
+    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<Bytes, io::Error> {
+    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<Vec<CurrentFile>, 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<Vec<CurrentFile>, Error> {
+    crawl_disk(base_path, base_path, None)
+}
+
+fn crawl_disk(
+    base_path: &PathBuf,
+    path: &PathBuf,
+    rsync_base: Option<&uri::Rsync>
+) -> Result<Vec<CurrentFile>, 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<uri::Rsync, Error> {
+    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<Error> 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));
+        });
+    }
+
+}

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<T: DeserializeOwned>(
+    uri: &str,
+    token: Option<&str>
+) -> Result<T, Error> {
+    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<String, Error> {
+    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<T: DeserializeOwned>(
+    uri: &str,
+    data: impl Serialize,
+    token: Option<&str>
+) -> Result<T, 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()?;
+    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<Bytes, Error> {
+    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<u8> = 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, Error> {
+    Client::builder()
+        .gzip(true)
+        .timeout(Duration::from_secs(300))
+        .build()
+        .map_err(Error::RequestError)
+}
+
+fn headers(
+    content_type: Option<&str>,
+    token: Option<&str>
+) -> Result<HeaderMap, Error> {
+    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<T: DeserializeOwned>(
+    res: Response
+) -> Result<T, Error> {
+    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<Option<String>, 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<reqwest::Error> for Error {
+    fn from(e: reqwest::Error) -> Self { Error::RequestError(e) }
+}
+
+impl From<serde_json::Error> for Error {
+    fn from(e: serde_json::Error) -> Self { Error::JsonError(e) }
+}
+
+impl From<InvalidHeaderValue> for Error {
+    fn from(v: InvalidHeaderValue) -> Self { Error::InvalidHeader(v) }
+}

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<Utc>);
+
+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<S>(
+        &self, serializer: S
+    ) -> Result<S::Ok, S::Error> where S: Serializer {
+        serializer.serialize_i64(self.0.timestamp_millis())
+    }
+}
+
+impl<'de> Deserialize<'de> for Time {
+    fn deserialize<D>(
+        deserializer: D
+    ) -> Result<Self, D::Error> where D: Deserializer<'de> {
+
+        let timestamp: i64 = i64::deserialize(deserializer)?;
+        Ok(Time(Utc.timestamp_millis(timestamp)))
+    }
+}
+

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<str> for SignerKeyId {
+    fn as_ref(&self) -> &str {
+        &self.0
+    }
+}
+
+impl Serialize for SignerKeyId {
+    fn serialize<S>(
+        &self,
+        serializer: S
+    ) -> Result<S::Ok, S::Error> where S: Serializer {
+        self.as_ref().serialize(serializer)
+    }
+}
+
+impl<'de> Deserialize<'de> for SignerKeyId {
+    fn deserialize<D>(
+        deserializer: D
+    ) -> Result<Self, D::Error> 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<Self, SignerError> {
+        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<D: AsRef<[u8]> + ?Sized>(
+        pkey: &PKeyRef<Private>,
+        data: &D
+    ) -> Result<Signature, SignerError>
+    {
+        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<OpenSslKeyPair, SignerError> {
+        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<Self::KeyId, Self::Error> {
+        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<PublicKey, KeyError<Self::Error>> {
+        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<Self::Error>> {
+        let path = self.key_path(key_id);
+        if path.exists() {
+            fs::remove_file(path).map_err(SignerError::IoError)?;
+        }
+        Ok(())
+    }
+
+    fn sign<D: AsRef<[u8]> + ?Sized>(
+        &self,
+        key_id: &Self::KeyId,
+        _algorithm: SignatureAlgorithm,
+        data: &D
+    ) -> Result<Signature, SigningError<Self::Error>> {
+        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<D: AsRef<[u8]> + ?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<Private>
+}
+
+impl Serialize for OpenSslKeyPair {
+    fn serialize<S>(
+        &self,
+        s: S
+    ) -> Result<S::Ok, S::Error> where
+        S: Serializer {
+        let bytes: Vec<u8> = 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: D
+    ) -> Result<OpenSslKeyPair, D::Error> 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<OpenSslKeyPair, SignerError> {
+        // 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<PublicKey, SignerError> {
+        // 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<ErrorStack> for SignerError {
+    fn from(e: ErrorStack) -> Self {
+        SignerError::OpenSslError(e)
+    }
+}
+
+impl From<serde_json::Error> for SignerError {
+    fn from(e: serde_json::Error) -> Self {
+        SignerError::JsonError(e)
+    }
+}
+
+impl From<io::Error> 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);
+    }
+}

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<F>(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();
+}

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<R: io::Read> {
+    /// The underlying xml-rs reader
+    reader: EventReader<R>,
+
+    /// 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<XmlEvent>,
+
+    /// Name of the next start element, if any
+    next_start_name: Option<String>
+}
+
+
+/// Reader methods
+impl <R: io::Read> XmlReader<R> {
+
+    /// Gets the next XmlEvent
+    ///
+    /// Will take cached event if there is one
+    fn next(&mut self) -> Result<XmlEvent, XmlReaderErr> {
+        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 <R: io::Read> XmlReader<R> {
+    /// 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 <R: io::Read> XmlReader<R> {
+    /// 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<F, T, E>(source: R, op: F) -> Result<T, E>
+    where F: FnOnce(&mut Self) -> Result<T, E>,
+          E: From<XmlReaderErr> {
+        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<F, T, E>(&mut self, op: F) -> Result<T, E>
+    where F: FnOnce(&Tag, Attributes, &mut Self) -> Result<T, E>,
+          E: From<XmlReaderErr> {
+        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<F, T, E>(
+        &mut self,
+        name: &str,
+        op: F
+    ) -> Result<T, E>
+    where
+        F: FnOnce(Attributes, &mut Self) -> Result<T, E>,
+        E: From<XmlReaderErr>
+    {
+        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<F, T, E>(&mut self, op: F) -> Result<Option<T>, E>
+    where F: FnOnce(&Tag, Attributes, &mut Self) -> Result<Option<T>, E>,
+          E: From<XmlReaderErr> {
+
+        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<String, XmlReaderErr> {
+        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<Bytes, XmlReaderErr> {
+        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<fs::File> {
+
+    /// Opens a file and decodes it as an XML file.
+    pub fn open<P, F, T, E>(path: P, op: F) -> Result<T, E>
+    where F: FnOnce(&mut Self) -> Result<T, E>,
+          P: AsRef<Path>,
+          E: From<XmlReaderErr> + From<io::Error> {
+        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<io::Error> for XmlReaderErr {
+    fn from(e: io::Error) -> XmlReaderErr{
+        XmlReaderErr::IoError(e)
+    }
+}
+
+impl From<AttributesError> for XmlReaderErr {
+    fn from(e: AttributesError) -> XmlReaderErr {
+        XmlReaderErr::AttributesError(e)
+    }
+}
+
+impl From<reader::Error> for XmlReaderErr {
+    fn from(e: reader::Error) -> XmlReaderErr {
+        XmlReaderErr::ReaderError(e)
+    }
+}
+
+impl From<DecodeError> 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<OwnedAttribute>
+}
+
+impl Attributes {
+
+    /// Takes an optional attribute by name
+    pub fn take_opt(&mut self, name: &str) -> Option<String> {
+        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<Bytes> {
+        self.take_req_hex(name).ok()
+    }
+
+    /// Takes a required attribute by name
+    pub fn take_req(&mut self, name: &str) -> Result<String, AttributesError> {
+        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<Bytes, AttributesError> {
+
+        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<W> {
+    /// The underlying xml-rs writer
+    writer: EventWriter<W>
+}
+
+
+/// Generate the XML.
+impl <W: io::Write> XmlWriter<W> {
+
+
+    fn unwrap_emitter_error<T>(r: Result<T, writer::Error>) -> Result<T, io::Error> {
+        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<F>(
+        &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<u8>, File, etc.).
+    fn encode<F>(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<u8>
+    pub fn encode_vec<F>(op: F) -> Vec<u8>
+    where F: FnOnce(&mut XmlWriter<&mut Vec<u8>>)
+        -> 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<F>(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(),
+            "<a xmlns=\"http://ns/\" c=\"d\">\n  <b>WA==</b>\n</a>"
+        );
+    }
+}

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"

pubc/Cargo.toml

@@ -6,7 +6,6 @@ authors = ["Tim Bruijnzeels <tim@nlnetlabs.nl>", "Martin Hoffmann <martin@nlnetl
 [dependencies]
 clap            = "^2.32"
 derive_more     = "^0.13"
-krill_commons   = "^0.2"
 rpki            = "^0.3"
 serde           = { version = "^1.0", features = ["rc"] }
 serde_derive    = "^1.0"
@@ -16,3 +15,7 @@ serde_json      = "^1.0"
 path = "../cms_proxy"
 version = "0.2.0"
 
+[dependencies.krill_commons]
+path = "../commons"
+version = "0.2.0"
+

pubd/Cargo.toml

@@ -6,8 +6,11 @@ authors = ["Tim Bruijnzeels <tim@nlnetlabs.nl>", "Martin Hoffmann <martin@nlnetl
 [dependencies]
 bytes           = "^0.4"
 derive_more     = "^0.13"
-krill_commons   = "^0.2"
 rand            = "^0.5"
 rpki            = "^0.3"
 serde           = { version = "^1.0", features = ["rc"] }
-serde_derive    = "^1.0"
+serde_derive    = "^1.0"
+
+[dependencies.krill_commons]
+path = "../commons"
+version = "0.2.0"