From e59c23a69df116737c428ccbfbe4dfeff4d956d5 Mon Sep 17 00:00:00 2001 From: Alex Auvolat Date: Mon, 3 Jan 2022 17:22:40 +0100 Subject: [PATCH] Refactor logic for setting/unsetting aliases --- src/garage/admin.rs | 482 +++++++++----------------------- src/model/bucket_alias_table.rs | 9 +- src/model/helper/bucket.rs | 386 ++++++++++++++++++++++++- src/model/migrate.rs | 91 +++--- src/model/permission.rs | 11 + 5 files changed, 568 insertions(+), 411 deletions(-) diff --git a/src/garage/admin.rs b/src/garage/admin.rs index 9328f46e..0c1e58f8 100644 --- a/src/garage/admin.rs +++ b/src/garage/admin.rs @@ -7,7 +7,7 @@ use serde::{Deserialize, Serialize}; use garage_util::crdt::*; use garage_util::data::*; -use garage_util::error::{Error as GarageError, OkOrMessage}; +use garage_util::error::Error as GarageError; use garage_util::time::*; use garage_table::replication::*; @@ -28,8 +28,6 @@ use crate::repair::Repair; pub const ADMIN_RPC_PATH: &str = "garage/admin_rpc.rs/Rpc"; -macro_rules! INVALID_BUCKET_NAME_MESSAGE { () => { "Invalid bucket name: {}. See AWS documentation for constraints on S3 bucket names:\nhttps://docs.aws.amazon.com/AmazonS3/latest/userguide/bucketnamingrules.html" }; } - #[derive(Debug, Serialize, Deserialize)] pub enum AdminRpc { BucketOperation(BucketOperation), @@ -134,47 +132,54 @@ impl AdminRpcHandler { #[allow(clippy::ptr_arg)] async fn handle_create_bucket(&self, name: &String) -> Result { - let mut bucket = Bucket::new(); - let alias = match self.garage.bucket_alias_table.get(&EmptyKey, name).await? { - Some(mut alias) => { - if !alias.state.get().is_deleted() { - return Err(Error::BadRequest(format!("Bucket {} already exists", name))); - } - alias.state.update(Deletable::Present(AliasParams { - bucket_id: bucket.id, - })); - alias + if !is_valid_bucket_name(name) { + return Err(Error::BadRequest(format!( + "{}: {}", + name, INVALID_BUCKET_NAME_MESSAGE + ))); + } + + if let Some(alias) = self.garage.bucket_alias_table.get(&EmptyKey, name).await? { + if !alias.state.get().is_deleted() { + return Err(Error::BadRequest(format!("Bucket {} already exists", name))); } - None => BucketAlias::new(name.clone(), bucket.id) - .ok_or_bad_request(format!(INVALID_BUCKET_NAME_MESSAGE!(), name))?, - }; - bucket.state.as_option_mut().unwrap().aliases.merge_raw( - name, - alias.state.timestamp(), - &true, - ); + } + + // ---- done checking, now commit ---- + + let bucket = Bucket::new(); self.garage.bucket_table.insert(&bucket).await?; - self.garage.bucket_alias_table.insert(&alias).await?; + + self.garage + .bucket_helper() + .set_global_bucket_alias(bucket.id, name) + .await?; + Ok(AdminRpc::Ok(format!("Bucket {} was created.", name))) } async fn handle_delete_bucket(&self, query: &DeleteBucketOpt) -> Result { - let mut bucket_alias = self + let helper = self.garage.bucket_helper(); + + let bucket_id = helper + .resolve_global_bucket_name(&query.name) + .await? + .ok_or_bad_request("Bucket not found")?; + + // Get the alias, but keep in minde here the bucket name + // given in parameter can also be directly the bucket's ID. + // In that case bucket_alias will be None, and + // we can still delete the bucket if it has zero aliases + // (a condition which we try to prevent but that could still happen somehow). + // We just won't try to delete an alias entry because there isn't one. + let bucket_alias = self .garage .bucket_alias_table .get(&EmptyKey, &query.name) - .await? - .filter(|a| !a.is_deleted()) - .ok_or_bad_request(format!("Bucket {} does not exist", query.name))?; - - let bucket_id = bucket_alias.state.get().as_option().unwrap().bucket_id; + .await?; // Check bucket doesn't have other aliases - let mut bucket = self - .garage - .bucket_helper() - .get_existing_bucket(bucket_id) - .await?; + let mut bucket = helper.get_existing_bucket(bucket_id).await?; let bucket_state = bucket.state.as_option().unwrap(); if bucket_state .aliases @@ -216,18 +221,18 @@ impl AdminRpcHandler { // --- done checking, now commit --- // 1. delete authorization from keys that had access for (key_id, _) in bucket.authorized_keys() { - if let Some(key) = self.garage.key_table.get(&EmptyKey, key_id).await? { - if !key.state.is_deleted() { - self.update_key_bucket(&key, bucket.id, false, false, false) - .await?; - } - } else { - return Err(Error::BadRequest(format!("Key not found: {}", key_id))); - } + helper + .set_bucket_key_permissions(bucket.id, key_id, BucketKeyPerm::no_permissions()) + .await?; } + // 2. delete bucket alias - bucket_alias.state.update(Deletable::Deleted); - self.garage.bucket_alias_table.insert(&bucket_alias).await?; + if bucket_alias.is_some() { + helper + .unset_global_bucket_alias(bucket_id, &query.name) + .await?; + } + // 3. delete bucket bucket.state = Deletable::delete(); self.garage.bucket_table.insert(&bucket).await?; @@ -236,125 +241,39 @@ impl AdminRpcHandler { } async fn handle_alias_bucket(&self, query: &AliasBucketOpt) -> Result { - let bucket_id = self - .garage - .bucket_helper() + let helper = self.garage.bucket_helper(); + + let bucket_id = helper .resolve_global_bucket_name(&query.existing_bucket) .await? .ok_or_bad_request("Bucket not found")?; - let mut bucket = self - .garage - .bucket_helper() - .get_existing_bucket(bucket_id) - .await?; - if let Some(key_local) = &query.local { - let mut key = self.get_existing_key(key_local).await?; - let mut key_param = key.state.as_option_mut().unwrap(); - - if let Some(Deletable::Present(existing_alias)) = - key_param.local_aliases.get(&query.new_name) - { - if *existing_alias == bucket_id { - return Ok(AdminRpc::Ok(format!( - "Alias {} already points to bucket {:?} in namespace of key {}", - query.new_name, bucket_id, key.key_id - ))); - } else { - return Err(Error::BadRequest(format!("Alias {} already exists and points to different bucket: {:?} in namespace of key {}", query.new_name, existing_alias, key.key_id))); - } - } - - if !is_valid_bucket_name(&query.new_name) { - return Err(Error::BadRequest(format!( - INVALID_BUCKET_NAME_MESSAGE!(), - query.new_name - ))); - } - - // Checks ok, add alias - let mut bucket_p = bucket.state.as_option_mut().unwrap(); - let bucket_p_local_alias_key = (key.key_id.clone(), query.new_name.clone()); - - // Calculate the timestamp to assign to this aliasing in the two local_aliases maps - // (the one from key to bucket, and the reverse one stored in the bucket iself) - // so that merges on both maps in case of a concurrent operation resolve - // to the same alias being set - let alias_ts = increment_logical_clock_2( - key_param.local_aliases.get_timestamp(&query.new_name), - bucket_p - .local_aliases - .get_timestamp(&bucket_p_local_alias_key), - ); - - key_param.local_aliases = LwwMap::raw_item( - query.new_name.clone(), - alias_ts, - Deletable::present(bucket_id), - ); - self.garage.key_table.insert(&key).await?; - - bucket_p.local_aliases = LwwMap::raw_item(bucket_p_local_alias_key, alias_ts, true); - self.garage.bucket_table.insert(&bucket).await?; + if let Some(key_pattern) = &query.local { + let key = helper.get_existing_matching_key(key_pattern).await?; + helper + .set_local_bucket_alias(bucket_id, &key.key_id, &query.new_name) + .await?; Ok(AdminRpc::Ok(format!( - "Alias {} created to bucket {:?} in namespace of key {}", + "Alias {} now points to bucket {:?} in namespace of key {}", query.new_name, bucket_id, key.key_id ))) } else { - let alias = self - .garage - .bucket_alias_table - .get(&EmptyKey, &query.new_name) + helper + .set_global_bucket_alias(bucket_id, &query.new_name) .await?; - - if let Some(existing_alias) = alias.as_ref() { - if let Some(p) = existing_alias.state.get().as_option() { - if p.bucket_id == bucket_id { - return Ok(AdminRpc::Ok(format!( - "Alias {} already points to bucket {:?}", - query.new_name, bucket_id - ))); - } else { - return Err(Error::BadRequest(format!( - "Alias {} already exists and points to different bucket: {:?}", - query.new_name, p.bucket_id - ))); - } - } - } - - // Checks ok, add alias - let mut bucket_p = bucket.state.as_option_mut().unwrap(); - - let alias_ts = increment_logical_clock_2( - bucket_p.aliases.get_timestamp(&query.new_name), - alias.as_ref().map(|a| a.state.timestamp()).unwrap_or(0), - ); - - let alias = match alias { - None => BucketAlias::new(query.new_name.clone(), bucket_id) - .ok_or_bad_request(format!(INVALID_BUCKET_NAME_MESSAGE!(), query.new_name))?, - Some(mut a) => { - a.state = Lww::raw(alias_ts, Deletable::present(AliasParams { bucket_id })); - a - } - }; - self.garage.bucket_alias_table.insert(&alias).await?; - - bucket_p.aliases = LwwMap::raw_item(query.new_name.clone(), alias_ts, true); - self.garage.bucket_table.insert(&bucket).await?; - Ok(AdminRpc::Ok(format!( - "Alias {} created to bucket {:?}", + "Alias {} now points to bucket {:?}", query.new_name, bucket_id ))) } } async fn handle_unalias_bucket(&self, query: &UnaliasBucketOpt) -> Result { - if let Some(key_local) = &query.local { - let mut key = self.get_existing_key(key_local).await?; + let helper = self.garage.bucket_helper(); + + if let Some(key_pattern) = &query.local { + let key = helper.get_existing_matching_key(key_pattern).await?; let bucket_id = key .state @@ -365,122 +284,56 @@ impl AdminRpcHandler { .map(|a| a.into_option()) .flatten() .ok_or_bad_request("Bucket not found")?; - let mut bucket = self - .garage - .bucket_helper() - .get_existing_bucket(bucket_id) + + helper + .unset_local_bucket_alias(bucket_id, &key.key_id, &query.name) .await?; - let mut bucket_p = bucket.state.as_option_mut().unwrap(); - - let has_other_aliases = bucket_p - .aliases - .items() - .iter() - .any(|(_, _, active)| *active) - || bucket_p - .local_aliases - .items() - .iter() - .any(|((k, n), _, active)| *k == key.key_id && *n == query.name && *active); - if !has_other_aliases { - return Err(Error::BadRequest(format!("Bucket {} doesn't have other aliases, please delete it instead of just unaliasing.", query.name))); - } - - // Checks ok, remove alias - let mut key_param = key.state.as_option_mut().unwrap(); - let bucket_p_local_alias_key = (key.key_id.clone(), query.name.clone()); - - let alias_ts = increment_logical_clock_2( - key_param.local_aliases.get_timestamp(&query.name), - bucket_p - .local_aliases - .get_timestamp(&bucket_p_local_alias_key), - ); - - key_param.local_aliases = - LwwMap::raw_item(query.name.clone(), alias_ts, Deletable::delete()); - self.garage.key_table.insert(&key).await?; - - bucket_p.local_aliases = LwwMap::raw_item(bucket_p_local_alias_key, alias_ts, false); - self.garage.bucket_table.insert(&bucket).await?; Ok(AdminRpc::Ok(format!( - "Bucket alias {} deleted from namespace of key {}", - query.name, key.key_id + "Alias {} no longer points to bucket {:?} in namespace of key {}", + &query.name, bucket_id, key.key_id ))) } else { - let bucket_id = self - .garage - .bucket_helper() + let bucket_id = helper .resolve_global_bucket_name(&query.name) .await? .ok_or_bad_request("Bucket not found")?; - let mut bucket = self - .garage - .bucket_helper() - .get_existing_bucket(bucket_id) + + helper + .unset_global_bucket_alias(bucket_id, &query.name) .await?; - let mut bucket_state = bucket.state.as_option_mut().unwrap(); - let has_other_aliases = bucket_state - .aliases - .items() - .iter() - .any(|(name, _, active)| *name != query.name && *active) - || bucket_state - .local_aliases - .items() - .iter() - .any(|(_, _, active)| *active); - if !has_other_aliases { - return Err(Error::BadRequest(format!("Bucket {} doesn't have other aliases, please delete it instead of just unaliasing.", query.name))); - } - - let mut alias = self - .garage - .bucket_alias_table - .get(&EmptyKey, &query.name) - .await? - .ok_or_message("Internal error: alias not found")?; - - // Checks ok, remove alias - let alias_ts = increment_logical_clock_2( - alias.state.timestamp(), - bucket_state.aliases.get_timestamp(&query.name), - ); - - alias.state = Lww::raw(alias_ts, Deletable::delete()); - self.garage.bucket_alias_table.insert(&alias).await?; - - bucket_state.aliases = LwwMap::raw_item(query.name.clone(), alias_ts, false); - self.garage.bucket_table.insert(&bucket).await?; - - Ok(AdminRpc::Ok(format!("Bucket alias {} deleted", query.name))) + Ok(AdminRpc::Ok(format!( + "Alias {} no longer points to bucket {:?}", + &query.name, bucket_id + ))) } } async fn handle_bucket_allow(&self, query: &PermBucketOpt) -> Result { - let bucket_id = self - .garage - .bucket_helper() + let helper = self.garage.bucket_helper(); + + let bucket_id = helper .resolve_global_bucket_name(&query.bucket) .await? .ok_or_bad_request("Bucket not found")?; - let bucket = self - .garage - .bucket_helper() - .get_existing_bucket(bucket_id) - .await?; - let key = self.get_existing_key(&query.key_pattern).await?; + let key = helper.get_existing_matching_key(&query.key_pattern).await?; let allow_read = query.read || key.allow_read(&bucket_id); let allow_write = query.write || key.allow_write(&bucket_id); let allow_owner = query.owner || key.allow_owner(&bucket_id); - let new_perm = self - .update_key_bucket(&key, bucket_id, allow_read, allow_write, allow_owner) - .await?; - self.update_bucket_key(bucket, &key.key_id, new_perm) + helper + .set_bucket_key_permissions( + bucket_id, + &key.key_id, + BucketKeyPerm { + timestamp: now_msec(), + allow_read, + allow_write, + allow_owner, + }, + ) .await?; Ok(AdminRpc::Ok(format!( @@ -490,27 +343,29 @@ impl AdminRpcHandler { } async fn handle_bucket_deny(&self, query: &PermBucketOpt) -> Result { - let bucket_id = self - .garage - .bucket_helper() + let helper = self.garage.bucket_helper(); + + let bucket_id = helper .resolve_global_bucket_name(&query.bucket) .await? .ok_or_bad_request("Bucket not found")?; - let bucket = self - .garage - .bucket_helper() - .get_existing_bucket(bucket_id) - .await?; - let key = self.get_existing_key(&query.key_pattern).await?; + let key = helper.get_existing_matching_key(&query.key_pattern).await?; let allow_read = !query.read && key.allow_read(&bucket_id); let allow_write = !query.write && key.allow_write(&bucket_id); let allow_owner = !query.owner && key.allow_owner(&bucket_id); - let new_perm = self - .update_key_bucket(&key, bucket_id, allow_read, allow_write, allow_owner) - .await?; - self.update_bucket_key(bucket, &key.key_id, new_perm) + helper + .set_bucket_key_permissions( + bucket_id, + &key.key_id, + BucketKeyPerm { + timestamp: now_msec(), + allow_read, + allow_write, + allow_owner, + }, + ) .await?; Ok(AdminRpc::Ok(format!( @@ -590,7 +445,11 @@ impl AdminRpcHandler { } async fn handle_key_info(&self, query: &KeyOpt) -> Result { - let key = self.get_existing_key(&query.key_pattern).await?; + let key = self + .garage + .bucket_helper() + .get_existing_matching_key(&query.key_pattern) + .await?; self.key_info_result(key).await } @@ -601,54 +460,43 @@ impl AdminRpcHandler { } async fn handle_rename_key(&self, query: &KeyRenameOpt) -> Result { - let mut key = self.get_existing_key(&query.key_pattern).await?; + let mut key = self + .garage + .bucket_helper() + .get_existing_matching_key(&query.key_pattern) + .await?; key.name.update(query.new_name.clone()); self.garage.key_table.insert(&key).await?; self.key_info_result(key).await } async fn handle_delete_key(&self, query: &KeyDeleteOpt) -> Result { - let mut key = self.get_existing_key(&query.key_pattern).await?; + let helper = self.garage.bucket_helper(); + + let mut key = helper.get_existing_matching_key(&query.key_pattern).await?; + if !query.yes { return Err(Error::BadRequest( "Add --yes flag to really perform this operation".to_string(), )); } + let state = key.state.as_option_mut().unwrap(); // --- done checking, now commit --- // 1. Delete local aliases for (alias, _, to) in state.local_aliases.items().iter() { if let Deletable::Present(bucket_id) = to { - if let Some(mut bucket) = self.garage.bucket_table.get(bucket_id, &EmptyKey).await? - { - if let Deletable::Present(bucket_state) = &mut bucket.state { - bucket_state.local_aliases = bucket_state - .local_aliases - .update_mutator((key.key_id.to_string(), alias.to_string()), false); - self.garage.bucket_table.insert(&bucket).await?; - } - } else { - // ignore - } + helper + .unset_local_bucket_alias(*bucket_id, &key.key_id, alias) + .await?; } } // 2. Delete authorized buckets - for (ab_id, auth) in state.authorized_buckets.items().iter() { - if let Some(bucket) = self.garage.bucket_table.get(ab_id, &EmptyKey).await? { - let new_perm = BucketKeyPerm { - timestamp: increment_logical_clock(auth.timestamp), - allow_read: false, - allow_write: false, - allow_owner: false, - }; - if !bucket.is_deleted() { - self.update_bucket_key(bucket, &key.key_id, new_perm) - .await?; - } - } else { - // ignore - } + for (ab_id, _auth) in state.authorized_buckets.items().iter() { + helper + .set_bucket_key_permissions(*ab_id, &key.key_id, BucketKeyPerm::no_permissions()) + .await?; } // 3. Actually delete key key.state = Deletable::delete(); @@ -671,30 +519,6 @@ impl AdminRpcHandler { self.key_info_result(imported_key).await } - async fn get_existing_key(&self, pattern: &str) -> Result { - let candidates = self - .garage - .key_table - .get_range( - &EmptyKey, - None, - Some(KeyFilter::Matches(pattern.to_string())), - 10, - ) - .await? - .into_iter() - .filter(|k| !k.state.is_deleted()) - .collect::>(); - if candidates.len() != 1 { - Err(Error::BadRequest(format!( - "{} matching keys", - candidates.len() - ))) - } else { - Ok(candidates.into_iter().next().unwrap()) - } - } - async fn key_info_result(&self, key: Key) -> Result { let mut relevant_buckets = HashMap::new(); @@ -714,54 +538,6 @@ impl AdminRpcHandler { Ok(AdminRpc::KeyInfo(key, relevant_buckets)) } - /// Update **key table** to inform of the new linked bucket - async fn update_key_bucket( - &self, - key: &Key, - bucket_id: Uuid, - allow_read: bool, - allow_write: bool, - allow_owner: bool, - ) -> Result { - let mut key = key.clone(); - let mut key_state = key.state.as_option_mut().unwrap(); - - let perm = key_state - .authorized_buckets - .get(&bucket_id) - .cloned() - .map(|old_perm| BucketKeyPerm { - timestamp: increment_logical_clock(old_perm.timestamp), - allow_read, - allow_write, - allow_owner, - }) - .unwrap_or(BucketKeyPerm { - timestamp: now_msec(), - allow_read, - allow_write, - allow_owner, - }); - - key_state.authorized_buckets = Map::put_mutator(bucket_id, perm); - - self.garage.key_table.insert(&key).await?; - Ok(perm) - } - - /// Update **bucket table** to inform of the new linked key - async fn update_bucket_key( - &self, - mut bucket: Bucket, - key_id: &str, - new_perm: BucketKeyPerm, - ) -> Result<(), Error> { - bucket.state.as_option_mut().unwrap().authorized_keys = - Map::put_mutator(key_id.to_string(), new_perm); - self.garage.bucket_table.insert(&bucket).await?; - Ok(()) - } - async fn handle_migrate(self: &Arc, opt: MigrateOpt) -> Result { if !opt.yes { return Err(Error::BadRequest( diff --git a/src/model/bucket_alias_table.rs b/src/model/bucket_alias_table.rs index caae76f1..45807178 100644 --- a/src/model/bucket_alias_table.rs +++ b/src/model/bucket_alias_table.rs @@ -1,7 +1,6 @@ use serde::{Deserialize, Serialize}; use garage_util::data::*; -use garage_util::time::*; use garage_table::crdt::*; use garage_table::*; @@ -24,10 +23,7 @@ impl AutoCrdt for AliasParams { } impl BucketAlias { - pub fn new(name: String, bucket_id: Uuid) -> Option { - Self::raw(name, now_msec(), bucket_id) - } - pub fn raw(name: String, ts: u64, bucket_id: Uuid) -> Option { + pub fn new(name: String, ts: u64, bucket_id: Uuid) -> Option { if !is_valid_bucket_name(&name) { None } else { @@ -101,3 +97,6 @@ pub fn is_valid_bucket_name(n: &str) -> bool { // Bucket names must not end with "-s3alias" && !n.ends_with("-s3alias") } + +/// Error message to return for invalid bucket names +pub const INVALID_BUCKET_NAME_MESSAGE: &str = "Invalid bucket name. See AWS documentation for constraints on S3 bucket names:\nhttps://docs.aws.amazon.com/AmazonS3/latest/userguide/bucketnamingrules.html"; diff --git a/src/model/helper/bucket.rs b/src/model/helper/bucket.rs index e89a723d..bb4b5b24 100644 --- a/src/model/helper/bucket.rs +++ b/src/model/helper/bucket.rs @@ -1,14 +1,20 @@ use garage_table::util::EmptyKey; +use garage_util::crdt::*; use garage_util::data::*; +use garage_util::error::{Error as GarageError, OkOrMessage}; +use garage_util::time::*; -use crate::bucket_table::Bucket; +use crate::bucket_alias_table::*; +use crate::bucket_table::*; use crate::garage::Garage; use crate::helper::error::*; +use crate::key_table::{Key, KeyFilter}; +use crate::permission::BucketKeyPerm; pub struct BucketHelper<'a>(pub(crate) &'a Garage); +#[allow(clippy::ptr_arg)] impl<'a> BucketHelper<'a> { - #[allow(clippy::ptr_arg)] pub async fn resolve_global_bucket_name( &self, bucket_name: &String, @@ -45,12 +51,386 @@ impl<'a> BucketHelper<'a> { } } + /// Returns a Bucket if it is present in bucket table, + /// even if it is in deleted state. Querying a non-existing + /// bucket ID returns an internal error. + pub async fn get_internal_bucket(&self, bucket_id: Uuid) -> Result { + Ok(self + .0 + .bucket_table + .get(&bucket_id, &EmptyKey) + .await? + .ok_or_message(format!("Bucket {:?} does not exist", bucket_id))?) + } + + /// Returns a Bucket if it is present in bucket table, + /// only if it is in non-deleted state. + /// Querying a non-existing bucket ID or a deleted bucket + /// returns a bad request error. pub async fn get_existing_bucket(&self, bucket_id: Uuid) -> Result { self.0 .bucket_table .get(&bucket_id, &EmptyKey) .await? .filter(|b| !b.is_deleted()) - .ok_or_bad_request(format!("Bucket {:?} does not exist", bucket_id)) + .ok_or_bad_request(format!( + "Bucket {:?} does not exist or has been deleted", + bucket_id + )) + } + + /// Returns a Key if it is present in key table, + /// even if it is in deleted state. Querying a non-existing + /// key ID returns an internal error. + pub async fn get_internal_key(&self, key_id: &String) -> Result { + Ok(self + .0 + .key_table + .get(&EmptyKey, key_id) + .await? + .ok_or_message(format!("Key {} does not exist", key_id))?) + } + + /// Returns a Key if it is present in key table, + /// only if it is in non-deleted state. + /// Querying a non-existing key ID or a deleted key + /// returns a bad request error. + pub async fn get_existing_key(&self, key_id: &String) -> Result { + self.0 + .key_table + .get(&EmptyKey, key_id) + .await? + .filter(|b| !b.state.is_deleted()) + .ok_or_bad_request(format!("Key {} does not exist or has been deleted", key_id)) + } + + /// Returns a Key if it is present in key table, + /// looking it up by key ID or by a match on its name, + /// only if it is in non-deleted state. + /// Querying a non-existing key ID or a deleted key + /// returns a bad request error. + pub async fn get_existing_matching_key(&self, pattern: &str) -> Result { + let candidates = self + .0 + .key_table + .get_range( + &EmptyKey, + None, + Some(KeyFilter::Matches(pattern.to_string())), + 10, + ) + .await? + .into_iter() + .filter(|k| !k.state.is_deleted()) + .collect::>(); + if candidates.len() != 1 { + Err(Error::BadRequest(format!( + "{} matching keys", + candidates.len() + ))) + } else { + Ok(candidates.into_iter().next().unwrap()) + } + } + + /// Sets a new alias for a bucket in global namespace. + /// This function fails if: + /// - alias name is not valid according to S3 spec + /// - bucket does not exist or is deleted + /// - alias already exists and points to another bucket + pub async fn set_global_bucket_alias( + &self, + bucket_id: Uuid, + alias_name: &String, + ) -> Result<(), Error> { + if !is_valid_bucket_name(alias_name) { + return Err(Error::BadRequest(format!( + "{}: {}", + alias_name, INVALID_BUCKET_NAME_MESSAGE + ))); + } + + let mut bucket = self.get_existing_bucket(bucket_id).await?; + + let alias = self.0.bucket_alias_table.get(&EmptyKey, alias_name).await?; + + if let Some(existing_alias) = alias.as_ref() { + if let Some(p) = existing_alias.state.get().as_option() { + if p.bucket_id != bucket_id { + return Err(Error::BadRequest(format!( + "Alias {} already exists and points to different bucket: {:?}", + alias_name, p.bucket_id + ))); + } + } + } + + // Checks ok, add alias + let mut bucket_p = bucket.state.as_option_mut().unwrap(); + + let alias_ts = increment_logical_clock_2( + bucket_p.aliases.get_timestamp(alias_name), + alias.as_ref().map(|a| a.state.timestamp()).unwrap_or(0), + ); + + // ---- timestamp-ensured causality barrier ---- + // writes are now done and all writes use timestamp alias_ts + + let alias = match alias { + None => BucketAlias::new(alias_name.clone(), alias_ts, bucket_id) + .ok_or_bad_request(format!("{}: {}", alias_name, INVALID_BUCKET_NAME_MESSAGE))?, + Some(mut a) => { + a.state = Lww::raw(alias_ts, Deletable::present(AliasParams { bucket_id })); + a + } + }; + self.0.bucket_alias_table.insert(&alias).await?; + + bucket_p.aliases = LwwMap::raw_item(alias_name.clone(), alias_ts, true); + self.0.bucket_table.insert(&bucket).await?; + + Ok(()) + } + + /// Unsets an alias for a bucket in global namespace. + /// This function fails if: + /// - bucket does not exist or is deleted + /// - alias does not exist or maps to another bucket (-> internal error) + /// - bucket has no other aliases (global or local) + pub async fn unset_global_bucket_alias( + &self, + bucket_id: Uuid, + alias_name: &String, + ) -> Result<(), Error> { + let mut bucket = self.get_existing_bucket(bucket_id).await?; + let mut bucket_state = bucket.state.as_option_mut().unwrap(); + + let mut alias = self + .0 + .bucket_alias_table + .get(&EmptyKey, alias_name) + .await? + .filter(|a| { + a.state + .get() + .as_option() + .map(|x| x.bucket_id == bucket_id) + .unwrap_or(false) + }) + .ok_or_message(format!( + "Internal error: alias not found or does not point to bucket {:?}", + bucket_id + ))?; + + let has_other_global_aliases = bucket_state + .aliases + .items() + .iter() + .any(|(name, _, active)| name != alias_name && *active); + let has_other_local_aliases = bucket_state + .local_aliases + .items() + .iter() + .any(|(_, _, active)| *active); + if !has_other_global_aliases && !has_other_local_aliases { + return Err(Error::BadRequest(format!("Bucket {} doesn't have other aliases, please delete it instead of just unaliasing.", alias_name))); + } + + // Checks ok, remove alias + let alias_ts = increment_logical_clock_2( + alias.state.timestamp(), + bucket_state.aliases.get_timestamp(alias_name), + ); + + // ---- timestamp-ensured causality barrier ---- + // writes are now done and all writes use timestamp alias_ts + + alias.state = Lww::raw(alias_ts, Deletable::delete()); + self.0.bucket_alias_table.insert(&alias).await?; + + bucket_state.aliases = LwwMap::raw_item(alias_name.clone(), alias_ts, false); + self.0.bucket_table.insert(&bucket).await?; + + Ok(()) + } + + /// Sets a new alias for a bucket in the local namespace of a key. + /// This function fails if: + /// - alias name is not valid according to S3 spec + /// - bucket does not exist or is deleted + /// - key does not exist or is deleted + /// - alias already exists and points to another bucket + pub async fn set_local_bucket_alias( + &self, + bucket_id: Uuid, + key_id: &String, + alias_name: &String, + ) -> Result<(), Error> { + if !is_valid_bucket_name(alias_name) { + return Err(Error::BadRequest(format!( + "{}: {}", + alias_name, INVALID_BUCKET_NAME_MESSAGE + ))); + } + + let mut bucket = self.get_existing_bucket(bucket_id).await?; + let mut key = self.get_existing_key(key_id).await?; + + let mut key_param = key.state.as_option_mut().unwrap(); + + if let Some(Deletable::Present(existing_alias)) = key_param.local_aliases.get(alias_name) { + if *existing_alias != bucket_id { + return Err(Error::BadRequest(format!("Alias {} already exists in namespace of key {} and points to different bucket: {:?}", alias_name, key.key_id, existing_alias))); + } + } + + // Checks ok, add alias + let mut bucket_p = bucket.state.as_option_mut().unwrap(); + let bucket_p_local_alias_key = (key.key_id.clone(), alias_name.clone()); + + // Calculate the timestamp to assign to this aliasing in the two local_aliases maps + // (the one from key to bucket, and the reverse one stored in the bucket iself) + // so that merges on both maps in case of a concurrent operation resolve + // to the same alias being set + let alias_ts = increment_logical_clock_2( + key_param.local_aliases.get_timestamp(alias_name), + bucket_p + .local_aliases + .get_timestamp(&bucket_p_local_alias_key), + ); + + // ---- timestamp-ensured causality barrier ---- + // writes are now done and all writes use timestamp alias_ts + + key_param.local_aliases = + LwwMap::raw_item(alias_name.clone(), alias_ts, Deletable::present(bucket_id)); + self.0.key_table.insert(&key).await?; + + bucket_p.local_aliases = LwwMap::raw_item(bucket_p_local_alias_key, alias_ts, true); + self.0.bucket_table.insert(&bucket).await?; + + Ok(()) + } + + /// Unsets an alias for a bucket in the local namespace of a key. + /// This function fails if: + /// - bucket does not exist or is deleted + /// - key does not exist or is deleted + /// - alias does not exist or maps to another bucket (-> internal error) + /// - bucket has no other aliases (global or local) + pub async fn unset_local_bucket_alias( + &self, + bucket_id: Uuid, + key_id: &String, + alias_name: &String, + ) -> Result<(), Error> { + let mut bucket = self.get_existing_bucket(bucket_id).await?; + let mut key = self.get_existing_key(key_id).await?; + + let mut bucket_p = bucket.state.as_option_mut().unwrap(); + + if key + .state + .as_option() + .unwrap() + .local_aliases + .get(alias_name) + .map(|x| x.as_option()) + .flatten() != Some(&bucket_id) + { + return Err(GarageError::Message(format!( + "Bucket {:?} does not have alias {} in namespace of key {}", + bucket_id, alias_name, key_id + )) + .into()); + } + + let has_other_global_aliases = bucket_p + .aliases + .items() + .iter() + .any(|(_, _, active)| *active); + let has_other_local_aliases = bucket_p + .local_aliases + .items() + .iter() + .any(|((k, n), _, active)| *k == key.key_id && n == alias_name && *active); + if !has_other_global_aliases && !has_other_local_aliases { + return Err(Error::BadRequest(format!("Bucket {} doesn't have other aliases, please delete it instead of just unaliasing.", alias_name))); + } + + // Checks ok, remove alias + let mut key_param = key.state.as_option_mut().unwrap(); + let bucket_p_local_alias_key = (key.key_id.clone(), alias_name.clone()); + + let alias_ts = increment_logical_clock_2( + key_param.local_aliases.get_timestamp(alias_name), + bucket_p + .local_aliases + .get_timestamp(&bucket_p_local_alias_key), + ); + + // ---- timestamp-ensured causality barrier ---- + // writes are now done and all writes use timestamp alias_ts + + key_param.local_aliases = + LwwMap::raw_item(alias_name.clone(), alias_ts, Deletable::delete()); + self.0.key_table.insert(&key).await?; + + bucket_p.local_aliases = LwwMap::raw_item(bucket_p_local_alias_key, alias_ts, false); + self.0.bucket_table.insert(&bucket).await?; + + Ok(()) + } + + /// Sets permissions for a key on a bucket. + /// This function fails if: + /// - bucket or key cannot be found at all (its ok if they are in deleted state) + /// - bucket or key is in deleted state and we are trying to set permissions other than "deny + /// all" + pub async fn set_bucket_key_permissions( + &self, + bucket_id: Uuid, + key_id: &String, + mut perm: BucketKeyPerm, + ) -> Result<(), Error> { + let mut bucket = self.get_internal_bucket(bucket_id).await?; + let mut key = self.get_internal_key(key_id).await?; + + let allow_any = perm.allow_read || perm.allow_write || perm.allow_owner; + + if let Some(bstate) = bucket.state.as_option() { + if let Some(kp) = bstate.authorized_keys.get(key_id) { + perm.timestamp = increment_logical_clock_2(perm.timestamp, kp.timestamp); + } + } else if allow_any { + return Err(Error::BadRequest( + "Trying to give permissions on a deleted bucket".into(), + )); + } + + if let Some(kstate) = key.state.as_option() { + if let Some(bp) = kstate.authorized_buckets.get(&bucket_id) { + perm.timestamp = increment_logical_clock_2(perm.timestamp, bp.timestamp); + } + } else if allow_any { + return Err(Error::BadRequest( + "Trying to give permissions to a deleted key".into(), + )); + } + + // ---- timestamp-ensured causality barrier ---- + + if let Some(bstate) = bucket.state.as_option_mut() { + bstate.authorized_keys = Map::put_mutator(key_id.clone(), perm); + self.0.bucket_table.insert(&bucket).await?; + } + + if let Some(kstate) = key.state.as_option_mut() { + kstate.authorized_buckets = Map::put_mutator(bucket_id, perm); + self.0.key_table.insert(&key).await?; + } + + Ok(()) } } diff --git a/src/model/migrate.rs b/src/model/migrate.rs index a5508c4d..65140c4b 100644 --- a/src/model/migrate.rs +++ b/src/model/migrate.rs @@ -1,9 +1,8 @@ use std::sync::Arc; -use garage_table::util::EmptyKey; use garage_util::crdt::*; use garage_util::data::*; -use garage_util::error::*; +use garage_util::error::Error as GarageError; use garage_util::time::*; use garage_model_050::bucket_table as old_bucket; @@ -11,6 +10,7 @@ use garage_model_050::bucket_table as old_bucket; use crate::bucket_alias_table::*; use crate::bucket_table::*; use crate::garage::Garage; +use crate::helper::error::*; use crate::permission::*; pub struct Migrate { @@ -19,11 +19,16 @@ pub struct Migrate { impl Migrate { pub async fn migrate_buckets050(&self) -> Result<(), Error> { - let tree = self.garage.db.open_tree("bucket:table")?; + let tree = self + .garage + .db + .open_tree("bucket:table") + .map_err(GarageError::from)?; for res in tree.iter() { - let (_k, v) = res?; - let bucket = rmp_serde::decode::from_read_ref::<_, old_bucket::Bucket>(&v[..])?; + let (_k, v) = res.map_err(GarageError::from)?; + let bucket = rmp_serde::decode::from_read_ref::<_, old_bucket::Bucket>(&v[..]) + .map_err(GarageError::from)?; if let old_bucket::BucketState::Present(p) = bucket.state.get() { self.migrate_buckets050_do_bucket(&bucket, p).await?; @@ -48,27 +53,6 @@ impl Migrate { hex::encode(&bucket_id.as_slice()[..16]) }; - let new_ak = old_bucket_p - .authorized_keys - .items() - .iter() - .map(|(k, ts, perm)| { - ( - k.to_string(), - BucketKeyPerm { - timestamp: *ts, - allow_read: perm.allow_read, - allow_write: perm.allow_write, - allow_owner: false, - }, - ) - }) - .collect::>(); - - let mut aliases = LwwMap::new(); - aliases.update_in_place(new_name.clone(), true); - let alias_ts = aliases.get_timestamp(&new_name); - let website = if *old_bucket_p.website.get() { Some(WebsiteConfig { index_document: "index.html".into(), @@ -78,32 +62,39 @@ impl Migrate { None }; - let new_bucket = Bucket { - id: bucket_id, - state: Deletable::Present(BucketParams { - creation_date: now_msec(), - authorized_keys: new_ak.clone(), - website_config: Lww::new(website), - aliases, - local_aliases: LwwMap::new(), - }), - }; - self.garage.bucket_table.insert(&new_bucket).await?; + self.garage + .bucket_table + .insert(&Bucket { + id: bucket_id, + state: Deletable::Present(BucketParams { + creation_date: now_msec(), + authorized_keys: Map::new(), + website_config: Lww::new(website), + aliases: LwwMap::new(), + local_aliases: LwwMap::new(), + }), + }) + .await?; - let new_alias = BucketAlias::raw(new_name.clone(), alias_ts, new_bucket.id).unwrap(); - self.garage.bucket_alias_table.insert(&new_alias).await?; + self.garage + .bucket_helper() + .set_global_bucket_alias(bucket_id, &new_name) + .await?; - for (k, perm) in new_ak.items().iter() { - let mut key = self - .garage - .key_table - .get(&EmptyKey, k) - .await? - .ok_or_message(format!("Missing key: {}", k))?; - if let Some(p) = key.state.as_option_mut() { - p.authorized_buckets.put(new_bucket.id, *perm); - } - self.garage.key_table.insert(&key).await?; + for (k, ts, perm) in old_bucket_p.authorized_keys.items().iter() { + self.garage + .bucket_helper() + .set_bucket_key_permissions( + bucket_id, + k, + BucketKeyPerm { + timestamp: *ts, + allow_read: perm.allow_read, + allow_write: perm.allow_write, + allow_owner: false, + }, + ) + .await?; } Ok(()) diff --git a/src/model/permission.rs b/src/model/permission.rs index ebb24a32..b8f7dd71 100644 --- a/src/model/permission.rs +++ b/src/model/permission.rs @@ -20,6 +20,17 @@ pub struct BucketKeyPerm { pub allow_owner: bool, } +impl BucketKeyPerm { + pub fn no_permissions() -> Self { + Self { + timestamp: 0, + allow_read: false, + allow_write: false, + allow_owner: false, + } + } +} + impl Crdt for BucketKeyPerm { fn merge(&mut self, other: &Self) { match other.timestamp.cmp(&self.timestamp) {