diff --git a/crates/e2e_test/src/replication_extension_test.rs b/crates/e2e_test/src/replication_extension_test.rs index ce4bc633d..e07cf7c22 100644 --- a/crates/e2e_test/src/replication_extension_test.rs +++ b/crates/e2e_test/src/replication_extension_test.rs @@ -2754,7 +2754,7 @@ async fn test_set_remote_target_update_requires_arn() -> Result<(), Box Result<(), Bo Ok(()) } +async fn send_set_replication_target_update_request( + source_env: &RustFSTestEnvironment, + source_bucket: &str, + ops: &[&str], + body: serde_json::Value, +) -> Result> { + 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> { + let response = list_replication_targets_request(env, Some(bucket)).await?; + assert_eq!(response.status(), StatusCode::OK); + let mut targets: Vec = 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> { + 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> { diff --git a/crates/replication/src/config.rs b/crates/replication/src/config.rs index ddfdad183..c94faab24 100644 --- a/crates/replication/src/config.rs +++ b/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 STANDARD. + let xml = br#" + + + + console-rule + Enabled + 1 + Enabled + + arn:aws:s3:::destination + STANDARD + + + + "#; + let mut deserializer = Deserializer::new(xml); + let config = ::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] diff --git a/rustfs/src/admin/handlers/replication.rs b/rustfs/src/admin/handlers/replication.rs index 2954f14e5..17fb3edea 100644 --- a/rustfs/src/admin/handlers/replication.rs +++ b/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) -> S3Result> { + 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 { 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, + #[serde(default)] expiration: Option, } @@ -111,9 +166,14 @@ impl From 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, - #[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 { + /// 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 { + 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 { + 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 { + 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::(&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::(&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 { + 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::(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] diff --git a/scripts/test/site_replication_smoke.py b/scripts/test/site_replication_smoke.py new file mode 100755 index 000000000..be6b94563 --- /dev/null +++ b/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()