diff --git a/Cargo.lock b/Cargo.lock index e56ed0e..f2e8546 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -2,6 +2,12 @@ # It is not intended for manual editing. version = 4 +[[package]] +name = "adler2" +version = "2.0.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "320119579fcad9c21884f5c4861d16174d0e06250625266f50fe6898340abefa" + [[package]] name = "ahash" version = "0.8.12" @@ -1349,6 +1355,16 @@ version = "0.1.9" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "5baebc0774151f905a1a2cc41989300b1e6fbb29aff0ceffa1064fdd3088d582" +[[package]] +name = "flate2" +version = "1.1.9" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "843fba2746e448b37e26a819579957415c8cef339bf08564fe8b7ddbd959573c" +dependencies = [ + "miniz_oxide", + "zlib-rs", +] + [[package]] name = "fluent-uri" version = "0.4.1" @@ -2188,6 +2204,16 @@ version = "0.2.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "68354c5c6bd36d73ff3feceb05efa59b6acb7626617f4962be322a825e61f79a" +[[package]] +name = "miniz_oxide" +version = "0.8.9" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "1fa76a2c86f704bdb222d66965fb3d63269ce38518b83cb0575fca855ebb6316" +dependencies = [ + "adler2", + "simd-adler32", +] + [[package]] name = "mio" version = "1.2.1" @@ -2934,6 +2960,7 @@ dependencies = [ "url", "urlencoding", "zeroize", + "zip", ] [[package]] @@ -3233,6 +3260,12 @@ dependencies = [ "rand_core 0.6.4", ] +[[package]] +name = "simd-adler32" +version = "0.3.10" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "3a219298ac11a56ea9a6d2120044824d6f01aeb034955e7af7bc16858527deea" + [[package]] name = "similar" version = "2.7.0" @@ -3663,6 +3696,12 @@ version = "0.2.5" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "e421abadd41a4225275504ea4d6566923418b7f05506fbc9c0fe86ba7396114b" +[[package]] +name = "typed-path" +version = "0.12.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "8e28f89b80c87b8fb0cf04ab448d5dd0dd0ade2f8891bae878de66a75a28600e" + [[package]] name = "typenum" version = "1.20.1" @@ -4166,8 +4205,40 @@ dependencies = [ "syn 2.0.118", ] +[[package]] +name = "zip" +version = "8.6.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "2d04a6b5381502aa6087c94c669499eb1602eb9c5e8198e534de571f7154809b" +dependencies = [ + "crc32fast", + "flate2", + "indexmap", + "memchr", + "typed-path", + "zopfli", +] + +[[package]] +name = "zlib-rs" +version = "0.6.6" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "b142a20ec14a91d5bc708c1dc21b080c550113d8aa77afa29635673a65dd02c5" + [[package]] name = "zmij" version = "1.0.21" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "b8848ee67ecc8aedbaf3e4122217aff892639231befc6a1b58d29fff4c2cabaa" + +[[package]] +name = "zopfli" +version = "0.8.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "f05cd8797d63865425ff89b5c4a48804f35ba0ce8d125800027ad6017d2b5249" +dependencies = [ + "bumpalo", + "crc32fast", + "log", + "simd-adler32", +] diff --git a/Cargo.toml b/Cargo.toml index ff88c32..53a150d 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -68,6 +68,7 @@ rustls-pki-types = "1.15" x509-parser = "0.18" zeroize = "1.8" getrandom = "0.4" +zip = { version = "8.6", default-features = false, features = ["deflate"] } # HTTP client for Admin API reqwest = { version = "0.12", default-features = false, features = ["rustls-tls-native-roots", "rustls-tls-webpki-roots", "json", "stream"] } diff --git a/crates/cli/Cargo.toml b/crates/cli/Cargo.toml index df220c6..890789c 100644 --- a/crates/cli/Cargo.toml +++ b/crates/cli/Cargo.toml @@ -56,6 +56,7 @@ tempfile.workspace = true glob.workspace = true shlex.workspace = true zeroize.workspace = true +zip.workspace = true urlencoding.workspace = true url.workspace = true diff --git a/crates/cli/src/commands/admin/bucket_metadata.rs b/crates/cli/src/commands/admin/bucket_metadata.rs new file mode 100644 index 0000000..d115fa2 --- /dev/null +++ b/crates/cli/src/commands/admin/bucket_metadata.rs @@ -0,0 +1,918 @@ +//! Validated, deterministic RustFS bucket-metadata archive commands. + +use clap::{Args, Subcommand, ValueEnum}; +use rc_core::Error; +use rc_core::admin::{ + BUCKET_METADATA_CAPABILITY, BucketMetadataApi, BucketMetadataArchive, + MAX_BUCKET_METADATA_ARCHIVE_BYTES, +}; +use serde::Serialize; +use std::collections::{BTreeMap, BTreeSet}; +use std::fs::File; +use std::future::Future; +use std::io::{Cursor, Read, Write, stdin}; +use std::path::{Path, PathBuf}; +use zeroize::Zeroizing; +use zip::write::SimpleFileOptions; +use zip::{CompressionMethod, DateTime, ZipArchive, ZipWriter}; + +use super::get_admin_client; +use crate::exit_code::ExitCode; +use crate::output::Formatter; + +const MAX_ARCHIVE_ENTRIES: usize = 4096; +const MAX_ENTRY_BYTES: u64 = 16 * 1024 * 1024; +const MAX_EXPANDED_ARCHIVE_BYTES: u64 = 128 * 1024 * 1024; +const TARGETS_CONFIG: &str = "bucket-targets.json"; +const CONFIG_FILES: [&str; 10] = [ + "policy.json", + "notification.xml", + "lifecycle.xml", + "bucket-encryption.xml", + "tagging.xml", + "quota.json", + "object-lock.xml", + "versioning.xml", + "replication.xml", + TARGETS_CONFIG, +]; + +#[derive(Subcommand, Debug)] +pub enum BucketMetadataCommands { + /// Export a deterministic, redacted metadata archive + Export(ExportArgs), + /// Validate and import metadata using an explicit conflict strategy + Import(ImportArgs), +} + +#[derive(Args, Debug)] +pub struct ExportArgs { + pub alias: String, + /// Select one or more buckets; omit to export every bucket + #[arg(long = "bucket")] + pub buckets: Vec, + /// Destination ZIP path + #[arg(long)] + pub file: PathBuf, + /// Atomically replace an existing destination + #[arg(long)] + pub force: bool, +} + +#[derive(Clone, Copy, Debug, Eq, PartialEq, ValueEnum)] +pub enum ConflictStrategy { + /// Stop before mutation when an imported config differs from current state + Fail, + /// Import differing configs + Overwrite, + /// Omit differing configs and import only non-conflicting entries + Skip, +} + +#[derive(Args, Debug)] +pub struct ImportArgs { + pub alias: String, + /// Protected ZIP path, or '-' to read from standard input + #[arg(long)] + pub file: String, + /// Select one or more buckets from the archive + #[arg(long = "bucket")] + pub buckets: Vec, + /// Required behavior when an existing config differs + #[arg(long, value_enum)] + pub conflict: ConflictStrategy, + /// Validate and report the import plan without mutation + #[arg(long)] + pub dry_run: bool, + /// Confirm a mutating import + #[arg(long)] + pub yes: bool, +} + +struct ArchiveEntry { + bytes: Zeroizing>, +} + +#[derive(Default)] +struct ValidatedArchive { + entries: BTreeMap<(String, String), ArchiveEntry>, +} + +#[derive(Debug, Serialize)] +struct OutputEnvelope { + schema_version: u8, + #[serde(rename = "type")] + output_type: &'static str, + status: &'static str, + data: OutputData, +} + +#[derive(Debug, Serialize)] +struct OutputData { + operations: Vec, +} + +#[derive(Debug, Serialize)] +struct OutputOperation { + operation: &'static str, + resource: String, + state: &'static str, + operation_id: Option, + changed: bool, + result: OperationResult, +} + +#[derive(Debug, Serialize)] +struct OperationResult { + bucket: String, + configs: usize, + planned_changes: usize, + conflicts: usize, + skipped: usize, + unchanged: usize, + dry_run: bool, + #[serde(skip_serializing_if = "Option::is_none")] + path: Option, +} + +#[derive(Debug, Serialize)] +struct ErrorEnvelope { + schema_version: u8, + #[serde(rename = "type")] + output_type: &'static str, + status: &'static str, + error: ErrorDetail, +} + +#[derive(Debug, Serialize)] +struct ErrorDetail { + #[serde(rename = "type")] + error_type: &'static str, + message: String, + retryable: bool, + #[serde(skip_serializing_if = "Option::is_none")] + capability: Option<&'static str>, + #[serde(skip_serializing_if = "Option::is_none")] + server: Option<&'static str>, + #[serde(skip_serializing_if = "Option::is_none")] + outcome: Option<&'static str>, + suggestion: &'static str, +} + +pub async fn execute(command: BucketMetadataCommands, formatter: &Formatter) -> ExitCode { + let alias = match &command { + BucketMetadataCommands::Export(args) => &args.alias, + BucketMetadataCommands::Import(args) => &args.alias, + }; + let client = match get_admin_client(alias, formatter) { + Ok(client) => client, + Err(code) => return code, + }; + let result = match command { + BucketMetadataCommands::Export(args) => export(args, &client, formatter).await, + BucketMetadataCommands::Import(args) => import(args, &client, formatter).await, + }; + match result { + Ok(()) => ExitCode::Success, + Err(error) => emit_error(&error, formatter), + } +} + +async fn export( + args: ExportArgs, + api: &dyn BucketMetadataApi, + formatter: &Formatter, +) -> rc_core::Result<()> { + let selected = validate_bucket_selection(&args.buckets)?; + let archive = fetch_server_archive(api, &selected).await?; + let buckets = if selected.is_empty() { + archive.bucket_names() + } else { + selected.iter().cloned().collect() + }; + let bucket_count = buckets.len(); + let bytes = archive.encode()?; + write_atomic_private_file(&args.file, &bytes, args.force)?; + + let operations = buckets + .into_iter() + .map(|bucket| OutputOperation { + operation: "bucket-metadata.export", + resource: format!("bucket/{bucket}"), + state: "succeeded", + operation_id: None, + changed: true, + result: OperationResult { + configs: archive.config_count(&bucket), + planned_changes: archive.config_count(&bucket), + bucket, + conflicts: 0, + skipped: 0, + unchanged: 0, + dry_run: false, + path: Some(args.file.display().to_string()), + }, + }) + .collect(); + emit_success(operations, formatter, || { + formatter.success(&format!( + "Exported validated metadata for {} bucket(s) to {}", + bucket_count, + args.file.display() + )); + }); + Ok(()) +} + +async fn import( + args: ImportArgs, + api: &dyn BucketMetadataApi, + formatter: &Formatter, +) -> rc_core::Result<()> { + if !args.dry_run && !args.yes { + return Err(Error::Config( + "Bucket metadata import requires --yes unless --dry-run is used".to_string(), + )); + } + let selected = validate_bucket_selection(&args.buckets)?; + let input = read_protected_archive(&args.file)?; + let mut requested = ValidatedArchive::read(&input)?; + requested.select(&selected)?; + requested.reject_redacted_target_credentials()?; + + let current = fetch_server_archive(api, &selected).await?; + let plan = ImportPlan::build(requested, ¤t, args.conflict)?; + let operations = plan.operations(args.dry_run); + if !args.dry_run && !plan.archive.entries.is_empty() { + import_with_cancellation( + api, + BucketMetadataArchive::new(plan.archive.encode()?)?, + tokio::signal::ctrl_c(), + ) + .await?; + } + emit_success(operations, formatter, || { + plan.print(args.dry_run, formatter) + }); + Ok(()) +} + +async fn import_with_cancellation( + api: &dyn BucketMetadataApi, + archive: BucketMetadataArchive, + cancellation: F, +) -> rc_core::Result<()> +where + F: Future>, +{ + tokio::pin!(cancellation); + tokio::select! { + result = api.import_bucket_metadata(archive) => result, + signal = &mut cancellation => match signal { + Ok(()) => Err(Error::Interrupted( + "Bucket metadata import was interrupted; the outcome may be partial, so inspect every selected bucket before retrying" + .to_string(), + )), + Err(_) => Err(Error::General( + "Failed to register bucket metadata import interruption handling".to_string(), + )), + }, + } +} + +async fn fetch_server_archive( + api: &dyn BucketMetadataApi, + selected: &BTreeSet, +) -> rc_core::Result { + if selected.is_empty() { + let archive = api.export_bucket_metadata(None).await?; + return ValidatedArchive::read_server(archive.as_bytes()); + } + let mut combined = ValidatedArchive::default(); + for bucket in selected { + let archive = api.export_bucket_metadata(Some(bucket)).await?; + let archive = ValidatedArchive::read_server(archive.as_bytes())?; + if archive + .entries + .keys() + .any(|(candidate, _)| candidate != bucket) + { + return Err(Error::General( + "RustFS returned metadata outside the selected bucket".to_string(), + )); + } + for (key, entry) in archive.entries { + if combined.entries.insert(key, entry).is_some() { + return Err(Error::General( + "RustFS returned duplicate bucket metadata".to_string(), + )); + } + } + } + Ok(combined) +} + +fn validate_bucket_selection(values: &[String]) -> rc_core::Result> { + let mut selected = BTreeSet::new(); + for value in values { + if value.is_empty() + || value.len() > 63 + || value.contains(['/', '\\', '\0']) + || value == "." + || value == ".." + { + return Err(Error::InvalidPath( + "Bucket selectors must be non-empty bucket names without path separators" + .to_string(), + )); + } + selected.insert(value.clone()); + } + Ok(selected) +} + +impl ValidatedArchive { + fn read(bytes: &[u8]) -> rc_core::Result { + Self::read_internal(bytes, false) + } + + fn read_server(bytes: &[u8]) -> rc_core::Result { + Self::read_internal(bytes, true) + } + + fn read_internal(bytes: &[u8], allow_empty: bool) -> rc_core::Result { + if bytes.len() > MAX_BUCKET_METADATA_ARCHIVE_BYTES { + return Err(Error::RequestRejected( + "Bucket metadata archive exceeds the 100 MiB limit".to_string(), + )); + } + let mut zip = ZipArchive::new(Cursor::new(bytes)).map_err(|_| { + Error::Config("Bucket metadata input is not a valid ZIP archive".to_string()) + })?; + if zip.len() > MAX_ARCHIVE_ENTRIES { + return Err(Error::Config(format!( + "Bucket metadata archive exceeds the {MAX_ARCHIVE_ENTRIES} entry limit" + ))); + } + let mut entries = BTreeMap::new(); + let mut expanded = 0_u64; + for index in 0..zip.len() { + let mut file = zip.by_index(index).map_err(|_| { + Error::Config("Bucket metadata archive contains an unreadable entry".to_string()) + })?; + if !file.is_file() || file.enclosed_name().is_none() { + return Err(Error::Config( + "Bucket metadata archive contains a non-file or unsafe entry".to_string(), + )); + } + if file.size() > MAX_ENTRY_BYTES { + return Err(Error::Config(format!( + "Bucket metadata archive entry exceeds the {MAX_ENTRY_BYTES} byte limit" + ))); + } + expanded = expanded.saturating_add(file.size()); + if expanded > MAX_EXPANDED_ARCHIVE_BYTES { + return Err(Error::Config( + "Bucket metadata archive expands beyond the 128 MiB limit".to_string(), + )); + } + let name = file.name().to_string(); + let mut parts = name.split('/'); + let bucket = parts.next().unwrap_or_default(); + let config = parts.next().unwrap_or_default(); + if bucket.is_empty() + || config.is_empty() + || parts.next().is_some() + || !CONFIG_FILES.contains(&config) + { + return Err(Error::Config( + "Bucket metadata archive contains an unsupported entry path".to_string(), + )); + } + validate_bucket_selection(&[bucket.to_string()])?; + let mut content = Zeroizing::new(Vec::new()); + file.by_ref() + .take(MAX_ENTRY_BYTES + 1) + .read_to_end(&mut content) + .map_err(|_| { + Error::Config( + "Bucket metadata archive contains an unreadable entry".to_string(), + ) + })?; + if content.len() as u64 > MAX_ENTRY_BYTES { + return Err(Error::Config( + "Bucket metadata archive entry exceeded its declared bound".to_string(), + )); + } + if content.is_empty() { + return Err(Error::Config( + "Bucket metadata archive contains an empty config entry".to_string(), + )); + } + if entries + .insert( + (bucket.to_string(), config.to_string()), + ArchiveEntry { bytes: content }, + ) + .is_some() + { + return Err(Error::Config( + "Bucket metadata archive contains a duplicate config entry".to_string(), + )); + } + } + if entries.is_empty() && !allow_empty { + return Err(Error::Config( + "Bucket metadata archive contains no supported configs".to_string(), + )); + } + Ok(Self { entries }) + } + + fn encode(&self) -> rc_core::Result> { + let mut writer = ZipWriter::new(Cursor::new(Vec::new())); + let options = SimpleFileOptions::default() + .compression_method(CompressionMethod::Deflated) + .last_modified_time(DateTime::default()) + .unix_permissions(0o600); + for ((bucket, config), entry) in &self.entries { + writer + .start_file(format!("{bucket}/{config}"), options) + .map_err(|_| { + Error::General("Failed to create bucket metadata archive".to_string()) + })?; + writer.write_all(entry.bytes.as_slice())?; + } + let bytes = writer + .finish() + .map_err(|_| Error::General("Failed to finalize bucket metadata archive".to_string()))? + .into_inner(); + if bytes.len() > MAX_BUCKET_METADATA_ARCHIVE_BYTES { + return Err(Error::RequestRejected( + "Bucket metadata archive exceeds the 100 MiB limit".to_string(), + )); + } + Ok(bytes) + } + + fn select(&mut self, selected: &BTreeSet) -> rc_core::Result<()> { + if selected.is_empty() { + return Ok(()); + } + let available = self.bucket_names().into_iter().collect::>(); + let missing = selected.difference(&available).cloned().collect::>(); + if !missing.is_empty() { + return Err(Error::NotFound(format!( + "Selected bucket is absent from the metadata archive: {}", + missing.join(", ") + ))); + } + self.entries + .retain(|(bucket, _), _| selected.contains(bucket)); + Ok(()) + } + + fn bucket_names(&self) -> Vec { + self.entries + .keys() + .map(|(bucket, _)| bucket.clone()) + .collect::>() + .into_iter() + .collect() + } + + fn config_count(&self, bucket: &str) -> usize { + self.entries + .keys() + .filter(|(candidate, _)| candidate == bucket) + .count() + } + + fn reject_redacted_target_credentials(&self) -> rc_core::Result<()> { + for ((_, config), entry) in &self.entries { + if config != TARGETS_CONFIG { + continue; + } + let value: serde_json::Value = + serde_json::from_slice(entry.bytes.as_slice()).map_err(|_| { + Error::Config("Bucket target metadata is malformed JSON".to_string()) + })?; + if contains_unusable_target_secret(&value) { + return Err(Error::Config( + "Bucket target metadata contains missing or redacted credentials and cannot be imported; provide a protected archive containing the real target credentials" + .to_string(), + )); + } + } + Ok(()) + } +} + +fn contains_unusable_target_secret(value: &serde_json::Value) -> bool { + match value { + serde_json::Value::Array(values) => values.iter().any(contains_unusable_target_secret), + serde_json::Value::Object(values) => values.iter().any(|(key, value)| { + let normalized = key + .chars() + .filter(|character| character.is_ascii_alphanumeric()) + .flat_map(char::to_lowercase) + .collect::(); + if normalized == "secretkey" { + return value.as_str().is_none_or(|secret| { + matches!( + secret.trim().to_ascii_lowercase().as_str(), + "" | "redacted" | "*redacted*" | "" + ) + }); + } + contains_unusable_target_secret(value) + }), + _ => false, + } +} + +struct ImportPlan { + archive: ValidatedArchive, + buckets: BTreeMap, +} + +#[derive(Default)] +struct BucketPlan { + configs: usize, + conflicts: usize, + skipped: usize, + unchanged: usize, +} + +impl ImportPlan { + fn build( + mut requested: ValidatedArchive, + current: &ValidatedArchive, + strategy: ConflictStrategy, + ) -> rc_core::Result { + let source_buckets = requested.bucket_names(); + let mut buckets = source_buckets + .iter() + .map(|bucket| (bucket.clone(), BucketPlan::default())) + .collect::>(); + let mut remove = Vec::new(); + for (key, requested_entry) in &requested.entries { + let bucket = buckets + .get_mut(&key.0) + .expect("bucket plan was created from requested entries"); + bucket.configs += 1; + if let Some(current_entry) = current.entries.get(key) { + if current_entry.bytes.as_slice() == requested_entry.bytes.as_slice() { + bucket.unchanged += 1; + remove.push(key.clone()); + } else { + bucket.conflicts += 1; + match strategy { + ConflictStrategy::Fail => {} + ConflictStrategy::Overwrite => {} + ConflictStrategy::Skip => { + bucket.skipped += 1; + remove.push(key.clone()); + } + } + } + } + } + let conflicts = buckets.values().map(|plan| plan.conflicts).sum::(); + if conflicts > 0 && strategy == ConflictStrategy::Fail { + return Err(Error::Conflict(format!( + "Bucket metadata import found {conflicts} conflicting config(s); use --conflict overwrite or --conflict skip" + ))); + } + for key in remove { + requested.entries.remove(&key); + } + Ok(Self { + archive: requested, + buckets, + }) + } + + fn operations(&self, dry_run: bool) -> Vec { + self.buckets + .iter() + .map(|(bucket, plan)| { + let applied = self.archive.config_count(bucket); + OutputOperation { + operation: "bucket-metadata.import", + resource: format!("bucket/{bucket}"), + state: "succeeded", + operation_id: None, + changed: !dry_run && applied > 0, + result: OperationResult { + bucket: bucket.clone(), + configs: plan.configs, + planned_changes: applied, + conflicts: plan.conflicts, + skipped: plan.skipped, + unchanged: plan.unchanged, + dry_run, + path: None, + }, + } + }) + .collect() + } + + fn print(&self, dry_run: bool, formatter: &Formatter) { + for (bucket, plan) in &self.buckets { + formatter.println(&format!( + "{bucket}: {} config(s), {} planned change(s), {} conflict(s), {} skipped, {} unchanged{}", + plan.configs, + self.archive.config_count(bucket), + plan.conflicts, + plan.skipped, + plan.unchanged, + if dry_run { " (dry-run)" } else { "" } + )); + } + } +} + +fn read_protected_archive(path: &str) -> rc_core::Result>> { + if path == "-" { + return read_bounded(stdin().lock()); + } + let path = Path::new(path); + let before = std::fs::symlink_metadata(path) + .map_err(|_| Error::InvalidPath("Failed to inspect bucket metadata input".to_string()))?; + if before.file_type().is_symlink() || !before.is_file() { + return Err(Error::InvalidPath( + "Bucket metadata input must be a regular file, not a symlink".to_string(), + )); + } + let file = File::open(path) + .map_err(|_| Error::InvalidPath("Failed to open bucket metadata input".to_string()))?; + let opened = file.metadata().map_err(|_| { + Error::InvalidPath("Failed to inspect opened bucket metadata input".to_string()) + })?; + #[cfg(unix)] + { + use std::os::unix::fs::{MetadataExt, PermissionsExt}; + if before.dev() != opened.dev() || before.ino() != opened.ino() { + return Err(Error::InvalidPath( + "Bucket metadata input changed while being opened".to_string(), + )); + } + if opened.permissions().mode() & 0o077 != 0 { + return Err(Error::InvalidPath( + "Bucket metadata input cannot grant group or other permissions".to_string(), + )); + } + } + read_bounded(file) +} + +fn read_bounded(reader: impl Read) -> rc_core::Result>> { + let mut bytes = Zeroizing::new(Vec::new()); + reader + .take(MAX_BUCKET_METADATA_ARCHIVE_BYTES as u64 + 1) + .read_to_end(&mut bytes) + .map_err(|_| Error::InvalidPath("Failed to read bucket metadata input".to_string()))?; + if bytes.len() > MAX_BUCKET_METADATA_ARCHIVE_BYTES { + return Err(Error::RequestRejected( + "Bucket metadata archive exceeds the 100 MiB limit".to_string(), + )); + } + Ok(bytes) +} + +fn write_atomic_private_file(path: &Path, bytes: &[u8], force: bool) -> rc_core::Result<()> { + let directory = path + .parent() + .filter(|path| !path.as_os_str().is_empty()) + .unwrap_or_else(|| Path::new(".")); + std::fs::create_dir_all(directory)?; + let mut temporary = tempfile::NamedTempFile::new_in(directory)?; + temporary.write_all(bytes)?; + temporary.as_file_mut().sync_all()?; + #[cfg(unix)] + { + use std::os::unix::fs::PermissionsExt; + temporary + .as_file_mut() + .set_permissions(std::fs::Permissions::from_mode(0o600))?; + } + if force { + temporary.persist(path).map_err(|error| error.error)?; + } else { + temporary + .persist_noclobber(path) + .map_err(|error| error.error)?; + } + Ok(()) +} + +fn emit_success(operations: Vec, formatter: &Formatter, human: impl FnOnce()) { + if formatter.is_json() { + formatter.json(&OutputEnvelope { + schema_version: 3, + output_type: "admin_operations", + status: "success", + data: OutputData { operations }, + }); + } else { + human(); + } +} + +fn emit_error(error: &Error, formatter: &Formatter) -> ExitCode { + let code = ExitCode::from_i32(error.exit_code()).unwrap_or(ExitCode::GeneralError); + let uncertain_import = matches!( + error, + Error::Network(message) + if message.contains("outcome is unknown") + || message.contains("partially applied") + || message.contains("inspect every selected bucket") + ); + if formatter.is_json() { + formatter.json_error(&ErrorEnvelope { + schema_version: 3, + output_type: "admin_operations", + status: "error", + error: ErrorDetail { + error_type: match code { + ExitCode::UsageError => "usage_error", + ExitCode::NetworkError => "network_error", + ExitCode::AuthError => "auth_error", + ExitCode::NotFound => "not_found", + ExitCode::Conflict => "conflict", + ExitCode::UnsupportedFeature => "unsupported_feature", + ExitCode::Interrupted => "interrupted", + _ => "general_error", + }, + message: error.to_string(), + retryable: matches!(code, ExitCode::NetworkError) && !uncertain_import, + capability: matches!(code, ExitCode::UnsupportedFeature) + .then_some(BUCKET_METADATA_CAPABILITY), + server: matches!(code, ExitCode::UnsupportedFeature).then_some("rustfs"), + outcome: uncertain_import.then_some("unknown_partial"), + suggestion: match code { + ExitCode::Conflict => { + "Choose an explicit overwrite or skip conflict strategy when appropriate." + } + ExitCode::NetworkError => { + "For import failures, inspect all selected buckets before retrying." + } + ExitCode::AuthError => { + "Verify ExportBucketMetadata or ImportBucketMetadata permission." + } + ExitCode::UnsupportedFeature => { + "Verify the RustFS server exposes the v3 bucket metadata routes." + } + _ => "Validate the protected archive, bucket selection, and command arguments.", + }, + }, + }); + } else { + formatter.error_with_code(code, &error.to_string()); + } + code +} + +#[cfg(test)] +mod tests { + use super::*; + use async_trait::async_trait; + use std::future::pending; + + fn archive(entries: &[(&str, &[u8])]) -> Vec { + let mut writer = ZipWriter::new(Cursor::new(Vec::new())); + for (name, bytes) in entries { + writer + .start_file(*name, SimpleFileOptions::default()) + .expect("start entry"); + writer.write_all(bytes).expect("write entry"); + } + writer.finish().expect("finish").into_inner() + } + + #[test] + fn deterministic_archive_sorts_entries_and_round_trips_multiple_buckets() { + let input = archive(&[ + ("zeta/quota.json", br#"{"quota":2}"#), + ("alpha/policy.json", br#"{"Version":"2012-10-17"}"#), + ]); + let parsed = ValidatedArchive::read(&input).expect("parse"); + let first = parsed.encode().expect("encode"); + let second = parsed.encode().expect("encode"); + assert_eq!(first, second); + let reparsed = ValidatedArchive::read(&first).expect("reparse"); + assert_eq!(reparsed.bucket_names(), ["alpha", "zeta"]); + } + + #[test] + fn selection_does_not_touch_unselected_buckets() { + let input = archive(&[("alpha/policy.json", b"{}"), ("beta/policy.json", b"{}")]); + let mut parsed = ValidatedArchive::read(&input).expect("parse"); + parsed + .select(&BTreeSet::from(["beta".to_string()])) + .expect("select"); + assert_eq!(parsed.bucket_names(), ["beta"]); + } + + #[test] + fn malformed_paths_and_unknown_configs_fail_closed() { + assert!(ValidatedArchive::read(&archive(&[("../policy.json", b"{}")])).is_err()); + assert!(ValidatedArchive::read(&archive(&[("alpha/credentials.json", b"{}")])).is_err()); + } + + #[test] + fn redacted_replication_target_credentials_are_never_imported() { + let input = archive(&[( + "alpha/bucket-targets.json", + br#"{"targets":[{"accessKey":"visible","secretKey":"","description":"server-secret"}]}"#, + )]); + let parsed = ValidatedArchive::read(&input).expect("parse"); + let error = parsed + .reject_redacted_target_credentials() + .expect_err("redacted input must fail"); + let message = error.to_string(); + assert!(!message.contains("server-secret")); + } + + #[test] + fn conflict_strategies_are_explicit_and_skip_is_per_entry() { + let requested = ValidatedArchive::read(&archive(&[ + ("alpha/policy.json", b"new"), + ("alpha/quota.json", b"same"), + ])) + .expect("requested"); + let current = ValidatedArchive::read(&archive(&[ + ("alpha/policy.json", b"old"), + ("alpha/quota.json", b"same"), + ])) + .expect("current"); + assert!(ImportPlan::build(requested, ¤t, ConflictStrategy::Fail).is_err()); + + let requested = ValidatedArchive::read(&archive(&[ + ("alpha/policy.json", b"new"), + ("alpha/quota.json", b"same"), + ])) + .expect("requested"); + let plan = + ImportPlan::build(requested, ¤t, ConflictStrategy::Skip).expect("skip plan"); + assert!(plan.archive.entries.is_empty()); + assert_eq!(plan.buckets["alpha"].conflicts, 1); + assert_eq!(plan.buckets["alpha"].skipped, 1); + } + + #[test] + fn export_is_private_atomic_and_refuses_implicit_overwrite() { + let directory = tempfile::tempdir().expect("tempdir"); + let path = directory.path().join("archive.zip"); + write_atomic_private_file(&path, b"first", false).expect("first"); + assert!(write_atomic_private_file(&path, b"second", false).is_err()); + assert_eq!(std::fs::read(&path).expect("read"), b"first"); + write_atomic_private_file(&path, b"second", true).expect("replace"); + assert_eq!(std::fs::read(&path).expect("read"), b"second"); + #[cfg(unix)] + { + use std::os::unix::fs::PermissionsExt; + assert_eq!( + std::fs::metadata(&path) + .expect("metadata") + .permissions() + .mode() + & 0o077, + 0 + ); + } + } + + struct PendingImportApi; + + #[async_trait] + impl BucketMetadataApi for PendingImportApi { + async fn export_bucket_metadata( + &self, + _bucket: Option<&str>, + ) -> rc_core::Result { + panic!("export is not used") + } + + async fn import_bucket_metadata( + &self, + _archive: BucketMetadataArchive, + ) -> rc_core::Result<()> { + pending().await + } + } + + #[tokio::test] + async fn interruption_reports_unknown_partial_outcome() { + let archive = + BucketMetadataArchive::new(archive(&[("alpha/policy.json", b"new")])).expect("archive"); + let error = + import_with_cancellation(&PendingImportApi, archive, std::future::ready(Ok(()))) + .await + .expect_err("interrupted"); + assert!(matches!(error, Error::Interrupted(_))); + assert!(error.to_string().contains("partial")); + } +} diff --git a/crates/cli/src/commands/admin/mod.rs b/crates/cli/src/commands/admin/mod.rs index aa6c4c6..5b61357 100644 --- a/crates/cli/src/commands/admin/mod.rs +++ b/crates/cli/src/commands/admin/mod.rs @@ -4,6 +4,7 @@ //! service accounts, and cluster operations through the RustFS Admin API. mod access_key; +mod bucket_metadata; mod capabilities; mod config; mod decommission; @@ -111,6 +112,10 @@ pub enum AdminCommands { #[command(name = "access-key", subcommand)] AccessKey(access_key::AccessKeyCommands), + /// Export or import validated per-bucket metadata archives + #[command(name = "bucket-metadata", subcommand)] + BucketMetadata(bucket_metadata::BucketMetadataCommands), + /// Control the server process (restart, stop, freeze, unfreeze) #[command(subcommand)] Service(service::ServiceCommands), @@ -150,6 +155,9 @@ pub async fn execute(cmd: AdminCommands, output_config: OutputConfig) -> ExitCod AdminCommands::AccessKey(access_key_cmd) => { access_key::execute(access_key_cmd, &formatter).await } + AdminCommands::BucketMetadata(command) => { + bucket_metadata::execute(command, &formatter).await + } AdminCommands::Service(service_cmd) => service::execute(service_cmd, &formatter).await, AdminCommands::Replicate(replicate_cmd) => { replicate::execute(replicate_cmd, &formatter).await diff --git a/crates/cli/tests/admin_bucket_metadata.rs b/crates/cli/tests/admin_bucket_metadata.rs new file mode 100644 index 0000000..f29f4f8 --- /dev/null +++ b/crates/cli/tests/admin_bucket_metadata.rs @@ -0,0 +1,309 @@ +#![cfg(not(windows))] + +mod admin_support; + +use std::io::{Cursor, Write}; +use std::process::Command; +use std::time::Duration; + +use admin_support::{rc_binary, rc_host_alias, start_admin_binary_sequence_test_server}; +use zip::write::SimpleFileOptions; +use zip::{ZipArchive, ZipWriter}; + +fn archive(entries: &[(&str, &[u8])]) -> Vec { + let mut writer = ZipWriter::new(Cursor::new(Vec::new())); + for (name, bytes) in entries { + writer + .start_file(*name, SimpleFileOptions::default()) + .expect("start ZIP entry"); + writer.write_all(bytes).expect("write ZIP entry"); + } + writer.finish().expect("finish ZIP").into_inner() +} + +fn protected_archive(directory: &tempfile::TempDir, bytes: &[u8]) -> String { + let path = directory.path().join("metadata.zip"); + std::fs::write(&path, bytes).expect("write archive"); + #[cfg(unix)] + { + use std::os::unix::fs::PermissionsExt; + std::fs::set_permissions(&path, std::fs::Permissions::from_mode(0o600)) + .expect("protect archive"); + } + path.display().to_string() +} + +fn run(args: &[&str], endpoint: &str, config_dir: &tempfile::TempDir) -> std::process::Output { + Command::new(rc_binary()) + .args(args) + .env("RC_CONFIG_DIR", config_dir.path()) + .env("RC_HOST_myalias", rc_host_alias(endpoint)) + .output() + .expect("run rc command") +} + +fn assert_v3(stdout: &[u8]) -> serde_json::Value { + let value: serde_json::Value = serde_json::from_slice(stdout).expect("JSON output"); + let schema_path = std::path::Path::new(env!("CARGO_MANIFEST_DIR")) + .join("../..") + .join("schemas/output_v3.json"); + let schema: serde_json::Value = + serde_json::from_slice(&std::fs::read(schema_path).expect("read schema")) + .expect("parse schema"); + let validator = jsonschema::validator_for(&schema).expect("compile schema"); + let errors = validator + .iter_errors(&value) + .map(|error| error.to_string()) + .collect::>(); + assert!(errors.is_empty(), "schema errors: {}", errors.join("\n")); + value +} + +#[test] +fn export_is_selected_deterministic_atomic_and_v3() { + let config_dir = tempfile::tempdir().expect("config dir"); + let response = archive(&[ + ("alpha/quota.json", br#"{"quota":10}"#), + ("alpha/policy.json", br#"{"Version":"2012-10-17"}"#), + ]); + let (endpoint, receiver, handle) = start_admin_binary_sequence_test_server(vec![ + ("200 OK", "application/zip", response.clone()), + ("200 OK", "application/zip", response), + ]); + let first = config_dir.path().join("first.zip"); + let second = config_dir.path().join("second.zip"); + + let output = run( + &[ + "--json", + "admin", + "bucket-metadata", + "export", + "myalias", + "--bucket", + "alpha", + "--file", + first.to_str().expect("UTF-8 path"), + ], + &endpoint, + &config_dir, + ); + assert!( + output.status.success(), + "stderr: {}", + String::from_utf8_lossy(&output.stderr) + ); + let value = assert_v3(&output.stdout); + assert_eq!(value["data"]["operations"][0]["result"]["bucket"], "alpha"); + + let output = run( + &[ + "--json", + "admin", + "bucket-metadata", + "export", + "myalias", + "--bucket", + "alpha", + "--file", + second.to_str().expect("UTF-8 path"), + ], + &endpoint, + &config_dir, + ); + assert!(output.status.success()); + assert_eq!( + std::fs::read(first).expect("first"), + std::fs::read(second).expect("second") + ); + for _ in 0..2 { + let request = receiver + .recv_timeout(Duration::from_secs(5)) + .expect("export request"); + assert_eq!(request.method, "GET"); + assert_eq!( + request.target, + "/rustfs/admin/v3/export-bucket-metadata?bucket=alpha" + ); + } + handle.join().expect("server finished"); +} + +#[test] +fn dry_run_reports_conflicts_without_mutation() { + let config_dir = tempfile::tempdir().expect("config dir"); + let source = protected_archive( + &config_dir, + &archive(&[("alpha/policy.json", b"new-policy")]), + ); + let current = archive(&[("alpha/policy.json", b"old-policy")]); + let (endpoint, receiver, handle) = + start_admin_binary_sequence_test_server(vec![("200 OK", "application/zip", current)]); + + let output = run( + &[ + "--json", + "admin", + "bucket-metadata", + "import", + "myalias", + "--file", + &source, + "--conflict", + "overwrite", + "--dry-run", + ], + &endpoint, + &config_dir, + ); + assert!( + output.status.success(), + "stderr: {}", + String::from_utf8_lossy(&output.stderr) + ); + let value = assert_v3(&output.stdout); + assert_eq!(value["data"]["operations"][0]["changed"], false); + assert_eq!(value["data"]["operations"][0]["result"]["conflicts"], 1); + let request = receiver + .recv_timeout(Duration::from_secs(5)) + .expect("preflight export"); + assert_eq!(request.method, "GET"); + assert!(receiver.recv_timeout(Duration::from_millis(100)).is_err()); + handle.join().expect("server finished"); +} + +#[test] +fn confirmed_import_sends_one_bounded_zip_and_never_retries() { + let config_dir = tempfile::tempdir().expect("config dir"); + let source_bytes = archive(&[("alpha/policy.json", b"new-policy")]); + let source = protected_archive(&config_dir, &source_bytes); + let current = archive(&[("alpha/policy.json", b"old-policy")]); + let (endpoint, receiver, handle) = start_admin_binary_sequence_test_server(vec![ + ("200 OK", "application/zip", current), + ( + "500 Internal Server Error", + "application/json", + b"{}".to_vec(), + ), + ]); + + let output = run( + &[ + "--json", + "admin", + "bucket-metadata", + "import", + "myalias", + "--file", + &source, + "--conflict", + "overwrite", + "--yes", + ], + &endpoint, + &config_dir, + ); + assert_eq!(output.status.code(), Some(3)); + let stderr = String::from_utf8_lossy(&output.stderr); + assert!(stderr.contains("partially applied")); + let error = assert_v3(stderr.as_bytes()); + assert_eq!(error["error"]["type"], "network_error"); + assert_eq!(error["error"]["outcome"], "unknown_partial"); + assert_eq!(error["error"]["retryable"], false); + let preflight = receiver + .recv_timeout(Duration::from_secs(5)) + .expect("preflight"); + let mutation = receiver + .recv_timeout(Duration::from_secs(5)) + .expect("mutation"); + assert_eq!(preflight.method, "GET"); + assert_eq!(mutation.method, "PUT"); + assert_eq!(mutation.target, "/rustfs/admin/v3/import-bucket-metadata"); + assert!( + mutation + .headers + .to_ascii_lowercase() + .contains("application/zip") + ); + let imported = ZipArchive::new(Cursor::new(mutation.body)).expect("valid uploaded ZIP"); + assert_eq!(imported.len(), 1); + assert!(receiver.recv_timeout(Duration::from_millis(200)).is_err()); + handle.join().expect("server finished"); +} + +#[test] +fn malformed_archive_and_redacted_target_fail_before_network() { + let config_dir = tempfile::tempdir().expect("config dir"); + let malformed = protected_archive(&config_dir, b"not-a-zip"); + let output = run( + &[ + "admin", + "bucket-metadata", + "import", + "myalias", + "--file", + &malformed, + "--conflict", + "fail", + "--dry-run", + ], + "http://127.0.0.1:9", + &config_dir, + ); + assert_eq!(output.status.code(), Some(2)); + + let redacted = archive(&[( + "alpha/bucket-targets.json", + br#"{"targets":[{"secretKey":"*redacted*"}]}"#, + )]); + let redacted = protected_archive(&config_dir, &redacted); + let output = run( + &[ + "admin", + "bucket-metadata", + "import", + "myalias", + "--file", + &redacted, + "--conflict", + "fail", + "--dry-run", + ], + "http://127.0.0.1:9", + &config_dir, + ); + assert_eq!(output.status.code(), Some(2)); + assert!(!String::from_utf8_lossy(&output.stderr).contains("secretKey")); +} + +#[test] +fn access_denial_and_unsupported_route_are_distinct() { + for (status, expected) in [("403 Forbidden", 4), ("404 Not Found", 7)] { + let config_dir = tempfile::tempdir().expect("config dir"); + let destination = config_dir.path().join("archive.zip"); + let (endpoint, receiver, handle) = start_admin_binary_sequence_test_server(vec![( + status, + "application/json", + b"{}".to_vec(), + )]); + let output = run( + &[ + "--json", + "admin", + "bucket-metadata", + "export", + "myalias", + "--file", + destination.to_str().expect("UTF-8 path"), + ], + &endpoint, + &config_dir, + ); + assert_eq!(output.status.code(), Some(expected)); + assert_v3(&output.stderr); + receiver + .recv_timeout(Duration::from_secs(5)) + .expect("request"); + handle.join().expect("server finished"); + } +} diff --git a/crates/cli/tests/admin_support/mod.rs b/crates/cli/tests/admin_support/mod.rs index 14ebf82..1440eb8 100644 --- a/crates/cli/tests/admin_support/mod.rs +++ b/crates/cli/tests/admin_support/mod.rs @@ -261,6 +261,53 @@ pub fn start_admin_sequence_test_server( (endpoint, receiver, handle) } +#[allow(dead_code)] +pub fn start_admin_binary_sequence_test_server( + responses: Vec<(&'static str, &'static str, Vec)>, +) -> (String, Receiver, JoinHandle<()>) { + let listener = TcpListener::bind("127.0.0.1:0").expect("bind admin test server"); + listener + .set_nonblocking(true) + .expect("set admin test server nonblocking"); + let endpoint = format!("http://{}", listener.local_addr().expect("server address")); + let (sender, receiver) = mpsc::channel(); + + let handle = thread::spawn(move || { + for (response_status, content_type, response_body) in responses { + let deadline = Instant::now() + Duration::from_secs(120); + let (mut stream, _) = loop { + match listener.accept() { + Ok(accepted) => break accepted, + Err(error) + if error.kind() == ErrorKind::WouldBlock && Instant::now() < deadline => + { + thread::sleep(Duration::from_millis(10)); + } + Err(error) => panic!("accept admin request: {error}"), + } + }; + stream + .set_nonblocking(false) + .expect("set admin request stream blocking"); + let request = read_admin_request(&mut stream); + sender.send(request).expect("send captured request"); + + let headers = format!( + "HTTP/1.1 {response_status}\r\ncontent-type: {content_type}\r\ncontent-length: {}\r\nconnection: close\r\n\r\n", + response_body.len() + ); + stream + .write_all(headers.as_bytes()) + .expect("write admin response headers"); + stream + .write_all(&response_body) + .expect("write admin response body"); + } + }); + + (endpoint, receiver, handle) +} + pub fn rc_host_alias(endpoint: &str) -> String { let (_, endpoint_authority) = endpoint.split_once("://").expect("endpoint has scheme"); format!("http://ACCESS_KEY:SECRET_KEY@{endpoint_authority}") diff --git a/crates/core/src/admin/bucket_metadata.rs b/crates/core/src/admin/bucket_metadata.rs new file mode 100644 index 0000000..75abbb4 --- /dev/null +++ b/crates/core/src/admin/bucket_metadata.rs @@ -0,0 +1,60 @@ +//! Typed, bounded RustFS bucket-metadata archive operations. + +use async_trait::async_trait; +use zeroize::Zeroizing; + +use crate::Result; + +/// Runtime capability required by bucket-metadata archive commands. +pub const BUCKET_METADATA_CAPABILITY: &str = "admin.bucket-metadata"; + +/// RustFS server limit for both exported and imported bucket-metadata archives. +pub const MAX_BUCKET_METADATA_ARCHIVE_BYTES: usize = 100 * 1024 * 1024; + +/// An owned archive whose storage is cleared when dropped. +pub struct BucketMetadataArchive { + bytes: Zeroizing>, +} + +impl BucketMetadataArchive { + pub fn new(bytes: Vec) -> Result { + if bytes.len() > MAX_BUCKET_METADATA_ARCHIVE_BYTES { + return Err(crate::Error::RequestRejected(format!( + "Bucket metadata archive size {} exceeds the {} byte limit", + bytes.len(), + MAX_BUCKET_METADATA_ARCHIVE_BYTES + ))); + } + Ok(Self { + bytes: Zeroizing::new(bytes), + }) + } + + pub fn as_bytes(&self) -> &[u8] { + self.bytes.as_slice() + } + + pub fn into_bytes(self) -> Zeroizing> { + self.bytes + } +} + +#[async_trait] +pub trait BucketMetadataApi: Send + Sync { + /// Export every bucket, or one selected bucket, as a bounded ZIP archive. + async fn export_bucket_metadata(&self, bucket: Option<&str>) -> Result; + + /// Import a prevalidated bounded archive. Mutations are never retried automatically. + async fn import_bucket_metadata(&self, archive: BucketMetadataArchive) -> Result<()>; +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn archive_rejects_oversized_input() { + let bytes = vec![0; MAX_BUCKET_METADATA_ARCHIVE_BYTES + 1]; + assert!(BucketMetadataArchive::new(bytes).is_err()); + } +} diff --git a/crates/core/src/admin/mod.rs b/crates/core/src/admin/mod.rs index cf70ec3..832eee7 100644 --- a/crates/core/src/admin/mod.rs +++ b/crates/core/src/admin/mod.rs @@ -3,6 +3,7 @@ //! This module provides the AdminApi trait and types for managing //! IAM users, policies, groups, service accounts, and cluster operations. +mod bucket_metadata; mod capabilities; mod cluster; mod configuration; @@ -17,6 +18,10 @@ mod site; pub mod tier; mod types; +pub use bucket_metadata::{ + BUCKET_METADATA_CAPABILITY, BucketMetadataApi, BucketMetadataArchive, + MAX_BUCKET_METADATA_ARCHIVE_BYTES, +}; pub use capabilities::{ CapabilityAvailability, CapabilityEntry, CapabilityReport, ClusterSnapshotMetadata, ClusterSnapshotSummary, DiagnosticCapability, DiagnosticCapabilityGuardError, diff --git a/crates/s3/src/admin.rs b/crates/s3/src/admin.rs index 3a424d2..08b0c5a 100644 --- a/crates/s3/src/admin.rs +++ b/crates/s3/src/admin.rs @@ -14,20 +14,21 @@ use aws_sigv4::sign::v4; use bytes::Bytes; use futures::StreamExt; use rc_core::admin::{ - AccessKeyInfo, AdminApi, BucketQuota, CapabilityApi, CapabilityAvailability, CapabilityEntry, - CapabilityReport, ClusterInfo, ClusterSnapshotDocument, ClusterSnapshotMetadata, - ClusterSnapshotSummary, ConfigApi, ConfigDocument, ConfigHelp, ConfigHistoryEntry, - ConfigMutationResult, CreateServiceAccountRequest, DecommissionPoolStatus, DecommissionStatus, + AccessKeyInfo, AdminApi, BucketMetadataApi, BucketMetadataArchive, BucketQuota, CapabilityApi, + CapabilityAvailability, CapabilityEntry, CapabilityReport, ClusterInfo, + ClusterSnapshotDocument, ClusterSnapshotMetadata, ClusterSnapshotSummary, ConfigApi, + ConfigDocument, ConfigHelp, ConfigHistoryEntry, ConfigMutationResult, + CreateServiceAccountRequest, DecommissionPoolStatus, DecommissionStatus, DetailedHealthSnapshot, DiagnosticCapability, DiagnosticReadApi, ExtensionsCatalog, Group, GroupStatus, HealRuntimeState, HealScanMode, HealStartRequest, HealStatus, HealTaskRequest, IAM_POLICY_DETACH_CAPABILITY, IAM_POLICY_ENTITIES_CAPABILITY, IamMutationApi, IamReadApi, KmsApi, KmsBackendKind, KmsCacheSummary, KmsCancelKeyDeletionResult, KmsConfigSummary, KmsConfigureRequest, KmsCreateKeyRequest, KmsCreateKeyResult, KmsDeleteKeyRequest, KmsDeleteKeyResult, KmsKey, KmsKeyPage, KmsKeyState, KmsKeyUsage, KmsServiceState, KmsStatus, - MAX_DIAGNOSTIC_RESPONSE_BYTES, MAX_IAM_POLICY_DETACH_REQUEST_BYTES, - MAX_IAM_POLICY_DETACH_RESPONSE_BYTES, MAX_IAM_POLICY_ENTITIES_RESPONSE_BYTES, - MAX_METRICS_LINE_BYTES, MAX_METRICS_RESPONSE_BYTES, MAX_METRICS_SAMPLES, - MAX_OIDC_RESPONSE_BYTES, MAX_REPLICATION_DIFF_RESPONSE_BYTES, + MAX_BUCKET_METADATA_ARCHIVE_BYTES, MAX_DIAGNOSTIC_RESPONSE_BYTES, + MAX_IAM_POLICY_DETACH_REQUEST_BYTES, MAX_IAM_POLICY_DETACH_RESPONSE_BYTES, + MAX_IAM_POLICY_ENTITIES_RESPONSE_BYTES, MAX_METRICS_LINE_BYTES, MAX_METRICS_RESPONSE_BYTES, + MAX_METRICS_SAMPLES, MAX_OIDC_RESPONSE_BYTES, MAX_REPLICATION_DIFF_RESPONSE_BYTES, MAX_SITE_REPLICATION_ERROR_RESPONSE_BYTES, MAX_SITE_REPLICATION_REQUEST_BYTES, MAX_SITE_REPLICATION_SUCCESS_RESPONSE_BYTES, ManualTransitionRunRequest, ManualTransitionRunResponse, MetricsBatch, MetricsQuery, ModuleSwitches, ObservabilityApi, @@ -102,6 +103,7 @@ static CAPABILITY_CACHE: OnceLock &'static Mutex> { CAPABILITY_CACHE.get_or_init(|| Mutex::new(HashMap::new())) @@ -3060,6 +3062,111 @@ fn normalize_policy_detach_response_names(kind: &str, names: &mut Vec) - Ok(()) } +#[async_trait] +impl BucketMetadataApi for AdminClient { + async fn export_bucket_metadata(&self, bucket: Option<&str>) -> Result { + let mut url = self.admin_url("/export-bucket-metadata"); + if let Some(bucket) = bucket { + url.push_str("?bucket="); + url.push_str(&urlencoding::encode(bucket)); + } + let headers = self.request_headers(&[])?; + let signed_headers = self.sign_request(&Method::GET, &url, &headers, &[]).await?; + let mut request = self.http_client.get(&url); + for (name, value) in &signed_headers { + request = request.header(name, value); + } + let response = request + .send() + .await + .map_err(|_| Error::Network("Bucket metadata export request failed".to_string()))?; + let status = response.status(); + let bytes = read_bounded_response_body( + response, + MAX_BUCKET_METADATA_ARCHIVE_BYTES, + "Bucket metadata export response", + ) + .await?; + if !status.is_success() { + return Err(bucket_metadata_status_error(status, false)); + } + BucketMetadataArchive::new(bytes) + } + + async fn import_bucket_metadata(&self, archive: BucketMetadataArchive) -> Result<()> { + let body = archive.into_bytes(); + let url = self.admin_url("/import-bucket-metadata"); + let mut headers = self.request_headers(body.as_slice())?; + headers.insert(CONTENT_TYPE, HeaderValue::from_static("application/zip")); + let signed_headers = self + .sign_request(&Method::PUT, &url, &headers, body.as_slice()) + .await?; + let mut request = self.http_client.put(&url); + for (name, value) in &signed_headers { + request = request.header(name, value); + } + let response = request + .body(Bytes::from_owner(SensitiveRequestBody(body))) + .send() + .await + .map_err(|_| { + Error::Network( + "Bucket metadata import outcome is unknown; the mutation was not retried" + .to_string(), + ) + })?; + let status = response.status(); + let response_body = read_bounded_response_body( + response, + MAX_BUCKET_METADATA_MUTATION_RESPONSE_BYTES, + "Bucket metadata import response", + ) + .await + .map_err(|_| { + Error::Network( + "Bucket metadata import outcome is unknown; inspect every selected bucket before retrying" + .to_string(), + ) + })?; + if !status.is_success() { + return Err(bucket_metadata_status_error(status, true)); + } + if !response_body.is_empty() + && serde_json::from_slice::(&response_body).is_err() + { + return Err(Error::Network( + "Bucket metadata import returned an unexpected response; inspect every selected bucket before retrying" + .to_string(), + )); + } + Ok(()) + } +} + +fn bucket_metadata_status_error(status: StatusCode, mutation: bool) -> Error { + match status { + StatusCode::UNAUTHORIZED | StatusCode::FORBIDDEN => { + Error::Auth("Bucket metadata administration permission was denied".to_string()) + } + StatusCode::NOT_FOUND | StatusCode::METHOD_NOT_ALLOWED | StatusCode::NOT_IMPLEMENTED => { + Error::UnsupportedFeature( + "The RustFS bucket metadata archive route is unavailable".to_string(), + ) + } + StatusCode::CONFLICT | StatusCode::PRECONDITION_FAILED => { + Error::Conflict("Bucket metadata changed concurrently".to_string()) + } + StatusCode::BAD_REQUEST | StatusCode::UNPROCESSABLE_ENTITY => { + Error::Config("RustFS rejected the bucket metadata archive".to_string()) + } + _ if mutation => Error::Network( + "Bucket metadata import may be partially applied; inspect every selected bucket before retrying" + .to_string(), + ), + _ => Error::Network("Bucket metadata export service is unavailable".to_string()), + } +} + #[async_trait] impl DiagnosticReadApi for AdminClient { async fn health_snapshot(&self) -> Result { diff --git a/docs/reference/rc/admin.md b/docs/reference/rc/admin.md index 6cef3af..2fa946f 100644 --- a/docs/reference/rc/admin.md +++ b/docs/reference/rc/admin.md @@ -49,6 +49,7 @@ rc admin service rc admin diagnostics client-devnull [--size ] [--timeout ] [--concurrency ] --yes rc admin config ... rc admin config module-switch ... +rc admin bucket-metadata ... rc admin replicate add [...] rc admin replicate [OPTIONS] rc admin replicate edit --site [EDIT OPTIONS] --yes @@ -78,6 +79,7 @@ rc admin replicate remove <--all|--site > | `service-account` | Manage service accounts. | | `service` | Control the server process: restart, stop, freeze, unfreeze. | | `config` | Inspect, plan, export, and mutate RustFS server configuration. | +| `bucket-metadata` | Export or import validated per-bucket configuration archives. | | `replicate` | Manage site replication across clusters. | ## Examples @@ -525,6 +527,23 @@ Exports are always redacted because secret values are not a portable client outp Value files are limited to 1 MiB of single-line UTF-8 text. On Unix they must be regular, non-symlink files without group or other permissions; trailing line endings are removed. If server help metadata is incomplete or unavailable, `set`, `delete`, and `import` emit a warning and defer final validation to the mutation endpoint. +## Bucket Metadata Archive Workflow + +`rc admin bucket-metadata` exports and imports the per-bucket configuration families supported by the RustFS v3 archive routes: policy, notification, lifecycle, encryption, tagging, quota, object lock, versioning, replication, and replication targets. + +| Command | Description | +| --- | --- | +| `rc admin bucket-metadata export --file [--bucket ...] [--force]` | Export all buckets or an explicit selection to a deterministic, owner-private ZIP. The file is created atomically; `--force` is required to replace an existing path. | +| `rc admin bucket-metadata import --file [--bucket ...] --conflict [--dry-run] [--yes]` | Validate a protected ZIP, compare it with current metadata, and apply the explicit conflict policy. Mutating imports require `--yes`; dry runs never send a mutation. | + +Archives are bounded to 100 MiB compressed, 128 MiB expanded, 4,096 entries, and 16 MiB per entry. Entry paths must be exactly `/`; duplicate, empty, nested, traversal, and unknown entries fail before mutation. On Unix, file input must be a regular non-symlink file with no group or other permissions. Standard input is available through `--file -` for an explicitly protected pipeline. + +Server exports redact replication-target credentials. The client never prints archive contents and rejects an import whose target metadata contains a missing or redacted secret, because importing it would overwrite a working credential. Supply real target credentials only through a protected archive or standard input. + +`--conflict fail` stops before mutation if any imported config differs from current state. `overwrite` sends differing configs, while `skip` removes only conflicting entries from the outgoing archive. Identical entries are never resent. Missing destination buckets remain eligible for the server's create-on-import behavior, while a selected bucket absent from the source archive is a distinct not-found error. + +An import is one bounded PUT and is never automatically retried. A transport failure or server error can mean that only part of the archive was applied; inspect every selected bucket before deciding whether to retry. Successful JSON output uses output schema v3 with one `admin_operations` result per selected bucket. + ## Site Replication Workflow `rc admin replicate` manages multi-cluster site replication. Peer sites are given as configured alias names; their endpoints and credentials are resolved from the local alias store, so every participating site needs an alias with root credentials before running `add`.