diff --git a/crates/ecstore/src/api/mod.rs b/crates/ecstore/src/api/mod.rs index 17fffac3d..3206e921e 100644 --- a/crates/ecstore/src/api/mod.rs +++ b/crates/ecstore/src/api/mod.rs @@ -199,12 +199,12 @@ pub mod bucket { VersionPurgeStatusType, XferStats, commit_force_delete_intent, complete_force_delete_intent, delete_replication_state_from_config, delete_replication_version_id, get_global_replication_pool, get_global_replication_stats, get_proxy_targets, init_background_replication, - invalid_replication_config_status_field, persist_force_delete_intent, read_durable_mrf_backlog, - replication_state_to_filemeta, replication_status_to_filemeta, replication_statuses_map, replication_target_arns, - resync_start_conflict_id, should_remove_replication_target, should_schedule_delete_replication, - should_use_existing_delete_replication_info, should_use_existing_delete_replication_source, - unsupported_replication_config_field, validate_replication_config_structure, validate_replication_config_target_arns, - version_purge_status_to_filemeta, + invalid_replication_config_status_field, is_site_replication_rule, merge_incoming_replication_config, + persist_force_delete_intent, read_durable_mrf_backlog, replication_state_to_filemeta, replication_status_to_filemeta, + replication_statuses_map, replication_target_arn_deployment_id, replication_target_arns, resync_start_conflict_id, + should_remove_replication_target, should_schedule_delete_replication, should_use_existing_delete_replication_info, + should_use_existing_delete_replication_source, unsupported_replication_config_field, + validate_replication_config_structure, validate_replication_config_target_arns, version_purge_status_to_filemeta, }; } diff --git a/crates/ecstore/src/bucket/replication/mod.rs b/crates/ecstore/src/bucket/replication/mod.rs index be0de0121..2279164bd 100644 --- a/crates/ecstore/src/bucket/replication/mod.rs +++ b/crates/ecstore/src/bucket/replication/mod.rs @@ -47,7 +47,8 @@ pub use replication_config_boundary::{ ObjectOpts, REMOTE_TARGET_CAPABILITY_CONTRACT_VERSION, REMOTE_TARGET_UNSUPPORTED_FIELDS, REMOTE_TARGET_WRITABLE_FIELDS, REPLICATION_CAPABILITY_CONTRACT_VERSION, REPLICATION_READ_ONLY_HISTORICAL_FIELDS, REPLICATION_WRITABLE_FIELDS, ReplicationConfigStructureError, ReplicationConfigurationExt, ReplicationTargetValidationError, - invalid_replication_config_status_field, replication_target_arns, should_remove_replication_target, + invalid_replication_config_status_field, is_site_replication_rule, merge_incoming_replication_config, + replication_target_arn_deployment_id, replication_target_arns, should_remove_replication_target, unsupported_replication_config_field, validate_replication_config_structure, validate_replication_config_target_arns, }; pub(crate) use replication_filemeta_boundary::version_purge_statuses_map; diff --git a/crates/ecstore/src/bucket/replication/replication_config_boundary.rs b/crates/ecstore/src/bucket/replication/replication_config_boundary.rs index 484fe7a25..4ef1eef51 100644 --- a/crates/ecstore/src/bucket/replication/replication_config_boundary.rs +++ b/crates/ecstore/src/bucket/replication/replication_config_boundary.rs @@ -16,6 +16,7 @@ pub use rustfs_replication::{ ObjectOpts, REMOTE_TARGET_CAPABILITY_CONTRACT_VERSION, REMOTE_TARGET_UNSUPPORTED_FIELDS, REMOTE_TARGET_WRITABLE_FIELDS, REPLICATION_CAPABILITY_CONTRACT_VERSION, REPLICATION_READ_ONLY_HISTORICAL_FIELDS, REPLICATION_WRITABLE_FIELDS, ReplicationConfigStructureError, ReplicationConfigurationExt, ReplicationRuleExt, ReplicationTargetValidationError, - invalid_replication_config_status_field, replication_target_arns, should_remove_replication_target, + invalid_replication_config_status_field, is_site_replication_rule, merge_incoming_replication_config, + replication_target_arn_deployment_id, replication_target_arns, should_remove_replication_target, unsupported_replication_config_field, validate_replication_config_structure, validate_replication_config_target_arns, }; diff --git a/crates/replication/src/config.rs b/crates/replication/src/config.rs index c8ca328a5..0aed40dbc 100644 --- a/crates/replication/src/config.rs +++ b/crates/replication/src/config.rs @@ -265,6 +265,78 @@ pub fn active_replication_rule_destination_arns(config: &ReplicationConfiguratio arns } +/// Deployment id extracted from a site-replication target ARN +/// (`arn:{rustfs|minio}:replication:::`), or `None` +/// for an operator-authored ARN. +pub fn replication_target_arn_deployment_id(arn: &str) -> Option { + let parts: Vec<_> = arn.split(':').collect(); + if parts.len() == 6 + && parts[0] == "arn" + && matches!(parts[1], "rustfs" | "minio") + && parts[2] == "replication" + && !parts[4].is_empty() + { + return Some(parts[4].to_string()); + } + + None +} + +/// Whether `rule` is a site-replication rule (`site-repl-*` id) owned by the +/// local site's reconciler rather than authored by an operator. +pub fn is_site_replication_rule(rule: &ReplicationRule) -> bool { + rule.id.as_deref().is_some_and(|id| id.starts_with("site-repl-")) +} + +/// Merge an incoming replication config into the local one. +/// +/// `site-repl-*` rules encode the *holder's* outbound direction — their +/// destination ARN names another site — so applying an external rule set +/// verbatim replaces the local reverse rule with one this site can never +/// satisfy (no bucket target backs it) and replication silently stops. Only +/// operator-authored rules travel: the site-replication peer ingestion path +/// and the S3 put/delete-bucket-replication path both keep the local site's +/// `site-repl-*` rules through this merge. `incoming == None` models a +/// delete of the operator-authored rules. +pub fn merge_incoming_replication_config( + incoming: Option, + local: Option, +) -> Option { + let incoming_role = incoming.as_ref().map(|config| config.role.clone()).unwrap_or_default(); + // Operator rules first, then the local site rules — the same order the + // site-replication reconciler produces, so its no-op check matches and + // the bucket metadata is written once per broadcast, not twice. + let mut rules: Vec = incoming + .into_iter() + .flat_map(|config| config.rules) + .filter(|rule| !is_site_replication_rule(rule)) + .collect(); + rules.extend( + local + .into_iter() + .flat_map(|config| config.rules) + .filter(is_site_replication_rule), + ); + + if rules.is_empty() { + return None; + } + + for (index, rule) in rules.iter_mut().enumerate() { + rule.priority = Some(i32::try_from(index + 1).unwrap_or(i32::MAX)); + } + + // A site-replication ARN in `role` is the sender's, and the reconciler's + // per-peer target lookup reads it — carrying it over would pin the + // receiver's targets to the sender's identity. + let role = match replication_target_arn_deployment_id(&incoming_role) { + Some(_) => String::new(), + None => incoming_role, + }; + + Some(ReplicationConfiguration { role, rules }) +} + pub fn replication_target_arns(config: &ReplicationConfiguration) -> HashSet { let role = config.role.trim(); if !role.is_empty() { diff --git a/crates/replication/src/lib.rs b/crates/replication/src/lib.rs index cb0761b5c..e65f3186f 100644 --- a/crates/replication/src/lib.rs +++ b/crates/replication/src/lib.rs @@ -32,7 +32,8 @@ pub use config::{ ObjectOpts, REMOTE_TARGET_CAPABILITY_CONTRACT_VERSION, REMOTE_TARGET_UNSUPPORTED_FIELDS, REMOTE_TARGET_WRITABLE_FIELDS, REPLICATION_CAPABILITY_CONTRACT_VERSION, REPLICATION_READ_ONLY_HISTORICAL_FIELDS, REPLICATION_WRITABLE_FIELDS, ReplicationConfigStructureError, ReplicationConfigurationExt, ReplicationTargetValidationError, - active_replication_rule_destination_arns, invalid_replication_config_status_field, replication_target_arns, + active_replication_rule_destination_arns, invalid_replication_config_status_field, is_site_replication_rule, + merge_incoming_replication_config, replication_target_arn_deployment_id, replication_target_arns, should_remove_replication_target, unsupported_replication_config_field, validate_replication_config_structure, validate_replication_config_target_arns, }; diff --git a/rustfs/src/admin/handlers/site_replication.rs b/rustfs/src/admin/handlers/site_replication.rs index 1bb0379df..86e8914c6 100644 --- a/rustfs/src/admin/handlers/site_replication.rs +++ b/rustfs/src/admin/handlers/site_replication.rs @@ -31,6 +31,9 @@ use crate::admin::storage_api::bucket::metadata::{ use crate::admin::storage_api::bucket::metadata_sys; use crate::admin::storage_api::bucket::quota::BucketQuota; use crate::admin::storage_api::bucket::replication; +use crate::admin::storage_api::bucket::replication::{ + is_site_replication_rule, merge_incoming_replication_config, replication_target_arn_deployment_id, +}; use crate::admin::storage_api::bucket::target::{ARN, BucketTarget, BucketTargetType, BucketTargets, Credentials}; use crate::admin::storage_api::bucket::target_sys::BucketTargetSys; use crate::admin::storage_api::bucket::utils::{deserialize, serialize}; @@ -1117,6 +1120,14 @@ async fn load_site_replication_state() -> S3Result { } } +/// Whether this deployment participates in site replication (two or more +/// peers in the persisted state). Read by the S3 interface layer to gate +/// replication-config edits (MinIO `ErrReplicationDenyEditError` semantics, +/// issue #1948); a state-read failure propagates so the gate fails closed. +pub(crate) async fn site_replication_enabled() -> S3Result { + Ok(load_site_replication_state().await?.enabled()) +} + async fn load_site_replication_state_no_lock(store: Arc) -> S3Result { match read_config_no_lock(store, SITE_REPLICATION_STATE_PATH).await { Ok(data) => parse_site_replication_state(&data), @@ -7748,20 +7759,6 @@ fn bucket_target_deployment_id(target: &BucketTarget) -> Option { replication_target_arn_deployment_id(&target.arn) } -fn replication_target_arn_deployment_id(arn: &str) -> Option { - let parts: Vec<_> = arn.split(':').collect(); - if parts.len() == 6 - && parts[0] == "arn" - && matches!(parts[1], "rustfs" | "minio") - && parts[2] == "replication" - && !parts[4].is_empty() - { - return Some(parts[4].to_string()); - } - - None -} - fn prune_removed_site_replication_bucket_targets( existing: BucketTargets, removed_deployment_ids: &HashSet, @@ -7786,10 +7783,6 @@ fn prune_removed_site_replication_bucket_targets( (BucketTargets { targets }, removed) } -fn is_site_replication_rule(rule: &ReplicationRule) -> bool { - rule.id.as_deref().is_some_and(|id| id.starts_with("site-repl-")) -} - /// Whether every `site-repl-*` rule on this bucket resolves to a live remote target. /// /// The rule set alone cannot answer this: a rule can be perfectly formed while the endpoint @@ -7815,52 +7808,6 @@ async fn site_replication_targets_online(bucket: &str, replication_config_xml: & true } -/// Merge a peer's replication config into the local one. -/// -/// `site-repl-*` rules encode the *sender's* outbound direction — their destination ARN -/// names the receiver — so applying a peer's rule set verbatim replaces the receiver's -/// reverse rule with one pointing at itself. No bucket target can satisfy that ARN -/// (`reconcile_site_replication_bucket_targets` skips the local peer), so the receiver -/// silently stops replicating back: the one-directional symptom. Only operator-authored -/// rules travel between sites; each site owns its own `site-repl-*` rules. -fn merge_incoming_replication_config( - incoming: Option, - local: Option, -) -> Option { - let incoming_role = incoming.as_ref().map(|config| config.role.clone()).unwrap_or_default(); - // Operator rules first, then the local site rules — the same order - // `ensure_site_replication_bucket_replication_config_with_runtime` produces, so its - // no-op check matches and the bucket metadata is written once per broadcast, not twice. - let mut rules: Vec = incoming - .into_iter() - .flat_map(|config| config.rules) - .filter(|rule| !is_site_replication_rule(rule)) - .collect(); - rules.extend( - local - .into_iter() - .flat_map(|config| config.rules) - .filter(is_site_replication_rule), - ); - - if rules.is_empty() { - return None; - } - - for (index, rule) in rules.iter_mut().enumerate() { - rule.priority = Some(i32::try_from(index + 1).unwrap_or(i32::MAX)); - } - - // A site-replication ARN in `role` is the sender's, and `site_replication_target_arns_by_peer` - // reads it — carrying it over would pin the receiver's targets to the sender's identity. - let role = match replication_target_arn_deployment_id(&incoming_role) { - Some(_) => String::new(), - None => incoming_role, - }; - - Some(ReplicationConfiguration { role, rules }) -} - /// Merge a peer's ILM expiry document into the local lifecycle config. /// /// Mirrors MinIO's `mergeWithCurrentLCConfig` with one hardening: incoming diff --git a/rustfs/src/admin/storage_api.rs b/rustfs/src/admin/storage_api.rs index 318718b49..3538d6329 100644 --- a/rustfs/src/admin/storage_api.rs +++ b/rustfs/src/admin/storage_api.rs @@ -443,6 +443,7 @@ pub(crate) mod replication { pub(crate) use super::ecstore_bucket::replication::{ REMOTE_TARGET_CAPABILITY_CONTRACT_VERSION, REMOTE_TARGET_UNSUPPORTED_FIELDS, REMOTE_TARGET_WRITABLE_FIELDS, REPLICATION_CAPABILITY_CONTRACT_VERSION, REPLICATION_READ_ONLY_HISTORICAL_FIELDS, REPLICATION_WRITABLE_FIELDS, + is_site_replication_rule, merge_incoming_replication_config, replication_target_arn_deployment_id, }; pub(crate) type BucketReplicationResyncStatus = super::ecstore_bucket::replication::BucketReplicationResyncStatus; pub(crate) type BucketStats = super::ecstore_bucket::replication::BucketStats; diff --git a/rustfs/src/app/bucket_usecase.rs b/rustfs/src/app/bucket_usecase.rs index 7d55035e9..92a70407a 100644 --- a/rustfs/src/app/bucket_usecase.rs +++ b/rustfs/src/app/bucket_usecase.rs @@ -38,9 +38,9 @@ use super::storage_api::bucket_usecase::bucket::{ metadata_sys, policy_sys::PolicySys, replication::{ - ReplicationTargetValidationError, invalid_replication_config_status_field, replication_target_arns, - should_remove_replication_target, unsupported_replication_config_field, validate_replication_config_structure, - validate_replication_config_target_arns, + ReplicationTargetValidationError, invalid_replication_config_status_field, is_site_replication_rule, + merge_incoming_replication_config, replication_target_arns, should_remove_replication_target, + unsupported_replication_config_field, validate_replication_config_structure, validate_replication_config_target_arns, }, target::{BucketTargetType, BucketTargets}, utils::serialize, @@ -623,11 +623,50 @@ async fn validate_bucket_replication_update(bucket: &str, config: &ReplicationCo validate_replication_config_targets(&targets, config) } -async fn replication_targets_without_config_targets( +/// Defense in depth for site-replication-managed buckets (issue #1948): an S3 +/// PutBucketReplication replaces the operator-authored rules but must not wipe +/// the local `site-repl-*` rules the reconciler owns — until its next pass +/// (600s period) every peer link on this bucket would be silently dead. The +/// same merge also drops incoming `site-repl-*` impostor rules, matching the +/// peer bucket-meta ingestion path. Buckets without site-replication rules +/// keep the verbatim overwrite semantics. +fn merge_user_replication_config_update( + incoming: ReplicationConfiguration, + existing: Option, +) -> ReplicationConfiguration { + let has_site_rules = existing + .as_ref() + .is_some_and(|config| config.rules.iter().any(is_site_replication_rule)); + if !has_site_rules { + return incoming; + } + // `existing` holds at least one site-replication rule the merge keeps, so + // the merged rule set is non-empty; the fallback only guards the type. + merge_incoming_replication_config(Some(incoming.clone()), existing).unwrap_or(incoming) +} + +/// Split of an S3 DeleteBucketReplication on the stored config (issue #1948): +/// the operator-authored rules are removed, the local `site-repl-*` rules +/// survive (`None` means nothing survives and the config is deleted), and the +/// returned ARNs are the ones whose bucket targets may be garbage-collected — +/// never an ARN a surviving site-replication rule still points at. +fn split_replication_config_for_user_delete( + config: ReplicationConfiguration, +) -> (Option, HashSet) { + let mut removable_arns = replication_target_arns(&config); + let remaining = merge_incoming_replication_config(None, Some(config)); + if let Some(remaining) = remaining.as_ref() { + for rule in &remaining.rules { + removable_arns.remove(rule.destination.bucket.trim()); + } + } + (remaining, removable_arns) +} + +async fn replication_targets_without_arns( bucket: &str, - config: &ReplicationConfiguration, + target_arns: &HashSet, ) -> S3Result> { - let target_arns = replication_target_arns(config); if target_arns.is_empty() { return Ok(None); } @@ -638,7 +677,7 @@ async fn replication_targets_without_config_targets( Err(err) => return Err(ApiError::from(err).into()), }; - let removed = remove_replication_targets_from_config_targets(&mut targets, &target_arns); + let removed = remove_replication_targets_from_config_targets(&mut targets, target_arns); if removed == 0 { return Ok(None); } @@ -1604,15 +1643,29 @@ impl DefaultBucketUsecase { Err(StorageError::ConfigNotFound) => None, Err(err) => return Err(ApiError::from(err).into()), }; - let updated_targets = if let Some(config) = replication_config.as_ref() { - replication_targets_without_config_targets(&bucket, config).await? + let (remaining_config, updated_targets) = if let Some(config) = replication_config.as_ref() { + let (remaining, removable_arns) = split_replication_config_for_user_delete(config.clone()); + let targets = replication_targets_without_arns(&bucket, &removable_arns).await?; + (remaining, targets) } else { - None + (None, None) }; - delete_bucket_config_for_incarnation(&bucket, BUCKET_REPLICATION_CONFIG, expected_incarnation_id) - .await - .map_err(ApiError::from)?; + match remaining_config { + // Site-replication rules and the targets backing them survive the + // S3 delete (issue #1948); only the operator-authored rules go. + Some(remaining) => { + let data = serialize_config(&remaining)?; + update_bucket_config_for_incarnation(&bucket, BUCKET_REPLICATION_CONFIG, data, expected_incarnation_id) + .await + .map_err(ApiError::from)?; + } + None => { + delete_bucket_config_for_incarnation(&bucket, BUCKET_REPLICATION_CONFIG, expected_incarnation_id) + .await + .map_err(ApiError::from)?; + } + } if let Some((targets, removed)) = updated_targets && let Err(err) = write_replication_targets_after_config_delete(&bucket, &targets, removed, expected_incarnation_id).await @@ -2485,6 +2538,12 @@ impl DefaultBucketUsecase { let targets_guard = lock_bucket_targets_metadata(&bucket).await; validate_bucket_replication_update(&bucket, &replication_configuration).await?; + let existing_config = match metadata_sys::get_replication_config(&bucket).await { + Ok((config, _)) => Some(config), + Err(StorageError::ConfigNotFound) => None, + Err(err) => return Err(ApiError::from(err).into()), + }; + let replication_configuration = merge_user_replication_config_update(replication_configuration, existing_config); let data = serialize_config(&replication_configuration)?; update_bucket_config_for_incarnation(&bucket, BUCKET_REPLICATION_CONFIG, data, expected_incarnation_id) .await @@ -3114,6 +3173,127 @@ mod tests { assert!(arns.contains(destination)); } + fn replication_rule_with_id(arn: &str, id: &str, priority: i32) -> ReplicationRule { + let mut rule = replication_rule_for_target(arn); + rule.id = Some(id.to_string()); + rule.priority = Some(priority); + rule + } + + #[test] + fn put_replication_merge_preserves_site_replication_rules() { + let existing = ReplicationConfiguration { + role: String::new(), + rules: vec![ + replication_rule_with_id("arn:rustfs:replication::peer-dep:bucket", "site-repl-peer-dep", 1), + replication_rule_with_id("arn:rustfs:replication:us-east-1:old:bucket", "old-user-rule", 2), + ], + }; + let incoming = ReplicationConfiguration { + role: String::new(), + rules: vec![ + replication_rule_with_id("arn:rustfs:replication:us-east-1:new:bucket", "new-user-rule", 1), + replication_rule_with_id("arn:rustfs:replication::forged-dep:bucket", "site-repl-forged", 2), + ], + }; + + let merged = merge_user_replication_config_update(incoming, Some(existing)); + + let ids: Vec<_> = merged + .rules + .iter() + .map(|rule| rule.id.as_deref().unwrap_or_default()) + .collect(); + assert_eq!( + ids, + vec!["new-user-rule", "site-repl-peer-dep"], + "user rules replaced, local site-replication rule preserved, forged incoming site-repl rule dropped" + ); + } + + #[test] + fn put_replication_merge_returns_incoming_verbatim_without_site_rules() { + let existing = ReplicationConfiguration { + role: String::new(), + rules: vec![replication_rule_with_id( + "arn:rustfs:replication:us-east-1:old:bucket", + "old-user-rule", + 7, + )], + }; + let incoming = ReplicationConfiguration { + role: String::new(), + rules: vec![replication_rule_with_id( + "arn:rustfs:replication:us-east-1:new:bucket", + "new-user-rule", + 5, + )], + }; + + let merged = merge_user_replication_config_update(incoming.clone(), Some(existing)); + + assert_eq!(merged.role, incoming.role); + assert_eq!(merged.rules, incoming.rules, "non-SR buckets keep the verbatim overwrite semantics"); + } + + #[test] + fn delete_replication_split_keeps_site_rules_and_their_targets() { + let sr_arn = "arn:rustfs:replication::peer-dep:bucket"; + let user_arn = "arn:rustfs:replication:us-east-1:user:bucket"; + let config = ReplicationConfiguration { + role: String::new(), + rules: vec![ + replication_rule_with_id(user_arn, "user-rule", 1), + replication_rule_with_id(sr_arn, "site-repl-peer-dep", 2), + ], + }; + + let (remaining, removable) = split_replication_config_for_user_delete(config); + + let remaining = remaining.expect("site-replication rules must survive a user delete"); + let ids: Vec<_> = remaining + .rules + .iter() + .map(|rule| rule.id.as_deref().unwrap_or_default()) + .collect(); + assert_eq!(ids, vec!["site-repl-peer-dep"]); + assert_eq!(removable, HashSet::from([user_arn.to_string()])); + } + + #[test] + fn delete_replication_split_protects_targets_shared_with_site_rules() { + let sr_arn = "arn:rustfs:replication::peer-dep:bucket"; + let config = ReplicationConfiguration { + role: String::new(), + rules: vec![ + replication_rule_with_id(sr_arn, "user-rule-on-sr-target", 1), + replication_rule_with_id(sr_arn, "site-repl-peer-dep", 2), + ], + }; + + let (remaining, removable) = split_replication_config_for_user_delete(config); + + assert!(remaining.is_some()); + assert!( + removable.is_empty(), + "a target still referenced by a surviving site-replication rule must not be removed" + ); + } + + #[test] + fn delete_replication_split_removes_everything_without_site_rules() { + let user_arn = "arn:rustfs:replication:us-east-1:user:bucket"; + let config = ReplicationConfiguration { + role: String::new(), + rules: vec![replication_rule_with_id(user_arn, "user-rule", 1)], + }; + + let (remaining, removable) = split_replication_config_for_user_delete(config); + + assert!(remaining.is_none(), "without site-replication rules the whole config is deleted"); + assert_eq!(removable, HashSet::from([user_arn.to_string()])); + } + fn replication_targets_with_arn(arns: &[&str]) -> BucketTargets { BucketTargets { targets: arns diff --git a/rustfs/src/app/storage_api.rs b/rustfs/src/app/storage_api.rs index d90cfb6a8..0a4515b6b 100644 --- a/rustfs/src/app/storage_api.rs +++ b/rustfs/src/app/storage_api.rs @@ -614,6 +614,8 @@ pub(crate) mod bucket { use crate::storage::storage_api::ecstore_bucket::replication as replication_contracts; + pub(crate) use replication_contracts::{is_site_replication_rule, merge_incoming_replication_config}; + type ReplicationObjectBridge = crate::storage::storage_api::ecstore_bucket::replication::ReplicationObjectBridge; pub(crate) type DeleteReplicationConfigSnapshot = crate::storage::storage_api::ecstore_bucket::replication::DeleteReplicationConfigSnapshot; diff --git a/rustfs/src/storage/ecfs.rs b/rustfs/src/storage/ecfs.rs index da0c9b775..0d2db854d 100644 --- a/rustfs/src/storage/ecfs.rs +++ b/rustfs/src/storage/ecfs.rs @@ -63,6 +63,54 @@ use crate::app::storage_api::object_usecase::bucket::replication::{ }; use crate::storage::storage_api::ecfs_consumer::StorageObjectOptions as ObjectOptions; +#[cfg(test)] +static SITE_REPLICATION_GATE_TEST_OVERRIDE: std::sync::atomic::AtomicU8 = std::sync::atomic::AtomicU8::new(0); +#[cfg(test)] +const SITE_REPLICATION_GATE_FORCE_DISABLED: u8 = 1; +#[cfg(test)] +const SITE_REPLICATION_GATE_FORCE_ENABLED: u8 = 2; + +async fn site_replication_gate_enabled() -> S3Result { + #[cfg(test)] + match SITE_REPLICATION_GATE_TEST_OVERRIDE.load(std::sync::atomic::Ordering::SeqCst) { + SITE_REPLICATION_GATE_FORCE_DISABLED => return Ok(false), + SITE_REPLICATION_GATE_FORCE_ENABLED => return Ok(true), + _ => {} + } + crate::admin::handlers::site_replication::site_replication_enabled().await +} + +/// MinIO `ErrReplicationDenyEditError`. +fn replication_deny_edit_error() -> S3Error { + let mut err = S3Error::with_message( + S3ErrorCode::Custom("XMinioReplicationDenyEdit".into()), + "Sub-User is not allowed to edit Replication configuration", + ); + err.set_status_code(StatusCode::BAD_REQUEST); + err +} + +/// Site-replication gate for S3 replication-config edits (issue #1948). +/// +/// On a site-replication deployment the bucket's replication config carries +/// the operator-managed `site-repl-*` rules that keep every peer in sync, and +/// a successful edit is broadcast to all peers — so a user holding only +/// bucket-scoped `s3:PutReplicationConfiguration` could rewrite or erase +/// replication net-wide. MinIO parity (`ErrReplicationDenyEditError`): only +/// owner credentials (root or root-parented) may edit. Runs after the policy +/// authorization in the access layer and only on the external S3 path — the +/// reconciler and peer bucket-meta ingestion never route through these +/// handlers. +async fn deny_replication_config_edit_for_non_owner(req: &S3Request) -> S3Result<()> { + if crate::storage::access::req_info_ref(req)?.is_owner { + return Ok(()); + } + if site_replication_gate_enabled().await? { + return Err(replication_deny_edit_error()); + } + Ok(()) +} + #[derive(Debug, Clone)] pub struct FS { /// This server's late-bound application-context slot (backlog#1052 S2). @@ -500,6 +548,7 @@ impl S3 for FS { &self, req: S3Request, ) -> S3Result> { + deny_replication_config_edit_for_non_owner(&req).await?; let usecase = s3_api::bucket_usecase_for(self); usecase.execute_delete_bucket_replication(req).await } @@ -1353,6 +1402,7 @@ impl S3 for FS { &self, req: S3Request, ) -> S3Result> { + deny_replication_config_edit_for_non_owner(&req).await?; let usecase = s3_api::bucket_usecase_for(self); usecase.execute_put_bucket_replication(req).await } @@ -1919,3 +1969,103 @@ impl S3 for FS { Box::pin(usecase.execute_upload_part_copy(req)).await } } + +#[cfg(test)] +mod tests { + use super::{ + FS, SITE_REPLICATION_GATE_FORCE_DISABLED, SITE_REPLICATION_GATE_FORCE_ENABLED, SITE_REPLICATION_GATE_TEST_OVERRIDE, + }; + use crate::storage::access::ReqInfo; + use http::Method; + use http::StatusCode; + use s3s::dto::{DeleteBucketReplicationInput, PutBucketReplicationInput, ReplicationConfiguration}; + use s3s::{S3, S3Error, S3ErrorCode, S3Request}; + use std::sync::atomic::Ordering; + + fn replication_config_edit_request(input: T, is_owner: bool) -> S3Request { + let mut req = S3Request { + input, + method: Method::PUT, + uri: http::Uri::from_static("/"), + headers: http::HeaderMap::new(), + extensions: http::Extensions::new(), + credentials: None, + region: None, + service: None, + trailing_headers: None, + }; + req.extensions.insert(ReqInfo { + is_owner, + ..Default::default() + }); + req + } + + fn put_bucket_replication_input() -> PutBucketReplicationInput { + PutBucketReplicationInput { + bucket: "test-bucket".to_string(), + checksum_algorithm: None, + content_md5: None, + expected_bucket_owner: None, + replication_configuration: ReplicationConfiguration { + role: String::new(), + rules: Vec::new(), + }, + token: None, + } + } + + fn delete_bucket_replication_input() -> DeleteBucketReplicationInput { + DeleteBucketReplicationInput { + bucket: "test-bucket".to_string(), + expected_bucket_owner: None, + } + } + + fn assert_replication_deny_edit(err: &S3Error) { + match err.code() { + S3ErrorCode::Custom(code) => assert_eq!(code, "XMinioReplicationDenyEdit"), + other => panic!("expected XMinioReplicationDenyEdit, got {other:?}"), + } + assert_eq!(err.status_code(), Some(StatusCode::BAD_REQUEST)); + } + + /// Single test on purpose: the branches share the process-wide gate + /// override, and parallel tests would race it. + #[tokio::test] + async fn replication_config_edit_gate_denies_only_non_owner_under_site_replication() { + let fs = FS::new(); + SITE_REPLICATION_GATE_TEST_OVERRIDE.store(SITE_REPLICATION_GATE_FORCE_ENABLED, Ordering::SeqCst); + + // Non-owner PUT/DELETE through the real S3 handlers: denied by the + // gate before the usecase (and thus the store) is ever touched. + let err = fs + .put_bucket_replication(replication_config_edit_request(put_bucket_replication_input(), false)) + .await + .expect_err("non-owner PutBucketReplication must be denied while site replication is enabled"); + assert_replication_deny_edit(&err); + let err = fs + .delete_bucket_replication(replication_config_edit_request(delete_bucket_replication_input(), false)) + .await + .expect_err("non-owner DeleteBucketReplication must be denied while site replication is enabled"); + assert_replication_deny_edit(&err); + + // Owner passes the gate (the usecase's empty-rules structure error + // proves the request reached the usecase instead of the deny path). + let err = fs + .put_bucket_replication(replication_config_edit_request(put_bucket_replication_input(), true)) + .await + .expect_err("owner request should pass the gate and fail later on config validation"); + assert_eq!(err.code(), &S3ErrorCode::InvalidRequest); + + // Without site replication the policy check alone still governs the edit. + SITE_REPLICATION_GATE_TEST_OVERRIDE.store(SITE_REPLICATION_GATE_FORCE_DISABLED, Ordering::SeqCst); + let err = fs + .put_bucket_replication(replication_config_edit_request(put_bucket_replication_input(), false)) + .await + .expect_err("non-owner request should pass the gate and fail later on config validation"); + assert_eq!(err.code(), &S3ErrorCode::InvalidRequest); + + SITE_REPLICATION_GATE_TEST_OVERRIDE.store(0, Ordering::SeqCst); + } +}