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

refactor(app): add application layer module entry (#1907)

Browse Source
This commit is contained in:
安正超
2026-02-22 22:15:37 +08:00
committed by GitHub
parent 4a6e81d427
commit 4211652991
15 changed files with 3009 additions and 1044 deletions
Download Patch File Download Diff File Expand all files Collapse all files
crates/ecstore/src/bucket/metadata.rs
+52 -1
View File
@@ -26,7 +26,8 @@ use rmp_serde::Serializer as rmpSerializer;
use rustfs_policy::policy::BucketPolicy;
use s3s::dto::{
BucketLifecycleConfiguration, CORSConfiguration, NotificationConfiguration, ObjectLockConfiguration,
ReplicationConfiguration, ServerSideEncryptionConfiguration, Tagging, VersioningConfiguration,
PublicAccessBlockConfiguration, ReplicationConfiguration, ServerSideEncryptionConfiguration, Tagging,
VersioningConfiguration,
};
use serde::Serializer;
use serde::{Deserialize, Serialize};
@@ -50,6 +51,8 @@ pub const BUCKET_VERSIONING_CONFIG: &str = "versioning.xml";
pub const BUCKET_REPLICATION_CONFIG: &str = "replication.xml";
pub const BUCKET_TARGETS_FILE: &str = "bucket-targets.json";
pub const BUCKET_CORS_CONFIG: &str = "cors.xml";
pub const BUCKET_PUBLIC_ACCESS_BLOCK_CONFIG: &str = "public-access-block.xml";
pub const BUCKET_ACL_CONFIG: &str = "bucket-acl.json";
#[derive(Debug, Deserialize, Serialize, Clone)]
#[serde(rename_all = "PascalCase", default)]
@@ -69,6 +72,8 @@ pub struct BucketMetadata {
pub bucket_targets_config_json: Vec<u8>,
pub bucket_targets_config_meta_json: Vec<u8>,
pub cors_config_xml: Vec<u8>,
pub public_access_block_config_xml: Vec<u8>,
pub bucket_acl_config_json: Vec<u8>,
pub policy_config_updated_at: OffsetDateTime,
pub object_lock_config_updated_at: OffsetDateTime,
@@ -82,6 +87,8 @@ pub struct BucketMetadata {
pub bucket_targets_config_updated_at: OffsetDateTime,
pub bucket_targets_config_meta_updated_at: OffsetDateTime,
pub cors_config_updated_at: OffsetDateTime,
pub public_access_block_config_updated_at: OffsetDateTime,
pub bucket_acl_config_updated_at: OffsetDateTime,
#[serde(skip)]
pub new_field_updated_at: OffsetDateTime,
@@ -110,6 +117,10 @@ pub struct BucketMetadata {
pub bucket_target_config_meta: Option<HashMap<String, String>>,
#[serde(skip)]
pub cors_config: Option<CORSConfiguration>,
#[serde(skip)]
pub public_access_block_config: Option<PublicAccessBlockConfiguration>,
#[serde(skip)]
pub bucket_acl_config: Option<String>,
}
impl Default for BucketMetadata {
@@ -130,6 +141,8 @@ impl Default for BucketMetadata {
bucket_targets_config_json: Default::default(),
bucket_targets_config_meta_json: Default::default(),
cors_config_xml: Default::default(),
public_access_block_config_xml: Default::default(),
bucket_acl_config_json: Default::default(),
policy_config_updated_at: OffsetDateTime::UNIX_EPOCH,
object_lock_config_updated_at: OffsetDateTime::UNIX_EPOCH,
encryption_config_updated_at: OffsetDateTime::UNIX_EPOCH,
@@ -142,6 +155,8 @@ impl Default for BucketMetadata {
bucket_targets_config_updated_at: OffsetDateTime::UNIX_EPOCH,
bucket_targets_config_meta_updated_at: OffsetDateTime::UNIX_EPOCH,
cors_config_updated_at: OffsetDateTime::UNIX_EPOCH,
public_access_block_config_updated_at: OffsetDateTime::UNIX_EPOCH,
bucket_acl_config_updated_at: OffsetDateTime::UNIX_EPOCH,
new_field_updated_at: OffsetDateTime::UNIX_EPOCH,
policy_config: Default::default(),
notification_config: Default::default(),
@@ -155,6 +170,8 @@ impl Default for BucketMetadata {
bucket_target_config: Default::default(),
bucket_target_config_meta: Default::default(),
cors_config: Default::default(),
public_access_block_config: Default::default(),
bucket_acl_config: Default::default(),
}
}
}
@@ -254,6 +271,12 @@ impl BucketMetadata {
if self.bucket_targets_config_meta_updated_at == OffsetDateTime::UNIX_EPOCH {
self.bucket_targets_config_meta_updated_at = self.created
}
if self.public_access_block_config_updated_at == OffsetDateTime::UNIX_EPOCH {
self.public_access_block_config_updated_at = self.created
}
if self.bucket_acl_config_updated_at == OffsetDateTime::UNIX_EPOCH {
self.bucket_acl_config_updated_at = self.created
}
}
pub fn update_config(&mut self, config_file: &str, data: Vec<u8>) -> Result<OffsetDateTime> {
@@ -307,6 +330,14 @@ impl BucketMetadata {
self.cors_config_xml = data;
self.cors_config_updated_at = updated;
}
BUCKET_PUBLIC_ACCESS_BLOCK_CONFIG => {
self.public_access_block_config_xml = data;
self.public_access_block_config_updated_at = updated;
}
BUCKET_ACL_CONFIG => {
self.bucket_acl_config_json = data;
self.bucket_acl_config_updated_at = updated;
}
_ => return Err(Error::other(format!("config file not found : {config_file}"))),
}
@@ -380,6 +411,15 @@ impl BucketMetadata {
if !self.cors_config_xml.is_empty() {
self.cors_config = Some(deserialize::<CORSConfiguration>(&self.cors_config_xml)?);
}
if !self.public_access_block_config_xml.is_empty() {
self.public_access_block_config =
Some(deserialize::<PublicAccessBlockConfiguration>(&self.public_access_block_config_xml)?);
}
if !self.bucket_acl_config_json.is_empty() {
let acl = String::from_utf8(self.bucket_acl_config_json.clone())
.map_err(|e| Error::other(format!("invalid UTF-8 in bucket ACL: {}", e)))?;
self.bucket_acl_config = Some(acl);
}
Ok(())
}
@@ -530,6 +570,15 @@ mod test {
bm.bucket_targets_config_meta_json = bucket_targets_meta_json.as_bytes().to_vec();
bm.bucket_targets_config_meta_updated_at = OffsetDateTime::now_utc();
// Add public access block configuration
let public_access_block_xml = r#"<PublicAccessBlockConfiguration><BlockPublicAcls>true</BlockPublicAcls><IgnorePublicAcls>true</IgnorePublicAcls><BlockPublicPolicy>true</BlockPublicPolicy><RestrictPublicBuckets>false</RestrictPublicBuckets></PublicAccessBlockConfiguration>"#;
bm.public_access_block_config_xml = public_access_block_xml.as_bytes().to_vec();
bm.public_access_block_config_updated_at = OffsetDateTime::now_utc();
let bucket_acl = r#"{"owner":{"id":"rustfsadmin","display_name":"RustFS Tester"},"grants":[{"grantee":{"grantee_type":"CanonicalUser","id":"rustfsadmin","display_name":"RustFS Tester","uri":null,"email_address":null},"permission":"FULL_CONTROL"}]}"#;
bm.bucket_acl_config_json = bucket_acl.as_bytes().to_vec();
bm.bucket_acl_config_updated_at = OffsetDateTime::now_utc();
// Test serialization
let buf = bm.marshal_msg().unwrap();
assert!(!buf.is_empty(), "Serialized buffer should not be empty");
@@ -549,6 +598,8 @@ mod test {
assert_eq!(bm.encryption_config_xml, deserialized_bm.encryption_config_xml);
assert_eq!(bm.tagging_config_xml, deserialized_bm.tagging_config_xml);
assert_eq!(bm.quota_config_json, deserialized_bm.quota_config_json);
assert_eq!(bm.public_access_block_config_xml, deserialized_bm.public_access_block_config_xml);
assert_eq!(bm.bucket_acl_config_json, deserialized_bm.bucket_acl_config_json);
assert_eq!(bm.object_lock_config_xml, deserialized_bm.object_lock_config_xml);
assert_eq!(bm.notification_config_xml, deserialized_bm.notification_config_xml);
assert_eq!(bm.replication_config_xml, deserialized_bm.replication_config_xml);
crates/ecstore/src/bucket/metadata_sys.rs
+35 -1
View File
@@ -29,7 +29,7 @@ use rustfs_policy::policy::BucketPolicy;
use s3s::dto::ReplicationConfiguration;
use s3s::dto::{
BucketLifecycleConfiguration, CORSConfiguration, NotificationConfiguration, ObjectLockConfiguration,
ServerSideEncryptionConfiguration, Tagging, VersioningConfiguration,
PublicAccessBlockConfiguration, ServerSideEncryptionConfiguration, Tagging, VersioningConfiguration,
};
use std::collections::HashSet;
use std::sync::OnceLock;
@@ -105,6 +105,13 @@ pub async fn get_bucket_policy_raw(bucket: &str) -> Result<(String, OffsetDateTi
bucket_meta_sys.get_bucket_policy_raw(bucket).await
}
pub async fn get_bucket_acl_config(bucket: &str) -> Result<(String, OffsetDateTime)> {
let bucket_meta_sys_lock = get_bucket_metadata_sys()?;
let bucket_meta_sys = bucket_meta_sys_lock.read().await;
bucket_meta_sys.get_bucket_acl_config(bucket).await
}
pub async fn get_quota_config(bucket: &str) -> Result<(BucketQuota, OffsetDateTime)> {
let bucket_meta_sys_lock = get_bucket_metadata_sys()?;
let bucket_meta_sys = bucket_meta_sys_lock.read().await;
@@ -133,6 +140,13 @@ pub async fn get_tagging_config(bucket: &str) -> Result<(Tagging, OffsetDateTime
bucket_meta_sys.get_tagging_config(bucket).await
}
pub async fn get_public_access_block_config(bucket: &str) -> Result<(PublicAccessBlockConfiguration, OffsetDateTime)> {
let bucket_meta_sys_lock = get_bucket_metadata_sys()?;
let bucket_meta_sys = bucket_meta_sys_lock.read().await;
bucket_meta_sys.get_public_access_block_config(bucket).await
}
pub async fn get_lifecycle_config(bucket: &str) -> Result<(BucketLifecycleConfiguration, OffsetDateTime)> {
let bucket_meta_sys_lock = get_bucket_metadata_sys()?;
let bucket_meta_sys = bucket_meta_sys_lock.read().await;
@@ -478,6 +492,16 @@ impl BucketMetadataSys {
}
}
pub async fn get_bucket_acl_config(&self, bucket: &str) -> Result<(String, OffsetDateTime)> {
let (bm, _) = self.get_config(bucket).await?;
if let Some(config) = &bm.bucket_acl_config {
Ok((config.clone(), bm.bucket_acl_config_updated_at))
} else {
Err(Error::ConfigNotFound)
}
}
pub async fn get_tagging_config(&self, bucket: &str) -> Result<(Tagging, OffsetDateTime)> {
let (bm, _) = self.get_config(bucket).await?;
@@ -488,6 +512,16 @@ impl BucketMetadataSys {
}
}
pub async fn get_public_access_block_config(&self, bucket: &str) -> Result<(PublicAccessBlockConfiguration, OffsetDateTime)> {
let (bm, _) = self.get_config(bucket).await?;
if let Some(config) = &bm.public_access_block_config {
Ok((config.clone(), bm.public_access_block_config_updated_at))
} else {
Err(Error::ConfigNotFound)
}
}
pub async fn get_object_lock_config(&self, bucket: &str) -> Result<(ObjectLockConfiguration, OffsetDateTime)> {
let (bm, _) = self.get_config(bucket).await?;
crates/ecstore/src/disk/local.rs
+107 -239
View File
@@ -339,9 +339,7 @@ impl LocalDisk {
#[tracing::instrument(level = "debug", skip(self))]
async fn check_format_json(&self) -> Result<Metadata> {
let md = tokio::fs::metadata(&self.format_path)
.await
.map_err(to_unformatted_disk_error)?;
let md = std::fs::metadata(&self.format_path).map_err(to_unformatted_disk_error)?;
Ok(md)
}
async fn make_meta_volumes(&self) -> Result<()> {
@@ -894,13 +892,7 @@ impl LocalDisk {
check_path_length(file_path.to_string_lossy().as_ref())?;
self.write_all_internal(&file_path, InternalBuf::Owned(buf), sync, skip_parent)
.await?;
// Invalidate file cache after successful write to ensure listing and other readers
// see the updated metadata immediately (e.g. delete markers created via delete_objects).
get_global_file_cache().invalidate(&file_path).await;
Ok(())
.await
}
// write_all_internal do write file
async fn write_all_internal(&self, file_path: &Path, data: InternalBuf<'_>, sync: bool, skip_parent: &Path) -> Result<()> {
@@ -982,7 +974,7 @@ impl LocalDisk {
opts: &WalkDirOptions,
out: &mut MetacacheWriter<W>,
objs_returned: &mut i32,
emit_current_object: bool,
skip_current_dir_object: bool,
) -> Result<()>
where
W: AsyncWrite + Unpin + Send,
@@ -1043,7 +1035,6 @@ impl LocalDisk {
let bucket = opts.bucket.as_str();
let mut dir_objes = HashSet::new();
let mut object_data_dirs = HashSet::new();
// First-level filtering
for item in entries.iter_mut() {
@@ -1080,6 +1071,10 @@ impl LocalDisk {
*item = "".to_owned();
if entry.ends_with(STORAGE_FORMAT_FILE) {
if skip_current_dir_object {
continue;
}
let metadata = self
.read_metadata(bucket, format!("{}/{}", &current, &entry).as_str())
.await?;
@@ -1088,39 +1083,19 @@ impl LocalDisk {
let name = entry.trim_end_matches(SLASH_SEPARATOR);
let name = decode_dir_object(format!("{}/{}", &current, &name).as_str());
if emit_current_object {
// if opts.limit > 0
// && let Ok(meta) = FileMeta::load(&metadata)
// && !meta.all_hidden(true)
// {
*objs_returned += 1;
// }
// if opts.limit > 0
// && let Ok(meta) = FileMeta::load(&metadata)
// && !meta.all_hidden(true)
// {
*objs_returned += 1;
// }
out.write_obj(&MetaCacheEntry {
name: name.clone(),
metadata: metadata.to_vec(),
..Default::default()
})
.await?;
}
// Keep scanning this directory so nested children like
// "foo/bar/xyzzy" are still discoverable when "foo/bar" exists.
if opts.recursive
&& let Ok(fi) = get_file_info(
&metadata,
bucket,
&name,
"",
FileInfoOpts {
data: false,
include_free_versions: false,
},
)
&& let Some(data_dir) = fi.data_dir
{
object_data_dirs.insert(data_dir.to_string());
}
out.write_obj(&MetaCacheEntry {
name: name.clone(),
metadata: metadata.to_vec(),
..Default::default()
})
.await?;
continue;
}
@@ -1137,7 +1112,7 @@ impl LocalDisk {
}
}
let mut dir_stack: Vec<String> = Vec::with_capacity(5);
let mut dir_stack: Vec<(String, bool)> = Vec::with_capacity(5);
prefix = "".to_owned();
for entry in entries.iter() {
@@ -1149,15 +1124,9 @@ impl LocalDisk {
continue;
}
// Skip object data directories. They are internal storage layout, not user objects.
if opts.recursive && object_data_dirs.contains(entry) {
continue;
}
let name = path_join_buf(&[current.as_str(), entry.as_str()]);
let object_dir_name = name.clone();
while let Some(pop) = dir_stack.last().cloned()
while let Some((pop, skip_object)) = dir_stack.last().cloned()
&& pop < name
{
out.write_obj(&MetaCacheEntry {
@@ -1167,7 +1136,7 @@ impl LocalDisk {
.await?;
if opts.recursive
&& let Err(er) = Box::pin(self.scan_dir(pop, prefix.clone(), opts, out, objs_returned, true)).await
&& let Err(er) = Box::pin(self.scan_dir(pop, prefix.clone(), opts, out, objs_returned, skip_object)).await
{
error!("scan_dir err {:?}", er);
}
@@ -1207,20 +1176,12 @@ impl LocalDisk {
*objs_returned += 1;
// }
// This object directory can also contain nested children objects.
// Recurse, but do not emit the current object again.
if opts.recursive
&& let Err(er) = Box::pin(self.scan_dir(
format!("{object_dir_name}{SLASH_SEPARATOR}"),
prefix.clone(),
opts,
out,
objs_returned,
false,
))
.await
{
warn!("scan_dir err {:?}", &er);
if opts.recursive {
let mut dir_name = meta.name.clone();
if !dir_name.ends_with(SLASH_SEPARATOR) {
dir_name.push_str(SLASH_SEPARATOR);
}
dir_stack.push((dir_name, true));
}
}
Err(err) => {
@@ -1229,7 +1190,7 @@ impl LocalDisk {
// If dirObject, but no metadata (which is unexpected) we skip it.
if !is_dir_obj && !is_empty_dir(self.get_object_path(&opts.bucket, &meta.name)?).await {
meta.name.push_str(SLASH_SEPARATOR);
dir_stack.push(meta.name);
dir_stack.push((meta.name, false));
}
}
@@ -1238,7 +1199,7 @@ impl LocalDisk {
};
}
while let Some(dir) = dir_stack.pop() {
while let Some((dir, skip_object)) = dir_stack.pop() {
if opts.limit > 0 && *objs_returned >= opts.limit {
return Ok(());
}
@@ -1250,7 +1211,7 @@ impl LocalDisk {
.await?;
if opts.recursive
&& let Err(er) = Box::pin(self.scan_dir(dir, prefix.clone(), opts, out, objs_returned, true)).await
&& let Err(er) = Box::pin(self.scan_dir(dir, prefix.clone(), opts, out, objs_returned, skip_object)).await
{
warn!("scan_dir err {:?}", &er);
}
@@ -1417,43 +1378,36 @@ impl DiskAPI for LocalDisk {
#[tracing::instrument(level = "debug", skip(self))]
async fn get_disk_id(&self) -> Result<Option<Uuid>> {
let (id, last_check, file_info) = {
let format_info = {
let format_info = self.format_info.read().await;
(format_info.id, format_info.last_check, format_info.file_info.clone())
format_info.clone()
};
// Check if we can use cached value without doing any I/O
// If we checked recently (within 1 second) and have valid cache, return immediately
if let (Some(id), Some(last_check)) = (id, last_check)
&& last_check.unix_timestamp() + 1 >= OffsetDateTime::now_utc().unix_timestamp()
{
return Ok(Some(id));
let id = format_info.id;
// if format_info.last_check_valid() {
// return Ok(id);
// }
if format_info.file_info.is_some() && id.is_some() {
// check last check time
if let Some(last_check) = format_info.last_check
&& last_check.unix_timestamp() + 1 < OffsetDateTime::now_utc().unix_timestamp()
{
return Ok(id);
}
}
// Get current file metadata (async I/O)
let file_meta = match self.check_format_json().await {
Ok(meta) => meta,
Err(e) => {
// file does not exist or cannot be accessed, clear cached format info
if matches!(e, DiskError::UnformattedDisk | DiskError::DiskNotFound) {
let mut format_info = self.format_info.write().await;
format_info.id = None;
format_info.file_info = None;
format_info.data = Bytes::new();
format_info.last_check = None;
}
return Err(e);
}
};
let file_meta = self.check_format_json().await?;
// Validate cache against current file metadata
if let (Some(cached_file_info), Some(id)) = (&file_info, id)
&& super::fs::same_file(&file_meta, cached_file_info)
if let Some(file_info) = &format_info.file_info
&& super::fs::same_file(&file_meta, file_info)
{
// Cache is still valid, update last_check and return
let mut format_info = self.format_info.write().await;
format_info.last_check = Some(OffsetDateTime::now_utc());
return Ok(Some(id));
drop(format_info);
return Ok(id);
}
debug!("get_disk_id: read format.json");
@@ -1462,7 +1416,7 @@ impl DiskAPI for LocalDisk {
let fm = FormatV3::try_from(b.as_slice()).map_err(|e| {
warn!("decode format.json err {:?}", e);
DiskError::UnformattedDisk
DiskError::CorruptedBackend
})?;
let (m, n) = fm.find_disk_index_by_disk_id(fm.erasure.this)?;
@@ -2008,7 +1962,7 @@ impl DiskAPI for LocalDisk {
&opts,
&mut out,
&mut objs_returned,
true,
false,
)
.await?;
@@ -2170,12 +2124,6 @@ impl DiskAPI for LocalDisk {
return Err(err);
}
// Invalidate cache entries for both source and destination xl.meta so that reads
// after rename_data (e.g. immediately after put_object) see the new version rather
// than stale cached data, and cannot obtain data via the old source path.
get_global_file_cache().invalidate(&src_file_path).await;
get_global_file_cache().invalidate(&dst_file_path).await;
if let Some(src_file_path_parent) = src_file_path.parent() {
if src_volume != super::RUSTFS_META_MULTIPART_BUCKET {
let _ = remove_std(src_file_path_parent);
@@ -2487,12 +2435,11 @@ impl DiskAPI for LocalDisk {
return self.write_metadata("", volume, path, fi).await;
}
let ret_err = if fi.version_id.is_some() {
DiskError::FileVersionNotFound
return if fi.version_id.is_some() {
Err(DiskError::FileVersionNotFound)
} else {
DiskError::FileNotFound
Err(DiskError::FileNotFound)
};
return Err(ret_err);
}
};
@@ -2692,7 +2639,6 @@ async fn get_disk_info(drive_path: PathBuf) -> Result<(rustfs_utils::os::DiskInf
#[cfg(test)]
mod test {
use super::*;
use rustfs_filemeta::MetacacheReader;
#[tokio::test]
async fn test_skip_access_checks() {
@@ -2712,6 +2658,56 @@ mod test {
}
}
#[tokio::test]
async fn test_scan_dir_includes_nested_object_dirs() {
use rustfs_filemeta::MetacacheReader;
use tempfile::tempdir;
let dir = tempdir().unwrap();
let bucket = "test-bucket";
let bucket_dir = dir.path().join(bucket);
fs::create_dir_all(bucket_dir.join("foo/bar/xyzzy")).await.unwrap();
fs::create_dir_all(bucket_dir.join("quux/thud")).await.unwrap();
fs::create_dir_all(bucket_dir.join("asdf")).await.unwrap();
fs::write(bucket_dir.join("foo/bar/xl.meta"), b"meta").await.unwrap();
fs::write(bucket_dir.join("foo/bar/xyzzy/xl.meta"), b"meta").await.unwrap();
fs::write(bucket_dir.join("quux/thud/xl.meta"), b"meta").await.unwrap();
fs::write(bucket_dir.join("asdf/xl.meta"), b"meta").await.unwrap();
let endpoint = Endpoint::try_from(dir.path().to_str().unwrap()).unwrap();
let disk = LocalDisk::new(&endpoint, false).await.unwrap();
let (reader, mut writer) = tokio::io::duplex(4096);
let mut out = MetacacheWriter::new(&mut writer);
let opts = WalkDirOptions {
bucket: bucket.to_string(),
base_dir: "".to_string(),
recursive: true,
..Default::default()
};
let mut objs_returned = 0;
disk.scan_dir("".to_string(), "".to_string(), &opts, &mut out, &mut objs_returned, false)
.await
.unwrap();
out.close().await.unwrap();
let mut reader = MetacacheReader::new(reader);
let entries = reader.read_all().await.unwrap();
let names: Vec<String> = entries
.into_iter()
.filter(|entry| !entry.metadata.is_empty())
.map(|entry| entry.name)
.collect();
assert!(names.contains(&"asdf".to_string()));
assert!(names.contains(&"foo/bar".to_string()));
assert!(names.contains(&"foo/bar/xyzzy".to_string()));
assert!(names.contains(&"quux/thud".to_string()));
}
#[tokio::test]
async fn test_make_volume() {
let p = "./testv0";
@@ -2992,134 +2988,6 @@ mod test {
let _ = fs::remove_file(test_file).await;
}
#[tokio::test]
async fn test_scan_dir_lists_nested_child_when_parent_has_xlmeta() {
let test_dir = format!("./test_scan_dir_nested_{}", uuid::Uuid::new_v4());
fs::create_dir_all(&test_dir).await.unwrap();
let endpoint = Endpoint::try_from(test_dir.as_str()).unwrap();
let disk = LocalDisk::new(&endpoint, false).await.unwrap();
let bucket = "test-volume";
disk.make_volume(bucket).await.unwrap();
// Parent object metadata: foo/bar
let parent_meta = disk.get_object_path(bucket, "foo/bar/xl.meta").unwrap();
if let Some(parent) = parent_meta.parent() {
fs::create_dir_all(parent).await.unwrap();
}
fs::write(&parent_meta, b"parent-object").await.unwrap();
// Child object metadata: foo/bar/xyzzy
let child_meta = disk.get_object_path(bucket, "foo/bar/xyzzy/xl.meta").unwrap();
if let Some(parent) = child_meta.parent() {
fs::create_dir_all(parent).await.unwrap();
}
fs::write(&child_meta, b"child-object").await.unwrap();
let (rd, mut wr) = tokio::io::duplex(16 * 1024);
let mut out = MetacacheWriter::new(&mut wr);
let mut objs_returned = 0;
let opts = WalkDirOptions {
bucket: bucket.to_string(),
base_dir: "foo/bar".to_string(),
recursive: true,
..Default::default()
};
disk.scan_dir("foo/bar".to_string(), String::new(), &opts, &mut out, &mut objs_returned, true)
.await
.unwrap();
out.close().await.unwrap();
let mut reader = MetacacheReader::new(rd);
let entries = reader.read_all().await.unwrap();
let object_names = entries
.iter()
.filter(|entry| !entry.metadata.is_empty())
.map(|entry| entry.name.clone())
.collect::<Vec<_>>();
assert!(
object_names.iter().any(|name| name == "foo/bar" || name == "foo/bar/"),
"expected parent object in scan result, got: {:?}",
object_names
);
assert!(
object_names
.iter()
.any(|name| name == "foo/bar/xyzzy" || name == "foo/bar/xyzzy/"),
"expected nested child object in scan result, got: {:?}",
object_names
);
let _ = fs::remove_dir_all(&test_dir).await;
}
#[tokio::test]
async fn test_scan_dir_lists_parent_and_child_from_bucket_root() {
let test_dir = format!("./test_scan_dir_root_nested_{}", uuid::Uuid::new_v4());
fs::create_dir_all(&test_dir).await.unwrap();
let endpoint = Endpoint::try_from(test_dir.as_str()).unwrap();
let disk = LocalDisk::new(&endpoint, false).await.unwrap();
let bucket = "test-volume";
disk.make_volume(bucket).await.unwrap();
// Parent object metadata: foo/bar
let parent_meta = disk.get_object_path(bucket, "foo/bar/xl.meta").unwrap();
if let Some(parent) = parent_meta.parent() {
fs::create_dir_all(parent).await.unwrap();
}
fs::write(&parent_meta, b"parent-object").await.unwrap();
// Child object metadata: foo/bar/xyzzy
let child_meta = disk.get_object_path(bucket, "foo/bar/xyzzy/xl.meta").unwrap();
if let Some(parent) = child_meta.parent() {
fs::create_dir_all(parent).await.unwrap();
}
fs::write(&child_meta, b"child-object").await.unwrap();
let (rd, mut wr) = tokio::io::duplex(16 * 1024);
let mut out = MetacacheWriter::new(&mut wr);
let mut objs_returned = 0;
let opts = WalkDirOptions {
bucket: bucket.to_string(),
base_dir: String::new(),
recursive: true,
..Default::default()
};
disk.scan_dir(String::new(), String::new(), &opts, &mut out, &mut objs_returned, true)
.await
.unwrap();
out.close().await.unwrap();
let mut reader = MetacacheReader::new(rd);
let entries = reader.read_all().await.unwrap();
let object_names = entries
.iter()
.filter(|entry| !entry.metadata.is_empty())
.map(|entry| entry.name.trim_start_matches('/').trim_end_matches('/').to_string())
.collect::<Vec<_>>();
assert!(
object_names.iter().any(|name| name == "foo/bar"),
"expected parent object in root scan result, got: {:?}",
object_names
);
assert!(
object_names.iter().any(|name| name == "foo/bar/xyzzy"),
"expected child object in root scan result, got: {:?}",
object_names
);
let _ = fs::remove_dir_all(&test_dir).await;
}
#[test]
fn test_is_root_path() {
// Unix root path
crates/policy/src/policy/action.rs
+14
View File
@@ -218,6 +218,8 @@ pub enum S3Action {
ForceDeleteBucketAction,
#[strum(serialize = "s3:DeleteBucketPolicy")]
DeleteBucketPolicyAction,
#[strum(serialize = "s3:DeleteBucketPublicAccessBlock")]
DeleteBucketPublicAccessBlockAction,
#[strum(serialize = "s3:DeleteBucketCors")]
DeleteBucketCorsAction,
#[strum(serialize = "s3:DeleteObject")]
@@ -228,10 +230,18 @@ pub enum S3Action {
GetBucketNotificationAction,
#[strum(serialize = "s3:GetBucketPolicy")]
GetBucketPolicyAction,
#[strum(serialize = "s3:GetBucketPublicAccessBlock")]
GetBucketPublicAccessBlockAction,
#[strum(serialize = "s3:GetBucketCors")]
GetBucketCorsAction,
#[strum(serialize = "s3:GetBucketAcl")]
GetBucketAclAction,
#[strum(serialize = "s3:PutBucketAcl")]
PutBucketAclAction,
#[strum(serialize = "s3:GetObject")]
GetObjectAction,
#[strum(serialize = "s3:GetObjectAcl")]
GetObjectAclAction,
#[strum(serialize = "s3:GetObjectAttributes")]
GetObjectAttributesAction,
#[strum(serialize = "s3:HeadBucket")]
@@ -260,10 +270,14 @@ pub enum S3Action {
PutBucketNotificationAction,
#[strum(serialize = "s3:PutBucketPolicy")]
PutBucketPolicyAction,
#[strum(serialize = "s3:PutBucketPublicAccessBlock")]
PutBucketPublicAccessBlockAction,
#[strum(serialize = "s3:PutBucketCors")]
PutBucketCorsAction,
#[strum(serialize = "s3:PutObject")]
PutObjectAction,
#[strum(serialize = "s3:PutObjectAcl")]
PutObjectAclAction,
#[strum(serialize = "s3:DeleteObjectVersion")]
DeleteObjectVersionAction,
#[strum(serialize = "s3:DeleteObjectVersionTagging")]
crates/policy/src/policy/function/key_name.rs
+3
View File
@@ -160,6 +160,9 @@ pub enum S3KeyName {
#[strum(serialize = "s3:x-amz-content-sha256")]
S3XAmzContentSha256,
#[strum(serialize = "s3:x-amz-acl")]
S3XAmzAcl,
#[strum(serialize = "s3:LocationConstraint")]
S3LocationConstraint,
crates/protocols/src/common/gateway.rs
+4 -4
View File
@@ -78,10 +78,10 @@ impl From<S3Action> for PolicyS3Action {
S3Action::AbortMultipartUpload => PolicyS3Action::AbortMultipartUploadAction,
S3Action::ListMultipartUploads => PolicyS3Action::ListBucketMultipartUploadsAction,
S3Action::ListParts => PolicyS3Action::ListMultipartUploadPartsAction,
S3Action::GetBucketAcl => PolicyS3Action::GetBucketPolicyAction,
S3Action::PutBucketAcl => PolicyS3Action::PutBucketPolicyAction,
S3Action::GetObjectAcl => PolicyS3Action::GetObjectAction,
S3Action::PutObjectAcl => PolicyS3Action::PutObjectAction,
S3Action::GetBucketAcl => PolicyS3Action::GetBucketAclAction,
S3Action::PutBucketAcl => PolicyS3Action::PutBucketAclAction,
S3Action::GetObjectAcl => PolicyS3Action::GetObjectAclAction,
S3Action::PutObjectAcl => PolicyS3Action::PutObjectAclAction,
S3Action::CopyObject => PolicyS3Action::PutObjectAction,
}
}
rustfs/src/app/mod.rs
+16
View File
@@ -0,0 +1,16 @@
// 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.
//! Application layer module entry.
//! Concrete use-case modules will be introduced incrementally in Phase 3.
rustfs/src/main.rs
+1
View File
@@ -13,6 +13,7 @@
// limitations under the License.
mod admin;
mod app;
mod auth;
mod config;
mod error;
rustfs/src/server/layer.rs
+4 -1
View File
@@ -228,7 +228,6 @@ where
if ConditionalCorsLayer::is_s3_path(&path)
&& !bucket.is_empty()
&& cors_origins.is_some()
&& let Some(cors_headers) = apply_cors_headers(&bucket, &method_clone, &request_headers_clone).await
{
for (key, value) in cors_headers.iter() {
@@ -246,6 +245,10 @@ where
});
}
if method == Method::OPTIONS && ConditionalCorsLayer::is_s3_path(&path) {
return Box::pin(async move { Ok(Response::builder().status(StatusCode::OK).body(ResBody::default()).unwrap()) });
}
let mut inner = self.inner.clone();
Box::pin(async move {
let mut response = inner.call(req).await.map_err(Into::into)?;
rustfs/src/storage/access.rs
+246 -15
View File
@@ -13,10 +13,21 @@
// limitations under the License.
use super::ecfs::FS;
use super::ecfs::{
ACL_GROUP_ALL_USERS, ACL_GROUP_AUTHENTICATED_USERS, INTERNAL_ACL_METADATA_KEY, StoredAcl, default_owner,
parse_acl_json_or_canned_bucket, parse_acl_json_or_canned_object, stored_acl_from_canned_bucket,
stored_acl_from_canned_object,
};
use super::options::get_opts;
use crate::auth::{check_key_valid, get_condition_values, get_session_token};
use crate::error::ApiError;
use crate::license::license_check;
use crate::server::RemoteAddr;
use rustfs_ecstore::StorageAPI;
use rustfs_ecstore::bucket::metadata_sys;
use rustfs_ecstore::bucket::policy_sys::PolicySys;
use rustfs_ecstore::error::StorageError;
use rustfs_ecstore::new_object_layer_fn;
use rustfs_iam::error::Error as IamError;
use rustfs_policy::policy::action::{Action, S3Action};
use rustfs_policy::policy::{Args, BucketPolicyArgs};
@@ -36,11 +47,164 @@ pub(crate) struct ReqInfo {
pub region: Option<String>,
}
#[derive(Clone, Copy)]
enum AclTarget {
Bucket,
Object,
}
#[derive(Clone, Copy)]
enum AclPermission {
Read,
Write,
ReadAcp,
WriteAcp,
}
fn acl_permission_for_action(action: &Action) -> Option<(AclTarget, AclPermission)> {
match action {
Action::S3Action(S3Action::ListBucketAction) => Some((AclTarget::Bucket, AclPermission::Read)),
Action::S3Action(S3Action::PutObjectAction) => Some((AclTarget::Bucket, AclPermission::Write)),
Action::S3Action(S3Action::GetBucketAclAction) => Some((AclTarget::Bucket, AclPermission::ReadAcp)),
Action::S3Action(S3Action::PutBucketAclAction) => Some((AclTarget::Bucket, AclPermission::WriteAcp)),
Action::S3Action(S3Action::GetObjectAction) => Some((AclTarget::Object, AclPermission::Read)),
Action::S3Action(S3Action::GetObjectAclAction) => Some((AclTarget::Object, AclPermission::ReadAcp)),
Action::S3Action(S3Action::PutObjectAclAction) => Some((AclTarget::Object, AclPermission::WriteAcp)),
_ => None,
}
}
fn permission_matches(grant_perm: &str, required: AclPermission) -> bool {
if grant_perm == Permission::FULL_CONTROL {
return true;
}
match required {
AclPermission::Read => grant_perm == Permission::READ,
AclPermission::Write => grant_perm == Permission::WRITE,
AclPermission::ReadAcp => grant_perm == Permission::READ_ACP,
AclPermission::WriteAcp => grant_perm == Permission::WRITE_ACP,
}
}
fn acl_allows(
acl: &StoredAcl,
user_id: Option<&str>,
is_authenticated: bool,
required: AclPermission,
ignore_public_acls: bool,
) -> bool {
for grant in &acl.grants {
if !permission_matches(grant.permission.as_str(), required) {
continue;
}
match grant.grantee.grantee_type.as_str() {
"CanonicalUser" => {
if user_id.is_some_and(|id| grant.grantee.id.as_deref() == Some(id)) {
return true;
}
}
"Group" => {
if ignore_public_acls {
continue;
}
if let Some(uri) = grant.grantee.uri.as_deref() {
if uri == ACL_GROUP_ALL_USERS {
return true;
}
if uri == ACL_GROUP_AUTHENTICATED_USERS && is_authenticated {
return true;
}
}
}
_ => {}
}
}
false
}
async fn load_bucket_acl(bucket: &str) -> S3Result<StoredAcl> {
let owner = default_owner();
match metadata_sys::get_bucket_acl_config(bucket).await {
Ok((acl, _)) => Ok(parse_acl_json_or_canned_bucket(&acl, &owner)),
Err(err) => {
if err == StorageError::ConfigNotFound {
Ok(stored_acl_from_canned_bucket(BucketCannedACL::PRIVATE, &owner))
} else {
Err(S3Error::with_message(S3ErrorCode::InternalError, err.to_string()))
}
}
}
}
async fn load_object_acl(bucket: &str, object: &str, version_id: Option<&str>, headers: &http::HeaderMap) -> S3Result<StoredAcl> {
let Some(store) = new_object_layer_fn() else {
return Err(S3Error::with_message(S3ErrorCode::InternalError, "Not init".to_string()));
};
let opts = get_opts(bucket, object, version_id.map(|v| v.to_string()), None, headers)
.await
.map_err(ApiError::from)?;
let info = store.get_object_info(bucket, object, &opts).await.map_err(ApiError::from)?;
let bucket_owner = default_owner();
let object_owner = info
.user_defined
.get(INTERNAL_ACL_METADATA_KEY)
.and_then(|acl| serde_json::from_str::<StoredAcl>(acl).ok())
.map(|acl| acl.owner)
.unwrap_or_else(default_owner);
Ok(info
.user_defined
.get(INTERNAL_ACL_METADATA_KEY)
.map(|acl| parse_acl_json_or_canned_object(acl, &bucket_owner, &object_owner))
.unwrap_or_else(|| stored_acl_from_canned_object(ObjectCannedACL::PRIVATE, &bucket_owner, &object_owner)))
}
async fn check_acl_access<T>(req: &S3Request<T>, req_info: &ReqInfo, action: &Action, policy_allowed: bool) -> S3Result<bool> {
if req_info.is_owner || policy_allowed {
return Ok(true);
}
let Some((target, permission)) = acl_permission_for_action(action) else {
return Ok(true);
};
let bucket = req_info.bucket.as_deref().unwrap_or("");
if bucket.is_empty() {
return Ok(true);
}
let ignore_public_acls = match metadata_sys::get_public_access_block_config(bucket).await {
Ok((config, _)) => config.ignore_public_acls.unwrap_or(false),
Err(_) => false,
};
let user_id = req_info.cred.as_ref().map(|cred| cred.access_key.as_str());
let is_authenticated = user_id.is_some();
let acl = match target {
AclTarget::Bucket => load_bucket_acl(bucket).await?,
AclTarget::Object => {
let object = req_info.object.as_deref().unwrap_or("");
if object.is_empty() {
return Ok(true);
}
load_object_acl(bucket, object, req_info.version_id.as_deref(), &req.headers).await?
}
};
Ok(acl_allows(&acl, user_id, is_authenticated, permission, ignore_public_acls))
}
/// Authorizes the request based on the action and credentials.
pub async fn authorize_request<T>(req: &mut S3Request<T>, action: Action) -> S3Result<()> {
let remote_addr = req.extensions.get::<Option<RemoteAddr>>().and_then(|opt| opt.map(|a| a.0));
let req_info = req.extensions.get_mut::<ReqInfo>().expect("ReqInfo not found");
let req_info = req.extensions.get::<ReqInfo>().expect("ReqInfo not found");
if let Some(cred) = &req_info.cred {
let Ok(iam_store) = rustfs_iam::get() else {
@@ -53,6 +217,23 @@ pub async fn authorize_request<T>(req: &mut S3Request<T>, action: Action) -> S3R
let default_claims = HashMap::new();
let claims = cred.claims.as_ref().unwrap_or(&default_claims);
let conditions = get_condition_values(&req.headers, cred, req_info.version_id.as_deref(), None, remote_addr);
let bucket_name = req_info.bucket.as_deref().unwrap_or("");
if !bucket_name.is_empty()
&& !PolicySys::is_allowed(&BucketPolicyArgs {
bucket: bucket_name,
action,
// Run this early check in deny-only mode so ACL/IAM fallbacks can still grant access.
is_owner: true,
account: &cred.access_key,
groups: &cred.groups,
conditions: &conditions,
object: req_info.object.as_deref().unwrap_or(""),
})
.await
{
return Err(s3_error!(AccessDenied, "Access Denied"));
}
if action == Action::S3Action(S3Action::DeleteObjectAction)
&& req_info.version_id.is_some()
@@ -83,7 +264,7 @@ pub async fn authorize_request<T>(req: &mut S3Request<T>, action: Action) -> S3R
return Err(s3_error!(AccessDenied, "Access Denied"));
}
if iam_store
let iam_allowed = iam_store
.is_allowed(&Args {
account: &cred.access_key,
groups: &cred.groups,
@@ -95,9 +276,29 @@ pub async fn authorize_request<T>(req: &mut S3Request<T>, action: Action) -> S3R
claims,
deny_only: false,
})
.await
{
return Ok(());
.await;
if iam_allowed {
let policy_allowed = if !bucket_name.is_empty() {
PolicySys::is_allowed(&BucketPolicyArgs {
bucket: bucket_name,
action,
is_owner: false,
account: &cred.access_key,
groups: &cred.groups,
conditions: &conditions,
object: req_info.object.as_deref().unwrap_or(""),
})
.await
} else {
false
};
if check_acl_access(req, req_info, &action, policy_allowed).await? {
return Ok(());
}
return Err(s3_error!(AccessDenied, "Access Denied"));
}
if PolicySys::is_allowed(&BucketPolicyArgs {
@@ -154,6 +355,23 @@ pub async fn authorize_request<T>(req: &mut S3Request<T>, action: Action) -> S3R
req.region.as_deref(),
remote_addr,
);
let bucket_name = req_info.bucket.as_deref().unwrap_or("");
if !bucket_name.is_empty()
&& !PolicySys::is_allowed(&BucketPolicyArgs {
bucket: bucket_name,
action,
// Run this early check in deny-only mode so ACL checks are not bypassed.
is_owner: true,
account: "",
groups: &None,
conditions: &conditions,
object: req_info.object.as_deref().unwrap_or(""),
})
.await
{
return Err(s3_error!(AccessDenied, "Access Denied"));
}
if action != Action::S3Action(S3Action::ListAllMyBucketsAction) {
if PolicySys::is_allowed(&BucketPolicyArgs {
@@ -184,6 +402,10 @@ pub async fn authorize_request<T>(req: &mut S3Request<T>, action: Action) -> S3R
{
return Ok(());
}
if acl_permission_for_action(&action).is_some() && check_acl_access(req, req_info, &action, false).await? {
return Ok(());
}
}
}
@@ -502,8 +724,11 @@ impl S3Access for FS {
/// Checks whether the DeletePublicAccessBlock request has accesses to the resources.
///
/// This method returns `Ok(())` by default.
async fn delete_public_access_block(&self, _req: &mut S3Request<DeletePublicAccessBlockInput>) -> S3Result<()> {
Ok(())
async fn delete_public_access_block(&self, req: &mut S3Request<DeletePublicAccessBlockInput>) -> S3Result<()> {
let req_info = req.extensions.get_mut::<ReqInfo>().expect("ReqInfo not found");
req_info.bucket = Some(req.input.bucket.clone());
authorize_request(req, Action::S3Action(S3Action::DeleteBucketPublicAccessBlockAction)).await
}
/// Checks whether the GetBucketAccelerateConfiguration request has accesses to the resources.
@@ -523,7 +748,7 @@ impl S3Access for FS {
let req_info = req.extensions.get_mut::<ReqInfo>().expect("ReqInfo not found");
req_info.bucket = Some(req.input.bucket.clone());
authorize_request(req, Action::S3Action(S3Action::GetBucketPolicyAction)).await
authorize_request(req, Action::S3Action(S3Action::GetBucketAclAction)).await
}
/// Checks whether the GetBucketAnalyticsConfiguration request has accesses to the resources.
@@ -640,7 +865,7 @@ impl S3Access for FS {
let req_info = req.extensions.get_mut::<ReqInfo>().expect("ReqInfo not found");
req_info.bucket = Some(req.input.bucket.clone());
authorize_request(req, Action::S3Action(S3Action::GetBucketPolicyAction)).await
authorize_request(req, Action::S3Action(S3Action::GetObjectAclAction)).await
}
/// Checks whether the GetBucketPolicyStatus request has accesses to the resources.
@@ -797,8 +1022,11 @@ impl S3Access for FS {
/// Checks whether the GetPublicAccessBlock request has accesses to the resources.
///
/// This method returns `Ok(())` by default.
async fn get_public_access_block(&self, _req: &mut S3Request<GetPublicAccessBlockInput>) -> S3Result<()> {
Ok(())
async fn get_public_access_block(&self, req: &mut S3Request<GetPublicAccessBlockInput>) -> S3Result<()> {
let req_info = req.extensions.get_mut::<ReqInfo>().expect("ReqInfo not found");
req_info.bucket = Some(req.input.bucket.clone());
authorize_request(req, Action::S3Action(S3Action::GetBucketPublicAccessBlockAction)).await
}
/// Checks whether the HeadBucket request has accesses to the resources.
@@ -935,7 +1163,7 @@ impl S3Access for FS {
let req_info = req.extensions.get_mut::<ReqInfo>().expect("ReqInfo not found");
req_info.bucket = Some(req.input.bucket.clone());
authorize_request(req, Action::S3Action(S3Action::PutBucketPolicyAction)).await
authorize_request(req, Action::S3Action(S3Action::PutBucketAclAction)).await
}
/// Checks whether the PutBucketAnalyticsConfiguration request has accesses to the resources.
@@ -1042,7 +1270,7 @@ impl S3Access for FS {
let req_info = req.extensions.get_mut::<ReqInfo>().expect("ReqInfo not found");
req_info.bucket = Some(req.input.bucket.clone());
authorize_request(req, Action::S3Action(S3Action::PutBucketPolicyAction)).await
authorize_request(req, Action::S3Action(S3Action::PutObjectAclAction)).await
}
/// Checks whether the PutBucketReplication request has accesses to the resources.
@@ -1169,8 +1397,11 @@ impl S3Access for FS {
/// Checks whether the PutPublicAccessBlock request has accesses to the resources.
///
/// This method returns `Ok(())` by default.
async fn put_public_access_block(&self, _req: &mut S3Request<PutPublicAccessBlockInput>) -> S3Result<()> {
Ok(())
async fn put_public_access_block(&self, req: &mut S3Request<PutPublicAccessBlockInput>) -> S3Result<()> {
let req_info = req.extensions.get_mut::<ReqInfo>().expect("ReqInfo not found");
req_info.bucket = Some(req.input.bucket.clone());
authorize_request(req, Action::S3Action(S3Action::PutBucketPublicAccessBlockAction)).await
}
/// Checks whether the RestoreObject request has accesses to the resources.
rustfs/src/storage/ecfs.rs
+2314 -614
View File
File diff suppressed because it is too large Load Diff
rustfs/src/storage/ecfs_extend.rs
+169 -2
View File
@@ -17,7 +17,7 @@ use crate::config::workload_profiles::{
};
use crate::error::ApiError;
use crate::server::cors;
use crate::storage::ecfs::ListObjectUnorderedQuery;
use crate::storage::ecfs::{InMemoryAsyncReader, ListObjectUnorderedQuery};
use http::{HeaderMap, HeaderValue, StatusCode};
use metrics::counter;
use rustfs_ecstore::bucket::metadata_sys;
@@ -27,6 +27,9 @@ use rustfs_ecstore::bucket::replication::ReplicationConfigurationExt;
use rustfs_ecstore::error::StorageError;
use rustfs_ecstore::store_api::{BucketOptions, ObjectInfo, ObjectToDelete};
use rustfs_ecstore::{StorageAPI, new_object_layer_fn};
use rustfs_filemeta::ObjectPartInfo;
use rustfs_kms::{EncryptionMetadata, ObjectEncryptionContext, get_global_encryption_service};
use rustfs_rio::{DecryptReader, Reader, WarpReader};
use rustfs_targets::EventName;
use rustfs_targets::arn::{TargetID, TargetIDError};
use rustfs_utils::http::{
@@ -36,7 +39,7 @@ use rustfs_utils::http::{
use s3s::dto::{
Delimiter, LambdaFunctionConfiguration, NotificationConfigurationFilter, ObjectLockConfiguration, ObjectLockEnabled,
ObjectLockLegalHold, ObjectLockLegalHoldStatus, ObjectLockRetention, ObjectLockRetentionMode, QueueConfiguration,
TopicConfiguration,
ServerSideEncryption, TopicConfiguration,
};
use s3s::{S3Error, S3ErrorCode, S3Response, S3Result};
use serde_urlencoded::from_bytes;
@@ -46,6 +49,7 @@ use std::sync::Arc;
use time::OffsetDateTime;
use time::format_description::well_known::Rfc3339;
use time::{format_description::FormatItem, macros::format_description};
use tokio::io::AsyncRead;
use tracing::{debug, warn};
pub const RFC1123: &[FormatItem<'_>] =
@@ -63,6 +67,7 @@ pub const RFC1123: &[FormatItem<'_>] =
/// # Arguments
/// * `object_lock_config` - Optional bucket Object Lock configuration. If None, no retention is applied.
/// * `metadata` - Mutable reference to object metadata HashMap. Retention headers are inserted here.
#[allow(dead_code)]
pub(crate) fn apply_lock_retention(object_lock_config: Option<ObjectLockConfiguration>, metadata: &mut HashMap<String, String>) {
if metadata.contains_key(AMZ_OBJECT_LOCK_MODE_LOWER) || metadata.contains_key(AMZ_OBJECT_LOCK_RETAIN_UNTIL_DATE_LOWER) {
return;
@@ -187,6 +192,168 @@ pub(crate) fn get_buffer_size_opt_in(file_size: i64) -> usize {
buffer_size
}
pub(crate) async fn create_managed_encryption_material(
bucket: &str,
key: &str,
algorithm: &ServerSideEncryption,
kms_key_id: Option<String>,
original_size: i64,
) -> Result<crate::storage::ecfs::ManagedEncryptionMaterial, ApiError> {
let Some(service) = get_global_encryption_service().await else {
return Err(ApiError::from(StorageError::other("KMS encryption service is not initialized")));
};
if !is_managed_sse(algorithm) {
return Err(ApiError::from(StorageError::other(format!(
"Unsupported server-side encryption algorithm: {}",
algorithm.as_str()
))));
}
let algorithm_str = algorithm.as_str();
let mut context = ObjectEncryptionContext::new(bucket.to_string(), key.to_string());
if original_size >= 0 {
context = context.with_size(original_size as u64);
}
let mut kms_key_candidate = kms_key_id;
if kms_key_candidate.is_none() {
kms_key_candidate = service.get_default_key_id().cloned();
}
let kms_key_to_use = kms_key_candidate
.clone()
.ok_or_else(|| ApiError::from(StorageError::other("No KMS key available for managed server-side encryption")))?;
let (data_key, encrypted_data_key) = service
.create_data_key(&kms_key_candidate, &context)
.await
.map_err(|e| ApiError::from(StorageError::other(format!("Failed to create data key: {e}"))))?;
let metadata = EncryptionMetadata {
algorithm: algorithm_str.to_string(),
key_id: kms_key_to_use.clone(),
key_version: 1,
iv: data_key.nonce.to_vec(),
tag: None,
encryption_context: context.encryption_context.clone(),
encrypted_at: jiff::Zoned::now(),
original_size: if original_size >= 0 { original_size as u64 } else { 0 },
encrypted_data_key,
};
let mut headers = service.metadata_to_headers(&metadata);
headers.insert("x-rustfs-encryption-original-size".to_string(), metadata.original_size.to_string());
Ok(crate::storage::ecfs::ManagedEncryptionMaterial {
data_key,
headers,
kms_key_id: kms_key_to_use,
})
}
pub(crate) async fn decrypt_managed_encryption_key(
bucket: &str,
key: &str,
metadata: &HashMap<String, String>,
) -> Result<Option<([u8; 32], [u8; 12], Option<i64>)>, ApiError> {
if !metadata.contains_key("x-rustfs-encryption-key") {
return Ok(None);
}
let Some(service) = get_global_encryption_service().await else {
return Err(ApiError::from(StorageError::other("KMS encryption service is not initialized")));
};
let parsed = service
.headers_to_metadata(metadata)
.map_err(|e| ApiError::from(StorageError::other(format!("Failed to parse encryption metadata: {e}"))))?;
if parsed.iv.len() != 12 {
return Err(ApiError::from(StorageError::other("Invalid encryption nonce length; expected 12 bytes")));
}
let context = ObjectEncryptionContext::new(bucket.to_string(), key.to_string());
let data_key = service
.decrypt_data_key(&parsed.encrypted_data_key, &context)
.await
.map_err(|e| ApiError::from(StorageError::other(format!("Failed to decrypt data key: {e}"))))?;
let key_bytes = data_key.plaintext_key;
let mut nonce = [0u8; 12];
nonce.copy_from_slice(&parsed.iv[..12]);
let original_size = metadata
.get("x-rustfs-encryption-original-size")
.and_then(|s| s.parse::<i64>().ok());
Ok(Some((key_bytes, nonce, original_size)))
}
pub(crate) fn derive_part_nonce(base: [u8; 12], part_number: usize) -> [u8; 12] {
let mut nonce = base;
let current = u32::from_be_bytes([nonce[8], nonce[9], nonce[10], nonce[11]]);
let incremented = current.wrapping_add(part_number as u32);
nonce[8..12].copy_from_slice(&incremented.to_be_bytes());
nonce
}
pub(crate) async fn decrypt_multipart_managed_stream(
mut encrypted_stream: Box<dyn AsyncRead + Unpin + Send + Sync>,
parts: &[ObjectPartInfo],
key_bytes: [u8; 32],
base_nonce: [u8; 12],
) -> Result<(Box<dyn Reader>, i64), StorageError> {
let total_plain_capacity: usize = parts.iter().map(|part| part.actual_size.max(0) as usize).sum();
let mut plaintext = Vec::with_capacity(total_plain_capacity);
for part in parts {
if part.size == 0 {
continue;
}
let mut encrypted_part = vec![0u8; part.size];
tokio::io::AsyncReadExt::read_exact(&mut encrypted_stream, &mut encrypted_part)
.await
.map_err(|e| StorageError::other(format!("failed to read encrypted multipart segment {}: {}", part.number, e)))?;
let part_nonce = derive_part_nonce(base_nonce, part.number);
let cursor = std::io::Cursor::new(encrypted_part);
let mut decrypt_reader = DecryptReader::new(WarpReader::new(cursor), key_bytes, part_nonce);
tokio::io::AsyncReadExt::read_to_end(&mut decrypt_reader, &mut plaintext)
.await
.map_err(|e| StorageError::other(format!("failed to decrypt multipart segment {}: {}", part.number, e)))?;
}
let total_plain_size = plaintext.len() as i64;
let reader = Box::new(WarpReader::new(InMemoryAsyncReader::new(plaintext))) as Box<dyn Reader>;
Ok((reader, total_plain_size))
}
pub(crate) fn strip_managed_encryption_metadata(metadata: &mut HashMap<String, String>) {
const KEYS: [&str; 7] = [
"x-amz-server-side-encryption",
"x-amz-server-side-encryption-aws-kms-key-id",
"x-rustfs-encryption-iv",
"x-rustfs-encryption-tag",
"x-rustfs-encryption-key",
"x-rustfs-encryption-context",
"x-rustfs-encryption-original-size",
];
for key in KEYS.iter() {
metadata.remove(*key);
}
}
pub(crate) fn is_managed_sse(algorithm: &ServerSideEncryption) -> bool {
matches!(algorithm.as_str(), "AES256" | "aws:kms")
}
/// Validate object key for control characters and log special characters
///
/// This function:
rustfs/src/storage/ecfs_test.rs
+31 -157
View File
@@ -16,7 +16,7 @@
mod tests {
use crate::config::workload_profiles::WorkloadProfile;
use crate::storage::ecfs::FS;
use crate::storage::s3_api::common::rustfs_owner;
use crate::storage::ecfs::RUSTFS_OWNER;
use crate::storage::{
apply_cors_headers, check_preconditions, get_adaptive_buffer_size_with_profile, get_buffer_size_opt_in, is_etag_equal,
matches_origin_pattern, parse_etag, parse_object_lock_legal_hold, parse_object_lock_retention,
@@ -27,7 +27,6 @@ mod tests {
use rustfs_config::MI_B;
use rustfs_ecstore::set_disk::DEFAULT_READ_BUFFER_SIZE;
use rustfs_ecstore::store_api::ObjectInfo;
use rustfs_policy::policy::{BucketPolicy, Validator};
use rustfs_utils::http::{AMZ_OBJECT_LOCK_LEGAL_HOLD_LOWER, RESERVED_METADATA_PREFIX_LOWER};
use rustfs_zip::CompressionFormat;
use s3s::dto::{
@@ -67,12 +66,11 @@ mod tests {
}
#[test]
fn test_rustfs_owner_helper() {
// Test that rustfs owner metadata remains stable for S3 compatibility.
let owner = rustfs_owner();
assert!(!owner.display_name.as_ref().unwrap().is_empty());
assert!(!owner.id.as_ref().unwrap().is_empty());
assert_eq!(owner.display_name.as_ref().unwrap(), "rustfs");
fn test_rustfs_owner_constant() {
// Test that RUSTFS_OWNER constant is properly defined
assert!(!RUSTFS_OWNER.display_name.as_ref().unwrap().is_empty());
assert!(!RUSTFS_OWNER.id.as_ref().unwrap().is_empty());
assert_eq!(RUSTFS_OWNER.display_name.as_ref().unwrap(), "RustFS Tester");
}
// Note: Most S3 API methods require complex setup with global state, storage backend,
@@ -800,127 +798,6 @@ mod tests {
assert_eq!(result13.unwrap_err().code(), &S3ErrorCode::InvalidArgument);
}
#[test]
fn test_apply_lock_retention() {
use crate::storage::ecfs_extend::apply_lock_retention;
use s3s::dto::{DefaultRetention, ObjectLockConfiguration, ObjectLockEnabled, ObjectLockRetentionMode, ObjectLockRule};
use std::collections::HashMap;
// [1] Normal case: Apply default retention with COMPLIANCE mode and days
let mut metadata = HashMap::new();
let config = Some(ObjectLockConfiguration {
object_lock_enabled: Some(ObjectLockEnabled::from_static(ObjectLockEnabled::ENABLED)),
rule: Some(ObjectLockRule {
default_retention: Some(DefaultRetention {
mode: Some(ObjectLockRetentionMode::from_static(ObjectLockRetentionMode::COMPLIANCE)),
days: Some(30),
years: None,
}),
}),
});
apply_lock_retention(config, &mut metadata);
assert_eq!(metadata.get("x-amz-object-lock-mode"), Some(&"COMPLIANCE".to_string()));
assert!(metadata.contains_key("x-amz-object-lock-retain-until-date"));
// [2] Normal case: Apply default retention with GOVERNANCE mode and years
let mut metadata = HashMap::new();
let config = Some(ObjectLockConfiguration {
object_lock_enabled: Some(ObjectLockEnabled::from_static(ObjectLockEnabled::ENABLED)),
rule: Some(ObjectLockRule {
default_retention: Some(DefaultRetention {
mode: Some(ObjectLockRetentionMode::from_static(ObjectLockRetentionMode::GOVERNANCE)),
days: None,
years: Some(1),
}),
}),
});
apply_lock_retention(config, &mut metadata);
assert_eq!(metadata.get("x-amz-object-lock-mode"), Some(&"GOVERNANCE".to_string()));
assert!(metadata.contains_key("x-amz-object-lock-retain-until-date"));
// [3] Skip case: No configuration provided
let mut metadata = HashMap::new();
apply_lock_retention(None, &mut metadata);
assert!(!metadata.contains_key("x-amz-object-lock-mode"));
assert!(!metadata.contains_key("x-amz-object-lock-retain-until-date"));
// [4] Skip case: Object Lock not enabled
let mut metadata = HashMap::new();
let config = Some(ObjectLockConfiguration {
object_lock_enabled: None,
rule: Some(ObjectLockRule {
default_retention: Some(DefaultRetention {
mode: Some(ObjectLockRetentionMode::from_static(ObjectLockRetentionMode::COMPLIANCE)),
days: Some(30),
years: None,
}),
}),
});
apply_lock_retention(config, &mut metadata);
assert!(!metadata.contains_key("x-amz-object-lock-mode"));
// [5] Skip case: Explicit retention already set (explicit takes precedence)
let mut metadata = HashMap::new();
metadata.insert("x-amz-object-lock-mode".to_string(), "GOVERNANCE".to_string());
metadata.insert("x-amz-object-lock-retain-until-date".to_string(), "2030-01-01T00:00:00Z".to_string());
let config = Some(ObjectLockConfiguration {
object_lock_enabled: Some(ObjectLockEnabled::from_static(ObjectLockEnabled::ENABLED)),
rule: Some(ObjectLockRule {
default_retention: Some(DefaultRetention {
mode: Some(ObjectLockRetentionMode::from_static(ObjectLockRetentionMode::COMPLIANCE)),
days: Some(30),
years: None,
}),
}),
});
apply_lock_retention(config, &mut metadata);
// Explicit retention should remain unchanged
assert_eq!(metadata.get("x-amz-object-lock-mode"), Some(&"GOVERNANCE".to_string()));
assert_eq!(
metadata.get("x-amz-object-lock-retain-until-date"),
Some(&"2030-01-01T00:00:00Z".to_string())
);
// [6] Skip case: No default retention configured
let mut metadata = HashMap::new();
let config = Some(ObjectLockConfiguration {
object_lock_enabled: Some(ObjectLockEnabled::from_static(ObjectLockEnabled::ENABLED)),
rule: Some(ObjectLockRule { default_retention: None }),
});
apply_lock_retention(config, &mut metadata);
assert!(!metadata.contains_key("x-amz-object-lock-mode"));
// [7] Skip case: No retention mode specified
let mut metadata = HashMap::new();
let config = Some(ObjectLockConfiguration {
object_lock_enabled: Some(ObjectLockEnabled::from_static(ObjectLockEnabled::ENABLED)),
rule: Some(ObjectLockRule {
default_retention: Some(DefaultRetention {
mode: None,
days: Some(30),
years: None,
}),
}),
});
apply_lock_retention(config, &mut metadata);
assert!(!metadata.contains_key("x-amz-object-lock-mode"));
// [8] Skip case: No retention period specified (neither days nor years)
let mut metadata = HashMap::new();
let config = Some(ObjectLockConfiguration {
object_lock_enabled: Some(ObjectLockEnabled::from_static(ObjectLockEnabled::ENABLED)),
rule: Some(ObjectLockRule {
default_retention: Some(DefaultRetention {
mode: Some(ObjectLockRetentionMode::from_static(ObjectLockRetentionMode::COMPLIANCE)),
days: None,
years: None,
}),
}),
});
apply_lock_retention(config, &mut metadata);
assert!(!metadata.contains_key("x-amz-object-lock-mode"));
}
// Note: S3Request structure is complex and requires many fields.
// For real testing, we would need proper integration test setup.
// Removing this test as it requires too much S3 infrastructure setup.
@@ -976,6 +853,31 @@ mod tests {
assert_eq!(formatted, "550e8400-e29b-41d4-a716-446655440000");
}
#[test]
fn test_delete_objects_version_id_normalization() {
use uuid::Uuid;
let fs = FS::new();
let (raw, uuid) = fs.normalize_delete_objects_version_id(Some("null".to_string())).unwrap();
assert_eq!(raw.as_deref(), Some("null"));
assert_eq!(uuid, Some(Uuid::nil()));
let valid = "550e8400-e29b-41d4-a716-446655440000".to_string();
let (raw, uuid) = fs.normalize_delete_objects_version_id(Some(valid.clone())).unwrap();
assert_eq!(raw.as_deref(), Some(valid.as_str()));
assert_eq!(uuid, Some(Uuid::parse_str(&valid).unwrap()));
let err = fs
.normalize_delete_objects_version_id(Some("not-a-uuid".to_string()))
.unwrap_err();
assert!(!err.is_empty());
let (raw, uuid) = fs.normalize_delete_objects_version_id(None).unwrap();
assert!(raw.is_none());
assert!(uuid.is_none());
}
/// Test that ListObjectVersionsOutput markers are correctly set
/// This verifies the fix for boto3 ParamValidationError
#[test]
@@ -1014,34 +916,6 @@ mod tests {
assert_eq!(filtered_version_marker.unwrap(), "null");
}
#[test]
fn test_bucket_policy_round_trip_preserves_original_json_text() {
let policy = r#"{
"Version": "2012-10-17",
"Statement": [{
"Effect": "Allow",
"Principal": {"AWS": "*"},
"Action": "s3:ListBucket",
"Resource": [
"arn:aws:s3:::example-bucket",
"arn:aws:s3:::example-bucket/*"
]
}]
}"#;
let parsed: BucketPolicy = serde_json::from_str(policy).unwrap();
assert!(parsed.is_valid().is_ok());
// Normalized serialization can differ (for example, Action becomes an array).
let normalized = serde_json::to_string(&parsed).unwrap();
assert_ne!(normalized, policy);
// Stored raw policy bytes must preserve exact text for GetBucketPolicy round trip.
let stored = policy.as_bytes().to_vec();
let round_trip = String::from_utf8(stored).unwrap();
assert_eq!(round_trip, policy);
}
#[test]
fn test_matches_origin_pattern_exact_match() {
// Test exact match
rustfs/src/storage/mod.rs
+6 -10
View File
@@ -14,21 +14,17 @@
pub mod access;
pub mod concurrency;
#[cfg(test)]
mod concurrent_get_object_test;
pub mod ecfs;
mod ecfs_extend;
pub(crate) mod entity;
pub(crate) mod helper;
pub mod options;
pub(crate) mod readers;
pub(crate) mod s3_api;
pub mod tonic_service;
pub(crate) use ecfs_extend::*;
#[cfg(test)]
mod concurrent_get_object_test;
mod ecfs_extend;
#[cfg(test)]
mod ecfs_test;
pub(crate) mod head_prefix;
mod objects;
mod sse;
#[cfg(test)]
mod sse_test;
pub(crate) use ecfs_extend::*;
scripts/s3-tests/implemented_tests.txt
+7
View File
@@ -101,6 +101,13 @@ test_bucketv2_notexist
test_bucketv2_policy_another_bucket
test_get_bucket_policy_status
test_get_nonpublicpolicy_principal_bucket_policy_status
test_get_public_acl_bucket_policy_status
test_get_authpublic_acl_bucket_policy_status
test_get_publicpolicy_acl_bucket_policy_status
test_get_nonpublicpolicy_acl_bucket_policy_status
test_block_public_put_bucket_acls
test_block_public_object_canned_acls
test_ignore_public_acls
test_set_get_del_bucket_policy
test_get_object_ifmatch_good
test_get_object_ifmodifiedsince_good