From 35e1f28f235e70fb24594fb123e1240d422fe2dc Mon Sep 17 00:00:00 2001 From: weisd Date: Thu, 19 Mar 2026 12:43:53 +0800 Subject: [PATCH] fix: support legacy bucket metadata decoding (#2227) --- crates/ecstore/src/bucket/metadata.rs | 198 ++++++++++++++++++-- crates/ecstore/src/bucket/metadata_test.rs | 137 ++++++++++++++ crates/filemeta/src/filemeta.rs | 50 ++++- crates/filemeta/src/filemeta/inline_data.rs | 6 +- 4 files changed, 364 insertions(+), 27 deletions(-) diff --git a/crates/ecstore/src/bucket/metadata.rs b/crates/ecstore/src/bucket/metadata.rs index 6680f6c15..b07c871eb 100644 --- a/crates/ecstore/src/bucket/metadata.rs +++ b/crates/ecstore/src/bucket/metadata.rs @@ -33,7 +33,7 @@ use serde::Serializer; use std::collections::HashMap; use std::io::{Read, Write}; use std::sync::Arc; -use time::OffsetDateTime; +use time::{Date, OffsetDateTime, PrimitiveDateTime, Time as CivilTime, UtcOffset}; use tracing::error; fn read_msgp_str(rd: &mut R) -> Result { @@ -43,19 +43,177 @@ fn read_msgp_str(rd: &mut R) -> Result { Ok(String::from_utf8(buf)?) } +fn read_msgp_bool(rd: &mut R) -> Result { + let marker = rmp::decode::read_marker(rd).map_err(|e| Error::other(format!("{e:?}")))?; + match marker { + rmp::Marker::True => Ok(true), + rmp::Marker::False => Ok(false), + rmp::Marker::FixPos(v) => Ok(v != 0), + rmp::Marker::U8 => Ok(read_u8(rd)? != 0), + rmp::Marker::U16 => Ok(read_u16_raw(rd)? != 0), + rmp::Marker::U32 => Ok(read_u32_raw(rd)? != 0), + rmp::Marker::U64 => Ok(read_u64_raw(rd)? != 0), + rmp::Marker::I8 => Ok(read_i8_raw(rd)? != 0), + rmp::Marker::I16 => Ok(read_i16_raw(rd)? != 0), + rmp::Marker::I32 => Ok(read_i32_raw(rd)? != 0), + rmp::Marker::I64 => Ok(read_i64_raw(rd)? != 0), + rmp::Marker::FixNeg(v) => Ok(v != 0), + _ => Err(Error::other(format!("expected bool or int-like bool, got marker: {marker:?}"))), + } +} + fn read_msgp_time_value(rd: &mut R) -> Result { let marker = rmp::decode::read_marker(rd).map_err(|e| Error::other(format!("{e:?}")))?; match marker { rmp::Marker::Null => Ok(OffsetDateTime::UNIX_EPOCH), rmp::Marker::Ext8 => read_msgp_ext8_time(rd), + rmp::Marker::FixArray(len) => read_msgp_legacy_compact_time(rd, u32::from(len)), + rmp::Marker::Array16 => { + let len = read_u16_raw(rd)?; + read_msgp_legacy_compact_time(rd, u32::from(len)) + } + rmp::Marker::Array32 => { + let len = read_u32_raw(rd)?; + read_msgp_legacy_compact_time(rd, len) + } + rmp::Marker::Bin8 => { + let len = usize::from(read_u8(rd)?); + read_msgp_time_value_from_embedded_bin(rd, len) + } + rmp::Marker::Bin16 => { + let len = usize::from(read_u16_raw(rd)?); + read_msgp_time_value_from_embedded_bin(rd, len) + } + rmp::Marker::Bin32 => { + let len = read_u32_raw(rd)? as usize; + read_msgp_time_value_from_embedded_bin(rd, len) + } _ => Err(Error::other(format!("expected time ext or nil, got marker: {marker:?}"))), } } -fn read_msgp_bin(rd: &mut R) -> Result> { - let len = rmp::decode::read_bin_len(rd)? as usize; +fn read_u8(rd: &mut R) -> Result { + let mut buf = [0u8; 1]; + rd.read_exact(&mut buf)?; + Ok(buf[0]) +} + +fn read_u16_raw(rd: &mut R) -> Result { + let mut buf = [0u8; 2]; + rd.read_exact(&mut buf)?; + Ok(BigEndian::read_u16(&buf)) +} + +fn read_u32_raw(rd: &mut R) -> Result { + let mut buf = [0u8; 4]; + rd.read_exact(&mut buf)?; + Ok(BigEndian::read_u32(&buf)) +} + +fn read_u64_raw(rd: &mut R) -> Result { + let mut buf = [0u8; 8]; + rd.read_exact(&mut buf)?; + Ok(BigEndian::read_u64(&buf)) +} + +fn read_i8_raw(rd: &mut R) -> Result { + Ok(read_u8(rd)? as i8) +} + +fn read_i16_raw(rd: &mut R) -> Result { + let mut buf = [0u8; 2]; + rd.read_exact(&mut buf)?; + Ok(BigEndian::read_i16(&buf)) +} + +fn read_i32_raw(rd: &mut R) -> Result { + let mut buf = [0u8; 4]; + rd.read_exact(&mut buf)?; + Ok(BigEndian::read_i32(&buf)) +} + +fn read_i64_raw(rd: &mut R) -> Result { + let mut buf = [0u8; 8]; + rd.read_exact(&mut buf)?; + Ok(BigEndian::read_i64(&buf)) +} + +fn read_msgp_time_value_from_embedded_bin(rd: &mut R, len: usize) -> Result { let mut buf = vec![0u8; len]; rd.read_exact(&mut buf)?; + let mut cur = std::io::Cursor::new(buf); + read_msgp_time_value(&mut cur) +} + +fn read_msgp_legacy_compact_time(rd: &mut R, len: u32) -> Result { + if len != 9 { + return Err(Error::other(format!("invalid legacy compact time len: {len}"))); + } + + let year: i32 = rmp::decode::read_int(rd).map_err(|e| Error::other(format!("{e:?}")))?; + let ordinal: u16 = rmp::decode::read_int(rd).map_err(|e| Error::other(format!("{e:?}")))?; + let hour: u8 = rmp::decode::read_int(rd).map_err(|e| Error::other(format!("{e:?}")))?; + let minute: u8 = rmp::decode::read_int(rd).map_err(|e| Error::other(format!("{e:?}")))?; + let second: u8 = rmp::decode::read_int(rd).map_err(|e| Error::other(format!("{e:?}")))?; + let nanosecond: u32 = rmp::decode::read_int(rd).map_err(|e| Error::other(format!("{e:?}")))?; + let offset_hour: i8 = rmp::decode::read_int(rd).map_err(|e| Error::other(format!("{e:?}")))?; + let offset_minute: i8 = rmp::decode::read_int(rd).map_err(|e| Error::other(format!("{e:?}")))?; + let offset_second: i8 = rmp::decode::read_int(rd).map_err(|e| Error::other(format!("{e:?}")))?; + + let date = + Date::from_ordinal_date(year, ordinal).map_err(|e| Error::other(format!("invalid legacy compact time date: {e}")))?; + let time = CivilTime::from_hms_nano(hour, minute, second, nanosecond) + .map_err(|e| Error::other(format!("invalid legacy compact time time: {e}")))?; + let offset = UtcOffset::from_hms(offset_hour, offset_minute, offset_second) + .map_err(|e| Error::other(format!("invalid legacy compact time offset: {e}")))?; + + Ok(PrimitiveDateTime::new(date, time) + .assume_offset(offset) + .to_offset(UtcOffset::UTC)) +} + +fn read_msgp_bin(rd: &mut R) -> Result> { + let marker = rmp::decode::read_marker(rd).map_err(|e| Error::other(format!("{e:?}")))?; + match marker { + rmp::Marker::Null => Ok(Vec::new()), + rmp::Marker::Bin8 => { + let len = usize::from(read_u8(rd)?); + read_exact_bytes(rd, len) + } + rmp::Marker::Bin16 => { + let len = usize::from(read_u16_raw(rd)?); + read_exact_bytes(rd, len) + } + rmp::Marker::Bin32 => { + let len = read_u32_raw(rd)? as usize; + read_exact_bytes(rd, len) + } + rmp::Marker::FixArray(len) => read_msgp_legacy_byte_array(rd, u32::from(len)), + rmp::Marker::Array16 => { + let len = read_u16_raw(rd)?; + read_msgp_legacy_byte_array(rd, u32::from(len)) + } + rmp::Marker::Array32 => { + let len = read_u32_raw(rd)?; + read_msgp_legacy_byte_array(rd, len) + } + _ => Err(Error::other(format!("expected bin or byte array, got marker: {marker:?}"))), + } +} + +fn read_exact_bytes(rd: &mut R, len: usize) -> Result> { + let mut buf = vec![0u8; len]; + rd.read_exact(&mut buf)?; + Ok(buf) +} + +fn read_msgp_legacy_byte_array(rd: &mut R, len: u32) -> Result> { + let mut buf = Vec::with_capacity(len as usize); + for _ in 0..len { + let value: i64 = rmp::decode::read_int(rd).map_err(|e| Error::other(format!("{e:?}")))?; + let byte = u8::try_from(value).map_err(|_| Error::other(format!("byte value out of range: {value}")))?; + buf.push(byte); + } Ok(buf) } @@ -227,18 +385,20 @@ impl BucketMetadata { match key.as_str() { "Name" => self.name = read_msgp_str(rd)?, "Created" => self.created = read_msgp_time_value(rd)?, - "LockEnabled" => self.lock_enabled = rmp::decode::read_bool(rd)?, - "PolicyConfigJSON" => self.policy_config_json = read_msgp_bin(rd)?, - "NotificationConfigXML" => self.notification_config_xml = read_msgp_bin(rd)?, - "LifecycleConfigXML" => self.lifecycle_config_xml = read_msgp_bin(rd)?, - "ObjectLockConfigXML" => self.object_lock_config_xml = read_msgp_bin(rd)?, - "VersioningConfigXML" => self.versioning_config_xml = read_msgp_bin(rd)?, - "EncryptionConfigXML" => self.encryption_config_xml = read_msgp_bin(rd)?, - "TaggingConfigXML" => self.tagging_config_xml = read_msgp_bin(rd)?, - "QuotaConfigJSON" => self.quota_config_json = read_msgp_bin(rd)?, - "ReplicationConfigXML" => self.replication_config_xml = read_msgp_bin(rd)?, - "BucketTargetsConfigJSON" => self.bucket_targets_config_json = read_msgp_bin(rd)?, - "BucketTargetsConfigMetaJSON" => self.bucket_targets_config_meta_json = read_msgp_bin(rd)?, + "LockEnabled" => self.lock_enabled = read_msgp_bool(rd)?, + "PolicyConfigJSON" | "PolicyConfigJson" => self.policy_config_json = read_msgp_bin(rd)?, + "NotificationConfigXML" | "NotificationConfigXml" => self.notification_config_xml = read_msgp_bin(rd)?, + "LifecycleConfigXML" | "LifecycleConfigXml" => self.lifecycle_config_xml = read_msgp_bin(rd)?, + "ObjectLockConfigXML" | "ObjectLockConfigXml" => self.object_lock_config_xml = read_msgp_bin(rd)?, + "VersioningConfigXML" | "VersioningConfigXml" => self.versioning_config_xml = read_msgp_bin(rd)?, + "EncryptionConfigXML" | "EncryptionConfigXml" => self.encryption_config_xml = read_msgp_bin(rd)?, + "TaggingConfigXML" | "TaggingConfigXml" => self.tagging_config_xml = read_msgp_bin(rd)?, + "QuotaConfigJSON" | "QuotaConfigJson" => self.quota_config_json = read_msgp_bin(rd)?, + "ReplicationConfigXML" | "ReplicationConfigXml" => self.replication_config_xml = read_msgp_bin(rd)?, + "BucketTargetsConfigJSON" | "BucketTargetsConfigJson" => self.bucket_targets_config_json = read_msgp_bin(rd)?, + "BucketTargetsConfigMetaJSON" | "BucketTargetsConfigMetaJson" => { + self.bucket_targets_config_meta_json = read_msgp_bin(rd)? + } "PolicyConfigUpdatedAt" => self.policy_config_updated_at = read_msgp_time_value(rd)?, "ObjectLockConfigUpdatedAt" => self.object_lock_config_updated_at = read_msgp_time_value(rd)?, "EncryptionConfigUpdatedAt" => self.encryption_config_updated_at = read_msgp_time_value(rd)?, @@ -250,9 +410,11 @@ impl BucketMetadata { "NotificationConfigUpdatedAt" => self.notification_config_updated_at = read_msgp_time_value(rd)?, "BucketTargetsConfigUpdatedAt" => self.bucket_targets_config_updated_at = read_msgp_time_value(rd)?, "BucketTargetsConfigMetaUpdatedAt" => self.bucket_targets_config_meta_updated_at = read_msgp_time_value(rd)?, - "CorsConfigXML" => self.cors_config_xml = read_msgp_bin(rd)?, - "PublicAccessBlockConfigXML" => self.public_access_block_config_xml = read_msgp_bin(rd)?, - "BucketAclConfigJSON" => self.bucket_acl_config_json = read_msgp_bin(rd)?, + "CorsConfigXML" | "CorsConfigXml" => self.cors_config_xml = read_msgp_bin(rd)?, + "PublicAccessBlockConfigXML" | "PublicAccessBlockConfigXml" => { + self.public_access_block_config_xml = read_msgp_bin(rd)? + } + "BucketAclConfigJSON" | "BucketAclConfigJson" => self.bucket_acl_config_json = read_msgp_bin(rd)?, "CorsConfigUpdatedAt" => self.cors_config_updated_at = read_msgp_time_value(rd)?, "PublicAccessBlockConfigUpdatedAt" => self.public_access_block_config_updated_at = read_msgp_time_value(rd)?, "BucketAclConfigUpdatedAt" => self.bucket_acl_config_updated_at = read_msgp_time_value(rd)?, diff --git a/crates/ecstore/src/bucket/metadata_test.rs b/crates/ecstore/src/bucket/metadata_test.rs index 335594275..eafd6d041 100644 --- a/crates/ecstore/src/bucket/metadata_test.rs +++ b/crates/ecstore/src/bucket/metadata_test.rs @@ -98,6 +98,143 @@ async fn unmarshal_test_bucket_metadata() { assert!(bm.bucket_acl_config_json.is_empty()); } +#[test] +fn unmarshal_legacy_compact_time_bucket_metadata() { + use faster_hex::hex_decode; + + let legacy_hex = concat!( + "83", + "a44e616d65", + "a474657374", + "a743726561746564", + "99cd07e9cd01100c1021ce2026b1fa000000", + "ab4c6f636b456e61626c6564", + "c2" + ); + + let mut bytes = vec![0u8; legacy_hex.len() / 2]; + hex_decode(legacy_hex.as_bytes(), &mut bytes).expect("valid hex"); + + let bm = BucketMetadata::unmarshal(&bytes).expect("legacy compact time should decode"); + + assert_eq!(bm.name, "test"); + assert_eq!(bm.created.unix_timestamp(), 1759148193); + assert_eq!(bm.created.nanosecond(), 539406842); + assert!(!bm.lock_enabled); +} + +#[test] +fn unmarshal_bin_wrapped_ext_time_bucket_metadata() { + use faster_hex::hex_decode; + + let wrapped_hex = concat!( + "83", + "a44e616d65", + "a464616461", + "a743726561746564", + "c40fc70c05fffffff1886e090000000000", + "ab4c6f636b456e61626c6564", + "c2" + ); + + let mut bytes = vec![0u8; wrapped_hex.len() / 2]; + hex_decode(wrapped_hex.as_bytes(), &mut bytes).expect("valid hex"); + + let bm = BucketMetadata::unmarshal(&bytes).expect("bin-wrapped ext time should decode"); + + assert_eq!(bm.name, "dada"); + assert_eq!(bm.created.unix_timestamp(), -62135596800); + assert!(!bm.lock_enabled); +} + +#[test] +fn unmarshal_legacy_rmp_serde_field_aliases_and_byte_arrays() { + use faster_hex::hex_decode; + + let legacy_hex = concat!( + "85", + "a44e616d65", + "a474657374", + "a743726561746564", + "99cd07e9cd01100c1021ce2026b1fa000000", + "ab4c6f636b456e61626c6564", + "c2", + "b0506f6c696379436f6e6669674a736f6e", + "93010203", + "bb4275636b657454617267657473436f6e6669674d6574614a736f6e", + "920405" + ); + + let mut bytes = vec![0u8; legacy_hex.len() / 2]; + hex_decode(legacy_hex.as_bytes(), &mut bytes).expect("valid hex"); + + let bm = BucketMetadata::unmarshal(&bytes).expect("legacy field aliases and byte arrays should decode"); + + assert_eq!(bm.name, "test"); + assert_eq!(bm.created.unix_timestamp(), 1759148193); + assert_eq!(bm.created.nanosecond(), 539406842); + assert_eq!(bm.policy_config_json, vec![1, 2, 3]); + assert_eq!(bm.bucket_targets_config_meta_json, vec![4, 5]); + assert!(!bm.lock_enabled); +} + +#[test] +fn unmarshal_legacy_bin16_and_array16_bucket_metadata() { + let policy = vec![b'x'; 257]; + let targets_meta: Vec = (0u8..=16).collect(); + + let mut bytes = Vec::new(); + rmp::encode::write_map_len(&mut bytes, 5).unwrap(); + rmp::encode::write_str(&mut bytes, "Name").unwrap(); + rmp::encode::write_str(&mut bytes, "test-bucket").unwrap(); + rmp::encode::write_str(&mut bytes, "Created").unwrap(); + bytes.extend_from_slice(&[ + 0xc7, 0x0c, 0x05, 0x00, 0x00, 0x00, 0x00, 0x65, 0x92, 0x00, 0x80, 0x00, 0x00, 0x00, 0x00, + ]); + rmp::encode::write_str(&mut bytes, "LockEnabled").unwrap(); + bytes.push(0x01); + rmp::encode::write_str(&mut bytes, "PolicyConfigJson").unwrap(); + rmp::encode::write_bin(&mut bytes, &policy).unwrap(); + rmp::encode::write_str(&mut bytes, "BucketTargetsConfigMetaJson").unwrap(); + rmp::encode::write_array_len(&mut bytes, targets_meta.len() as u32).unwrap(); + for byte in &targets_meta { + rmp::encode::write_uint(&mut bytes, u64::from(*byte)).unwrap(); + } + + let bm = BucketMetadata::unmarshal(&bytes).expect("legacy bin16 and array16 should decode"); + + assert_eq!(bm.name, "test-bucket"); + assert_eq!(bm.created.unix_timestamp(), 1704067200); + assert!(bm.lock_enabled); + assert_eq!(bm.policy_config_json, policy); + assert_eq!(bm.bucket_targets_config_meta_json, targets_meta); +} + +#[test] +fn unmarshal_legacy_numeric_bool_bucket_metadata() { + use faster_hex::hex_decode; + + let legacy_hex = concat!( + "83", + "a44e616d65", + "ab746573742d6275636b6574", + "a743726561746564", + "c70c05000000006592008000000000", + "ab4c6f636b456e61626c6564", + "01" + ); + + let mut bytes = vec![0u8; legacy_hex.len() / 2]; + hex_decode(legacy_hex.as_bytes(), &mut bytes).expect("valid hex"); + + let bm = BucketMetadata::unmarshal(&bytes).expect("legacy numeric bool should decode"); + + assert_eq!(bm.name, "test-bucket"); + assert_eq!(bm.created.unix_timestamp(), 1704067200); + assert_eq!(bm.created.nanosecond(), 0); + assert!(bm.lock_enabled); +} + #[tokio::test] async fn marshal_msg_complete_example() { // Create a complete BucketMetadata with various configurations diff --git a/crates/filemeta/src/filemeta.rs b/crates/filemeta/src/filemeta.rs index 0ef91d991..8213aa694 100644 --- a/crates/filemeta/src/filemeta.rs +++ b/crates/filemeta/src/filemeta.rs @@ -74,6 +74,14 @@ pub(crate) fn data_key_for_version(version_id: Option) -> String { } } +fn legacy_data_key_for_version(version_id: Option) -> Option { + if version_id.is_none() || version_id == Some(Uuid::nil()) { + Some(Uuid::nil().to_string()) + } else { + None + } +} + pub const TRANSITION_COMPLETE: &str = "complete"; pub const TRANSITION_PENDING: &str = "pending"; @@ -164,6 +172,21 @@ impl FileMeta { }); } + fn find_inline_data_for_version(&self, version_id: Option) -> Result>> { + let key = data_key_for_version(version_id); + if let Some(data) = self.data.find(key.as_str())? { + return Ok(Some(data)); + } + + if let Some(legacy_key) = legacy_data_key_for_version(version_id) + && legacy_key != key + { + return self.data.find(legacy_key.as_str()); + } + + Ok(None) + } + // Find version pub fn find_version(&self, vid: Option) -> Result<(usize, FileMetaVersion)> { let vid = vid.unwrap_or_default(); @@ -709,10 +732,7 @@ impl FileMeta { } if read_data && fi.inline_data() { - fi.data = self - .data - .find(data_key_for_version(fi.version_id).as_str())? - .map(bytes::Bytes::from); + fi.data = self.find_inline_data_for_version(fi.version_id)?.map(bytes::Bytes::from); } found_fi = Some(fi); @@ -1071,6 +1091,28 @@ mod test { assert!(!inline_data.is_empty(), "Inline data should not be empty"); } + #[test] + fn test_into_fileinfo_reads_legacy_nil_uuid_inline_key() { + let mut fm = FileMeta::new(); + let mut fi = FileInfo::new("test", 2, 1); + fi.mod_time = Some(OffsetDateTime::now_utc()); + fi.data = Some(Bytes::from_static(b"legacy-inline")); + fi.set_inline_data(); + fm.add_version(fi).unwrap(); + + let current_key = data_key_for_version(Some(Uuid::nil())); + let legacy_key = legacy_data_key_for_version(Some(Uuid::nil())).unwrap(); + let payload = fm.data.find(current_key.as_str()).unwrap().unwrap(); + + fm.data.remove_key(current_key.as_str()).unwrap(); + fm.data.replace(legacy_key.as_str(), payload.clone()).unwrap(); + + let restored = fm.into_fileinfo("bucket", "test", "", true, false, true).unwrap(); + + assert!(restored.inline_data()); + assert_eq!(restored.data, Some(Bytes::from(payload))); + } + #[test] fn test_error_handling_and_recovery() { // Test error handling and recovery diff --git a/crates/filemeta/src/filemeta/inline_data.rs b/crates/filemeta/src/filemeta/inline_data.rs index e8dca577f..9f57e1fb0 100644 --- a/crates/filemeta/src/filemeta/inline_data.rs +++ b/crates/filemeta/src/filemeta/inline_data.rs @@ -43,11 +43,7 @@ impl FileMeta { let vid = version_id.unwrap_or_default(); if self.data.entries().unwrap_or_default() > 0 - && self - .data - .find(super::data_key_for_version(version_id).as_str()) - .unwrap_or_default() - .is_some() + && self.find_inline_data_for_version(version_id).unwrap_or_default().is_some() { return 0; }