diff --git a/Cargo.lock b/Cargo.lock index b7192728a..e252ba92c 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -3313,9 +3313,9 @@ checksum = "ac6b926516df9c60bfa16e107b21086399f8285a44ca9711344b9e553c5146e2" [[package]] name = "defmt" -version = "1.1.0" +version = "1.1.1" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "a6e524506490a1953d237cb87b1cfc1e46f88c18f10a22dfe0f507dc6bfc7f7f" +checksum = "e2953bfe4f93bbd20cc71198842756f77d161884c99ebbabc41d80231ded88d1" dependencies = [ "bitflags 1.3.2", "defmt-macros", @@ -3323,12 +3323,11 @@ dependencies = [ [[package]] name = "defmt-macros" -version = "1.1.0" +version = "1.1.1" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "f0a27770e9c8f719a79d8b638281f4d828f77d8fd61e0bd94451b9b85e576a0b" +checksum = "bad9c72e7ca2137e0dc3813245a0d282fd6daad32fd800af018306a9169b5fe8" dependencies = [ "defmt-parser", - "proc-macro-error2", "proc-macro2", "quote", "syn 2.0.118", @@ -7650,28 +7649,6 @@ dependencies = [ "toml_edit", ] -[[package]] -name = "proc-macro-error-attr2" -version = "2.0.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "96de42df36bb9bba5542fe9f1a054b8cc87e172759a1868aa05c1f3acc89dfc5" -dependencies = [ - "proc-macro2", - "quote", -] - -[[package]] -name = "proc-macro-error2" -version = "2.0.1" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "11ec05c52be0a07b08061f7dd003e7d7092e0472bc731b4af7bb1ef876109802" -dependencies = [ - "proc-macro-error-attr2", - "proc-macro2", - "quote", - "syn 2.0.118", -] - [[package]] name = "proc-macro-utils" version = "0.10.0" @@ -8154,9 +8131,9 @@ dependencies = [ [[package]] name = "rapidhash" -version = "4.5.0" +version = "4.5.1" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "b1a224897b65b7ce38bf7b0adb569e7d77e11460d0cc3577908b263d8b5a89aa" +checksum = "5da7e78a036ce858e8d55b7e7dc8ba3a88b78350fd2155d3591bbd966b58589e" dependencies = [ "rustversion", ] @@ -10174,8 +10151,9 @@ checksum = "9774ba4a74de5f7b1c1451ed6cd5285a32eddb5cccb8cc655a4e50009e06477f" [[package]] name = "s3s" -version = "0.14.0" -source = "git+https://github.com/rustfs/s3s?rev=b365107038a633ce0f7a199f507507a8c5e64665#b365107038a633ce0f7a199f507507a8c5e64665" +version = "0.14.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "abe1bd31748cb69848c2cf4028cecdadf5f8299ef3b02a83e21b20e7bda3aba9" dependencies = [ "arc-swap", "arrayvec", diff --git a/Cargo.toml b/Cargo.toml index 743df4099..e232aa894 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -283,7 +283,7 @@ redis = { version = "1.3.0", features = ["connection-manager", "tokio-rustls-com rustix = { version = "1.1.4", features = ["fs"] } rust-embed = { version = "8.11.0" } rustc-hash = { version = "2.1.3" } -s3s = { git = "https://github.com/rustfs/s3s", rev = "b365107038a633ce0f7a199f507507a8c5e64665", features = ["minio"] } +s3s = { version = "0.14.1", features = ["minio"] } serial_test = "3.5.0" shadow-rs = { version = "2.0.0", default-features = false } siphasher = "1.0.3" diff --git a/crates/protocols/src/swift/versioning.rs b/crates/protocols/src/swift/versioning.rs index 68b30355e..6c57e9d79 100644 --- a/crates/protocols/src/swift/versioning.rs +++ b/crates/protocols/src/swift/versioning.rs @@ -58,6 +58,7 @@ use super::resolve_swift_object_store_handle; use super::storage_api::versioning::{ListOperations as _, ObjectOperations as _}; use super::{SwiftError, SwiftResult}; use rustfs_credentials::Credentials; +use std::sync::atomic::{AtomicU64, Ordering}; use std::time::{SystemTime, UNIX_EPOCH}; use tracing::{debug, error}; @@ -66,6 +67,11 @@ const LOG_SUBSYSTEM_SWIFT_VERSIONING: &str = "swift_versioning"; const EVENT_SWIFT_VERSIONING_ARCHIVE_STATE: &str = "swift_versioning_archive_state"; const EVENT_SWIFT_VERSIONING_RESTORE_STATE: &str = "swift_versioning_restore_state"; const EVENT_SWIFT_VERSIONING_LIST_STATE: &str = "swift_versioning_list_state"; +const NANOS_PER_SECOND: u64 = 1_000_000_000; +const VERSION_TIMESTAMP_MAX_SECONDS: u64 = 9_999_999_999; +const VERSION_TIMESTAMP_MAX_NANOS: u64 = VERSION_TIMESTAMP_MAX_SECONDS * NANOS_PER_SECOND + (NANOS_PER_SECOND - 1); + +static LAST_VERSION_UNIX_NANOS: AtomicU64 = AtomicU64::new(0); /// Generate a version name for an archived object /// @@ -91,22 +97,37 @@ const EVENT_SWIFT_VERSIONING_LIST_STATE: &str = "swift_versioning_list_state"; /// # Returns /// Versioned object name with inverted timestamp prefix pub fn generate_version_name(container: &str, object: &str) -> String { - // Get current timestamp + let unix_nanos = next_version_unix_nanos(); + let inverted_nanos = VERSION_TIMESTAMP_MAX_NANOS.saturating_sub(unix_nanos); + let inverted_seconds = inverted_nanos / NANOS_PER_SECOND; + let inverted_subsec_nanos = inverted_nanos % NANOS_PER_SECOND; + + // Format: {inverted_timestamp}/{container}/{object} + // 9 decimal places = nanosecond precision (prevents collisions up to 1B ops/sec) + format!("{inverted_seconds:010}.{inverted_subsec_nanos:09}/{container}/{object}") +} + +fn current_unix_nanos() -> u64 { let now = SystemTime::now() .duration_since(UNIX_EPOCH) .unwrap_or_else(|_| std::time::Duration::from_secs(0)); - let timestamp = now.as_secs_f64(); + now.as_secs() + .saturating_mul(NANOS_PER_SECOND) + .saturating_add(u64::from(now.subsec_nanos())) +} - // Invert timestamp so newer versions sort first - // Max reasonable timestamp: 9999999999 (year 2286) - // Using 9 decimal places (nanosecond precision) to prevent collisions - // in high-throughput scenarios where objects are uploaded rapidly - let inverted = 9999999999.999999999 - timestamp; +fn next_version_unix_nanos() -> u64 { + let now = current_unix_nanos(); + let mut observed = LAST_VERSION_UNIX_NANOS.load(Ordering::Acquire); - // Format: {inverted_timestamp}/{container}/{object} - // 9 decimal places = nanosecond precision (prevents collisions up to 1B ops/sec) - format!("{:.9}/{}/{}", inverted, container, object) + loop { + let candidate = now.max(observed.saturating_add(1)).min(VERSION_TIMESTAMP_MAX_NANOS); + match LAST_VERSION_UNIX_NANOS.compare_exchange_weak(observed, candidate, Ordering::AcqRel, Ordering::Acquire) { + Ok(_) => return candidate, + Err(actual) => observed = actual, + } + } } /// Archive the current version of an object before overwriting @@ -677,8 +698,6 @@ mod tests { timestamps.insert(ts); } - // Should have at least some unique timestamps - // (May not be 100 due to system clock granularity) - assert!(timestamps.len() > 1, "Timestamps should be mostly unique"); + assert_eq!(timestamps.len(), 100, "Version timestamps should be unique"); } } diff --git a/crates/protocols/tests/swift_versioning_integration.rs b/crates/protocols/tests/swift_versioning_integration.rs index b904968a1..3c215ef01 100644 --- a/crates/protocols/tests/swift_versioning_integration.rs +++ b/crates/protocols/tests/swift_versioning_integration.rs @@ -109,18 +109,9 @@ fn test_version_timestamp_precision() { std::thread::sleep(std::time::Duration::from_micros(10)); } - // Check uniqueness - allow some collisions on low-precision systems + // Check uniqueness even when the system clock has coarse precision. let unique_count = versions.iter().collect::>().len(); - let collision_rate = (versions.len() - unique_count) as f64 / versions.len() as f64; - - // Allow up to 10% collision rate on low-precision systems - assert!( - collision_rate < 0.1, - "High collision rate: {} collisions out of {} ({}%)", - versions.len() - unique_count, - versions.len(), - collision_rate * 100.0 - ); + assert_eq!(unique_count, versions.len(), "Version names should be unique"); } /// Test inverted timestamp calculation @@ -171,24 +162,10 @@ fn test_version_uniqueness_stress() { handle.join().unwrap(); } - // Check uniqueness - allow some collisions on low-precision systems + // Check uniqueness even when multiple threads generate versions in the same clock tick. let versions_vec = versions.lock().unwrap(); let unique_count = versions_vec.iter().collect::>().len(); - let collision_rate = (versions_vec.len() - unique_count) as f64 / versions_vec.len() as f64; - - // Allow up to 15% collision rate on low-precision systems with concurrent generation - // This is acceptable because in production: - // 1. Versions are generated with more time between them - // 2. Swift uses additional mechanisms (UUIDs) to ensure uniqueness - // 3. The timestamp is primarily for ordering, not uniqueness - // 4. Concurrent generation from multiple threads on low-precision clocks can cause higher collision rates - assert!( - collision_rate < 0.15, - "High collision rate: {} unique out of {} total ({}% collisions)", - unique_count, - versions_vec.len(), - collision_rate * 100.0 - ); + assert_eq!(unique_count, versions_vec.len(), "Version names should be unique"); } /// Test that archive and restore preserve object path structure @@ -366,18 +343,9 @@ fn test_version_high_count_performance() { duration.as_millis() ); - // Check uniqueness - allow some collisions on low-precision systems + // Check uniqueness. let unique_count = versions.iter().collect::>().len(); - let collision_rate = (versions.len() - unique_count) as f64 / versions.len() as f64; - - // Allow up to 5% collision rate - assert!( - collision_rate < 0.05, - "High collision rate: {} collisions out of {} ({}%)", - versions.len() - unique_count, - versions.len(), - collision_rate * 100.0 - ); + assert_eq!(unique_count, versions.len(), "Version names should be unique"); } /// Test version name format stability diff --git a/rustfs/src/app/bucket_usecase.rs b/rustfs/src/app/bucket_usecase.rs index b8f29db05..9ead490bb 100644 --- a/rustfs/src/app/bucket_usecase.rs +++ b/rustfs/src/app/bucket_usecase.rs @@ -93,30 +93,31 @@ use s3s::dto::{ DeleteBucketCorsInput, DeleteBucketCorsOutput, DeleteBucketEncryptionInput, DeleteBucketEncryptionOutput, DeleteBucketInput, DeleteBucketLifecycleInput, DeleteBucketLifecycleOutput, DeleteBucketOutput, DeleteBucketPolicyInput, DeleteBucketPolicyOutput, DeleteBucketReplicationInput, DeleteBucketReplicationOutput, DeleteBucketTaggingInput, - DeleteBucketTaggingOutput, DeleteMarkerM, DeletePublicAccessBlockInput, DeletePublicAccessBlockOutput, EncodingType, + DeleteBucketTaggingOutput, DeleteMarkerEntry, DeletePublicAccessBlockInput, DeletePublicAccessBlockOutput, EncodingType, ExpirationStatus, GetBucketCorsInput, GetBucketCorsOutput, GetBucketEncryptionInput, GetBucketEncryptionOutput, GetBucketLifecycleConfigurationInput, GetBucketLifecycleConfigurationOutput, GetBucketLocationInput, GetBucketLocationOutput, GetBucketNotificationConfigurationInput, GetBucketNotificationConfigurationOutput, GetBucketPolicyInput, GetBucketPolicyOutput, GetBucketPolicyStatusInput, GetBucketPolicyStatusOutput, GetBucketReplicationInput, GetBucketReplicationOutput, GetBucketTaggingInput, GetBucketTaggingOutput, GetBucketVersioningInput, GetBucketVersioningOutput, GetPublicAccessBlockInput, GetPublicAccessBlockOutput, HeadBucketInput, HeadBucketOutput, - LifecycleRule, ListBucketsInput, ListBucketsOutput, ListObjectVersionMEntry, ListObjectVersionsInput, - ListObjectVersionsMOutput, ListObjectVersionsOutput, ListObjectsInput, ListObjectsOutput, ListObjectsV2Input, - ListObjectsV2MOutput, ListObjectsV2Output, NotificationConfiguration, NotificationConfigurationFilter, ObjectInternalInfo, - ObjectLockConfiguration, ObjectM, ObjectStorageClass, ObjectVersionM, ObjectVersionStorageClass, PolicyStatus, - PutBucketCorsInput, PutBucketCorsOutput, PutBucketEncryptionInput, PutBucketEncryptionOutput, - PutBucketLifecycleConfigurationInput, PutBucketLifecycleConfigurationOutput, PutBucketNotificationConfigurationInput, - PutBucketNotificationConfigurationOutput, PutBucketPolicyInput, PutBucketPolicyOutput, PutBucketReplicationInput, - PutBucketReplicationOutput, PutBucketTaggingInput, PutBucketTaggingOutput, PutBucketVersioningInput, - PutBucketVersioningOutput, PutPublicAccessBlockInput, PutPublicAccessBlockOutput, ReplicationConfiguration, - ServerSideEncryption, Tagging, Timestamp, UserMetadataCollection, UserMetadataEntry, VersioningConfiguration, + LifecycleRule, ListBucketsInput, ListBucketsOutput, ListObjectVersionsInput, ListObjectVersionsOutput, ListObjectsInput, + ListObjectsOutput, ListObjectsV2Input, ListObjectsV2Output, MetadataEntry, NotificationConfiguration, + NotificationConfigurationFilter, Object, ObjectLockConfiguration, ObjectStorageClass, ObjectVersion, + ObjectVersionStorageClass, PolicyStatus, PutBucketCorsInput, PutBucketCorsOutput, PutBucketEncryptionInput, + PutBucketEncryptionOutput, PutBucketLifecycleConfigurationInput, PutBucketLifecycleConfigurationOutput, + PutBucketNotificationConfigurationInput, PutBucketNotificationConfigurationOutput, PutBucketPolicyInput, + PutBucketPolicyOutput, PutBucketReplicationInput, PutBucketReplicationOutput, PutBucketTaggingInput, PutBucketTaggingOutput, + PutBucketVersioningInput, PutBucketVersioningOutput, PutPublicAccessBlockInput, PutPublicAccessBlockOutput, + ReplicationConfiguration, ServerSideEncryption, Tagging, Timestamp, UserMetadata, VersioningConfiguration, }; use s3s::region::Region; use s3s::xml; +use s3s::xml::{SerResult, SerializeContent}; use s3s::{S3Error, S3ErrorCode, S3Request, S3Response, S3Result, s3_error}; use std::{ collections::{HashMap, HashSet}, fmt::Display, + io::Write, sync::Arc, }; use tracing::{debug, error, info, instrument, warn}; @@ -128,6 +129,208 @@ use urlencoding::encode; type ListObjectVersionsInfo = StorageListObjectVersionsInfo; type ListObjectsV2Info = StorageListObjectsV2Info; +const XMLNS_S3: &str = "http://s3.amazonaws.com/doc/2006-03-01/"; + +#[derive(Clone, Debug, Default, PartialEq)] +pub(crate) struct ObjectInternalInfo { + pub k: i32, + pub m: i32, +} + +#[derive(Clone, Debug, Default, PartialEq)] +pub(crate) struct ObjectMetadataExtension { + pub user_metadata: Option, + pub user_tags: Option, + pub internal: Option, +} + +#[derive(Clone, Debug, PartialEq)] +pub(crate) enum ListObjectVersionMetadataEntry { + DeleteMarker(DeleteMarkerEntry, ObjectMetadataExtension), + Version(ObjectVersion, ObjectMetadataExtension), +} + +#[derive(Clone, Debug, Default, PartialEq)] +pub(crate) struct ListObjectVersionsMetadataOutput { + pub output: ListObjectVersionsOutput, + pub entries: Vec, +} + +#[derive(Clone, Debug, Default, PartialEq)] +pub(crate) struct ObjectMetadataEntry { + pub object: Object, + pub extension: ObjectMetadataExtension, +} + +#[derive(Clone, Debug, Default, PartialEq)] +pub(crate) struct ListObjectsV2MetadataOutput { + pub output: ListObjectsV2Output, + pub contents: Vec, +} + +impl xml::Serialize for ListObjectVersionsMetadataOutput { + fn serialize(&self, s: &mut xml::Serializer) -> SerResult { + s.content_with_ns("ListVersionsResult", XMLNS_S3, self) + } +} + +impl SerializeContent for ListObjectVersionsMetadataOutput { + fn serialize_content(&self, s: &mut xml::Serializer) -> SerResult { + if let Some(iter) = &self.output.common_prefixes { + s.flattened_list("CommonPrefixes", iter)?; + } + if let Some(ref val) = self.output.delimiter { + s.content("Delimiter", val)?; + } + if let Some(ref val) = self.output.encoding_type { + s.content("EncodingType", val)?; + } + if let Some(ref val) = self.output.is_truncated { + s.content("IsTruncated", val)?; + } + if let Some(ref val) = self.output.key_marker { + s.content("KeyMarker", val)?; + } + if let Some(ref val) = self.output.max_keys { + s.content("MaxKeys", val)?; + } + if let Some(ref val) = self.output.name { + s.content("Name", val)?; + } + if let Some(ref val) = self.output.next_key_marker { + s.content("NextKeyMarker", val)?; + } + if let Some(ref val) = self.output.next_version_id_marker { + s.content("NextVersionIdMarker", val)?; + } + if let Some(ref val) = self.output.prefix { + s.content("Prefix", val)?; + } + if let Some(ref val) = self.output.version_id_marker { + s.content("VersionIdMarker", val)?; + } + for entry in &self.entries { + match entry { + ListObjectVersionMetadataEntry::DeleteMarker(marker, extension) => { + s.content( + "DeleteMarker", + &VersionMetadataContent { + value: marker, + extension, + }, + )?; + } + ListObjectVersionMetadataEntry::Version(version, extension) => { + s.content( + "Version", + &VersionMetadataContent { + value: version, + extension, + }, + )?; + } + } + } + Ok(()) + } +} + +impl xml::Serialize for ListObjectsV2MetadataOutput { + fn serialize(&self, s: &mut xml::Serializer) -> SerResult { + s.content_with_ns("ListBucketResult", XMLNS_S3, self) + } +} + +impl SerializeContent for ListObjectsV2MetadataOutput { + fn serialize_content(&self, s: &mut xml::Serializer) -> SerResult { + if let Some(ref val) = self.output.name { + s.content("Name", val)?; + } + if let Some(ref val) = self.output.prefix { + s.content("Prefix", val)?; + } + if let Some(ref val) = self.output.max_keys { + s.content("MaxKeys", val)?; + } + if let Some(ref val) = self.output.key_count { + s.content("KeyCount", val)?; + } + if let Some(ref val) = self.output.continuation_token { + s.content("ContinuationToken", val)?; + } + if let Some(ref val) = self.output.is_truncated { + s.content("IsTruncated", val)?; + } + if let Some(ref val) = self.output.next_continuation_token { + s.content("NextContinuationToken", val)?; + } + for entry in &self.contents { + s.content("Contents", entry)?; + } + if let Some(iter) = &self.output.common_prefixes { + s.flattened_list("CommonPrefixes", iter)?; + } + if let Some(ref val) = self.output.delimiter { + s.content("Delimiter", val)?; + } + if let Some(ref val) = self.output.encoding_type { + s.content("EncodingType", val)?; + } + if let Some(ref val) = self.output.start_after { + s.content("StartAfter", val)?; + } + Ok(()) + } +} + +struct VersionMetadataContent<'a, T> { + value: &'a T, + extension: &'a ObjectMetadataExtension, +} + +impl SerializeContent for VersionMetadataContent<'_, T> { + fn serialize_content(&self, s: &mut xml::Serializer) -> SerResult { + self.value.serialize_content(s)?; + self.extension.serialize_content(s) + } +} + +impl SerializeContent for ObjectMetadataEntry { + fn serialize_content(&self, s: &mut xml::Serializer) -> SerResult { + self.object.serialize_content(s)?; + self.extension.serialize_content(s) + } +} + +impl SerializeContent for ObjectMetadataExtension { + fn serialize_content(&self, s: &mut xml::Serializer) -> SerResult { + if let Some(metadata) = &self.user_metadata { + s.element("UserMetadata", |s| { + for entry in metadata { + if let (Some(name), Some(value)) = (&entry.name, &entry.value) { + s.content(name, value)?; + } + } + Ok(()) + })?; + } + if let Some(tags) = &self.user_tags { + s.content("UserTags", tags)?; + } + if let Some(internal) = &self.internal { + s.content("Internal", internal)?; + } + Ok(()) + } +} + +impl SerializeContent for ObjectInternalInfo { + fn serialize_content(&self, s: &mut xml::Serializer) -> SerResult { + s.content("K", &self.k)?; + s.content("M", &self.m) + } +} + fn serialize_config(value: &T) -> S3Result> { serialize(value).map_err(to_internal_error) } @@ -421,19 +624,22 @@ fn encode_list_objects_v2_value(value: &str, encoding_type: Option<&EncodingType } } -fn build_metadata_extension_user_metadata(user_defined: &HashMap) -> Option { +fn build_metadata_extension_user_metadata(user_defined: &HashMap) -> Option { let mut items = extract_user_defined_metadata(user_defined) .into_iter() .filter(|(key, _)| !key.is_empty()) - .map(|(key, value)| UserMetadataEntry { key, value }) + .map(|(key, value)| MetadataEntry { + name: Some(key), + value: Some(value), + }) .collect::>(); - items.sort_by(|left, right| left.key.cmp(&right.key)); + items.sort_by(|left, right| left.name.cmp(&right.name)); - if items.is_empty() { - None - } else { - Some(UserMetadataCollection { items }) - } + if items.is_empty() { None } else { Some(items) } +} + +fn metadata_count_to_i32(value: usize) -> i32 { + i32::try_from(value).unwrap_or(i32::MAX) } async fn is_list_objects_metadata_action_allowed( @@ -496,13 +702,13 @@ async fn collect_list_objects_metadata_permissions( Ok(permissions) } -fn build_list_object_versions_m_output( +fn build_list_object_versions_metadata_output( object_infos: ListObjectVersionsInfo, bucket: &str, params: &ListObjectVersionsParams, encoding_type: Option<&EncodingType>, permissions: &HashMap, -) -> ListObjectVersionsMOutput { +) -> ListObjectVersionsMetadataOutput { let owner = rustfs_owner(); let common_prefixes = object_infos .prefixes @@ -535,42 +741,49 @@ fn build_list_object_versions_m_output( }; let internal = if permission.metadata_allowed && (object.data_blocks > 0 || object.parity_blocks > 0) { Some(ObjectInternalInfo { - k: object.data_blocks as i32, - m: object.parity_blocks as i32, + k: metadata_count_to_i32(object.data_blocks), + m: metadata_count_to_i32(object.parity_blocks), }) } else { None }; + let extension = ObjectMetadataExtension { + user_metadata, + user_tags, + internal, + }; + if object.delete_marker { - ListObjectVersionMEntry::DeleteMarker(DeleteMarkerM { - key: Some(object_name), - last_modified: object.mod_time.map(Timestamp::from), - owner: Some(owner.clone()), - version_id: Some(version_id), - is_latest: Some(object.is_latest), - user_metadata, - user_tags, - internal, - }) + ListObjectVersionMetadataEntry::DeleteMarker( + DeleteMarkerEntry { + key: Some(object_name), + last_modified: object.mod_time.map(Timestamp::from), + owner: Some(owner.clone()), + version_id: Some(version_id), + is_latest: Some(object.is_latest), + }, + extension, + ) } else { - ListObjectVersionMEntry::Version(ObjectVersionM { - key: Some(object_name), - last_modified: object.mod_time.map(Timestamp::from), - size: Some(object.size), - version_id: Some(version_id), - is_latest: Some(object.is_latest), - e_tag: object.etag.clone().map(|etag| to_s3s_etag(&etag)), - storage_class: Some(ObjectVersionStorageClass::from( - object - .storage_class - .unwrap_or_else(|| ObjectVersionStorageClass::STANDARD.to_string()), - )), - owner: Some(owner.clone()), - user_metadata, - user_tags, - internal, - }) + ListObjectVersionMetadataEntry::Version( + ObjectVersion { + key: Some(object_name), + last_modified: object.mod_time.map(Timestamp::from), + size: Some(object.size), + version_id: Some(version_id), + is_latest: Some(object.is_latest), + e_tag: object.etag.clone().map(|etag| to_s3s_etag(&etag)), + storage_class: Some(ObjectVersionStorageClass::from( + object + .storage_class + .unwrap_or_else(|| ObjectVersionStorageClass::STANDARD.to_string()), + )), + owner: Some(owner.clone()), + ..Default::default() + }, + extension, + ) } }) .collect::>(); @@ -581,37 +794,40 @@ fn build_list_object_versions_m_output( .map(|marker| encode_list_versions_value(&marker, encoding_type)); let next_version_id_marker = object_infos.next_version_idmarker.filter(|marker| !marker.is_empty()); - ListObjectVersionsMOutput { - common_prefixes: Some(common_prefixes), - delimiter: params - .delimiter - .clone() - .map(|value| encode_list_versions_value(&value, encoding_type)), - encoding_type: encoding_type.cloned(), - is_truncated: Some(object_infos.is_truncated), - key_marker: Some(encode_list_versions_value( - params.key_marker.as_deref().unwrap_or_default(), - encoding_type, - )), - max_keys: Some(params.max_keys), - name: Some(bucket.to_owned()), - next_key_marker, - next_version_id_marker, - prefix: Some(encode_list_versions_value(¶ms.prefix, encoding_type)), - request_charged: None, - version_id_marker: Some(params.version_id_marker.clone().unwrap_or_default()), + ListObjectVersionsMetadataOutput { + output: ListObjectVersionsOutput { + common_prefixes: Some(common_prefixes), + delimiter: params + .delimiter + .clone() + .map(|value| encode_list_versions_value(&value, encoding_type)), + encoding_type: encoding_type.cloned(), + is_truncated: Some(object_infos.is_truncated), + key_marker: Some(encode_list_versions_value( + params.key_marker.as_deref().unwrap_or_default(), + encoding_type, + )), + max_keys: Some(params.max_keys), + name: Some(bucket.to_owned()), + next_key_marker, + next_version_id_marker, + prefix: Some(encode_list_versions_value(¶ms.prefix, encoding_type)), + request_charged: None, + version_id_marker: Some(params.version_id_marker.clone().unwrap_or_default()), + ..Default::default() + }, entries, } } -fn build_list_objects_v2m_output( +fn build_list_objects_v2_metadata_output( object_infos: ListObjectsV2Info, bucket: &str, params: &ListObjectsV2Params, encoding_type: Option<&EncodingType>, fetch_owner: bool, permissions: &HashMap, -) -> ListObjectsV2MOutput { +) -> ListObjectsV2MetadataOutput { let owner = rustfs_owner(); let contents = object_infos @@ -632,28 +848,33 @@ fn build_list_objects_v2m_output( }; let internal = if permission.metadata_allowed && (object.data_blocks > 0 || object.parity_blocks > 0) { Some(ObjectInternalInfo { - k: object.data_blocks as i32, - m: object.parity_blocks as i32, + k: metadata_count_to_i32(object.data_blocks), + m: metadata_count_to_i32(object.parity_blocks), }) } else { None }; - ObjectM { - key: Some(encode_list_objects_v2_value(&object.name, encoding_type)), - last_modified: object.mod_time.map(Timestamp::from), - size: Some(object.get_actual_size().unwrap_or_default()), - e_tag: object.etag.clone().map(|etag| to_s3s_etag(&etag)), - storage_class: Some(ObjectStorageClass::from( - object - .storage_class - .clone() - .unwrap_or_else(|| ObjectStorageClass::STANDARD.to_string()), - )), - owner: fetch_owner.then_some(owner.clone()), - user_metadata, - user_tags, - internal, + ObjectMetadataEntry { + object: Object { + key: Some(encode_list_objects_v2_value(&object.name, encoding_type)), + last_modified: object.mod_time.map(Timestamp::from), + size: Some(object.get_actual_size().unwrap_or_default()), + e_tag: object.etag.clone().map(|etag| to_s3s_etag(&etag)), + storage_class: Some(ObjectStorageClass::from( + object + .storage_class + .clone() + .unwrap_or_else(|| ObjectStorageClass::STANDARD.to_string()), + )), + owner: fetch_owner.then_some(owner.clone()), + ..Default::default() + }, + extension: ObjectMetadataExtension { + user_metadata, + user_tags, + internal, + }, } }) .collect::>(); @@ -666,25 +887,27 @@ fn build_list_objects_v2m_output( }) .collect::>(); - let key_count = (contents.len() + common_prefixes.len()) as i32; + let key_count = metadata_count_to_i32(contents.len() + common_prefixes.len()); let next_continuation_token = object_infos .next_continuation_token .map(|token| base64_simd::STANDARD.encode_to_string(token.as_bytes())); - ListObjectsV2MOutput { - name: Some(bucket.to_owned()), - prefix: Some(params.prefix.clone()), - max_keys: Some(params.max_keys), - key_count: Some(key_count), - continuation_token: params.response_continuation_token.clone(), - is_truncated: Some(object_infos.is_truncated), - next_continuation_token, - contents: Some(contents), - common_prefixes: Some(common_prefixes), - delimiter: params.delimiter.clone(), - encoding_type: encoding_type.cloned(), - start_after: params.response_start_after.clone(), - ..Default::default() + ListObjectsV2MetadataOutput { + output: ListObjectsV2Output { + name: Some(bucket.to_owned()), + prefix: Some(params.prefix.clone()), + max_keys: Some(params.max_keys), + key_count: Some(key_count), + continuation_token: params.response_continuation_token.clone(), + is_truncated: Some(object_infos.is_truncated), + next_continuation_token, + common_prefixes: Some(common_prefixes), + delimiter: params.delimiter.clone(), + encoding_type: encoding_type.cloned(), + start_after: params.response_start_after.clone(), + ..Default::default() + }, + contents, } } @@ -2156,10 +2379,10 @@ impl DefaultBucketUsecase { Ok(S3Response::new(output)) } - pub async fn execute_list_objects_v2m( + pub(crate) async fn execute_list_objects_v2m( &self, req: S3Request, - ) -> S3Result> { + ) -> S3Result> { let input = req.input.clone(); let ListObjectsV2Input { bucket, @@ -2197,7 +2420,7 @@ impl DefaultBucketUsecase { .map_err(ApiError::from)?; let permissions = collect_list_objects_metadata_permissions(&req, &bucket, &object_infos.objects).await?; - let output = build_list_objects_v2m_output( + let output = build_list_objects_v2_metadata_output( object_infos, &bucket, ¶ms, @@ -2245,10 +2468,10 @@ impl DefaultBucketUsecase { Ok(S3Response::new(output)) } - pub async fn execute_list_object_versions_m( + pub(crate) async fn execute_list_object_versions_m( &self, req: S3Request, - ) -> S3Result> { + ) -> S3Result> { let input = req.input.clone(); let ListObjectVersionsInput { bucket, @@ -2277,7 +2500,8 @@ impl DefaultBucketUsecase { .map_err(ApiError::from)?; let permissions = collect_list_objects_metadata_permissions(&req, &bucket, &object_infos.objects).await?; - let output = build_list_object_versions_m_output(object_infos, &bucket, ¶ms, encoding_type.as_ref(), &permissions); + let output = + build_list_object_versions_metadata_output(object_infos, &bucket, ¶ms, encoding_type.as_ref(), &permissions); Ok(S3Response::new(output)) } @@ -3009,7 +3233,7 @@ mod tests { } #[test] - fn build_list_object_versions_m_output_maps_metadata_and_preserves_entry_order() { + fn build_list_object_versions_metadata_output_maps_metadata_and_preserves_entry_order() { use time::macros::datetime; use uuid::Uuid; @@ -3069,7 +3293,7 @@ mod tests { version_id_marker: Some("vid-1".to_string()), max_keys: 1000, }; - let output = build_list_object_versions_m_output( + let output = build_list_object_versions_metadata_output( object_infos, "demo-bucket", ¶ms, @@ -3077,24 +3301,24 @@ mod tests { &permissions, ); - assert_eq!(output.name.as_deref(), Some("demo-bucket")); - assert_eq!(output.prefix.as_deref(), Some("pre")); - assert_eq!(output.key_marker.as_deref(), Some("start%20marker")); - assert_eq!(output.next_key_marker.as_deref(), Some("obj-z")); - assert_eq!(output.next_version_id_marker.as_deref(), Some("null")); + assert_eq!(output.output.name.as_deref(), Some("demo-bucket")); + assert_eq!(output.output.prefix.as_deref(), Some("pre")); + assert_eq!(output.output.key_marker.as_deref(), Some("start%20marker")); + assert_eq!(output.output.next_key_marker.as_deref(), Some("obj-z")); + assert_eq!(output.output.next_version_id_marker.as_deref(), Some("null")); assert_eq!(output.entries.len(), 2); match &output.entries[0] { - ListObjectVersionMEntry::Version(version) => { + ListObjectVersionMetadataEntry::Version(version, extension) => { assert_eq!(version.key.as_deref(), Some("obj-a")); assert_eq!(version.version_id.as_deref(), Some(Uuid::nil().to_string().as_str())); - assert_eq!(version.user_tags.as_deref(), Some("env=prod")); - assert_eq!(version.internal, Some(ObjectInternalInfo { k: 4, m: 2 })); + assert_eq!(extension.user_tags.as_deref(), Some("env=prod")); + assert_eq!(extension.internal, Some(ObjectInternalInfo { k: 4, m: 2 })); assert_eq!( - version.user_metadata.as_ref().map(|metadata| metadata.items.clone()), - Some(vec![UserMetadataEntry { - key: "project".to_string(), - value: "alpha".to_string(), + extension.user_metadata.clone(), + Some(vec![MetadataEntry { + name: Some("project".to_string()), + value: Some("alpha".to_string()), }]) ); } @@ -3102,15 +3326,15 @@ mod tests { } match &output.entries[1] { - ListObjectVersionMEntry::DeleteMarker(marker) => { + ListObjectVersionMetadataEntry::DeleteMarker(marker, extension) => { assert_eq!(marker.key.as_deref(), Some("obj-b")); assert_eq!(marker.version_id.as_deref(), Some("null")); - assert!(marker.user_tags.is_none()); + assert!(extension.user_tags.is_none()); assert_eq!( - marker.user_metadata.as_ref().map(|metadata| metadata.items.clone()), - Some(vec![UserMetadataEntry { - key: "marker".to_string(), - value: "true".to_string(), + extension.user_metadata.clone(), + Some(vec![MetadataEntry { + name: Some("marker".to_string()), + value: Some("true".to_string()), }]) ); } @@ -3119,7 +3343,7 @@ mod tests { } #[test] - fn build_list_object_versions_m_output_uses_params_and_hides_metadata_without_permissions() { + fn build_list_object_versions_metadata_output_uses_params_and_hides_metadata_without_permissions() { use time::macros::datetime; let object_infos = ListObjectVersionsInfo { @@ -3148,7 +3372,7 @@ mod tests { max_keys: 25, }; - let output = build_list_object_versions_m_output( + let output = build_list_object_versions_metadata_output( object_infos, "demo-bucket", ¶ms, @@ -3156,20 +3380,20 @@ mod tests { &HashMap::new(), ); - assert_eq!(output.name.as_deref(), Some("demo-bucket")); - assert_eq!(output.prefix.as_deref(), Some("logs%20and%20more%2F")); - assert_eq!(output.delimiter.as_deref(), Some("%20")); - assert_eq!(output.key_marker.as_deref(), Some("marker%20value")); - assert_eq!(output.version_id_marker.as_deref(), Some("")); - assert_eq!(output.next_key_marker, None); - assert_eq!(output.next_version_id_marker, None); + assert_eq!(output.output.name.as_deref(), Some("demo-bucket")); + assert_eq!(output.output.prefix.as_deref(), Some("logs%20and%20more%2F")); + assert_eq!(output.output.delimiter.as_deref(), Some("%20")); + assert_eq!(output.output.key_marker.as_deref(), Some("marker%20value")); + assert_eq!(output.output.version_id_marker.as_deref(), Some("")); + assert_eq!(output.output.next_key_marker, None); + assert_eq!(output.output.next_version_id_marker, None); match &output.entries[0] { - ListObjectVersionMEntry::Version(version) => { + ListObjectVersionMetadataEntry::Version(version, extension) => { assert_eq!(version.key.as_deref(), Some("logs%20and%20more%2Fobject%20one.txt")); - assert!(version.user_metadata.is_none()); - assert!(version.user_tags.is_none()); - assert!(version.internal.is_none()); + assert!(extension.user_metadata.is_none()); + assert!(extension.user_tags.is_none()); + assert!(extension.internal.is_none()); } other => panic!("expected version entry, got {other:?}"), } @@ -3201,7 +3425,7 @@ mod tests { } #[test] - fn build_list_objects_v2m_output_maps_metadata_and_key_count() { + fn build_list_objects_v2_metadata_output_maps_metadata_and_key_count() { use time::macros::datetime; let object_infos = ListObjectsV2Info { @@ -3241,7 +3465,7 @@ mod tests { decoded_continuation_token: None, }; - let output = build_list_objects_v2m_output( + let output = build_list_objects_v2_metadata_output( object_infos, "demo-bucket", ¶ms, @@ -3250,34 +3474,82 @@ mod tests { &permissions, ); - assert_eq!(output.name.as_deref(), Some("demo-bucket")); - assert_eq!(output.prefix.as_deref(), Some("logs/")); - assert_eq!(output.continuation_token.as_deref(), Some("start token")); - assert_eq!(output.start_after.as_deref(), Some("logs/start after")); - assert_eq!(output.next_continuation_token.as_deref(), Some("bmV4dC10b2tlbg==")); - assert_eq!(output.key_count, Some(2)); - assert_eq!(output.contents.as_ref().map(Vec::len), Some(1)); - assert_eq!(output.common_prefixes.as_ref().map(Vec::len), Some(1)); + assert_eq!(output.output.name.as_deref(), Some("demo-bucket")); + assert_eq!(output.output.prefix.as_deref(), Some("logs/")); + assert_eq!(output.output.continuation_token.as_deref(), Some("start token")); + assert_eq!(output.output.start_after.as_deref(), Some("logs/start after")); + assert_eq!(output.output.next_continuation_token.as_deref(), Some("bmV4dC10b2tlbg==")); + assert_eq!(output.output.key_count, Some(2)); + assert_eq!(output.contents.len(), 1); + assert_eq!(output.output.common_prefixes.as_ref().map(Vec::len), Some(1)); - let object = output.contents.as_ref().unwrap().first().unwrap(); - assert_eq!(object.key.as_deref(), Some("logs/obj%20a.txt")); - assert_eq!(object.user_tags.as_deref(), Some("env=prod")); - assert_eq!(object.internal, Some(ObjectInternalInfo { k: 4, m: 2 })); - assert!(object.owner.is_some()); + let entry = output.contents.first().unwrap(); + assert_eq!(entry.object.key.as_deref(), Some("logs/obj%20a.txt")); + assert_eq!(entry.extension.user_tags.as_deref(), Some("env=prod")); + assert_eq!(entry.extension.internal, Some(ObjectInternalInfo { k: 4, m: 2 })); + assert!(entry.object.owner.is_some()); assert_eq!( - object.user_metadata.as_ref().map(|metadata| metadata.items.clone()), - Some(vec![UserMetadataEntry { - key: "project".to_string(), - value: "alpha".to_string(), + entry.extension.user_metadata.clone(), + Some(vec![MetadataEntry { + name: Some("project".to_string()), + value: Some("alpha".to_string()), }]) ); - let prefix = output.common_prefixes.as_ref().unwrap().first().unwrap(); + let prefix = output.output.common_prefixes.as_ref().unwrap().first().unwrap(); assert_eq!(prefix.prefix.as_deref(), Some("logs/archive/")); } #[test] - fn build_list_objects_v2m_output_uses_params_and_hides_owner_without_fetch_owner() { + fn list_objects_v2_metadata_output_serializes_minio_extension_xml() { + use time::macros::datetime; + + let object_infos = ListObjectsV2Info { + is_truncated: false, + objects: vec![ObjectInfo { + bucket: "demo-bucket".to_string(), + name: "logs/obj-a.txt".to_string(), + mod_time: Some(datetime!(2025-01-03 00:00 UTC)), + size: 11, + user_defined: Arc::new(HashMap::from([("project".to_string(), "alpha".to_string())])), + parity_blocks: 2, + data_blocks: 4, + user_tags: Arc::new("env=prod&project=alpha".to_string()), + ..Default::default() + }], + ..Default::default() + }; + let permissions = HashMap::from([( + "logs/obj-a.txt".to_string(), + ObjectMetadataPermissions { + metadata_allowed: true, + tags_allowed: true, + }, + )]); + let params = ListObjectsV2Params { + prefix: "logs/".to_string(), + max_keys: 1000, + delimiter: None, + response_start_after: None, + start_after_for_query: None, + response_continuation_token: None, + decoded_continuation_token: None, + }; + + let output = build_list_objects_v2_metadata_output(object_infos, "demo-bucket", ¶ms, None, false, &permissions); + let xml = String::from_utf8(serialize_config(&output).expect("metadata output should serialize")) + .expect("metadata output should be UTF-8"); + + assert!(xml.contains("")); + assert!(xml.contains("alpha")); + assert!(xml.contains("env=prod&project=alpha")); + assert!(xml.contains("42")); + assert!(!xml.contains("")); + } + + #[test] + fn build_list_objects_v2_metadata_output_uses_params_and_hides_owner_without_fetch_owner() { use time::macros::datetime; let object_infos = ListObjectsV2Info { @@ -3308,7 +3580,7 @@ mod tests { decoded_continuation_token: Some("decoded token".to_string()), }; - let output = build_list_objects_v2m_output( + let output = build_list_objects_v2_metadata_output( object_infos, "demo-bucket", ¶ms, @@ -3317,22 +3589,22 @@ mod tests { &HashMap::new(), ); - assert_eq!(output.name.as_deref(), Some("demo-bucket")); - assert_eq!(output.prefix.as_deref(), Some("logs and more/")); - assert_eq!(output.delimiter.as_deref(), Some("/")); - assert_eq!(output.continuation_token.as_deref(), Some("opaque token")); - assert_eq!(output.start_after.as_deref(), Some("logs and more/start after")); - assert_eq!(output.key_count, Some(2)); - assert_eq!(output.encoding_type.as_ref().map(EncodingType::as_str), Some(EncodingType::URL)); + assert_eq!(output.output.name.as_deref(), Some("demo-bucket")); + assert_eq!(output.output.prefix.as_deref(), Some("logs and more/")); + assert_eq!(output.output.delimiter.as_deref(), Some("/")); + assert_eq!(output.output.continuation_token.as_deref(), Some("opaque token")); + assert_eq!(output.output.start_after.as_deref(), Some("logs and more/start after")); + assert_eq!(output.output.key_count, Some(2)); + assert_eq!(output.output.encoding_type.as_ref().map(EncodingType::as_str), Some(EncodingType::URL)); - let object = output.contents.as_ref().unwrap().first().unwrap(); - assert_eq!(object.key.as_deref(), Some("logs%20and%20more/object%20one.txt")); - assert!(object.owner.is_none()); - assert!(object.user_metadata.is_none()); - assert!(object.user_tags.is_none()); - assert!(object.internal.is_none()); + let entry = output.contents.first().unwrap(); + assert_eq!(entry.object.key.as_deref(), Some("logs%20and%20more/object%20one.txt")); + assert!(entry.object.owner.is_none()); + assert!(entry.extension.user_metadata.is_none()); + assert!(entry.extension.user_tags.is_none()); + assert!(entry.extension.internal.is_none()); - let prefix = output.common_prefixes.as_ref().unwrap().first().unwrap(); + let prefix = output.output.common_prefixes.as_ref().unwrap().first().unwrap(); assert_eq!(prefix.prefix.as_deref(), Some("logs%20and%20more/archive/")); } diff --git a/rustfs/src/app/metadata_route.rs b/rustfs/src/app/metadata_route.rs new file mode 100644 index 000000000..f1607982a --- /dev/null +++ b/rustfs/src/app/metadata_route.rs @@ -0,0 +1,386 @@ +// 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. + +//! MinIO-compatible metadata listing extension routes. + +use super::bucket_usecase::DefaultBucketUsecase; +use crate::storage::access::{ReqInfo, authorize_request, req_info_mut}; +use async_trait::async_trait; +use http::header::CONTENT_TYPE; +use http::header::HOST; +use http::{Extensions, HeaderMap, HeaderValue, Method, Uri}; +use rustfs_policy::policy::action::{Action, S3Action}; +use s3s::dto::{EncodingType, ListObjectVersionsInput, ListObjectsV2Input}; +use s3s::host::{MultiDomain, S3Host}; +use s3s::route::S3Route; +use s3s::xml; +use s3s::{Body, S3Error, S3Request, S3Response, S3Result, s3_error}; +use url::form_urlencoded; + +pub(crate) struct MetadataRoute { + admin: A, + host: Option, +} + +pub(crate) fn with_metadata_route(admin: A, host: Option) -> MetadataRoute { + MetadataRoute { admin, host } +} + +#[async_trait] +impl S3Route for MetadataRoute +where + A: S3Route, +{ + fn is_match(&self, method: &Method, uri: &Uri, headers: &HeaderMap, extensions: &mut Extensions) -> bool { + metadata_operation(method, uri, headers, self.host.as_ref()).is_some() + || self.admin.is_match(method, uri, headers, extensions) + } + + async fn check_access(&self, req: &mut S3Request) -> S3Result<()> { + if let Some(target) = metadata_operation(&req.method, &req.uri, &req.headers, self.host.as_ref()) { + check_metadata_access(req, target).await + } else { + self.admin.check_access(req).await + } + } + + async fn call(&self, req: S3Request) -> S3Result> { + match metadata_operation(&req.method, &req.uri, &req.headers, self.host.as_ref()) { + Some(BucketTarget { + operation: MetadataOperation::ListObjectVersions, + bucket, + }) => call_list_object_versions(req, bucket).await, + Some(BucketTarget { + operation: MetadataOperation::ListObjectsV2, + bucket, + }) => call_list_objects_v2(req, bucket).await, + None => self.admin.call(req).await, + } + } +} + +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +enum MetadataOperation { + ListObjectVersions, + ListObjectsV2, +} + +struct BucketTarget { + operation: MetadataOperation, + bucket: String, +} + +fn metadata_operation(method: &Method, uri: &Uri, headers: &HeaderMap, host: Option<&MultiDomain>) -> Option { + if method != Method::GET { + return None; + } + let bucket = bucket_from_request(uri, headers, host).ok()?; + + let query = uri.query()?; + if query_value(query, "metadata").as_deref() != Some("true") { + return None; + } + let operation = if query_has(query, "versions") { + MetadataOperation::ListObjectVersions + } else if query_value(query, "list-type").as_deref() == Some("2") { + MetadataOperation::ListObjectsV2 + } else { + return None; + }; + + Some(BucketTarget { operation, bucket }) +} + +async fn check_metadata_access(req: &mut S3Request, target: BucketTarget) -> S3Result<()> { + { + if req_info_mut(req).is_err() { + req.extensions.insert(ReqInfo::default()); + } + let req_info = req_info_mut(req)?; + req_info.bucket = Some(target.bucket); + } + + let action = match target.operation { + MetadataOperation::ListObjectVersions => Action::S3Action(S3Action::ListBucketVersionsAction), + MetadataOperation::ListObjectsV2 => Action::S3Action(S3Action::ListBucketAction), + }; + authorize_request(req, action).await +} + +async fn call_list_object_versions(req: S3Request, bucket: String) -> S3Result> { + let input = list_object_versions_input(bucket, &req.uri, &req.headers)?; + let request = req.map_input(|_| input); + let output = DefaultBucketUsecase::from_global() + .execute_list_object_versions_m(request) + .await?; + xml_response(&output.output) +} + +async fn call_list_objects_v2(req: S3Request, bucket: String) -> S3Result> { + let input = list_objects_v2_input(bucket, &req.uri, &req.headers)?; + let request = req.map_input(|_| input); + let output = DefaultBucketUsecase::from_global().execute_list_objects_v2m(request).await?; + xml_response(&output.output) +} + +fn xml_response(output: &T) -> S3Result> { + let mut body = Vec::with_capacity(1024); + { + let mut serializer = xml::Serializer::new(&mut body); + serializer + .decl() + .and_then(|()| output.serialize(&mut serializer)) + .map_err(S3Error::internal_error)?; + } + + let mut response = S3Response::new(Body::from(body)); + response + .headers + .insert(CONTENT_TYPE, HeaderValue::from_static("application/xml")); + Ok(response) +} + +fn list_object_versions_input(bucket: String, uri: &Uri, headers: &HeaderMap) -> S3Result { + let query = uri.query().unwrap_or_default(); + + Ok(ListObjectVersionsInput { + bucket, + delimiter: query_value(query, "delimiter"), + encoding_type: parse_encoding_type(query_value(query, "encoding-type"))?, + expected_bucket_owner: header_value(headers, "x-amz-expected-bucket-owner")?, + key_marker: query_value(query, "key-marker"), + max_keys: query_i32(query, "max-keys")?, + optional_object_attributes: None, + prefix: query_value(query, "prefix"), + request_payer: header_value(headers, "x-amz-request-payer")?.map(Into::into), + version_id_marker: query_value(query, "version-id-marker"), + }) +} + +fn list_objects_v2_input(bucket: String, uri: &Uri, headers: &HeaderMap) -> S3Result { + let query = uri.query().unwrap_or_default(); + + Ok(ListObjectsV2Input { + bucket, + continuation_token: query_value(query, "continuation-token"), + delimiter: query_value(query, "delimiter"), + encoding_type: parse_encoding_type(query_value(query, "encoding-type"))?, + expected_bucket_owner: header_value(headers, "x-amz-expected-bucket-owner")?, + fetch_owner: query_bool(query, "fetch-owner")?, + max_keys: query_i32(query, "max-keys")?, + optional_object_attributes: None, + prefix: query_value(query, "prefix"), + request_payer: header_value(headers, "x-amz-request-payer")?.map(Into::into), + start_after: query_value(query, "start-after"), + }) +} + +fn bucket_from_request(uri: &Uri, headers: &HeaderMap, host: Option<&MultiDomain>) -> S3Result { + if let Some(host) = host + && let Some(host_header) = headers.get(HOST).and_then(|value| value.to_str().ok()) + { + let virtual_host = host.parse_host_header(host_header)?; + if let Some(bucket) = virtual_host.bucket() { + return if uri.path() == "/" { + Ok(bucket.to_owned()) + } else { + Err(s3_error!(InvalidRequest, "bucket-level metadata route requires a bucket path")) + }; + } + } + + path_style_bucket(uri)?.ok_or_else(|| s3_error!(InvalidRequest, "bucket name is required")) +} + +fn path_style_bucket(uri: &Uri) -> S3Result> { + let path = uri.path().trim_matches('/'); + if path.is_empty() { + return Ok(None); + } + if path.contains('/') { + return Err(s3_error!(InvalidRequest, "bucket-level metadata route does not accept an object path")); + } + urlencoding::decode(path) + .map(|bucket| Some(bucket.into_owned())) + .map_err(S3Error::internal_error) +} + +fn query_has(query: &str, key: &str) -> bool { + form_urlencoded::parse(query.as_bytes()).any(|(name, _)| name == key) +} + +fn query_value(query: &str, key: &str) -> Option { + form_urlencoded::parse(query.as_bytes()) + .find(|(name, _)| name == key) + .map(|(_, value)| value.into_owned()) +} + +fn query_i32(query: &str, key: &str) -> S3Result> { + query_value(query, key) + .map(|value| { + value + .parse::() + .map_err(|_| s3_error!(InvalidArgument, "invalid integer query value")) + }) + .transpose() +} + +fn query_bool(query: &str, key: &str) -> S3Result> { + query_value(query, key) + .map(|value| { + value + .parse::() + .map_err(|_| s3_error!(InvalidArgument, "invalid boolean query value")) + }) + .transpose() +} + +fn parse_encoding_type(value: Option) -> S3Result> { + value + .map(|value| { + if value == EncodingType::URL { + Ok(EncodingType::from_static(EncodingType::URL)) + } else { + Err(s3_error!(InvalidArgument, "invalid encoding-type")) + } + }) + .transpose() +} + +fn header_value(headers: &HeaderMap, name: &str) -> S3Result> { + headers + .get(name) + .map(|value| { + value + .to_str() + .map(str::to_owned) + .map_err(|_| s3_error!(InvalidArgument, "invalid header value")) + }) + .transpose() +} + +#[cfg(test)] +mod tests { + use super::{MetadataOperation, list_object_versions_input, list_objects_v2_input, metadata_operation}; + use http::header::HOST; + use http::{HeaderMap, Method, Uri}; + use s3s::dto::EncodingType; + use s3s::host::MultiDomain; + + fn uri(value: &str) -> Uri { + value.parse().expect("test URI should parse") + } + + #[test] + fn metadata_operation_matches_path_style_extensions_only() { + assert_eq!( + metadata_operation(&Method::GET, &uri("/bucket?versions&metadata=true"), &HeaderMap::new(), None) + .map(|target| target.operation), + Some(MetadataOperation::ListObjectVersions) + ); + assert_eq!( + metadata_operation(&Method::GET, &uri("/bucket?list-type=2&metadata=true"), &HeaderMap::new(), None) + .map(|target| target.operation), + Some(MetadataOperation::ListObjectsV2) + ); + assert_eq!( + metadata_operation(&Method::GET, &uri("/?list-type=2&metadata=true"), &HeaderMap::new(), None) + .map(|target| target.operation), + None + ); + assert_eq!( + metadata_operation(&Method::GET, &uri("/bucket/key?list-type=2&metadata=true"), &HeaderMap::new(), None) + .map(|target| target.operation), + None + ); + assert_eq!( + metadata_operation(&Method::GET, &uri("/bucket?list-type=2"), &HeaderMap::new(), None).map(|target| target.operation), + None + ); + assert_eq!( + metadata_operation(&Method::PUT, &uri("/bucket?list-type=2&metadata=true"), &HeaderMap::new(), None) + .map(|target| target.operation), + None + ); + } + + #[test] + fn metadata_operation_matches_virtual_hosted_bucket_root() { + let host = MultiDomain::new(["example.com", "example.com:9000"]).expect("valid test host domain"); + let mut headers = HeaderMap::new(); + headers.insert(HOST, "demo-bucket.example.com:9000".parse().expect("valid host header")); + + let target = metadata_operation(&Method::GET, &uri("/?list-type=2&metadata=true"), &headers, Some(&host)) + .expect("virtual-hosted bucket root should match metadata route"); + assert_eq!(target.operation, MetadataOperation::ListObjectsV2); + assert_eq!(target.bucket, "demo-bucket"); + + assert_eq!( + metadata_operation(&Method::GET, &uri("/object.txt?list-type=2&metadata=true"), &headers, Some(&host)) + .map(|target| target.operation), + None + ); + } + + #[test] + fn list_objects_v2_input_parses_query_headers_and_decodes_bucket() { + let mut headers = HeaderMap::new(); + headers.insert("x-amz-expected-bucket-owner", "123456789012".parse().expect("valid header")); + headers.insert("x-amz-request-payer", "requester".parse().expect("valid header")); + + let input = list_objects_v2_input( + "demo bucket".to_string(), + &uri("/demo%20bucket?list-type=2&metadata=true&prefix=logs%2F&delimiter=%2F&encoding-type=url&fetch-owner=true&max-keys=25&continuation-token=opaque&start-after=start"), + &headers, + ) + .expect("list objects v2 input should parse"); + + assert_eq!(input.bucket, "demo bucket"); + assert_eq!(input.prefix.as_deref(), Some("logs/")); + assert_eq!(input.delimiter.as_deref(), Some("/")); + assert_eq!(input.encoding_type.as_ref().map(EncodingType::as_str), Some(EncodingType::URL)); + assert_eq!(input.fetch_owner, Some(true)); + assert_eq!(input.max_keys, Some(25)); + assert_eq!(input.continuation_token.as_deref(), Some("opaque")); + assert_eq!(input.start_after.as_deref(), Some("start")); + assert_eq!(input.expected_bucket_owner.as_deref(), Some("123456789012")); + assert_eq!(input.request_payer.as_ref().map(|payer| payer.as_str()), Some("requester")); + } + + #[test] + fn list_object_versions_input_parses_markers_and_rejects_invalid_encoding() { + let input = list_object_versions_input( + "bucket".to_string(), + &uri("/bucket?versions&metadata=true&prefix=logs%2F&delimiter=%2F&encoding-type=url&key-marker=start&version-id-marker=v1&max-keys=10"), + &HeaderMap::new(), + ) + .expect("list versions input should parse"); + + assert_eq!(input.bucket, "bucket"); + assert_eq!(input.prefix.as_deref(), Some("logs/")); + assert_eq!(input.delimiter.as_deref(), Some("/")); + assert_eq!(input.encoding_type.as_ref().map(EncodingType::as_str), Some(EncodingType::URL)); + assert_eq!(input.key_marker.as_deref(), Some("start")); + assert_eq!(input.version_id_marker.as_deref(), Some("v1")); + assert_eq!(input.max_keys, Some(10)); + + let err = list_object_versions_input( + "bucket".to_string(), + &uri("/bucket?versions&metadata=true&encoding-type=xml"), + &HeaderMap::new(), + ) + .expect_err("invalid encoding-type should be rejected"); + assert_eq!(err.code(), &s3s::S3ErrorCode::InvalidArgument); + } +} diff --git a/rustfs/src/app/mod.rs b/rustfs/src/app/mod.rs index dcfd01a51..955ce507b 100644 --- a/rustfs/src/app/mod.rs +++ b/rustfs/src/app/mod.rs @@ -18,6 +18,7 @@ pub mod admin_usecase; pub mod bucket_usecase; pub mod context; +pub(crate) mod metadata_route; pub mod multipart_usecase; pub(crate) mod object_data_cache; pub mod object_usecase; diff --git a/rustfs/src/server/http.rs b/rustfs/src/server/http.rs index d93705b5b..5edd5f372 100644 --- a/rustfs/src/server/http.rs +++ b/rustfs/src/server/http.rs @@ -529,22 +529,7 @@ pub async fn start_http_server(config: &config::Config, readiness: Arc) -> S3Result<()> { - let req_info = ext_req_info_mut(&mut req.extensions)?; - req_info.bucket = Some(req.input.bucket.clone()); - authorize_request(req, Action::S3Action(S3Action::ListBucketVersionsAction)).await - } - /// Checks whether the ListObjects request has accesses to the resources. /// /// This method returns `Ok(())` by default. @@ -1667,13 +1661,6 @@ impl S3Access for FS { authorize_request(req, Action::S3Action(S3Action::ListBucketAction)).await } - async fn list_objects_v2m(&self, req: &mut S3Request) -> S3Result<()> { - let req_info = ext_req_info_mut(&mut req.extensions)?; - req_info.bucket = Some(req.input.bucket.clone()); - - authorize_request(req, Action::S3Action(S3Action::ListBucketAction)).await - } - /// Checks whether the ListParts request has accesses to the resources. /// /// This method returns `Ok(())` by default. diff --git a/rustfs/src/storage/ecfs.rs b/rustfs/src/storage/ecfs.rs index 8b9bb36d7..0f97d59fb 100644 --- a/rustfs/src/storage/ecfs.rs +++ b/rustfs/src/storage/ecfs.rs @@ -970,15 +970,6 @@ impl S3 for FS { usecase.execute_list_object_versions(req).await } - async fn list_object_versions_m( - &self, - req: S3Request, - ) -> S3Result> { - record_s3_op(S3Operation::ListObjectVersions, &req.input.bucket); - let usecase = s3_api::default_bucket_usecase(); - usecase.execute_list_object_versions_m(req).await - } - #[instrument(level = "debug", skip(self, req))] async fn list_objects(&self, req: S3Request) -> S3Result> { record_s3_op(S3Operation::ListObjects, &req.input.bucket); @@ -993,13 +984,6 @@ impl S3 for FS { usecase.execute_list_objects_v2(req).await } - #[instrument(level = "debug", skip(self, req))] - async fn list_objects_v2m(&self, req: S3Request) -> S3Result> { - record_s3_op(S3Operation::ListObjectsV2, &req.input.bucket); - let usecase = s3_api::default_bucket_usecase(); - usecase.execute_list_objects_v2m(req).await - } - #[instrument(level = "debug", skip(self, req))] async fn list_parts(&self, req: S3Request) -> S3Result> { record_s3_op(S3Operation::ListParts, &req.input.bucket); diff --git a/rustfs/src/storage/s3_api/bucket.rs b/rustfs/src/storage/s3_api/bucket.rs index 85161d295..467a6e7ed 100644 --- a/rustfs/src/storage/s3_api/bucket.rs +++ b/rustfs/src/storage/s3_api/bucket.rs @@ -350,6 +350,10 @@ fn calculate_next_marker(v2: &ListObjectsV2Output) -> Option { return None; } + if let Some(marker) = decoded_next_continuation_marker(v2) { + return Some(encode_list_output_value(marker, v2.encoding_type.as_ref())); + } + let last_key = v2 .contents .as_ref() @@ -388,6 +392,12 @@ fn calculate_next_marker(v2: &ListObjectsV2Output) -> Option { } } +fn decoded_next_continuation_marker(v2: &ListObjectsV2Output) -> Option { + let token = v2.next_continuation_token.as_ref()?; + let bytes = base64_simd::STANDARD.decode_to_vec(token.as_bytes()).ok()?; + String::from_utf8(bytes).ok() +} + #[cfg(test)] mod tests { use super::{ @@ -476,6 +486,44 @@ mod tests { assert_eq!(output.next_marker, Some("zebra/".to_string())); } + #[test] + fn test_list_objects_next_marker_preserves_internal_continuation_marker() { + let marker = "key-998[rustfs_cache:v2,id:list-cache-id,p:0,s:0]"; + let v2 = ListObjectsV2Output { + is_truncated: Some(true), + next_continuation_token: Some(base64_simd::STANDARD.encode_to_string(marker.as_bytes())), + contents: Some(vec![Object { + key: Some("key-998".to_string()), + ..Default::default() + }]), + ..Default::default() + }; + + let output = build_list_objects_output(v2, None); + assert_eq!(output.next_marker.as_deref(), Some(marker)); + } + + #[test] + fn test_list_objects_next_marker_encodes_internal_marker_when_url_encoded() { + let marker = "dir a/key[rustfs_cache:v2,id:list-cache-id,p:0,s:0]"; + let v2 = ListObjectsV2Output { + is_truncated: Some(true), + encoding_type: Some(EncodingType::from_static(EncodingType::URL)), + next_continuation_token: Some(base64_simd::STANDARD.encode_to_string(marker.as_bytes())), + contents: Some(vec![Object { + key: Some("dir%20a/key".to_string()), + ..Default::default() + }]), + ..Default::default() + }; + + let output = build_list_objects_output(v2, None); + assert_eq!( + output.next_marker.as_deref(), + Some("dir%20a/key%5Brustfs_cache%3Av2%2Cid%3Alist-cache-id%2Cp%3A0%2Cs%3A0%5D") + ); + } + #[test] fn test_list_objects_next_marker_compares_raw_values_when_url_encoded() { let v2 = ListObjectsV2Output { diff --git a/rustfs/src/storage_api.rs b/rustfs/src/storage_api.rs index 4a650ab44..7785efc4b 100644 --- a/rustfs/src/storage_api.rs +++ b/rustfs/src/storage_api.rs @@ -99,6 +99,15 @@ pub(crate) mod server { pub(crate) type FS = crate::storage::storage_api::FS; } + pub(crate) mod metadata_route { + pub(crate) fn with_metadata_route(admin: A, host: Option) -> impl s3s::route::S3Route + where + A: s3s::route::S3Route, + { + crate::app::metadata_route::with_metadata_route(admin, host) + } + } + pub(crate) mod request_context { pub(crate) use crate::storage::storage_api::request_context_consumer::{ RequestContext, extract_request_id_from_headers,