diff --git a/src/bayou.rs b/src/bayou.rs index 1361e49..7253a30 100644 --- a/src/bayou.rs +++ b/src/bayou.rs @@ -9,9 +9,8 @@ use tokio::sync::{watch, Notify}; use crate::cryptoblob::*; use crate::login::Credentials; -use crate::timestamp::*; use crate::storage; - +use crate::timestamp::*; const KEEP_STATE_EVERY: usize = 64; @@ -94,7 +93,11 @@ impl Bayou { } else { debug!("(sync) loading checkpoint: {}", key); - let buf = self.storage.blob_fetch(&storage::BlobRef(key.to_string())).await?.value; + let buf = self + .storage + .blob_fetch(&storage::BlobRef(key.to_string())) + .await? + .value; debug!("(sync) checkpoint body length: {}", buf.len()); let ck = open_deserialize::(&buf, &self.key)?; @@ -125,17 +128,22 @@ impl Bayou { // 3. List all operations starting from checkpoint let ts_ser = self.checkpoint.0.to_string(); debug!("(sync) looking up operations starting at {}", ts_ser); - let ops_map = self.storage.row_fetch(&storage::Selector::Range { - shard: &self.path, - sort_begin: &ts_ser, - sort_end: WATCH_SK - }).await?; + let ops_map = self + .storage + .row_fetch(&storage::Selector::Range { + shard: &self.path, + sort_begin: &ts_ser, + sort_end: WATCH_SK, + }) + .await?; let mut ops = vec![]; for row_value in ops_map { let row = row_value.row_ref; let sort_key = row.uid.sort; - let ts = sort_key.parse::().map_err(|_| anyhow!("Invalid operation timestamp: {}", sort_key))?; + let ts = sort_key + .parse::() + .map_err(|_| anyhow!("Invalid operation timestamp: {}", sort_key))?; let val = row_value.value; if val.len() != 1 { @@ -211,7 +219,7 @@ impl Bayou { // Save info that sync has been done self.last_sync = new_last_sync; - self.last_sync_watch_ct = new_last_sync_watch_ct; + self.last_sync_watch_ct = new_last_sync_watch_ct; Ok(()) } @@ -362,16 +370,20 @@ impl Bayou { // Delete blobs for (_ts, key) in existing_checkpoints[..last_to_keep].iter() { debug!("(cp) drop old checkpoint {}", key); - self.storage.blob_rm(&storage::BlobRef(key.to_string())).await?; + self.storage + .blob_rm(&storage::BlobRef(key.to_string())) + .await?; } // Delete corresponding range of operations let ts_ser = existing_checkpoints[last_to_keep].0.to_string(); - self.storage.row_rm(&storage::Selector::Range { - shard: &self.path, - sort_begin: "", - sort_end: &ts_ser - }).await? + self.storage + .row_rm(&storage::Selector::Range { + shard: &self.path, + sort_begin: "", + sort_end: &ts_ser, + }) + .await? } Ok(()) @@ -426,11 +438,7 @@ impl K2vWatch { let watch = Arc::new(K2vWatch { target, rx, notify }); - tokio::spawn(Self::background_task( - Arc::downgrade(&watch), - storage, - tx, - )); + tokio::spawn(Self::background_task(Arc::downgrade(&watch), storage, tx)); Ok(watch) } @@ -444,8 +452,8 @@ impl K2vWatch { Some(this) => this.target.clone(), None => { error!("can't start loop"); - return - }, + return; + } }; while let Some(this) = Weak::upgrade(&self_weak) { diff --git a/src/login/ldap_provider.rs b/src/login/ldap_provider.rs index a7f56e4..81e5879 100644 --- a/src/login/ldap_provider.rs +++ b/src/login/ldap_provider.rs @@ -30,7 +30,10 @@ enum BucketSource { enum StorageSpecific { InMemory, - Garage { from_config: LdapGarageConfig, bucket_source: BucketSource }, + Garage { + from_config: LdapGarageConfig, + bucket_source: BucketSource, + }, } impl LdapLoginProvider { @@ -57,22 +60,24 @@ impl LdapLoginProvider { let specific = match config.storage { LdapStorage::InMemory => StorageSpecific::InMemory, LdapStorage::Garage(grgconf) => { - let bucket_source = match (grgconf.default_bucket.clone(), grgconf.bucket_attr.clone()) { - (Some(b), None) => BucketSource::Constant(b), - (None, Some(a)) => BucketSource::Attr(a), - _ => bail!("Must set `bucket` or `bucket_attr`, but not both"), - }; + let bucket_source = + match (grgconf.default_bucket.clone(), grgconf.bucket_attr.clone()) { + (Some(b), None) => BucketSource::Constant(b), + (None, Some(a)) => BucketSource::Attr(a), + _ => bail!("Must set `bucket` or `bucket_attr`, but not both"), + }; if let BucketSource::Attr(a) = &bucket_source { attrs_to_retrieve.push(a.clone()); } - StorageSpecific::Garage { from_config: grgconf, bucket_source } - }, + StorageSpecific::Garage { + from_config: grgconf, + bucket_source, + } + } }; - - Ok(Self { ldap_server: config.ldap_server, pre_bind_on_login: config.pre_bind_on_login, @@ -89,27 +94,32 @@ impl LdapLoginProvider { async fn storage_creds_from_ldap_user(&self, user: &SearchEntry) -> Result { let storage: Builder = match &self.storage_specific { - StorageSpecific::InMemory => self.in_memory_store.builder( - &get_attr(user, &self.username_attr)? - ).await, - StorageSpecific::Garage { from_config, bucket_source } => { + StorageSpecific::InMemory => { + self.in_memory_store + .builder(&get_attr(user, &self.username_attr)?) + .await + } + StorageSpecific::Garage { + from_config, + bucket_source, + } => { let aws_access_key_id = get_attr(user, &from_config.aws_access_key_id_attr)?; - let aws_secret_access_key = get_attr(user, &from_config.aws_secret_access_key_attr)?; + let aws_secret_access_key = + get_attr(user, &from_config.aws_secret_access_key_attr)?; let bucket = match bucket_source { BucketSource::Constant(b) => b.clone(), BucketSource::Attr(a) => get_attr(user, &a)?, }; - storage::garage::GarageBuilder::new(storage::garage::GarageConf { - region: from_config.aws_region.clone(), + region: from_config.aws_region.clone(), s3_endpoint: from_config.s3_endpoint.clone(), k2v_endpoint: from_config.k2v_endpoint.clone(), - aws_access_key_id, - aws_secret_access_key, + aws_access_key_id, + aws_secret_access_key, bucket, })? - }, + } }; Ok(storage) @@ -172,7 +182,6 @@ impl LoginProvider for LdapLoginProvider { drop(ldap); - Ok(Credentials { storage, keys }) } @@ -215,7 +224,7 @@ impl LoginProvider for LdapLoginProvider { let cr = CryptoRoot(crstr); let public_key = cr.public_key()?; - // storage + // storage let storage = self.storage_creds_from_ldap_user(&user).await?; drop(ldap); diff --git a/src/login/mod.rs b/src/login/mod.rs index 3369ac2..2926738 100644 --- a/src/login/mod.rs +++ b/src/login/mod.rs @@ -1,8 +1,8 @@ pub mod ldap_provider; pub mod static_provider; -use std::sync::Arc; use base64::Engine; +use std::sync::Arc; use anyhow::{anyhow, bail, Context, Result}; use async_trait::async_trait; @@ -45,7 +45,7 @@ pub struct PublicCredentials { pub public_key: PublicKey, } -use serde::{Serialize, Deserialize}; +use serde::{Deserialize, Serialize}; #[derive(Serialize, Deserialize, Debug, Clone)] pub struct CryptoRoot(pub String); @@ -73,47 +73,59 @@ impl CryptoRoot { pub fn public_key(&self) -> Result { match self.0.splitn(4, ':').collect::>()[..] { - [ "aero", "cryptoroot", "pass", b64blob ] => { + ["aero", "cryptoroot", "pass", b64blob] => { let blob = base64::engine::general_purpose::STANDARD_NO_PAD.decode(b64blob)?; if blob.len() < 32 { - bail!("Decoded data is {} bytes long, expect at least 32 bytes", blob.len()); + bail!( + "Decoded data is {} bytes long, expect at least 32 bytes", + blob.len() + ); } PublicKey::from_slice(&blob[..32]).context("must be a valid public key") - }, - [ "aero", "cryptoroot", "cleartext", b64blob ] => { + } + ["aero", "cryptoroot", "cleartext", b64blob] => { let blob = base64::engine::general_purpose::STANDARD_NO_PAD.decode(b64blob)?; Ok(CryptoKeys::deserialize(&blob)?.public) - }, - [ "aero", "cryptoroot", "incoming", b64blob ] => { + } + ["aero", "cryptoroot", "incoming", b64blob] => { let blob = base64::engine::general_purpose::STANDARD_NO_PAD.decode(b64blob)?; if blob.len() < 32 { - bail!("Decoded data is {} bytes long, expect at least 32 bytes", blob.len()); + bail!( + "Decoded data is {} bytes long, expect at least 32 bytes", + blob.len() + ); } PublicKey::from_slice(&blob[..32]).context("must be a valid public key") - }, - [ "aero", "cryptoroot", "keyring", _ ] => { + } + ["aero", "cryptoroot", "keyring", _] => { bail!("keyring is not yet implemented!") - }, - _ => bail!(format!("passed string '{}' is not a valid cryptoroot", self.0)), + } + _ => bail!(format!( + "passed string '{}' is not a valid cryptoroot", + self.0 + )), } } pub fn crypto_keys(&self, password: &str) -> Result { match self.0.splitn(4, ':').collect::>()[..] { - [ "aero", "cryptoroot", "pass", b64blob ] => { + ["aero", "cryptoroot", "pass", b64blob] => { let blob = base64::engine::general_purpose::STANDARD_NO_PAD.decode(b64blob)?; CryptoKeys::password_open(password, &blob) - }, - [ "aero", "cryptoroot", "cleartext", b64blob ] => { + } + ["aero", "cryptoroot", "cleartext", b64blob] => { let blob = base64::engine::general_purpose::STANDARD_NO_PAD.decode(b64blob)?; CryptoKeys::deserialize(&blob) - }, - [ "aero", "cryptoroot", "incoming", _ ] => { + } + ["aero", "cryptoroot", "incoming", _] => { bail!("incoming cryptoroot does not contain a crypto key!") - }, - [ "aero", "cryptoroot", "keyring", _ ] =>{ + } + ["aero", "cryptoroot", "keyring", _] => { bail!("keyring is not yet implemented!") - }, - _ => bail!(format!("passed string '{}' is not a valid cryptoroot", self.0)), + } + _ => bail!(format!( + "passed string '{}' is not a valid cryptoroot", + self.0 + )), } } } @@ -132,9 +144,6 @@ pub struct CryptoKeys { // ---- - - - impl CryptoKeys { /// Initialize a new cryptography root pub fn init() -> Self { @@ -202,7 +211,11 @@ fn derive_password_key(kdf_salt: &[u8], password: &str) -> Result { Ok(Key::from_slice(&argon2_kdf(kdf_salt, password.as_bytes(), 32)?).unwrap()) } -fn try_open_encrypted_keys(kdf_salt: &[u8], password: &str, encrypted_keys: &[u8]) -> Result> { +fn try_open_encrypted_keys( + kdf_salt: &[u8], + password: &str, + encrypted_keys: &[u8], +) -> Result> { let password_key = derive_password_key(kdf_salt, password)?; open(encrypted_keys, &password_key) } @@ -210,7 +223,7 @@ fn try_open_encrypted_keys(kdf_salt: &[u8], password: &str, encrypted_keys: &[u8 // ---- UTIL ---- pub fn argon2_kdf(salt: &[u8], password: &[u8], output_len: usize) -> Result> { - use argon2::{Algorithm, Argon2, ParamsBuilder, PasswordHasher, Version, password_hash}; + use argon2::{password_hash, Algorithm, Argon2, ParamsBuilder, PasswordHasher, Version}; let params = ParamsBuilder::new() .output_len(output_len) @@ -219,7 +232,8 @@ pub fn argon2_kdf(salt: &[u8], password: &[u8], output_len: usize) -> Result) -> Result<()> { - let mut stream = signal(SignalKind::user_defined1()).expect("failed to install SIGUSR1 signal hander for reload"); + let mut stream = signal(SignalKind::user_defined1()) + .expect("failed to install SIGUSR1 signal hander for reload"); loop { let ulist: UserList = match read_config(config.clone()) { @@ -42,7 +43,12 @@ pub async fn update_user_list(config: PathBuf, up: watch::Sender) let users = ulist .into_iter() - .map(|(username, config)| (username.clone() , Arc::new(ContextualUserEntry { username, config }))) + .map(|(username, config)| { + ( + username.clone(), + Arc::new(ContextualUserEntry { username, config }), + ) + }) .collect::>(); let mut users_by_email = HashMap::new(); @@ -51,14 +57,18 @@ pub async fn update_user_list(config: PathBuf, up: watch::Sender) if users_by_email.contains_key(m) { tracing::warn!("Several users have the same email address: {}", m); stream.recv().await; - continue + continue; } users_by_email.insert(m.clone(), u.clone()); } } tracing::info!("{} users loaded", users.len()); - up.send(UserDatabase { users, users_by_email }).context("update user db config")?; + up.send(UserDatabase { + users, + users_by_email, + }) + .context("update user db config")?; stream.recv().await; tracing::info!("Received SIGUSR1, reloading"); } @@ -71,7 +81,10 @@ impl StaticLoginProvider { tokio::spawn(update_user_list(config.user_list, tx)); rx.changed().await?; - Ok(Self { user_db: rx, in_memory_store: storage::in_memory::MemDb::new() }) + Ok(Self { + user_db: rx, + in_memory_store: storage::in_memory::MemDb::new(), + }) } } @@ -95,14 +108,16 @@ impl LoginProvider for StaticLoginProvider { tracing::debug!(user=%username, "fetch keys"); let storage: storage::Builder = match &user.config.storage { StaticStorage::InMemory => self.in_memory_store.builder(username).await, - StaticStorage::Garage(grgconf) => storage::garage::GarageBuilder::new(storage::garage::GarageConf { - region: grgconf.aws_region.clone(), - k2v_endpoint: grgconf.k2v_endpoint.clone(), - s3_endpoint: grgconf.s3_endpoint.clone(), - aws_access_key_id: grgconf.aws_access_key_id.clone(), - aws_secret_access_key: grgconf.aws_secret_access_key.clone(), - bucket: grgconf.bucket.clone(), - })?, + StaticStorage::Garage(grgconf) => { + storage::garage::GarageBuilder::new(storage::garage::GarageConf { + region: grgconf.aws_region.clone(), + k2v_endpoint: grgconf.k2v_endpoint.clone(), + s3_endpoint: grgconf.s3_endpoint.clone(), + aws_access_key_id: grgconf.aws_access_key_id.clone(), + aws_secret_access_key: grgconf.aws_secret_access_key.clone(), + bucket: grgconf.bucket.clone(), + })? + } }; let cr = CryptoRoot(user.config.crypto_root.clone()); @@ -124,14 +139,16 @@ impl LoginProvider for StaticLoginProvider { let storage: storage::Builder = match &user.config.storage { StaticStorage::InMemory => self.in_memory_store.builder(&user.username).await, - StaticStorage::Garage(grgconf) => storage::garage::GarageBuilder::new(storage::garage::GarageConf { - region: grgconf.aws_region.clone(), - k2v_endpoint: grgconf.k2v_endpoint.clone(), - s3_endpoint: grgconf.s3_endpoint.clone(), - aws_access_key_id: grgconf.aws_access_key_id.clone(), - aws_secret_access_key: grgconf.aws_secret_access_key.clone(), - bucket: grgconf.bucket.clone(), - })?, + StaticStorage::Garage(grgconf) => { + storage::garage::GarageBuilder::new(storage::garage::GarageConf { + region: grgconf.aws_region.clone(), + k2v_endpoint: grgconf.k2v_endpoint.clone(), + s3_endpoint: grgconf.s3_endpoint.clone(), + aws_access_key_id: grgconf.aws_access_key_id.clone(), + aws_secret_access_key: grgconf.aws_secret_access_key.clone(), + bucket: grgconf.bucket.clone(), + })? + } }; let cr = CryptoRoot(user.config.crypto_root.clone()); diff --git a/src/mail/incoming.rs b/src/mail/incoming.rs index 7e33a9a..04d2ef1 100644 --- a/src/mail/incoming.rs +++ b/src/mail/incoming.rs @@ -51,7 +51,10 @@ async fn incoming_mail_watch_process_internal( creds: Credentials, mut rx_inbox_id: watch::Receiver>, ) -> Result<()> { - let mut lock_held = k2v_lock_loop(creds.storage.build().await?, storage::RowRef::new(INCOMING_PK, INCOMING_LOCK_SK)); + let mut lock_held = k2v_lock_loop( + creds.storage.build().await?, + storage::RowRef::new(INCOMING_PK, INCOMING_LOCK_SK), + ); let storage = creds.storage.build().await?; let mut inbox: Option> = None; @@ -63,8 +66,7 @@ async fn incoming_mail_watch_process_internal( let wait_new_mail = async { loop { - match storage.row_poll(&incoming_key).await - { + match storage.row_poll(&incoming_key).await { Ok(row_val) => break row_val.row_ref, Err(e) => { error!("Error in wait_new_mail: {}", e); @@ -360,7 +362,10 @@ async fn k2v_lock_loop_internal( Some(existing) => existing, None => row_ref.clone(), }; - if let Err(e) = storage.row_insert(vec![storage::RowVal::new(row, lock)]).await { + if let Err(e) = storage + .row_insert(vec![storage::RowVal::new(row, lock)]) + .await + { error!("Could not take lock: {}", e); tokio::time::sleep(Duration::from_secs(30)).await; } @@ -428,14 +433,12 @@ impl EncryptedMessage { let blob_val = storage::BlobVal::new( storage::BlobRef(format!("incoming/{}", gen_ident())), self.encrypted_body.clone().into(), - ).with_meta(MESSAGE_KEY.to_string(), key_header); + ) + .with_meta(MESSAGE_KEY.to_string(), key_header); storage.blob_insert(blob_val).await?; // Update watch key to signal new mail - let watch_val = storage::RowVal::new( - watch_ct.clone(), - gen_ident().0.to_vec(), - ); + let watch_val = storage::RowVal::new(watch_ct.clone(), gen_ident().0.to_vec()); storage.row_insert(vec![watch_val]).await?; Ok(()) } diff --git a/src/mail/mailbox.rs b/src/mail/mailbox.rs index 6fb7dea..e424ba3 100644 --- a/src/mail/mailbox.rs +++ b/src/mail/mailbox.rs @@ -8,7 +8,7 @@ use crate::login::Credentials; use crate::mail::uidindex::*; use crate::mail::unique_ident::*; use crate::mail::IMF; -use crate::storage::{Store, RowRef, RowVal, BlobRef, BlobVal, Selector, self}; +use crate::storage::{self, BlobRef, BlobVal, RowRef, RowVal, Selector, Store}; use crate::timestamp::now_msec; pub struct Mailbox { @@ -196,7 +196,10 @@ impl MailboxInternal { async fn fetch_meta(&self, ids: &[UniqueIdent]) -> Result> { let ids = ids.iter().map(|x| x.to_string()).collect::>(); - let ops = ids.iter().map(|id| RowRef::new(self.mail_path.as_str(), id.as_str())).collect::>(); + let ops = ids + .iter() + .map(|id| RowRef::new(self.mail_path.as_str(), id.as_str())) + .collect::>(); let res_vec = self.storage.row_fetch(&Selector::List(ops)).await?; let mut meta_vec = vec![]; @@ -231,7 +234,10 @@ impl MailboxInternal { } async fn fetch_full(&self, id: UniqueIdent, message_key: &Key) -> Result> { - let obj_res = self.storage.blob_fetch(&BlobRef(format!("{}/{}", self.mail_path, id))).await?; + let obj_res = self + .storage + .blob_fetch(&BlobRef(format!("{}/{}", self.mail_path, id))) + .await?; let body = obj_res.value; cryptoblob::open(&body, message_key) } @@ -266,10 +272,12 @@ impl MailboxInternal { async { // Encrypt and save mail body let message_blob = cryptoblob::seal(mail.raw, &message_key)?; - self.storage.blob_insert(BlobVal::new( - BlobRef(format!("{}/{}", self.mail_path, ident)), - message_blob, - )).await?; + self.storage + .blob_insert(BlobVal::new( + BlobRef(format!("{}/{}", self.mail_path, ident)), + message_blob, + )) + .await?; Ok::<_, anyhow::Error>(()) }, async { @@ -281,10 +289,12 @@ impl MailboxInternal { rfc822_size: mail.raw.len(), }; let meta_blob = seal_serialize(&meta, &self.encryption_key)?; - self.storage.row_insert(vec![RowVal::new( - RowRef::new(&self.mail_path, &ident.to_string()), - meta_blob, - )]).await?; + self.storage + .row_insert(vec![RowVal::new( + RowRef::new(&self.mail_path, &ident.to_string()), + meta_blob, + )]) + .await?; Ok::<_, anyhow::Error>(()) }, self.uid_index.opportunistic_sync() @@ -328,10 +338,12 @@ impl MailboxInternal { rfc822_size: mail.raw.len(), }; let meta_blob = seal_serialize(&meta, &self.encryption_key)?; - self.storage.row_insert(vec![RowVal::new( + self.storage + .row_insert(vec![RowVal::new( RowRef::new(&self.mail_path, &ident.to_string()), meta_blob, - )]).await?; + )]) + .await?; Ok::<_, anyhow::Error>(()) }, self.uid_index.opportunistic_sync() @@ -355,17 +367,25 @@ impl MailboxInternal { futures::try_join!( async { // Delete mail body from S3 - self.storage.blob_rm(&BlobRef(format!("{}/{}", self.mail_path, ident))).await?; + self.storage + .blob_rm(&BlobRef(format!("{}/{}", self.mail_path, ident))) + .await?; Ok::<_, anyhow::Error>(()) }, async { // Delete mail meta from K2V let sk = ident.to_string(); - let res = self.storage - .row_fetch(&storage::Selector::Single(&RowRef::new(&self.mail_path, &sk))) + let res = self + .storage + .row_fetch(&storage::Selector::Single(&RowRef::new( + &self.mail_path, + &sk, + ))) .await?; if let Some(row_val) = res.into_iter().next() { - self.storage.row_rm(&storage::Selector::Single(&row_val.row_ref)).await?; + self.storage + .row_rm(&storage::Selector::Single(&row_val.row_ref)) + .await?; } Ok::<_, anyhow::Error>(()) } @@ -421,10 +441,12 @@ impl MailboxInternal { // Copy mail meta in K2V let meta = &from.fetch_meta(&[source_id]).await?[0]; let meta_blob = seal_serialize(meta, &self.encryption_key)?; - self.storage.row_insert(vec![RowVal::new( - RowRef::new(&self.mail_path, &new_id.to_string()), - meta_blob, - )]).await?; + self.storage + .row_insert(vec![RowVal::new( + RowRef::new(&self.mail_path, &new_id.to_string()), + meta_blob, + )]) + .await?; Ok::<_, anyhow::Error>(()) }, self.uid_index.opportunistic_sync(), diff --git a/src/mail/user.rs b/src/mail/user.rs index 8d12c58..da0d509 100644 --- a/src/mail/user.rs +++ b/src/mail/user.rs @@ -226,7 +226,11 @@ impl User { async fn load_mailbox_list(&self) -> Result<(MailboxList, Option)> { let row_ref = storage::RowRef::new(MAILBOX_LIST_PK, MAILBOX_LIST_SK); - let (mut list, row) = match self.storage.row_fetch(&storage::Selector::Single(&row_ref)).await { + let (mut list, row) = match self + .storage + .row_fetch(&storage::Selector::Single(&row_ref)) + .await + { Err(storage::StorageError::NotFound) => (MailboxList::new(), None), Err(e) => return Err(e.into()), Ok(rv) => { diff --git a/src/main.rs b/src/main.rs index f08f1a3..3221c2e 100644 --- a/src/main.rs +++ b/src/main.rs @@ -1,6 +1,5 @@ #![feature(async_fn_in_trait)] -mod timestamp; mod bayou; mod config; mod cryptoblob; @@ -11,17 +10,18 @@ mod login; mod mail; mod server; mod storage; +mod timestamp; -use std::path::PathBuf; use std::io::Read; +use std::path::PathBuf; -use anyhow::{bail, Result, Context}; +use anyhow::{bail, Context, Result}; use clap::{Parser, Subcommand}; -use nix::{unistd::Pid, sys::signal}; +use nix::{sys::signal, unistd::Pid}; use config::*; -use server::Server; use login::{static_provider::*, *}; +use server::Server; #[derive(Parser, Debug)] #[clap(author, version, about, long_about = None)] @@ -58,7 +58,7 @@ enum ToolsCommand { PasswordHash { #[clap(env = "AEROGRAMME_PASSWORD")] maybe_password: Option, - } + }, } #[derive(Subcommand, Debug)] @@ -138,7 +138,7 @@ enum AccountManagement { maybe_new_password: Option, #[clap(short, long)] - login: String + login: String, }, } @@ -165,11 +165,11 @@ async fn main() -> Result<()> { CompanionCommand::Daemon => { let server = Server::from_companion_config(config).await?; server.run().await?; - }, + } CompanionCommand::Reload { pid } => reload(*pid, config.pid)?, CompanionCommand::Wizard => { unimplemented!(); - }, + } CompanionCommand::Account(cmd) => { let user_file = config.users.user_list; account_management(&args.command, cmd, user_file)?; @@ -179,22 +179,24 @@ async fn main() -> Result<()> { ProviderCommand::Daemon => { let server = Server::from_provider_config(config).await?; server.run().await?; - }, + } ProviderCommand::Reload { pid } => reload(*pid, config.pid)?, ProviderCommand::Account(cmd) => { let user_file = match config.users { UserManagement::Static(conf) => conf.user_list, - UserManagement::Ldap(_) => panic!("LDAP account management is not supported from Aerogramme.") + UserManagement::Ldap(_) => { + panic!("LDAP account management is not supported from Aerogramme.") + } }; account_management(&args.command, cmd, user_file)?; } }, (Command::Provider(_), AnyConfig::Companion(_)) => { bail!("Your want to run a 'Provider' command but your configuration file has role 'Companion'."); - }, + } (Command::Companion(_), AnyConfig::Provider(_)) => { bail!("Your want to run a 'Companion' command but your configuration file has role 'Provider'."); - }, + } (Command::Tools(subcommand), _) => match subcommand { ToolsCommand::PasswordHash { maybe_password } => { let password = match maybe_password { @@ -202,60 +204,64 @@ async fn main() -> Result<()> { None => rpassword::prompt_password("Enter password: ")?, }; println!("{}", hash_password(&password)?); - }, - ToolsCommand::CryptoRoot(crcommand) => { - match crcommand { - CryptoRootCommand::New { maybe_password } => { - let password = match maybe_password { - Some(pwd) => pwd.clone(), - None => { - let password = rpassword::prompt_password("Enter password: ")?; - let password_confirm = rpassword::prompt_password("Confirm password: ")?; - if password != password_confirm { - bail!("Passwords don't match."); - } - password + } + ToolsCommand::CryptoRoot(crcommand) => match crcommand { + CryptoRootCommand::New { maybe_password } => { + let password = match maybe_password { + Some(pwd) => pwd.clone(), + None => { + let password = rpassword::prompt_password("Enter password: ")?; + let password_confirm = + rpassword::prompt_password("Confirm password: ")?; + if password != password_confirm { + bail!("Passwords don't match."); } - }; - let crypto_keys = CryptoKeys::init(); - let cr = CryptoRoot::create_pass(&password, &crypto_keys)?; - println!("{}", cr.0); - }, - CryptoRootCommand::NewClearText => { - let crypto_keys = CryptoKeys::init(); - let cr = CryptoRoot::create_cleartext(&crypto_keys); - println!("{}", cr.0); - }, - CryptoRootCommand::ChangePassword { maybe_old_password, maybe_new_password, crypto_root } => { - let old_password = match maybe_old_password { - Some(pwd) => pwd.to_string(), - None => rpassword::prompt_password("Enter old password: ")?, - }; + password + } + }; + let crypto_keys = CryptoKeys::init(); + let cr = CryptoRoot::create_pass(&password, &crypto_keys)?; + println!("{}", cr.0); + } + CryptoRootCommand::NewClearText => { + let crypto_keys = CryptoKeys::init(); + let cr = CryptoRoot::create_cleartext(&crypto_keys); + println!("{}", cr.0); + } + CryptoRootCommand::ChangePassword { + maybe_old_password, + maybe_new_password, + crypto_root, + } => { + let old_password = match maybe_old_password { + Some(pwd) => pwd.to_string(), + None => rpassword::prompt_password("Enter old password: ")?, + }; - let new_password = match maybe_new_password { - Some(pwd) => pwd.to_string(), - None => { - let password = rpassword::prompt_password("Enter new password: ")?; - let password_confirm = rpassword::prompt_password("Confirm new password: ")?; - if password != password_confirm { - bail!("Passwords don't match."); - } - password + let new_password = match maybe_new_password { + Some(pwd) => pwd.to_string(), + None => { + let password = rpassword::prompt_password("Enter new password: ")?; + let password_confirm = + rpassword::prompt_password("Confirm new password: ")?; + if password != password_confirm { + bail!("Passwords don't match."); } - }; + password + } + }; - let keys = CryptoRoot(crypto_root.to_string()).crypto_keys(&old_password)?; - let cr = CryptoRoot::create_pass(&new_password, &keys)?; - println!("{}", cr.0); - }, - CryptoRootCommand::DeriveIncoming { crypto_root } => { - let pubkey = CryptoRoot(crypto_root.to_string()).public_key()?; - let cr = CryptoRoot::create_incoming(&pubkey); - println!("{}", cr.0); - }, + let keys = CryptoRoot(crypto_root.to_string()).crypto_keys(&old_password)?; + let cr = CryptoRoot::create_pass(&new_password, &keys)?; + println!("{}", cr.0); + } + CryptoRootCommand::DeriveIncoming { crypto_root } => { + let pubkey = CryptoRoot(crypto_root.to_string()).public_key()?; + let cr = CryptoRoot::create_incoming(&pubkey); + println!("{}", cr.0); } }, - } + }, } Ok(()) @@ -264,12 +270,12 @@ async fn main() -> Result<()> { fn reload(pid: Option, pid_path: Option) -> Result<()> { let final_pid = match (pid, pid_path) { (Some(pid), _) => pid, - (_, Some(path)) => { + (_, Some(path)) => { let mut f = std::fs::OpenOptions::new().read(true).open(path)?; let mut pidstr = String::new(); f.read_to_string(&mut pidstr)?; pidstr.parse::()? - }, + } _ => bail!("Unable to infer your daemon's PID"), }; let pid = Pid::from_raw(final_pid); @@ -278,13 +284,15 @@ fn reload(pid: Option, pid_path: Option) -> Result<()> { } fn account_management(root: &Command, cmd: &AccountManagement, users: PathBuf) -> Result<()> { - let mut ulist: UserList = read_config(users.clone()).context(format!("'{:?}' must be a user database", users))?; + let mut ulist: UserList = + read_config(users.clone()).context(format!("'{:?}' must be a user database", users))?; match cmd { AccountManagement::Add { login, setup } => { - tracing::debug!(user=login, "will-create"); - let stp: SetupEntry = read_config(setup.clone()).context(format!("'{:?}' must be a setup file", setup))?; - tracing::debug!(user=login, "loaded setup entry"); + tracing::debug!(user = login, "will-create"); + let stp: SetupEntry = read_config(setup.clone()) + .context(format!("'{:?}' must be a setup file", setup))?; + tracing::debug!(user = login, "loaded setup entry"); let password = match stp.clear_password { Some(pwd) => pwd, @@ -307,21 +315,28 @@ fn account_management(root: &Command, cmd: &AccountManagement, users: PathBuf) - let hash = hash_password(password.as_str()).context("unable to hash password")?; - ulist.insert(login.clone(), UserEntry { - email_addresses: stp.email_addresses, - password: hash, - crypto_root: crypto_root.0, - storage: stp.storage, - }); + ulist.insert( + login.clone(), + UserEntry { + email_addresses: stp.email_addresses, + password: hash, + crypto_root: crypto_root.0, + storage: stp.storage, + }, + ); write_config(users.clone(), &ulist)?; - }, + } AccountManagement::Delete { login } => { - tracing::debug!(user=login, "will-delete"); + tracing::debug!(user = login, "will-delete"); ulist.remove(login); write_config(users.clone(), &ulist)?; - }, - AccountManagement::ChangePassword { maybe_old_password, maybe_new_password, login } => { + } + AccountManagement::ChangePassword { + maybe_old_password, + maybe_new_password, + login, + } => { let mut user = ulist.remove(login).context("user must exist first")?; let old_password = match maybe_old_password { @@ -345,16 +360,16 @@ fn account_management(root: &Command, cmd: &AccountManagement, users: PathBuf) - } password } - }; + }; let new_hash = hash_password(&new_password)?; let new_crypto_root = CryptoRoot::create_pass(&new_password, &crypto_keys)?; - + user.password = new_hash; user.crypto_root = new_crypto_root.0; ulist.insert(login.clone(), user); write_config(users.clone(), &ulist)?; - }, + } }; Ok(()) diff --git a/src/server.rs b/src/server.rs index 552a0e6..28e0b27 100644 --- a/src/server.rs +++ b/src/server.rs @@ -1,6 +1,6 @@ -use std::sync::Arc; -use std::path::PathBuf; use std::io::Write; +use std::path::PathBuf; +use std::sync::Arc; use anyhow::Result; use futures::try_join; @@ -26,7 +26,11 @@ impl Server { let lmtp_server = None; let imap_server = Some(imap::new(config.imap, login.clone()).await?); - Ok(Self { lmtp_server, imap_server, pid_file: config.pid }) + Ok(Self { + lmtp_server, + imap_server, + pid_file: config.pid, + }) } pub async fn from_provider_config(config: ProviderConfig) -> Result { @@ -39,12 +43,16 @@ impl Server { let lmtp_server = Some(LmtpServer::new(config.lmtp, login.clone())); let imap_server = Some(imap::new(config.imap, login.clone()).await?); - Ok(Self { lmtp_server, imap_server, pid_file: config.pid }) + Ok(Self { + lmtp_server, + imap_server, + pid_file: config.pid, + }) } pub async fn run(self) -> Result<()> { let pid = std::process::id(); - tracing::info!(pid=pid, "Starting main loops"); + tracing::info!(pid = pid, "Starting main loops"); // write the pid file if let Some(pid_file) = self.pid_file { @@ -57,7 +65,6 @@ impl Server { drop(file); } - let (exit_signal, provoke_exit) = watch_ctrl_c(); let _exit_on_err = move |err: anyhow::Error| { error!("Error: {}", err); diff --git a/src/storage/garage.rs b/src/storage/garage.rs index f9ba756..d08585f 100644 --- a/src/storage/garage.rs +++ b/src/storage/garage.rs @@ -1,10 +1,6 @@ use crate::storage::*; +use aws_sdk_s3::{self as s3, error::SdkError, operation::get_object::GetObjectError}; use serde::Serialize; -use aws_sdk_s3::{ - self as s3, - error::SdkError, - operation::get_object::GetObjectError, -}; #[derive(Clone, Debug, Serialize)] pub struct GarageConf { @@ -28,18 +24,18 @@ impl GarageBuilder { unicity.extend_from_slice(file!().as_bytes()); unicity.append(&mut rmp_serde::to_vec(&conf)?); Ok(Arc::new(Self { conf, unicity })) - } + } } #[async_trait] impl IBuilder for GarageBuilder { async fn build(&self) -> Result { let s3_creds = s3::config::Credentials::new( - self.conf.aws_access_key_id.clone(), - self.conf.aws_secret_access_key.clone(), - None, - None, - "aerogramme" + self.conf.aws_access_key_id.clone(), + self.conf.aws_secret_access_key.clone(), + None, + None, + "aerogramme", ); let s3_config = aws_config::from_env() @@ -51,12 +47,12 @@ impl IBuilder for GarageBuilder { let s3_client = aws_sdk_s3::Client::new(&s3_config); let k2v_config = k2v_client::K2vClientConfig { - endpoint: self.conf.k2v_endpoint.clone(), - region: self.conf.region.clone(), - aws_access_key_id: self.conf.aws_access_key_id.clone(), - aws_secret_access_key: self.conf.aws_secret_access_key.clone(), - bucket: self.conf.bucket.clone(), - user_agent: None, + endpoint: self.conf.k2v_endpoint.clone(), + region: self.conf.region.clone(), + aws_access_key_id: self.conf.aws_access_key_id.clone(), + aws_secret_access_key: self.conf.aws_secret_access_key.clone(), + bucket: self.conf.bucket.clone(), + user_agent: None, }; let k2v_client = match k2v_client::K2vClient::new(k2v_config) { @@ -67,7 +63,7 @@ impl IBuilder for GarageBuilder { Ok(v) => v, }; - Ok(Box::new(GarageStore { + Ok(Box::new(GarageStore { bucket: self.conf.bucket.clone(), s3: s3_client, k2v: k2v_client, @@ -86,19 +82,30 @@ pub struct GarageStore { fn causal_to_row_val(row_ref: RowRef, causal_value: k2v_client::CausalValue) -> RowVal { let new_row_ref = row_ref.with_causality(causal_value.causality.into()); - let row_values = causal_value.value.into_iter().map(|k2v_value| match k2v_value { - k2v_client::K2vValue::Tombstone => Alternative::Tombstone, - k2v_client::K2vValue::Value(v) => Alternative::Value(v), - }).collect::>(); + let row_values = causal_value + .value + .into_iter() + .map(|k2v_value| match k2v_value { + k2v_client::K2vValue::Tombstone => Alternative::Tombstone, + k2v_client::K2vValue::Value(v) => Alternative::Value(v), + }) + .collect::>(); - RowVal { row_ref: new_row_ref, value: row_values } + RowVal { + row_ref: new_row_ref, + value: row_values, + } } #[async_trait] impl IStore for GarageStore { async fn row_fetch<'a>(&self, select: &Selector<'a>) -> Result, StorageError> { let (pk_list, batch_op) = match select { - Selector::Range { shard, sort_begin, sort_end } => ( + Selector::Range { + shard, + sort_begin, + sort_end, + } => ( vec![shard.to_string()], vec![k2v_client::BatchReadOp { partition_key: shard, @@ -108,49 +115,71 @@ impl IStore for GarageStore { ..k2v_client::Filter::default() }, ..k2v_client::BatchReadOp::default() - }] + }], ), Selector::List(row_ref_list) => ( - row_ref_list.iter().map(|row_ref| row_ref.uid.shard.to_string()).collect::>(), - row_ref_list.iter().map(|row_ref| k2v_client::BatchReadOp { - partition_key: &row_ref.uid.shard, - filter: k2v_client::Filter { - start: Some(&row_ref.uid.sort), - ..k2v_client::Filter::default() - }, - single_item: true, - ..k2v_client::BatchReadOp::default() - }).collect::>() + row_ref_list + .iter() + .map(|row_ref| row_ref.uid.shard.to_string()) + .collect::>(), + row_ref_list + .iter() + .map(|row_ref| k2v_client::BatchReadOp { + partition_key: &row_ref.uid.shard, + filter: k2v_client::Filter { + start: Some(&row_ref.uid.sort), + ..k2v_client::Filter::default() + }, + single_item: true, + ..k2v_client::BatchReadOp::default() + }) + .collect::>(), ), Selector::Prefix { shard, sort_prefix } => ( vec![shard.to_string()], vec![k2v_client::BatchReadOp { - partition_key: shard, - filter: k2v_client::Filter { - prefix: Some(sort_prefix), - ..k2v_client::Filter::default() - }, - ..k2v_client::BatchReadOp::default() - }]), - Selector::Single(row_ref) => { - let causal_value = match self.k2v.read_item(&row_ref.uid.shard, &row_ref.uid.sort).await { - Err(e) => { - tracing::error!("K2V read item shard={}, sort={}, bucket={} failed: {}", row_ref.uid.shard, row_ref.uid.sort, self.bucket, e); - return Err(StorageError::Internal); + partition_key: shard, + filter: k2v_client::Filter { + prefix: Some(sort_prefix), + ..k2v_client::Filter::default() }, + ..k2v_client::BatchReadOp::default() + }], + ), + Selector::Single(row_ref) => { + let causal_value = match self + .k2v + .read_item(&row_ref.uid.shard, &row_ref.uid.sort) + .await + { + Err(e) => { + tracing::error!( + "K2V read item shard={}, sort={}, bucket={} failed: {}", + row_ref.uid.shard, + row_ref.uid.sort, + self.bucket, + e + ); + return Err(StorageError::Internal); + } Ok(v) => v, }; let row_val = causal_to_row_val((*row_ref).clone(), causal_value); - return Ok(vec![row_val]) - }, + return Ok(vec![row_val]); + } }; let all_raw_res = match self.k2v.read_batch(&batch_op).await { Err(e) => { - tracing::error!("k2v read batch failed for {:?}, bucket {} with err: {}", select, self.bucket, e); + tracing::error!( + "k2v read batch failed for {:?}, bucket {} with err: {}", + select, + self.bucket, + e + ); return Err(StorageError::Internal); - }, + } Ok(v) => v, }; @@ -163,13 +192,17 @@ impl IStore for GarageStore { .into_iter() .zip(pk_list.into_iter()) .map(|((sk, cv), pk)| causal_to_row_val(RowRef::new(&pk, &sk), cv)) - .collect::>(); + .collect::>(); Ok(row_vals) } async fn row_rm<'a>(&self, select: &Selector<'a>) -> Result<(), StorageError> { let del_op = match select { - Selector::Range { shard, sort_begin, sort_end } => vec![k2v_client::BatchDeleteOp { + Selector::Range { + shard, + sort_begin, + sort_end, + } => vec![k2v_client::BatchDeleteOp { partition_key: shard, prefix: None, start: Some(sort_begin), @@ -178,21 +211,24 @@ impl IStore for GarageStore { }], Selector::List(row_ref_list) => { // Insert null values with causality token = delete - let batch_op = row_ref_list.iter().map(|v| k2v_client::BatchInsertOp { - partition_key: &v.uid.shard, - sort_key: &v.uid.sort, - causality: v.causality.clone().map(|ct| ct.into()), - value: k2v_client::K2vValue::Tombstone, - }).collect::>(); - + let batch_op = row_ref_list + .iter() + .map(|v| k2v_client::BatchInsertOp { + partition_key: &v.uid.shard, + sort_key: &v.uid.sort, + causality: v.causality.clone().map(|ct| ct.into()), + value: k2v_client::K2vValue::Tombstone, + }) + .collect::>(); + return match self.k2v.insert_batch(&batch_op).await { Err(e) => { tracing::error!("Unable to delete the list of values: {}", e); Err(StorageError::Internal) - }, + } Ok(_) => Ok(()), }; - }, + } Selector::Prefix { shard, sort_prefix } => vec![k2v_client::BatchDeleteOp { partition_key: shard, prefix: Some(sort_prefix), @@ -208,15 +244,15 @@ impl IStore for GarageStore { causality: row_ref.causality.clone().map(|ct| ct.into()), value: k2v_client::K2vValue::Tombstone, }]; - + return match self.k2v.insert_batch(&batch_op).await { Err(e) => { tracing::error!("Unable to delete the list of values: {}", e); Err(StorageError::Internal) - }, + } Ok(_) => Ok(()), }; - }, + } }; // Finally here we only have prefix & range @@ -224,34 +260,46 @@ impl IStore for GarageStore { Err(e) => { tracing::error!("delete batch error: {}", e); Err(StorageError::Internal) - }, + } Ok(_) => Ok(()), } } async fn row_insert(&self, values: Vec) -> Result<(), StorageError> { - let batch_ops = values.iter().map(|v| k2v_client::BatchInsertOp { - partition_key: &v.row_ref.uid.shard, - sort_key: &v.row_ref.uid.sort, - causality: v.row_ref.causality.clone().map(|ct| ct.into()), - value: v.value.iter().next().map(|cv| match cv { - Alternative::Value(buff) => k2v_client::K2vValue::Value(buff.clone()), - Alternative::Tombstone => k2v_client::K2vValue::Tombstone, - }).unwrap_or(k2v_client::K2vValue::Tombstone) - }).collect::>(); + let batch_ops = values + .iter() + .map(|v| k2v_client::BatchInsertOp { + partition_key: &v.row_ref.uid.shard, + sort_key: &v.row_ref.uid.sort, + causality: v.row_ref.causality.clone().map(|ct| ct.into()), + value: v + .value + .iter() + .next() + .map(|cv| match cv { + Alternative::Value(buff) => k2v_client::K2vValue::Value(buff.clone()), + Alternative::Tombstone => k2v_client::K2vValue::Tombstone, + }) + .unwrap_or(k2v_client::K2vValue::Tombstone), + }) + .collect::>(); match self.k2v.insert_batch(&batch_ops).await { Err(e) => { tracing::error!("k2v can't insert some value: {}", e); Err(StorageError::Internal) - }, + } Ok(v) => Ok(v), } } async fn row_poll(&self, value: &RowRef) -> Result { loop { if let Some(ct) = &value.causality { - match self.k2v.poll_item(&value.uid.shard, &value.uid.sort, ct.clone().into(), None).await { + match self + .k2v + .poll_item(&value.uid.shard, &value.uid.sort, ct.clone().into(), None) + .await + { Err(e) => { tracing::error!("Unable to poll item: {}", e); return Err(StorageError::Internal); @@ -262,8 +310,7 @@ impl IStore for GarageStore { } else { match self.k2v.read_item(&value.uid.shard, &value.uid.sort).await { Err(k2v_client::Error::NotFound) => { - self - .k2v + self.k2v .insert_item(&value.uid.shard, &value.uid.sort, vec![0u8], None) .await .map_err(|e| { @@ -273,8 +320,8 @@ impl IStore for GarageStore { } Err(e) => { tracing::error!("Unable to read item in polling logic: {}", e); - return Err(StorageError::Internal) - }, + return Err(StorageError::Internal); + } Ok(cv) => return Ok(causal_to_row_val(value.clone(), cv)), } } @@ -282,7 +329,8 @@ impl IStore for GarageStore { } async fn blob_fetch(&self, blob_ref: &BlobRef) -> Result { - let maybe_out = self.s3 + let maybe_out = self + .s3 .get_object() .bucket(self.bucket.to_string()) .key(blob_ref.0.to_string()) @@ -296,12 +344,12 @@ impl IStore for GarageStore { e => { tracing::warn!("Blob Fetch Error, Service Error: {}", e); return Err(StorageError::Internal); - }, + } }, Err(e) => { tracing::warn!("Blob Fetch Error, {}", e); return Err(StorageError::Internal); - }, + } }; let buffer = match object_output.body.collect().await { @@ -316,9 +364,10 @@ impl IStore for GarageStore { Ok(BlobVal::new(blob_ref.clone(), buffer)) } async fn blob_insert(&self, blob_val: BlobVal) -> Result<(), StorageError> { - let streamable_value = s3::primitives::ByteStream::from(blob_val.value); + let streamable_value = s3::primitives::ByteStream::from(blob_val.value); - let maybe_send = self.s3 + let maybe_send = self + .s3 .put_object() .bucket(self.bucket.to_string()) .key(blob_val.blob_ref.0.to_string()) @@ -338,7 +387,8 @@ impl IStore for GarageStore { } } async fn blob_copy(&self, src: &BlobRef, dst: &BlobRef) -> Result<(), StorageError> { - let maybe_copy = self.s3 + let maybe_copy = self + .s3 .copy_object() .bucket(self.bucket.to_string()) .key(dst.0.clone()) @@ -348,18 +398,24 @@ impl IStore for GarageStore { match maybe_copy { Err(e) => { - tracing::error!("unable to copy object {} to {} (bucket: {}), error: {}", src.0, dst.0, self.bucket, e); + tracing::error!( + "unable to copy object {} to {} (bucket: {}), error: {}", + src.0, + dst.0, + self.bucket, + e + ); Err(StorageError::Internal) - }, + } Ok(_) => { tracing::debug!("copied {} to {} (bucket: {})", src.0, dst.0, self.bucket); Ok(()) } } - } async fn blob_list(&self, prefix: &str) -> Result, StorageError> { - let maybe_list = self.s3 + let maybe_list = self + .s3 .list_objects_v2() .bucket(self.bucket.to_string()) .prefix(prefix) @@ -370,7 +426,12 @@ impl IStore for GarageStore { match maybe_list { Err(e) => { - tracing::error!("listing prefix {} on bucket {} failed: {}", prefix, self.bucket, e); + tracing::error!( + "listing prefix {} on bucket {} failed: {}", + prefix, + self.bucket, + e + ); Err(StorageError::Internal) } Ok(pagin_list_out) => Ok(pagin_list_out @@ -382,7 +443,8 @@ impl IStore for GarageStore { } } async fn blob_rm(&self, blob_ref: &BlobRef) -> Result<(), StorageError> { - let maybe_delete = self.s3 + let maybe_delete = self + .s3 .delete_object() .bucket(self.bucket.to_string()) .key(blob_ref.0.clone()) @@ -391,9 +453,14 @@ impl IStore for GarageStore { match maybe_delete { Err(e) => { - tracing::error!("unable to delete {} (bucket: {}), error {}", blob_ref.0, self.bucket, e); + tracing::error!( + "unable to delete {} (bucket: {}), error {}", + blob_ref.0, + self.bucket, + e + ); Err(StorageError::Internal) - }, + } Ok(_) => { tracing::debug!("deleted {} (bucket: {})", blob_ref.0, self.bucket); Ok(()) @@ -401,4 +468,3 @@ impl IStore for GarageStore { } } } - diff --git a/src/storage/in_memory.rs b/src/storage/in_memory.rs index ee7c9a6..3c3a94c 100644 --- a/src/storage/in_memory.rs +++ b/src/storage/in_memory.rs @@ -1,6 +1,6 @@ use crate::storage::*; -use std::collections::{HashMap, BTreeMap}; -use std::ops::Bound::{Included, Unbounded, Excluded, self}; +use std::collections::{BTreeMap, HashMap}; +use std::ops::Bound::{self, Excluded, Included, Unbounded}; use std::sync::{Arc, RwLock}; use tokio::sync::Notify; @@ -16,7 +16,7 @@ impl MemDb { Self(tokio::sync::Mutex::new(HashMap::new())) } - pub async fn builder(&self, username: &str) -> Arc { + pub async fn builder(&self, username: &str) -> Arc { let mut global_storage = self.0.lock().await; global_storage .entry(username.to_string()) @@ -60,8 +60,8 @@ impl InternalRowVal { } fn to_row_val(&self, row_ref: RowRef) -> RowVal { - RowVal{ - row_ref: row_ref.with_causality(self.version.to_string()), + RowVal { + row_ref: row_ref.with_causality(self.version.to_string()), value: self.concurrent_values(), } } @@ -75,7 +75,7 @@ struct InternalBlobVal { impl InternalBlobVal { fn to_blob_val(&self, bref: &BlobRef) -> BlobVal { BlobVal { - blob_ref: bref.clone(), + blob_ref: bref.clone(), meta: self.metadata.clone(), value: self.data.clone(), } @@ -113,7 +113,7 @@ impl IBuilder for MemBuilder { row: self.row.clone(), blob: self.blob.clone(), })) - } + } fn unique(&self) -> UnicityBuffer { UnicityBuffer(self.unicity.clone()) @@ -170,24 +170,32 @@ impl IStore for MemStore { let store = self.row.read().or(Err(StorageError::Internal))?; match select { - Selector::Range { shard, sort_begin, sort_end } => { - Ok(store - .get(*shard) - .unwrap_or(&BTreeMap::new()) - .range((Included(sort_begin.to_string()), Excluded(sort_end.to_string()))) - .map(|(k, v)| v.to_row_val(RowRef::new(shard, k))) - .collect::>()) - }, + Selector::Range { + shard, + sort_begin, + sort_end, + } => Ok(store + .get(*shard) + .unwrap_or(&BTreeMap::new()) + .range(( + Included(sort_begin.to_string()), + Excluded(sort_end.to_string()), + )) + .map(|(k, v)| v.to_row_val(RowRef::new(shard, k))) + .collect::>()), Selector::List(rlist) => { let mut acc = vec![]; for row_ref in rlist { - let maybe_intval = store.get(&row_ref.uid.shard).map(|v| v.get(&row_ref.uid.sort)).flatten(); + let maybe_intval = store + .get(&row_ref.uid.shard) + .map(|v| v.get(&row_ref.uid.sort)) + .flatten(); if let Some(intval) = maybe_intval { acc.push(intval.to_row_val(row_ref.clone())); } } Ok(acc) - }, + } Selector::Prefix { shard, sort_prefix } => { let last_bound = prefix_last_bound(sort_prefix); @@ -197,13 +205,13 @@ impl IStore for MemStore { .range((Included(sort_prefix.to_string()), last_bound)) .map(|(k, v)| v.to_row_val(RowRef::new(shard, k))) .collect::>()) - }, + } Selector::Single(row_ref) => { let intval = store - .get(&row_ref.uid.shard) - .ok_or(StorageError::NotFound)? - .get(&row_ref.uid.sort) - .ok_or(StorageError::NotFound)?; + .get(&row_ref.uid.shard) + .ok_or(StorageError::NotFound)? + .get(&row_ref.uid.sort) + .ok_or(StorageError::NotFound)?; Ok(vec![intval.to_row_val((*row_ref).clone())]) } } @@ -213,7 +221,12 @@ impl IStore for MemStore { tracing::trace!(select=%select, command="row_rm"); let values = match select { - Selector::Range { .. } | Selector::Prefix { .. } => self.row_fetch(select).await?.into_iter().map(|rv| rv.row_ref).collect::>(), + Selector::Range { .. } | Selector::Prefix { .. } => self + .row_fetch(select) + .await? + .into_iter() + .map(|rv| rv.row_ref) + .collect::>(), Selector::List(rlist) => rlist.clone(), Selector::Single(row_ref) => vec![(*row_ref).clone()], }; @@ -282,7 +295,10 @@ impl IStore for MemStore { async fn blob_fetch(&self, blob_ref: &BlobRef) -> Result { tracing::trace!(entry=%blob_ref, command="blob_fetch"); let store = self.blob.read().or(Err(StorageError::Internal))?; - store.get(&blob_ref.0).ok_or(StorageError::NotFound).map(|v| v.to_blob_val(blob_ref)) + store + .get(&blob_ref.0) + .ok_or(StorageError::NotFound) + .map(|v| v.to_blob_val(blob_ref)) } async fn blob_insert(&self, blob_val: BlobVal) -> Result<(), StorageError> { tracing::trace!(entry=%blob_val.blob_ref, command="blob_insert"); @@ -300,10 +316,13 @@ impl IStore for MemStore { Ok(()) } async fn blob_list(&self, prefix: &str) -> Result, StorageError> { - tracing::trace!(prefix=prefix, command="blob_list"); + tracing::trace!(prefix = prefix, command = "blob_list"); let store = self.blob.read().or(Err(StorageError::Internal))?; let last_bound = prefix_last_bound(prefix); - let blist = store.range((Included(prefix.to_string()), last_bound)).map(|(k, _)| BlobRef(k.to_string())).collect::>(); + let blist = store + .range((Included(prefix.to_string()), last_bound)) + .map(|(k, _)| BlobRef(k.to_string())) + .collect::>(); Ok(blist) } async fn blob_rm(&self, blob_ref: &BlobRef) -> Result<(), StorageError> { diff --git a/src/storage/mod.rs b/src/storage/mod.rs index c81ffe4..1f86f71 100644 --- a/src/storage/mod.rs +++ b/src/storage/mod.rs @@ -8,13 +8,13 @@ * into the object system so it is not exposed. */ -pub mod in_memory; pub mod garage; +pub mod in_memory; -use std::sync::Arc; -use std::hash::Hash; -use std::collections::HashMap; use async_trait::async_trait; +use std::collections::HashMap; +use std::hash::Hash; +use std::sync::Arc; #[derive(Debug, Clone)] pub enum Alternative { @@ -52,14 +52,18 @@ pub struct RowRef { } impl std::fmt::Display for RowRef { fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { - write!(f, "RowRef({}, {}, {:?})", self.uid.shard, self.uid.sort, self.causality) + write!( + f, + "RowRef({}, {}, {:?})", + self.uid.shard, self.uid.sort, self.causality + ) } } impl RowRef { pub fn new(shard: &str, sort: &str) -> Self { Self { - uid: RowUid { + uid: RowUid { shard: shard.to_string(), sort: sort.to_string(), }, @@ -87,7 +91,6 @@ impl RowVal { } } - #[derive(Debug, Clone)] pub struct BlobRef(pub String); impl std::fmt::Display for BlobRef { @@ -105,7 +108,8 @@ pub struct BlobVal { impl BlobVal { pub fn new(blob_ref: BlobRef, value: Vec) -> Self { Self { - blob_ref, value, + blob_ref, + value, meta: HashMap::new(), } } @@ -118,16 +122,27 @@ impl BlobVal { #[derive(Debug)] pub enum Selector<'a> { - Range { shard: &'a str, sort_begin: &'a str, sort_end: &'a str }, - List (Vec), // list of (shard_key, sort_key) + Range { + shard: &'a str, + sort_begin: &'a str, + sort_end: &'a str, + }, + List(Vec), // list of (shard_key, sort_key) #[allow(dead_code)] - Prefix { shard: &'a str, sort_prefix: &'a str }, + Prefix { + shard: &'a str, + sort_prefix: &'a str, + }, Single(&'a RowRef), } impl<'a> std::fmt::Display for Selector<'a> { fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { match self { - Self::Range { shard, sort_begin, sort_end } => write!(f, "Range({}, [{}, {}[)", shard, sort_begin, sort_end), + Self::Range { + shard, + sort_begin, + sort_end, + } => write!(f, "Range({}, [{}, {}[)", shard, sort_begin, sort_end), Self::List(list) => write!(f, "List({:?})", list), Self::Prefix { shard, sort_prefix } => write!(f, "Prefix({}, {})", shard, sort_prefix), Self::Single(row_ref) => write!(f, "Single({})", row_ref), @@ -149,7 +164,7 @@ pub trait IStore { async fn blob_rm(&self, blob_ref: &BlobRef) -> Result<(), StorageError>; } -#[derive(Clone,Debug,PartialEq,Eq,Hash)] +#[derive(Clone, Debug, PartialEq, Eq, Hash)] pub struct UnicityBuffer(Vec); #[async_trait]