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

* 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. * fix(replication): scope site-owned rule detection to reconciler-derived rules The `site-repl-*` prefix alone classified any rule as site-owned, so on a bucket outside site replication an owner's `site-repl-user` rule survived DeleteBucketReplication (rule and target kept, success returned). Rule ids do not reserve that namespace. A rule is reconciler-owned only when it matches what the reconciler derives: id `site-repl-<deployment id>` for a current remote site replication peer and a destination ARN naming that same deployment id. The S3 put/delete path reads the remote peer set (empty when site replication is disabled) and keeps exactly those rules; everything else is operator state the request replaces or deletes. An incoming rule that claims a current peer's id is dropped so the reconciler rule's id stays unique. The peer ingestion path and the reconciler keep their prefix predicate unchanged. * fix(replication): keep operator rule priorities across site rule merges Merging stored site-replication rules into a PutBucketReplication body renumbered every rule 1..n in list order, rewriting the submitted policy: overlapping same-target rules submitted as priority 5 then 1 became 1 then 2, so the delete-marker-disabled rule won the replication decision. The reconciler and the peer-removal prune renumbered the same way. Operator priorities now stay verbatim everywhere; only the reconciler's derived rules move, to the lowest priorities no operator rule uses, via one pure helper shared by the S3 edit merge, the peer ingestion merge, the reconciler pass and the prune. Being a pure function of the rule list it is idempotent, so the reconciler's no-op check still holds after a merged write, and an on-disk config in the historical layout (operator rules 1..k, site rules k+1..n) yields the same bytes, so nothing is rewritten on upgrade. * fix(replication): pass site peer ids into the bucket usecase from the interface layer The review fix made the bucket usecase read the site-replication peer set through the admin handlers, an app->interface import the layer guard rejects. The S3 handlers (interface) now read the peer set and pass it in, so the usecase stays a pure function of its inputs; a state-read failure still fails the edit closed, just one layer up. * fix(replication): classify peer-ingested rules by the derived id/ARN contract The peer ingestion merge still treated every incoming `site-repl-*` id as reconciler-owned, so an owner-authored `site-repl-user` rule that the S3 merge now keeps on the editing site was dropped on every peer and the sites persisted different operator configs. The ingestion merge now classifies by the same derived contract as the S3 merge: a rule is the reconciler's only when its `site-repl-<id>` names the deployment its destination ARN targets and that deployment is a site of the cluster (the receiver's own id included, since the sender's rule towards the receiver names it). The reconciler, the peer-removal prune and the target-online probe switch from the id prefix to the derived shape as well, so the rule survives their passes too; rules in the derived shape that name a removed peer or this site are still rebuilt away. Regression: a PutBucketReplication merged on site A and ingested on site B keeps `site-repl-user` on both and the operator rule sets agree. * fix(replication): keep an operator role target through site rule merges The S3 and peer-ingestion merges cleared `Role` whenever it parsed as a site-replication ARN, which an owner-submitted remote target with an empty region (`arn:minio:replication::<id>:<bucket>`) also does. The merged config then selected the rule destination ARNs instead of the validated role target. Only a role naming a current site of the cluster is the holder's identity (the reconciler's per-peer target lookup reads it); every other role passed target validation and stays. The reconciler's repair pass applies the same rule. Regression: an owner role target survives both merges and `filter_target_arns` / `replication_target_arns` select it; a role naming a current peer is still cleared. * fix(replication): gate operator priority preservation on a peer contract probe Keeping operator rule priorities verbatim is not rolling-upgrade safe: a peer still running the pre-contract code renumbers every rule 1..n in list order on ingest and on each reconciler pass, so an upgraded site broadcasting `5,1` leaves that peer on `1,2` — which can select the other overlapping rule — and the sites never reconverge. Operator rules now merge under an explicit contract: - `OperatorRuleContract::Derived`: site rules are the derived id/ARN shape, operator priorities stay verbatim (the behavior of the previous commits). - `OperatorRuleContract::Legacy`: byte-for-byte what a pre-contract peer does — `site-repl-*` ids are all site rules, a site-replication-shaped `Role` is dropped, every rule is renumbered 1..n in list order. The S3 merge additionally lists the operator rules in priority order first, so the renumbering keeps their relative order and the winning rule per target is the one the operator submitted. The S3 PutBucketReplication/DeleteBucketReplication path probes every remote peer through the existing `peer/edit-capabilities` endpoint (capability `derived-rule-contract`; pre-contract peers answer `success:false` or 404) and merges under Derived only when every peer supports it; any refusal or probe failure pins that edit to Legacy. Every bucket-meta item this site sends (S3 hooks, bootstrap plan, retry snapshots, tombstones) carries `derivedRuleContract: true`; a receiver merges a payload without the marker the Legacy way, so an item from a pre-contract sender is handled exactly as its own peers handle it. Rolling upgrade: while any site runs the older code every edit is canonicalized cluster-wide (numbers lost, order kept); once the last site is upgraded the next edit keeps its priorities. Configs canonicalized during the mixed period are not renumbered back — the derived priority assignment is a no-op on the canonical layout — so an operator who wants the original values re-submits the config after the upgrade completes. Adding a site that runs the older code after priorities were preserved is not gated and would desynchronize that bucket until the next edit. --------- Co-authored-by: houseme <housemecn@gmail.com>
2026-08-26 05:56:50 +00:00 · 2026-08-23 16:42:30 +08:00
parent dc8177c2b8
commit 4ddc728c9d
11 changed files with 1291 additions and 158 deletions
@@ -187,22 +187,24 @@ pub mod bucket {
        pub use crate::bucket::replication::{
            BucketReplicationResyncStatus, BucketReplicationStat, BucketReplicationStats, BucketStats,
            DeleteReplicationConfigSnapshot, DeletedObjectReplicationInfo, DurableMrfBacklog, DynReplicationPool, InQueueMetric,
-            MrfOpKind, MrfReplicateEntry, MustReplicateOptions, ObjectOpts, REMOTE_TARGET_CAPABILITY_CONTRACT_VERSION,
-            REMOTE_TARGET_UNSUPPORTED_FIELDS, REMOTE_TARGET_WRITABLE_FIELDS, REPLICATE_INCOMING_DELETE,
-            REPLICATION_CAPABILITY_CONTRACT_VERSION, REPLICATION_READ_ONLY_HISTORICAL_FIELDS, REPLICATION_WRITABLE_FIELDS,
-            ReplicateDecision, ReplicateObjectInfo, ReplicationBatchAdmission, ReplicationConfig,
+            MrfOpKind, MrfReplicateEntry, MustReplicateOptions, ObjectOpts, OperatorRuleContract,
+            REMOTE_TARGET_CAPABILITY_CONTRACT_VERSION, REMOTE_TARGET_UNSUPPORTED_FIELDS, REMOTE_TARGET_WRITABLE_FIELDS,
+            REPLICATE_INCOMING_DELETE, REPLICATION_CAPABILITY_CONTRACT_VERSION, REPLICATION_READ_ONLY_HISTORICAL_FIELDS,
+            REPLICATION_WRITABLE_FIELDS, ReplicateDecision, ReplicateObjectInfo, ReplicationBatchAdmission, ReplicationConfig,
            ReplicationConfigStructureError, ReplicationConfigurationExt, ReplicationDeleteScheduleInput,
            ReplicationDeleteStateSource, ReplicationHealQueueResult, ReplicationObjectBridge, ReplicationObjectIO,
            ReplicationOperation, ReplicationPoolTrait, ReplicationPriority, ReplicationQueueAdmission, ReplicationScannerBridge,
            ReplicationState, ReplicationStats, ReplicationStatusType, ReplicationStorage, ReplicationTargetValidationError,
            ReplicationType, ResyncOpts, ResyncStatusType, RuntimeReplicationTargetBacklog, TargetReplicationResyncStatus,
-            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,
+            VersionPurgeStatusType, XferStats, assign_site_replication_rule_priorities, 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, is_site_replication_role, is_site_replication_rule,
+            merge_incoming_replication_config, merge_user_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, site_replication_rule_deployment_id,
            unsupported_replication_config_field, validate_replication_config_structure, validate_replication_config_target_arns,
            version_purge_status_to_filemeta,
        };
@@ -44,11 +44,14 @@ mod replication_versioning_boundary;
 mod runtime_boundary;

 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,
-    unsupported_replication_config_field, validate_replication_config_structure, validate_replication_config_target_arns,
+    ObjectOpts, OperatorRuleContract, 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,
+    assign_site_replication_rule_priorities, invalid_replication_config_status_field, is_site_replication_role,
+    is_site_replication_rule, merge_incoming_replication_config, merge_user_replication_config,
+    replication_target_arn_deployment_id, replication_target_arns, should_remove_replication_target,
+    site_replication_rule_deployment_id, unsupported_replication_config_field, validate_replication_config_structure,
+    validate_replication_config_target_arns,
 };
 pub(crate) use replication_filemeta_boundary::version_purge_statuses_map;
 pub use replication_filemeta_boundary::{
@@ -13,9 +13,12 @@
 // limitations under the License.

 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,
-    unsupported_replication_config_field, validate_replication_config_structure, validate_replication_config_target_arns,
+    ObjectOpts, OperatorRuleContract, 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, assign_site_replication_rule_priorities, invalid_replication_config_status_field,
+    is_site_replication_role, is_site_replication_rule, merge_incoming_replication_config, merge_user_replication_config,
+    replication_target_arn_deployment_id, replication_target_arns, should_remove_replication_target,
+    site_replication_rule_deployment_id, unsupported_replication_config_field, validate_replication_config_structure,
+    validate_replication_config_target_arns,
 };
@@ -431,6 +431,12 @@ pub struct SRBucketMeta {
    pub cors: Option<String>,
    #[serde(rename = "apiVersion", skip_serializing_if = "Option::is_none")]
    pub api_version: Option<String>,
+    /// Set by a sender that merges replication configs under the derived
+    /// site-rule contract (operator rule priorities verbatim, `site-repl-*`
+    /// ids classified by id/ARN). A receiver merges a payload without it the
+    /// pre-contract way; a pre-contract receiver ignores the field.
+    #[serde(rename = "derivedRuleContract", default, skip_serializing_if = "std::ops::Not::not")]
+    pub derived_rule_contract: bool,
 }

 #[derive(Debug, Clone, Serialize, Deserialize, Default)]
@@ -270,6 +270,218 @@ 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
+}
+
+/// Rule id prefix the site-replication reconciler stamps on the rules it
+/// derives (`site-repl-<peer deployment id>`).
+pub const SITE_REPLICATION_RULE_ID_PREFIX: &str = "site-repl-";
+
+/// Whether `rule` carries a site-replication rule id (`site-repl-*`). Rule
+/// ids are not reserved, so this is only the classification of
+/// [`OperatorRuleContract::Legacy`]; every other path classifies by
+/// [`site_replication_rule_deployment_id`].
+pub fn is_site_replication_rule(rule: &ReplicationRule) -> bool {
+    rule.id
+        .as_deref()
+        .is_some_and(|id| id.starts_with(SITE_REPLICATION_RULE_ID_PREFIX))
+}
+
+/// Deployment id of the peer a reconciler-derived rule replicates to, or
+/// `None` for any other rule. The reconciler builds each rule from one peer:
+/// the id is `site-repl-<deployment id>` and the destination ARN names that
+/// same deployment id — an operator-authored `site-repl-user` rule, or a
+/// `site-repl-<peer>` id pasted onto a foreign ARN, fails the agreement check.
+/// Callers that know the current peer set must also confirm the id is one of
+/// those peers before treating the rule as reconciler-owned.
+pub fn site_replication_rule_deployment_id(rule: &ReplicationRule) -> Option<&str> {
+    let deployment_id = rule.id.as_deref()?.strip_prefix(SITE_REPLICATION_RULE_ID_PREFIX)?;
+    (!deployment_id.is_empty()
+        && replication_target_arn_deployment_id(&rule.destination.bucket).as_deref() == Some(deployment_id))
+    .then_some(deployment_id)
+}
+
+/// Whether `rule` is one the local reconciler derived for a current remote
+/// site-replication peer in `peer_deployment_ids`. With an empty peer set
+/// (site replication disabled) nothing qualifies, so a bucket outside site
+/// replication keeps the verbatim S3 put/delete semantics.
+pub fn is_reconciler_owned_site_replication_rule(rule: &ReplicationRule, peer_deployment_ids: &HashSet<String>) -> bool {
+    site_replication_rule_deployment_id(rule).is_some_and(|deployment_id| peer_deployment_ids.contains(deployment_id))
+}
+
+/// Whether a config's `Role` is a site-replication ARN naming a site in
+/// `deployment_ids`. Such a role is the holder's identity, not policy: the
+/// reconciler's per-peer target lookup reads it, so carrying it across sites
+/// would pin the receiver's targets to the sender's. Any other role — an IAM
+/// role, or an operator remote target whose ARN happens to carry an empty
+/// region — passed target validation and drives target selection.
+pub fn is_site_replication_role(role: &str, deployment_ids: &HashSet<String>) -> bool {
+    replication_target_arn_deployment_id(role).is_some_and(|deployment_id| deployment_ids.contains(&deployment_id))
+}
+
+/// How the sites of a cluster treat the operator rules of a replication
+/// config merge. Every site must apply the same contract to the same
+/// payload or the sites persist different configs, so the S3 edit path
+/// probes the peers before merging and a peer payload carries the contract
+/// its sender applied.
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub enum OperatorRuleContract {
+    /// Site rules are the derived id/ARN shape; operator rule priorities are
+    /// kept verbatim.
+    Derived,
+    /// Some site still runs the pre-contract code: every `site-repl-*` id is
+    /// a site rule, a site-replication-shaped `Role` is dropped, and every
+    /// rule is renumbered 1..n in list order on ingest and on each reconciler
+    /// pass. Merging the same way keeps a mixed cluster on one config; the
+    /// operator's priority values are lost for that edit but their order —
+    /// what decides the winning rule per target — is not, because the S3
+    /// merge lists the operator rules in priority order first.
+    Legacy,
+}
+
+/// Merge a peer's replication config into the local one.
+///
+/// Reconciler-derived 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 sender's derived rules are dropped
+/// and the local site's survive. `site_deployment_ids` is every site of the
+/// cluster, the receiver included — the sender's rule towards the receiver
+/// names the receiver's own id. Rules are classified by the derived id/ARN
+/// contract ([`is_reconciler_owned_site_replication_rule`]), the same one
+/// the S3 edit merge applies, so an operator-authored `site-repl-*` id
+/// persists on every site. `incoming == None` models a delete of the
+/// operator-authored rules.
+pub fn merge_incoming_replication_config(
+    incoming: Option<ReplicationConfiguration>,
+    local: Option<ReplicationConfiguration>,
+    site_deployment_ids: &HashSet<String>,
+    contract: OperatorRuleContract,
+) -> Option<ReplicationConfiguration> {
+    merge_replication_config_keeping_site_rules(incoming, local, site_deployment_ids, contract)
+}
+
+/// [`merge_incoming_replication_config`] for the S3 put/delete-bucket-replication
+/// path (issue #1948): only rules the local reconciler derived for a current
+/// peer in `peer_deployment_ids` survive as site rules; every other stored
+/// rule — including an operator-authored `site-repl-*` id — is operator state
+/// that the request replaces or deletes. An incoming rule whose id is a
+/// current peer's `site-repl-<id>` is dropped whatever its ARN: accepting it
+/// would duplicate the reconciler rule's id. Under
+/// [`OperatorRuleContract::Legacy`] the merge instead reproduces what the
+/// pre-contract peers will do with the broadcast, listing the operator rules
+/// in priority order so their relative order survives the renumbering.
+pub fn merge_user_replication_config(
+    incoming: Option<ReplicationConfiguration>,
+    local: Option<ReplicationConfiguration>,
+    peer_deployment_ids: &HashSet<String>,
+    contract: OperatorRuleContract,
+) -> Option<ReplicationConfiguration> {
+    let incoming = incoming.map(|mut config| {
+        match contract {
+            OperatorRuleContract::Derived => config.rules.retain(|rule| {
+                !rule
+                    .id
+                    .as_deref()
+                    .and_then(|id| id.strip_prefix(SITE_REPLICATION_RULE_ID_PREFIX))
+                    .is_some_and(|deployment_id| peer_deployment_ids.contains(deployment_id))
+            }),
+            // Pre-contract peers renumber in list order, so listing the
+            // operator rules in priority order keeps their relative order —
+            // and the replication decision — through that renumbering.
+            OperatorRuleContract::Legacy => config.rules.sort_by_key(|rule| rule.priority.unwrap_or(0)),
+        }
+        config
+    });
+    merge_replication_config_keeping_site_rules(incoming, local, peer_deployment_ids, contract)
+}
+
+fn merge_replication_config_keeping_site_rules(
+    incoming: Option<ReplicationConfiguration>,
+    local: Option<ReplicationConfiguration>,
+    deployment_ids: &HashSet<String>,
+    contract: OperatorRuleContract,
+) -> Option<ReplicationConfiguration> {
+    let is_site_rule = |rule: &ReplicationRule| match contract {
+        OperatorRuleContract::Derived => is_reconciler_owned_site_replication_rule(rule, deployment_ids),
+        OperatorRuleContract::Legacy => is_site_replication_rule(rule),
+    };
+    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_rule(rule))
+        .collect();
+    rules.extend(
+        local
+            .into_iter()
+            .flat_map(|config| config.rules)
+            .filter(|rule| is_site_rule(rule)),
+    );
+
+    if rules.is_empty() {
+        return None;
+    }
+
+    let drop_role = match contract {
+        OperatorRuleContract::Derived => {
+            assign_site_replication_rule_priorities(&mut rules, is_site_rule);
+            is_site_replication_role(&incoming_role, deployment_ids)
+        }
+        OperatorRuleContract::Legacy => {
+            for (index, rule) in rules.iter_mut().enumerate() {
+                rule.priority = Some(i32::try_from(index + 1).unwrap_or(i32::MAX));
+            }
+            replication_target_arn_deployment_id(&incoming_role).is_some()
+        }
+    };
+    let role = if drop_role { String::new() } else { incoming_role };
+
+    Some(ReplicationConfiguration { role, rules })
+}
+
+/// Give the site rules in `rules` the lowest priorities no operator rule uses,
+/// in rule order, leaving every operator rule's priority untouched. Operator
+/// priorities decide which rule wins per target, so they are part of the
+/// submitted policy; site rules are derived state and only need to be unique
+/// (`validate_replication_config_structure` rejects duplicates). The result
+/// is a pure function of the rule list, so the site-replication reconciler,
+/// the peer ingestion merge and the S3 edit merge all converge on the same
+/// bytes and the reconciler's no-op check holds.
+pub fn assign_site_replication_rule_priorities(rules: &mut [ReplicationRule], is_site_rule: impl Fn(&ReplicationRule) -> bool) {
+    let taken: HashSet<i32> = rules
+        .iter()
+        .filter(|rule| !is_site_rule(rule))
+        .map(|rule| rule.priority.unwrap_or(0))
+        .collect();
+    let mut next = 1;
+    for rule in rules.iter_mut().filter(|rule| is_site_rule(rule)) {
+        while taken.contains(&next) {
+            next += 1;
+        }
+        rule.priority = Some(next);
+        next = next.saturating_add(1);
+    }
+}
+
 pub fn replication_target_arns(config: &ReplicationConfiguration) -> HashSet<String> {
    let role = config.role.trim();
    if !role.is_empty() {
@@ -1544,4 +1756,276 @@ mod tests {
            "the child rule must win for target A while the overlapping child target B remains eligible"
        );
    }
+
+    #[test]
+    fn site_replication_rule_deployment_id_requires_id_and_arn_agreement() {
+        let reconciler_rule = replication_rule("site-repl-peer-dep", "arn:rustfs:replication::peer-dep:bucket");
+        assert_eq!(site_replication_rule_deployment_id(&reconciler_rule), Some("peer-dep"));
+
+        // A remote-target ARN carries the remote's deployment id (or a random
+        // uuid), never the operator's rule id.
+        let operator_named_rule = replication_rule("site-repl-user", "arn:minio:replication:us-east-1:2f1c-remote:bucket");
+        assert_eq!(site_replication_rule_deployment_id(&operator_named_rule), None);
+
+        let foreign_arn = replication_rule("site-repl-peer-dep", "arn:rustfs:replication::other-dep:bucket");
+        assert_eq!(site_replication_rule_deployment_id(&foreign_arn), None);
+
+        let empty_id = replication_rule("site-repl-", "arn:rustfs:replication::peer-dep:bucket");
+        assert_eq!(site_replication_rule_deployment_id(&empty_id), None);
+
+        let peers = HashSet::from(["peer-dep".to_string()]);
+        assert!(is_reconciler_owned_site_replication_rule(&reconciler_rule, &peers));
+        assert!(!is_reconciler_owned_site_replication_rule(&reconciler_rule, &HashSet::new()));
+        let removed_peer = replication_rule("site-repl-gone-dep", "arn:rustfs:replication::gone-dep:bucket");
+        assert!(!is_reconciler_owned_site_replication_rule(&removed_peer, &peers));
+    }
+
+    // The merge must not rewrite the operator's priorities: with the
+    // priority-5 rule listed first and renumbered 1 then 2, the priority-1
+    // delete-marker-disabled rule would win the replication decision.
+    #[test]
+    fn merge_keeps_operator_priorities_and_replication_decision() {
+        let user_arn = "arn:minio:replication:us-east-1:2f1c-remote:bucket";
+        let peer_arn = "arn:rustfs:replication::peer-dep:bucket";
+        let incoming = ReplicationConfiguration {
+            role: String::new(),
+            rules: vec![
+                delete_marker_rule("dm-enabled", user_arn, "logs/", 5, true),
+                delete_marker_rule("dm-disabled", user_arn, "logs/2026/", 1, false),
+            ],
+        };
+        let mut site_rule = delete_marker_rule("site-repl-peer-dep", peer_arn, "", 7, true);
+        site_rule.prefix = None;
+        let local = structure_config(vec![site_rule]);
+        let opts = ObjectOpts {
+            name: "logs/2026/app.log".to_string(),
+            op_type: ReplicationType::Delete,
+            delete_marker: true,
+            version_id: None,
+            ..Default::default()
+        };
+        let submitted: Vec<_> = incoming.filter_target_replication_decisions(&opts);
+
+        let peers = HashSet::from(["peer-dep".to_string()]);
+        let merged =
+            merge_user_replication_config(Some(incoming.clone()), Some(local.clone()), &peers, OperatorRuleContract::Derived)
+                .expect("rules");
+
+        let priorities: Vec<_> = merged
+            .rules
+            .iter()
+            .map(|rule| (rule.id.as_deref().unwrap(), rule.priority))
+            .collect();
+        assert_eq!(
+            priorities,
+            vec![
+                ("dm-enabled", Some(5)),
+                ("dm-disabled", Some(1)),
+                ("site-repl-peer-dep", Some(2))
+            ],
+            "operator priorities are kept verbatim; the site rule takes the lowest free slot"
+        );
+        assert!(validate_replication_config_structure(&merged).is_ok());
+        let mut decisions = merged.filter_target_replication_decisions(&opts);
+        decisions.retain(|(arn, _)| arn == user_arn);
+        assert_eq!(decisions, submitted, "the merged config must replicate exactly as the operator submitted");
+        assert_eq!(decisions, vec![(user_arn.to_string(), true)]);
+
+        // The peer ingestion merge follows the same rule.
+        let merged =
+            merge_incoming_replication_config(Some(incoming), Some(local), &peers, OperatorRuleContract::Derived).expect("rules");
+        let priorities: Vec<_> = merged.rules.iter().map(|rule| rule.priority).collect();
+        assert_eq!(priorities, vec![Some(5), Some(1), Some(2)]);
+    }
+
+    #[test]
+    fn site_rule_priorities_skip_every_operator_priority() {
+        let mut rules = vec![
+            delete_marker_rule("a", "arn:a", "", 2, true),
+            delete_marker_rule("site-repl-x", "arn:rustfs:replication::x:b", "", 9, true),
+            delete_marker_rule("b", "arn:a", "", 1, true),
+            delete_marker_rule("site-repl-y", "arn:rustfs:replication::y:b", "", 9, true),
+            delete_marker_rule("c", "arn:a", "", 4, true),
+        ];
+        assign_site_replication_rule_priorities(&mut rules, is_site_replication_rule);
+        let priorities: Vec<_> = rules.iter().map(|rule| rule.priority).collect();
+        assert_eq!(priorities, vec![Some(2), Some(3), Some(1), Some(5), Some(4)]);
+        assert!(validate_replication_config_structure(&structure_config(rules.clone())).is_ok());
+
+        // Idempotent, so the reconciler's pass over an already-merged config
+        // is a byte-stable no-op rather than a rewrite every period.
+        let settled = rules.clone();
+        assign_site_replication_rule_priorities(&mut rules, is_site_replication_rule);
+        assert_eq!(rules, settled);
+    }
+
+    fn operator_rule_ids(config: &ReplicationConfiguration) -> Vec<(&str, Option<i32>)> {
+        config
+            .rules
+            .iter()
+            .filter(|rule| site_replication_rule_deployment_id(rule).is_none())
+            .map(|rule| (rule.id.as_deref().unwrap(), rule.priority))
+            .collect()
+    }
+
+    // Issue #1948 review: an owner-authored `site-repl-user` rule is operator
+    // state. Site A's S3 merge keeps it; the broadcast payload must survive
+    // site B's peer ingestion too, or the sites persist different configs.
+    #[test]
+    fn peer_ingestion_keeps_owner_site_repl_user_rule_and_sites_agree() {
+        let user_arn = "arn:minio:replication:us-east-1:2f1c-remote:bucket";
+        let put = structure_config(vec![
+            delete_marker_rule("site-repl-user", user_arn, "logs/", 3, true),
+            delete_marker_rule("nightly", user_arn, "", 1, true),
+        ]);
+        let a_local = structure_config(vec![replication_rule("site-repl-b-dep", "arn:rustfs:replication::b-dep:bucket")]);
+        let a_peers = HashSet::from(["b-dep".to_string()]);
+        let a_merged =
+            merge_user_replication_config(Some(put), Some(a_local), &a_peers, OperatorRuleContract::Derived).expect("rules");
+        assert_eq!(operator_rule_ids(&a_merged), vec![("site-repl-user", Some(3)), ("nightly", Some(1))]);
+
+        // Site B ingests A's broadcast; its own reverse rule names A.
+        let b_local = structure_config(vec![replication_rule("site-repl-a-dep", "arn:rustfs:replication::a-dep:bucket")]);
+        let b_sites = HashSet::from(["a-dep".to_string(), "b-dep".to_string()]);
+        let b_merged =
+            merge_incoming_replication_config(Some(a_merged.clone()), Some(b_local), &b_sites, OperatorRuleContract::Derived)
+                .expect("rules");
+
+        let ids: Vec<_> = b_merged.rules.iter().map(|rule| rule.id.as_deref().unwrap()).collect();
+        assert_eq!(ids, vec!["site-repl-user", "nightly", "site-repl-a-dep"]);
+        assert_eq!(
+            operator_rule_ids(&b_merged),
+            operator_rule_ids(&a_merged),
+            "both sites must persist the same operator rules"
+        );
+    }
+
+    // Issue #1948 review: `Role` is only the sender's when it names a
+    // current site-replication peer; an owner-submitted role target has
+    // already passed target validation and drives target selection.
+    #[test]
+    fn merge_keeps_operator_role_target_for_target_selection() {
+        let role = "arn:minio:replication::operator-dep:bucket";
+        let peers = HashSet::from(["peer-dep".to_string()]);
+        let incoming = ReplicationConfiguration {
+            role: role.to_string(),
+            rules: vec![delete_marker_rule("nightly", role, "", 1, true)],
+        };
+        let local = structure_config(vec![replication_rule(
+            "site-repl-peer-dep",
+            "arn:rustfs:replication::peer-dep:bucket",
+        )]);
+        let opts = ObjectOpts {
+            name: "logs/app.log".to_string(),
+            ..Default::default()
+        };
+
+        let merged =
+            merge_user_replication_config(Some(incoming.clone()), Some(local.clone()), &peers, OperatorRuleContract::Derived)
+                .expect("rules");
+        assert_eq!(merged.role, role);
+        assert_eq!(replication_target_arns(&merged), HashSet::from([role.to_string()]));
+        assert_eq!(merged.filter_target_arns(&opts), vec![role.to_string()]);
+
+        let ingested =
+            merge_incoming_replication_config(Some(incoming.clone()), Some(local.clone()), &peers, OperatorRuleContract::Derived)
+                .expect("rules");
+        assert_eq!(ingested.role, role);
+        assert_eq!(ingested.filter_target_arns(&opts), vec![role.to_string()]);
+
+        // A role naming a current peer is the sender's identity and still goes.
+        let mut derived_role = incoming;
+        derived_role.role = "arn:rustfs:replication::peer-dep:bucket".to_string();
+        let merged =
+            merge_user_replication_config(Some(derived_role), Some(local), &peers, OperatorRuleContract::Derived).expect("rules");
+        assert!(merged.role.is_empty());
+    }
+
+    // Issue #1948 review: while a site still runs the pre-contract code the
+    // cluster must stay on one config. A new site broadcasting `5,1` would
+    // be renumbered `1,2` by that peer — selecting the other overlapping
+    // rule — so the new sites merge the legacy way and list the operator
+    // rules in priority order first, which keeps the decision.
+    #[test]
+    fn legacy_contract_matches_pre_contract_peers_and_keeps_the_decision() {
+        let user_arn = "arn:minio:replication:us-east-1:2f1c-remote:bucket";
+        let put = ReplicationConfiguration {
+            role: "arn:minio:replication::operator-dep:bucket".to_string(),
+            rules: vec![
+                delete_marker_rule("dm-enabled", user_arn, "logs/", 5, true),
+                delete_marker_rule("dm-disabled", user_arn, "logs/2026/", 1, false),
+                delete_marker_rule("site-repl-user", user_arn, "tmp/", 2, true),
+            ],
+        };
+        let a_local = structure_config(vec![replication_rule("site-repl-b-dep", "arn:rustfs:replication::b-dep:bucket")]);
+        let opts = ObjectOpts {
+            name: "logs/2026/app.log".to_string(),
+            op_type: ReplicationType::Delete,
+            delete_marker: true,
+            ..Default::default()
+        };
+        // Decisions per rule destination: the role is dropped by the legacy
+        // merge, so compare against the rules alone.
+        let submitted: Vec<_> = structure_config(put.rules.clone()).filter_target_replication_decisions(&opts);
+
+        let a_peers = HashSet::from(["b-dep".to_string()]);
+        let a_merged =
+            merge_user_replication_config(Some(put.clone()), Some(a_local.clone()), &a_peers, OperatorRuleContract::Legacy)
+                .expect("rules");
+        let layout: Vec<_> = a_merged
+            .rules
+            .iter()
+            .map(|rule| (rule.id.as_deref().unwrap(), rule.priority))
+            .collect();
+        assert_eq!(
+            layout,
+            vec![
+                ("dm-disabled", Some(1)),
+                ("dm-enabled", Some(2)),
+                ("site-repl-b-dep", Some(3))
+            ],
+            "legacy: operator rules in priority order, every rule renumbered 1..n, `site-repl-*` ids dropped"
+        );
+        assert!(a_merged.role.is_empty(), "legacy peers drop any site-replication-shaped role");
+        let mut decisions = a_merged.filter_target_replication_decisions(&opts);
+        decisions.retain(|(arn, _)| arn == user_arn);
+        assert_eq!(decisions, submitted, "the renumbering must not flip the winning rule");
+
+        // A pre-contract peer renumbers A's payload in list order: same bytes.
+        let mut pre_contract = a_merged
+            .rules
+            .iter()
+            .filter(|rule| !is_site_replication_rule(rule))
+            .cloned()
+            .collect::<Vec<_>>();
+        pre_contract.push(replication_rule("site-repl-a-dep", "arn:rustfs:replication::a-dep:bucket"));
+        for (index, rule) in pre_contract.iter_mut().enumerate() {
+            rule.priority = Some(index as i32 + 1);
+        }
+        // A new peer told the payload is legacy produces the same bytes too.
+        let b_local = structure_config(vec![replication_rule("site-repl-a-dep", "arn:rustfs:replication::a-dep:bucket")]);
+        let b_sites = HashSet::from(["a-dep".to_string(), "b-dep".to_string()]);
+        let b_merged = merge_incoming_replication_config(Some(a_merged), Some(b_local), &b_sites, OperatorRuleContract::Legacy)
+            .expect("rules");
+        assert_eq!(b_merged.rules, pre_contract);
+
+        // Every site on the derived contract: the submitted policy is kept.
+        let a_merged =
+            merge_user_replication_config(Some(put), Some(a_local), &a_peers, OperatorRuleContract::Derived).expect("rules");
+        let layout: Vec<_> = a_merged
+            .rules
+            .iter()
+            .map(|rule| (rule.id.as_deref().unwrap(), rule.priority))
+            .collect();
+        assert_eq!(
+            layout,
+            vec![
+                ("dm-enabled", Some(5)),
+                ("dm-disabled", Some(1)),
+                ("site-repl-user", Some(2)),
+                ("site-repl-b-dep", Some(3))
+            ]
+        );
+        assert_eq!(a_merged.role, "arn:minio:replication::operator-dep:bucket");
+    }
 }
@@ -29,12 +29,14 @@ mod storage_api;
 pub mod tagging;

 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,
-    should_remove_replication_target, unsupported_replication_config_field, validate_replication_config_structure,
-    validate_replication_config_target_arns,
+    ObjectOpts, OperatorRuleContract, 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, assign_site_replication_rule_priorities, invalid_replication_config_status_field,
+    is_reconciler_owned_site_replication_rule, is_site_replication_role, is_site_replication_rule,
+    merge_incoming_replication_config, merge_user_replication_config, replication_target_arn_deployment_id,
+    replication_target_arns, should_remove_replication_target, site_replication_rule_deployment_id,
+    unsupported_replication_config_field, validate_replication_config_structure, validate_replication_config_target_arns,
 };
 pub use delete::{
    DeletedObjectReplicationInfo, delete_marker_purge_mrf_entry, delete_marker_purge_version_id,
@@ -31,6 +31,10 @@ 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::{
+    OperatorRuleContract, assign_site_replication_rule_priorities, is_site_replication_role, merge_incoming_replication_config,
+    replication_target_arn_deployment_id, site_replication_rule_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};
@@ -160,6 +164,8 @@ const SITE_REPLICATION_PEER_EDIT_CAPABILITY_PATH: &str =
    "/rustfs/admin/v3/site-replication/peer/edit-capabilities?capability=endpoint-target-refresh";
 const SITE_REPLICATION_PEER_TLS_CAPABILITY_PATH: &str =
    "/rustfs/admin/v3/site-replication/peer/edit-capabilities?capability=peer-tls-settings";
+const SITE_REPLICATION_PEER_DERIVED_RULE_CONTRACT_CAPABILITY_PATH: &str =
+    "/rustfs/admin/v3/site-replication/peer/edit-capabilities?capability=derived-rule-contract";
 const SITE_REPLICATION_PEER_EDIT_REFRESH_PATH: &str = "/rustfs/admin/v3/site-replication/peer/edit?refresh-targets=true";
 /// Peer-edit fencing token, carried as query parameters so a peer that predates
 /// the fence simply ignores them (unknown query keys are dropped) and keeps the
@@ -1118,6 +1124,114 @@ async fn load_site_replication_state() -> S3Result<SiteReplicationState> {
    }
 }

+/// 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<bool> {
+    Ok(load_site_replication_state().await?.enabled())
+}
+
+/// Deployment ids of the remote peers the reconciler derives a
+/// `site-repl-<id>` rule for on every bucket (the same peer filter as
+/// `build_site_replication_config`); empty when site replication is not
+/// enabled. Read by the bucket usecase so an S3 replication-config edit keeps
+/// exactly the reconciler-owned rules (issue #1948); a state-read failure
+/// propagates so the edit fails closed.
+pub(crate) async fn site_replication_edit_context() -> S3Result<(HashSet<String>, OperatorRuleContract)> {
+    let Some(runtime) = runtime_site_replication_targets().await? else {
+        // Enabled without a service account is a state this site cannot
+        // broadcast from either; the peers are still the reconciler's.
+        let state = load_site_replication_state().await?;
+        if !state.enabled() {
+            return Ok((HashSet::new(), OperatorRuleContract::Derived));
+        }
+        let peers = remote_peer_deployment_ids(&state, &current_local_runtime_peer(&state));
+        return Ok((peers, OperatorRuleContract::Legacy));
+    };
+    let peers = remote_peer_deployment_ids(&runtime.state, &runtime.local_peer);
+    let contract = site_replication_operator_rule_contract(&runtime).await;
+    Ok((peers, contract))
+}
+
+/// Whether every remote peer merges replication configs under the derived
+/// contract, probed through the peer capability endpoint. A peer that does
+/// not (or cannot be asked) pins the cluster to [`OperatorRuleContract::Legacy`]
+/// for this edit: consistency across sites wins over keeping the operator's
+/// priority values, and the legacy merge keeps their order anyway.
+async fn site_replication_operator_rule_contract(runtime: &SiteReplicationRuntime) -> OperatorRuleContract {
+    let remote_peers: Vec<&PeerInfo> = runtime
+        .state
+        .peers
+        .values()
+        .filter(|peer| {
+            peer.deployment_id != runtime.local_peer.deployment_id
+                && !same_identity_endpoint(&peer.endpoint, &runtime.local_peer.endpoint)
+        })
+        .collect();
+    let probes = futures::future::join_all(remote_peers.iter().map(|peer| async move {
+        let transport = PeerTransport::for_runtime_peer(peer).await?;
+        let (status, body) = send_peer_admin_request_raw_with_client(
+            &transport.client,
+            &transport.connection,
+            SITE_REPLICATION_PEER_DERIVED_RULE_CONTRACT_CAPABILITY_PATH,
+            &runtime.state.service_account_access_key,
+            &runtime.service_account_secret_key,
+            &(),
+        )
+        .await?;
+        peer_capability_response_supported(peer, status, &body)
+    }))
+    .await;
+    operator_rule_contract_from_probes(remote_peers.into_iter().zip(probes))
+}
+
+fn operator_rule_contract_from_probes<'a>(
+    probes: impl IntoIterator<Item = (&'a PeerInfo, S3Result<bool>)>,
+) -> OperatorRuleContract {
+    for (peer, probe) in probes {
+        match probe {
+            Ok(true) => {}
+            Ok(false) => return OperatorRuleContract::Legacy,
+            Err(err) => {
+                warn!(
+                    event = EVENT_ADMIN_SITE_REPLICATION_STATE,
+                    component = LOG_COMPONENT_ADMIN,
+                    subsystem = LOG_SUBSYSTEM_SITE_REPLICATION,
+                    result = "derived_rule_contract_probe_failed",
+                    peer = %peer.endpoint,
+                    error = %err,
+                    "admin site replication state"
+                );
+                return OperatorRuleContract::Legacy;
+            }
+        }
+    }
+    OperatorRuleContract::Derived
+}
+
+fn remote_peer_deployment_ids(state: &SiteReplicationState, local_peer: &PeerInfo) -> HashSet<String> {
+    state
+        .peers
+        .values()
+        .filter(|peer| {
+            peer.deployment_id != local_peer.deployment_id && !same_identity_endpoint(&peer.endpoint, &local_peer.endpoint)
+        })
+        .map(|peer| peer.deployment_id.clone())
+        .collect()
+}
+
+/// Deployment ids of every site in the cluster, this one included: the set
+/// a peer's derived rules can name (its rule towards this site carries this
+/// site's id). Empty when site replication is not enabled.
+async fn site_replication_deployment_ids() -> S3Result<HashSet<String>> {
+    let state = load_site_replication_state().await?;
+    if !state.enabled() {
+        return Ok(HashSet::new());
+    }
+    Ok(state.peers.values().map(|peer| peer.deployment_id.clone()).collect())
+}
+
 async fn load_site_replication_state_no_lock(store: Arc<ECStore>) -> S3Result<SiteReplicationState> {
    match read_config_no_lock(store, SITE_REPLICATION_STATE_PATH).await {
        Ok(data) => parse_site_replication_state(&data),
@@ -1984,7 +2098,7 @@ fn peer_tls_settings_changed(existing: Option<&PeerInfo>, proposed: &PeerInfo) -
 }

 fn peer_edit_capability_supported(capability: &str) -> bool {
-    matches!(capability, "endpoint-target-refresh" | "peer-tls-settings")
+    matches!(capability, "endpoint-target-refresh" | "peer-tls-settings" | "derived-rule-contract")
 }

 fn validate_add_sites(sites: &[PeerSite], local_peer: &PeerInfo) -> S3Result<()> {
@@ -2251,6 +2365,7 @@ fn bootstrap_bucket_meta_item(bucket: &SRBucketInfo, item_type: &str, updated_at
        r#type: item_type.to_string(),
        updated_at,
        api_version: Some(SITE_REPL_API_VERSION.to_string()),
+        derived_rule_contract: true,
        ..Default::default()
    }
 }
@@ -6632,6 +6747,7 @@ fn bucket_metadata_snapshot_tombstone(item: &SRBucketMeta, observed_at: OffsetDa
        updated_at: Some(observed_at),
        expiry_updated_at: Some(observed_at),
        api_version: item.api_version.clone(),
+        derived_rule_contract: item.derived_rule_contract,
        ..Default::default()
    }
 }
@@ -7762,20 +7878,6 @@ fn bucket_target_deployment_id(target: &BucketTarget) -> Option<String> {
    replication_target_arn_deployment_id(&target.arn)
 }

-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
-}
-
 fn prune_removed_site_replication_bucket_targets(
    existing: BucketTargets,
    removed_deployment_ids: &HashSet<String>,
@@ -7800,10 +7902,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
@@ -7816,7 +7914,7 @@ async fn site_replication_targets_online(bucket: &str, replication_config_xml: &
        return true;
    };

-    for rule in config.rules.iter().filter(|rule| is_site_replication_rule(rule)) {
+    for rule in config.rules.iter().filter(|rule| is_derived_site_replication_rule(rule)) {
        if BucketTargetSys::get()
            .get_remote_target_client_by_arn(bucket, &rule.destination.bucket)
            .await
@@ -7829,52 +7927,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<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
-    // `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<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 `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
@@ -8169,6 +8221,17 @@ fn lifecycle_expiry_statement(
    }
 }

+/// Whether `rule` is in the shape the reconciler derives (`site-repl-<id>`
+/// naming the deployment its ARN targets). The reconciler rebuilds every such
+/// rule from the current peer set — current peer or not, so a leftover from a
+/// removed peer or a self-pointing rule is rebuilt away — while the merges
+/// keep only the current peers' rules and treat a leftover as operator state
+/// the edit replaces. An operator-authored `site-repl-*` id on an operator
+/// ARN is outside the shape and survives every pass.
+fn is_derived_site_replication_rule(rule: &ReplicationRule) -> bool {
+    site_replication_rule_deployment_id(rule).is_some()
+}
+
 fn replication_rule_deployment_id(rule: &ReplicationRule) -> Option<String> {
    if let Some(rule_id) = rule.id.as_deref() {
        if let Some(deployment_id) = rule_id.strip_prefix("site-repl-")
@@ -8213,9 +8276,7 @@ fn prune_removed_site_replication_rules(
        return (None, removed);
    }

-    for (index, rule) in config.rules.iter_mut().enumerate() {
-        rule.priority = Some(i32::try_from(index + 1).unwrap_or(i32::MAX));
-    }
+    assign_site_replication_rule_priorities(&mut config.rules, is_derived_site_replication_rule);

    (Some(config), removed)
 }
@@ -8375,30 +8436,32 @@ async fn ensure_site_replication_bucket_replication_config_with_runtime(
        return Ok(());
    };

-    // `site-repl-*` rules are derived state owned by this site: rebuild them from the
-    // current peer set on every pass instead of preserving whatever is on disk. A rule
-    // left over from a removed peer — or one whose destination ARN names this very
-    // deployment, which no bucket target can ever satisfy — must not survive, otherwise
-    // objects are queued against an ARN that resolves to nothing.
+    // Derived rules are state owned by this site: rebuild them from the current peer
+    // set on every pass instead of preserving whatever is on disk. A rule left over
+    // from a removed peer — or one whose destination ARN names this very deployment,
+    // which no bucket target can ever satisfy — must not survive, otherwise objects
+    // are queued against an ARN that resolves to nothing.
    let (existing_role, existing_rules) = existing
        .map(|config| (config.role, config.rules))
        .unwrap_or_else(|| (String::new(), Vec::new()));
    let mut rules: Vec<ReplicationRule> = existing_rules
        .iter()
-        .filter(|rule| !is_site_replication_rule(rule))
+        .filter(|rule| !is_derived_site_replication_rule(rule))
        .cloned()
        .collect();
    rules.extend(desired.rules);
-    for (index, rule) in rules.iter_mut().enumerate() {
-        rule.priority = Some(i32::try_from(index + 1).unwrap_or(i32::MAX));
-    }
+    // Operator priorities are the operator's policy; only the derived rules
+    // take free slots, by the same function as the config merges so a merged
+    // write and this pass agree byte for byte.
+    assign_site_replication_rule_priorities(&mut rules, is_derived_site_replication_rule);

-    // Only a site-replication ARN in `role` is ours to drop — an operator-authored role is
+    // Only a `role` naming a current peer is ours to drop — an operator-authored role is
    // part of the bucket's S3-visible configuration, and repairing a reverse rule must not
    // quietly rewrite it. Same rule as `merge_incoming_replication_config`.
-    let role = match replication_target_arn_deployment_id(&existing_role) {
-        Some(_) => String::new(),
-        None => existing_role.clone(),
+    let role = if is_site_replication_role(&existing_role, &remote_peer_deployment_ids(state, local_peer)) {
+        String::new()
+    } else {
+        existing_role.clone()
    };

    if rules == existing_rules && role == existing_role {
@@ -9284,7 +9347,13 @@ async fn apply_bucket_meta_item(item: SRBucketMeta) -> S3Result<()> {
                Err(err) => return Err(ApiError::from(err).into()),
            };
            let local_absent = local.is_none();
-            match merge_incoming_replication_config(incoming, local) {
+            let site_deployment_ids = site_replication_deployment_ids().await?;
+            let contract = if item.derived_rule_contract {
+                OperatorRuleContract::Derived
+            } else {
+                OperatorRuleContract::Legacy
+            };
+            match merge_incoming_replication_config(incoming, local, &site_deployment_ids, contract) {
                Some(config) => Some(serialize(&config).map_err(|e| {
                    S3Error::with_message(S3ErrorCode::InternalError, format!("serialize replication failed: {e}"))
                })?),
@@ -13395,6 +13464,7 @@ mod tests {

        assert!(peer_edit_capability_supported("peer-tls-settings"));
        assert!(peer_edit_capability_supported("endpoint-target-refresh"));
+        assert!(peer_edit_capability_supported("derived-rule-contract"));
        assert!(!peer_edit_capability_supported("unknown"));
        assert!(peer_capability_response_supported(&remote, StatusCode::OK, br#"{"success":true}"#).expect("supported"));
        assert!(!peer_capability_response_supported(&remote, StatusCode::NOT_FOUND, b"").expect("legacy peer"));
@@ -16186,6 +16256,10 @@ mod tests {
        assert_eq!(deployment_id.as_deref(), Some("remote-dep"));
    }

+    fn home_office() -> HashSet<String> {
+        HashSet::from(["home".to_string(), "office".to_string()])
+    }
+
    fn site_repl_config(peer: &str) -> ReplicationConfiguration {
        ReplicationConfiguration {
            role: String::new(),
@@ -16284,8 +16358,13 @@ mod tests {
    // itself. No bucket target backs that ARN, so every object was dropped without a log.
    #[test]
    fn test_merge_incoming_replication_config_keeps_local_reverse_rule() {
-        let merged = merge_incoming_replication_config(Some(site_repl_config("home")), Some(site_repl_config("office")))
-            .expect("merge should keep the local rule");
+        let merged = merge_incoming_replication_config(
+            Some(site_repl_config("home")),
+            Some(site_repl_config("office")),
+            &home_office(),
+            OperatorRuleContract::Derived,
+        )
+        .expect("merge should keep the local rule");

        assert_eq!(merged.rules.len(), 1);
        assert_eq!(merged.rules[0].id.as_deref(), Some("site-repl-office"));
@@ -16296,8 +16375,13 @@ mod tests {
    // either — the delete travels as `replication-config` with no payload.
    #[test]
    fn test_merge_incoming_replication_config_survives_peer_delete() {
-        let merged = merge_incoming_replication_config(None, Some(site_repl_config("office")))
-            .expect("local site rules must survive a peer delete");
+        let merged = merge_incoming_replication_config(
+            None,
+            Some(site_repl_config("office")),
+            &home_office(),
+            OperatorRuleContract::Derived,
+        )
+        .expect("local site rules must survive a peer delete");

        assert_eq!(merged.rules.len(), 1);
        assert_eq!(merged.rules[0].id.as_deref(), Some("site-repl-office"));
@@ -16309,8 +16393,13 @@ mod tests {
        incoming.rules.push(operator_rule("nightly-backup"));
        incoming.role = "arn:rustfs:replication::home:photos".to_string();

-        let merged = merge_incoming_replication_config(Some(incoming), Some(site_repl_config("office")))
-            .expect("merge should produce rules");
+        let merged = merge_incoming_replication_config(
+            Some(incoming),
+            Some(site_repl_config("office")),
+            &home_office(),
+            OperatorRuleContract::Derived,
+        )
+        .expect("merge should produce rules");

        let ids: Vec<_> = merged.rules.iter().filter_map(|rule| rule.id.as_deref()).collect();
        assert_eq!(ids, vec!["nightly-backup", "site-repl-office"]);
@@ -16324,7 +16413,15 @@ mod tests {

    #[test]
    fn test_merge_incoming_replication_config_returns_none_when_nothing_remains() {
-        assert!(merge_incoming_replication_config(Some(site_repl_config("home")), None).is_none());
+        assert!(
+            merge_incoming_replication_config(
+                Some(site_repl_config("home")),
+                None,
+                &home_office(),
+                OperatorRuleContract::Derived
+            )
+            .is_none()
+        );
    }

    fn lc_rule(id: &str, expiry_days: Option<i32>, transition_days: Option<i32>) -> s3s::dto::LifecycleRule {
@@ -16727,26 +16824,31 @@ mod tests {
    }

    // `role` is part of the bucket's S3-visible configuration. Repairing a reverse rule must
-    // drop only a sender-owned site-replication ARN, never an operator's own role — the same
-    // rule the merge path applies, so both paths agree on what is ours to rewrite.
+    // drop only a role naming a current peer, never an operator's own role — an IAM role or
+    // a remote target whose ARN carries an empty region — the same rule the merge path
+    // applies, so both paths agree on what is ours to rewrite.
    #[test]
-    fn test_replication_role_is_only_cleared_when_it_is_a_site_replication_arn() {
-        let operator_role = "arn:aws:iam::123456789012:role/replication";
-        assert!(
-            replication_target_arn_deployment_id(operator_role).is_none(),
-            "an operator IAM role is not a site-replication ARN and must be preserved"
-        );
-        assert_eq!(
-            replication_target_arn_deployment_id("arn:rustfs:replication::home:photos").as_deref(),
-            Some("home"),
-            "a site-replication ARN is sender-owned and gets cleared"
-        );
+    fn test_replication_role_is_only_cleared_when_it_names_a_peer() {
+        let sites = home_office();
+        assert!(!is_site_replication_role("arn:aws:iam::123456789012:role/replication", &sites));
+        assert!(!is_site_replication_role("arn:minio:replication::operator-dep:photos", &sites));
+        assert!(is_site_replication_role("arn:rustfs:replication::home:photos", &sites));

-        let mut incoming = site_repl_config("home");
-        incoming.role = operator_role.to_string();
-        let merged = merge_incoming_replication_config(Some(incoming), Some(site_repl_config("office")))
+        for operator_role in [
+            "arn:aws:iam::123456789012:role/replication",
+            "arn:minio:replication::operator-dep:photos",
+        ] {
+            let mut incoming = site_repl_config("home");
+            incoming.role = operator_role.to_string();
+            let merged = merge_incoming_replication_config(
+                Some(incoming),
+                Some(site_repl_config("office")),
+                &sites,
+                OperatorRuleContract::Derived,
+            )
            .expect("merge should produce rules");
-        assert_eq!(merged.role, operator_role, "operator role must survive the merge");
+            assert_eq!(merged.role, operator_role, "operator role must survive the merge");
+        }
    }

    // Rules and targets are keyed off the same ARN. Minting a fresh one while
@@ -17089,7 +17191,7 @@ mod tests {
    }

    #[test]
-    fn test_prune_removed_site_replication_rules_removes_site_rule_and_reorders_priorities() {
+    fn test_prune_removed_site_replication_rules_removes_site_rule_and_keeps_operator_priority() {
        let removed_deployment_ids = HashSet::from(["removed-dep".to_string()]);
        let kept_rule = build_site_replication_rule("arn:rustfs:replication::kept-dep:photos", 3, "site-repl-kept-dep");
        let removed_rule = build_site_replication_rule("arn:rustfs:replication::removed-dep:photos", 1, "site-repl-removed-dep");
@@ -17106,9 +17208,90 @@ mod tests {
        assert!(updated.role.is_empty());
        assert_eq!(updated.rules.len(), 2);
        assert_eq!(updated.rules[0].id.as_deref(), Some("user-managed-rule"));
-        assert_eq!(updated.rules[0].priority, Some(1));
+        assert_eq!(updated.rules[0].priority, Some(9), "the operator's priority is policy and stays");
        assert_eq!(updated.rules[1].id.as_deref(), Some("site-repl-kept-dep"));
-        assert_eq!(updated.rules[1].priority, Some(2));
+        assert_eq!(updated.rules[1].priority, Some(1), "the derived rule moves to the lowest free slot");
+    }
+
+    // Issue #1948 review: one pre-contract peer pins an S3 edit to the legacy
+    // merge; only a cluster where every remote peer answered the probe moves
+    // to the derived contract. A probe error counts as a pre-contract peer.
+    #[test]
+    fn test_operator_rule_contract_requires_every_remote_peer() {
+        let home = normalize_peer_info(PeerInfo {
+            endpoint: "https://home.example.com".to_string(),
+            ..Default::default()
+        });
+        let office = normalize_peer_info(PeerInfo {
+            endpoint: "https://office.example.com".to_string(),
+            ..Default::default()
+        });
+
+        assert_eq!(operator_rule_contract_from_probes([]), OperatorRuleContract::Derived);
+        assert_eq!(
+            operator_rule_contract_from_probes([(&home, Ok(true)), (&office, Ok(true))]),
+            OperatorRuleContract::Derived
+        );
+        assert_eq!(
+            operator_rule_contract_from_probes([(&home, Ok(true)), (&office, Ok(false))]),
+            OperatorRuleContract::Legacy
+        );
+        assert_eq!(
+            operator_rule_contract_from_probes([(&home, Err(s3_error!(InternalError, "unreachable"))), (&office, Ok(true))]),
+            OperatorRuleContract::Legacy
+        );
+    }
+
+    // The contract travels with the payload: a pre-contract sender's item has
+    // no marker and is merged the legacy way; every item this site sends is
+    // marked, bootstrap snapshots included, so a preserved config is never
+    // renumbered by a peer on the derived contract.
+    #[test]
+    fn test_bucket_meta_items_carry_the_derived_rule_contract() {
+        let legacy: SRBucketMeta = serde_json::from_str(r#"{"type":"replication-config","bucket":"photos"}"#).expect("item");
+        assert!(!legacy.derived_rule_contract);
+
+        let bucket = SRBucketInfo {
+            bucket: "photos".to_string(),
+            ..Default::default()
+        };
+        let item = bootstrap_bucket_meta_item(&bucket, "replication-config", None);
+        assert!(item.derived_rule_contract);
+        let wire = serde_json::to_value(&item).expect("json");
+        assert_eq!(wire["derivedRuleContract"], serde_json::Value::Bool(true));
+        assert!(bucket_metadata_snapshot_tombstone(&item, OffsetDateTime::now_utc()).derived_rule_contract);
+    }
+
+    // Issue #1948 review: an owner's `site-repl-user` rule on an operator ARN
+    // is outside the derived shape, so neither the prune nor the reconciler
+    // treats it as theirs; a leftover in the derived shape still is.
+    #[test]
+    fn test_derived_shape_excludes_owner_site_repl_user_rule() {
+        let owner_rule = build_site_replication_rule("arn:minio:replication:us-east-1:2f1c-remote:photos", 9, "site-repl-user");
+        assert!(!is_derived_site_replication_rule(&owner_rule));
+        assert!(is_derived_site_replication_rule(&build_site_replication_rule(
+            "arn:rustfs:replication::gone-dep:photos",
+            1,
+            "site-repl-gone-dep"
+        )));
+
+        let config = ReplicationConfiguration {
+            role: String::new(),
+            rules: vec![
+                build_site_replication_rule("arn:rustfs:replication::removed-dep:photos", 1, "site-repl-removed-dep"),
+                owner_rule,
+                build_site_replication_rule("arn:rustfs:replication::kept-dep:photos", 2, "site-repl-kept-dep"),
+            ],
+        };
+        let (updated, removed) = prune_removed_site_replication_rules(config, &HashSet::from(["removed-dep".to_string()]));
+        let updated = updated.expect("rules remain");
+        assert_eq!(removed, 1);
+        let rules: Vec<_> = updated
+            .rules
+            .iter()
+            .map(|rule| (rule.id.as_deref().unwrap(), rule.priority))
+            .collect();
+        assert_eq!(rules, vec![("site-repl-user", Some(9)), ("site-repl-kept-dep", Some(1))]);
    }

    #[test]
@@ -441,8 +441,10 @@ pub(crate) mod quota {

 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,
+        OperatorRuleContract, 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, assign_site_replication_rule_priorities, is_site_replication_role,
+        merge_incoming_replication_config, replication_target_arn_deployment_id, site_replication_rule_deployment_id,
    };
    pub(crate) type BucketReplicationResyncStatus = super::ecstore_bucket::replication::BucketReplicationResyncStatus;
    pub(crate) type BucketStats = super::ecstore_bucket::replication::BucketStats;
@@ -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,
+        OperatorRuleContract, ReplicationTargetValidationError, invalid_replication_config_status_field,
+        merge_user_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,
@@ -509,6 +509,7 @@ fn sr_bucket_meta_item(bucket: String, item_type: &str) -> SRBucketMeta {
        r#type: item_type.to_string(),
        updated_at: Some(time::OffsetDateTime::now_utc()),
        api_version: Some(SITE_REPL_API_VERSION.to_string()),
+        derived_rule_contract: true,
        ..Default::default()
    }
 }
@@ -623,11 +624,56 @@ 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 rules the reconciler derived for the current remote peers
+/// (`site_peer_deployment_ids`) — until its next pass (600s period) every
+/// peer link on this bucket would be silently dead. The same merge also drops
+/// incoming impostors of those rules. An empty peer set (site replication
+/// disabled) keeps the verbatim overwrite semantics: rule ids are not
+/// reserved, so an operator's own `site-repl-*` rule is ordinary state there.
+/// `contract` is what the peers were probed to support; the merged config is
+/// what gets broadcast, so it is built the way every peer will merge it.
+fn merge_user_replication_config_update(
+    incoming: ReplicationConfiguration,
+    existing: Option<ReplicationConfiguration>,
+    site_peer_deployment_ids: &HashSet<String>,
+    contract: OperatorRuleContract,
+) -> ReplicationConfiguration {
+    if site_peer_deployment_ids.is_empty() {
+        return incoming;
+    }
+    // `incoming` passed structure validation, so it holds at least one rule;
+    // `None` is only reachable when every incoming rule impersonates a
+    // reconciler rule, and then the stored reconciler rules are what remains.
+    merge_user_replication_config(Some(incoming.clone()), existing, site_peer_deployment_ids, contract).unwrap_or(incoming)
+}
+
+/// Split of an S3 DeleteBucketReplication on the stored config (issue #1948):
+/// the operator-authored rules are removed, the rules the reconciler derived
+/// for the current remote peers 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 reconciler
+/// rule still points at.
+fn split_replication_config_for_user_delete(
+    config: ReplicationConfiguration,
+    site_peer_deployment_ids: &HashSet<String>,
+    contract: OperatorRuleContract,
+) -> (Option<ReplicationConfiguration>, HashSet<String>) {
+    let mut removable_arns = replication_target_arns(&config);
+    let remaining = merge_user_replication_config(None, Some(config), site_peer_deployment_ids, contract);
+    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<String>,
 ) -> S3Result<Option<(BucketTargets, usize)>> {
-    let target_arns = replication_target_arns(config);
    if target_arns.is_empty() {
        return Ok(None);
    }
@@ -638,7 +684,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);
    }
@@ -1582,9 +1628,16 @@ impl DefaultBucketUsecase {
        Ok(S3Response::new(DeleteBucketPolicyOutput {}))
    }

+    /// `site_peers` is the set of remote site-replication peer deployment ids
+    /// (empty when site replication is disabled). The interface layer reads it
+    /// from the persisted state and fails closed on a read error, so this
+    /// usecase stays a pure function of its inputs (layer rule: app never
+    /// imports interface).
    pub async fn execute_delete_bucket_replication(
        &self,
        req: S3Request<DeleteBucketReplicationInput>,
+        site_peers: HashSet<String>,
+        contract: OperatorRuleContract,
    ) -> S3Result<S3Response<DeleteBucketReplicationOutput>> {
        let expected_incarnation_id = bucket_config_mutation_incarnation(&req, &req.input.bucket)?;
        let request_context = req.extensions.get::<request_context::RequestContext>().cloned();
@@ -1604,15 +1657,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(), &site_peers, contract);
+            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
@@ -2459,9 +2526,13 @@ impl DefaultBucketUsecase {
        Ok(S3Response::new(PutBucketCorsOutput::default()))
    }

+    /// See [`Self::execute_delete_bucket_replication`] for `site_peers` and
+    /// `contract`.
    pub async fn execute_put_bucket_replication(
        &self,
        req: S3Request<PutBucketReplicationInput>,
+        site_peers: HashSet<String>,
+        contract: OperatorRuleContract,
    ) -> S3Result<S3Response<PutBucketReplicationOutput>> {
        let expected_incarnation_id = bucket_config_mutation_incarnation(&req, &req.input.bucket)?;
        let request_context = req.extensions.get::<request_context::RequestContext>().cloned();
@@ -2485,6 +2556,13 @@ 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, &site_peers, contract);
        let data = serialize_config(&replication_configuration)?;
        update_bucket_config_for_incarnation(&bucket, BUCKET_REPLICATION_CONFIG, data, expected_incarnation_id)
            .await
@@ -3114,6 +3192,200 @@ 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
+    }
+
+    fn site_peers(deployment_ids: &[&str]) -> HashSet<String> {
+        deployment_ids.iter().map(|id| id.to_string()).collect()
+    }
+
+    #[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-peer-dep", 2),
+                replication_rule_with_id("arn:rustfs:replication::other-dep:bucket", "site-repl-other", 3),
+            ],
+        };
+
+        let merged = merge_user_replication_config_update(
+            incoming,
+            Some(existing),
+            &site_peers(&["peer-dep"]),
+            OperatorRuleContract::Derived,
+        );
+
+        let rules: Vec<_> = merged
+            .rules
+            .iter()
+            .map(|rule| (rule.id.as_deref().unwrap_or_default(), rule.destination.bucket.as_str()))
+            .collect();
+        assert_eq!(
+            rules,
+            vec![
+                ("new-user-rule", "arn:rustfs:replication:us-east-1:new:bucket"),
+                ("site-repl-other", "arn:rustfs:replication::other-dep:bucket"),
+                ("site-repl-peer-dep", "arn:rustfs:replication::peer-dep:bucket"),
+            ],
+            "user rules replaced, the reconciler rule for the current peer kept over the incoming impostor, \
+             a site-repl-* id that names no current peer is ordinary operator state"
+        );
+    }
+
+    // Rule ids do not reserve `site-repl-*`: outside site replication an
+    // owner's `site-repl-user` rule is ordinary state, so PUT stores it
+    // verbatim and DELETE removes it and garbage-collects its target.
+    #[test]
+    fn put_then_delete_replication_without_site_replication_treats_site_repl_id_as_user_rule() {
+        let user_arn = "arn:minio:replication:us-east-1:2f1c-remote:bucket";
+        let incoming = ReplicationConfiguration {
+            role: String::new(),
+            rules: vec![replication_rule_with_id(user_arn, "site-repl-user", 1)],
+        };
+
+        let stored = merge_user_replication_config_update(incoming.clone(), None, &HashSet::new(), OperatorRuleContract::Derived);
+        assert_eq!(stored, incoming, "PUT on a non-site-replication bucket is verbatim");
+
+        let (remaining, removable) =
+            split_replication_config_for_user_delete(stored, &HashSet::new(), OperatorRuleContract::Derived);
+        assert!(remaining.is_none(), "DELETE must remove the operator's site-repl-* rule");
+        assert_eq!(removable, HashSet::from([user_arn.to_string()]));
+    }
+
+    // Under site replication only a rule the reconciler would derive — id
+    // `site-repl-<peer>` for a current peer, destination ARN naming the same
+    // peer — is reconciler-owned. Everything else is operator state.
+    #[test]
+    fn delete_replication_split_keeps_only_reconciler_derived_rules() {
+        let peer_arn = "arn:rustfs:replication::peer-dep:bucket";
+        let user_arn = "arn:minio:replication:us-east-1:2f1c-remote:bucket";
+        let config = ReplicationConfiguration {
+            role: String::new(),
+            rules: vec![
+                replication_rule_with_id(user_arn, "site-repl-user", 1),
+                replication_rule_with_id(user_arn, "site-repl-peer-dep", 2),
+                replication_rule_with_id("arn:rustfs:replication::gone-dep:bucket", "site-repl-gone-dep", 3),
+                replication_rule_with_id(peer_arn, "site-repl-peer-dep", 4),
+            ],
+        };
+
+        let (remaining, removable) =
+            split_replication_config_for_user_delete(config, &site_peers(&["peer-dep"]), OperatorRuleContract::Derived);
+
+        let remaining = remaining.expect("the reconciler-derived rule must survive");
+        assert_eq!(remaining.rules.len(), 1);
+        assert_eq!(remaining.rules[0].destination.bucket, peer_arn);
+        assert_eq!(
+            removable,
+            HashSet::from([user_arn.to_string(), "arn:rustfs:replication::gone-dep:bucket".to_string()]),
+            "targets of operator rules and of a removed peer are garbage-collected"
+        );
+    }
+
+    #[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),
+            &HashSet::new(),
+            OperatorRuleContract::Derived,
+        );
+
+        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, &site_peers(&["peer-dep"]), OperatorRuleContract::Derived);
+
+        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, &site_peers(&["peer-dep"]), OperatorRuleContract::Derived);
+
+        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, &site_peers(&["peer-dep"]), OperatorRuleContract::Derived);
+
+        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
@@ -3451,7 +3723,10 @@ mod tests {
        let req = build_request(input, Method::DELETE);
        let usecase = DefaultBucketUsecase::without_context();

-        let err = usecase.execute_delete_bucket_replication(req).await.unwrap_err();
+        let err = usecase
+            .execute_delete_bucket_replication(req, HashSet::new(), OperatorRuleContract::Derived)
+            .await
+            .unwrap_err();
        assert_eq!(err.code(), &S3ErrorCode::InternalError);
    }

@@ -4537,7 +4812,10 @@ mod tests {
        let req = build_request(input, Method::PUT);
        let usecase = DefaultBucketUsecase::without_context();

-        let err = usecase.execute_put_bucket_replication(req).await.unwrap_err();
+        let err = usecase
+            .execute_put_bucket_replication(req, HashSet::new(), OperatorRuleContract::Derived)
+            .await
+            .unwrap_err();
        assert_eq!(err.code(), &S3ErrorCode::InternalError);
    }

@@ -4555,7 +4833,7 @@ mod tests {
            .unwrap();

        let err = DefaultBucketUsecase::without_context()
-            .execute_put_bucket_replication(build_request(input, Method::PUT))
+            .execute_put_bucket_replication(build_request(input, Method::PUT), HashSet::new(), OperatorRuleContract::Derived)
            .await
            .expect_err("unsupported fields must be rejected before store access");

@@ -619,6 +619,8 @@ pub(crate) mod bucket {

        use crate::storage::storage_api::ecstore_bucket::replication as replication_contracts;

+        pub(crate) use replication_contracts::{OperatorRuleContract, merge_user_replication_config};
+
        type ReplicationObjectBridge = crate::storage::storage_api::ecstore_bucket::replication::ReplicationObjectBridge;
        pub(crate) type DeleteReplicationConfigSnapshot =
            crate::storage::storage_api::ecstore_bucket::replication::DeleteReplicationConfigSnapshot;
@@ -59,10 +59,74 @@ const LOG_SUBSYSTEM_OBJECT_LOCK: &str = "object_lock";
 const LOG_SUBSYSTEM_TAGGING: &str = "tagging";

 use crate::app::storage_api::object_usecase::bucket::replication::{
-    ReplicateDecision, get_read_proxy_targets, must_replicate_metadata, schedule_metadata_replication,
+    OperatorRuleContract, ReplicateDecision, get_read_proxy_targets, must_replicate_metadata, schedule_metadata_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<bool> {
+    #[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
+}
+
+/// Remote site-replication peer deployment ids and the operator-rule contract
+/// the peers support, handed to the bucket usecase so an S3 replication-config
+/// edit keeps exactly the reconciler-owned rules and merges the way every
+/// peer will (issue #1948). Read here, in the interface layer, because the
+/// usecase must not import the admin handlers (layer guard); a state-read
+/// failure propagates so the edit fails closed.
+async fn site_replication_edit_context() -> S3Result<(std::collections::HashSet<String>, OperatorRuleContract)> {
+    // While the gate override is in effect the test exercises the deny/allow
+    // branch, not the peer set; there is no persisted state to read.
+    #[cfg(test)]
+    if SITE_REPLICATION_GATE_TEST_OVERRIDE.load(std::sync::atomic::Ordering::SeqCst) != 0 {
+        return Ok((std::collections::HashSet::new(), OperatorRuleContract::Derived));
+    }
+    crate::admin::handlers::site_replication::site_replication_edit_context().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<T>(req: &S3Request<T>) -> 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,8 +564,10 @@ impl S3 for FS {
        &self,
        req: S3Request<DeleteBucketReplicationInput>,
    ) -> S3Result<S3Response<DeleteBucketReplicationOutput>> {
+        deny_replication_config_edit_for_non_owner(&req).await?;
+        let (site_peers, contract) = site_replication_edit_context().await?;
        let usecase = s3_api::bucket_usecase_for(self);
-        usecase.execute_delete_bucket_replication(req).await
+        usecase.execute_delete_bucket_replication(req, site_peers, contract).await
    }

    #[instrument(level = "debug", skip(self))]
@@ -1353,8 +1419,10 @@ impl S3 for FS {
        &self,
        req: S3Request<PutBucketReplicationInput>,
    ) -> S3Result<S3Response<PutBucketReplicationOutput>> {
+        deny_replication_config_edit_for_non_owner(&req).await?;
+        let (site_peers, contract) = site_replication_edit_context().await?;
        let usecase = s3_api::bucket_usecase_for(self);
-        usecase.execute_put_bucket_replication(req).await
+        usecase.execute_put_bucket_replication(req, site_peers, contract).await
    }

    async fn put_bucket_request_payment(
@@ -1919,3 +1987,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<T>(input: T, is_owner: bool) -> S3Request<T> {
+        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);
+    }
+}