Merge branch 'main' into fix/b5-t1-a1-replication-deny-edit

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.

.config/coverage-baselines.toml

@@ -1,20 +0,0 @@
-# Report-only calibration baseline from https://github.com/rustfs/rustfs/actions/runs/29394996173.
-# Update counts only with a linked coverage run and a reviewed explanation.
-phase = "report-only"
-allowed_drop_percentage_points = 1.0
-
-[crates."crates/iam"]
-covered = 5149
-count = 8131
-
-[crates."crates/kms"]
-covered = 2950
-count = 4200
-
-[crates."crates/policy"]
-covered = 4636
-count = 5464
-
-[crates."crates/crypto"]
-covered = 469
-count = 494

.config/make/tests.mak

@@ -36,7 +36,6 @@ script-tests: ## Run shell script tests
 	./scripts/test_manual_transition_runbooks.sh
 	./scripts/check_embedded_secrets.sh --self-test
 	python3 ./scripts/check_test_wiring.py --self-test
-	python3 ./scripts/check_security_coverage.py --self-test
 	python3 ./scripts/check_scheduled_validation_freshness.py --self-test
 	python3 ./scripts/s3-tests/test_report_compat.py
 	bash -n ./scripts/validate_object_data_cache_cold_stampede.sh

.github/workflows/coverage.yml

@@ -12,12 +12,14 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
-# Workspace line-coverage baseline and security-crate calibration
-# (backlog#1153 infra-5/infra-6).
+# Weekly workspace line-coverage baseline (backlog#1153 infra-5).
 #
-# NON-BLOCKING by design: the weekly job gives coverage a visible baseline and
-# trend, while relevant pull requests run a report-only security-crate
-# comparison. Neither job is a required check during calibration.
+# NON-BLOCKING by design: this workflow only runs on schedule and manual
+# dispatch, so it never attaches a status to a PR and must never be made a
+# required check. It exists to give coverage a visible baseline and trend
+# (per-crate table in the job summary, lcov artifact kept 90 days) — the
+# per-crate ratchet for the security-critical crates builds on it later
+# (backlog#1153 infra-6, report-only first per the ci-11 ladder).
 #
 # Measurement scope matches the PR test gate (ci.yml "Run tests"):
 # `--workspace --exclude e2e_test` with the `ci` nextest profile. Doctests are
@@ -29,17 +31,6 @@
 name: coverage
 
 on:
-  pull_request:
-    branches: [main]
-    paths:
-      - "crates/iam/**"
-      - "crates/kms/**"
-      - "crates/policy/**"
-      - "crates/crypto/**"
-      - ".config/coverage-baselines.toml"
-      - "scripts/coverage_per_crate.py"
-      - "scripts/check_security_coverage.py"
-      - ".github/workflows/coverage.yml"
   workflow_dispatch:
   schedule:
     # 07:00 UTC Sunday — staggered clear of the other Sunday crons: ci (00:00),
@@ -48,10 +39,6 @@ on:
     # e2e-replication-nightly (04:00) and performance-ab (06:00) lanes.
     - cron: "43 7 * * 0"
 
-concurrency:
-  group: ${{ github.workflow }}-${{ github.event_name }}-${{ github.event.pull_request.number || github.ref }}
-  cancel-in-progress: ${{ github.event_name != 'schedule' }}
-
 # Only alert-on-failure needs more than read access; it declares its own
 # job-level `issues: write`.
 permissions:
@@ -59,13 +46,12 @@ permissions:
 
 jobs:
   coverage:
-    name: Workspace line coverage
+    name: Workspace coverage (weekly)
     runs-on: sm-standard-4
     # The instrumented build cannot reuse the regular CI cache (different
-    # RUSTFLAGS), so a cold run rebuilds the workspace before running the
-    # full suite. Exact-head run 32573798257 needed 119m42s including reports
-    # and artifact upload, so keep a bounded 30-minute publication margin.
-    timeout-minutes: 150
+    # RUSTFLAGS), so a cold week rebuilds the workspace before running the
+    # full suite; give it double the test job's 60-minute budget.
+    timeout-minutes: 120
     env:
       FORCE_JAVASCRIPT_ACTIONS_TO_NODE24: "true"
       # Match the PR gate's nextest semantics (ci.yml runs `--profile ci`):
@@ -105,9 +91,7 @@ jobs:
           cargo llvm-cov report --json --output-path target/llvm-cov/coverage.json
 
       - name: Write per-crate summary
-        run: |
-          python3 scripts/coverage_per_crate.py target/llvm-cov/coverage.json >> "$GITHUB_STEP_SUMMARY"
-          python3 scripts/check_security_coverage.py target/llvm-cov/coverage.json >> "$GITHUB_STEP_SUMMARY"
+        run: python3 scripts/coverage_per_crate.py target/llvm-cov/coverage.json >> "$GITHUB_STEP_SUMMARY"
 
       - name: Upload coverage artifact
         if: always()

.github/workflows/nightly-gnu.yml

@@ -39,10 +39,11 @@ jobs:
     env:
       FORCE_JAVASCRIPT_ACTIONS_TO_NODE24: "true"
     steps:
-      - name: Checkout repository
+      - name: Checkout main branch
         uses: actions/checkout@9c091bb21b7c1c1d1991bb908d89e4e9dddfe3e0 # v7
         with:
           persist-credentials: false
+          ref: main
 
       - name: Setup Rust environment
         uses: ./.github/actions/setup
@@ -88,10 +89,11 @@ jobs:
       # either casing.
       NO_PROXY: 127.0.0.1,localhost
     steps:
-      - name: Checkout repository
+      - name: Checkout main branch
         uses: actions/checkout@9c091bb21b7c1c1d1991bb908d89e4e9dddfe3e0 # v7
         with:
           persist-credentials: false
+          ref: main
 
       - name: Setup Rust environment
         uses: ./.github/actions/setup
@@ -176,10 +178,11 @@ jobs:
       FORCE_JAVASCRIPT_ACTIONS_TO_NODE24: "true"
       NO_PROXY: 127.0.0.1,localhost
     steps:
-      - name: Checkout repository
+      - name: Checkout main branch
         uses: actions/checkout@9c091bb21b7c1c1d1991bb908d89e4e9dddfe3e0 # v7
         with:
           persist-credentials: false
+          ref: main
 
       - name: Setup Rust environment
         uses: ./.github/actions/setup

crates/ecstore/src/api/mod.rs

@@ -196,15 +196,16 @@ pub mod bucket {
             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,
-            unsupported_replication_config_field, validate_replication_config_structure, validate_replication_config_target_arns,
-            version_purge_status_to_filemeta,
+            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_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, unsupported_replication_config_field,
+            validate_replication_config_structure, validate_replication_config_target_arns, version_purge_status_to_filemeta,
         };
     }
 

crates/ecstore/src/bucket/replication/mod.rs

@@ -47,8 +47,10 @@ 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,
+    assign_site_replication_rule_priorities, invalid_replication_config_status_field, is_site_replication_rule,
+    merge_incoming_replication_config, merge_user_replication_config, replication_target_arn_deployment_id,
+    replication_target_arns, should_remove_replication_target, unsupported_replication_config_field,
+    validate_replication_config_structure, validate_replication_config_target_arns,
 };
 pub(crate) use replication_filemeta_boundary::version_purge_statuses_map;
 pub use replication_filemeta_boundary::{

crates/ecstore/src/bucket/replication/replication_config_boundary.rs

@@ -16,6 +16,8 @@ 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,
+    assign_site_replication_rule_priorities, invalid_replication_config_status_field, is_site_replication_rule,
+    merge_incoming_replication_config, merge_user_replication_config, replication_target_arn_deployment_id,
+    replication_target_arns, should_remove_replication_target, unsupported_replication_config_field,
+    validate_replication_config_structure, validate_replication_config_target_arns,
 };

crates/ecstore/src/io_support/bitrot.rs

@@ -784,24 +784,6 @@ pub(crate) fn create_deferred_bitrot_reader_with_stripe_handle(
 ///
 /// # Returns
 /// A Result containing the BitrotWriterWrapper or an error
-/// Size hint handed to `DiskAPI::create_file` for a bitrot-wrapped shard.
-///
-/// A known length is grown by one checksum per shard so the on-disk file size
-/// matches what the bitrot writer emits. A negative length is the
-/// unknown-size sentinel (`HashReader::SIZE_PRESERVE_LAYER`, used by SSE and
-/// compression) and must be preserved: `RemoteDisk::create_file` forwards it
-/// in the `put_file_stream` query, and the receiver only treats `size > 0` as
-/// a fixed body length when locating the authenticated trailer. Clamping it
-/// to `0` would claim an empty body and misframe the stream. `0` stays `0`
-/// because a genuinely empty object still means an empty body.
-fn bitrot_create_file_size(length: i64, shard_size: usize, checksum_algo: &HashAlgorithm) -> i64 {
-    if length <= 0 {
-        return length;
-    }
-    let length = length as usize;
-    (length.div_ceil(shard_size) * checksum_algo.size() + length) as i64
-}
-
 pub async fn create_bitrot_writer(
     is_inline_buffer: bool,
     disk: Option<&DiskStore>,
@@ -814,7 +796,12 @@ pub async fn create_bitrot_writer(
     let writer = if is_inline_buffer {
         CustomWriter::new_inline_buffer()
     } else if let Some(disk) = disk {
-        let length = bitrot_create_file_size(length, shard_size, &checksum_algo);
+        let length = if length > 0 {
+            let length = length as usize;
+            (length.div_ceil(shard_size) * checksum_algo.size() + length) as i64
+        } else {
+            0
+        };
 
         let file = disk.create_file("", volume, path, length).await?;
         #[cfg(feature = "hotpath")]
@@ -833,25 +820,6 @@ mod tests {
     use rustfs_rio::ChunkReader;
     use std::collections::VecDeque;
 
-    #[test]
-    fn bitrot_create_file_size_grows_known_length_by_checksums() {
-        // 10 bytes over 4-byte shards = 3 shards, each followed by a 32-byte hash.
-        assert_eq!(bitrot_create_file_size(10, 4, &HashAlgorithm::HighwayHash256), 10 + 3 * 32);
-        assert_eq!(bitrot_create_file_size(10, 4, &HashAlgorithm::None), 10);
-    }
-
-    #[test]
-    fn bitrot_create_file_size_keeps_empty_and_unknown_distinct() {
-        assert_eq!(bitrot_create_file_size(0, 4, &HashAlgorithm::HighwayHash256), 0);
-        // SSE/compression streams advertise SIZE_PRESERVE_LAYER (-1); the remote
-        // put_file_stream receiver relies on a non-positive size to parse the auth
-        // trailer from the stream tail, so the sentinel must survive untouched.
-        assert_eq!(
-            bitrot_create_file_size(rustfs_rio::HashReader::SIZE_PRESERVE_LAYER, 4, &HashAlgorithm::HighwayHash256),
-            rustfs_rio::HashReader::SIZE_PRESERVE_LAYER
-        );
-    }
-
     struct TestChunkReader {
         chunks: VecDeque<Bytes>,
     }

crates/kms/tests/vault_ha_failover_live.rs

@@ -16,14 +16,14 @@
 //!
 //! `scripts/test/vault_ha_kms_live.sh` owns the official Vault containers and
 //! kills the active node while this test continuously decrypts through a
-//! surviving standby. KV2 and Transit must recover after the bounded circuit
-//! interval, use a bounded number of attempts, and leave the circuit and
-//! in-flight gauges at zero after a new leader is elected.
+//! surviving standby. KV2 and Transit requests must remain successful, use a
+//! bounded number of attempts, and leave the circuit and in-flight gauges at
+//! zero after a new leader is elected.
 
 use std::collections::HashMap;
 use std::path::{Path, PathBuf};
+use std::sync::Arc;
 use std::sync::atomic::{AtomicBool, AtomicU64, Ordering};
-use std::sync::{Arc, Mutex};
 use std::time::Duration;
 
 use metrics_util::MetricKind;
@@ -43,11 +43,6 @@ const OPERATION_ATTEMPTS: &str = "rustfs_kms_backend_operation_attempts";
 const IN_FLIGHT: &str = "rustfs_kms_backend_in_flight";
 const CIRCUIT_OPEN: &str = "rustfs_kms_backend_circuit_open";
 const MAX_ATTEMPTS: u32 = 10;
-const ATTEMPT_TIMEOUT: Duration = Duration::from_secs(2);
-const HEALTHY_PROGRESS_TIMEOUT: Duration = Duration::from_secs(20);
-// The circuit remains open for 30s after five failed attempts.
-const POST_FAILOVER_PROGRESS_TIMEOUT: Duration = Duration::from_secs(35);
-const FAILOVER_ERROR_POLL_INTERVAL: Duration = Duration::from_millis(100);
 
 type MetricEntry = (
     metrics_util::CompositeKey,
@@ -69,7 +64,7 @@ fn config(backend: KmsBackend, backend_config: BackendConfig) -> KmsConfig {
         backend,
         backend_config,
         allow_insecure_dev_defaults: true,
-        timeout: ATTEMPT_TIMEOUT,
+        timeout: Duration::from_secs(2),
         retry_attempts: MAX_ATTEMPTS,
         enable_cache: false,
         ..KmsConfig::default()
@@ -169,31 +164,14 @@ fn retryable_failures(snapshot: &[MetricEntry], operation: &str) -> u64 {
         .sum()
 }
 
-async fn wait_for_count(
-    counter: &AtomicU64,
-    failure: &Mutex<Option<String>>,
-    minimum: u64,
-    description: &str,
-    timeout: Duration,
-) {
-    tokio::time::timeout(timeout, async {
+async fn wait_for_count(counter: &AtomicU64, minimum: u64, description: &str) {
+    tokio::time::timeout(Duration::from_secs(20), async {
         while counter.load(Ordering::SeqCst) < minimum {
-            if let Some(error) = failure.lock().expect("decrypt failure lock poisoned").as_ref() {
-                panic!(
-                    "{description} worker failed after {} successful decrypts: {error}",
-                    counter.load(Ordering::SeqCst)
-                );
-            }
             tokio::time::sleep(Duration::from_millis(25)).await;
         }
     })
     .await
-    .unwrap_or_else(|_| {
-        panic!(
-            "timed out after {timeout:?} waiting for {description}: completed {}, expected {minimum}",
-            counter.load(Ordering::SeqCst)
-        )
-    });
+    .unwrap_or_else(|_| panic!("timed out waiting for {description}"));
 }
 
 async fn wait_for_file(path: &Path, description: &str) {
@@ -211,8 +189,7 @@ async fn decrypt_loop<B: KmsBackendTrait + Send + Sync + 'static>(
     request: DecryptRequest,
     expected: Vec<u8>,
     completed: Arc<AtomicU64>,
-    allow_failover_errors: Arc<AtomicBool>,
-    failure: Arc<Mutex<Option<String>>>,
+    failed: Arc<AtomicBool>,
     stop: CancellationToken,
 ) {
     while !stop.is_cancelled() {
@@ -220,18 +197,8 @@ async fn decrypt_loop<B: KmsBackendTrait + Send + Sync + 'static>(
             Ok(response) if response.plaintext == expected => {
                 completed.fetch_add(1, Ordering::SeqCst);
             }
-            Ok(_) => {
-                *failure.lock().expect("decrypt failure lock poisoned") =
-                    Some("decrypt returned unexpected plaintext".to_string());
-                return;
-            }
-            Err(rustfs_kms::KmsError::BackendError { .. } | rustfs_kms::KmsError::OperationTimedOut { .. })
-                if allow_failover_errors.load(Ordering::SeqCst) =>
-            {
-                tokio::time::sleep(FAILOVER_ERROR_POLL_INTERVAL).await;
-            }
-            Err(error) => {
-                *failure.lock().expect("decrypt failure lock poisoned") = Some(error.to_string());
+            Ok(_) | Err(_) => {
+                failed.store(true, Ordering::SeqCst);
                 return;
             }
         }
@@ -329,9 +296,7 @@ async fn exercise_failover(snapshotter: &Snapshotter) {
     );
 
     let stop = CancellationToken::new();
-    let allow_failover_errors = Arc::new(AtomicBool::new(false));
-    let kv2_failure = Arc::new(Mutex::new(None));
-    let transit_failure = Arc::new(Mutex::new(None));
+    let failed = Arc::new(AtomicBool::new(false));
     let kv2_completed = Arc::new(AtomicU64::new(0));
     let transit_completed = Arc::new(AtomicU64::new(0));
     let kv2_worker = tokio::spawn(decrypt_loop(
@@ -339,8 +304,7 @@ async fn exercise_failover(snapshotter: &Snapshotter) {
         kv2_request,
         kv2_data_key.plaintext_key,
         Arc::clone(&kv2_completed),
-        Arc::clone(&allow_failover_errors),
-        Arc::clone(&kv2_failure),
+        Arc::clone(&failed),
         stop.clone(),
     ));
     let transit_worker = tokio::spawn(decrypt_loop(
@@ -348,21 +312,12 @@ async fn exercise_failover(snapshotter: &Snapshotter) {
         transit_request,
         transit_data_key.plaintext_key,
         Arc::clone(&transit_completed),
-        Arc::clone(&allow_failover_errors),
-        Arc::clone(&transit_failure),
+        Arc::clone(&failed),
         stop.clone(),
     ));
 
-    wait_for_count(&kv2_completed, &kv2_failure, 2, "two healthy KV2 decrypts", HEALTHY_PROGRESS_TIMEOUT).await;
-    wait_for_count(
-        &transit_completed,
-        &transit_failure,
-        2,
-        "two healthy Transit decrypts",
-        HEALTHY_PROGRESS_TIMEOUT,
-    )
-    .await;
-    allow_failover_errors.store(true, Ordering::SeqCst);
+    wait_for_count(&kv2_completed, 2, "two healthy KV2 decrypts").await;
+    wait_for_count(&transit_completed, 2, "two healthy Transit decrypts").await;
     std::fs::write(&marker, b"ready").expect("publish failover readiness marker");
 
     wait_for_file(&elected, "the replacement Vault leader").await;
@@ -371,39 +326,18 @@ async fn exercise_failover(snapshotter: &Snapshotter) {
 
     let kv2_after_election = kv2_completed.load(Ordering::SeqCst) + 2;
     let transit_after_election = transit_completed.load(Ordering::SeqCst) + 2;
-    wait_for_count(
-        &kv2_completed,
-        &kv2_failure,
-        kv2_after_election,
-        "post-failover KV2 decrypts",
-        POST_FAILOVER_PROGRESS_TIMEOUT,
-    )
-    .await;
-    wait_for_count(
-        &transit_completed,
-        &transit_failure,
-        transit_after_election,
-        "post-failover Transit decrypts",
-        POST_FAILOVER_PROGRESS_TIMEOUT,
-    )
-    .await;
+    wait_for_count(&kv2_completed, kv2_after_election, "post-failover KV2 decrypts").await;
+    wait_for_count(&transit_completed, transit_after_election, "post-failover Transit decrypts").await;
 
     stop.cancel();
     kv2_worker.await.expect("KV2 decrypt worker must join");
     transit_worker.await.expect("Transit decrypt worker must join");
-    assert!(
-        kv2_failure.lock().expect("KV2 failure lock poisoned").is_none(),
-        "no KV2 decrypt may fail or return different plaintext"
-    );
-    assert!(
-        transit_failure.lock().expect("Transit failure lock poisoned").is_none(),
-        "no Transit decrypt may fail or return different plaintext"
-    );
+    assert!(!failed.load(Ordering::SeqCst), "no decrypt may fail or return different plaintext");
 }
 
 #[test]
 #[ignore = "requires a real three-node Vault Raft cluster; run scripts/test/vault_ha_kms_live.sh"]
-fn vault_raft_leader_failure_recovers_kv2_and_transit_decrypts() {
+fn vault_raft_leader_failure_preserves_kv2_and_transit_decrypts() {
     let recorder = DebuggingRecorder::new();
     let snapshotter = recorder.snapshotter();
     metrics::with_local_recorder(&recorder, || {
@@ -415,6 +349,11 @@ fn vault_raft_leader_failure_recovers_kv2_and_transit_decrypts() {
     });
     let snapshot = snapshotter.snapshot().into_vec();
 
+    assert_eq!(
+        counter_value(&snapshot, OPERATIONS_TOTAL, &[("outcome", "circuit_open")]),
+        0,
+        "a bounded leader election must not open the circuit"
+    );
     assert_eq!(
         counter_value(&snapshot, OPERATIONS_TOTAL, &[("outcome", "budget_exhausted")]),
         0,

crates/replication/src/config.rs

@@ -270,6 +270,160 @@ 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-*`). The
+/// reconciler and the peer ingestion path treat the whole namespace as theirs
+/// on a site-replication bucket; the S3 edit path must not — rule ids are not
+/// reserved, so see [`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))
+}
+
+/// Merge an incoming replication config into the local one.
+///
+/// `site-repl-*` rules encode the *holder's* outbound direction — their
+/// destination ARN names another site — so applying an external rule set
+/// verbatim replaces the local reverse rule with one this site can never
+/// satisfy (no bucket target backs it) and replication silently stops. Only
+/// operator-authored rules travel: the site-replication peer ingestion path
+/// and the S3 put/delete-bucket-replication path both keep the local site's
+/// `site-repl-*` rules through this merge. `incoming == None` models a
+/// delete of the operator-authored rules.
+pub fn merge_incoming_replication_config(
+    incoming: Option<ReplicationConfiguration>,
+    local: Option<ReplicationConfiguration>,
+) -> Option<ReplicationConfiguration> {
+    merge_replication_config_keeping_site_rules(incoming, local, is_site_replication_rule)
+}
+
+/// [`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.
+pub fn merge_user_replication_config(
+    incoming: Option<ReplicationConfiguration>,
+    local: Option<ReplicationConfiguration>,
+    peer_deployment_ids: &HashSet<String>,
+) -> Option<ReplicationConfiguration> {
+    let incoming = incoming.map(|mut config| {
+        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))
+        });
+        config
+    });
+    merge_replication_config_keeping_site_rules(incoming, local, |rule| {
+        is_reconciler_owned_site_replication_rule(rule, peer_deployment_ids)
+    })
+}
+
+fn merge_replication_config_keeping_site_rules(
+    incoming: Option<ReplicationConfiguration>,
+    local: Option<ReplicationConfiguration>,
+    is_site_rule: impl Fn(&ReplicationRule) -> bool,
+) -> Option<ReplicationConfiguration> {
+    let incoming_role = incoming.as_ref().map(|config| config.role.clone()).unwrap_or_default();
+    // Operator rules first, then the local site rules — the same order the
+    // site-replication reconciler produces, so its no-op check matches and
+    // the bucket metadata is written once per broadcast, not twice.
+    let mut rules: Vec<ReplicationRule> = incoming
+        .into_iter()
+        .flat_map(|config| config.rules)
+        .filter(|rule| !is_site_rule(rule))
+        .collect();
+    rules.extend(local.into_iter().flat_map(|config| config.rules).filter(&is_site_rule));
+
+    if rules.is_empty() {
+        return None;
+    }
+
+    assign_site_replication_rule_priorities(&mut rules, &is_site_rule);
+
+    // A site-replication ARN in `role` is the sender's, and the reconciler's
+    // per-peer target lookup reads it — carrying it over would pin the
+    // receiver's targets to the sender's identity.
+    let role = match replication_target_arn_deployment_id(&incoming_role) {
+        Some(_) => String::new(),
+        None => incoming_role,
+    };
+
+    Some(ReplicationConfiguration { role, rules })
+}
+
+/// 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 +1698,103 @@ 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).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)).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);
+    }
 }

crates/replication/src/lib.rs

@@ -32,9 +32,11 @@ 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,
+    active_replication_rule_destination_arns, assign_site_replication_rule_priorities, invalid_replication_config_status_field,
+    is_reconciler_owned_site_replication_rule, 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,

docs/testing/README.md

@@ -158,11 +158,10 @@ added by backlog#1153 infra-4.
 
 ## Coverage
 
-Workspace line coverage is measured weekly. Pull requests that touch iam, kms,
-policy, or crypto also run a non-required, report-only comparison against
-`.config/coverage-baselines.toml`. During calibration, a regression is recorded
-in the job summary without failing the job; missing or malformed coverage
-evidence still fails closed (backlog#1153 infra-6).
+Line coverage is measured **weekly, not per-PR**, and is non-blocking: it
+exists for visibility and trend, never as a required check. Per-crate ratchets
+for the security-critical crates (iam / kms / policy / crypto) build on this
+baseline later (backlog#1153 infra-6, report-only first).
 
 - **CI**: `.github/workflows/coverage.yml` runs every Sunday and on manual
   dispatch: `cargo llvm-cov nextest --workspace --exclude e2e_test` under the
@@ -175,12 +174,6 @@ evidence still fails closed (backlog#1153 infra-6).
   plus the full suite). It prints the same per-crate table via
   `scripts/coverage_per_crate.py` and writes `target/llvm-cov/lcov.info` and
   `coverage.json`.
-- **Security-critical ratchet**: relevant pull requests compare iam / kms /
-  policy / crypto line coverage with the versioned baseline. Drops greater than
-  the configured one-percentage-point calibration threshold are marked
-  `REGRESSION (report-only)`. The weekly summary runs the same comparison so
-  calibration continues even when no relevant pull request is open. Baseline
-  changes require a linked coverage run and a reviewed explanation.
 - **Trend comparison**: each run's job summary is the weekly per-crate
   snapshot — open two runs from the Actions history (workflow "coverage") and
   compare their tables. For line-level diffs, download the two runs'

rustfs/src/admin/handlers/site_replication.rs

@@ -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::{
+    assign_site_replication_rule_priorities, is_site_replication_rule, merge_incoming_replication_config,
+    replication_target_arn_deployment_id,
+};
 use crate::admin::storage_api::bucket::target::{ARN, BucketTarget, BucketTargetType, BucketTargets, Credentials};
 use crate::admin::storage_api::bucket::target_sys::BucketTargetSys;
 use crate::admin::storage_api::bucket::utils::{deserialize, serialize};
@@ -1118,6 +1122,36 @@ 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_remote_peer_deployment_ids() -> S3Result<HashSet<String>> {
+    let state = load_site_replication_state().await?;
+    if !state.enabled() {
+        return Ok(HashSet::new());
+    }
+    let local_peer = current_local_runtime_peer(&state);
+    Ok(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())
+}
+
 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),
@@ -7762,20 +7796,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 +7820,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
@@ -7829,52 +7845,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
@@ -8213,9 +8183,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_site_replication_rule);
 
     (Some(config), removed)
 }
@@ -8389,9 +8357,10 @@ async fn ensure_site_replication_bucket_replication_config_with_runtime(
         .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_site_replication_rule);
 
     // Only a site-replication ARN in `role` is ours to drop — an operator-authored role is
     // part of the bucket's S3-visible configuration, and repairing a reverse rule must not
@@ -17089,7 +17058,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 +17075,9 @@ 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");
     }
 
     #[test]

rustfs/src/admin/storage_api.rs

@@ -443,6 +443,8 @@ 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,
+        assign_site_replication_rule_priorities, is_site_replication_rule, merge_incoming_replication_config,
+        replication_target_arn_deployment_id,
     };
     pub(crate) type BucketReplicationResyncStatus = super::ecstore_bucket::replication::BucketReplicationResyncStatus;
     pub(crate) type BucketStats = super::ecstore_bucket::replication::BucketStats;

rustfs/src/app/bucket_usecase.rs

@@ -38,9 +38,9 @@ use super::storage_api::bucket_usecase::bucket::{
     metadata_sys,
     policy_sys::PolicySys,
     replication::{
-        ReplicationTargetValidationError, invalid_replication_config_status_field, replication_target_arns,
-        should_remove_replication_target, unsupported_replication_config_field, validate_replication_config_structure,
-        validate_replication_config_target_arns,
+        ReplicationTargetValidationError, invalid_replication_config_status_field, 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,
@@ -623,11 +623,52 @@ 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.
+fn merge_user_replication_config_update(
+    incoming: ReplicationConfiguration,
+    existing: Option<ReplicationConfiguration>,
+    site_peer_deployment_ids: &HashSet<String>,
+) -> 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).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>,
+) -> (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);
+    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 +679,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 +1623,15 @@ 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>,
     ) -> 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 +1651,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);
+            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 +2520,11 @@ impl DefaultBucketUsecase {
         Ok(S3Response::new(PutBucketCorsOutput::default()))
     }
 
+    /// See [`Self::execute_delete_bucket_replication`] for `site_peers`.
     pub async fn execute_put_bucket_replication(
         &self,
         req: S3Request<PutBucketReplicationInput>,
+        site_peers: HashSet<String>,
     ) -> 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 +2548,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);
         let data = serialize_config(&replication_configuration)?;
         update_bucket_config_for_incarnation(&bucket, BUCKET_REPLICATION_CONFIG, data, expected_incarnation_id)
             .await
@@ -3114,6 +3184,185 @@ 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"]));
+
+        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());
+        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());
+        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"]));
+
+        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());
+
+        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"]));
+
+        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"]));
+
+        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"]));
+
+        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 +3700,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())
+            .await
+            .unwrap_err();
         assert_eq!(err.code(), &S3ErrorCode::InternalError);
     }
 
@@ -4537,7 +4789,7 @@ 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()).await.unwrap_err();
         assert_eq!(err.code(), &S3ErrorCode::InternalError);
     }
 
@@ -4555,7 +4807,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())
             .await
             .expect_err("unsupported fields must be rejected before store access");
 

rustfs/src/app/storage_api.rs

@@ -619,6 +619,8 @@ pub(crate) mod bucket {
 
         use crate::storage::storage_api::ecstore_bucket::replication as replication_contracts;
 
+        pub(crate) use replication_contracts::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;

rustfs/src/storage/ecfs.rs

@@ -63,6 +63,69 @@ use crate::app::storage_api::object_usecase::bucket::replication::{
 };
 use crate::storage::storage_api::ecfs_consumer::StorageObjectOptions as ObjectOptions;
 
+#[cfg(test)]
+static SITE_REPLICATION_GATE_TEST_OVERRIDE: std::sync::atomic::AtomicU8 = std::sync::atomic::AtomicU8::new(0);
+#[cfg(test)]
+const SITE_REPLICATION_GATE_FORCE_DISABLED: u8 = 1;
+#[cfg(test)]
+const SITE_REPLICATION_GATE_FORCE_ENABLED: u8 = 2;
+
+async fn site_replication_gate_enabled() -> S3Result<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 handed to the bucket usecase
+/// so an S3 replication-config edit keeps exactly the reconciler-owned rules
+/// (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_peer_deployment_ids_for_edit() -> S3Result<std::collections::HashSet<String>> {
+    // 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());
+    }
+    crate::admin::handlers::site_replication::site_replication_remote_peer_deployment_ids().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 +563,10 @@ impl S3 for FS {
         &self,
         req: S3Request<DeleteBucketReplicationInput>,
     ) -> S3Result<S3Response<DeleteBucketReplicationOutput>> {
+        deny_replication_config_edit_for_non_owner(&req).await?;
+        let site_peers = site_replication_peer_deployment_ids_for_edit().await?;
         let usecase = s3_api::bucket_usecase_for(self);
-        usecase.execute_delete_bucket_replication(req).await
+        usecase.execute_delete_bucket_replication(req, site_peers).await
     }
 
     #[instrument(level = "debug", skip(self))]
@@ -1353,8 +1418,10 @@ impl S3 for FS {
         &self,
         req: S3Request<PutBucketReplicationInput>,
     ) -> S3Result<S3Response<PutBucketReplicationOutput>> {
+        deny_replication_config_edit_for_non_owner(&req).await?;
+        let site_peers = site_replication_peer_deployment_ids_for_edit().await?;
         let usecase = s3_api::bucket_usecase_for(self);
-        usecase.execute_put_bucket_replication(req).await
+        usecase.execute_put_bucket_replication(req, site_peers).await
     }
 
     async fn put_bucket_request_payment(
@@ -1919,3 +1986,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);
+    }
+}

scripts/check_security_coverage.py

@@ -1,264 +0,0 @@
-#!/usr/bin/env python3
-# Copyright 2024 RustFS Team
-#
-# Licensed under the Apache License, Version 2.0 (the "License");
-# you may not use this file except in compliance with the License.
-# You may obtain a copy of the License at
-#
-#      http://www.apache.org/licenses/LICENSE-2.0
-#
-# Unless required by applicable law or agreed to in writing, software
-# distributed under the License is distributed on an "AS IS" BASIS,
-# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-# See the License for the specific language governing permissions and
-# limitations under the License.
-
-"""Compare security-critical crate line coverage with the report-only baseline."""
-
-import argparse
-import json
-import math
-import os
-import sys
-import tempfile
-import tomllib
-from pathlib import Path
-
-from coverage_per_crate import fmt_pct, load_coverage
-
-
-SECURITY_CRATES = ("crates/iam", "crates/kms", "crates/policy", "crates/crypto")
-
-
-def load_baselines(path: str) -> tuple[float, dict[str, tuple[int, int]]]:
-    with open(path, "rb") as fh:
-        config = tomllib.load(fh)
-
-    if config.get("phase") != "report-only":
-        raise ValueError("coverage baseline phase must be report-only")
-
-    raw_allowed_drop = config["allowed_drop_percentage_points"]
-    if isinstance(raw_allowed_drop, bool) or not isinstance(raw_allowed_drop, (int, float)):
-        raise ValueError("allowed_drop_percentage_points must be a number")
-    allowed_drop = float(raw_allowed_drop)
-    if not math.isfinite(allowed_drop) or allowed_drop < 0:
-        raise ValueError("allowed_drop_percentage_points must be finite and non-negative")
-
-    baselines: dict[str, tuple[int, int]] = {}
-    for crate, values in config["crates"].items():
-        covered = values["covered"]
-        count = values["count"]
-        if type(covered) is not int or type(count) is not int:
-            raise ValueError(f"invalid baseline for {crate}: covered and count must be integers")
-        if covered < 0 or count <= 0 or covered > count:
-            raise ValueError(f"invalid baseline for {crate}: {covered}/{count}")
-        baselines[crate] = (covered, count)
-    missing = [crate for crate in SECURITY_CRATES if crate not in baselines]
-    unexpected = sorted(set(baselines).difference(SECURITY_CRATES))
-    if missing or unexpected:
-        raise ValueError(f"coverage baseline crate set mismatch: missing={missing}, unexpected={unexpected}")
-    return allowed_drop, baselines
-
-
-def compare(
-    current: dict[str, list[int]],
-    baselines: dict[str, tuple[int, int]],
-    allowed_drop: float,
-) -> list[tuple[str, int, int, int, int, float, bool]]:
-    rows = []
-    for crate, (baseline_covered, baseline_count) in baselines.items():
-        if crate not in current:
-            raise ValueError(f"coverage report is missing {crate}")
-        covered, count = current[crate]
-        if type(covered) is not int or type(count) is not int:
-            raise ValueError(f"invalid coverage for {crate}: covered and count must be integers")
-        if covered < 0 or count <= 0 or covered > count:
-            raise ValueError(f"invalid coverage for {crate}: {covered}/{count}")
-        current_pct = 100.0 * covered / count
-        baseline_pct = 100.0 * baseline_covered / baseline_count
-        delta = current_pct - baseline_pct
-        rows.append((crate, covered, count, baseline_covered, baseline_count, delta, delta < -allowed_drop))
-    return rows
-
-
-def print_report(rows: list[tuple[str, int, int, int, int, float, bool]], allowed_drop: float) -> None:
-    print("## Security-critical coverage ratchet (report-only)")
-    print()
-    print(f"Calibration threshold: a drop greater than {allowed_drop:.2f} percentage points is reported as a regression.")
-    print()
-    print("| Crate | Current | Baseline | Delta | Status |")
-    print("|---|---:|---:|---:|---|")
-    for crate, covered, count, baseline_covered, baseline_count, delta, regressed in rows:
-        status = "REGRESSION (report-only)" if regressed else "OK"
-        print(
-            f"| `{crate}` | {fmt_pct(covered, count)} ({covered}/{count}) "
-            f"| {fmt_pct(baseline_covered, baseline_count)} ({baseline_covered}/{baseline_count}) "
-            f"| {delta:+.2f} pp | {status} |"
-        )
-    print()
-    print("This calibration phase records regressions without failing the job; malformed or incomplete evidence still fails closed.")
-
-
-def self_test() -> None:
-    with tempfile.TemporaryDirectory() as tmp:
-        root = Path(tmp)
-        coverage = root / "coverage.json"
-        baseline = root / "baseline.toml"
-        coverage_data = {
-            "data": [
-                {
-                    "files": [
-                        {
-                            "filename": str(root / "crates/iam/src/lib.rs"),
-                            "summary": {"lines": {"covered": 80, "count": 100}},
-                        },
-                        {
-                            "filename": str(root / "crates/kms/src/lib.rs"),
-                            "summary": {"lines": {"covered": 90, "count": 100}},
-                        },
-                        {
-                            "filename": str(root / "crates/policy/src/lib.rs"),
-                            "summary": {"lines": {"covered": 90, "count": 100}},
-                        },
-                        {
-                            "filename": str(root / "crates/crypto/src/lib.rs"),
-                            "summary": {"lines": {"covered": 90, "count": 100}},
-                        },
-                    ],
-                    "totals": {"lines": {"covered": 350, "count": 400}},
-                }
-            ]
-        }
-        coverage.write_text(json.dumps(coverage_data), encoding="utf-8")
-        baseline_text = """phase = "report-only"
-allowed_drop_percentage_points = 1.0
-[crates."crates/iam"]
-covered = 90
-count = 100
-[crates."crates/kms"]
-covered = 85
-count = 100
-[crates."crates/policy"]
-covered = 90
-count = 100
-[crates."crates/crypto"]
-covered = 90
-count = 100
-"""
-        baseline.write_text(baseline_text, encoding="utf-8")
-        current, _ = load_coverage(str(coverage), str(root))
-        allowed_drop, baselines = load_baselines(str(baseline))
-        rows = compare(current, baselines, allowed_drop)
-        assert [row[-1] for row in rows] == [True, False, False, False]
-        try:
-            compare({"crates/iam": current["crates/iam"]}, baselines, allowed_drop)
-        except ValueError as error:
-            assert str(error) == "coverage report is missing crates/kms"
-        else:
-            raise AssertionError("missing crate must fail closed")
-        try:
-            compare({**current, "crates/iam": [101, 100]}, baselines, allowed_drop)
-        except ValueError as error:
-            assert str(error) == "invalid coverage for crates/iam: 101/100"
-        else:
-            raise AssertionError("invalid coverage must fail closed")
-        for invalid_threshold in ("true", '"1.0"', "nan", "inf", "-inf"):
-            baseline.write_text(
-                baseline_text.replace("allowed_drop_percentage_points = 1.0", f"allowed_drop_percentage_points = {invalid_threshold}"),
-                encoding="utf-8",
-            )
-            try:
-                load_baselines(str(baseline))
-            except ValueError:
-                pass
-            else:
-                raise AssertionError(f"non-finite threshold {invalid_threshold} must fail closed")
-        for field, invalid_values in (
-            ("covered", ("true", '"90"', "90.0", "90.5")),
-            ("count", ("true", '"100"', "100.0", "100.5")),
-        ):
-            for invalid_value in invalid_values:
-                baseline.write_text(
-                    baseline_text.replace(f"{field} = {90 if field == 'covered' else 100}", f"{field} = {invalid_value}", 1),
-                    encoding="utf-8",
-                )
-                try:
-                    load_baselines(str(baseline))
-                except ValueError:
-                    pass
-                else:
-                    raise AssertionError(f"non-integer baseline {field} {invalid_value} must fail closed")
-        for covered, count in (
-            (True, 100),
-            (80, True),
-            (80.0, 100),
-            (80, 100.0),
-            (float("nan"), 100),
-            (80, float("inf")),
-        ):
-            try:
-                compare({**current, "crates/iam": [covered, count]}, baselines, allowed_drop)
-            except ValueError:
-                pass
-            else:
-                raise AssertionError(f"invalid aggregate coverage {covered}/{count} must fail closed")
-        lines = coverage_data["data"][0]["files"][0]["summary"]["lines"]
-        for field, invalid_values in (
-            ("covered", (True, "80", 80.0, 80.5, float("nan"), float("inf"), float("-inf"))),
-            ("count", (True, "100", 100.0, 100.5, float("nan"), float("inf"), float("-inf"))),
-        ):
-            original = lines[field]
-            for invalid_value in invalid_values:
-                lines[field] = invalid_value
-                coverage.write_text(json.dumps(coverage_data), encoding="utf-8")
-                try:
-                    load_coverage(str(coverage), str(root))
-                except ValueError:
-                    pass
-                else:
-                    raise AssertionError(f"invalid raw coverage {field} {invalid_value} must fail closed")
-            lines[field] = original
-        baseline.write_text(
-            baseline_text.replace(
-                '[crates."crates/crypto"]\ncovered = 90\ncount = 100\n',
-                "",
-            ),
-            encoding="utf-8",
-        )
-        try:
-            load_baselines(str(baseline))
-        except ValueError:
-            pass
-        else:
-            raise AssertionError("missing security-crate baseline must fail closed")
-
-
-def main() -> int:
-    parser = argparse.ArgumentParser(description=__doc__)
-    parser.add_argument("coverage_json", nargs="?")
-    parser.add_argument("--baseline", default=".config/coverage-baselines.toml")
-    parser.add_argument("--repo-root", default=os.getcwd())
-    parser.add_argument("--self-test", action="store_true")
-    args = parser.parse_args()
-
-    if args.self_test:
-        self_test()
-        print("security coverage self-test passed")
-        return 0
-    if not args.coverage_json:
-        parser.error("coverage_json is required unless --self-test is used")
-
-    try:
-        current, _ = load_coverage(args.coverage_json, os.path.abspath(args.repo_root))
-        allowed_drop, baselines = load_baselines(args.baseline)
-        rows = compare(current, baselines, allowed_drop)
-    except (OSError, ValueError, KeyError, IndexError, json.JSONDecodeError, tomllib.TOMLDecodeError) as error:
-        print(f"error: {error}", file=sys.stderr)
-        return 1
-
-    print_report(rows, allowed_drop)
-    return 0
-
-
-if __name__ == "__main__":
-    sys.exit(main())

scripts/coverage_per_crate.py

@@ -47,31 +47,6 @@ def fmt_pct(covered: int, count: int) -> str:
     return f"{100.0 * covered / count:.2f}%" if count else "—"
 
 
-def _line_counts(lines: dict[str, int], source: str) -> tuple[int, int]:
-    covered = lines["covered"]
-    count = lines["count"]
-    if type(covered) is not int or type(count) is not int or covered < 0 or count < 0 or covered > count:
-        raise ValueError(f"invalid line coverage for {source}: {covered}/{count}")
-    return covered, count
-
-
-def load_coverage(path: str, root: str) -> tuple[dict[str, list[int]], dict[str, int]]:
-    with open(path, encoding="utf-8") as fh:
-        export = json.load(fh)
-
-    data = export["data"][0]
-    files = data["files"]
-    total_covered, total_count = _line_counts(data["totals"]["lines"], "totals")
-
-    crates: dict[str, list[int]] = {}
-    for f in files:
-        covered, count = _line_counts(f["summary"]["lines"], f["filename"])
-        acc = crates.setdefault(crate_label(f["filename"], root), [0, 0])
-        acc[0] += covered
-        acc[1] += count
-    return crates, {"covered": total_covered, "count": total_count}
-
-
 def main() -> int:
     if len(sys.argv) < 2 or len(sys.argv) > 3:
         print(__doc__.strip(), file=sys.stderr)
@@ -79,12 +54,24 @@ def main() -> int:
     path = sys.argv[1]
     root = os.path.abspath(sys.argv[2] if len(sys.argv) == 3 else os.getcwd())
 
+    with open(path, encoding="utf-8") as fh:
+        export = json.load(fh)
+
     try:
-        crates, totals = load_coverage(path, root)
-    except (KeyError, IndexError, ValueError) as exc:
+        data = export["data"][0]
+        files = data["files"]
+        totals = data["totals"]["lines"]
+    except (KeyError, IndexError) as exc:
         print(f"error: unexpected llvm-cov JSON shape ({exc})", file=sys.stderr)
         return 1
 
+    crates: dict[str, list[int]] = {}
+    for f in files:
+        lines = f["summary"]["lines"]
+        acc = crates.setdefault(crate_label(f["filename"], root), [0, 0])
+        acc[0] += lines["covered"]
+        acc[1] += lines["count"]
+
     rows = sorted(
         crates.items(),
         key=lambda kv: (100.0 * kv[1][0] / kv[1][1]) if kv[1][1] else 101.0,

scripts/test/vault_ha_kms_live.sh

@@ -241,7 +241,7 @@ env \
   RUSTFS_TEST_VAULT_FAILOVER_MARKER="$MARKER" \
   RUSTFS_TEST_VAULT_OLD_LEADER="$OLD_LEADER" \
   cargo test -p rustfs-kms --test vault_ha_failover_live \
-    vault_raft_leader_failure_recovers_kv2_and_transit_decrypts -- \
+    vault_raft_leader_failure_preserves_kv2_and_transit_decrypts -- \
     --ignored --nocapture --test-threads=1 &
 TEST_PID=$!