gsadmin/rustfs
1
0
Fork 0
mirror of https://github.com/rustfs/rustfs.git synced 2026-07-26 16:28:15 +00:00
Code Issues Packages Projects Releases Wiki Activity

fix(s3): preserve metadata listing extensions (#4261)

Browse Source 
* chore(deps): update s3s to 0.14.1

* fix(s3): preserve metadata listing extensions

* fix(swift): make version names monotonic

* fix(s3): preserve v1 list pagination markers
This commit is contained in:
houseme
2026-07-05 05:03:21 +08:00
committed by GitHub
parent d0a965f2ee
commit 9b69c6d14c
12 changed files with 972 additions and 307 deletions
Download Patch File Download Diff File Expand all files Collapse all files
Cargo.lock
Generated
+9 -31
View File
@@ -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",
Cargo.toml
+1 -1
View File
@@ -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"
crates/protocols/src/swift/versioning.rs
+32 -13
View File
@@ -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");
}
}
crates/protocols/tests/swift_versioning_integration.rs
+6 -38
View File
@@ -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::<std::collections::HashSet<_>>().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::<std::collections::HashSet<_>>().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::<std::collections::HashSet<_>>().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
rustfs/src/app/bucket_usecase.rs
+450 -178
View File
@@ -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<ObjectInfo>;
type ListObjectsV2Info = StorageListObjectsV2Info<ObjectInfo>;
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<UserMetadata>,
pub user_tags: Option<String>,
pub internal: Option<ObjectInternalInfo>,
}
#[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<ListObjectVersionMetadataEntry>,
}
#[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<ObjectMetadataEntry>,
}
impl xml::Serialize for ListObjectVersionsMetadataOutput {
fn serialize<W: Write>(&self, s: &mut xml::Serializer<W>) -> SerResult {
s.content_with_ns("ListVersionsResult", XMLNS_S3, self)
}
}
impl SerializeContent for ListObjectVersionsMetadataOutput {
fn serialize_content<W: Write>(&self, s: &mut xml::Serializer<W>) -> 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<W: Write>(&self, s: &mut xml::Serializer<W>) -> SerResult {
s.content_with_ns("ListBucketResult", XMLNS_S3, self)
}
}
impl SerializeContent for ListObjectsV2MetadataOutput {
fn serialize_content<W: Write>(&self, s: &mut xml::Serializer<W>) -> 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<T: SerializeContent> SerializeContent for VersionMetadataContent<'_, T> {
fn serialize_content<W: Write>(&self, s: &mut xml::Serializer<W>) -> SerResult {
self.value.serialize_content(s)?;
self.extension.serialize_content(s)
}
}
impl SerializeContent for ObjectMetadataEntry {
fn serialize_content<W: Write>(&self, s: &mut xml::Serializer<W>) -> SerResult {
self.object.serialize_content(s)?;
self.extension.serialize_content(s)
}
}
impl SerializeContent for ObjectMetadataExtension {
fn serialize_content<W: Write>(&self, s: &mut xml::Serializer<W>) -> 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<W: Write>(&self, s: &mut xml::Serializer<W>) -> SerResult {
s.content("K", &self.k)?;
s.content("M", &self.m)
}
}
fn serialize_config<T: xml::Serialize>(value: &T) -> S3Result<Vec<u8>> {
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<String, String>) -> Option<UserMetadataCollection> {
fn build_metadata_extension_user_metadata(user_defined: &HashMap<String, String>) -> Option<UserMetadata> {
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::<Vec<_>>();
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<T>(
@@ -496,13 +702,13 @@ async fn collect_list_objects_metadata_permissions<T>(
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<String, ObjectMetadataPermissions>,
) -> 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::<Vec<_>>();
@@ -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(&params.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(&params.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<String, ObjectMetadataPermissions>,
) -> 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::<Vec<_>>();
@@ -666,25 +887,27 @@ fn build_list_objects_v2m_output(
})
.collect::<Vec<_>>();
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<ListObjectsV2Input>,
) -> S3Result<S3Response<ListObjectsV2MOutput>> {
) -> S3Result<S3Response<ListObjectsV2MetadataOutput>> {
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,
&params,
@@ -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<ListObjectVersionsInput>,
) -> S3Result<S3Response<ListObjectVersionsMOutput>> {
) -> S3Result<S3Response<ListObjectVersionsMetadataOutput>> {
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, &params, encoding_type.as_ref(), &permissions);
let output =
build_list_object_versions_metadata_output(object_infos, &bucket, &params, 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",
&params,
@@ -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",
&params,
@@ -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",
&params,
@@ -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", &params, 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("<ListBucketResult"));
assert!(xml.contains("<Contents>"));
assert!(xml.contains("<UserMetadata><project>alpha</project></UserMetadata>"));
assert!(xml.contains("<UserTags>env=prod&amp;project=alpha</UserTags>"));
assert!(xml.contains("<Internal><K>4</K><M>2</M></Internal>"));
assert!(!xml.contains("<MetadataEntry>"));
}
#[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",
&params,
@@ -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/"));
}
rustfs/src/app/metadata_route.rs
+386
View File
@@ -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<A> {
admin: A,
host: Option<MultiDomain>,
}
pub(crate) fn with_metadata_route<A>(admin: A, host: Option<MultiDomain>) -> MetadataRoute<A> {
MetadataRoute { admin, host }
}
#[async_trait]
impl<A> S3Route for MetadataRoute<A>
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<Body>) -> 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<Body>) -> S3Result<S3Response<Body>> {
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<BucketTarget> {
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<Body>, 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<Body>, bucket: String) -> S3Result<S3Response<Body>> {
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<Body>, bucket: String) -> S3Result<S3Response<Body>> {
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<T: xml::Serialize>(output: &T) -> S3Result<S3Response<Body>> {
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<ListObjectVersionsInput> {
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<ListObjectsV2Input> {
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<String> {
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<Option<String>> {
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<String> {
form_urlencoded::parse(query.as_bytes())
.find(|(name, _)| name == key)
.map(|(_, value)| value.into_owned())
}
fn query_i32(query: &str, key: &str) -> S3Result<Option<i32>> {
query_value(query, key)
.map(|value| {
value
.parse::<i32>()
.map_err(|_| s3_error!(InvalidArgument, "invalid integer query value"))
})
.transpose()
}
fn query_bool(query: &str, key: &str) -> S3Result<Option<bool>> {
query_value(query, key)
.map(|value| {
value
.parse::<bool>()
.map_err(|_| s3_error!(InvalidArgument, "invalid boolean query value"))
})
.transpose()
}
fn parse_encoding_type(value: Option<String>) -> S3Result<Option<EncodingType>> {
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<Option<String>> {
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);
}
}
rustfs/src/app/mod.rs
+1
View File
@@ -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;
rustfs/src/server/http.rs
+30 -17
View File
@@ -529,22 +529,7 @@ pub async fn start_http_server(config: &config::Config, readiness: Arc<GlobalRea
let access_key = config.access_key.clone();
let secret_key = config.secret_key.clone();
b.set_auth(IAMAuth::new(access_key, secret_key));
b.set_access(store);
b.set_route(admin::make_admin_route(config.console_enable)?);
// Normalize leading/duplicate forward slashes in object keys (MinIO parity).
// AWS S3 accepts keys such as "/foo/bar"; without this, requests like
// `PUT /bucket//foo/bar` are rejected downstream with InvalidArgument
// (ObjectNamePrefixAsSlash, issue #2427). MinIO collapses these slashes instead of preserving them,
// so `//foo/bar` is stored and served as `foo/bar`.
let mut s3_config = S3Config::default();
s3_config.normalize_forward_slash_path = true;
b.set_config(Arc::new(StaticConfigProvider::new(Arc::new(s3_config))));
// Virtual-hosted-style requests are only set up for S3 API when server domains are configured and console is disabled
if !config.server_domains.is_empty() && !config.console_enable {
let host_domain_sets = if !config.server_domains.is_empty() && !config.console_enable {
MultiDomain::new(&config.server_domains).map_err(Error::other)?; // validate domains
// add the default port number to the given server domains
@@ -558,6 +543,34 @@ pub async fn start_http_server(config: &config::Config, readiness: Arc<GlobalRea
}
}
Some(domain_sets)
} else {
None
};
let metadata_route_host = host_domain_sets
.as_ref()
.map(MultiDomain::new)
.transpose()
.map_err(Error::other)?;
b.set_auth(IAMAuth::new(access_key, secret_key));
b.set_access(store);
b.set_route(storage::metadata_route::with_metadata_route(
admin::make_admin_route(config.console_enable)?,
metadata_route_host,
));
// Normalize leading/duplicate forward slashes in object keys (MinIO parity).
// AWS S3 accepts keys such as "/foo/bar"; without this, requests like
// `PUT /bucket//foo/bar` are rejected downstream with InvalidArgument
// (ObjectNamePrefixAsSlash, issue #2427). MinIO collapses these slashes instead of preserving them,
// so `//foo/bar` is stored and served as `foo/bar`.
let mut s3_config = S3Config::default();
s3_config.normalize_forward_slash_path = true;
b.set_config(Arc::new(StaticConfigProvider::new(Arc::new(s3_config))));
// Virtual-hosted-style requests are only set up for S3 API when server domains are configured and console is disabled
if let Some(domain_sets) = host_domain_sets {
info!(
event = EVENT_HTTP_HOST_ROUTING,
component = LOG_COMPONENT_SERVER,
@@ -567,7 +580,7 @@ pub async fn start_http_server(config: &config::Config, readiness: Arc<GlobalRea
domains = ?domain_sets,
"Virtual-hosted-style routing configured"
);
b.set_host(MultiDomain::new(domain_sets).map_err(Error::other)?);
b.set_host(MultiDomain::new(&domain_sets).map_err(Error::other)?);
}
b.build()
rustfs/src/storage/access.rs
-13
View File
@@ -1641,12 +1641,6 @@ impl S3Access for FS {
authorize_request(req, Action::S3Action(S3Action::ListBucketVersionsAction)).await
}
async fn list_object_versions_m(&self, req: &mut S3Request<ListObjectVersionsInput>) -> 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<ListObjectsV2Input>) -> 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.
rustfs/src/storage/ecfs.rs
-16
View File
@@ -970,15 +970,6 @@ impl S3 for FS {
usecase.execute_list_object_versions(req).await
}
async fn list_object_versions_m(
&self,
req: S3Request<ListObjectVersionsInput>,
) -> S3Result<S3Response<ListObjectVersionsMOutput>> {
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<ListObjectsInput>) -> S3Result<S3Response<ListObjectsOutput>> {
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<ListObjectsV2Input>) -> S3Result<S3Response<ListObjectsV2MOutput>> {
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<ListPartsInput>) -> S3Result<S3Response<ListPartsOutput>> {
record_s3_op(S3Operation::ListParts, &req.input.bucket);
rustfs/src/storage/s3_api/bucket.rs
+48
View File
@@ -350,6 +350,10 @@ fn calculate_next_marker(v2: &ListObjectsV2Output) -> Option<String> {
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<String> {
}
}
fn decoded_next_continuation_marker(v2: &ListObjectsV2Output) -> Option<String> {
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 {
rustfs/src/storage_api.rs
+9
View File
@@ -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<A>(admin: A, host: Option<s3s::host::MultiDomain>) -> 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,