src/login/mod.rs

@@ -1,8 +1,8 @@
 pub mod ldap_provider;
 pub mod static_provider;
 
-use std::collections::BTreeMap;
 use std::sync::Arc;
+use futures::try_join;
 
 use anyhow::{anyhow, bail, Context, Result};
 use async_trait::async_trait;
@@ -122,12 +122,13 @@ impl CryptoKeys {
         let password_blob = [&kdf_salt[..], &password_sealed].concat();
 
         // Write values to storage
-        k2v.insert_batch(&[
+        // @FIXME Implement insert batch in the storage API
-            k2v_insert_single_key("keys", "salt", salt_ct, ident_salt),
+        let (salt, public, passwd) = (
-            k2v_insert_single_key("keys", "public", public_ct, keys.public),
+            salt_ct.set_value(&ident_salt),
-            k2v_insert_single_key("keys", &password_sortkey, None, &password_blob),
+            public_ct.set_value(keys.public.as_ref()),
-        ])
+            k2v.row("keys", &password_sortkey).set_value(&password_blob)
-        .await
+        );
+        try_join!(salt.push(), public.push(), passwd.push())
             .context("InsertBatch for salt, public, and password")?;
 
         Ok(keys)
@@ -155,12 +156,13 @@ impl CryptoKeys {
         };
 
         // Write values to storage
-        k2v.insert_batch(&[
+        // @FIXME implement insert batch in the storage API
-            k2v_insert_single_key("keys", "salt", salt_ct, ident_salt),
+        let (salt, public) = (
-            k2v_insert_single_key("keys", "public", public_ct, keys.public),
+            salt_ct.set_value(&ident_salt),
-        ])
+            public_ct.set_value(keys.public.as_ref()),
-        .await
+        );
-        .context("InsertBatch for salt and public")?;
+
+        try_join!(salt.push(), public.push()).context("InsertBatch for salt and public")?;
 
         Ok(keys)
     }
@@ -170,7 +172,7 @@ impl CryptoKeys {
         user_secrets: &UserSecrets,
         password: &str,
     ) -> Result<Self> {
-        let k2v = storage.k2v_client()?;
+        let k2v = storage.row_store()?;
         let (ident_salt, expected_public) = Self::load_salt_and_public(&k2v).await?;
 
         // Generate short password digest (= password identity)
@@ -178,20 +180,21 @@ impl CryptoKeys {
 
         // Lookup password blob
         let password_sortkey = format!("password:{}", hex::encode(&ident));
+        let password_ref = k2v.row("keys", &password_sortkey);
 
         let password_blob = {
-            let mut val = match k2v.read_item("keys", &password_sortkey).await {
+            let val = match password_ref.fetch().await {
-                Err(k2v_client::Error::NotFound) => {
+                Err(StorageError::NotFound) => {
                     bail!("invalid password")
                 }
                 x => x?,
             };
-            if val.value.len() != 1 {
+            if val.content().len() != 1 {
                 bail!("multiple values for password in storage");
             }
-            match val.value.pop().unwrap() {
+            match val.content().pop().unwrap() {
-                K2vValue::Value(v) => v,
+                Alternative::Value(v) => v,
-                K2vValue::Tombstone => bail!("invalid password"),
+                Alternative::Tombstone => bail!("invalid password"),
             }
         };
 
@@ -258,24 +261,22 @@ impl CryptoKeys {
         let password_blob = [&kdf_salt[..], &password_sealed].concat();
 
         // List existing passwords to overwrite existing entry if necessary
-        let ct = match k2v.read_item("keys", &password_sortkey).await {
+        let pass_key = k2v.row("keys", &password_sortkey);
-            Err(k2v_client::Error::NotFound) => None,
+        let passwd = match pass_key.fetch().await {
+            Err(StorageError::NotFound) => pass_key,
             v => {
                 let entry = v?;
-                if entry.value.iter().any(|x| matches!(x, K2vValue::Value(_))) {
+                if entry.content().iter().any(|x| matches!(x, Alternative::Value(_))) {
                     bail!("password already exists");
                 }
-                Some(entry.causality)
+                entry.to_ref()
             }
         };
 
         // Write values to storage
-        k2v.insert_batch(&[k2v_insert_single_key(
+        passwd
-            "keys",
+            .set_value(&password_blob)
-            &password_sortkey,
+            .push()
-            ct,
-            &password_blob,
-        )])
             .await
             .context("InsertBatch for new password")?;
 
@@ -287,7 +288,7 @@ impl CryptoKeys {
         password: &str,
         allow_delete_all: bool,
     ) -> Result<()> {
-        let k2v = storage.row_client()?;
+        let k2v = storage.row_store()?;
         let (ident_salt, _public) = Self::load_salt_and_public(&k2v).await?;
 
         // Generate short password digest (= password identity)
@@ -299,22 +300,23 @@ impl CryptoKeys {
 
         // Check password is there
         let pw = existing_passwords
-            .get(&password_sortkey)
+            .iter()
+            .map(|x| x.to_ref())
+            .find(|x| x.key().1 == &password_sortkey)
+            //.get(&password_sortkey)
             .ok_or(anyhow!("password does not exist"))?;
 
         if !allow_delete_all && existing_passwords.len() < 2 {
             bail!("No other password exists, not deleting last password.");
         }
 
-        k2v.delete_item("keys", &password_sortkey, pw.causality.clone())
+        pw.rm().await.context("DeleteItem for password")?;
-            .await
-            .context("DeleteItem for password")?;
 
         Ok(())
     }
 
     // ---- STORAGE UTIL ----
-
+    //
     async fn check_uninitialized(
         k2v: &RowStore,
     ) -> Result<(RowRef, RowRef)> {
@@ -346,32 +348,29 @@ impl CryptoKeys {
         Ok((salt_ct, public_ct))
     }
 
-    pub async fn load_salt_and_public(k2v: &K2vClient) -> Result<([u8; 32], PublicKey)> {
+    pub async fn load_salt_and_public(k2v: &RowStore) -> Result<([u8; 32], PublicKey)> {
-        let mut params = k2v
+        let params = k2v
-            .read_batch(&[
+            .select(Selector::List(vec![
-                k2v_read_single_key("keys", "salt", false),
+                ("keys", "salt"),
-                k2v_read_single_key("keys", "public", false),
+                ("keys", "public"),
-            ])
+            ]))
             .await
             .context("ReadBatch for salt and public in load_salt_and_public")?;
+
         if params.len() != 2 {
             bail!(
                 "Invalid response from k2v storage: {:?} (expected two items)",
                 params
             );
         }
-        if params[0].items.len() != 1 || params[1].items.len() != 1 {
+        if params[0].content().len() != 1 || params[1].content().len() != 1 {
             bail!("cryptographic keys not initialized for user");
         }
 
         // Retrieve salt from given response
-        let salt_vals = &mut params[0].items.iter_mut().next().unwrap().1.value;
+        let salt: Vec<u8> = match &mut params[0].content().iter_mut().next().unwrap() {
-        if salt_vals.len() != 1 {
+            Alternative::Value(v) => std::mem::take(v),
-            bail!("Multiple values for `salt`");
+            Alternative::Tombstone => bail!("salt is a tombstone"),
-        }
-        let salt: Vec<u8> = match &mut salt_vals[0] {
-            K2vValue::Value(v) => std::mem::take(v),
-            K2vValue::Tombstone => bail!("salt is a tombstone"),
         };
         if salt.len() != 32 {
             bail!("`salt` is not 32 bytes long");
@@ -380,40 +379,21 @@ impl CryptoKeys {
         salt_constlen.copy_from_slice(&salt);
 
         // Retrieve public from given response
-        let public_vals = &mut params[1].items.iter_mut().next().unwrap().1.value;
+        let public: Vec<u8> = match &mut params[1].content().iter_mut().next().unwrap() {
-        if public_vals.len() != 1 {
+            Alternative::Value(v) => std::mem::take(v),
-            bail!("Multiple values for `public`");
+            Alternative::Tombstone => bail!("public is a tombstone"),
-        }
-        let public: Vec<u8> = match &mut public_vals[0] {
-            K2vValue::Value(v) => std::mem::take(v),
-            K2vValue::Tombstone => bail!("public is a tombstone"),
         };
         let public = PublicKey::from_slice(&public).ok_or(anyhow!("Invalid public key length"))?;
 
         Ok((salt_constlen, public))
     }
 
-    async fn list_existing_passwords(k2v: &K2vClient) -> Result<BTreeMap<String, CausalValue>> {
+    async fn list_existing_passwords(k2v: &RowStore) -> Result<Vec<RowValue>> {
-        let mut res = k2v
+        let res = k2v.select(Selector::Prefix { shard_key: "keys", prefix: "password:" })
-            .read_batch(&[BatchReadOp {
-                partition_key: "keys",
-                filter: Filter {
-                    start: None,
-                    end: None,
-                    prefix: Some("password:"),
-                    limit: None,
-                    reverse: false,
-                },
-                conflicts_only: false,
-                tombstones: false,
-                single_item: false,
-            }])
             .await
             .context("ReadBatch for prefix password: in list_existing_passwords")?;
-        if res.len() != 1 {
+
-            bail!("unexpected k2v result: {:?}, expected one item", res);
+        Ok(res)
-        }
-        Ok(res.pop().unwrap().items)
     }
 
     fn serialize(&self) -> [u8; 64] {

src/login/static_provider.rs

@@ -51,31 +51,12 @@ impl LoginProvider for StaticLoginProvider {
             bail!("Wrong password");
         }
 
-        /*
-        tracing::debug!(user=%username, "fetch bucket");
-        let bucket = user
-            .bucket
-            .clone()
-            .or_else(|| self.default_bucket.clone())
-            .ok_or(anyhow!(
-                "No bucket configured and no default bucket specified"
-            ))?;*/
-
         tracing::debug!(user=%username, "fetch keys");
         let storage: storage::Builders = match user.storage {
             StaticStorage::InMemory => Box::new(storage::in_memory::FullMem {}),
             StaticStorage::Garage(c) => Box::new(storage::garage::GrgCreds {}),
         };
 
-        /*
-        StorageCredentials {
-            k2v_region: self.k2v_region.clone(),
-            s3_region: self.s3_region.clone(),
-            aws_access_key_id: user.aws_access_key_id.clone(),
-            aws_secret_access_key: user.aws_secret_access_key.clone(),
-            bucket,
-        };*/
-
         let keys = match (&user.master_key, &user.secret_key) {
             (Some(m), Some(s)) => {
                 let master_key =

src/storage/garage.rs

@@ -53,7 +53,7 @@ impl IRowRef for GrgRef {
         unimplemented!();
     }
 
-    fn set_value(&self, content: Vec<u8>) -> RowValue {
+    fn set_value(&self, content: &[u8]) -> RowValue {
         unimplemented!();
     }
     fn fetch(&self) -> AsyncResult<RowValue> {

src/storage/in_memory.rs

@@ -55,7 +55,7 @@ impl IRowRef for MemRef {
         unimplemented!();
     }*/
 
-    fn set_value(&self, content: Vec<u8>) -> RowValue {
+    fn set_value(&self, content: &[u8]) -> RowValue {
         unimplemented!();
     }
     fn fetch(&self) -> AsyncResult<RowValue> {

src/storage/mod.rs

@@ -95,21 +95,14 @@ pub type RowStore = Box<dyn IRowStore + Sync + Send>;
 
 pub trait IRowRef: std::fmt::Debug
 {
-    /*fn clone_boxed(&self) -> RowRef;*/
     fn to_orphan(&self) -> OrphanRowRef;
     fn key(&self) -> (&str, &str);
-    fn set_value(&self, content: Vec<u8>) -> RowValue;
+    fn set_value(&self, content: &[u8]) -> RowValue;
     fn fetch(&self) -> AsyncResult<RowValue>;
     fn rm(&self) -> AsyncResult<()>;
     fn poll(&self) -> AsyncResult<RowValue>;
 }
 pub type RowRef = Box<dyn IRowRef + Send + Sync>;
-/*impl Clone for RowRef {
-    fn clone(&self) -> Self {
-        return self.clone_boxed()
-    }
-}*/
-
 
 pub trait IRowValue: std::fmt::Debug
 {