mirror of
https://github.com/rustfs/rustfs.git
synced 2026-07-26 08:18:18 +00:00
fix(replication): persist sanitized resync errors (#5200)
This commit is contained in:
cxymds
@@ -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;
|
||||
|
||||
@@ -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<E, R>(error: &SdkError<E, R>) -> Option<String>
|
||||
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<tokio::sync::mpsc::Sender<ReplicateObjectInfo>>,
|
||||
results_tx: tokio::sync::mpsc::Sender<TargetReplicationResyncStatus>,
|
||||
@@ -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::<HeadObjectError, ()>::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 {
|
||||
|
||||
@@ -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::{
|
||||
|
||||
@@ -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<T> = std::result::Result<T, Error>;
|
||||
|
||||
@@ -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<String> {
|
||||
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<Self> {
|
||||
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]);
|
||||
|
||||
Reference in New Issue
Block a user