Refactor logic for setting/unsetting aliases
This commit is contained in:
parent
2140cd7205
commit
e59c23a69d
@ -7,7 +7,7 @@ use serde::{Deserialize, Serialize};
|
|||||||
|
|
||||||
use garage_util::crdt::*;
|
use garage_util::crdt::*;
|
||||||
use garage_util::data::*;
|
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_util::time::*;
|
||||||
|
|
||||||
use garage_table::replication::*;
|
use garage_table::replication::*;
|
||||||
@ -28,8 +28,6 @@ use crate::repair::Repair;
|
|||||||
|
|
||||||
pub const ADMIN_RPC_PATH: &str = "garage/admin_rpc.rs/Rpc";
|
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)]
|
#[derive(Debug, Serialize, Deserialize)]
|
||||||
pub enum AdminRpc {
|
pub enum AdminRpc {
|
||||||
BucketOperation(BucketOperation),
|
BucketOperation(BucketOperation),
|
||||||
@ -134,47 +132,54 @@ impl AdminRpcHandler {
|
|||||||
|
|
||||||
#[allow(clippy::ptr_arg)]
|
#[allow(clippy::ptr_arg)]
|
||||||
async fn handle_create_bucket(&self, name: &String) -> Result<AdminRpc, Error> {
|
async fn handle_create_bucket(&self, name: &String) -> Result<AdminRpc, Error> {
|
||||||
let mut bucket = Bucket::new();
|
if !is_valid_bucket_name(name) {
|
||||||
let alias = match self.garage.bucket_alias_table.get(&EmptyKey, name).await? {
|
return Err(Error::BadRequest(format!(
|
||||||
Some(mut alias) => {
|
"{}: {}",
|
||||||
|
name, INVALID_BUCKET_NAME_MESSAGE
|
||||||
|
)));
|
||||||
|
}
|
||||||
|
|
||||||
|
if let Some(alias) = self.garage.bucket_alias_table.get(&EmptyKey, name).await? {
|
||||||
if !alias.state.get().is_deleted() {
|
if !alias.state.get().is_deleted() {
|
||||||
return Err(Error::BadRequest(format!("Bucket {} already exists", name)));
|
return Err(Error::BadRequest(format!("Bucket {} already exists", name)));
|
||||||
}
|
}
|
||||||
alias.state.update(Deletable::Present(AliasParams {
|
|
||||||
bucket_id: bucket.id,
|
|
||||||
}));
|
|
||||||
alias
|
|
||||||
}
|
}
|
||||||
None => BucketAlias::new(name.clone(), bucket.id)
|
|
||||||
.ok_or_bad_request(format!(INVALID_BUCKET_NAME_MESSAGE!(), name))?,
|
// ---- done checking, now commit ----
|
||||||
};
|
|
||||||
bucket.state.as_option_mut().unwrap().aliases.merge_raw(
|
let bucket = Bucket::new();
|
||||||
name,
|
|
||||||
alias.state.timestamp(),
|
|
||||||
&true,
|
|
||||||
);
|
|
||||||
self.garage.bucket_table.insert(&bucket).await?;
|
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)))
|
Ok(AdminRpc::Ok(format!("Bucket {} was created.", name)))
|
||||||
}
|
}
|
||||||
|
|
||||||
async fn handle_delete_bucket(&self, query: &DeleteBucketOpt) -> Result<AdminRpc, Error> {
|
async fn handle_delete_bucket(&self, query: &DeleteBucketOpt) -> Result<AdminRpc, Error> {
|
||||||
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
|
.garage
|
||||||
.bucket_alias_table
|
.bucket_alias_table
|
||||||
.get(&EmptyKey, &query.name)
|
.get(&EmptyKey, &query.name)
|
||||||
.await?
|
.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;
|
|
||||||
|
|
||||||
// Check bucket doesn't have other aliases
|
// Check bucket doesn't have other aliases
|
||||||
let mut bucket = self
|
let mut bucket = helper.get_existing_bucket(bucket_id).await?;
|
||||||
.garage
|
|
||||||
.bucket_helper()
|
|
||||||
.get_existing_bucket(bucket_id)
|
|
||||||
.await?;
|
|
||||||
let bucket_state = bucket.state.as_option().unwrap();
|
let bucket_state = bucket.state.as_option().unwrap();
|
||||||
if bucket_state
|
if bucket_state
|
||||||
.aliases
|
.aliases
|
||||||
@ -216,18 +221,18 @@ impl AdminRpcHandler {
|
|||||||
// --- done checking, now commit ---
|
// --- done checking, now commit ---
|
||||||
// 1. delete authorization from keys that had access
|
// 1. delete authorization from keys that had access
|
||||||
for (key_id, _) in bucket.authorized_keys() {
|
for (key_id, _) in bucket.authorized_keys() {
|
||||||
if let Some(key) = self.garage.key_table.get(&EmptyKey, key_id).await? {
|
helper
|
||||||
if !key.state.is_deleted() {
|
.set_bucket_key_permissions(bucket.id, key_id, BucketKeyPerm::no_permissions())
|
||||||
self.update_key_bucket(&key, bucket.id, false, false, false)
|
|
||||||
.await?;
|
.await?;
|
||||||
}
|
}
|
||||||
} else {
|
|
||||||
return Err(Error::BadRequest(format!("Key not found: {}", key_id)));
|
|
||||||
}
|
|
||||||
}
|
|
||||||
// 2. delete bucket alias
|
// 2. delete bucket alias
|
||||||
bucket_alias.state.update(Deletable::Deleted);
|
if bucket_alias.is_some() {
|
||||||
self.garage.bucket_alias_table.insert(&bucket_alias).await?;
|
helper
|
||||||
|
.unset_global_bucket_alias(bucket_id, &query.name)
|
||||||
|
.await?;
|
||||||
|
}
|
||||||
|
|
||||||
// 3. delete bucket
|
// 3. delete bucket
|
||||||
bucket.state = Deletable::delete();
|
bucket.state = Deletable::delete();
|
||||||
self.garage.bucket_table.insert(&bucket).await?;
|
self.garage.bucket_table.insert(&bucket).await?;
|
||||||
@ -236,125 +241,39 @@ impl AdminRpcHandler {
|
|||||||
}
|
}
|
||||||
|
|
||||||
async fn handle_alias_bucket(&self, query: &AliasBucketOpt) -> Result<AdminRpc, Error> {
|
async fn handle_alias_bucket(&self, query: &AliasBucketOpt) -> Result<AdminRpc, Error> {
|
||||||
let bucket_id = self
|
let helper = self.garage.bucket_helper();
|
||||||
.garage
|
|
||||||
.bucket_helper()
|
let bucket_id = helper
|
||||||
.resolve_global_bucket_name(&query.existing_bucket)
|
.resolve_global_bucket_name(&query.existing_bucket)
|
||||||
.await?
|
.await?
|
||||||
.ok_or_bad_request("Bucket not found")?;
|
.ok_or_bad_request("Bucket not found")?;
|
||||||
let mut bucket = self
|
|
||||||
.garage
|
if let Some(key_pattern) = &query.local {
|
||||||
.bucket_helper()
|
let key = helper.get_existing_matching_key(key_pattern).await?;
|
||||||
.get_existing_bucket(bucket_id)
|
|
||||||
|
helper
|
||||||
|
.set_local_bucket_alias(bucket_id, &key.key_id, &query.new_name)
|
||||||
.await?;
|
.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?;
|
|
||||||
|
|
||||||
Ok(AdminRpc::Ok(format!(
|
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
|
query.new_name, bucket_id, key.key_id
|
||||||
)))
|
)))
|
||||||
} else {
|
} else {
|
||||||
let alias = self
|
helper
|
||||||
.garage
|
.set_global_bucket_alias(bucket_id, &query.new_name)
|
||||||
.bucket_alias_table
|
|
||||||
.get(&EmptyKey, &query.new_name)
|
|
||||||
.await?;
|
.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!(
|
Ok(AdminRpc::Ok(format!(
|
||||||
"Alias {} created to bucket {:?}",
|
"Alias {} now points to bucket {:?}",
|
||||||
query.new_name, bucket_id
|
query.new_name, bucket_id
|
||||||
)))
|
)))
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
async fn handle_unalias_bucket(&self, query: &UnaliasBucketOpt) -> Result<AdminRpc, Error> {
|
async fn handle_unalias_bucket(&self, query: &UnaliasBucketOpt) -> Result<AdminRpc, Error> {
|
||||||
if let Some(key_local) = &query.local {
|
let helper = self.garage.bucket_helper();
|
||||||
let mut key = self.get_existing_key(key_local).await?;
|
|
||||||
|
if let Some(key_pattern) = &query.local {
|
||||||
|
let key = helper.get_existing_matching_key(key_pattern).await?;
|
||||||
|
|
||||||
let bucket_id = key
|
let bucket_id = key
|
||||||
.state
|
.state
|
||||||
@ -365,122 +284,56 @@ impl AdminRpcHandler {
|
|||||||
.map(|a| a.into_option())
|
.map(|a| a.into_option())
|
||||||
.flatten()
|
.flatten()
|
||||||
.ok_or_bad_request("Bucket not found")?;
|
.ok_or_bad_request("Bucket not found")?;
|
||||||
let mut bucket = self
|
|
||||||
.garage
|
helper
|
||||||
.bucket_helper()
|
.unset_local_bucket_alias(bucket_id, &key.key_id, &query.name)
|
||||||
.get_existing_bucket(bucket_id)
|
|
||||||
.await?;
|
.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!(
|
Ok(AdminRpc::Ok(format!(
|
||||||
"Bucket alias {} deleted from namespace of key {}",
|
"Alias {} no longer points to bucket {:?} in namespace of key {}",
|
||||||
query.name, key.key_id
|
&query.name, bucket_id, key.key_id
|
||||||
)))
|
)))
|
||||||
} else {
|
} else {
|
||||||
let bucket_id = self
|
let bucket_id = helper
|
||||||
.garage
|
|
||||||
.bucket_helper()
|
|
||||||
.resolve_global_bucket_name(&query.name)
|
.resolve_global_bucket_name(&query.name)
|
||||||
.await?
|
.await?
|
||||||
.ok_or_bad_request("Bucket not found")?;
|
.ok_or_bad_request("Bucket not found")?;
|
||||||
let mut bucket = self
|
|
||||||
.garage
|
helper
|
||||||
.bucket_helper()
|
.unset_global_bucket_alias(bucket_id, &query.name)
|
||||||
.get_existing_bucket(bucket_id)
|
|
||||||
.await?;
|
.await?;
|
||||||
let mut bucket_state = bucket.state.as_option_mut().unwrap();
|
|
||||||
|
|
||||||
let has_other_aliases = bucket_state
|
Ok(AdminRpc::Ok(format!(
|
||||||
.aliases
|
"Alias {} no longer points to bucket {:?}",
|
||||||
.items()
|
&query.name, bucket_id
|
||||||
.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)))
|
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
async fn handle_bucket_allow(&self, query: &PermBucketOpt) -> Result<AdminRpc, Error> {
|
async fn handle_bucket_allow(&self, query: &PermBucketOpt) -> Result<AdminRpc, Error> {
|
||||||
let bucket_id = self
|
let helper = self.garage.bucket_helper();
|
||||||
.garage
|
|
||||||
.bucket_helper()
|
let bucket_id = helper
|
||||||
.resolve_global_bucket_name(&query.bucket)
|
.resolve_global_bucket_name(&query.bucket)
|
||||||
.await?
|
.await?
|
||||||
.ok_or_bad_request("Bucket not found")?;
|
.ok_or_bad_request("Bucket not found")?;
|
||||||
let bucket = self
|
let key = helper.get_existing_matching_key(&query.key_pattern).await?;
|
||||||
.garage
|
|
||||||
.bucket_helper()
|
|
||||||
.get_existing_bucket(bucket_id)
|
|
||||||
.await?;
|
|
||||||
let key = self.get_existing_key(&query.key_pattern).await?;
|
|
||||||
|
|
||||||
let allow_read = query.read || key.allow_read(&bucket_id);
|
let allow_read = query.read || key.allow_read(&bucket_id);
|
||||||
let allow_write = query.write || key.allow_write(&bucket_id);
|
let allow_write = query.write || key.allow_write(&bucket_id);
|
||||||
let allow_owner = query.owner || key.allow_owner(&bucket_id);
|
let allow_owner = query.owner || key.allow_owner(&bucket_id);
|
||||||
|
|
||||||
let new_perm = self
|
helper
|
||||||
.update_key_bucket(&key, bucket_id, allow_read, allow_write, allow_owner)
|
.set_bucket_key_permissions(
|
||||||
.await?;
|
bucket_id,
|
||||||
self.update_bucket_key(bucket, &key.key_id, new_perm)
|
&key.key_id,
|
||||||
|
BucketKeyPerm {
|
||||||
|
timestamp: now_msec(),
|
||||||
|
allow_read,
|
||||||
|
allow_write,
|
||||||
|
allow_owner,
|
||||||
|
},
|
||||||
|
)
|
||||||
.await?;
|
.await?;
|
||||||
|
|
||||||
Ok(AdminRpc::Ok(format!(
|
Ok(AdminRpc::Ok(format!(
|
||||||
@ -490,27 +343,29 @@ impl AdminRpcHandler {
|
|||||||
}
|
}
|
||||||
|
|
||||||
async fn handle_bucket_deny(&self, query: &PermBucketOpt) -> Result<AdminRpc, Error> {
|
async fn handle_bucket_deny(&self, query: &PermBucketOpt) -> Result<AdminRpc, Error> {
|
||||||
let bucket_id = self
|
let helper = self.garage.bucket_helper();
|
||||||
.garage
|
|
||||||
.bucket_helper()
|
let bucket_id = helper
|
||||||
.resolve_global_bucket_name(&query.bucket)
|
.resolve_global_bucket_name(&query.bucket)
|
||||||
.await?
|
.await?
|
||||||
.ok_or_bad_request("Bucket not found")?;
|
.ok_or_bad_request("Bucket not found")?;
|
||||||
let bucket = self
|
let key = helper.get_existing_matching_key(&query.key_pattern).await?;
|
||||||
.garage
|
|
||||||
.bucket_helper()
|
|
||||||
.get_existing_bucket(bucket_id)
|
|
||||||
.await?;
|
|
||||||
let key = self.get_existing_key(&query.key_pattern).await?;
|
|
||||||
|
|
||||||
let allow_read = !query.read && key.allow_read(&bucket_id);
|
let allow_read = !query.read && key.allow_read(&bucket_id);
|
||||||
let allow_write = !query.write && key.allow_write(&bucket_id);
|
let allow_write = !query.write && key.allow_write(&bucket_id);
|
||||||
let allow_owner = !query.owner && key.allow_owner(&bucket_id);
|
let allow_owner = !query.owner && key.allow_owner(&bucket_id);
|
||||||
|
|
||||||
let new_perm = self
|
helper
|
||||||
.update_key_bucket(&key, bucket_id, allow_read, allow_write, allow_owner)
|
.set_bucket_key_permissions(
|
||||||
.await?;
|
bucket_id,
|
||||||
self.update_bucket_key(bucket, &key.key_id, new_perm)
|
&key.key_id,
|
||||||
|
BucketKeyPerm {
|
||||||
|
timestamp: now_msec(),
|
||||||
|
allow_read,
|
||||||
|
allow_write,
|
||||||
|
allow_owner,
|
||||||
|
},
|
||||||
|
)
|
||||||
.await?;
|
.await?;
|
||||||
|
|
||||||
Ok(AdminRpc::Ok(format!(
|
Ok(AdminRpc::Ok(format!(
|
||||||
@ -590,7 +445,11 @@ impl AdminRpcHandler {
|
|||||||
}
|
}
|
||||||
|
|
||||||
async fn handle_key_info(&self, query: &KeyOpt) -> Result<AdminRpc, Error> {
|
async fn handle_key_info(&self, query: &KeyOpt) -> Result<AdminRpc, Error> {
|
||||||
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
|
self.key_info_result(key).await
|
||||||
}
|
}
|
||||||
|
|
||||||
@ -601,55 +460,44 @@ impl AdminRpcHandler {
|
|||||||
}
|
}
|
||||||
|
|
||||||
async fn handle_rename_key(&self, query: &KeyRenameOpt) -> Result<AdminRpc, Error> {
|
async fn handle_rename_key(&self, query: &KeyRenameOpt) -> Result<AdminRpc, Error> {
|
||||||
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());
|
key.name.update(query.new_name.clone());
|
||||||
self.garage.key_table.insert(&key).await?;
|
self.garage.key_table.insert(&key).await?;
|
||||||
self.key_info_result(key).await
|
self.key_info_result(key).await
|
||||||
}
|
}
|
||||||
|
|
||||||
async fn handle_delete_key(&self, query: &KeyDeleteOpt) -> Result<AdminRpc, Error> {
|
async fn handle_delete_key(&self, query: &KeyDeleteOpt) -> Result<AdminRpc, Error> {
|
||||||
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 {
|
if !query.yes {
|
||||||
return Err(Error::BadRequest(
|
return Err(Error::BadRequest(
|
||||||
"Add --yes flag to really perform this operation".to_string(),
|
"Add --yes flag to really perform this operation".to_string(),
|
||||||
));
|
));
|
||||||
}
|
}
|
||||||
|
|
||||||
let state = key.state.as_option_mut().unwrap();
|
let state = key.state.as_option_mut().unwrap();
|
||||||
|
|
||||||
// --- done checking, now commit ---
|
// --- done checking, now commit ---
|
||||||
// 1. Delete local aliases
|
// 1. Delete local aliases
|
||||||
for (alias, _, to) in state.local_aliases.items().iter() {
|
for (alias, _, to) in state.local_aliases.items().iter() {
|
||||||
if let Deletable::Present(bucket_id) = to {
|
if let Deletable::Present(bucket_id) = to {
|
||||||
if let Some(mut bucket) = self.garage.bucket_table.get(bucket_id, &EmptyKey).await?
|
helper
|
||||||
{
|
.unset_local_bucket_alias(*bucket_id, &key.key_id, alias)
|
||||||
if let Deletable::Present(bucket_state) = &mut bucket.state {
|
.await?;
|
||||||
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
|
|
||||||
}
|
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
// 2. Delete authorized buckets
|
// 2. Delete authorized buckets
|
||||||
for (ab_id, auth) in state.authorized_buckets.items().iter() {
|
for (ab_id, _auth) in state.authorized_buckets.items().iter() {
|
||||||
if let Some(bucket) = self.garage.bucket_table.get(ab_id, &EmptyKey).await? {
|
helper
|
||||||
let new_perm = BucketKeyPerm {
|
.set_bucket_key_permissions(*ab_id, &key.key_id, BucketKeyPerm::no_permissions())
|
||||||
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?;
|
.await?;
|
||||||
}
|
}
|
||||||
} else {
|
|
||||||
// ignore
|
|
||||||
}
|
|
||||||
}
|
|
||||||
// 3. Actually delete key
|
// 3. Actually delete key
|
||||||
key.state = Deletable::delete();
|
key.state = Deletable::delete();
|
||||||
self.garage.key_table.insert(&key).await?;
|
self.garage.key_table.insert(&key).await?;
|
||||||
@ -671,30 +519,6 @@ impl AdminRpcHandler {
|
|||||||
self.key_info_result(imported_key).await
|
self.key_info_result(imported_key).await
|
||||||
}
|
}
|
||||||
|
|
||||||
async fn get_existing_key(&self, pattern: &str) -> Result<Key, Error> {
|
|
||||||
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::<Vec<_>>();
|
|
||||||
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<AdminRpc, Error> {
|
async fn key_info_result(&self, key: Key) -> Result<AdminRpc, Error> {
|
||||||
let mut relevant_buckets = HashMap::new();
|
let mut relevant_buckets = HashMap::new();
|
||||||
|
|
||||||
@ -714,54 +538,6 @@ impl AdminRpcHandler {
|
|||||||
Ok(AdminRpc::KeyInfo(key, relevant_buckets))
|
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<BucketKeyPerm, Error> {
|
|
||||||
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<Self>, opt: MigrateOpt) -> Result<AdminRpc, Error> {
|
async fn handle_migrate(self: &Arc<Self>, opt: MigrateOpt) -> Result<AdminRpc, Error> {
|
||||||
if !opt.yes {
|
if !opt.yes {
|
||||||
return Err(Error::BadRequest(
|
return Err(Error::BadRequest(
|
||||||
|
@ -1,7 +1,6 @@
|
|||||||
use serde::{Deserialize, Serialize};
|
use serde::{Deserialize, Serialize};
|
||||||
|
|
||||||
use garage_util::data::*;
|
use garage_util::data::*;
|
||||||
use garage_util::time::*;
|
|
||||||
|
|
||||||
use garage_table::crdt::*;
|
use garage_table::crdt::*;
|
||||||
use garage_table::*;
|
use garage_table::*;
|
||||||
@ -24,10 +23,7 @@ impl AutoCrdt for AliasParams {
|
|||||||
}
|
}
|
||||||
|
|
||||||
impl BucketAlias {
|
impl BucketAlias {
|
||||||
pub fn new(name: String, bucket_id: Uuid) -> Option<Self> {
|
pub fn new(name: String, ts: u64, bucket_id: Uuid) -> Option<Self> {
|
||||||
Self::raw(name, now_msec(), bucket_id)
|
|
||||||
}
|
|
||||||
pub fn raw(name: String, ts: u64, bucket_id: Uuid) -> Option<Self> {
|
|
||||||
if !is_valid_bucket_name(&name) {
|
if !is_valid_bucket_name(&name) {
|
||||||
None
|
None
|
||||||
} else {
|
} else {
|
||||||
@ -101,3 +97,6 @@ pub fn is_valid_bucket_name(n: &str) -> bool {
|
|||||||
// Bucket names must not end with "-s3alias"
|
// Bucket names must not end with "-s3alias"
|
||||||
&& !n.ends_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";
|
||||||
|
@ -1,14 +1,20 @@
|
|||||||
use garage_table::util::EmptyKey;
|
use garage_table::util::EmptyKey;
|
||||||
|
use garage_util::crdt::*;
|
||||||
use garage_util::data::*;
|
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::garage::Garage;
|
||||||
use crate::helper::error::*;
|
use crate::helper::error::*;
|
||||||
|
use crate::key_table::{Key, KeyFilter};
|
||||||
|
use crate::permission::BucketKeyPerm;
|
||||||
|
|
||||||
pub struct BucketHelper<'a>(pub(crate) &'a Garage);
|
pub struct BucketHelper<'a>(pub(crate) &'a Garage);
|
||||||
|
|
||||||
impl<'a> BucketHelper<'a> {
|
|
||||||
#[allow(clippy::ptr_arg)]
|
#[allow(clippy::ptr_arg)]
|
||||||
|
impl<'a> BucketHelper<'a> {
|
||||||
pub async fn resolve_global_bucket_name(
|
pub async fn resolve_global_bucket_name(
|
||||||
&self,
|
&self,
|
||||||
bucket_name: &String,
|
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<Bucket, Error> {
|
||||||
|
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<Bucket, Error> {
|
pub async fn get_existing_bucket(&self, bucket_id: Uuid) -> Result<Bucket, Error> {
|
||||||
self.0
|
self.0
|
||||||
.bucket_table
|
.bucket_table
|
||||||
.get(&bucket_id, &EmptyKey)
|
.get(&bucket_id, &EmptyKey)
|
||||||
.await?
|
.await?
|
||||||
.filter(|b| !b.is_deleted())
|
.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<Key, Error> {
|
||||||
|
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<Key, Error> {
|
||||||
|
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<Key, Error> {
|
||||||
|
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::<Vec<_>>();
|
||||||
|
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(())
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
@ -1,9 +1,8 @@
|
|||||||
use std::sync::Arc;
|
use std::sync::Arc;
|
||||||
|
|
||||||
use garage_table::util::EmptyKey;
|
|
||||||
use garage_util::crdt::*;
|
use garage_util::crdt::*;
|
||||||
use garage_util::data::*;
|
use garage_util::data::*;
|
||||||
use garage_util::error::*;
|
use garage_util::error::Error as GarageError;
|
||||||
use garage_util::time::*;
|
use garage_util::time::*;
|
||||||
|
|
||||||
use garage_model_050::bucket_table as old_bucket;
|
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_alias_table::*;
|
||||||
use crate::bucket_table::*;
|
use crate::bucket_table::*;
|
||||||
use crate::garage::Garage;
|
use crate::garage::Garage;
|
||||||
|
use crate::helper::error::*;
|
||||||
use crate::permission::*;
|
use crate::permission::*;
|
||||||
|
|
||||||
pub struct Migrate {
|
pub struct Migrate {
|
||||||
@ -19,11 +19,16 @@ pub struct Migrate {
|
|||||||
|
|
||||||
impl Migrate {
|
impl Migrate {
|
||||||
pub async fn migrate_buckets050(&self) -> Result<(), Error> {
|
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() {
|
for res in tree.iter() {
|
||||||
let (_k, v) = res?;
|
let (_k, v) = res.map_err(GarageError::from)?;
|
||||||
let bucket = rmp_serde::decode::from_read_ref::<_, old_bucket::Bucket>(&v[..])?;
|
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() {
|
if let old_bucket::BucketState::Present(p) = bucket.state.get() {
|
||||||
self.migrate_buckets050_do_bucket(&bucket, p).await?;
|
self.migrate_buckets050_do_bucket(&bucket, p).await?;
|
||||||
@ -48,27 +53,6 @@ impl Migrate {
|
|||||||
hex::encode(&bucket_id.as_slice()[..16])
|
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::<Map<_, _>>();
|
|
||||||
|
|
||||||
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() {
|
let website = if *old_bucket_p.website.get() {
|
||||||
Some(WebsiteConfig {
|
Some(WebsiteConfig {
|
||||||
index_document: "index.html".into(),
|
index_document: "index.html".into(),
|
||||||
@ -78,32 +62,39 @@ impl Migrate {
|
|||||||
None
|
None
|
||||||
};
|
};
|
||||||
|
|
||||||
let new_bucket = Bucket {
|
self.garage
|
||||||
|
.bucket_table
|
||||||
|
.insert(&Bucket {
|
||||||
id: bucket_id,
|
id: bucket_id,
|
||||||
state: Deletable::Present(BucketParams {
|
state: Deletable::Present(BucketParams {
|
||||||
creation_date: now_msec(),
|
creation_date: now_msec(),
|
||||||
authorized_keys: new_ak.clone(),
|
authorized_keys: Map::new(),
|
||||||
website_config: Lww::new(website),
|
website_config: Lww::new(website),
|
||||||
aliases,
|
aliases: LwwMap::new(),
|
||||||
local_aliases: LwwMap::new(),
|
local_aliases: LwwMap::new(),
|
||||||
}),
|
}),
|
||||||
};
|
})
|
||||||
self.garage.bucket_table.insert(&new_bucket).await?;
|
.await?;
|
||||||
|
|
||||||
let new_alias = BucketAlias::raw(new_name.clone(), alias_ts, new_bucket.id).unwrap();
|
self.garage
|
||||||
self.garage.bucket_alias_table.insert(&new_alias).await?;
|
.bucket_helper()
|
||||||
|
.set_global_bucket_alias(bucket_id, &new_name)
|
||||||
|
.await?;
|
||||||
|
|
||||||
for (k, perm) in new_ak.items().iter() {
|
for (k, ts, perm) in old_bucket_p.authorized_keys.items().iter() {
|
||||||
let mut key = self
|
self.garage
|
||||||
.garage
|
.bucket_helper()
|
||||||
.key_table
|
.set_bucket_key_permissions(
|
||||||
.get(&EmptyKey, k)
|
bucket_id,
|
||||||
.await?
|
k,
|
||||||
.ok_or_message(format!("Missing key: {}", k))?;
|
BucketKeyPerm {
|
||||||
if let Some(p) = key.state.as_option_mut() {
|
timestamp: *ts,
|
||||||
p.authorized_buckets.put(new_bucket.id, *perm);
|
allow_read: perm.allow_read,
|
||||||
}
|
allow_write: perm.allow_write,
|
||||||
self.garage.key_table.insert(&key).await?;
|
allow_owner: false,
|
||||||
|
},
|
||||||
|
)
|
||||||
|
.await?;
|
||||||
}
|
}
|
||||||
|
|
||||||
Ok(())
|
Ok(())
|
||||||
|
@ -20,6 +20,17 @@ pub struct BucketKeyPerm {
|
|||||||
pub allow_owner: bool,
|
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 {
|
impl Crdt for BucketKeyPerm {
|
||||||
fn merge(&mut self, other: &Self) {
|
fn merge(&mut self, other: &Self) {
|
||||||
match other.timestamp.cmp(&self.timestamp) {
|
match other.timestamp.cmp(&self.timestamp) {
|
||||||
|
Loading…
Reference in New Issue
Block a user