From c9622611ed5d5832a36f1bf2588a9b1f7f2151f2 Mon Sep 17 00:00:00 2001 From: houseme Date: Wed, 29 Apr 2026 11:27:55 +0800 Subject: [PATCH] fix(admin): harden site-replication identity handling and local verification for issue 2723 (#2730) --- docker-compose-simple.yml | 6 + docker-compose.yml | 13 +- rustfs/src/admin/handlers/site_replication.rs | 307 +++++++++++++----- rustfs/src/admin/mod.rs | 1 + rustfs/src/admin/service/site_replication.rs | 183 ++++++++++- rustfs/src/admin/site_replication_identity.rs | 140 ++++++++ .../validate_issue_2723_site_replication.sh | 298 +++++++++++++++++ 7 files changed, 864 insertions(+), 84 deletions(-) create mode 100644 rustfs/src/admin/site_replication_identity.rs create mode 100755 scripts/validate_issue_2723_site_replication.sh diff --git a/docker-compose-simple.yml b/docker-compose-simple.yml index 96b9dc232..645320e68 100644 --- a/docker-compose-simple.yml +++ b/docker-compose-simple.yml @@ -49,6 +49,9 @@ services: - rustfs-network restart: unless-stopped healthcheck: + # Production strict TLS example (SAN/FQDN aligned, no `-k`): + # curl -f --cacert /opt/tls/ca.crt --resolve rustfs-a.example.com:9000:127.0.0.1 https://rustfs-a.example.com:9000/health + # curl -f --cacert /opt/tls/ca.crt --resolve rustfs-a.example.com:9001:127.0.0.1 https://rustfs-a.example.com:9001/rustfs/console/health test: [ "CMD", @@ -75,6 +78,9 @@ services: echo 'Volume Permissions fixed' && exit 0 " + # Permission baseline: + # - default RustFS runtime user is 10001:10001 + # - alternatively, run rustfs service with host-matched `user: \":\"` restart: "no" networks: diff --git a/docker-compose.yml b/docker-compose.yml index 2a668070f..391840a6b 100644 --- a/docker-compose.yml +++ b/docker-compose.yml @@ -42,11 +42,22 @@ services: volumes: - ./deploy/data/pro:/data - ./deploy/logs:/app/logs - - ./deploy/data/certs/:/opt/tls # TLS configuration, you should create tls directory and put your tls files in it and then specify the path here + # TLS configuration directory. + # Place at least: + # - /opt/tls/ca.crt + # - /opt/tls/rustfs_cert.pem + # - /opt/tls/rustfs_key.pem + - ./deploy/data/certs/:/opt/tls + # Permission baseline: + # - default RustFS runtime user is 10001:10001 + # - ensure host mounts are writable by that user, or run with host-matched user networks: - rustfs-network restart: unless-stopped healthcheck: + # Production strict TLS example (SAN/FQDN aligned, no `-k`): + # curl -f --cacert /opt/tls/ca.crt --resolve rustfs-a.example.com:9000:127.0.0.1 https://rustfs-a.example.com:9000/health + # curl -f --cacert /opt/tls/ca.crt --resolve rustfs-a.example.com:9001:127.0.0.1 https://rustfs-a.example.com:9001/rustfs/console/health test: [ "CMD", diff --git a/rustfs/src/admin/handlers/site_replication.rs b/rustfs/src/admin/handlers/site_replication.rs index 3e662932e..2f22678c9 100644 --- a/rustfs/src/admin/handlers/site_replication.rs +++ b/rustfs/src/admin/handlers/site_replication.rs @@ -14,6 +14,10 @@ use crate::admin::auth::validate_admin_request; use crate::admin::router::{AdminOperation, Operation, S3Router}; +use crate::admin::site_replication_identity::{ + canonical_endpoint, deployment_id_for_endpoint, normalize_peer_map_by_identity_with, same_identity_endpoint, + site_identity_key, +}; use crate::admin::utils::{encode_compatible_admin_payload, read_compatible_admin_body}; use crate::auth::{check_key_valid, get_session_token}; use crate::error::ApiError; @@ -25,7 +29,10 @@ use http::header::{CONTENT_TYPE, HOST}; use http::{HeaderMap, HeaderValue, Uri}; use hyper::{Method, StatusCode}; use matchit::Params; -use rustfs_config::{DEFAULT_DELIMITER, MAX_ADMIN_REQUEST_BODY_SIZE}; +use rustfs_config::{ + DEFAULT_DELIMITER, DEFAULT_RUSTFS_TLS_PATH, DEFAULT_TRUST_LEAF_CERT_AS_CA, ENV_RUSTFS_TLS_PATH, ENV_TRUST_LEAF_CERT_AS_CA, + MAX_ADMIN_REQUEST_BODY_SIZE, RUSTFS_CA_CERT, RUSTFS_TLS_CERT, +}; use rustfs_ecstore::bucket::bucket_target_sys::BucketTargetSys; use rustfs_ecstore::bucket::metadata::{ BUCKET_CORS_CONFIG, BUCKET_LIFECYCLE_CONFIG, BUCKET_POLICY_CONFIG, BUCKET_QUOTA_CONFIG_FILE, BUCKET_REPLICATION_CONFIG, @@ -60,6 +67,7 @@ use rustfs_policy::policy::{ }; use rustfs_signer::constants::UNSIGNED_PAYLOAD; use rustfs_signer::sign_v4; +use rustfs_utils::http::get_source_scheme; use s3s::dto::{ BucketVersioningStatus, DeleteMarkerReplication, DeleteMarkerReplicationStatus, DeleteReplication, DeleteReplicationStatus, Destination, ExistingObjectReplication, ExistingObjectReplicationStatus, ReplicationConfiguration, ReplicationRule, @@ -71,8 +79,7 @@ use serde::Serialize; use serde::de::DeserializeOwned; use serde_json::Value; use sha2::{Digest, Sha256}; -use std::collections::{BTreeMap, HashMap, HashSet, hash_map::DefaultHasher}; -use std::hash::{Hash, Hasher}; +use std::collections::{BTreeMap, HashMap, HashSet}; use std::sync::OnceLock; use std::time::{Duration, Instant}; use time::OffsetDateTime; @@ -94,7 +101,7 @@ const SITE_REPLICATOR_SERVICE_ACCOUNT: &str = "site-replicator-0"; const SITE_REPLICATION_PEER_JOIN_PATH: &str = "/rustfs/admin/v3/site-replication/peer/join"; const SITE_REPLICATION_PEER_EDIT_PATH: &str = "/rustfs/admin/v3/site-replication/peer/edit"; const SITE_REPLICATION_PEER_REMOVE_PATH: &str = "/rustfs/admin/v3/site-replication/peer/remove"; -static SITE_REPLICATION_PEER_CLIENT: OnceLock = OnceLock::new(); +static SITE_REPLICATION_PEER_CLIENT: OnceLock> = OnceLock::new(); #[derive(Debug, Clone, Serialize, Deserialize, Default)] struct SiteReplicationState { @@ -378,9 +385,22 @@ async fn load_site_replication_state() -> S3Result { return Err(S3Error::with_message(S3ErrorCode::InternalError, "Not init".to_string())); }; - match read_config(store, SITE_REPLICATION_STATE_PATH).await { - Ok(data) => serde_json::from_slice(&data) - .map_err(|e| S3Error::with_message(S3ErrorCode::InternalError, format!("invalid site replication state: {e}"))), + match read_config(store.clone(), SITE_REPLICATION_STATE_PATH).await { + Ok(data) => { + let mut state: SiteReplicationState = serde_json::from_slice(&data) + .map_err(|e| S3Error::with_message(S3ErrorCode::InternalError, format!("invalid site replication state: {e}")))?; + let original_state = serde_json::to_vec(&state).ok(); + state.peers = normalize_peer_map_by_identity(state.peers); + let normalized_state = serde_json::to_vec(&state).ok(); + if original_state != normalized_state + && let Some(data) = normalized_state + { + save_config(store, SITE_REPLICATION_STATE_PATH, data).await.map_err(|e| { + S3Error::with_message(S3ErrorCode::InternalError, format!("normalize site replication state failed: {e}")) + })?; + } + Ok(state) + } Err(StorageError::ConfigNotFound) => Ok(SiteReplicationState::default()), Err(err) => Err(S3Error::with_message( S3ErrorCode::InternalError, @@ -394,7 +414,10 @@ async fn save_site_replication_state(state: &SiteReplicationState) -> S3Result<( return Err(S3Error::with_message(S3ErrorCode::InternalError, "Not init".to_string())); }; - let data = serde_json::to_vec(state) + let mut normalized = state.clone(); + normalized.peers = normalize_peer_map_by_identity(normalized.peers); + + let data = serde_json::to_vec(&normalized) .map_err(|e| S3Error::with_message(S3ErrorCode::InternalError, format!("serialize state failed: {e}")))?; save_config(store, SITE_REPLICATION_STATE_PATH, data) .await @@ -414,24 +437,98 @@ async fn clear_site_replication_state() -> S3Result<()> { } async fn persist_site_replication_state(state: &SiteReplicationState) -> S3Result<()> { - if state.peers.len() <= 1 { + let mut normalized = state.clone(); + normalized.peers = normalize_peer_map_by_identity(normalized.peers); + if normalized.peers.len() <= 1 { clear_site_replication_state().await } else { - save_site_replication_state(state).await + save_site_replication_state(&normalized).await } } -fn site_replication_peer_client() -> &'static reqwest::Client { - SITE_REPLICATION_PEER_CLIENT.get_or_init(|| { - reqwest::Client::builder() - .timeout(SITE_REPLICATION_PEER_REQUEST_TIMEOUT) - .connect_timeout(SITE_REPLICATION_PEER_CONNECT_TIMEOUT) - .pool_idle_timeout(Some(Duration::from_secs(60))) - .build() - .unwrap_or_else(|_| reqwest::Client::new()) +fn add_root_certificates_from_file( + mut builder: reqwest::ClientBuilder, + cert_path: &std::path::Path, + description: &str, +) -> S3Result { + if !cert_path.exists() { + return Ok(builder); + } + + std::fs::read(cert_path).map_err(|e| { + S3Error::with_message( + S3ErrorCode::InternalError, + format!("failed to read {description} {}: {e}", cert_path.display()), + ) + })?; + + let certs_der = rustfs_utils::load_cert_bundle_der_bytes(cert_path.to_string_lossy().as_ref()).map_err(|e| { + S3Error::with_message( + S3ErrorCode::InternalError, + format!("failed to parse {description} {}: {e}", cert_path.display()), + ) + })?; + + for cert_der in certs_der { + let cert = reqwest::Certificate::from_der(&cert_der).map_err(|e| { + S3Error::with_message( + S3ErrorCode::InternalError, + format!("failed to load {description} {}: {e}", cert_path.display()), + ) + })?; + builder = builder.add_root_certificate(cert); + } + + Ok(builder) +} + +fn build_site_replication_peer_client() -> S3Result { + let mut builder = reqwest::Client::builder() + .timeout(SITE_REPLICATION_PEER_REQUEST_TIMEOUT) + .connect_timeout(SITE_REPLICATION_PEER_CONNECT_TIMEOUT) + .pool_idle_timeout(Some(Duration::from_secs(60))); + + let tls_path = rustfs_utils::get_env_str(ENV_RUSTFS_TLS_PATH, DEFAULT_RUSTFS_TLS_PATH); + if !tls_path.is_empty() { + let tls_dir = std::path::Path::new(&tls_path); + builder = add_root_certificates_from_file(builder, &tls_dir.join(RUSTFS_CA_CERT), "site-replication CA cert")?; + + if rustfs_utils::get_env_bool(ENV_TRUST_LEAF_CERT_AS_CA, DEFAULT_TRUST_LEAF_CERT_AS_CA) { + builder = + add_root_certificates_from_file(builder, &tls_dir.join(RUSTFS_TLS_CERT), "site-replication leaf cert as CA")?; + } + } + + builder + .build() + .map_err(|e| S3Error::with_message(S3ErrorCode::InternalError, format!("build site replication peer client failed: {e}"))) +} + +fn site_replication_peer_client() -> S3Result<&'static reqwest::Client> { + let result = SITE_REPLICATION_PEER_CLIENT.get_or_init(|| build_site_replication_peer_client().map_err(|e| e.to_string())); + result.as_ref().map_err(|err| { + S3Error::with_message( + S3ErrorCode::InternalError, + format!("initialize site replication peer client failed: {err}"), + ) }) } +fn runtime_tls_enabled() -> bool { + if let Some(tls_enabled) = get_global_endpoints_opt().and_then(|endpoints| { + endpoints + .as_ref() + .iter() + .flat_map(|pool| pool.endpoints.as_ref().iter()) + .find(|endpoint| endpoint.is_local) + .map(|endpoint| endpoint.url.scheme().eq_ignore_ascii_case("https")) + }) { + return tls_enabled; + } + + !rustfs_utils::get_env_str(ENV_RUSTFS_TLS_PATH, DEFAULT_RUSTFS_TLS_PATH).is_empty() +} + fn query_pairs(uri: &Uri) -> HashMap { uri.query() .map(|query| { @@ -554,17 +651,30 @@ fn load_ldap_idp_settings() -> (LDAPSettings, LDAPConfigSettings) { } fn request_endpoint(uri: &Uri, headers: &HeaderMap) -> String { - let scheme = headers - .get("x-forwarded-proto") - .and_then(|value| value.to_str().ok()) - .filter(|value| !value.is_empty()) - .unwrap_or("http"); + let scheme = get_source_scheme(headers) + .and_then(|value| { + value + .split(',') + .next() + .map(str::trim) + .filter(|value| !value.is_empty()) + .map(str::to_ascii_lowercase) + }) + .or_else(|| uri.scheme_str().map(str::to_ascii_lowercase)) + .unwrap_or_else(|| { + if runtime_tls_enabled() { + "https".to_string() + } else { + "http".to_string() + } + }); let host = headers .get(http::header::HOST) .and_then(|value| value.to_str().ok()) .filter(|value| !value.is_empty()) .map(str::to_string) + .or_else(|| uri.authority().map(|value| value.as_str().to_string())) .or_else(|| { get_global_endpoints_opt().and_then(|endpoints| { endpoints @@ -577,10 +687,6 @@ fn request_endpoint(uri: &Uri, headers: &HeaderMap) -> String { }) .unwrap_or_else(|| format!("127.0.0.1:{}", global_rustfs_port())); - if uri.scheme_str().is_some() { - return format!("{scheme}://{host}"); - } - format!("{scheme}://{host}") } @@ -601,12 +707,6 @@ fn infer_site_name(endpoint: &str) -> String { .to_string() } -fn deployment_id_for_endpoint(endpoint: &str) -> String { - let mut hasher = DefaultHasher::new(); - endpoint.hash(&mut hasher); - format!("{:016x}", hasher.finish()) -} - fn qstat(count: i64, bytes: i64) -> QStat { QStat { count: count as f64, @@ -666,37 +766,15 @@ fn current_local_runtime_peer(state: &SiteReplicationState) -> PeerInfo { } } -fn canonical_endpoint(endpoint: &str) -> String { - let trimmed = endpoint.trim().trim_end_matches('/'); - let candidate = if trimmed.starts_with("http://") || trimmed.starts_with("https://") { - trimmed.to_string() - } else { - format!("http://{trimmed}") - }; - - Url::parse(&candidate) - .ok() - .map(|url| { - let scheme = url.scheme().to_ascii_lowercase(); - let host = url.host_str().unwrap_or_default().to_ascii_lowercase(); - let port = url.port_or_known_default(); - match port { - Some(port) => format!("{scheme}://{host}:{port}"), - None => format!("{scheme}://{host}"), - } - }) - .unwrap_or_else(|| trimmed.to_ascii_lowercase()) -} - -fn same_endpoint(left: &str, right: &str) -> bool { - canonical_endpoint(left) == canonical_endpoint(right) +fn normalize_peer_map_by_identity(peers: BTreeMap) -> BTreeMap { + normalize_peer_map_by_identity_with(peers, normalize_peer_info) } fn existing_peer_for_endpoint(state: &SiteReplicationState, endpoint: &str) -> Option { state .peers .values() - .find(|peer| same_endpoint(&peer.endpoint, endpoint)) + .find(|peer| same_identity_endpoint(&peer.endpoint, endpoint)) .cloned() } @@ -744,11 +822,11 @@ fn build_join_peers( let mut normalized_local = local_peer.clone(); normalized_local.replicate_ilm_expiry = replicate_ilm_expiry; normalized_local = normalize_peer_info(normalized_local); - seen_endpoints.insert(canonical_endpoint(&normalized_local.endpoint)); + seen_endpoints.insert(site_identity_key(&normalized_local.endpoint)); peers.insert(normalized_local.deployment_id.clone(), normalized_local); for site in sites { - let endpoint_key = canonical_endpoint(&site.endpoint); + let endpoint_key = site_identity_key(&site.endpoint); if !seen_endpoints.insert(endpoint_key) { continue; } @@ -764,7 +842,7 @@ fn build_join_peers( peers.insert(peer.deployment_id.clone(), peer); } - peers + normalize_peer_map_by_identity(peers) } fn normalize_join_peers_for_local(local_peer: &PeerInfo, peers: BTreeMap) -> BTreeMap { @@ -772,7 +850,7 @@ fn normalize_join_peers_for_local(local_peer: &PeerInfo, peers: BTreeMap SiteReplicationState { let actual_peer = normalize_peer_info(actual_peer); state .peers - .retain(|_, peer| !same_endpoint(&peer.endpoint, &actual_peer.endpoint)); + .retain(|_, peer| !same_identity_endpoint(&peer.endpoint, &actual_peer.endpoint)); state.peers.insert(actual_peer.deployment_id.clone(), actual_peer); + state.peers = normalize_peer_map_by_identity(state.peers); state } @@ -888,16 +967,26 @@ async fn send_peer_admin_request( .unwrap_or("us-east-1"), ); - let mut req = site_replication_peer_client().request(reqwest::Method::PUT, &url); + let mut req = site_replication_peer_client()?.request(reqwest::Method::PUT, &url); for (name, value) in signed.headers() { req = req.header(name, value); } - let response = req - .body(payload) - .send() - .await - .map_err(|e| S3Error::with_message(S3ErrorCode::InternalError, format!("peer request failed: {e}")))?; + let response = req.body(payload).send().await.map_err(|e| { + let classify = if e.is_timeout() { + "timeout" + } else if e.is_connect() && e.to_string().to_ascii_lowercase().contains("dns") { + "dns resolution" + } else if e.to_string().to_ascii_lowercase().contains("certificate") || e.to_string().to_ascii_lowercase().contains("tls") + { + "tls handshake" + } else if e.is_connect() { + "connect" + } else { + "request" + }; + S3Error::with_message(S3ErrorCode::InternalError, format!("peer request to {url} failed ({classify}): {e}")) + })?; let status = response.status(); let body = response @@ -931,7 +1020,7 @@ async fn broadcast_site_replication_json(path: &str, body: &T) -> }; for peer in state.peers.values() { - if peer.deployment_id == local_peer.deployment_id || same_endpoint(&peer.endpoint, &local_peer.endpoint) { + if peer.deployment_id == local_peer.deployment_id || same_identity_endpoint(&peer.endpoint, &local_peer.endpoint) { continue; } @@ -1473,7 +1562,7 @@ fn sync_state_name_for_local_peer( local_peer: &PeerInfo, incoming: &PeerInfo, ) -> SiteReplicationState { - if same_endpoint(&incoming.endpoint, &local_peer.endpoint) && !incoming.name.is_empty() { + if same_identity_endpoint(&incoming.endpoint, &local_peer.endpoint) && !incoming.name.is_empty() { state.name = incoming.name.clone(); } state @@ -1634,7 +1723,7 @@ fn reconcile_site_replication_bucket_targets( let mut targets = existing.targets; for peer in state.peers.values() { - if peer.deployment_id == local_peer.deployment_id || same_endpoint(&peer.endpoint, &local_peer.endpoint) { + if peer.deployment_id == local_peer.deployment_id || same_identity_endpoint(&peer.endpoint, &local_peer.endpoint) { continue; } @@ -1705,7 +1794,7 @@ fn build_site_replication_config( ) -> Option { let mut rules = Vec::new(); for peer in state.peers.values() { - if peer.deployment_id == local_peer.deployment_id || same_endpoint(&peer.endpoint, &local_peer.endpoint) { + if peer.deployment_id == local_peer.deployment_id || same_identity_endpoint(&peer.endpoint, &local_peer.endpoint) { continue; } @@ -2283,8 +2372,9 @@ impl Operation for SiteReplicationAddHandler { let mut joined_endpoints = HashSet::new(); for site in &sites { - let endpoint_key = canonical_endpoint(&site.endpoint); - if same_endpoint(&site.endpoint, &local_peer.endpoint) || !joined_endpoints.insert(endpoint_key) { + if same_identity_endpoint(&site.endpoint, &local_peer.endpoint) + || !joined_endpoints.insert(site_identity_key(&site.endpoint)) + { continue; } @@ -2330,7 +2420,7 @@ impl Operation for SiteReplicationRemoveHandler { if !current_state.service_account_access_key.is_empty() && !current_state.service_account_secret_key.is_empty() { for peer in current_state.peers.values() { - if same_endpoint(&peer.endpoint, &local_peer.endpoint) { + if same_identity_endpoint(&peer.endpoint, &local_peer.endpoint) { continue; } send_peer_admin_request( @@ -2745,7 +2835,7 @@ impl Operation for SRPeerEditHandler { let state = load_site_replication_state().await?; let local_peer = current_local_peer(&req, &state); let mut incoming: PeerInfo = read_site_replication_json(req, "", false).await?; - if same_endpoint(&incoming.endpoint, &local_peer.endpoint) { + if same_identity_endpoint(&incoming.endpoint, &local_peer.endpoint) { incoming.deployment_id = local_peer.deployment_id.clone(); if incoming.name.is_empty() { incoming.name = local_peer.name.clone(); @@ -2863,7 +2953,8 @@ impl Operation for SRStateEditHandler { #[cfg(test)] mod tests { use super::*; - use http::Uri; + use http::{HeaderMap, HeaderValue, Uri}; + use temp_env::with_var; fn peer(name: &str, endpoint: &str) -> PeerInfo { PeerInfo { @@ -2991,6 +3082,64 @@ mod tests { assert!(normalized.contains_key("hash-remote")); } + #[test] + fn test_site_identity_key_deduplicates_scheme_drift_on_same_host_port() { + assert_eq!( + site_identity_key("https://node-a.example.com:9000"), + site_identity_key("http://NODE-A.example.com:9000/"), + ); + } + + #[test] + fn test_normalize_peer_map_by_identity_prefers_https_endpoint() { + let peers = BTreeMap::from([ + ( + "peer-http".to_string(), + PeerInfo { + deployment_id: "peer-http".to_string(), + ..peer("peer", "http://node-a.example.com:9000") + }, + ), + ( + "peer-https".to_string(), + PeerInfo { + deployment_id: "peer-https".to_string(), + ..peer("peer", "https://node-a.example.com:9000") + }, + ), + ]); + + let normalized = normalize_peer_map_by_identity(peers); + assert_eq!(normalized.len(), 1); + let normalized_peer = normalized.values().next().expect("normalized peer"); + assert!(normalized_peer.endpoint.starts_with("https://")); + } + + #[test] + fn test_request_endpoint_prefers_forwarded_proto() { + let uri: Uri = "/rustfs/admin/v3/site-replication/status".parse().unwrap(); + let mut headers = HeaderMap::new(); + headers.insert("x-forwarded-scheme", HeaderValue::from_static("http")); + headers.insert("x-forwarded-proto", HeaderValue::from_static("https")); + headers.insert("host", HeaderValue::from_static("node-a.example.com:9000")); + + let endpoint = request_endpoint(&uri, &headers); + + assert_eq!(endpoint, "https://node-a.example.com:9000"); + } + + #[test] + fn test_request_endpoint_falls_back_to_https_when_tls_path_is_configured() { + with_var(ENV_RUSTFS_TLS_PATH, Some("/tmp/tls"), || { + let uri: Uri = "/rustfs/admin/v3/site-replication/status".parse().unwrap(); + let headers = HeaderMap::new(); + + let endpoint = request_endpoint(&uri, &headers); + + assert!(endpoint.starts_with("https://")); + }); + } + #[test] fn test_reconcile_peer_with_actual_identity_replaces_endpoint_hash_key() { let mut state = SiteReplicationState::default(); diff --git a/rustfs/src/admin/mod.rs b/rustfs/src/admin/mod.rs index 69f382cfa..62efb8c8a 100644 --- a/rustfs/src/admin/mod.rs +++ b/rustfs/src/admin/mod.rs @@ -17,6 +17,7 @@ pub mod console; pub mod handlers; pub mod router; pub mod service; +pub mod site_replication_identity; pub mod utils; #[cfg(test)] diff --git a/rustfs/src/admin/service/site_replication.rs b/rustfs/src/admin/service/site_replication.rs index b8e3c78b8..71acbcaa7 100644 --- a/rustfs/src/admin/service/site_replication.rs +++ b/rustfs/src/admin/service/site_replication.rs @@ -12,13 +12,80 @@ // See the License for the specific language governing permissions and // limitations under the License. -use rustfs_ecstore::config::com::read_config; +use crate::admin::site_replication_identity::{deployment_id_for_endpoint, normalize_peer_map_by_identity_with}; +use rustfs_ecstore::config::com::{read_config, save_config}; use rustfs_ecstore::error::Error as StorageError; use rustfs_ecstore::new_object_layer_fn; +use rustfs_madmin::PeerInfo; use s3s::{S3Error, S3ErrorCode, S3Result}; +use serde_json::{Map, Value}; +use tracing::info; const SITE_REPLICATION_STATE_PATH: &str = "config/site-replication/state.json"; +fn normalize_peers_map(peers: &Map) -> Map { + let mut valid_peers = std::collections::BTreeMap::::new(); + let mut passthrough_invalid = Vec::<(String, Value)>::new(); + + for (key, value) in peers { + match serde_json::from_value::(value.clone()) { + Ok(mut peer) => { + if peer.endpoint.is_empty() { + passthrough_invalid.push((key.clone(), value.clone())); + continue; + } + if peer.deployment_id.is_empty() { + peer.deployment_id = deployment_id_for_endpoint(&peer.endpoint); + } + // Keep all parsed entries for identity-level normalization. Using the + // original JSON key avoids dropping records early on temporary + // deployment_id collisions. + valid_peers.insert(key.clone(), peer); + } + Err(_) => passthrough_invalid.push((key.clone(), value.clone())), + } + } + + let deduped_by_deployment = normalize_peer_map_by_identity_with(valid_peers, |peer| peer); + + let mut normalized = Map::new(); + for (_, peer) in deduped_by_deployment { + let key = peer.deployment_id.clone(); + if let Ok(value) = serde_json::to_value(peer) { + normalized.insert(key, value); + } + } + for (key, value) in passthrough_invalid { + normalized.entry(key).or_insert(value); + } + normalized +} + +fn normalize_site_replication_state_json(data: &[u8]) -> Result>, String> { + let mut state: Value = serde_json::from_slice(data).map_err(|e| format!("invalid site replication state: {e}"))?; + let Some(obj) = state.as_object_mut() else { + return Ok(None); + }; + + let before = obj.get("peers").and_then(|v| v.as_object()).map(|v| v.len()).unwrap_or(0); + + let Some(peers_obj) = obj.get("peers").and_then(|v| v.as_object()) else { + return Ok(None); + }; + + let normalized_peers = normalize_peers_map(peers_obj); + if normalized_peers == *peers_obj { + return Ok(None); + } + + let after = normalized_peers.len(); + obj.insert("peers".to_string(), Value::Object(normalized_peers)); + let normalized = + serde_json::to_vec(&state).map_err(|e| format!("serialize normalized site replication state failed: {e}"))?; + info!("normalized site-replication peers during reload: {before} -> {after}"); + Ok(Some(normalized)) +} + /// Reload persisted site-replication state. /// /// RustFS does not currently keep a separate in-memory cache for this state, @@ -28,10 +95,17 @@ pub async fn reload_site_replication_runtime_state() -> S3Result<()> { return Err(S3Error::with_message(S3ErrorCode::InternalError, "Not init".to_string())); }; - match read_config(store, SITE_REPLICATION_STATE_PATH).await { + match read_config(store.clone(), SITE_REPLICATION_STATE_PATH).await { Ok(data) => { - let _: serde_json::Value = serde_json::from_slice(&data) - .map_err(|e| S3Error::with_message(S3ErrorCode::InternalError, format!("invalid site replication state: {e}")))?; + if let Some(normalized) = + normalize_site_replication_state_json(&data).map_err(|e| S3Error::with_message(S3ErrorCode::InternalError, e))? + { + save_config(store, SITE_REPLICATION_STATE_PATH, normalized) + .await + .map_err(|e| { + S3Error::with_message(S3ErrorCode::InternalError, format!("normalize site replication state failed: {e}")) + })?; + } Ok(()) } Err(StorageError::ConfigNotFound) => Ok(()), @@ -41,3 +115,104 @@ pub async fn reload_site_replication_runtime_state() -> S3Result<()> { )), } } + +#[cfg(test)] +mod tests { + use super::*; + + fn peer_value(name: &str, endpoint: &str, deployment_id: &str) -> Value { + serde_json::json!({ + "name": name, + "endpoint": endpoint, + "deployment_id": deployment_id, + "sync_state": "", + "default_bandwidth": {}, + "replicate_ilm_expiry": false, + "object_naming_mode": "", + "api_version": "1" + }) + } + + #[test] + fn test_normalize_state_json_deduplicates_http_https_peer() { + let data = serde_json::to_vec(&serde_json::json!({ + "name": "local", + "peers": { + "remote-http": peer_value("remote", "http://node-a.example.com:9000", "remote-http"), + "remote-https": peer_value("remote", "https://node-a.example.com:9000/", "remote-https") + } + })) + .unwrap(); + + let normalized = normalize_site_replication_state_json(&data) + .unwrap() + .expect("state should be normalized"); + let value: Value = serde_json::from_slice(&normalized).unwrap(); + let peers = value.get("peers").and_then(Value::as_object).unwrap(); + + assert_eq!(peers.len(), 1); + let endpoint = peers + .values() + .next() + .and_then(|peer| peer.get("endpoint")) + .and_then(Value::as_str) + .unwrap(); + assert!(endpoint.starts_with("https://")); + } + + #[test] + fn test_normalize_state_json_is_idempotent() { + let data = serde_json::to_vec(&serde_json::json!({ + "name": "local", + "peers": { + "remote": peer_value("remote", "https://node-a.example.com:9000", "remote") + } + })) + .unwrap(); + + let first = normalize_site_replication_state_json(&data).unwrap(); + let normalized_once = first.unwrap_or(data); + let second = normalize_site_replication_state_json(&normalized_once).unwrap(); + assert!(second.is_none()); + } + + #[test] + fn test_normalize_state_json_tolerates_malformed_peer_entries() { + let data = serde_json::to_vec(&serde_json::json!({ + "name": "local", + "peers": { + "broken": {"endpoint": 123}, + "remote": peer_value("remote", "https://node-a.example.com:9000", "remote") + } + })) + .unwrap(); + + let normalized = normalize_site_replication_state_json(&data).unwrap(); + let out = normalized.unwrap_or(data); + let value: Value = serde_json::from_slice(&out).unwrap(); + let peers = value.get("peers").and_then(Value::as_object).unwrap(); + + assert!(peers.contains_key("broken")); + assert!(!peers.is_empty()); + } + + #[test] + fn test_normalize_state_json_preserves_entries_before_identity_dedupe() { + let data = serde_json::to_vec(&serde_json::json!({ + "name": "local", + "peers": { + "peer-a": peer_value("remote-a", "https://node-a.example.com:9000", "dup"), + "peer-b": peer_value("remote-b", "https://node-b.example.com:9000", "dup") + } + })) + .unwrap(); + + let normalized = normalize_site_replication_state_json(&data) + .unwrap() + .expect("state should be normalized"); + let value: Value = serde_json::from_slice(&normalized).unwrap(); + let peers = value.get("peers").and_then(Value::as_object).unwrap(); + + assert_eq!(peers.len(), 2); + } +} diff --git a/rustfs/src/admin/site_replication_identity.rs b/rustfs/src/admin/site_replication_identity.rs new file mode 100644 index 000000000..6992f88bd --- /dev/null +++ b/rustfs/src/admin/site_replication_identity.rs @@ -0,0 +1,140 @@ +// 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. + +use rustfs_madmin::PeerInfo; +use std::collections::{BTreeMap, hash_map::DefaultHasher}; +use std::hash::{Hash, Hasher}; +use url::Url; + +pub fn canonical_endpoint(endpoint: &str) -> String { + let trimmed = endpoint.trim().trim_end_matches('/'); + let candidate = if trimmed.starts_with("http://") || trimmed.starts_with("https://") { + trimmed.to_string() + } else { + format!("http://{trimmed}") + }; + + Url::parse(&candidate) + .ok() + .map(|url| { + let scheme = url.scheme().to_ascii_lowercase(); + let host = url.host_str().unwrap_or_default().to_ascii_lowercase(); + let port = url.port_or_known_default(); + match port { + Some(port) => format!("{scheme}://{host}:{port}"), + None => format!("{scheme}://{host}"), + } + }) + .unwrap_or_else(|| trimmed.to_ascii_lowercase()) +} + +pub fn site_identity_key(endpoint: &str) -> String { + let trimmed = endpoint.trim().trim_end_matches('/'); + let candidate = if trimmed.starts_with("http://") || trimmed.starts_with("https://") { + trimmed.to_string() + } else { + format!("http://{trimmed}") + }; + + Url::parse(&candidate) + .ok() + .map(|url| { + let host = url.host_str().unwrap_or_default().to_ascii_lowercase(); + match url.port_or_known_default() { + Some(port) => format!("{host}:{port}"), + None => host, + } + }) + .unwrap_or_else(|| trimmed.to_ascii_lowercase()) +} + +pub fn deployment_id_for_endpoint(endpoint: &str) -> String { + let mut hasher = DefaultHasher::new(); + endpoint.hash(&mut hasher); + format!("{:016x}", hasher.finish()) +} + +pub fn same_identity_endpoint(left: &str, right: &str) -> bool { + site_identity_key(left) == site_identity_key(right) +} + +fn is_https_endpoint(endpoint: &str) -> bool { + canonical_endpoint(endpoint).starts_with("https://") +} + +fn merge_identity_peer(existing: PeerInfo, incoming: PeerInfo) -> PeerInfo { + let existing_https = is_https_endpoint(&existing.endpoint); + let incoming_https = is_https_endpoint(&incoming.endpoint); + let mut merged = if incoming_https && !existing_https { + incoming.clone() + } else { + existing.clone() + }; + let fallback = if merged.deployment_id == incoming.deployment_id { + existing + } else { + incoming + }; + + if merged.deployment_id.is_empty() { + merged.deployment_id = fallback.deployment_id; + } + if merged.name.is_empty() { + merged.name = fallback.name; + } + if merged.api_version.is_none() { + merged.api_version = fallback.api_version; + } + merged.replicate_ilm_expiry |= fallback.replicate_ilm_expiry; + merged +} + +pub fn normalize_peer_map_by_identity_with(peers: BTreeMap, mut normalize: F) -> BTreeMap +where + F: FnMut(PeerInfo) -> PeerInfo, +{ + let mut peers_by_identity = BTreeMap::::new(); + for (_, peer) in peers { + let normalized_peer = normalize(peer); + let identity = site_identity_key(&normalized_peer.endpoint); + if let Some(existing) = peers_by_identity.remove(&identity) { + peers_by_identity.insert(identity, normalize(merge_identity_peer(existing, normalized_peer))); + } else { + peers_by_identity.insert(identity, normalized_peer); + } + } + + let mut normalized = BTreeMap::::new(); + for (_, mut peer) in peers_by_identity { + if peer.deployment_id.is_empty() { + peer.deployment_id = deployment_id_for_endpoint(&peer.endpoint); + } + + let mut deployment_id = peer.deployment_id.clone(); + if let Some(existing) = normalized.get(&deployment_id) + && site_identity_key(&existing.endpoint) != site_identity_key(&peer.endpoint) + { + deployment_id = format!("{deployment_id}-{}", deployment_id_for_endpoint(&peer.endpoint)); + peer.deployment_id = deployment_id.clone(); + } + + if let Some(existing) = normalized.get(&deployment_id).cloned() { + normalized.insert(deployment_id, normalize(merge_identity_peer(existing, peer))); + } else { + normalized.insert(deployment_id, peer); + } + } + + normalized +} diff --git a/scripts/validate_issue_2723_site_replication.sh b/scripts/validate_issue_2723_site_replication.sh new file mode 100755 index 000000000..c58798277 --- /dev/null +++ b/scripts/validate_issue_2723_site_replication.sh @@ -0,0 +1,298 @@ +#!/usr/bin/env bash +set -euo pipefail + +# Issue 2723 verification runner +# Validates site-replication behavior against Task 07 matrix/evidence. + +SITE_A_ENDPOINT="${SITE_A_ENDPOINT:-}" +SITE_B_ENDPOINT="${SITE_B_ENDPOINT:-}" +ACCESS_KEY="${ACCESS_KEY:-rustfsadmin}" +SECRET_KEY="${SECRET_KEY:-rustfsadmin}" +REGION="${REGION:-us-east-1}" +CA_CERT="${CA_CERT:-}" +OUT_DIR="${OUT_DIR:-target/verify/issue-2723-$(date +%Y%m%d-%H%M%S)}" +SITE_A_RESTART_CMD="${SITE_A_RESTART_CMD:-}" +SITE_B_RESTART_CMD="${SITE_B_RESTART_CMD:-}" +BUCKET="${BUCKET:-}" +REPL_OBJECT_KEY="${REPL_OBJECT_KEY:-issue-2723-e2e-object.txt}" +REPL_OBJECT_BODY="${REPL_OBJECT_BODY:-issue-2723-replication-check}" +AWS_PROFILE="${AWS_PROFILE:-}" + +AWSCURL_BIN="${AWSCURL_BIN:-awscurl}" +AWS_BIN="${AWS_BIN:-aws}" +HEALTHCHECK_FORCE_LOOPBACK_RESOLVE="${HEALTHCHECK_FORCE_LOOPBACK_RESOLVE:-false}" + +usage() { + cat <<'USAGE' +Usage: + scripts/validate_issue_2723_site_replication.sh [options] + +Required: + --site-a-endpoint Site A admin endpoint, e.g. https://site-a.example.com:9000 + --site-b-endpoint Site B admin endpoint, e.g. https://site-b.example.com:9000 + --access-key + --secret-key + +Optional: + --region AWS region for signing (default: us-east-1) + --ca-cert CA cert for strict HTTPS health checks + --out-dir Artifact output directory + --site-a-restart-cmd Restart command for site A (optional) + --site-b-restart-cmd Restart command for site B (optional) + --bucket Replication validation bucket (optional) + --repl-object-key Replication validation object key + --repl-object-body Replication validation object body + --awscurl-bin awscurl binary (default: awscurl) + --aws-bin aws cli binary (default: aws) + --aws-profile AWS CLI profile for object-flow checks + --healthcheck-force-loopback-resolve + Force HTTPS healthcheck `--resolve host:port:127.0.0.1` + (default: false; intended for single-host local Docker) + -h, --help Show help + +Notes: + 1) If --bucket is provided and aws cli is available, script will run optional + object-flow checks on both sites. + 2) Restart verification is skipped unless both restart commands are provided. +USAGE +} + +log() { + printf '[%s] %s\n' "$(date +%H:%M:%S)" "$*" +} + +fail() { + log "ERROR: $*" >&2 + exit 1 +} + +require_cmd() { + if ! command -v "$1" >/dev/null 2>&1; then + fail "required command not found: $1" + fi +} + +parse_args() { + while [[ $# -gt 0 ]]; do + case "$1" in + --site-a-endpoint) SITE_A_ENDPOINT="$2"; shift 2 ;; + --site-b-endpoint) SITE_B_ENDPOINT="$2"; shift 2 ;; + --access-key) ACCESS_KEY="$2"; shift 2 ;; + --secret-key) SECRET_KEY="$2"; shift 2 ;; + --region) REGION="$2"; shift 2 ;; + --ca-cert) CA_CERT="$2"; shift 2 ;; + --out-dir) OUT_DIR="$2"; shift 2 ;; + --site-a-restart-cmd) SITE_A_RESTART_CMD="$2"; shift 2 ;; + --site-b-restart-cmd) SITE_B_RESTART_CMD="$2"; shift 2 ;; + --bucket) BUCKET="$2"; shift 2 ;; + --repl-object-key) REPL_OBJECT_KEY="$2"; shift 2 ;; + --repl-object-body) REPL_OBJECT_BODY="$2"; shift 2 ;; + --awscurl-bin) AWSCURL_BIN="$2"; shift 2 ;; + --aws-bin) AWS_BIN="$2"; shift 2 ;; + --aws-profile) AWS_PROFILE="$2"; shift 2 ;; + --healthcheck-force-loopback-resolve) HEALTHCHECK_FORCE_LOOPBACK_RESOLVE="true"; shift ;; + -h|--help) usage; exit 0 ;; + *) + fail "unknown argument: $1" + ;; + esac + done +} + +endpoint_scheme() { + local endpoint="$1" + if [[ "$endpoint" == https://* ]]; then + echo "https" + elif [[ "$endpoint" == http://* ]]; then + echo "http" + else + fail "endpoint must include scheme http:// or https:// : $endpoint" + fi +} + +endpoint_hostport() { + local endpoint="$1" + local hostport + hostport="${endpoint#http://}" + hostport="${hostport#https://}" + hostport="${hostport%%/*}" + echo "$hostport" +} + +admin_get() { + local endpoint="$1" + local path="$2" + local out_file="$3" + local url="${endpoint%/}${path}" + if [[ -n "$CA_CERT" ]]; then + REQUESTS_CA_BUNDLE="$CA_CERT" SSL_CERT_FILE="$CA_CERT" \ + "$AWSCURL_BIN" --service s3 --region "$REGION" --access_key "$ACCESS_KEY" --secret_key "$SECRET_KEY" "$url" >"$out_file" + else + "$AWSCURL_BIN" --service s3 --region "$REGION" --access_key "$ACCESS_KEY" --secret_key "$SECRET_KEY" "$url" >"$out_file" + fi +} + +strict_healthcheck() { + local endpoint="$1" + local label="$2" + local out_file="$3" + local scheme hostport host port + scheme="$(endpoint_scheme "$endpoint")" + hostport="$(endpoint_hostport "$endpoint")" + host="${hostport%:*}" + port="${hostport##*:}" + if [[ "$port" == "$hostport" ]]; then + port=$([[ "$scheme" == "https" ]] && echo "443" || echo "80") + fi + local url="${scheme}://${host}:${port}/health" + + if [[ "$scheme" == "https" ]]; then + if [[ -z "$CA_CERT" ]]; then + fail "HTTPS endpoint requires --ca-cert for strict validation: $endpoint" + fi + if [[ "$HEALTHCHECK_FORCE_LOOPBACK_RESOLVE" == "true" ]]; then + curl -fsS --cacert "$CA_CERT" --resolve "${host}:${port}:127.0.0.1" "$url" >"$out_file" + else + curl -fsS --cacert "$CA_CERT" "$url" >"$out_file" + fi + else + curl -fsS "$url" >"$out_file" + fi + log "health check passed for ${label}: $url" +} + +analyze_duplicates() { + local status_json="$1" + local out_file="$2" + jq -r ' + def identity_key($e): + ($e | sub("^https?://";"") | sub("/$";"") | ascii_downcase); + (.sites // {}) + | to_entries + | map(.value.endpoint // "") + | map(select(. != "")) + | map(identity_key(.)) + | group_by(.) + | map({identity: .[0], count: length}) + | map(select(.count > 1)) + ' "$status_json" >"$out_file" +} + +optional_object_flow_check() { + local endpoint="$1" + local label="$2" + local put_out="$3" + local get_out="$4" + + if [[ -z "$BUCKET" ]]; then + log "skip object-flow check for ${label}: --bucket not provided" + return 0 + fi + if ! command -v "$AWS_BIN" >/dev/null 2>&1; then + log "skip object-flow check for ${label}: aws cli not found" + return 0 + fi + + local common=( + --endpoint-url "$endpoint" + --region "$REGION" + --no-cli-pager + ) + if [[ -n "$AWS_PROFILE" ]]; then + common+=(--profile "$AWS_PROFILE") + fi + if [[ -n "$CA_CERT" ]]; then + common+=(--ca-bundle "$CA_CERT") + fi + + AWS_ACCESS_KEY_ID="$ACCESS_KEY" AWS_SECRET_ACCESS_KEY="$SECRET_KEY" \ + "$AWS_BIN" "${common[@]}" s3api put-object \ + --bucket "$BUCKET" --key "$REPL_OBJECT_KEY" --body <(printf '%s' "$REPL_OBJECT_BODY") >"$put_out" + + AWS_ACCESS_KEY_ID="$ACCESS_KEY" AWS_SECRET_ACCESS_KEY="$SECRET_KEY" \ + "$AWS_BIN" "${common[@]}" s3api head-object \ + --bucket "$BUCKET" --key "$REPL_OBJECT_KEY" >"$get_out" + + log "object-flow check passed for ${label} on bucket=${BUCKET}, key=${REPL_OBJECT_KEY}" +} + +restart_if_configured() { + if [[ -z "$SITE_A_RESTART_CMD" || -z "$SITE_B_RESTART_CMD" ]]; then + log "skip restart verification: restart commands not fully provided" + return 0 + fi + log "running restart command for site A" + bash -lc "$SITE_A_RESTART_CMD" + log "running restart command for site B" + bash -lc "$SITE_B_RESTART_CMD" +} + +main() { + parse_args "$@" + [[ -n "$SITE_A_ENDPOINT" ]] || fail "--site-a-endpoint is required" + [[ -n "$SITE_B_ENDPOINT" ]] || fail "--site-b-endpoint is required" + [[ -n "$ACCESS_KEY" ]] || fail "--access-key is required" + [[ -n "$SECRET_KEY" ]] || fail "--secret-key is required" + + require_cmd "$AWSCURL_BIN" + require_cmd curl + require_cmd jq + + mkdir -p "$OUT_DIR" + log "output directory: $OUT_DIR" + + local a_status="$OUT_DIR/site-a.status.json" + local a_info="$OUT_DIR/site-a.info.json" + local b_status="$OUT_DIR/site-b.status.json" + local b_info="$OUT_DIR/site-b.info.json" + local a_health="$OUT_DIR/site-a.health.txt" + local b_health="$OUT_DIR/site-b.health.txt" + local a_dupes="$OUT_DIR/site-a.duplicates.json" + local b_dupes="$OUT_DIR/site-b.duplicates.json" + local summary="$OUT_DIR/summary.txt" + + log "step 1/6: strict health checks" + strict_healthcheck "$SITE_A_ENDPOINT" "site-a" "$a_health" + strict_healthcheck "$SITE_B_ENDPOINT" "site-b" "$b_health" + + log "step 2/6: collect site-replication status/info" + admin_get "$SITE_A_ENDPOINT" "/rustfs/admin/v3/site-replication/status" "$a_status" + admin_get "$SITE_A_ENDPOINT" "/rustfs/admin/v3/site-replication/info" "$a_info" + admin_get "$SITE_B_ENDPOINT" "/rustfs/admin/v3/site-replication/status" "$b_status" + admin_get "$SITE_B_ENDPOINT" "/rustfs/admin/v3/site-replication/info" "$b_info" + + log "step 3/6: duplicate identity analysis" + analyze_duplicates "$a_status" "$a_dupes" + analyze_duplicates "$b_status" "$b_dupes" + + log "step 4/6: optional object-flow checks" + optional_object_flow_check "$SITE_A_ENDPOINT" "site-a" "$OUT_DIR/site-a.put.json" "$OUT_DIR/site-a.head.json" + optional_object_flow_check "$SITE_B_ENDPOINT" "site-b" "$OUT_DIR/site-b.put.json" "$OUT_DIR/site-b.head.json" + + log "step 5/6: optional restart verification" + restart_if_configured + if [[ -n "$SITE_A_RESTART_CMD" && -n "$SITE_B_RESTART_CMD" ]]; then + admin_get "$SITE_A_ENDPOINT" "/rustfs/admin/v3/site-replication/status" "$OUT_DIR/site-a.status.after-restart.json" + admin_get "$SITE_B_ENDPOINT" "/rustfs/admin/v3/site-replication/status" "$OUT_DIR/site-b.status.after-restart.json" + fi + + log "step 6/6: write summary" + { + echo "Issue 2723 verification summary" + echo "site-a endpoint: $SITE_A_ENDPOINT" + echo "site-b endpoint: $SITE_B_ENDPOINT" + echo "region: $REGION" + echo "ca-cert: ${CA_CERT:-}" + echo + echo "Duplicate identities (site-a): $(jq 'length' "$a_dupes")" + echo "Duplicate identities (site-b): $(jq 'length' "$b_dupes")" + echo + echo "Artifacts:" + find "$OUT_DIR" -maxdepth 1 -type f | sort + } >"$summary" + + log "done. summary: $summary" + cat "$summary" +} + +main "$@"