fix(admin): harden site-replication identity handling and local verification for issue 2723 (#2730)

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<reqwest::Client> = OnceLock::new();
+static SITE_REPLICATION_PEER_CLIENT: OnceLock<Result<reqwest::Client, String>> = OnceLock::new();
 
 #[derive(Debug, Clone, Serialize, Deserialize, Default)]
 struct SiteReplicationState {
@@ -378,9 +385,22 @@ async fn load_site_replication_state() -> S3Result<SiteReplicationState> {
         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<reqwest::ClientBuilder> {
+    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<reqwest::Client> {
+    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<String, String> {
     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<String, PeerInfo>) -> BTreeMap<String, PeerInfo> {
+    normalize_peer_map_by_identity_with(peers, normalize_peer_info)
 }
 
 fn existing_peer_for_endpoint(state: &SiteReplicationState, endpoint: &str) -> Option<PeerInfo> {
     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<String, PeerInfo>) -> BTreeMap<String, PeerInfo> {
@@ -772,7 +850,7 @@ fn normalize_join_peers_for_local(local_peer: &PeerInfo, peers: BTreeMap<String,
 
     for (_, incoming_peer) in peers {
         let mut peer = normalize_peer_info(incoming_peer);
-        if same_endpoint(&peer.endpoint, &local_peer.endpoint) {
+        if same_identity_endpoint(&peer.endpoint, &local_peer.endpoint) {
             peer.deployment_id = local_peer.deployment_id.clone();
             if peer.name.is_empty() {
                 peer.name = local_peer.name.clone();
@@ -785,15 +863,16 @@ fn normalize_join_peers_for_local(local_peer: &PeerInfo, peers: BTreeMap<String,
         normalized.insert(local_peer.deployment_id.clone(), local_peer.clone());
     }
 
-    normalized
+    normalize_peer_map_by_identity(normalized)
 }
 
 fn reconcile_peer_with_actual_identity(mut state: SiteReplicationState, actual_peer: PeerInfo) -> 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<T: Serialize>(
             .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<T: Serialize>(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<ReplicationConfiguration> {
     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();

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)]

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<String, Value>) -> Map<String, Value> {
+    let mut valid_peers = std::collections::BTreeMap::<String, PeerInfo>::new();
+    let mut passthrough_invalid = Vec::<(String, Value)>::new();
+
+    for (key, value) in peers {
+        match serde_json::from_value::<PeerInfo>(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<Option<Vec<u8>>, 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);
+    }
+}

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<F>(peers: BTreeMap<String, PeerInfo>, mut normalize: F) -> BTreeMap<String, PeerInfo>
+where
+    F: FnMut(PeerInfo) -> PeerInfo,
+{
+    let mut peers_by_identity = BTreeMap::<String, PeerInfo>::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::<String, PeerInfo>::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
+}