diff --git a/crates/ecstore/src/bucket/replication/replication_resync_boundary.rs b/crates/ecstore/src/bucket/replication/replication_resync_boundary.rs index 7df357ad3..3c4da2413 100644 --- a/crates/ecstore/src/bucket/replication/replication_resync_boundary.rs +++ b/crates/ecstore/src/bucket/replication/replication_resync_boundary.rs @@ -17,7 +17,8 @@ use super::replication_filemeta_boundary::MrfReplicateEntry; pub use rustfs_replication::{BucketReplicationResyncStatus, ResyncOpts, ResyncStatusType, TargetReplicationResyncStatus}; pub(crate) use rustfs_replication::{ - is_version_id_mismatch, resync_state_accepts_update, should_auto_resume_resync, should_count_head_proxy_failure, + is_version_id_mismatch, resync_state_accepts_update, sanitize_resync_error_detail, should_auto_resume_resync, + should_count_head_proxy_failure, }; pub(crate) const RESYNC_META_FORMAT: u16 = rustfs_replication::resync::RESYNC_META_FORMAT; diff --git a/crates/ecstore/src/bucket/replication/replication_resyncer.rs b/crates/ecstore/src/bucket/replication/replication_resyncer.rs index 247623de0..f9ecfced0 100644 --- a/crates/ecstore/src/bucket/replication/replication_resyncer.rs +++ b/crates/ecstore/src/bucket/replication/replication_resyncer.rs @@ -37,7 +37,7 @@ use super::replication_queue_boundary::DeletedObjectReplicationInfo; use super::replication_resync_boundary::ResyncStatusType; use super::replication_resync_boundary::{ BucketReplicationResyncStatus, ResyncOpts, TargetReplicationResyncStatus, encode_resync_file, is_version_id_mismatch, - resync_state_accepts_update, should_count_head_proxy_failure, + resync_state_accepts_update, sanitize_resync_error_detail, should_count_head_proxy_failure, }; #[cfg(test)] use super::replication_resync_boundary::{RESYNC_META_FORMAT, RESYNC_META_VERSION, WIRE_ZERO_TIME_UNIX, decode_resync_file}; @@ -117,6 +117,20 @@ const RESYNC_TIME_INTERVAL: TokioDuration = TokioDuration::from_secs(60); static WARNED_MONITOR_UNINIT: std::sync::Once = std::sync::Once::new(); +fn resync_target_error_detail(error: &SdkError) -> Option +where + E: ProvideErrorMetadata, +{ + sanitize_resync_error_detail(error.code().unwrap_or(match error { + SdkError::ConstructionFailure(_) => "failed to construct target request", + SdkError::TimeoutError(_) => "target request timed out", + SdkError::DispatchFailure(_) => "target dispatch failed", + SdkError::ResponseError(_) => "invalid target response", + SdkError::ServiceError(_) => "target service error", + _ => "target request failed", + })) +} + async fn finish_resync_workers( worker_txs: Vec>, results_tx: tokio::sync::mpsc::Sender, @@ -428,6 +442,9 @@ impl ReplicationResyncer { state.replicated_size += status.replicated_size; state.failed_count += status.failed_count; state.failed_size += status.failed_size; + if state.error.is_none() && status.failed_count > 0 { + state.error = status.error.as_deref().and_then(sanitize_resync_error_detail); + } state.last_update = Some(now); bucket_status.last_update = Some(now); } @@ -885,6 +902,7 @@ impl ReplicationResyncer { "Processed resync object" ); } + st.error = err.as_ref().and_then(resync_target_error_detail); if cancel_token.is_cancelled() { return; @@ -3227,6 +3245,7 @@ mod tests { assert_eq!(tgt.start_time, Some(start)); assert_eq!(tgt.last_update, Some(last)); assert_eq!(tgt.resync_before_date, Some(before)); + assert_eq!(tgt.error, None); } #[test] @@ -3681,6 +3700,59 @@ mod tests { assert!(resyncer.target_has_resync_failures(&opts).await); } + #[tokio::test] + async fn test_inc_stats_retains_first_sanitized_error_across_success() { + let resyncer = ReplicationResyncer::new().await; + let opts = ResyncOpts { + bucket: "bucket".to_string(), + arn: "arn:replication::dest".to_string(), + resync_id: "run-new".to_string(), + resync_before: None, + }; + let failed = TargetReplicationResyncStatus { + failed_count: 1, + object: "failed-object".to_string(), + error: Some("Authorization: Bearer status-secret".to_string()), + ..Default::default() + }; + let later_failure = TargetReplicationResyncStatus { + failed_count: 1, + object: "later-failed-object".to_string(), + error: Some("AccessDenied".to_string()), + ..Default::default() + }; + let succeeded = TargetReplicationResyncStatus { + replicated_count: 1, + object: "successful-object".to_string(), + ..Default::default() + }; + + resyncer.inc_stats(&failed, opts.clone()).await; + resyncer.inc_stats(&later_failure, opts.clone()).await; + resyncer.inc_stats(&succeeded, opts.clone()).await; + + let status_map = resyncer.status_map.read().await; + let target = &status_map["bucket"].targets_map["arn:replication::dest"]; + assert_eq!(target.failed_count, 2); + assert_eq!(target.replicated_count, 1); + assert_eq!(target.object, "successful-object"); + assert_eq!(target.error.as_deref(), Some("[redacted sensitive resync error detail]")); + } + + #[test] + fn test_resync_target_error_detail_uses_safe_service_code_and_fallback() { + let metadata = aws_smithy_types::error::ErrorMetadata::builder() + .code("AccessDenied") + .message("Authorization: Bearer status-secret") + .build(); + let service_error = SdkError::service_error(HeadObjectError::generic(metadata), ()); + let timeout_error = + SdkError::::timeout_error(std::io::Error::new(std::io::ErrorKind::TimedOut, "status-secret")); + + assert_eq!(resync_target_error_detail(&service_error).as_deref(), Some("AccessDenied")); + assert_eq!(resync_target_error_detail(&timeout_error).as_deref(), Some("target request timed out")); + } + #[test] fn test_resync_state_accepts_update_only_for_matching_run() { let current = TargetReplicationResyncStatus { diff --git a/crates/replication/src/lib.rs b/crates/replication/src/lib.rs index 89e646c39..93a4eb18d 100644 --- a/crates/replication/src/lib.rs +++ b/crates/replication/src/lib.rs @@ -67,8 +67,8 @@ pub use queue::{ }; pub use resync::{ BucketReplicationResyncStatus, Error, Result, ResyncOpts, ResyncStatusType, TargetReplicationResyncStatus, - decode_resync_file, encode_resync_file, is_version_id_mismatch, resync_state_accepts_update, should_auto_resume_resync, - should_count_head_proxy_failure, + decode_resync_file, encode_resync_file, is_version_id_mismatch, resync_state_accepts_update, sanitize_resync_error_detail, + should_auto_resume_resync, should_count_head_proxy_failure, }; pub use rule::ReplicationRuleExt; pub use runtime::{ diff --git a/crates/replication/src/resync.rs b/crates/replication/src/resync.rs index f000de47c..f65d57ac6 100644 --- a/crates/replication/src/resync.rs +++ b/crates/replication/src/resync.rs @@ -27,6 +27,18 @@ pub const RESYNC_META_VERSION: u16 = 1; const MSGP_TIME_EXT_TYPE: i8 = 5; const MSGP_TIME_LEN: u8 = 12; pub const WIRE_ZERO_TIME_UNIX: i64 = -62_135_596_800; +const RESYNC_ERROR_DETAIL_MAX_CHARS: usize = 512; +const RESYNC_ERROR_REDACTED: &str = "[redacted sensitive resync error detail]"; +const RESYNC_ERROR_TRUNCATED_SUFFIX: &str = "... (truncated)"; +const RESYNC_ERROR_SENSITIVE_MARKERS: &[&str] = &[ + "authorization", + "bearer", + "credential", + "secret", + "token", + "signature", + "password", +]; pub type Result = std::result::Result; @@ -178,7 +190,8 @@ impl TargetReplicationResyncStatus { rmp::encode::write_str(wr, "obj")?; rmp::encode::write_str(wr, &self.object)?; rmp::encode::write_str(wr, "err")?; - rmp::encode::write_str(wr, self.error.as_deref().unwrap_or_default())?; + let error = self.error.as_deref().and_then(sanitize_resync_error_detail); + rmp::encode::write_str(wr, error.as_deref().unwrap_or_default())?; Ok(()) } @@ -206,7 +219,7 @@ impl TargetReplicationResyncStatus { "obj" => out.object = read_msgp_str(rd)?, "err" => { let error = read_msgp_str(rd)?; - out.error = (!error.is_empty()).then_some(error); + out.error = sanitize_resync_error_detail(&error); } _ => skip_msgp_value(rd)?, } @@ -215,6 +228,54 @@ impl TargetReplicationResyncStatus { } } +pub fn sanitize_resync_error_detail(detail: &str) -> Option { + let mut normalized = String::new(); + let mut char_count = 0usize; + let mut truncated = false; + 'words: for word in detail.split_whitespace() { + if !normalized.is_empty() { + if char_count == RESYNC_ERROR_DETAIL_MAX_CHARS { + truncated = true; + break; + } + normalized.push(' '); + char_count += 1; + } + for ch in word.chars() { + if char_count == RESYNC_ERROR_DETAIL_MAX_CHARS { + truncated = true; + break 'words; + } + normalized.push(ch); + char_count += 1; + } + } + if normalized.is_empty() { + return None; + } + + let lower = normalized.to_ascii_lowercase(); + let is_service_code = normalized.len() <= 128 + && normalized + .bytes() + .all(|byte| byte.is_ascii_alphanumeric() || matches!(byte, b'-' | b'_')); + if !is_service_code + && (RESYNC_ERROR_SENSITIVE_MARKERS.iter().any(|marker| lower.contains(marker)) + || (lower.contains("://") && lower.contains('@'))) + { + return Some(RESYNC_ERROR_REDACTED.to_string()); + } + + if !truncated { + return Some(normalized); + } + + let keep = RESYNC_ERROR_DETAIL_MAX_CHARS.saturating_sub(RESYNC_ERROR_TRUNCATED_SUFFIX.chars().count()); + let mut summary: String = normalized.chars().take(keep).collect(); + summary.push_str(RESYNC_ERROR_TRUNCATED_SUFFIX); + Some(summary) +} + pub fn resync_state_accepts_update(state: &TargetReplicationResyncStatus, opts: &ResyncOpts) -> bool { state.resync_id.is_empty() || opts.resync_id.is_empty() || state.resync_id == opts.resync_id } @@ -313,7 +374,11 @@ impl BucketReplicationResyncStatus { } pub fn unmarshal_legacy_msg(data: &[u8]) -> Result { - Ok(rmp_serde::from_slice(data)?) + let mut status: Self = rmp_serde::from_slice(data)?; + for target in status.targets_map.values_mut() { + target.error = target.error.as_deref().and_then(sanitize_resync_error_detail); + } + Ok(status) } } @@ -644,6 +709,91 @@ mod tests { assert_eq!(got.targets_map["arn:replication:a"].error.as_deref(), Some("durable failure")); } + #[test] + fn resync_error_detail_is_bounded_and_unicode_safe() { + let detail = format!("{}尾", "x".repeat(RESYNC_ERROR_DETAIL_MAX_CHARS + 32)); + + let sanitized = sanitize_resync_error_detail(&detail).expect("non-empty detail should remain"); + + assert_eq!(sanitized.chars().count(), RESYNC_ERROR_DETAIL_MAX_CHARS); + assert!(sanitized.ends_with(RESYNC_ERROR_TRUNCATED_SUFFIX)); + assert!(!sanitized.contains('尾')); + } + + #[test] + fn resync_error_detail_redacts_credentials_and_headers() { + for detail in [ + "Authorization: Bearer status-secret", + "request failed with secret_key=status-secret", + "x-amz-security-token=status-secret", + "https://user:status-secret@example.test", + ] { + assert_eq!( + sanitize_resync_error_detail(detail).as_deref(), + Some(RESYNC_ERROR_REDACTED), + "sensitive detail should be replaced: {detail}" + ); + } + assert_eq!(sanitize_resync_error_detail("ExpiredToken").as_deref(), Some("ExpiredToken")); + } + + #[test] + fn resync_file_sanitizes_legacy_error_on_decode() { + let mut status = BucketReplicationResyncStatus::new(); + status.targets_map.insert( + "arn:replication:a".to_string(), + TargetReplicationResyncStatus { + resync_id: "rid-legacy".to_string(), + resync_status: ResyncStatusType::ResyncFailed, + error: Some("Authorization: Bearer persisted-secret".to_string()), + ..Default::default() + }, + ); + let legacy_payload = rmp_serde::to_vec(&status).expect("legacy status should encode"); + let direct = + BucketReplicationResyncStatus::unmarshal_legacy_msg(&legacy_payload).expect("legacy status should decode directly"); + let mut data = Vec::new(); + data.extend_from_slice(&RESYNC_META_FORMAT.to_le_bytes()); + data.extend_from_slice(&RESYNC_META_VERSION.to_le_bytes()); + data.extend_from_slice(&legacy_payload); + + let decoded = decode_resync_file(&data).expect("legacy status should decode"); + + assert_eq!(direct.targets_map["arn:replication:a"].error.as_deref(), Some(RESYNC_ERROR_REDACTED)); + assert_eq!(decoded.targets_map["arn:replication:a"].error.as_deref(), Some(RESYNC_ERROR_REDACTED)); + } + + #[test] + fn resync_file_retains_error_for_restartable_and_failed_states() { + let mut status = BucketReplicationResyncStatus::new(); + for (arn, resync_status) in [ + ("arn:replication:pending", ResyncStatusType::ResyncPending), + ("arn:replication:ongoing", ResyncStatusType::ResyncStarted), + ("arn:replication:failed", ResyncStatusType::ResyncFailed), + ] { + status.targets_map.insert( + arn.to_string(), + TargetReplicationResyncStatus { + resync_id: format!("{arn}-run"), + resync_status, + failed_count: 1, + error: Some("AccessDenied".to_string()), + ..Default::default() + }, + ); + } + + let data = encode_resync_file(&status).expect("resync status should encode"); + let decoded = decode_resync_file(&data).expect("resync status should decode after restart"); + + for arn in ["arn:replication:pending", "arn:replication:ongoing", "arn:replication:failed"] { + assert_eq!(decoded.targets_map[arn].error.as_deref(), Some("AccessDenied")); + } + assert!(should_auto_resume_resync(decoded.targets_map["arn:replication:pending"].resync_status)); + assert!(should_auto_resume_resync(decoded.targets_map["arn:replication:ongoing"].resync_status)); + assert!(!should_auto_resume_resync(decoded.targets_map["arn:replication:failed"].resync_status)); + } + #[test] fn skip_msgp_value_consumes_ext_type_and_payload() { let mut ext16 = Cursor::new(vec![0xc8, 0, 2, MSGP_TIME_EXT_TYPE as u8, 0xaa, 0xbb, 0x01]);