fix(replication): deny non-owner replication config edits under site replication

Under site replication a user holding only bucket-scoped
s3:PutReplicationConfiguration could rewrite or erase the operator-managed
site-repl-* rules, with the change broadcast to every peer (backlog#1948,
audit A1/P2-17).

- Gate PutBucketReplication/DeleteBucketReplication in the S3 handlers:
  when site replication is enabled and the requester is not the owner,
  return MinIO-parity XMinioReplicationDenyEdit (HTTP 400). The gate runs
  after policy authorization and only on the external S3 path; the
  reconciler and peer bucket-meta ingestion are unaffected.
- Defense in depth in the bucket usecase: PUT merges the incoming config
  with the stored site-repl-* rules (same merge as peer ingestion) instead
  of overwriting verbatim; DELETE keeps the site-repl-* rules and never
  garbage-collects a bucket target a surviving site-replication rule still
  references.
- Move is_site_replication_rule / merge_incoming_replication_config /
  replication_target_arn_deployment_id from the admin site-replication
  handler down to rustfs-replication so the app layer can reuse them
  without new layering violations.

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,
         };
     }
 

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;

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,
 };

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::<deployment-id>:<bucket>`), or `None`
+/// for an operator-authored ARN.
+pub fn replication_target_arn_deployment_id(arn: &str) -> Option<String> {
+    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<ReplicationConfiguration>,
+    local: Option<ReplicationConfiguration>,
+) -> Option<ReplicationConfiguration> {
+    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<ReplicationRule> = 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<String> {
     let role = config.role.trim();
     if !role.is_empty() {

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,
 };