Support async storage

Cargo.toml

@@ -21,6 +21,7 @@ exclude = [
 ]
 
 [dependencies]
+async-trait     = "0.1.74"
 backoff         = { version = "0.4.0", optional = true }
 base64          = "^0.13"
 basic-cookies   = { version = "^0.1", optional = true }
@@ -39,6 +40,7 @@ intervaltree    = "0.2.6"
 jmespatch       = { version = "^0.3", features = ["sync"], optional = true }
 kmip            = { version = "0.4.2", package = "kmip-protocol", features = [ "tls-with-openssl" ], optional = true }
 kvx             = { version = "0.9.3", features = ["macros"] }
+lazy_static     = "1.4.0"
 libflate        = "^1"
 log             = "^0.4"
 once_cell       = { version = "^1.7.2", optional = true }
@@ -51,11 +53,13 @@ regex           = { version = "1.5.5", optional = true, default_features = false
 reqwest         = { version = "0.11", features = ["json"] }
 rpassword       = { version = "^5.0", optional = true }
 #rpki            = { version = "0.18.0", features = ["ca", "compat", "rrdp"] }
-rpki            = { git = "https://github.com/nLnetLabs/rpki-rs", features = [ "ca", "compat", "rrdp" ] }
+rpki            = { git = "https://github.com/nLnetLabs/rpki-rs", branch = "async-signer-trait", features = [ "ca", "compat", "rrdp" ] }
+rustix          = { version = "0.38.28", features = ["fs"] }
 rustls-pemfile  = "1.0.4"
 scrypt          = { version = "^0.6", optional = true, default-features = false }
 serde           = { version = "^1.0", features = ["derive", "rc"] }
 serde_json      = "^1.0"
+tempfile        = "3.1.0"
 tokio           = { version = "1", features = [ "macros", "rt", "rt-multi-thread", "signal", "time" ] }
 tokio-rustls    = "0.24.1"
 toml            = "^0.5"

src/bin/krillup.rs

@@ -13,7 +13,8 @@ use krill::{
 };
 use url::Url;
 
-fn main() {
+#[tokio::main]
+async fn main() {
     let matches = make_matches();
 
     match parse_matches(matches) {
@@ -36,7 +37,7 @@ fn main() {
                     }
                 };
 
-                match prepare_upgrade_data_migrations(UpgradeMode::PrepareOnly, &config, &properties_manager) {
+                match prepare_upgrade_data_migrations(UpgradeMode::PrepareOnly, &config, &properties_manager).await {
                     Err(e) => {
                         eprintln!("*** Error Preparing Data Migration ***");
                         eprintln!("{}", e);
@@ -62,7 +63,7 @@ fn main() {
                 }
             }
             KrillUpMode::Migrate { config, target } => {
-                if let Err(e) = migrate(config, target) {
+                if let Err(e) = migrate(config, target).await {
                     eprintln!("*** Error Migrating DATA ***");
                     eprintln!("{}", e);
                     eprintln!();

src/cli/ta_client.rs

@@ -19,7 +19,8 @@ use crate::{
         api::{AddChildRequest, ApiRepositoryContact, CertAuthInfo, IdCertInfo, RepositoryContact, Token},
         crypto::KrillSigner,
         error::Error as KrillError,
-        eventsourcing::{namespace, AggregateStore, AggregateStoreError, Namespace},
+        eventsourcing::{AggregateStore, AggregateStoreError},
+        storage::NamespaceBuf,
         util::{file, httpclient},
     },
     constants::{KRILL_CLI_API_ENV, KRILL_CLI_FORMAT_ENV, KRILL_TA_CLIENT_APP, KRILL_VERSION},
@@ -912,14 +913,14 @@ impl TrustAnchorClient {
                 let signer_manager = TrustAnchorSignerManager::create(signer_command.config)?;
 
                 match signer_command.details {
-                    SignerCommandDetails::Init(info) => signer_manager.init(info),
-                    SignerCommandDetails::ShowInfo => signer_manager.show(),
+                    SignerCommandDetails::Init(info) => signer_manager.init(info).await,
+                    SignerCommandDetails::ShowInfo => signer_manager.show().await,
                     SignerCommandDetails::ProcessRequest {
                         signed_request,
                         ta_mft_number_override,
-                    } => signer_manager.process(signed_request, ta_mft_number_override),
-                    SignerCommandDetails::ShowLastResponse => signer_manager.show_last_response(),
-                    SignerCommandDetails::ShowExchanges => signer_manager.show_exchanges(),
+                    } => signer_manager.process(signed_request, ta_mft_number_override).await,
+                    SignerCommandDetails::ShowLastResponse => signer_manager.show_last_response().await,
+                    SignerCommandDetails::ShowExchanges => signer_manager.show_exchanges().await,
                 }
             }
         }
@@ -1044,8 +1045,12 @@ struct TrustAnchorSignerManager {
 
 impl TrustAnchorSignerManager {
     fn create(config: Config) -> Result<Self, TaClientError> {
-        let store = AggregateStore::create(&config.storage_uri, namespace!("signer"), config.use_history_cache)
-            .map_err(KrillError::AggregateStoreError)?;
+        let store = AggregateStore::create(
+            &config.storage_uri,
+            NamespaceBuf::parse_lossy("signer").as_ref(),
+            config.use_history_cache,
+        )
+        .map_err(KrillError::AggregateStoreError)?;
         let ta_handle = TrustAnchorHandle::new("ta".into());
         let config = Arc::new(config);
         let signer = config.signer()?;
@@ -1060,8 +1065,8 @@ impl TrustAnchorSignerManager {
         })
     }
 
-    fn init(&self, info: SignerInitInfo) -> Result<TrustAnchorClientApiResponse, TaClientError> {
-        if self.store.has(&self.ta_handle)? {
+    async fn init(&self, info: SignerInitInfo) -> Result<TrustAnchorClientApiResponse, TaClientError> {
+        if self.store.has(&self.ta_handle).await? {
             Err(TaClientError::other("Trust Anchor Signer was already initialised."))
         } else {
             let cmd = TrustAnchorSignerInitCommand::new(
@@ -1079,19 +1084,19 @@ impl TrustAnchorSignerManager {
                 &self.actor,
             );
 
-            self.store.add(cmd)?;
+            self.store.add(cmd).await?;
 
             Ok(TrustAnchorClientApiResponse::Empty)
         }
     }
 
-    fn show(&self) -> Result<TrustAnchorClientApiResponse, TaClientError> {
-        let ta_signer = self.get_signer()?;
+    async fn show(&self) -> Result<TrustAnchorClientApiResponse, TaClientError> {
+        let ta_signer = self.get_signer().await?;
         let info = ta_signer.get_signer_info();
         Ok(TrustAnchorClientApiResponse::TrustAnchorProxySignerInfo(info))
     }
 
-    fn process(
+    async fn process(
         &self,
         signed_request: TrustAnchorSignedRequest,
         ta_mft_number_override: Option<u64>,
@@ -1104,20 +1109,21 @@ impl TrustAnchorSignerManager {
             self.signer.clone(),
             &self.actor,
         );
-        self.store.command(cmd)?;
+        self.store.command(cmd).await?;
 
-        self.show_last_response()
+        self.show_last_response().await
     }
 
-    fn show_last_response(&self) -> Result<TrustAnchorClientApiResponse, TaClientError> {
-        self.get_signer()?
+    async fn show_last_response(&self) -> Result<TrustAnchorClientApiResponse, TaClientError> {
+        self.get_signer()
+            .await?
             .get_latest_exchange()
             .map(|exchange| TrustAnchorClientApiResponse::SignerResponse(exchange.response.clone()))
             .ok_or_else(|| TaClientError::other("No response found."))
     }
 
-    fn show_exchanges(&self) -> Result<TrustAnchorClientApiResponse, TaClientError> {
-        let signer = self.get_signer()?;
+    async fn show_exchanges(&self) -> Result<TrustAnchorClientApiResponse, TaClientError> {
+        let signer = self.get_signer().await?;
         // In this context it's okay to clone the exchanges.
         // If we are afraid that this would become too expensive, then we will
         // need to rethink the model where we return data in the enum that we
@@ -1130,10 +1136,11 @@ impl TrustAnchorSignerManager {
         Ok(TrustAnchorClientApiResponse::ProxySignerExchanges(exchanges))
     }
 
-    fn get_signer(&self) -> Result<Arc<TrustAnchorSigner>, TaClientError> {
-        if self.store.has(&self.ta_handle)? {
+    async fn get_signer(&self) -> Result<Arc<TrustAnchorSigner>, TaClientError> {
+        if self.store.has(&self.ta_handle).await? {
             self.store
                 .get_latest(&self.ta_handle)
+                .await
                 .map_err(TaClientError::KrillError)
         } else {
             Err(TaClientError::other("Trust Anchor Signer is not initialised."))

src/commons/api/ca.rs

@@ -2192,29 +2192,28 @@ mod test {
         assert_eq!(base_uri(), signed_objects_uri)
     }
 
-    #[test]
-    fn mft_uri() {
-        test::test_in_memory(|storage_uri| {
-            let signer = OpenSslSigner::build(storage_uri, "dummy", None).unwrap();
-            let key_id = signer.create_key(PublicKeyFormat::Rsa).unwrap();
-            let pub_key = signer.get_key_info(&key_id).unwrap();
+    #[tokio::test]
+    async fn mft_uri() {
+        let storage_uri = test::mem_storage();
+        let signer = OpenSslSigner::build(&storage_uri, "dummy", None).unwrap();
+        let key_id = signer.create_key(PublicKeyFormat::Rsa).await.unwrap();
+        let pub_key = signer.get_key_info(&key_id).await.unwrap();
 
-            let mft_uri = info().resolve("", ObjectName::mft_for_key(&pub_key.key_identifier()).as_ref());
+        let mft_uri = info().resolve("", ObjectName::mft_for_key(&pub_key.key_identifier()).as_ref());
 
-            let mft_path = mft_uri.relative_to(&base_uri()).unwrap();
+        let mft_path = mft_uri.relative_to(&base_uri()).unwrap();
 
-            assert_eq!(44, mft_path.len());
+        assert_eq!(44, mft_path.len());
 
-            // the file name should be the hexencoded pub key info
-            // not repeating that here, but checking that the name
-            // part is validly hex encoded.
-            let name = &mft_path[..40];
-            hex::decode(name).unwrap();
+        // the file name should be the hexencoded pub key info
+        // not repeating that here, but checking that the name
+        // part is validly hex encoded.
+        let name = &mft_path[..40];
+        hex::decode(name).unwrap();
 
-            // and the extension is '.mft'
-            let ext = &mft_path[40..];
-            assert_eq!(ext, ".mft");
-        });
+        // and the extension is '.mft'
+        let ext = &mft_path[40..];
+        assert_eq!(ext, ".mft");
     }
 
     #[test]

src/commons/api/import.rs

@@ -163,6 +163,10 @@ impl ImportCa {
         ImportCa { handle, parents, roas }
     }
 
+    pub fn handle(&self) -> &CaHandle {
+        &self.handle
+    }
+
     pub fn unpack(self) -> (CaHandle, Vec<ImportParent>, Vec<RoaConfiguration>) {
         (self.handle, self.parents, self.roas)
     }
@@ -220,11 +224,7 @@ pub struct ImportChildCertificate {
 impl fmt::Display for ImportChild {
     fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
         writeln!(f, "Name:         {}", self.name)?;
-        writeln!(
-            f,
-            "Id Key:       {}",
-            self.id_cert.public_key().key_identifier().to_string()
-        )?;
+        writeln!(f, "Id Key:       {}", self.id_cert.public_key().key_identifier())?;
         writeln!(f, "Resources:    {}", self.resources)?;
         if let Some(class_name) = &self.issued_cert.class_name {
             writeln!(f, "Classname:    {}", class_name)?;