fix(replication): make bucket replication rules editable from clients (#5715)

* fix(replication): accept explicit STANDARD destination storage class

The replication engine never reads Rule.Destination.StorageClass (replica
placement comes from the bucket-target config or the source object), yet the
validator rejected any config carrying the field. The console's add-rule form
always sends StorageClass=STANDARD, so every rule created through it failed
with InvalidRequest.

Tolerate exactly STANDARD as a no-op — semantically identical to omitting
the field — and keep rejecting every other value, which would be silently
ignored rather than honored. Document the deliberate omission from the
replication capability contract.

* feat(admin): support MinIO-style partial updates for set-remote-target

set-remote-target?update=true previously replaced every stored field and
required complete credentials in the body, so flipping a target's sync mode
from the console forced operators to re-enter the secret key, and real
mc replicate update bodies (madmin Clone() strips the secret) failed to
deserialize at all.

Adopt MinIO's TargetUpdateType contract: query params creds/sync/bandwidth/
path name the field groups to overlay onto the stored target, everything
else keeps its persisted value, and unsupported groups (proxy, healthcheck,
edge, edgeSyncBeforeExpiry) fail loudly. Credentials updates are skipped for
site-replication peer targets — probed by both scheme derivations of the
stored endpoint and the stored deployment id — because an operator never
knows the site replicator's credentials, and a body-supplied deployment id
is ignored on update since it anchors peer identity. madmin JSON aliases
(bandwidthlimit, storageclass, resetID, deploymentID, sessionToken) let mc
bodies parse under deny_unknown_fields.

e2e: cover a credential-free sync-only update preserving the stored
connection and the zero-ops no-op contract; align the missing-arn assertion
with the earlier validation error.

* chore(scripts): add two-site replication lab manager

site_replication_smoke.py spawns and manages two local rustfs processes,
pairs them via the site-replication admin API (idempotent), and verifies
bidirectional object replication. Subcommands: up/down/restart/status/logs/
smoke/info/remove/clean. Stdlib-only; requests are SigV4-signed the same
way as crates/e2e_test.

* chore(scripts): rename direction-suffixed payload variables for typos check

The typos linter reads the _ba suffix in payload_ba as a misspelling of
"by"; use payload_a_to_b / payload_b_to_a instead.

---------

Co-authored-by: overtrue <anzhengchao@gmail.com>

crates/e2e_test/src/replication_extension_test.rs

@@ -2754,7 +2754,7 @@ async fn test_set_remote_target_update_requires_arn() -> Result<(), Box<dyn Erro
 
     assert_eq!(status, StatusCode::BAD_REQUEST);
     assert!(body.contains("InvalidRequest"), "unexpected response: {body}");
-    assert!(body.to_ascii_lowercase().contains("arn is empty"), "unexpected response: {body}");
+    assert!(body.to_ascii_lowercase().contains("arn is required"), "unexpected response: {body}");
 
     Ok(())
 }
@@ -2812,6 +2812,128 @@ async fn test_set_remote_target_update_rejects_missing_target() -> Result<(), Bo
     Ok(())
 }
 
+async fn send_set_replication_target_update_request(
+    source_env: &RustFSTestEnvironment,
+    source_bucket: &str,
+    ops: &[&str],
+    body: serde_json::Value,
+) -> Result<reqwest::Response, Box<dyn Error + Send + Sync>> {
+    let mut url = format!(
+        "{}/rustfs/admin/v3/set-remote-target?bucket={}&update=true",
+        source_env.url,
+        urlencoding::encode(source_bucket)
+    );
+    for op in ops {
+        url.push_str(&format!("&{op}=true"));
+    }
+    signed_request(
+        http::Method::PUT,
+        &url,
+        &source_env.access_key,
+        &source_env.secret_key,
+        Some(body.to_string().into_bytes()),
+        Some("application/json"),
+    )
+    .await
+}
+
+async fn fetch_single_target(
+    env: &RustFSTestEnvironment,
+    bucket: &str,
+) -> Result<serde_json::Value, Box<dyn Error + Send + Sync>> {
+    let response = list_replication_targets_request(env, Some(bucket)).await?;
+    assert_eq!(response.status(), StatusCode::OK);
+    let mut targets: Vec<serde_json::Value> = response.json().await?;
+    assert_eq!(targets.len(), 1, "expected exactly one remote target");
+    Ok(targets.remove(0))
+}
+
+#[tokio::test]
+#[serial]
+async fn test_set_remote_target_partial_update_preserves_credentials() -> Result<(), Box<dyn Error + Send + Sync>> {
+    init_logging();
+
+    let mut source_env = RustFSTestEnvironment::new().await?;
+    source_env
+        .start_rustfs_server_with_env(vec![], LOOPBACK_REPLICATION_TARGET_ENV)
+        .await?;
+
+    let mut target_env = RustFSTestEnvironment::new().await?;
+    target_env.start_rustfs_server_without_cleanup(vec![]).await?;
+
+    let source_bucket = "replication-partial-update-src";
+    let target_bucket = "replication-partial-update-dst";
+
+    let source_client = source_env.create_s3_client();
+    let target_client = target_env.create_s3_client();
+
+    source_client.create_bucket().bucket(source_bucket).send().await?;
+    target_client.create_bucket().bucket(target_bucket).send().await?;
+
+    enable_bucket_versioning(&source_env, source_bucket).await?;
+    enable_bucket_versioning(&target_env, target_bucket).await?;
+
+    let arn = set_replication_target(&source_env, source_bucket, &target_env, target_bucket).await?;
+
+    // A sync-only update whose body omits credentials entirely must succeed and
+    // leave the stored connection settings untouched.
+    let response = send_set_replication_target_update_request(
+        &source_env,
+        source_bucket,
+        &["sync"],
+        serde_json::json!({
+            "arn": arn,
+            "type": "replication",
+            "replicationSync": true
+        }),
+    )
+    .await?;
+    assert_eq!(response.status(), StatusCode::OK, "sync-only update failed: {}", response.text().await?);
+
+    let target = fetch_single_target(&source_env, source_bucket).await?;
+    assert_eq!(target["replicationSync"], serde_json::json!(true));
+    assert_eq!(target["endpoint"], serde_json::json!(target_env.address));
+    assert_eq!(target["credentials"]["accessKey"], serde_json::json!(target_env.access_key));
+
+    // An update naming no field groups is a no-op: a body carrying a different
+    // endpoint and credentials must not leak into the stored target.
+    let response = send_set_replication_target_update_request(
+        &source_env,
+        source_bucket,
+        &[],
+        serde_json::json!({
+            "arn": arn,
+            "type": "replication",
+            "endpoint": "203.0.113.1:9000",
+            "credentials": { "accessKey": "other-access", "secretKey": "other-secret" },
+            "targetbucket": "elsewhere",
+            "secure": false,
+            "replicationSync": false
+        }),
+    )
+    .await?;
+    assert_eq!(response.status(), StatusCode::OK, "no-op update failed: {}", response.text().await?);
+
+    let target = fetch_single_target(&source_env, source_bucket).await?;
+    assert_eq!(
+        target["replicationSync"],
+        serde_json::json!(true),
+        "no-op update must not change sync mode"
+    );
+    assert_eq!(
+        target["endpoint"],
+        serde_json::json!(target_env.address),
+        "no-op update must not change endpoint"
+    );
+    assert_eq!(
+        target["credentials"]["accessKey"],
+        serde_json::json!(target_env.access_key),
+        "no-op update must not change credentials"
+    );
+
+    Ok(())
+}
+
 #[tokio::test]
 #[serial]
 async fn test_set_remote_target_rejects_invalid_target_url() -> Result<(), Box<dyn Error + Send + Sync>> {

crates/replication/src/config.rs

@@ -20,7 +20,7 @@ use s3s::dto::DeleteReplicationStatus;
 use s3s::dto::Destination;
 use s3s::dto::{
     ExistingObjectReplicationStatus, ReplicaModificationsStatus, ReplicationConfiguration, ReplicationRule,
-    ReplicationRuleStatus, ReplicationRules,
+    ReplicationRuleStatus, ReplicationRules, StorageClass,
 };
 use serde::{Deserialize, Serialize};
 use std::collections::{BTreeMap, HashMap, HashSet};
@@ -28,6 +28,10 @@ use uuid::Uuid;
 
 pub const REPLICATION_CAPABILITY_CONTRACT_VERSION: u32 = 1;
 
+// `Rule.Destination.StorageClass` is deliberately absent from both lists below:
+// clients should keep omitting it, but the validator tolerates an explicit
+// `STANDARD` as a no-op (see `unsupported_replication_config_field`) because the
+// console's rule form always sends it.
 pub const REPLICATION_WRITABLE_FIELDS: &[&str] = &[
     "Role",
     "Rule.ID",
@@ -164,7 +168,18 @@ pub fn unsupported_replication_config_field(config: &ReplicationConfiguration) -
         if rule.destination.replication_time.is_some() {
             return Some("Destination.ReplicationTime");
         }
-        if rule.destination.storage_class.is_some() {
+        // The replication engine never reads this field (replica placement comes from
+        // the bucket-target config or the source object), so an explicit STANDARD —
+        // which the console's rule form always sends — is indistinguishable from
+        // omitting it and is tolerated as a no-op. Any other value would be silently
+        // ignored rather than honored, so it stays rejected. Exact match: S3 storage
+        // class enums are case-sensitive.
+        if rule
+            .destination
+            .storage_class
+            .as_ref()
+            .is_some_and(|class| class.as_str() != StorageClass::STANDARD)
+        {
             return Some("Destination.StorageClass");
         }
     }
@@ -952,6 +967,47 @@ mod tests {
         config.rules[0].destination.storage_class =
             Some(s3s::dto::StorageClass::from_static(s3s::dto::StorageClass::STANDARD_IA));
         assert_eq!(unsupported_replication_config_field(&config), Some("Destination.StorageClass"));
+
+        // The exact-match contract is deliberate: S3 storage class enums are
+        // case-sensitive, so a lowercase variant must stay rejected.
+        config.rules[0].destination.storage_class = Some(StorageClass::from("standard".to_string()));
+        assert_eq!(unsupported_replication_config_field(&config), Some("Destination.StorageClass"));
+
+        // Explicit STANDARD is a no-op (the engine never reads the field) and must
+        // pass: the console's rule form always sends it, and rejecting it makes the
+        // form unusable.
+        config.rules[0].destination.storage_class = Some(StorageClass::from_static(StorageClass::STANDARD));
+        assert_eq!(unsupported_replication_config_field(&config), None);
+    }
+
+    #[test]
+    fn explicit_standard_storage_class_is_accepted_from_wire_xml() {
+        // The exact request shape the console's add-replication-rule form sends:
+        // a rule whose Destination carries <StorageClass>STANDARD</StorageClass>.
+        let xml = br#"
+            <ReplicationConfiguration>
+              <Role></Role>
+              <Rule>
+                <ID>console-rule</ID>
+                <Status>Enabled</Status>
+                <Priority>1</Priority>
+                <DeleteMarkerReplication><Status>Enabled</Status></DeleteMarkerReplication>
+                <Destination>
+                  <Bucket>arn:aws:s3:::destination</Bucket>
+                  <StorageClass>STANDARD</StorageClass>
+                </Destination>
+              </Rule>
+            </ReplicationConfiguration>
+        "#;
+        let mut deserializer = Deserializer::new(xml);
+        let config = <ReplicationConfiguration as s3s::xml::Deserialize>::deserialize(&mut deserializer)
+            .expect("console-shaped config should parse");
+        deserializer
+            .expect_eof()
+            .expect("console-shaped config should consume the whole body");
+
+        assert_eq!(config.rules[0].destination.storage_class.as_ref().map(|c| c.as_str()), Some("STANDARD"));
+        assert_eq!(unsupported_replication_config_field(&config), None);
     }
 
     #[test]

rustfs/src/admin/handlers/replication.rs

@@ -56,6 +56,55 @@ use url::Host;
 
 const SUPPORTED_REMOTE_TARGET_API: &str = "s3v4";
 
+/// Field groups a `set-remote-target?update=true` request may modify, mirroring
+/// MinIO's `TargetUpdateType` / `GetTargetUpdateOps` query contract: the update
+/// overlays only the requested groups onto the stored target, so a client can
+/// e.g. flip sync mode without knowing or re-sending the target credentials.
+#[derive(Clone, Copy, PartialEq, Eq, Debug)]
+enum TargetUpdateOp {
+    /// Connection group: credentials plus endpoint, target bucket, and TLS settings.
+    Credentials,
+    Sync,
+    Bandwidth,
+    Path,
+}
+
+fn parse_remote_target_update_ops(queries: &HashMap<String, String>) -> S3Result<Vec<TargetUpdateOp>> {
+    const SUPPORTED_OPS: &[(&str, TargetUpdateOp)] = &[
+        ("creds", TargetUpdateOp::Credentials),
+        ("sync", TargetUpdateOp::Sync),
+        ("bandwidth", TargetUpdateOp::Bandwidth),
+        ("path", TargetUpdateOp::Path),
+    ];
+    // Present in the MinIO wire contract, but they drive target fields this
+    // version rejects as unsupported — fail loudly instead of silently ignoring.
+    const UNSUPPORTED_OPS: &[&str] = &["proxy", "healthcheck", "edge", "edgeSyncBeforeExpiry"];
+
+    for key in UNSUPPORTED_OPS {
+        if queries.get(*key).is_some_and(|value| value == "true") {
+            return Err(s3_error!(
+                InvalidRequest,
+                "remote target update op {key} is not supported by this RustFS version"
+            ));
+        }
+    }
+    Ok(SUPPORTED_OPS
+        .iter()
+        .filter(|(key, _)| queries.get(*key).is_some_and(|value| value == "true"))
+        .map(|(_, op)| *op)
+        .collect())
+}
+
+fn site_endpoint_for(endpoint: &str, secure: bool) -> String {
+    if endpoint.starts_with("http://") || endpoint.starts_with("https://") {
+        endpoint.to_string()
+    } else if secure {
+        format!("https://{endpoint}")
+    } else {
+        format!("http://{endpoint}")
+    }
+}
+
 fn extract_query_params(uri: &Uri) -> HashMap<String, String> {
     let mut params = HashMap::new();
 
@@ -84,14 +133,20 @@ fn map_bucket_target_error(err: BucketTargetError) -> S3Error {
     }
 }
 
-#[derive(Deserialize)]
+#[derive(Deserialize, Default)]
 #[serde(deny_unknown_fields)]
 struct RemoteTargetCredentialsRequest {
-    #[serde(rename = "accessKey")]
+    #[serde(rename = "accessKey", default)]
     access_key: String,
-    #[serde(rename = "secretKey")]
+    // madmin's BucketTarget::Clone() strips the secret before mc round-trips a
+    // target, so a non-creds `mc replicate update` body carries accessKey
+    // without secretKey; validate_connection_fields still rejects that shape
+    // for create and creds updates.
+    #[serde(rename = "secretKey", default)]
     secret_key: String,
+    #[serde(alias = "sessionToken", default)]
     session_token: Option<String>,
+    #[serde(default)]
     expiration: Option<Timestamp>,
 }
 
@@ -111,9 +166,14 @@ impl From<RemoteTargetCredentialsRequest> for TargetCredentials {
 struct RemoteTargetRequest {
     #[serde(rename = "sourcebucket", default)]
     source_bucket: String,
+    #[serde(default)]
     endpoint: String,
+    // Defaulted so a partial `update=true` body can omit credentials (or, as mc
+    // does, send accessKey without the secret); the create path and creds
+    // updates still reject incomplete ones.
+    #[serde(default)]
     credentials: RemoteTargetCredentialsRequest,
-    #[serde(rename = "targetbucket")]
+    #[serde(rename = "targetbucket", default)]
     target_bucket: String,
     #[serde(default)]
     secure: bool,
@@ -127,11 +187,12 @@ struct RemoteTargetRequest {
     target_type: BucketTargetType,
     #[serde(default)]
     region: String,
-    #[serde(alias = "bandwidth", default)]
+    // The extra aliases accept madmin's JSON tags so mc bodies deserialize.
+    #[serde(alias = "bandwidth", alias = "bandwidthlimit", default)]
     bandwidth_limit: i64,
     #[serde(rename = "replicationSync", default)]
     replication_sync: bool,
-    #[serde(default)]
+    #[serde(alias = "storageclass", default)]
     storage_class: String,
     #[serde(rename = "skipTlsVerify", default)]
     skip_tls_verify: bool,
@@ -143,7 +204,7 @@ struct RemoteTargetRequest {
     disable_proxy: bool,
     #[serde(rename = "resetBeforeDate", with = "time::serde::rfc3339::option", default)]
     reset_before_date: Option<OffsetDateTime>,
-    #[serde(default)]
+    #[serde(alias = "resetID", default)]
     reset_id: String,
     #[serde(rename = "totalDowntime", default)]
     total_downtime: u64,
@@ -153,7 +214,7 @@ struct RemoteTargetRequest {
     online: bool,
     #[serde(default)]
     latency: LatencyStat,
-    #[serde(default)]
+    #[serde(alias = "deploymentID", default)]
     deployment_id: String,
     #[serde(default)]
     edge: bool,
@@ -164,7 +225,9 @@ struct RemoteTargetRequest {
 }
 
 impl RemoteTargetRequest {
-    fn into_bucket_target(self) -> S3Result<BucketTarget> {
+    /// Connection-group requirements: enforced on create and on `creds` updates,
+    /// where the endpoint/credentials in the body replace the stored ones.
+    fn validate_connection_fields(&self) -> S3Result<()> {
         if self.endpoint.trim().is_empty() {
             return Err(s3_error!(InvalidRequest, "endpoint is required"));
         }
@@ -173,10 +236,6 @@ impl RemoteTargetRequest {
             return Err(s3_error!(InvalidRequest, "targetbucket is required"));
         }
 
-        if !self.target_type.is_valid() {
-            return Err(s3_error!(InvalidRequest, "type is invalid"));
-        }
-
         if self.credentials.access_key.trim().is_empty() {
             return Err(s3_error!(InvalidRequest, "credentials.accessKey is required"));
         }
@@ -185,6 +244,31 @@ impl RemoteTargetRequest {
             return Err(s3_error!(InvalidRequest, "credentials.secretKey is required"));
         }
 
+        Ok(())
+    }
+
+    fn into_bucket_target(self) -> S3Result<BucketTarget> {
+        self.validate_connection_fields()?;
+        self.into_bucket_target_common()
+    }
+
+    /// Partial-update parse: only the field groups named by `ops` are validated;
+    /// everything else may be absent from the body (MinIO clients omit it).
+    fn into_update_bucket_target(self, ops: &[TargetUpdateOp]) -> S3Result<BucketTarget> {
+        if self.arn.trim().is_empty() {
+            return Err(s3_error!(InvalidRequest, "arn is required for update"));
+        }
+        if ops.contains(&TargetUpdateOp::Credentials) {
+            self.validate_connection_fields()?;
+        }
+        self.into_bucket_target_common()
+    }
+
+    fn into_bucket_target_common(self) -> S3Result<BucketTarget> {
+        if !self.target_type.is_valid() {
+            return Err(s3_error!(InvalidRequest, "type is invalid"));
+        }
+
         if self
             .credentials
             .session_token
@@ -455,40 +539,56 @@ impl Operation for SetRemoteTargetHandler {
                 }
             };
 
-        let mut remote_target = serde_json::from_slice::<RemoteTargetRequest>(&body)
-            .map_err(|e| {
-                error!("Failed to parse remote target request body: {}", e);
-                S3Error::with_message(S3ErrorCode::InvalidRequest, format!("invalid remote target request: {e}"))
-            })?
-            .into_bucket_target()?;
-        validate_remote_target_tls_settings(&remote_target)?;
+        let request = serde_json::from_slice::<RemoteTargetRequest>(&body).map_err(|e| {
+            error!("Failed to parse remote target request body: {}", e);
+            S3Error::with_message(S3ErrorCode::InvalidRequest, format!("invalid remote target request: {e}"))
+        })?;
 
-        let Ok(target_url) = remote_target.url() else {
-            return Err(s3_error!(InvalidRequest, "invalid target url"));
-        };
-
-        let same_target = rustfs_utils::net::is_local_host(
-            target_url.host().unwrap_or(Host::Domain("localhost")),
-            target_url.port().unwrap_or(80),
-            current_runtime_port(),
-        )
-        .unwrap_or_default();
-
-        if same_target && bucket == &remote_target.target_bucket {
-            return Err(S3Error::with_message(S3ErrorCode::IncorrectEndpoint, "Same target".to_string()));
-        }
-
-        remote_target.source_bucket = bucket.clone();
-        let site_endpoint = if remote_target.endpoint.starts_with("http://") || remote_target.endpoint.starts_with("https://") {
-            remote_target.endpoint.clone()
-        } else if remote_target.secure {
-            format!("https://{}", remote_target.endpoint)
+        let update_ops = if update {
+            parse_remote_target_update_ops(&queries)?
         } else {
-            format!("http://{}", remote_target.endpoint)
+            Vec::new()
         };
-        if let Some(deployment_id) = site_replication_peer_deployment_id_for_endpoint(&site_endpoint).await {
-            remote_target.deployment_id = deployment_id;
+        let replacing_connection = !update || update_ops.contains(&TargetUpdateOp::Credentials);
+
+        let mut remote_target = if update {
+            request.into_update_bucket_target(&update_ops)?
+        } else {
+            request.into_bucket_target()?
+        };
+
+        // Endpoint, TLS, and credential fields from the body only take effect on
+        // create or a `creds` update; a partial update body may omit them.
+        if replacing_connection {
+            validate_remote_target_tls_settings(&remote_target)?;
+
+            let Ok(target_url) = remote_target.url() else {
+                return Err(s3_error!(InvalidRequest, "invalid target url"));
+            };
+
+            let same_target = rustfs_utils::net::is_local_host(
+                target_url.host().unwrap_or(Host::Domain("localhost")),
+                target_url.port().unwrap_or(80),
+                current_runtime_port(),
+            )
+            .unwrap_or_default();
+
+            if same_target && bucket == &remote_target.target_bucket {
+                return Err(S3Error::with_message(S3ErrorCode::IncorrectEndpoint, "Same target".to_string()));
+            }
+
+            if update {
+                // Never trust a body-supplied deployment id on update: it is a
+                // peer-identity anchor, so derive it from the new endpoint's peer
+                // lookup below (empty when the endpoint is not a peer).
+                remote_target.deployment_id = String::new();
+            }
+            let site_endpoint = site_endpoint_for(&remote_target.endpoint, remote_target.secure);
+            if let Some(deployment_id) = site_replication_peer_deployment_id_for_endpoint(&site_endpoint).await {
+                remote_target.deployment_id = deployment_id;
+            }
         }
+        remote_target.source_bucket = bucket.clone();
 
         let bucket_target_sys = BucketTargetSys::get();
 
@@ -520,17 +620,50 @@ impl Operation for SetRemoteTargetHandler {
                 return Err(S3Error::with_message(S3ErrorCode::InvalidRequest, "Target not found".to_string()));
             };
 
-            target.credentials = remote_target.credentials;
-            target.endpoint = remote_target.endpoint;
-            target.secure = remote_target.secure;
-            target.target_bucket = remote_target.target_bucket;
-
-            target.path = remote_target.path;
-            target.replication_sync = remote_target.replication_sync;
-            target.bandwidth_limit = remote_target.bandwidth_limit;
-            target.skip_tls_verify = remote_target.skip_tls_verify;
-            target.ca_cert_pem = remote_target.ca_cert_pem;
-            target.health_check_duration = remote_target.health_check_duration;
+            // Overlay only the requested field groups onto the stored target
+            // (MinIO `TargetUpdateType` semantics); everything else — including
+            // credentials — stays as persisted.
+            for op in &update_ops {
+                match op {
+                    TargetUpdateOp::Credentials => {
+                        // Mirror MinIO: an operator never knows the site
+                        // replicator's credentials, so overwriting a peer-owned
+                        // target's connection settings would silently break sync.
+                        // Peer ownership is probed three ways — either scheme
+                        // derivation of the stored endpoint (a stale `secure` flag
+                        // must not bypass the guard via a default-port mismatch)
+                        // and the stored deployment id (covers peers registered
+                        // under an alternate NAT/rewritten address).
+                        let https_endpoint = site_endpoint_for(&target.endpoint, true);
+                        let http_endpoint = site_endpoint_for(&target.endpoint, false);
+                        let peer_owned = !target.deployment_id.trim().is_empty()
+                            || site_replication_peer_deployment_id_for_endpoint(&https_endpoint)
+                                .await
+                                .is_some()
+                            || site_replication_peer_deployment_id_for_endpoint(&http_endpoint)
+                                .await
+                                .is_some();
+                        if peer_owned {
+                            warn!(
+                                bucket = %bucket,
+                                arn = %target.arn,
+                                "skip credentials update for site-replication peer target"
+                            );
+                            continue;
+                        }
+                        target.credentials = remote_target.credentials.clone();
+                        target.endpoint = remote_target.endpoint.clone();
+                        target.secure = remote_target.secure;
+                        target.target_bucket = remote_target.target_bucket.clone();
+                        target.skip_tls_verify = remote_target.skip_tls_verify;
+                        target.ca_cert_pem = remote_target.ca_cert_pem.clone();
+                        target.deployment_id = remote_target.deployment_id.clone();
+                    }
+                    TargetUpdateOp::Sync => target.replication_sync = remote_target.replication_sync,
+                    TargetUpdateOp::Bandwidth => target.bandwidth_limit = remote_target.bandwidth_limit,
+                    TargetUpdateOp::Path => target.path = remote_target.path.clone(),
+                }
+            }
 
             warn!(
                 bucket = %bucket,
@@ -539,6 +672,7 @@ impl Operation for SetRemoteTargetHandler {
                 secure = target.secure,
                 skip_tls_verify = target.skip_tls_verify,
                 has_custom_ca = !target.ca_cert_pem.trim().is_empty(),
+                ops = ?update_ops,
                 "update remote target"
             );
             remote_target = target;
@@ -1052,8 +1186,8 @@ impl Operation for ReplicationMrfHandler {
 mod tests {
     use super::{
         REMOTE_TARGET_UNSUPPORTED_FIELDS, REMOTE_TARGET_WRITABLE_FIELDS, RemoteTargetCredentialsRequest, RemoteTargetRequest,
-        SUPPORTED_REMOTE_TARGET_API, build_mrf_response, extract_query_params, unique_replication_peers,
-        validate_remote_target_tls_settings,
+        SUPPORTED_REMOTE_TARGET_API, TargetUpdateOp, build_mrf_response, extract_query_params, parse_remote_target_update_ops,
+        unique_replication_peers, validate_remote_target_tls_settings,
     };
     use crate::admin::storage_api::bucket::target::BucketTarget;
     use crate::admin::storage_api::replication::{BucketStats, DurableMrfBacklog, MrfOpKind, MrfReplicateEntry};
@@ -1072,6 +1206,113 @@ mod tests {
         })
     }
 
+    fn query_map(pairs: &[(&str, &str)]) -> std::collections::HashMap<String, String> {
+        pairs.iter().map(|(k, v)| (k.to_string(), v.to_string())).collect()
+    }
+
+    #[test]
+    fn update_ops_parse_minio_query_contract() {
+        let ops = parse_remote_target_update_ops(&query_map(&[
+            ("update", "true"),
+            ("creds", "true"),
+            ("sync", "true"),
+            ("bandwidth", "true"),
+            ("path", "true"),
+        ]))
+        .expect("supported ops should parse");
+        assert_eq!(
+            ops,
+            vec![
+                TargetUpdateOp::Credentials,
+                TargetUpdateOp::Sync,
+                TargetUpdateOp::Bandwidth,
+                TargetUpdateOp::Path
+            ]
+        );
+
+        assert!(
+            parse_remote_target_update_ops(&query_map(&[("update", "true")]))
+                .expect("no ops should parse")
+                .is_empty()
+        );
+
+        let err = parse_remote_target_update_ops(&query_map(&[("update", "true"), ("healthcheck", "true")]))
+            .expect_err("unsupported op must be rejected");
+        assert!(err.to_string().contains("not supported"), "unexpected error: {err}");
+    }
+
+    #[test]
+    fn update_body_without_creds_op_may_omit_credentials() {
+        let body = serde_json::json!({
+            "arn": "arn:rustfs:replication:us-east-1:dep:target",
+            "type": "replication",
+            "replicationSync": true
+        });
+        let request: RemoteTargetRequest = serde_json::from_value(body).expect("partial update body should deserialize");
+        let target = request
+            .into_update_bucket_target(&[TargetUpdateOp::Sync])
+            .expect("sync-only update must not require credentials");
+        assert!(target.replication_sync);
+
+        // The same partial body must stay rejected when it claims a creds update.
+        let body = serde_json::json!({
+            "arn": "arn:rustfs:replication:us-east-1:dep:target",
+            "type": "replication"
+        });
+        let request: RemoteTargetRequest = serde_json::from_value(body).expect("body should deserialize");
+        let err = request
+            .into_update_bucket_target(&[TargetUpdateOp::Credentials])
+            .expect_err("creds update requires connection fields");
+        assert!(err.to_string().contains("endpoint is required"), "unexpected error: {err}");
+    }
+
+    #[test]
+    fn update_body_accepts_mc_wire_shape() {
+        // What `mc replicate update --sync` actually sends: a madmin
+        // BucketTarget round-trip with the secret stripped by Clone() and
+        // madmin's own JSON tags (bandwidthlimit, deploymentID, ...).
+        let body = serde_json::json!({
+            "sourcebucket": "src",
+            "endpoint": "192.168.1.10:9000",
+            "credentials": { "accessKey": "access" },
+            "targetbucket": "target",
+            "secure": false,
+            "path": "auto",
+            "api": "s3v4",
+            "arn": "arn:rustfs:replication:us-east-1:dep:target",
+            "type": "replication",
+            "bandwidthlimit": 1073741824i64,
+            "replicationSync": true,
+            "totalDowntime": 0,
+            "lastOnline": "2024-01-01T00:00:00Z",
+            "isOnline": true,
+            "latency": { "curr": 0, "avg": 0, "max": 0 },
+            "deploymentID": "dep",
+            "edge": false,
+            "edgeSyncBeforeExpiry": false,
+            "offlineCount": 0
+        });
+        let request: RemoteTargetRequest = serde_json::from_value(body).expect("mc-shaped body should deserialize");
+        let target = request
+            .into_update_bucket_target(&[TargetUpdateOp::Sync, TargetUpdateOp::Bandwidth])
+            .expect("non-creds update must not require the stripped secret");
+        assert!(target.replication_sync);
+        assert_eq!(target.bandwidth_limit, 1073741824);
+    }
+
+    #[test]
+    fn update_body_requires_arn() {
+        let body = serde_json::json!({
+            "type": "replication",
+            "replicationSync": true
+        });
+        let request: RemoteTargetRequest = serde_json::from_value(body).expect("body should deserialize");
+        let err = request
+            .into_update_bucket_target(&[TargetUpdateOp::Sync])
+            .expect_err("update without arn must fail");
+        assert!(err.to_string().contains("arn is required"), "unexpected error: {err}");
+    }
+
     #[test]
     fn cluster_peer_plan_deduplicates_nodes_and_counts_unavailable_slots() {
         let peer = crate::admin::storage_api::runtime::PeerRestClient::new(
@@ -1323,18 +1564,19 @@ mod tests {
 
     #[test]
     fn remote_target_request_rejects_missing_credentials() {
+        // Credentials may be absent at the serde layer (partial update bodies omit
+        // them), but the create path must still reject their absence.
         let mut request = valid_remote_target_request();
         request
             .as_object_mut()
             .expect("request should be an object")
             .remove("credentials");
 
-        let err = match serde_json::from_value::<RemoteTargetRequest>(request) {
-            Ok(_) => panic!("remote target request should require credentials"),
-            Err(err) => err,
-        };
-
-        assert!(err.to_string().contains("missing field"));
+        let request: RemoteTargetRequest = serde_json::from_value(request).expect("body without credentials should deserialize");
+        let err = request
+            .into_bucket_target()
+            .expect_err("create without credentials must fail");
+        assert!(err.to_string().contains("credentials.accessKey is required"), "unexpected error: {err}");
     }
 
     #[test]

scripts/test/site_replication_smoke.py

@@ -0,0 +1,507 @@
+#!/usr/bin/env -S uv run --script
+# /// script
+# requires-python = ">=3.10"
+# dependencies = []
+# ///
+# 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.
+"""Local two-site replication lab: manages two rustfs processes and pairs them.
+
+Standalone replacement for the "Site Replication: A + B" compound in
+.vscode/launch.json — it spawns site A (:9000) and site B (:9020) from
+target/debug/rustfs, waits for both to come up, then calls the admin API to
+configure site replication (idempotent: skipped when the pair already exists).
+
+Usage:
+    ./scripts/test/site_replication_smoke.py            # up: start both + pair
+    ./scripts/test/site_replication_smoke.py status     # process + pair status
+    ./scripts/test/site_replication_smoke.py smoke      # bidirectional object check
+    ./scripts/test/site_replication_smoke.py logs       # tail both server logs
+    ./scripts/test/site_replication_smoke.py down       # stop both processes
+    ./scripts/test/site_replication_smoke.py clean      # down + wipe site data
+
+State lives under target/: volumes in target/volume/site-{a,b}/test{1..4},
+logs and pidfiles in target/logs/site-{a,b}. Build the server first with
+`cargo build --bin rustfs`. Requests are SigV4-signed the same way as
+crates/e2e_test (service "s3", region "us-east-1", UNSIGNED-PAYLOAD).
+"""
+
+from __future__ import annotations
+
+import argparse
+import datetime
+import hashlib
+import hmac
+import json
+import os
+import shutil
+import signal
+import socket
+import subprocess
+import sys
+import time
+import urllib.error
+import urllib.parse
+import urllib.request
+import uuid
+from dataclasses import dataclass
+from pathlib import Path
+
+REPO_ROOT = Path(__file__).resolve().parents[2]
+ADMIN_PREFIX = "/rustfs/admin/v3"
+REGION = "us-east-1"
+SERVICE = "s3"
+UNSIGNED_PAYLOAD = "UNSIGNED-PAYLOAD"
+STOP_GRACE_SECONDS = 10.0
+
+
+@dataclass
+class Site:
+    name: str
+    port: int
+    console_port: int
+    access_key: str
+    secret_key: str
+
+    @property
+    def endpoint(self) -> str:
+        return f"http://127.0.0.1:{self.port}"
+
+    @property
+    def volume_dir(self) -> Path:
+        return REPO_ROOT / "target" / "volume" / self.name
+
+    @property
+    def log_dir(self) -> Path:
+        return REPO_ROOT / "target" / "logs" / self.name
+
+    @property
+    def stdout_log(self) -> Path:
+        return self.log_dir / "stdout.log"
+
+    @property
+    def pid_file(self) -> Path:
+        return self.log_dir / "rustfs.pid"
+
+
+# ---------------------------------------------------------------------------
+# SigV4 signing (stdlib only)
+# ---------------------------------------------------------------------------
+
+
+def _hmac(key: bytes, msg: str) -> bytes:
+    return hmac.new(key, msg.encode(), hashlib.sha256).digest()
+
+
+def _uri_encode(value: str, encode_slash: bool) -> str:
+    safe = "-._~" + ("" if encode_slash else "/")
+    return urllib.parse.quote(value, safe=safe)
+
+
+def _canonical_query(query: str) -> str:
+    if not query:
+        return ""
+    pairs = urllib.parse.parse_qsl(query, keep_blank_values=True)
+    encoded = sorted((_uri_encode(k, True), _uri_encode(v, True)) for k, v in pairs)
+    return "&".join(f"{k}={v}" for k, v in encoded)
+
+
+def signed_request(
+    site: Site,
+    method: str,
+    path: str,
+    query: str = "",
+    body: bytes | None = None,
+    content_type: str | None = None,
+    timeout: float = 15.0,
+) -> tuple[int, bytes]:
+    """Send a SigV4-signed request; returns (status_code, body_bytes)."""
+    now = datetime.datetime.now(datetime.timezone.utc)
+    amz_date = now.strftime("%Y%m%dT%H%M%SZ")
+    date_stamp = now.strftime("%Y%m%d")
+
+    headers = {
+        "host": f"127.0.0.1:{site.port}",
+        "x-amz-content-sha256": UNSIGNED_PAYLOAD,
+        "x-amz-date": amz_date,
+    }
+    if content_type:
+        headers["content-type"] = content_type
+
+    signed_names = ";".join(sorted(headers))
+    canonical_headers = "".join(f"{k}:{headers[k].strip()}\n" for k in sorted(headers))
+    canonical_request = "\n".join(
+        [
+            method,
+            _uri_encode(path, False),
+            _canonical_query(query),
+            canonical_headers,
+            signed_names,
+            UNSIGNED_PAYLOAD,
+        ]
+    )
+
+    scope = f"{date_stamp}/{REGION}/{SERVICE}/aws4_request"
+    string_to_sign = "\n".join(
+        [
+            "AWS4-HMAC-SHA256",
+            amz_date,
+            scope,
+            hashlib.sha256(canonical_request.encode()).hexdigest(),
+        ]
+    )
+    key = _hmac(_hmac(_hmac(_hmac(f"AWS4{site.secret_key}".encode(), date_stamp), REGION), SERVICE), "aws4_request")
+    signature = hmac.new(key, string_to_sign.encode(), hashlib.sha256).hexdigest()
+    headers["authorization"] = (
+        f"AWS4-HMAC-SHA256 Credential={site.access_key}/{scope}, SignedHeaders={signed_names}, Signature={signature}"
+    )
+
+    url = f"{site.endpoint}{urllib.parse.quote(path, safe='/-._~')}"
+    if query:
+        url += f"?{query}"
+    request = urllib.request.Request(url, data=body, method=method)
+    for name, value in headers.items():
+        if name != "host":
+            request.add_header(name, value)
+    try:
+        with urllib.request.urlopen(request, timeout=timeout) as response:
+            return response.status, response.read()
+    except urllib.error.HTTPError as err:
+        return err.code, err.read()
+
+
+def admin(site: Site, method: str, subpath: str, query: str = "", payload: object | None = None) -> tuple[int, bytes]:
+    body = None
+    content_type = None
+    if payload is not None:
+        body = json.dumps(payload).encode()
+        content_type = "application/json"
+    return signed_request(site, method, f"{ADMIN_PREFIX}/{subpath}", query, body, content_type)
+
+
+# ---------------------------------------------------------------------------
+# Process management
+# ---------------------------------------------------------------------------
+
+
+def read_pid(site: Site) -> int | None:
+    try:
+        pid = int(site.pid_file.read_text().strip())
+    except (FileNotFoundError, ValueError):
+        return None
+    try:
+        os.kill(pid, 0)
+    except (ProcessLookupError, PermissionError):
+        site.pid_file.unlink(missing_ok=True)
+        return None
+    return pid
+
+
+def port_in_use(port: int) -> bool:
+    with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as sock:
+        sock.settimeout(0.5)
+        return sock.connect_ex(("127.0.0.1", port)) == 0
+
+
+def start_site(site: Site, binary: Path, console: bool) -> None:
+    if (pid := read_pid(site)) is not None:
+        print(f"[ok] {site.name} already running (pid {pid}, {site.endpoint})")
+        return
+    if port_in_use(site.port):
+        raise SystemExit(
+            f"[fail] port {site.port} is in use but not managed by this script; "
+            f"stop the other process first (lsof -iTCP:{site.port} -sTCP:LISTEN)"
+        )
+
+    for index in range(1, 5):
+        (site.volume_dir / f"test{index}").mkdir(parents=True, exist_ok=True)
+    site.log_dir.mkdir(parents=True, exist_ok=True)
+
+    env = os.environ.copy()
+    env.setdefault("RUST_LOG", "rustfs=info,ecstore=warn,s3s=warn,iam=info")
+    env.update(
+        {
+            "RUSTFS_ACCESS_KEY": site.access_key,
+            "RUSTFS_SECRET_KEY": site.secret_key,
+            "RUSTFS_VOLUMES": f"./target/volume/{site.name}/test{{1...4}}",
+            "RUSTFS_ADDRESS": f":{site.port}",
+            "RUSTFS_SERVER_DOMAINS": f"127.0.0.1:{site.port}",
+            "RUSTFS_CONSOLE_ENABLE": "true" if console else "false",
+            "RUSTFS_CONSOLE_ADDRESS": f"127.0.0.1:{site.console_port}",
+            "RUSTFS_OBS_LOG_DIRECTORY": f"./target/logs/{site.name}",
+            "RUSTFS_UNSAFE_BYPASS_DISK_CHECK": "true",
+            "RUSTFS_REPLICATION_ALLOW_LOOPBACK_TARGET": "true",
+            # Let a locally running console dev server (pnpm dev) reach the S3/admin API.
+            "RUSTFS_CORS_ALLOWED_ORIGINS": "http://localhost:3000,http://127.0.0.1:3000",
+        }
+    )
+
+    with site.stdout_log.open("ab") as log:
+        process = subprocess.Popen(
+            [str(binary)],
+            cwd=REPO_ROOT,
+            env=env,
+            stdout=log,
+            stderr=subprocess.STDOUT,
+            start_new_session=True,
+        )
+    site.pid_file.write_text(f"{process.pid}\n")
+    print(f"[ok] started {site.name} (pid {process.pid}, {site.endpoint}, log {site.stdout_log.relative_to(REPO_ROOT)})")
+
+
+def stop_site(site: Site) -> None:
+    pid = read_pid(site)
+    if pid is None:
+        print(f"[ok] {site.name} not running")
+        return
+    os.kill(pid, signal.SIGTERM)
+    deadline = time.monotonic() + STOP_GRACE_SECONDS
+    while time.monotonic() < deadline:
+        try:
+            os.kill(pid, 0)
+        except ProcessLookupError:
+            break
+        time.sleep(0.2)
+    else:
+        print(f"[warn] {site.name} (pid {pid}) ignored SIGTERM, sending SIGKILL")
+        try:
+            os.kill(pid, signal.SIGKILL)
+        except ProcessLookupError:
+            pass
+    site.pid_file.unlink(missing_ok=True)
+    print(f"[ok] stopped {site.name} (pid {pid})")
+
+
+def wait_ready(sites: list[Site], timeout: float) -> None:
+    deadline = time.monotonic() + timeout
+    for site in sites:
+        while True:
+            if read_pid(site) is None:
+                raise SystemExit(f"[fail] {site.name} exited during startup; check {site.stdout_log.relative_to(REPO_ROOT)}")
+            try:
+                status, _ = signed_request(site, "GET", "/", timeout=3.0)
+                if status < 500:
+                    print(f"[ok] {site.name} is ready at {site.endpoint}")
+                    break
+            except (urllib.error.URLError, OSError, TimeoutError):
+                pass
+            if time.monotonic() > deadline:
+                raise SystemExit(f"[fail] {site.name} ({site.endpoint}) not ready within {timeout:.0f}s")
+            time.sleep(1.0)
+
+
+# ---------------------------------------------------------------------------
+# Site replication
+# ---------------------------------------------------------------------------
+
+
+def pair_state(site: Site) -> dict:
+    status, body = admin(site, "GET", "site-replication/info")
+    if status != 200:
+        raise SystemExit(f"[fail] site-replication info: HTTP {status} {body.decode(errors='replace')}")
+    return json.loads(body)
+
+
+def ensure_pair(site_a: Site, site_b: Site) -> None:
+    info = pair_state(site_a)
+    if info.get("enabled") and len(info.get("sites", [])) >= 2:
+        endpoints = ", ".join(peer.get("endpoint", "?") for peer in info["sites"])
+        print(f"[ok] site replication already configured ({endpoints})")
+        return
+
+    peers = [
+        {"name": s.name, "endpoints": s.endpoint, "accessKey": s.access_key, "secretKey": s.secret_key}
+        for s in (site_a, site_b)
+    ]
+    status, body = admin(site_a, "PUT", "site-replication/add", "replicateILMExpiry=false", peers)
+    if status != 200:
+        text = body.decode(errors="replace")
+        hint = ""
+        if "non-empty" in text:
+            hint = f"\n  hint: both sites already hold data; run `{sys.argv[0]} clean` for a fresh pair"
+        raise SystemExit(f"[fail] site-replication add: HTTP {status} {text}{hint}")
+    result = json.loads(body)
+    if not result.get("success", False):
+        raise SystemExit(f"[fail] site-replication add rejected: {json.dumps(result, indent=2)}")
+    print(f"[ok] site replication configured: {result.get('status', '')}")
+
+
+def remove_pair(site: Site) -> None:
+    status, body = admin(site, "PUT", "site-replication/remove", payload={"all": True})
+    if status != 200:
+        raise SystemExit(f"[fail] site-replication remove: HTTP {status} {body.decode(errors='replace')}")
+    print(f"[ok] site replication removed: {body.decode(errors='replace')}")
+
+
+# ---------------------------------------------------------------------------
+# Smoke test
+# ---------------------------------------------------------------------------
+
+
+def put_object(site: Site, bucket: str, key: str, data: bytes) -> None:
+    status, body = signed_request(site, "PUT", f"/{bucket}/{key}", body=data, content_type="application/octet-stream")
+    if status != 200:
+        raise SystemExit(f"[fail] PUT {site.name}/{bucket}/{key}: HTTP {status} {body.decode(errors='replace')}")
+
+
+def wait_object(site: Site, bucket: str, key: str, expected: bytes, timeout: float) -> None:
+    deadline = time.monotonic() + timeout
+    last = "no response yet"
+    while time.monotonic() < deadline:
+        status, body = signed_request(site, "GET", f"/{bucket}/{key}")
+        if status == 200 and body == expected:
+            print(f"[ok] {key} replicated to {site.name}")
+            return
+        last = f"HTTP {status}" if status != 200 else "body mismatch"
+        time.sleep(1.0)
+    raise SystemExit(f"[fail] {key} did not appear on {site.name} within {timeout:.0f}s (last: {last})")
+
+
+def smoke(site_a: Site, site_b: Site, timeout: float) -> None:
+    bucket = f"sr-smoke-{uuid.uuid4().hex[:8]}"
+    status, body = signed_request(site_a, "PUT", f"/{bucket}")
+    if status != 200:
+        raise SystemExit(f"[fail] create bucket {bucket} on {site_a.name}: HTTP {status} {body.decode(errors='replace')}")
+    print(f"[ok] created bucket {bucket} on {site_a.name}")
+
+    # Bucket creation itself must replicate before objects can flow.
+    deadline = time.monotonic() + timeout
+    while time.monotonic() < deadline:
+        status, _ = signed_request(site_b, "GET", f"/{bucket}", query="location=")
+        if status == 200:
+            print(f"[ok] bucket {bucket} replicated to {site_b.name}")
+            break
+        time.sleep(1.0)
+    else:
+        raise SystemExit(f"[fail] bucket {bucket} did not replicate to {site_b.name} within {timeout:.0f}s")
+
+    payload_a_to_b = f"hello from {site_a.name} {uuid.uuid4()}".encode()
+    put_object(site_a, bucket, "from-a.txt", payload_a_to_b)
+    wait_object(site_b, bucket, "from-a.txt", payload_a_to_b, timeout)
+
+    payload_b_to_a = f"hello from {site_b.name} {uuid.uuid4()}".encode()
+    put_object(site_b, bucket, "from-b.txt", payload_b_to_a)
+    wait_object(site_a, bucket, "from-b.txt", payload_b_to_a, timeout)
+
+    print(f"[ok] bidirectional replication verified via bucket {bucket}")
+
+
+# ---------------------------------------------------------------------------
+# Commands
+# ---------------------------------------------------------------------------
+
+
+def cmd_up(sites: list[Site], binary: Path, console: bool, timeout: float) -> None:
+    if not binary.is_file():
+        raise SystemExit(f"[fail] {binary} not found; build it first: cargo build --bin rustfs")
+    for site in sites:
+        start_site(site, binary, console)
+    wait_ready(sites, timeout)
+    ensure_pair(sites[0], sites[1])
+    print("[ok] lab is up:")
+    for site in sites:
+        print(f"      {site.name}: {site.endpoint}  (admin {site.access_key}/{site.secret_key})")
+
+
+def cmd_status(sites: list[Site]) -> None:
+    any_up = False
+    for site in sites:
+        pid = read_pid(site)
+        if pid is not None:
+            any_up = True
+            print(f"[ok] {site.name}: running (pid {pid}, {site.endpoint})")
+        else:
+            print(f"[--] {site.name}: stopped")
+    if not any_up:
+        return
+    try:
+        info = pair_state(sites[0])
+    except SystemExit as err:
+        print(err)
+        return
+    if info.get("enabled"):
+        print(f"[ok] site replication enabled, peers: {', '.join(p.get('endpoint', '?') for p in info.get('sites', []))}")
+    else:
+        print("[--] site replication not configured")
+
+
+def cmd_logs(sites: list[Site], lines: int) -> None:
+    for site in sites:
+        print(f"===== {site.name} ({site.stdout_log.relative_to(REPO_ROOT)}) =====")
+        try:
+            content = site.stdout_log.read_text(errors="replace").splitlines()
+        except FileNotFoundError:
+            print("(no log yet)")
+            continue
+        for line in content[-lines:]:
+            print(line)
+
+
+def cmd_clean(sites: list[Site]) -> None:
+    for site in sites:
+        stop_site(site)
+    for site in sites:
+        if site.volume_dir.exists():
+            shutil.rmtree(site.volume_dir)
+            print(f"[ok] wiped {site.volume_dir.relative_to(REPO_ROOT)}")
+    print("[ok] clean; next `up` starts a fresh pair")
+
+
+def main() -> None:
+    parser = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter)
+    parser.add_argument(
+        "command",
+        nargs="?",
+        default="up",
+        choices=["up", "down", "restart", "status", "logs", "smoke", "info", "remove", "clean"],
+    )
+    parser.add_argument("--port-a", type=int, default=9000, help="site A S3 port (default: %(default)s)")
+    parser.add_argument("--port-b", type=int, default=9020, help="site B S3 port (default: %(default)s)")
+    parser.add_argument("--access-key", default="rustfsadmin")
+    parser.add_argument("--secret-key", default="rustfsadmin")
+    parser.add_argument("--binary", type=Path, default=REPO_ROOT / "target" / "debug" / "rustfs")
+    parser.add_argument("--console", action="store_true", help="also start the web console (ports 9001/9021)")
+    parser.add_argument("--timeout", type=float, default=60.0, help="per-step wait timeout in seconds")
+    parser.add_argument("--lines", type=int, default=30, help="log lines per site for `logs`")
+    args = parser.parse_args()
+
+    site_a = Site("site-a", args.port_a, args.port_a + 1, args.access_key, args.secret_key)
+    site_b = Site("site-b", args.port_b, args.port_b + 1, args.access_key, args.secret_key)
+    sites = [site_a, site_b]
+
+    if args.command == "up":
+        cmd_up(sites, args.binary, args.console, args.timeout)
+    elif args.command == "down":
+        for site in sites:
+            stop_site(site)
+    elif args.command == "restart":
+        for site in sites:
+            stop_site(site)
+        cmd_up(sites, args.binary, args.console, args.timeout)
+    elif args.command == "status":
+        cmd_status(sites)
+    elif args.command == "logs":
+        cmd_logs(sites, args.lines)
+    elif args.command == "smoke":
+        smoke(site_a, site_b, args.timeout)
+    elif args.command == "info":
+        print(json.dumps(pair_state(site_a), indent=2, ensure_ascii=False))
+    elif args.command == "remove":
+        remove_pair(site_a)
+    elif args.command == "clean":
+        cmd_clean(sites)
+
+
+if __name__ == "__main__":
+    main()