diff --git a/rustfs/src/app/bucket_usecase.rs b/rustfs/src/app/bucket_usecase.rs index add47dc28..e42518962 100644 --- a/rustfs/src/app/bucket_usecase.rs +++ b/rustfs/src/app/bucket_usecase.rs @@ -16,29 +16,64 @@ #![allow(dead_code)] use crate::app::context::{AppContext, get_global_app_context}; +use crate::auth::get_condition_values; use crate::error::ApiError; +use crate::server::RemoteAddr; use crate::storage::access::authorize_request; -use crate::storage::ecfs::{default_owner, serialize_acl, stored_acl_from_canned_bucket, stored_acl_from_grant_headers}; +use crate::storage::ecfs::{ + default_owner, is_public_grant, parse_acl_json_or_canned_bucket, serialize_acl, stored_acl_from_canned_bucket, + stored_acl_from_grant_headers, +}; use crate::storage::helper::OperationHelper; use crate::storage::*; use metrics::counter; -use rustfs_ecstore::bucket::{metadata::BUCKET_ACL_CONFIG, metadata_sys}; +use rustfs_ecstore::bucket::{ + lifecycle::bucket_lifecycle_ops::validate_transition_tier, + metadata::{ + BUCKET_ACL_CONFIG, BUCKET_LIFECYCLE_CONFIG, BUCKET_NOTIFICATION_CONFIG, BUCKET_POLICY_CONFIG, BUCKET_SSECONFIG, + BUCKET_TAGGING_CONFIG, BUCKET_VERSIONING_CONFIG, + }, + metadata_sys, + policy_sys::PolicySys, + utils::serialize, + versioning_sys::BucketVersioningSys, +}; use rustfs_ecstore::client::object_api_utils::to_s3s_etag; use rustfs_ecstore::error::StorageError; use rustfs_ecstore::new_object_layer_fn; use rustfs_ecstore::store_api::{BucketOptions, DeleteBucketOptions, MakeBucketOptions, StorageAPI}; -use rustfs_policy::policy::action::{Action, S3Action}; -use rustfs_targets::EventName; +use rustfs_notify::notifier_global; +use rustfs_policy::policy::{ + action::{Action, S3Action}, + {BucketPolicy, BucketPolicyArgs, Effect, Validator}, +}; +use rustfs_targets::{ + EventName, + arn::{ARN, TargetIDError}, +}; use rustfs_utils::http::RUSTFS_FORCE_DELETE; use rustfs_utils::string::parse_bool; use s3s::dto::*; +use s3s::xml; use s3s::{S3Error, S3ErrorCode, S3Request, S3Response, S3Result, s3_error}; -use std::sync::Arc; -use tracing::{debug, instrument}; +use std::{fmt::Display, sync::Arc}; +use tracing::{debug, info, instrument, warn}; use urlencoding::encode; pub type BucketUsecaseResult = Result; +fn serialize_config(value: &T) -> S3Result> { + serialize(value).map_err(to_internal_error) +} + +fn to_internal_error(err: impl Display) -> S3Error { + S3Error::with_message(S3ErrorCode::InternalError, format!("{err}")) +} + +fn resolve_notification_region(global_region: Option, request_region: Option) -> String { + global_region.unwrap_or_else(|| request_region.unwrap_or_default()) +} + #[derive(Debug, Clone, PartialEq, Eq)] pub struct CreateBucketRequest { pub bucket: String, @@ -264,6 +299,649 @@ impl DefaultBucketUsecase { Ok(S3Response::new(HeadBucketOutput::default())) } + pub async fn execute_delete_bucket_encryption( + &self, + req: S3Request, + ) -> S3Result> { + if let Some(context) = &self.context { + let _ = context.object_store(); + } + + let DeleteBucketEncryptionInput { bucket, .. } = req.input; + + let Some(store) = new_object_layer_fn() else { + return Err(S3Error::with_message(S3ErrorCode::InternalError, "Not init".to_string())); + }; + + store + .get_bucket_info(&bucket, &BucketOptions::default()) + .await + .map_err(ApiError::from)?; + + metadata_sys::delete(&bucket, BUCKET_SSECONFIG) + .await + .map_err(ApiError::from)?; + + Ok(S3Response::new(DeleteBucketEncryptionOutput::default())) + } + + #[instrument(level = "debug", skip(self))] + pub async fn execute_delete_bucket_lifecycle( + &self, + req: S3Request, + ) -> S3Result> { + if let Some(context) = &self.context { + let _ = context.object_store(); + } + + let DeleteBucketLifecycleInput { bucket, .. } = req.input; + + let Some(store) = new_object_layer_fn() else { + return Err(S3Error::with_message(S3ErrorCode::InternalError, "Not init".to_string())); + }; + + store + .get_bucket_info(&bucket, &BucketOptions::default()) + .await + .map_err(ApiError::from)?; + + metadata_sys::delete(&bucket, BUCKET_LIFECYCLE_CONFIG) + .await + .map_err(ApiError::from)?; + + Ok(S3Response::new(DeleteBucketLifecycleOutput::default())) + } + + pub async fn execute_delete_bucket_policy( + &self, + req: S3Request, + ) -> S3Result> { + if let Some(context) = &self.context { + let _ = context.object_store(); + } + + let DeleteBucketPolicyInput { bucket, .. } = req.input; + + let Some(store) = new_object_layer_fn() else { + return Err(S3Error::with_message(S3ErrorCode::InternalError, "Not init".to_string())); + }; + + store + .get_bucket_info(&bucket, &BucketOptions::default()) + .await + .map_err(ApiError::from)?; + + metadata_sys::delete(&bucket, BUCKET_POLICY_CONFIG) + .await + .map_err(ApiError::from)?; + + Ok(S3Response::new(DeleteBucketPolicyOutput {})) + } + + #[instrument(level = "debug", skip(self))] + pub async fn execute_delete_bucket_tagging( + &self, + req: S3Request, + ) -> S3Result> { + if let Some(context) = &self.context { + let _ = context.object_store(); + } + + let DeleteBucketTaggingInput { bucket, .. } = req.input; + + metadata_sys::delete(&bucket, BUCKET_TAGGING_CONFIG) + .await + .map_err(ApiError::from)?; + + Ok(S3Response::new(DeleteBucketTaggingOutput {})) + } + + pub async fn execute_get_bucket_encryption( + &self, + req: S3Request, + ) -> S3Result> { + if let Some(context) = &self.context { + let _ = context.object_store(); + } + + let GetBucketEncryptionInput { bucket, .. } = req.input; + + let Some(store) = new_object_layer_fn() else { + return Err(S3Error::with_message(S3ErrorCode::InternalError, "Not init".to_string())); + }; + + store + .get_bucket_info(&bucket, &BucketOptions::default()) + .await + .map_err(ApiError::from)?; + + let server_side_encryption_configuration = match metadata_sys::get_sse_config(&bucket).await { + Ok((cfg, _)) => Some(cfg), + Err(err) => { + warn!("get_sse_config err {:?}", err); + None + } + }; + + Ok(S3Response::new(GetBucketEncryptionOutput { + server_side_encryption_configuration, + })) + } + + #[instrument(level = "debug", skip(self))] + pub async fn execute_get_bucket_lifecycle_configuration( + &self, + req: S3Request, + ) -> S3Result> { + if let Some(context) = &self.context { + let _ = context.object_store(); + } + + let GetBucketLifecycleConfigurationInput { bucket, .. } = req.input; + + let Some(store) = new_object_layer_fn() else { + return Err(S3Error::with_message(S3ErrorCode::InternalError, "Not init".to_string())); + }; + + store + .get_bucket_info(&bucket, &BucketOptions::default()) + .await + .map_err(ApiError::from)?; + + let rules = match metadata_sys::get_lifecycle_config(&bucket).await { + Ok((cfg, _)) => cfg.rules, + Err(_) => { + return Err(s3_error!(NoSuchLifecycleConfiguration)); + } + }; + + Ok(S3Response::new(GetBucketLifecycleConfigurationOutput { + rules: Some(rules), + ..Default::default() + })) + } + + pub async fn execute_get_bucket_notification_configuration( + &self, + req: S3Request, + ) -> S3Result> { + if let Some(context) = &self.context { + let _ = context.object_store(); + } + + let GetBucketNotificationConfigurationInput { bucket, .. } = req.input; + + let Some(store) = new_object_layer_fn() else { + return Err(S3Error::with_message(S3ErrorCode::InternalError, "Not init".to_string())); + }; + + store + .get_bucket_info(&bucket, &BucketOptions::default()) + .await + .map_err(ApiError::from)?; + + let has_notification_config = metadata_sys::get_notification_config(&bucket).await.unwrap_or_else(|err| { + warn!("get_notification_config err {:?}", err); + None + }); + + if let Some(NotificationConfiguration { + event_bridge_configuration, + lambda_function_configurations, + queue_configurations, + topic_configurations, + }) = has_notification_config + { + Ok(S3Response::new(GetBucketNotificationConfigurationOutput { + event_bridge_configuration, + lambda_function_configurations, + queue_configurations, + topic_configurations, + })) + } else { + Ok(S3Response::new(GetBucketNotificationConfigurationOutput::default())) + } + } + + pub async fn execute_get_bucket_policy( + &self, + req: S3Request, + ) -> S3Result> { + if let Some(context) = &self.context { + let _ = context.object_store(); + } + + let GetBucketPolicyInput { bucket, .. } = req.input; + + let Some(store) = new_object_layer_fn() else { + return Err(S3Error::with_message(S3ErrorCode::InternalError, "Not init".to_string())); + }; + + store + .get_bucket_info(&bucket, &BucketOptions::default()) + .await + .map_err(ApiError::from)?; + + let (policy_str, _) = match metadata_sys::get_bucket_policy_raw(&bucket).await { + Ok(res) => res, + Err(err) => { + if StorageError::ConfigNotFound == err { + return Err(s3_error!(NoSuchBucketPolicy)); + } + return Err(S3Error::with_message(S3ErrorCode::InternalError, err.to_string())); + } + }; + + Ok(S3Response::new(GetBucketPolicyOutput { + policy: Some(policy_str), + })) + } + + pub async fn execute_get_bucket_policy_status( + &self, + req: S3Request, + ) -> S3Result> { + if let Some(context) = &self.context { + let _ = context.object_store(); + } + + let GetBucketPolicyStatusInput { bucket, .. } = req.input; + + let Some(store) = new_object_layer_fn() else { + return Err(S3Error::with_message(S3ErrorCode::InternalError, "Not init".to_string())); + }; + + store + .get_bucket_info(&bucket, &BucketOptions::default()) + .await + .map_err(ApiError::from)?; + + let remote_addr = req.extensions.get::>().and_then(|opt| opt.map(|a| a.0)); + let conditions = get_condition_values(&req.headers, &rustfs_credentials::Credentials::default(), None, None, remote_addr); + + let read_allowed = PolicySys::is_allowed(&BucketPolicyArgs { + bucket: &bucket, + action: Action::S3Action(S3Action::ListBucketAction), + is_owner: false, + account: "", + groups: &None, + conditions: &conditions, + object: "", + }) + .await; + + let write_allowed = PolicySys::is_allowed(&BucketPolicyArgs { + bucket: &bucket, + action: Action::S3Action(S3Action::PutObjectAction), + is_owner: false, + account: "", + groups: &None, + conditions: &conditions, + object: "", + }) + .await; + + let mut is_public = read_allowed || write_allowed; + let ignore_public_acls = match metadata_sys::get_public_access_block_config(&bucket).await { + Ok((config, _)) => config.ignore_public_acls.unwrap_or(false), + Err(_) => false, + }; + + let owner = default_owner(); + let acl_public = match metadata_sys::get_bucket_acl_config(&bucket).await { + Ok((acl, _)) => { + let stored_acl = parse_acl_json_or_canned_bucket(&acl, &owner); + stored_acl + .grants + .iter() + .any(|grant| is_public_grant(grant) && !ignore_public_acls) + } + Err(_) => false, + }; + + if acl_public { + is_public = true; + } + + let policy_public = match metadata_sys::get_bucket_policy(&bucket).await { + Ok((cfg, _)) => cfg.statements.iter().any(|statement| { + matches!(statement.effect, Effect::Allow) + && statement.principal.is_match("*") + && statement.conditions.is_empty() + && statement.actions.is_match(&Action::S3Action(S3Action::ListBucketAction)) + }), + Err(err) => { + if err == StorageError::ConfigNotFound { + false + } else { + return Err(ApiError::from(err).into()); + } + } + }; + + if policy_public { + is_public = true; + } + + let output = GetBucketPolicyStatusOutput { + policy_status: Some(PolicyStatus { + is_public: Some(is_public), + }), + }; + + Ok(S3Response::new(output)) + } + + #[instrument(level = "debug", skip(self))] + pub async fn execute_get_bucket_tagging( + &self, + req: S3Request, + ) -> S3Result> { + if let Some(context) = &self.context { + let _ = context.object_store(); + } + + let GetBucketTaggingInput { bucket, .. } = req.input; + + let Some(store) = new_object_layer_fn() else { + return Err(S3Error::with_message(S3ErrorCode::InternalError, "Not init".to_string())); + }; + + store + .get_bucket_info(&bucket, &BucketOptions::default()) + .await + .map_err(ApiError::from)?; + + let Tagging { tag_set } = match metadata_sys::get_tagging_config(&bucket).await { + Ok((tags, _)) => tags, + Err(err) => { + if err == StorageError::ConfigNotFound { + return Err(S3Error::with_message(S3ErrorCode::NoSuchTagSet, "The TagSet does not exist".to_string())); + } + warn!("get_tagging_config err {:?}", &err); + return Err(ApiError::from(err).into()); + } + }; + + Ok(S3Response::new(GetBucketTaggingOutput { tag_set })) + } + + #[instrument(level = "debug", skip(self))] + pub async fn execute_get_bucket_versioning( + &self, + req: S3Request, + ) -> S3Result> { + if let Some(context) = &self.context { + let _ = context.object_store(); + } + + let GetBucketVersioningInput { bucket, .. } = req.input; + let Some(store) = new_object_layer_fn() else { + return Err(S3Error::with_message(S3ErrorCode::InternalError, "Not init".to_string())); + }; + + store + .get_bucket_info(&bucket, &BucketOptions::default()) + .await + .map_err(ApiError::from)?; + + let VersioningConfiguration { status, .. } = BucketVersioningSys::get(&bucket).await.map_err(ApiError::from)?; + + Ok(S3Response::new(GetBucketVersioningOutput { + status, + ..Default::default() + })) + } + + pub async fn execute_put_bucket_encryption( + &self, + req: S3Request, + ) -> S3Result> { + if let Some(context) = &self.context { + let _ = context.object_store(); + } + + let PutBucketEncryptionInput { + bucket, + server_side_encryption_configuration, + .. + } = req.input; + + info!("sse_config {:?}", &server_side_encryption_configuration); + + let Some(store) = new_object_layer_fn() else { + return Err(S3Error::with_message(S3ErrorCode::InternalError, "Not init".to_string())); + }; + + store + .get_bucket_info(&bucket, &BucketOptions::default()) + .await + .map_err(ApiError::from)?; + + let data = serialize_config(&server_side_encryption_configuration)?; + metadata_sys::update(&bucket, BUCKET_SSECONFIG, data) + .await + .map_err(ApiError::from)?; + Ok(S3Response::new(PutBucketEncryptionOutput::default())) + } + + #[instrument(level = "debug", skip(self))] + pub async fn execute_put_bucket_lifecycle_configuration( + &self, + req: S3Request, + ) -> S3Result> { + if let Some(context) = &self.context { + let _ = context.object_store(); + } + + let PutBucketLifecycleConfigurationInput { + bucket, + lifecycle_configuration, + .. + } = req.input; + + let Some(input_cfg) = lifecycle_configuration else { return Err(s3_error!(InvalidArgument)) }; + + let rcfg = metadata_sys::get_object_lock_config(&bucket).await; + if let Ok(rcfg) = rcfg + && let Err(err) = rustfs_ecstore::bucket::lifecycle::lifecycle::Lifecycle::validate(&input_cfg, &rcfg.0).await + { + return Err(S3Error::with_message(S3ErrorCode::Custom("ValidateFailed".into()), err.to_string())); + } + + if let Err(err) = validate_transition_tier(&input_cfg).await { + return Err(S3Error::with_message(S3ErrorCode::Custom("CustomError".into()), err.to_string())); + } + + let data = serialize_config(&input_cfg)?; + metadata_sys::update(&bucket, BUCKET_LIFECYCLE_CONFIG, data) + .await + .map_err(ApiError::from)?; + + Ok(S3Response::new(PutBucketLifecycleConfigurationOutput::default())) + } + + pub async fn execute_put_bucket_notification_configuration( + &self, + req: S3Request, + ) -> S3Result> { + if let Some(context) = &self.context { + let _ = context.object_store(); + } + let request_region = req.region.clone(); + + let PutBucketNotificationConfigurationInput { + bucket, + notification_configuration, + .. + } = req.input; + + let Some(store) = new_object_layer_fn() else { + return Err(S3Error::with_message(S3ErrorCode::InternalError, "Not init".to_string())); + }; + + store + .get_bucket_info(&bucket, &BucketOptions::default()) + .await + .map_err(ApiError::from)?; + + let data = serialize_config(¬ification_configuration)?; + metadata_sys::update(&bucket, BUCKET_NOTIFICATION_CONFIG, data) + .await + .map_err(ApiError::from)?; + + let region = resolve_notification_region(rustfs_ecstore::global::get_global_region(), request_region); + let clear_rules = notifier_global::clear_bucket_notification_rules(&bucket); + let parse_rules = async { + let mut event_rules = Vec::new(); + + process_queue_configurations(&mut event_rules, notification_configuration.queue_configurations.clone(), |arn_str| { + ARN::parse(arn_str) + .map(|arn| arn.target_id) + .map_err(|e| TargetIDError::InvalidFormat(e.to_string())) + })?; + process_topic_configurations(&mut event_rules, notification_configuration.topic_configurations.clone(), |arn_str| { + ARN::parse(arn_str) + .map(|arn| arn.target_id) + .map_err(|e| TargetIDError::InvalidFormat(e.to_string())) + })?; + process_lambda_configurations( + &mut event_rules, + notification_configuration.lambda_function_configurations.clone(), + |arn_str| { + ARN::parse(arn_str) + .map(|arn| arn.target_id) + .map_err(|e| TargetIDError::InvalidFormat(e.to_string())) + }, + )?; + + Ok::<_, TargetIDError>(event_rules) + }; + + let (clear_result, event_rules_result) = tokio::join!(clear_rules, parse_rules); + + clear_result.map_err(|e| s3_error!(InternalError, "Failed to clear rules: {e}"))?; + let event_rules = + event_rules_result.map_err(|e| s3_error!(InvalidArgument, "Invalid ARN in notification configuration: {e}"))?; + warn!("notify event rules: {:?}", &event_rules); + + notifier_global::add_event_specific_rules(&bucket, ®ion, &event_rules) + .await + .map_err(|e| s3_error!(InternalError, "Failed to add rules: {e}"))?; + + Ok(S3Response::new(PutBucketNotificationConfigurationOutput {})) + } + + pub async fn execute_put_bucket_policy( + &self, + req: S3Request, + ) -> S3Result> { + if let Some(context) = &self.context { + let _ = context.object_store(); + } + + let PutBucketPolicyInput { bucket, policy, .. } = req.input; + + let Some(store) = new_object_layer_fn() else { + return Err(S3Error::with_message(S3ErrorCode::InternalError, "Not init".to_string())); + }; + + store + .get_bucket_info(&bucket, &BucketOptions::default()) + .await + .map_err(ApiError::from)?; + + let cfg: BucketPolicy = + serde_json::from_str(&policy).map_err(|e| s3_error!(InvalidArgument, "parse policy failed {:?}", e))?; + + if let Err(err) = cfg.is_valid() { + warn!("put_bucket_policy err input {:?}, {:?}", &policy, err); + return Err(s3_error!(MalformedPolicy)); + } + + let is_public_policy = cfg + .statements + .iter() + .any(|statement| matches!(statement.effect, Effect::Allow) && statement.principal.is_match("*")); + + if is_public_policy { + match metadata_sys::get_public_access_block_config(&bucket).await { + Ok((config, _)) => { + if config.block_public_policy.unwrap_or(false) { + return Err(s3_error!(AccessDenied, "Access Denied")); + } + } + Err(err) => { + if err != StorageError::ConfigNotFound { + warn!("get_public_access_block_config err {:?}", &err); + return Err(ApiError::from(err).into()); + } + } + } + } + + let data = policy.as_bytes().to_vec(); + + metadata_sys::update(&bucket, BUCKET_POLICY_CONFIG, data) + .await + .map_err(ApiError::from)?; + + Ok(S3Response::new(PutBucketPolicyOutput {})) + } + + #[instrument(level = "debug", skip(self))] + pub async fn execute_put_bucket_tagging( + &self, + req: S3Request, + ) -> S3Result> { + if let Some(context) = &self.context { + let _ = context.object_store(); + } + + let PutBucketTaggingInput { bucket, tagging, .. } = req.input; + + let Some(store) = new_object_layer_fn() else { + return Err(S3Error::with_message(S3ErrorCode::InternalError, "Not init".to_string())); + }; + + store + .get_bucket_info(&bucket, &BucketOptions::default()) + .await + .map_err(ApiError::from)?; + + let data = serialize_config(&tagging)?; + + metadata_sys::update(&bucket, BUCKET_TAGGING_CONFIG, data) + .await + .map_err(ApiError::from)?; + + Ok(S3Response::new(Default::default())) + } + + #[instrument(level = "debug", skip(self))] + pub async fn execute_put_bucket_versioning( + &self, + req: S3Request, + ) -> S3Result> { + if let Some(context) = &self.context { + let _ = context.object_store(); + } + + let PutBucketVersioningInput { + bucket, + versioning_configuration, + .. + } = req.input; + + let data = serialize_config(&versioning_configuration)?; + + metadata_sys::update(&bucket, BUCKET_VERSIONING_CONFIG, data) + .await + .map_err(ApiError::from)?; + + Ok(S3Response::new(PutBucketVersioningOutput {})) + } + #[instrument(level = "debug", skip(self, req))] pub async fn execute_list_objects_v2(&self, req: S3Request) -> S3Result> { if let Some(context) = &self.context { @@ -473,6 +1151,24 @@ mod tests { } } + #[test] + fn resolve_notification_region_prefers_global_region() { + let region = resolve_notification_region(Some("us-east-1".to_string()), Some("ap-southeast-1".to_string())); + assert_eq!(region, "us-east-1"); + } + + #[test] + fn resolve_notification_region_falls_back_to_request_region() { + let region = resolve_notification_region(None, Some("ap-southeast-1".to_string())); + assert_eq!(region, "ap-southeast-1"); + } + + #[test] + fn resolve_notification_region_defaults_to_empty() { + let region = resolve_notification_region(None, None); + assert!(region.is_empty()); + } + #[tokio::test] async fn execute_create_bucket_returns_internal_error_when_store_uninitialized() { let input = CreateBucketInput::builder() @@ -512,6 +1208,63 @@ mod tests { assert_eq!(err.code(), &S3ErrorCode::InternalError); } + #[tokio::test] + async fn execute_get_bucket_policy_returns_internal_error_when_store_uninitialized() { + let input = GetBucketPolicyInput::builder() + .bucket("test-bucket".to_string()) + .build() + .unwrap(); + + let req = build_request(input, Method::GET); + let usecase = DefaultBucketUsecase::without_context(); + + let err = usecase.execute_get_bucket_policy(req).await.unwrap_err(); + assert_eq!(err.code(), &S3ErrorCode::InternalError); + } + + #[tokio::test] + async fn execute_get_bucket_versioning_returns_internal_error_when_store_uninitialized() { + let input = GetBucketVersioningInput::builder() + .bucket("test-bucket".to_string()) + .build() + .unwrap(); + + let req = build_request(input, Method::GET); + let usecase = DefaultBucketUsecase::without_context(); + + let err = usecase.execute_get_bucket_versioning(req).await.unwrap_err(); + assert_eq!(err.code(), &S3ErrorCode::InternalError); + } + + #[tokio::test] + async fn execute_put_bucket_lifecycle_configuration_rejects_missing_configuration() { + let input = PutBucketLifecycleConfigurationInput::builder() + .bucket("test-bucket".to_string()) + .build() + .unwrap(); + + let req = build_request(input, Method::PUT); + let usecase = DefaultBucketUsecase::without_context(); + + let err = usecase.execute_put_bucket_lifecycle_configuration(req).await.unwrap_err(); + assert_eq!(err.code(), &S3ErrorCode::InvalidArgument); + } + + #[tokio::test] + async fn execute_put_bucket_policy_returns_internal_error_when_store_uninitialized() { + let input = PutBucketPolicyInput::builder() + .bucket("test-bucket".to_string()) + .policy("{}".to_string()) + .build() + .unwrap(); + + let req = build_request(input, Method::PUT); + let usecase = DefaultBucketUsecase::without_context(); + + let err = usecase.execute_put_bucket_policy(req).await.unwrap_err(); + assert_eq!(err.code(), &S3ErrorCode::InternalError); + } + #[tokio::test] async fn execute_list_objects_v2_rejects_negative_max_keys() { let input = ListObjectsV2Input::builder() diff --git a/rustfs/src/storage/ecfs.rs b/rustfs/src/storage/ecfs.rs index ed199d99b..8c86bfbe4 100644 --- a/rustfs/src/storage/ecfs.rs +++ b/rustfs/src/storage/ecfs.rs @@ -14,9 +14,7 @@ use crate::app::bucket_usecase::DefaultBucketUsecase; use crate::app::object_usecase::DefaultObjectUsecase; -use crate::auth::get_condition_values; use crate::error::ApiError; -use crate::server::RemoteAddr; use crate::storage::concurrency::{ConcurrencyManager, get_concurrency_manager}; use crate::storage::entity; use crate::storage::helper::OperationHelper; @@ -31,7 +29,6 @@ use crate::storage::{ use crate::storage::{ create_managed_encryption_material, decrypt_managed_encryption_key, derive_part_nonce, get_buffer_size_opt_in, get_validated_store, has_replication_rules, is_managed_sse, parse_object_lock_legal_hold, parse_object_lock_retention, - process_lambda_configurations, process_queue_configurations, process_topic_configurations, validate_bucket_object_lock_enabled, }; use bytes::Bytes; @@ -45,18 +42,16 @@ use rustfs_ecstore::bucket::quota::checker::QuotaChecker; use rustfs_ecstore::{ bucket::{ lifecycle::{ - bucket_lifecycle_ops::{RestoreRequestOps, post_restore_opts, validate_transition_tier}, - lifecycle::{self, Lifecycle, TransitionOptions}, + bucket_lifecycle_ops::{RestoreRequestOps, post_restore_opts}, + lifecycle::{self, TransitionOptions}, }, metadata::{ - BUCKET_ACL_CONFIG, BUCKET_CORS_CONFIG, BUCKET_LIFECYCLE_CONFIG, BUCKET_NOTIFICATION_CONFIG, BUCKET_POLICY_CONFIG, - BUCKET_PUBLIC_ACCESS_BLOCK_CONFIG, BUCKET_REPLICATION_CONFIG, BUCKET_SSECONFIG, BUCKET_TAGGING_CONFIG, + BUCKET_ACL_CONFIG, BUCKET_CORS_CONFIG, BUCKET_PUBLIC_ACCESS_BLOCK_CONFIG, BUCKET_REPLICATION_CONFIG, BUCKET_VERSIONING_CONFIG, OBJECT_LOCK_CONFIG, }, metadata_sys, metadata_sys::get_replication_config, object_lock::objectlock_sys::{check_object_lock_for_deletion, check_retention_for_modification}, - policy_sys::PolicySys, quota::QuotaOperation, replication::{ DeletedObjectReplicationInfo, check_replicate_delete, get_must_replicate_options, must_replicate, @@ -91,17 +86,11 @@ use rustfs_filemeta::RestoreStatusOps; use rustfs_filemeta::{ReplicationStatusType, ReplicationType, VersionPurgeStatusType}; use rustfs_kms::DataKey; use rustfs_notify::{EventArgsBuilder, notifier_global}; -use rustfs_policy::policy::{ - action::{Action, S3Action}, - {BucketPolicy, BucketPolicyArgs, Effect, Validator}, -}; +use rustfs_policy::policy::action::{Action, S3Action}; use rustfs_rio::{CompressReader, DecryptReader, EncryptReader, HashReader, Reader, WarpReader}; use rustfs_s3select_api::query::{Context, Query}; use rustfs_s3select_query::get_global_db; -use rustfs_targets::{ - EventName, - arn::{ARN, TargetIDError}, -}; +use rustfs_targets::EventName; use rustfs_utils::{ CompressionAlgorithm, extract_params_header, extract_resp_elements, get_request_host, get_request_port, get_request_user_agent, @@ -1518,21 +1507,8 @@ impl S3 for FS { &self, req: S3Request, ) -> S3Result> { - let DeleteBucketEncryptionInput { bucket, .. } = req.input; - - let Some(store) = new_object_layer_fn() else { - return Err(S3Error::with_message(S3ErrorCode::InternalError, "Not init".to_string())); - }; - - store - .get_bucket_info(&bucket, &BucketOptions::default()) - .await - .map_err(ApiError::from)?; - metadata_sys::delete(&bucket, BUCKET_SSECONFIG) - .await - .map_err(ApiError::from)?; - - Ok(S3Response::new(DeleteBucketEncryptionOutput::default())) + let usecase = DefaultBucketUsecase::from_global(); + usecase.execute_delete_bucket_encryption(req).await } #[instrument(level = "debug", skip(self))] @@ -1540,44 +1516,16 @@ impl S3 for FS { &self, req: S3Request, ) -> S3Result> { - let DeleteBucketLifecycleInput { bucket, .. } = req.input; - - let Some(store) = new_object_layer_fn() else { - return Err(S3Error::with_message(S3ErrorCode::InternalError, "Not init".to_string())); - }; - - store - .get_bucket_info(&bucket, &BucketOptions::default()) - .await - .map_err(ApiError::from)?; - - metadata_sys::delete(&bucket, BUCKET_LIFECYCLE_CONFIG) - .await - .map_err(ApiError::from)?; - - Ok(S3Response::new(DeleteBucketLifecycleOutput::default())) + let usecase = DefaultBucketUsecase::from_global(); + usecase.execute_delete_bucket_lifecycle(req).await } async fn delete_bucket_policy( &self, req: S3Request, ) -> S3Result> { - let DeleteBucketPolicyInput { bucket, .. } = req.input; - - let Some(store) = new_object_layer_fn() else { - return Err(S3Error::with_message(S3ErrorCode::InternalError, "Not init".to_string())); - }; - - store - .get_bucket_info(&bucket, &BucketOptions::default()) - .await - .map_err(ApiError::from)?; - - metadata_sys::delete(&bucket, BUCKET_POLICY_CONFIG) - .await - .map_err(ApiError::from)?; - - Ok(S3Response::new(DeleteBucketPolicyOutput {})) + let usecase = DefaultBucketUsecase::from_global(); + usecase.execute_delete_bucket_policy(req).await } async fn delete_bucket_replication( @@ -1609,13 +1557,8 @@ impl S3 for FS { &self, req: S3Request, ) -> S3Result> { - let DeleteBucketTaggingInput { bucket, .. } = req.input; - - metadata_sys::delete(&bucket, BUCKET_TAGGING_CONFIG) - .await - .map_err(ApiError::from)?; - - Ok(S3Response::new(DeleteBucketTaggingOutput {})) + let usecase = DefaultBucketUsecase::from_global(); + usecase.execute_delete_bucket_tagging(req).await } #[instrument(level = "debug", skip(self))] @@ -2095,31 +2038,8 @@ impl S3 for FS { &self, req: S3Request, ) -> S3Result> { - let GetBucketEncryptionInput { bucket, .. } = req.input; - - let Some(store) = new_object_layer_fn() else { - return Err(S3Error::with_message(S3ErrorCode::InternalError, "Not init".to_string())); - }; - - store - .get_bucket_info(&bucket, &BucketOptions::default()) - .await - .map_err(ApiError::from)?; - - let server_side_encryption_configuration = match metadata_sys::get_sse_config(&bucket).await { - Ok((cfg, _)) => Some(cfg), - Err(err) => { - // if BucketMetadataError::BucketLifecycleNotFound.is(&err) { - // return Err(s3_error!(ErrNoSuchBucketSSEConfig)); - // } - warn!("get_sse_config err {:?}", err); - None - } - }; - - Ok(S3Response::new(GetBucketEncryptionOutput { - server_side_encryption_configuration, - })) + let usecase = DefaultBucketUsecase::from_global(); + usecase.execute_get_bucket_encryption(req).await } #[instrument(level = "debug", skip(self))] @@ -2127,30 +2047,8 @@ impl S3 for FS { &self, req: S3Request, ) -> S3Result> { - let GetBucketLifecycleConfigurationInput { bucket, .. } = req.input; - - let Some(store) = new_object_layer_fn() else { - return Err(S3Error::with_message(S3ErrorCode::InternalError, "Not init".to_string())); - }; - - store - .get_bucket_info(&bucket, &BucketOptions::default()) - .await - .map_err(ApiError::from)?; - - let rules = match metadata_sys::get_lifecycle_config(&bucket).await { - Ok((cfg, _)) => cfg.rules, - Err(_err) => { - // Return NoSuchLifecycleConfiguration error as expected by S3 clients - // This fixes issue #990 where Ansible S3 roles fail with KeyError: 'Rules' - return Err(s3_error!(NoSuchLifecycleConfiguration)); - } - }; - - Ok(S3Response::new(GetBucketLifecycleConfigurationOutput { - rules: Some(rules), - ..Default::default() - })) + let usecase = DefaultBucketUsecase::from_global(); + usecase.execute_get_bucket_lifecycle_configuration(req).await } /// Get bucket location @@ -2182,160 +2080,21 @@ impl S3 for FS { &self, req: S3Request, ) -> S3Result> { - let GetBucketNotificationConfigurationInput { bucket, .. } = req.input; - - let Some(store) = new_object_layer_fn() else { - return Err(S3Error::with_message(S3ErrorCode::InternalError, "Not init".to_string())); - }; - - store - .get_bucket_info(&bucket, &BucketOptions::default()) - .await - .map_err(ApiError::from)?; - - let has_notification_config = metadata_sys::get_notification_config(&bucket).await.unwrap_or_else(|err| { - warn!("get_notification_config err {:?}", err); - None - }); - - // TODO: valid target list - - if let Some(NotificationConfiguration { - event_bridge_configuration, - lambda_function_configurations, - queue_configurations, - topic_configurations, - }) = has_notification_config - { - Ok(S3Response::new(GetBucketNotificationConfigurationOutput { - event_bridge_configuration, - lambda_function_configurations, - queue_configurations, - topic_configurations, - })) - } else { - Ok(S3Response::new(GetBucketNotificationConfigurationOutput::default())) - } + let usecase = DefaultBucketUsecase::from_global(); + usecase.execute_get_bucket_notification_configuration(req).await } async fn get_bucket_policy(&self, req: S3Request) -> S3Result> { - let GetBucketPolicyInput { bucket, .. } = req.input; - - let Some(store) = new_object_layer_fn() else { - return Err(S3Error::with_message(S3ErrorCode::InternalError, "Not init".to_string())); - }; - - store - .get_bucket_info(&bucket, &BucketOptions::default()) - .await - .map_err(ApiError::from)?; - - // Return the raw policy JSON as originally stored to preserve exact format. - // This ensures GET returns the same document that was PUT, matching S3 behavior. - let (policy_str, _) = match metadata_sys::get_bucket_policy_raw(&bucket).await { - Ok(res) => res, - Err(err) => { - if StorageError::ConfigNotFound == err { - return Err(s3_error!(NoSuchBucketPolicy)); - } - return Err(S3Error::with_message(S3ErrorCode::InternalError, err.to_string())); - } - }; - - Ok(S3Response::new(GetBucketPolicyOutput { - policy: Some(policy_str), - })) + let usecase = DefaultBucketUsecase::from_global(); + usecase.execute_get_bucket_policy(req).await } async fn get_bucket_policy_status( &self, req: S3Request, ) -> S3Result> { - let GetBucketPolicyStatusInput { bucket, .. } = req.input; - - let Some(store) = new_object_layer_fn() else { - return Err(S3Error::with_message(S3ErrorCode::InternalError, "Not init".to_string())); - }; - - store - .get_bucket_info(&bucket, &BucketOptions::default()) - .await - .map_err(ApiError::from)?; - - let remote_addr = req.extensions.get::>().and_then(|opt| opt.map(|a| a.0)); - let conditions = get_condition_values(&req.headers, &rustfs_credentials::Credentials::default(), None, None, remote_addr); - - let read_allowed = PolicySys::is_allowed(&BucketPolicyArgs { - bucket: &bucket, - action: Action::S3Action(S3Action::ListBucketAction), - is_owner: false, - account: "", - groups: &None, - conditions: &conditions, - object: "", - }) - .await; - - let write_allowed = PolicySys::is_allowed(&BucketPolicyArgs { - bucket: &bucket, - action: Action::S3Action(S3Action::PutObjectAction), - is_owner: false, - account: "", - groups: &None, - conditions: &conditions, - object: "", - }) - .await; - - let mut is_public = read_allowed || write_allowed; - let ignore_public_acls = match metadata_sys::get_public_access_block_config(&bucket).await { - Ok((config, _)) => config.ignore_public_acls.unwrap_or(false), - Err(_) => false, - }; - - let owner = default_owner(); - let acl_public = match metadata_sys::get_bucket_acl_config(&bucket).await { - Ok((acl, _)) => { - let stored_acl = parse_acl_json_or_canned_bucket(&acl, &owner); - stored_acl - .grants - .iter() - .any(|grant| is_public_grant(grant) && !ignore_public_acls) - } - Err(_) => false, - }; - - if acl_public { - is_public = true; - } - - let policy_public = match metadata_sys::get_bucket_policy(&bucket).await { - Ok((cfg, _)) => cfg.statements.iter().any(|statement| { - matches!(statement.effect, Effect::Allow) - && statement.principal.is_match("*") - && statement.conditions.is_empty() - && statement.actions.is_match(&Action::S3Action(S3Action::ListBucketAction)) - }), - Err(err) => { - if err == StorageError::ConfigNotFound { - false - } else { - return Err(ApiError::from(err).into()); - } - } - }; - - if policy_public { - is_public = true; - } - - let output = GetBucketPolicyStatusOutput { - policy_status: Some(PolicyStatus { - is_public: Some(is_public), - }), - }; - - Ok(S3Response::new(output)) + let usecase = DefaultBucketUsecase::from_global(); + usecase.execute_get_bucket_policy_status(req).await } async fn get_bucket_replication( @@ -2395,27 +2154,8 @@ impl S3 for FS { #[instrument(level = "debug", skip(self))] async fn get_bucket_tagging(&self, req: S3Request) -> S3Result> { - let bucket = req.input.bucket.clone(); - // check bucket exists. - let _bucket = self - .head_bucket(req.map_input(|input| HeadBucketInput { - bucket: input.bucket, - expected_bucket_owner: None, - })) - .await?; - - let Tagging { tag_set } = match metadata_sys::get_tagging_config(&bucket).await { - Ok((tags, _)) => tags, - Err(err) => { - if err == StorageError::ConfigNotFound { - return Err(S3Error::with_message(S3ErrorCode::NoSuchTagSet, "The TagSet does not exist".to_string())); - } - warn!("get_tagging_config err {:?}", &err); - return Err(ApiError::from(err).into()); - } - }; - - Ok(S3Response::new(GetBucketTaggingOutput { tag_set })) + let usecase = DefaultBucketUsecase::from_global(); + usecase.execute_get_bucket_tagging(req).await } #[instrument(level = "debug", skip(self))] @@ -2455,22 +2195,8 @@ impl S3 for FS { &self, req: S3Request, ) -> S3Result> { - let GetBucketVersioningInput { bucket, .. } = req.input; - let Some(store) = new_object_layer_fn() else { - return Err(S3Error::with_message(S3ErrorCode::InternalError, "Not init".to_string())); - }; - - store - .get_bucket_info(&bucket, &BucketOptions::default()) - .await - .map_err(ApiError::from)?; - - let VersioningConfiguration { status, .. } = BucketVersioningSys::get(&bucket).await.map_err(ApiError::from)?; - - Ok(S3Response::new(GetBucketVersioningOutput { - status, - ..Default::default() - })) + let usecase = DefaultBucketUsecase::from_global(); + usecase.execute_get_bucket_versioning(req).await } /// Get bucket notification @@ -3206,30 +2932,8 @@ impl S3 for FS { &self, req: S3Request, ) -> S3Result> { - let PutBucketEncryptionInput { - bucket, - server_side_encryption_configuration, - .. - } = req.input; - - info!("sse_config {:?}", &server_side_encryption_configuration); - - let Some(store) = new_object_layer_fn() else { - return Err(S3Error::with_message(S3ErrorCode::InternalError, "Not init".to_string())); - }; - - store - .get_bucket_info(&bucket, &BucketOptions::default()) - .await - .map_err(ApiError::from)?; - - // TODO: check kms - - let data = try_!(serialize(&server_side_encryption_configuration)); - metadata_sys::update(&bucket, BUCKET_SSECONFIG, data) - .await - .map_err(ApiError::from)?; - Ok(S3Response::new(PutBucketEncryptionOutput::default())) + let usecase = DefaultBucketUsecase::from_global(); + usecase.execute_put_bucket_encryption(req).await } #[instrument(level = "debug", skip(self))] @@ -3237,157 +2941,21 @@ impl S3 for FS { &self, req: S3Request, ) -> S3Result> { - let PutBucketLifecycleConfigurationInput { - bucket, - lifecycle_configuration, - .. - } = req.input; - - let Some(input_cfg) = lifecycle_configuration else { return Err(s3_error!(InvalidArgument)) }; - - let rcfg = metadata_sys::get_object_lock_config(&bucket).await; - if let Ok(rcfg) = rcfg - && let Err(err) = input_cfg.validate(&rcfg.0).await - { - //return Err(S3Error::with_message(S3ErrorCode::Custom("BucketLockValidateFailed".into()), err.to_string())); - return Err(S3Error::with_message(S3ErrorCode::Custom("ValidateFailed".into()), err.to_string())); - } - - if let Err(err) = validate_transition_tier(&input_cfg).await { - //warn!("lifecycle_configuration add failed, err: {:?}", err); - return Err(S3Error::with_message(S3ErrorCode::Custom("CustomError".into()), err.to_string())); - } - - let data = try_!(serialize(&input_cfg)); - metadata_sys::update(&bucket, BUCKET_LIFECYCLE_CONFIG, data) - .await - .map_err(ApiError::from)?; - - Ok(S3Response::new(PutBucketLifecycleConfigurationOutput::default())) + let usecase = DefaultBucketUsecase::from_global(); + usecase.execute_put_bucket_lifecycle_configuration(req).await } async fn put_bucket_notification_configuration( &self, req: S3Request, ) -> S3Result> { - let PutBucketNotificationConfigurationInput { - bucket, - notification_configuration, - .. - } = req.input; - - let Some(store) = new_object_layer_fn() else { - return Err(S3Error::with_message(S3ErrorCode::InternalError, "Not init".to_string())); - }; - - // Verify that the bucket exists - store - .get_bucket_info(&bucket, &BucketOptions::default()) - .await - .map_err(ApiError::from)?; - - // Persist the new notification configuration - let data = try_!(serialize(¬ification_configuration)); - metadata_sys::update(&bucket, BUCKET_NOTIFICATION_CONFIG, data) - .await - .map_err(ApiError::from)?; - - // Determine region (BucketInfo has no region field) -> use global region or default - let region = rustfs_ecstore::global::get_global_region().unwrap_or_else(|| req.region.clone().unwrap_or_default()); - - // Purge old rules and resolve new rules in parallel - let clear_rules = notifier_global::clear_bucket_notification_rules(&bucket); - let parse_rules = async { - let mut event_rules = Vec::new(); - - process_queue_configurations(&mut event_rules, notification_configuration.queue_configurations.clone(), |arn_str| { - ARN::parse(arn_str) - .map(|arn| arn.target_id) - .map_err(|e| TargetIDError::InvalidFormat(e.to_string())) - })?; - process_topic_configurations(&mut event_rules, notification_configuration.topic_configurations.clone(), |arn_str| { - ARN::parse(arn_str) - .map(|arn| arn.target_id) - .map_err(|e| TargetIDError::InvalidFormat(e.to_string())) - })?; - process_lambda_configurations( - &mut event_rules, - notification_configuration.lambda_function_configurations.clone(), - |arn_str| { - ARN::parse(arn_str) - .map(|arn| arn.target_id) - .map_err(|e| TargetIDError::InvalidFormat(e.to_string())) - }, - )?; - - Ok::<_, TargetIDError>(event_rules) - }; - - let (clear_result, event_rules_result) = tokio::join!(clear_rules, parse_rules); - - clear_result.map_err(|e| s3_error!(InternalError, "Failed to clear rules: {e}"))?; - let event_rules = - event_rules_result.map_err(|e| s3_error!(InvalidArgument, "Invalid ARN in notification configuration: {e}"))?; - warn!("notify event rules: {:?}", &event_rules); - - // Add a new notification rule - notifier_global::add_event_specific_rules(&bucket, ®ion, &event_rules) - .await - .map_err(|e| s3_error!(InternalError, "Failed to add rules: {e}"))?; - - Ok(S3Response::new(PutBucketNotificationConfigurationOutput {})) + let usecase = DefaultBucketUsecase::from_global(); + usecase.execute_put_bucket_notification_configuration(req).await } async fn put_bucket_policy(&self, req: S3Request) -> S3Result> { - let PutBucketPolicyInput { bucket, policy, .. } = req.input; - - let Some(store) = new_object_layer_fn() else { - return Err(S3Error::with_message(S3ErrorCode::InternalError, "Not init".to_string())); - }; - - store - .get_bucket_info(&bucket, &BucketOptions::default()) - .await - .map_err(ApiError::from)?; - - // warn!("input policy {}", &policy); - - let cfg: BucketPolicy = - serde_json::from_str(&policy).map_err(|e| s3_error!(InvalidArgument, "parse policy failed {:?}", e))?; - - if let Err(err) = cfg.is_valid() { - warn!("put_bucket_policy err input {:?}, {:?}", &policy, err); - return Err(s3_error!(MalformedPolicy)); - } - - let is_public_policy = cfg - .statements - .iter() - .any(|statement| matches!(statement.effect, Effect::Allow) && statement.principal.is_match("*")); - - if is_public_policy { - match metadata_sys::get_public_access_block_config(&bucket).await { - Ok((config, _)) => { - if config.block_public_policy.unwrap_or(false) { - return Err(s3_error!(AccessDenied, "Access Denied")); - } - } - Err(err) => { - if err != StorageError::ConfigNotFound { - warn!("get_public_access_block_config err {:?}", &err); - return Err(ApiError::from(err).into()); - } - } - } - } - - let data = policy.as_bytes().to_vec(); - - metadata_sys::update(&bucket, BUCKET_POLICY_CONFIG, data) - .await - .map_err(ApiError::from)?; - - Ok(S3Response::new(PutBucketPolicyOutput {})) + let usecase = DefaultBucketUsecase::from_global(); + usecase.execute_put_bucket_policy(req).await } async fn put_bucket_replication( @@ -3451,24 +3019,8 @@ impl S3 for FS { #[instrument(level = "debug", skip(self))] async fn put_bucket_tagging(&self, req: S3Request) -> S3Result> { - let PutBucketTaggingInput { bucket, tagging, .. } = req.input; - - let Some(store) = new_object_layer_fn() else { - return Err(S3Error::with_message(S3ErrorCode::InternalError, "Not init".to_string())); - }; - - store - .get_bucket_info(&bucket, &BucketOptions::default()) - .await - .map_err(ApiError::from)?; - - let data = try_!(serialize(&tagging)); - - metadata_sys::update(&bucket, BUCKET_TAGGING_CONFIG, data) - .await - .map_err(ApiError::from)?; - - Ok(S3Response::new(Default::default())) + let usecase = DefaultBucketUsecase::from_global(); + usecase.execute_put_bucket_tagging(req).await } #[instrument(level = "debug", skip(self))] @@ -3476,26 +3028,8 @@ impl S3 for FS { &self, req: S3Request, ) -> S3Result> { - let PutBucketVersioningInput { - bucket, - versioning_configuration, - .. - } = req.input; - - // TODO: check other sys - // check site replication enable - // check bucket object lock enable - // check replication suspended - - let data = try_!(serialize(&versioning_configuration)); - - metadata_sys::update(&bucket, BUCKET_VERSIONING_CONFIG, data) - .await - .map_err(ApiError::from)?; - - // TODO: globalSiteReplicationSys.BucketMetaHook - - Ok(S3Response::new(PutBucketVersioningOutput {})) + let usecase = DefaultBucketUsecase::from_global(); + usecase.execute_put_bucket_versioning(req).await } #[instrument(level = "debug", skip(self, req))]