diff --git a/Cargo.lock b/Cargo.lock index f2e8546..93efc62 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -8,6 +8,41 @@ version = "2.0.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "320119579fcad9c21884f5c4861d16174d0e06250625266f50fe6898340abefa" +[[package]] +name = "aead" +version = "0.5.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "d122413f284cf2d62fb1b7db97e02edb8cda96d769b16e443a4f6195e35662b0" +dependencies = [ + "crypto-common 0.1.7", + "generic-array", +] + +[[package]] +name = "aes" +version = "0.8.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "b169f7a6d4742236a0a00c541b845991d0ac43e546831af1249753ab4c3aa3a0" +dependencies = [ + "cfg-if", + "cipher", + "cpufeatures 0.2.17", +] + +[[package]] +name = "aes-gcm" +version = "0.10.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "831010a0f742e1209b3bcea8fab6a8e149051ba6099432c8cb2cc117dec3ead1" +dependencies = [ + "aead", + "aes", + "cipher", + "ctr", + "ghash", + "subtle", +] + [[package]] name = "ahash" version = "0.8.12" @@ -873,6 +908,16 @@ dependencies = [ "half", ] +[[package]] +name = "cipher" +version = "0.4.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "773f3b9af64447d2ce9850330c473515014aa235e6a783b02db81ff39e4a3dad" +dependencies = [ + "crypto-common 0.1.7", + "inout", +] + [[package]] name = "clap" version = "4.6.1" @@ -1019,7 +1064,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "e75b2483e97a5a7da73ac68a05b629f9c53cff58d8ed1c77866079e18b00dba5" dependencies = [ "digest 0.10.7", - "spin", + "spin 0.10.0", ] [[package]] @@ -1079,6 +1124,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "78c8292055d1c1df0cce5d180393dc8cce0abec0a7102adb6c7b1eef6016d60a" dependencies = [ "generic-array", + "rand_core 0.6.4", "typenum", ] @@ -1091,6 +1137,15 @@ dependencies = [ "hybrid-array", ] +[[package]] +name = "ctr" +version = "0.9.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "0369ee1ad671834580515889b80f2ea915f23b8be8d0daa4bbaf2ac5c7590835" +dependencies = [ + "cipher", +] + [[package]] name = "ctutils" version = "0.4.2" @@ -1349,6 +1404,16 @@ dependencies = [ "subtle", ] +[[package]] +name = "filetime" +version = "0.2.29" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "5c287a33c7f0a620c38e641e7f60827713987b3c0f26e8ddc9462cc69cf75759" +dependencies = [ + "cfg-if", + "libc", +] + [[package]] name = "find-msvc-tools" version = "0.1.9" @@ -1562,6 +1627,16 @@ dependencies = [ "wasm-bindgen", ] +[[package]] +name = "ghash" +version = "0.5.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "f0d8a4362ccb29cb0b265253fb0a2728f592895ee6854fd9bc13f2ffda266ff1" +dependencies = [ + "opaque-debug", + "polyval", +] + [[package]] name = "glob" version = "0.3.3" @@ -1935,6 +2010,15 @@ dependencies = [ "web-time", ] +[[package]] +name = "inout" +version = "0.1.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "879f10e63c20629ecabbb64a8010319738c66a5cd0c29b02d63d272b03751d01" +dependencies = [ + "generic-array", +] + [[package]] name = "insta" version = "1.48.0" @@ -2071,6 +2155,9 @@ name = "lazy_static" version = "1.5.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "bbd2bcb4c963f2ddae06a2efc7e9f3591312473c50c6685e1f298068316e66fe" +dependencies = [ + "spin 0.9.9", +] [[package]] name = "libc" @@ -2294,6 +2381,22 @@ dependencies = [ "num-traits", ] +[[package]] +name = "num-bigint-dig" +version = "0.8.6" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "e661dda6640fad38e827a6d4a310ff4763082116fe217f279885c97f511bb0b7" +dependencies = [ + "lazy_static", + "libm", + "num-integer", + "num-iter", + "num-traits", + "rand 0.8.7", + "smallvec", + "zeroize", +] + [[package]] name = "num-cmp" version = "0.1.0" @@ -2352,6 +2455,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "071dfc062690e90b734c0b2273ce72ad0ffa95f0c74596bc250dcfd960262841" dependencies = [ "autocfg", + "libm", ] [[package]] @@ -2375,6 +2479,12 @@ version = "1.70.2" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "384b8ab6d37215f3c5301a95a4accb5d64aa607f1fcb26a11b5303878451b4fe" +[[package]] +name = "opaque-debug" +version = "0.3.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "c08d65885ee38876c4f86fa503fb49d7b507c2b62552df7c70b2fce627e06381" + [[package]] name = "openssl-probe" version = "0.2.1" @@ -2455,6 +2565,17 @@ version = "0.1.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "8b870d8c151b6f2fb93e84a13146138f05d02ed11c7e7c54f8826aaaf7c9f184" +[[package]] +name = "pkcs1" +version = "0.7.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "c8ffb9f10fa047879315e6625af03c164b16962a5368d724ed16323b68ace47f" +dependencies = [ + "der", + "pkcs8", + "spki", +] + [[package]] name = "pkcs8" version = "0.10.2" @@ -2471,6 +2592,18 @@ version = "0.3.33" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "19f132c84eca552bf34cab8ec81f1c1dcc229b811638f9d283dceabe58c5569e" +[[package]] +name = "polyval" +version = "0.6.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "9d1fe60d06143b2430aa532c94cfe9e29783047f06c0d7fd359a9a51b729fa25" +dependencies = [ + "cfg-if", + "cpufeatures 0.2.17", + "opaque-debug", + "universal-hash", +] + [[package]] name = "portable-atomic" version = "1.13.1" @@ -2501,6 +2634,15 @@ version = "0.2.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "439ee305def115ba05938db6eb1644ff94165c5ab5e9420d1c1bcedbba909391" +[[package]] +name = "ppv-lite86" +version = "0.2.21" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "85eae3c4ed2f50dcfe72643da4befc30deadb458a9b590d720cde2f2b1e97da9" +dependencies = [ + "zerocopy", +] + [[package]] name = "predicates" version = "3.1.4" @@ -2594,7 +2736,7 @@ dependencies = [ "bytes", "getrandom 0.4.3", "lru-slab", - "rand", + "rand 0.10.2", "rand_pcg", "ring", "rustc-hash", @@ -2642,6 +2784,17 @@ version = "6.0.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "f8dcc9c7d52a811697d2151c701e0d08956f92b0e24136cf4cf27b57a6a0d9bf" +[[package]] +name = "rand" +version = "0.8.7" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "22f6172bdec972074665ed81ed53b71da00bfc44b65a753cfde883ec4c702a1a" +dependencies = [ + "libc", + "rand_chacha", + "rand_core 0.6.4", +] + [[package]] name = "rand" version = "0.10.2" @@ -2653,6 +2806,16 @@ dependencies = [ "rand_core 0.10.1", ] +[[package]] +name = "rand_chacha" +version = "0.3.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "e6c10a63a0fa32252be49d21e7709d4d4baf8d231c2dbce1eaa8141b9b127d88" +dependencies = [ + "ppv-lite86", + "rand_core 0.6.4", +] + [[package]] name = "rand_core" version = "0.6.4" @@ -2681,6 +2844,7 @@ dependencies = [ name = "rc-core" version = "0.1.24" dependencies = [ + "aes-gcm", "anyhow", "async-trait", "base64", @@ -2688,13 +2852,18 @@ dependencies = [ "futures", "getrandom 0.4.3", "glob", + "hex", "http 1.4.2", "humansize", "jiff", "mockall", + "rand 0.8.7", + "rsa", "rustls-pki-types", "serde", "serde_json", + "sha2 0.10.9", + "tar", "tempfile", "thiserror", "tokio", @@ -2912,6 +3081,27 @@ dependencies = [ "xmlparser", ] +[[package]] +name = "rsa" +version = "0.9.10" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "b8573f03f5883dcaebdfcf4725caa1ecb9c15b2ef50c43a07b816e06799bb12d" +dependencies = [ + "const-oid 0.9.6", + "digest 0.10.7", + "num-bigint-dig", + "num-integer", + "num-traits", + "pkcs1", + "pkcs8", + "rand_core 0.6.4", + "sha2 0.10.9", + "signature", + "spki", + "subtle", + "zeroize", +] + [[package]] name = "rustc-hash" version = "2.1.3" @@ -3294,6 +3484,12 @@ dependencies = [ "windows-sys 0.61.2", ] +[[package]] +name = "spin" +version = "0.9.9" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "3763264f6b73151db08c50ff20d7d8a0b8796e021cdea7ceedad07b80155fa0e" + [[package]] name = "spin" version = "0.10.0" @@ -3391,6 +3587,17 @@ dependencies = [ "syn 2.0.118", ] +[[package]] +name = "tar" +version = "0.4.46" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "3f6221d9a6003c78398e3b239969f352578258df48c8eb051caadae0015bc840" +dependencies = [ + "filetime", + "libc", + "xattr", +] + [[package]] name = "tempfile" version = "3.27.0" @@ -3744,6 +3951,16 @@ version = "0.5.2" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "81e544489bf3d8ef66c953931f56617f423cd4b5494be343d9b9d3dda037b9a3" +[[package]] +name = "universal-hash" +version = "0.5.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "fc1de2c688dc15305988b563c3854064043356019f97a4b46276fe734c4f07ea" +dependencies = [ + "crypto-common 0.1.7", + "subtle", +] + [[package]] name = "untrusted" version = "0.9.0" @@ -4090,6 +4307,16 @@ dependencies = [ "time", ] +[[package]] +name = "xattr" +version = "1.6.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "32e45ad4206f6d2479085147f02bc2ef834ac85886624a23575ae137c8aa8156" +dependencies = [ + "libc", + "rustix", +] + [[package]] name = "xmlparser" version = "0.13.6" diff --git a/Cargo.toml b/Cargo.toml index 53a150d..377b0d9 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -68,6 +68,10 @@ rustls-pki-types = "1.15" x509-parser = "0.18" zeroize = "1.8" getrandom = "0.4" +aes-gcm = "0.10" +rsa = { version = "0.9", features = ["sha2", "pem"] } +rand = "0.8" +tar = "0.4" zip = { version = "8.6", default-features = false, features = ["deflate"] } # HTTP client for Admin API diff --git a/crates/cli/src/commands/admin/diagnostics.rs b/crates/cli/src/commands/admin/diagnostics.rs index 5f4db3a..25abcbc 100644 --- a/crates/cli/src/commands/admin/diagnostics.rs +++ b/crates/cli/src/commands/admin/diagnostics.rs @@ -1,6 +1,7 @@ //! RustFS read-only snapshots and explicitly confirmed bounded active diagnostics. mod client_devnull; +mod inspect_archive; use std::collections::BTreeMap; @@ -18,6 +19,7 @@ use crate::exit_code::ExitCode; use crate::output::Formatter; pub use client_devnull::ClientDevnullArgs; +pub use inspect_archive::InspectArchiveArgs; #[derive(Subcommand, Debug, Clone)] pub enum DiagnosticsCommands { @@ -30,6 +32,9 @@ pub enum DiagnosticsCommands { /// Measure bounded client-to-server upload throughput without storing data #[command(name = "client-devnull")] ClientDevnull(ClientDevnullArgs), + /// Retrieve and verify an encrypted drive-metadata archive + #[command(name = "inspect-archive")] + InspectArchive(InspectArchiveArgs), } #[derive(clap::Args, Debug, Clone)] @@ -46,6 +51,7 @@ impl DiagnosticsCommands { Self::Cluster(_) => "cluster", Self::Extensions(_) => "extensions", Self::ClientDevnull(_) => "client-devnull", + Self::InspectArchive(_) => "inspect-archive", } } @@ -53,6 +59,10 @@ impl DiagnosticsCommands { match self { Self::Health(args) | Self::Cluster(args) | Self::Extensions(args) => &args.alias, Self::ClientDevnull(args) => &args.alias, + Self::InspectArchive(args) => args + .target + .split_once('/') + .map_or(args.target.as_str(), |(alias, _)| alias), } } @@ -62,6 +72,7 @@ impl DiagnosticsCommands { Self::Cluster(_) => DiagnosticCapability::ClusterSnapshot, Self::Extensions(_) => DiagnosticCapability::ExtensionsCatalog, Self::ClientDevnull(_) => DiagnosticCapability::ClientDevnull, + Self::InspectArchive(_) => DiagnosticCapability::InspectArchive, } } @@ -71,6 +82,7 @@ impl DiagnosticsCommands { Self::Cluster(_) => "cluster_snapshot", Self::Extensions(_) => "extension_catalog", Self::ClientDevnull(_) => "admin_operations", + Self::InspectArchive(_) => "diagnostic_archive", } } } @@ -120,6 +132,9 @@ pub async fn execute(command: DiagnosticsCommands, formatter: &Formatter) -> Exi DiagnosticsCommands::ClientDevnull(args) => { client_devnull::execute_client_devnull(args, formatter).await } + DiagnosticsCommands::InspectArchive(args) => { + inspect_archive::execute_inspect_archive(args, formatter).await + } read_command => { let client = match get_admin_client(read_command.alias(), formatter) { Ok(client) => client, @@ -196,6 +211,12 @@ async fn execute_with_api( ), formatter, ), + DiagnosticsCommands::InspectArchive(_) => emit_diagnostic_error( + &command, + "Failed to route diagnostic archive retrieval", + &Error::General("Inspect archive must use the encrypted archive executor".to_string()), + formatter, + ), } } diff --git a/crates/cli/src/commands/admin/diagnostics/inspect_archive.rs b/crates/cli/src/commands/admin/diagnostics/inspect_archive.rs new file mode 100644 index 0000000..f028983 --- /dev/null +++ b/crates/cli/src/commands/admin/diagnostics/inspect_archive.rs @@ -0,0 +1,721 @@ +//! Retrieve and locally verify an encrypted RustFS diagnostic archive. + +use std::future::Future; +use std::io::Read; +use std::path::{Component, Path, PathBuf}; + +use clap::Args; +use rc_core::admin::{ + CapabilityApi, CapabilityAvailability, DiagnosticCapability, InspectArchiveApi, + InspectArchiveCancellation, InspectArchiveKey, InspectArchiveTransportRequest, + PublishedInspectArchive, decrypt_and_validate_inspect_archive_with_cancel, + publish_inspect_archive, validate_inspect_archive_output_directory, +}; +use rc_core::{Error, parse_object_path}; +use serde::Serialize; +use zeroize::Zeroizing; + +use super::super::get_admin_client; +use crate::exit_code::ExitCode; +use crate::output::Formatter; + +const MAX_PRIVATE_KEY_BYTES: u64 = 64 * 1024; + +/// Options for retrieving one encrypted diagnostic metadata archive. +#[derive(Args, Debug, Clone)] +pub struct InspectArchiveArgs { + /// Exact RustFS object target (ALIAS/BUCKET/OBJECT) + pub target: String, + + /// Destination for the verified plaintext tar archive + #[arg(short, long)] + pub output: PathBuf, + + /// Read a caller-managed PKCS#8 RSA private key from a protected file + #[arg(long, value_name = "FILE")] + pub private_key: Option, +} + +pub(super) async fn execute_inspect_archive( + args: InspectArchiveArgs, + formatter: &Formatter, +) -> ExitCode { + let target = match validate_args(&args) { + Ok(target) => target, + Err(error) => return emit_archive_error(&error, None, formatter), + }; + let client = match get_admin_client(&target.alias, formatter) { + Ok(client) => client, + Err(code) => return code, + }; + run_inspect_archive( + args, + target.bucket, + target.key, + &client, + &client, + formatter, + tokio::signal::ctrl_c(), + ) + .await +} + +fn validate_args(args: &InspectArchiveArgs) -> rc_core::Result { + let target = parse_object_path(&args.target)?; + if target + .key + .split('/') + .any(|component| component.is_empty() || matches!(component, "." | "..")) + { + return Err(Error::InvalidPath( + "Diagnostic archive object contains an unsafe path component".to_string(), + )); + } + if args.output.as_os_str().is_empty() { + return Err(Error::InvalidPath( + "Diagnostic archive output path cannot be empty".to_string(), + )); + } + if args + .output + .components() + .any(|component| matches!(component, Component::ParentDir)) + { + return Err(Error::InvalidPath( + "Diagnostic archive output cannot contain parent-directory components".to_string(), + )); + } + if args.output.exists() { + return Err(Error::Conflict( + "Diagnostic archive output already exists".to_string(), + )); + } + let parent = output_parent(&args.output); + validate_inspect_archive_output_directory(parent)?; + Ok(target) +} + +fn output_parent(path: &Path) -> &Path { + path.parent() + .filter(|parent| !parent.as_os_str().is_empty()) + .unwrap_or_else(|| Path::new(".")) +} + +async fn run_inspect_archive( + args: InspectArchiveArgs, + bucket: String, + object: String, + capabilities: &dyn CapabilityApi, + api: &dyn InspectArchiveApi, + formatter: &Formatter, + interrupt: F, +) -> ExitCode +where + F: Future>, +{ + let report = match capabilities.discover_capabilities(false).await { + Ok(report) => report, + Err(error) => return emit_archive_error(&error, None, formatter), + }; + if let Err(error) = report.require_diagnostic_capability(DiagnosticCapability::InspectArchive) { + let (mapped, kind) = match error.availability() { + CapabilityAvailability::PermissionDenied => ( + Error::Auth("Permission denied during capability discovery".to_string()), + Some("auth_error"), + ), + CapabilityAvailability::Disabled => ( + Error::RequestRejected( + error + .reason() + .unwrap_or("Diagnostic archive capability is disabled") + .to_string(), + ), + Some("disabled"), + ), + _ => ( + Error::UnsupportedFeature(error.to_string()), + Some("unsupported_feature"), + ), + }; + return emit_archive_error(&mapped, kind, formatter); + } + let contract = match api.inspect_archive_capability().await { + Ok(contract) => contract, + Err(error) => return emit_archive_error(&error, None, formatter), + }; + if let Err(error) = contract.validate() { + let kind = (contract.state.availability() == CapabilityAvailability::Disabled) + .then_some("disabled"); + return emit_archive_error(&error, kind, formatter); + } + + let private_key_path = args.private_key.clone(); + let key = match tokio::task::spawn_blocking(move || { + if let Some(path) = private_key_path { + let pem = read_private_key(&path)?; + InspectArchiveKey::from_pkcs8_pem(&pem) + } else { + InspectArchiveKey::generate() + } + }) + .await + { + Ok(Ok(key)) => key, + Ok(Err(error)) => return emit_archive_error(&error, None, formatter), + Err(_) => { + return emit_archive_error( + &Error::General("Diagnostic archive key preparation failed".to_string()), + None, + formatter, + ); + } + }; + + let temporary_directory = output_parent(&args.output).to_path_buf(); + let request = InspectArchiveTransportRequest { + bucket, + object, + public_key_pem: key.public_key_pem().to_string(), + max_bytes: contract.max_bytes, + timeout: contract.timeout(), + }; + tokio::pin!(interrupt); + let encrypted = tokio::select! { + biased; + interrupt_result = &mut interrupt => { + return interrupt_result.map_or_else( + |_| emit_archive_error( + &Error::General("Failed to register diagnostic archive interruption handler".to_string()), + None, + formatter, + ), + |_| emit_archive_error( + &Error::Interrupted("Diagnostic archive retrieval was cancelled".to_string()), + Some("interrupted"), + formatter, + ), + ); + } + result = api.download_inspect_archive(request, &temporary_directory) => { + match result { + Ok(encrypted) => encrypted, + Err(error) => return emit_archive_error(&error, None, formatter), + } + } + }; + + let cancellation = InspectArchiveCancellation::default(); + let blocking_cancellation = cancellation.clone(); + let maximum = contract.max_bytes; + let maximum_metadata = contract.max_metadata_bytes_per_drive; + let directory = temporary_directory.clone(); + let verification = tokio::task::spawn_blocking(move || { + decrypt_and_validate_inspect_archive_with_cancel( + encrypted, + &key, + &directory, + maximum, + maximum_metadata, + maximum, + &blocking_cancellation, + ) + }); + tokio::pin!(verification); + let verified = tokio::select! { + biased; + interrupt_result = &mut interrupt => { + cancellation.cancel(); + let _ = verification.await; + return interrupt_result.map_or_else( + |_| emit_archive_error( + &Error::General("Failed to monitor diagnostic archive interruption".to_string()), + None, + formatter, + ), + |_| emit_archive_error( + &Error::Interrupted("Diagnostic archive verification was cancelled".to_string()), + Some("interrupted"), + formatter, + ), + ); + } + result = &mut verification => { + match result { + Ok(Ok(verified)) => verified, + Ok(Err(error)) => return emit_archive_error(&error, None, formatter), + Err(_) => return emit_archive_error( + &Error::General("Diagnostic archive verification task failed".to_string()), + Some("cryptographic_verification"), + formatter, + ), + } + } + }; + + match publish_inspect_archive(verified, &args.output) { + Ok(published) => { + emit_archive_success(published, formatter); + ExitCode::Success + } + Err(error) => emit_archive_error(&error, Some("local_io"), formatter), + } +} + +fn read_private_key(path: &Path) -> rc_core::Result> { + let metadata = std::fs::symlink_metadata(path).map_err(|_| { + Error::InvalidPath("Failed to inspect diagnostic archive private-key file".to_string()) + })?; + if metadata.file_type().is_symlink() || !metadata.is_file() { + return Err(Error::InvalidPath( + "Diagnostic archive private key must be a regular file, not a symlink".to_string(), + )); + } + #[cfg(unix)] + { + use std::os::unix::fs::PermissionsExt; + if metadata.permissions().mode() & 0o077 != 0 { + return Err(Error::InvalidPath( + "Diagnostic archive private key cannot grant group or other permissions" + .to_string(), + )); + } + } + let file = std::fs::File::open(path).map_err(|_| { + Error::InvalidPath("Failed to open diagnostic archive private-key file".to_string()) + })?; + let mut bytes = Zeroizing::new(Vec::new()); + file.take(MAX_PRIVATE_KEY_BYTES + 1) + .read_to_end(&mut bytes) + .map_err(|_| { + Error::InvalidPath("Failed to read diagnostic archive private-key file".to_string()) + })?; + if bytes.len() as u64 > MAX_PRIVATE_KEY_BYTES { + return Err(Error::InvalidPath( + "Diagnostic archive private-key file exceeds the 64 KiB limit".to_string(), + )); + } + std::str::from_utf8(&bytes) + .map(|pem| Zeroizing::new(pem.to_string())) + .map_err(|_| { + Error::InvalidPath( + "Diagnostic archive private-key file must contain UTF-8 PEM".to_string(), + ) + }) +} + +#[derive(Debug, Serialize)] +struct ArchiveSuccessOutput { + schema_version: u8, + #[serde(rename = "type")] + output_type: &'static str, + status: &'static str, + data: ArchiveSuccessData, +} + +#[derive(Debug, Serialize)] +struct ArchiveSuccessData { + operation: &'static str, + state: &'static str, + output: PathBuf, + archive_version: u16, + drive_count: usize, + encrypted_bytes: u64, + plaintext_bytes: u64, + plaintext_sha256: String, +} + +fn emit_archive_success(published: PublishedInspectArchive, formatter: &Formatter) { + if formatter.is_json() { + formatter.json(&ArchiveSuccessOutput { + schema_version: 3, + output_type: "diagnostic_archive", + status: "success", + data: ArchiveSuccessData { + operation: "retrieve", + state: "verified", + output: published.path, + archive_version: published.archive_version, + drive_count: published.drive_count, + encrypted_bytes: published.encrypted_bytes, + plaintext_bytes: published.plaintext_bytes, + plaintext_sha256: published.plaintext_sha256, + }, + }); + } else { + formatter.success(&format!( + "Verified diagnostic archive written to '{}' (version {}, {} drive artifacts, {} bytes)", + formatter.sanitize_text(&published.path.display().to_string()), + published.archive_version, + published.drive_count, + published.plaintext_bytes + )); + } +} + +#[derive(Debug, Serialize)] +struct ArchiveErrorOutput { + schema_version: u8, + #[serde(rename = "type")] + output_type: &'static str, + status: &'static str, + error: ArchiveErrorBody, +} + +#[derive(Debug, Serialize)] +struct ArchiveErrorBody { + #[serde(rename = "type")] + error_type: &'static str, + message: String, + retryable: bool, + capability: Option<&'static str>, + server: Option<&'static str>, + suggestion: Option<&'static str>, +} + +fn emit_archive_error( + error: &Error, + override_kind: Option<&'static str>, + formatter: &Formatter, +) -> ExitCode { + let code = ExitCode::from_i32(error.exit_code()).unwrap_or(ExitCode::GeneralError); + let error_type = override_kind.unwrap_or_else(|| archive_error_type(error)); + let message = format!("Diagnostic archive operation failed: {error}"); + if formatter.is_json() { + formatter.json_error(&ArchiveErrorOutput { + schema_version: 3, + output_type: "diagnostic_archive", + status: "error", + error: ArchiveErrorBody { + error_type, + message, + retryable: matches!(error, Error::Network(_) | Error::Interrupted(_)), + capability: matches!(error, Error::UnsupportedFeature(_)) + .then_some(rc_core::admin::INSPECT_ARCHIVE_CAPABILITY), + server: matches!(error, Error::UnsupportedFeature(_)).then_some("rustfs"), + suggestion: archive_error_suggestion(error_type), + }, + }); + } else { + formatter.error_with_code(code, &message); + } + code +} + +fn archive_error_type(error: &Error) -> &'static str { + match error { + Error::InvalidPath(message) + if message.contains("staging") + || message.contains("publish") + || message.contains("sync") => + { + "local_io" + } + Error::InvalidPath(_) | Error::Config(_) => "invalid_target", + Error::Network(_) => "network_error", + Error::Auth(_) => "auth_error", + Error::NotFound(_) | Error::AliasNotFound(_) => "not_found", + Error::Conflict(_) | Error::AliasExists(_) => "conflict", + Error::UnsupportedFeature(_) => "unsupported_feature", + Error::RequestRejected(_) => "server_limit", + Error::Interrupted(_) => "interrupted", + Error::General(message) + if message.starts_with("Diagnostic archive verification failed") => + { + "cryptographic_verification" + } + Error::Io(_) => "local_io", + _ => "general_error", + } +} + +fn archive_error_suggestion(kind: &str) -> Option<&'static str> { + match kind { + "invalid_target" => Some("Verify the exact object target, output path, and private key."), + "auth_error" => Some("Verify credentials include the InspectData admin permission."), + "unsupported_feature" => { + Some("Upgrade RustFS to a release with encrypted inspect archives.") + } + "disabled" => Some("Initialize local RustFS drives before requesting an archive."), + "network_error" => Some("Verify connectivity and retry; incomplete files were removed."), + "server_limit" => { + Some("Retry after server capacity is available or reduce target metadata.") + } + "cryptographic_verification" => Some("Do not use the output; retrieve a fresh archive."), + "local_io" => Some("Verify destination permissions and available disk space."), + "interrupted" => Some("Retry when ready; incomplete files were removed."), + _ => None, + } +} + +#[cfg(test)] +mod tests { + use std::collections::BTreeMap; + use std::future::{pending, ready}; + use std::sync::Arc; + use std::sync::atomic::{AtomicBool, AtomicUsize, Ordering}; + + use async_trait::async_trait; + use jsonschema::Validator; + use rc_core::Result; + use rc_core::admin::{ + CapabilityEntry, CapabilityReport, ClusterSnapshotMetadata, EncryptedInspectArchive, + INSPECT_ARCHIVE_COMPLETION, INSPECT_ARCHIVE_CONTENT_TYPE, INSPECT_ARCHIVE_ENCRYPTION, + INSPECT_ARCHIVE_ROUTE, InspectArchiveCapabilityContract, RuntimeCapabilityState, + RuntimeCapabilityStatus, + }; + use serde_json::Value; + use tempfile::tempdir; + + use super::*; + use crate::output::OutputConfig; + + struct DropMarker(Arc); + + impl Drop for DropMarker { + fn drop(&mut self) { + self.0.store(true, Ordering::SeqCst); + } + } + + struct StubArchiveApi { + availability: CapabilityAvailability, + downloads: AtomicUsize, + dropped: Arc, + } + + #[async_trait] + impl CapabilityApi for StubArchiveApi { + async fn discover_capabilities(&self, _refresh: bool) -> Result { + Ok(CapabilityReport { + server_version: Some("test".to_string()), + runtime_path: "/rustfs/admin/v4/runtime/capabilities".to_string(), + extensions_path: "/rustfs/admin/v4/extensions/catalog".to_string(), + cluster_snapshot_path: "/rustfs/admin/v4/cluster/snapshot".to_string(), + capabilities: vec![CapabilityEntry { + name: DiagnosticCapability::InspectArchive.name().to_string(), + availability: self.availability, + reason: None, + }], + extensions: Vec::new(), + cluster: ClusterSnapshotMetadata { + summary: None, + runtime_capabilities_path: None, + extensions_catalog_path: None, + }, + }) + } + } + + #[async_trait] + impl InspectArchiveApi for StubArchiveApi { + async fn inspect_archive_capability(&self) -> Result { + Ok(contract()) + } + + async fn download_inspect_archive( + &self, + _request: InspectArchiveTransportRequest, + _temporary_directory: &Path, + ) -> Result { + self.downloads.fetch_add(1, Ordering::SeqCst); + let _drop_marker = DropMarker(Arc::clone(&self.dropped)); + pending::<()>().await; + Err(Error::General("pending transport completed".to_string())) + } + } + + fn contract() -> InspectArchiveCapabilityContract { + InspectArchiveCapabilityContract { + state: RuntimeCapabilityStatus { + state: RuntimeCapabilityState::Supported, + reason: None, + extra: BTreeMap::new(), + }, + route: INSPECT_ARCHIVE_ROUTE.to_string(), + archive_version: 1, + content_type: INSPECT_ARCHIVE_CONTENT_TYPE.to_string(), + encryption: INSPECT_ARCHIVE_ENCRYPTION.to_string(), + completion_contract: INSPECT_ARCHIVE_COMPLETION.to_string(), + max_bytes: 16 * 1024 * 1024, + max_duration_secs: 30, + max_metadata_bytes_per_drive: 1024 * 1024, + extra: BTreeMap::new(), + } + } + + fn stub(availability: CapabilityAvailability) -> StubArchiveApi { + StubArchiveApi { + availability, + downloads: AtomicUsize::new(0), + dropped: Arc::new(AtomicBool::new(false)), + } + } + + fn formatter() -> Formatter { + Formatter::new(OutputConfig { + quiet: true, + ..OutputConfig::default() + }) + } + + fn args(directory: &Path) -> InspectArchiveArgs { + InspectArchiveArgs { + target: "local/diagnostics/node.json".to_string(), + output: directory.join("inspect.tar"), + private_key: None, + } + } + + fn output_v3_validator() -> Validator { + let schema_path = Path::new(env!("CARGO_MANIFEST_DIR")) + .join("../..") + .join("schemas/output_v3.json"); + let schema = std::fs::read_to_string(&schema_path) + .unwrap_or_else(|error| panic!("failed to read {}: {error}", schema_path.display())); + let schema: Value = serde_json::from_str(&schema).expect("output v3 schema should parse"); + jsonschema::validator_for(&schema).expect("output v3 schema should compile") + } + + #[test] + fn inspect_archive_rejects_unsafe_targets_and_existing_output() { + let directory = tempdir().expect("temporary directory"); + for target in ["local/bucket", "local/bucket/a//b", "local/bucket/a/../b"] { + let mut value = args(directory.path()); + value.target = target.to_string(); + assert!(validate_args(&value).is_err(), "{target} must fail"); + } + let value = args(directory.path()); + std::fs::write(&value.output, b"existing").expect("existing output"); + assert!(matches!(validate_args(&value), Err(Error::Conflict(_)))); + } + + #[cfg(unix)] + #[test] + fn inspect_archive_private_key_requires_regular_private_file() { + use std::os::unix::fs::{PermissionsExt, symlink}; + + let directory = tempdir().expect("temporary directory"); + let key = directory.path().join("key.pem"); + std::fs::write(&key, b"not-a-key").expect("key fixture"); + std::fs::set_permissions(&key, std::fs::Permissions::from_mode(0o644)) + .expect("public permissions"); + assert!(read_private_key(&key).is_err()); + + std::fs::set_permissions(&key, std::fs::Permissions::from_mode(0o600)) + .expect("private permissions"); + let link = directory.path().join("key-link.pem"); + symlink(&key, &link).expect("key symlink"); + assert!(read_private_key(&link).is_err()); + } + + #[tokio::test] + async fn inspect_archive_fails_closed_before_transport_when_unavailable() { + let directory = tempdir().expect("temporary directory"); + for (availability, expected) in [ + ( + CapabilityAvailability::Unsupported, + ExitCode::UnsupportedFeature, + ), + (CapabilityAvailability::Disabled, ExitCode::GeneralError), + ( + CapabilityAvailability::PermissionDenied, + ExitCode::AuthError, + ), + ] { + let api = stub(availability); + let code = run_inspect_archive( + args(directory.path()), + "diagnostics".to_string(), + "node.json".to_string(), + &api, + &api, + &formatter(), + pending::>(), + ) + .await; + assert_eq!(code, expected); + assert_eq!(api.downloads.load(Ordering::SeqCst), 0); + } + } + + #[tokio::test] + async fn inspect_archive_interrupt_drops_in_flight_transport() { + let directory = tempdir().expect("temporary directory"); + let api = stub(CapabilityAvailability::Available); + let dropped = Arc::clone(&api.dropped); + let interrupt = async { + tokio::task::yield_now().await; + Ok(()) + }; + let code = run_inspect_archive( + args(directory.path()), + "diagnostics".to_string(), + "node.json".to_string(), + &api, + &api, + &formatter(), + interrupt, + ) + .await; + assert_eq!(code, ExitCode::Interrupted); + assert!(dropped.load(Ordering::SeqCst)); + assert!(!directory.path().join("inspect.tar").exists()); + } + + #[tokio::test] + async fn inspect_archive_ready_interrupt_does_not_start_transport() { + let directory = tempdir().expect("temporary directory"); + let api = stub(CapabilityAvailability::Available); + let code = run_inspect_archive( + args(directory.path()), + "diagnostics".to_string(), + "node.json".to_string(), + &api, + &api, + &formatter(), + ready(Ok(())), + ) + .await; + assert_eq!(code, ExitCode::Interrupted); + assert_eq!(api.downloads.load(Ordering::SeqCst), 0); + } + + #[test] + fn inspect_archive_json_success_and_error_satisfy_output_v3() { + let success = ArchiveSuccessOutput { + schema_version: 3, + output_type: "diagnostic_archive", + status: "success", + data: ArchiveSuccessData { + operation: "retrieve", + state: "verified", + output: PathBuf::from("inspect.tar"), + archive_version: 1, + drive_count: 2, + encrypted_bytes: 2048, + plaintext_bytes: 1024, + plaintext_sha256: "a".repeat(64), + }, + }; + let error = ArchiveErrorOutput { + schema_version: 3, + output_type: "diagnostic_archive", + status: "error", + error: ArchiveErrorBody { + error_type: "cryptographic_verification", + message: "Diagnostic archive verification failed".to_string(), + retryable: false, + capability: None, + server: None, + suggestion: Some("Do not use the output; retrieve a fresh archive."), + }, + }; + let validator = output_v3_validator(); + for value in [ + serde_json::to_value(success).expect("success JSON"), + serde_json::to_value(error).expect("error JSON"), + ] { + assert!(validator.is_valid(&value), "invalid output: {value}"); + } + } +} diff --git a/crates/cli/src/commands/admin/mod.rs b/crates/cli/src/commands/admin/mod.rs index df9d4b6..9cb93d8 100644 --- a/crates/cli/src/commands/admin/mod.rs +++ b/crates/cli/src/commands/admin/mod.rs @@ -295,6 +295,8 @@ pub fn get_admin_alias(alias_name: &str, formatter: &Formatter) -> Result { + assert_eq!(args.target, "local/diagnostics/node.json"); + assert_eq!(args.output, PathBuf::from("inspect.tar")); + assert_eq!(args.private_key, Some(PathBuf::from("inspect-key.pem"))); + } + _ => panic!("Unexpected command parsing result"), + } + } + #[test] fn test_parse_admin_access_key_info() { let cli = diff --git a/crates/core/Cargo.toml b/crates/core/Cargo.toml index 7861406..0cfc859 100644 --- a/crates/core/Cargo.toml +++ b/crates/core/Cargo.toml @@ -40,9 +40,15 @@ rustls-pki-types.workspace = true x509-parser.workspace = true zeroize.workspace = true getrandom.workspace = true +aes-gcm.workspace = true +rsa.workspace = true +rand.workspace = true +tar.workspace = true +sha2.workspace = true +hex.workspace = true base64.workspace = true http.workspace = true +tempfile.workspace = true [dev-dependencies] -tempfile.workspace = true mockall.workspace = true diff --git a/crates/core/src/admin/capabilities.rs b/crates/core/src/admin/capabilities.rs index 8acf8df..273cde0 100644 --- a/crates/core/src/admin/capabilities.rs +++ b/crates/core/src/admin/capabilities.rs @@ -90,6 +90,8 @@ pub struct RuntimeCapabilitiesSummary { #[derive(Debug, Clone, PartialEq, Serialize, Deserialize)] pub struct RuntimeCapabilitiesSnapshot { pub summary: RuntimeCapabilitiesSummary, + #[serde(default)] + pub inspect_archive: Option, pub cluster_snapshot_path: String, pub cluster_snapshot_summary: Option, pub topology_status: RuntimeCapabilityStatus, diff --git a/crates/core/src/admin/inspect_archive.rs b/crates/core/src/admin/inspect_archive.rs new file mode 100644 index 0000000..5a172b0 --- /dev/null +++ b/crates/core/src/admin/inspect_archive.rs @@ -0,0 +1,1000 @@ +//! Encrypted RustFS diagnostic archive contracts and local verification. + +use std::collections::{BTreeMap, BTreeSet}; +use std::fs::File; +use std::io::{BufReader, Read, Seek, SeekFrom, Write}; +use std::path::{Component, Path, PathBuf}; +use std::sync::Arc; +use std::sync::atomic::{AtomicBool, Ordering}; +use std::time::Duration; + +use aes_gcm::Aes256Gcm; +use aes_gcm::aead::{Aead, KeyInit, Payload}; +use async_trait::async_trait; +use rand::rngs::OsRng; +use rsa::pkcs8::{DecodePrivateKey, EncodePublicKey, LineEnding}; +use rsa::traits::PublicKeyParts; +use rsa::{Oaep, RsaPrivateKey, RsaPublicKey}; +use serde::{Deserialize, Serialize}; +use sha2::{Digest, Sha256}; +use tempfile::NamedTempFile; +use zeroize::Zeroizing; + +use crate::{Error, Result}; + +pub const INSPECT_ARCHIVE_CAPABILITY: &str = "admin.diagnostics.inspect-archive"; +pub const INSPECT_ARCHIVE_ROUTE: &str = "/rustfs/admin/v4/inspect/archive"; +pub const INSPECT_ARCHIVE_CONTENT_TYPE: &str = "application/vnd.rustfs.inspect-archive.v1"; +pub const INSPECT_ARCHIVE_ENCRYPTION: &str = "RSA-OAEP-SHA256+AES-256-GCM-CHUNKED"; +pub const INSPECT_ARCHIVE_COMPLETION: &str = "authenticated-final-record-required"; +pub const INSPECT_ARCHIVE_VERSION: u16 = 1; +pub const MAX_INSPECT_ARCHIVE_BYTES: usize = 16 * 1024 * 1024; +pub const MAX_INSPECT_ARCHIVE_DURATION: Duration = Duration::from_secs(30); +pub const MAX_INSPECT_ARCHIVE_METADATA_BYTES_PER_DRIVE: usize = 4 * 1024 * 1024; + +const FORMAT_MAGIC: &[u8; 8] = b"RFSINSP1"; +const ARCHIVE_CHUNK_SIZE: usize = 64 * 1024; +const RECORD_DATA: u8 = 1; +const RECORD_FINAL: u8 = 2; +const MAX_MANIFEST_BYTES: u64 = 64 * 1024; + +/// Exact capability fields required before invoking the archive route. +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +pub struct InspectArchiveCapabilityContract { + pub state: super::RuntimeCapabilityStatus, + pub route: String, + pub archive_version: u16, + pub content_type: String, + pub encryption: String, + pub completion_contract: String, + pub max_bytes: usize, + pub max_duration_secs: u64, + pub max_metadata_bytes_per_drive: usize, + #[serde(flatten, default)] + pub extra: BTreeMap, +} + +impl InspectArchiveCapabilityContract { + /// Validate the server advertisement against the only contract understood by this client. + pub fn validate(&self) -> Result<()> { + if self.state.availability() != super::CapabilityAvailability::Available { + return Err(Error::UnsupportedFeature( + self.state + .reason + .clone() + .unwrap_or_else(|| "Diagnostic archive capability is unavailable".to_string()), + )); + } + if self.route != INSPECT_ARCHIVE_ROUTE + || self.archive_version != INSPECT_ARCHIVE_VERSION + || self.content_type != INSPECT_ARCHIVE_CONTENT_TYPE + || self.encryption != INSPECT_ARCHIVE_ENCRYPTION + || self.completion_contract != INSPECT_ARCHIVE_COMPLETION + { + return Err(Error::UnsupportedFeature( + "RustFS advertised an incompatible diagnostic archive contract".to_string(), + )); + } + if self.max_bytes == 0 || self.max_bytes > MAX_INSPECT_ARCHIVE_BYTES { + return Err(Error::RequestRejected( + "RustFS advertised an invalid diagnostic archive byte limit".to_string(), + )); + } + if self.max_duration_secs == 0 + || self.max_duration_secs > MAX_INSPECT_ARCHIVE_DURATION.as_secs() + { + return Err(Error::RequestRejected( + "RustFS advertised an invalid diagnostic archive duration limit".to_string(), + )); + } + if self.max_metadata_bytes_per_drive == 0 + || self.max_metadata_bytes_per_drive > MAX_INSPECT_ARCHIVE_METADATA_BYTES_PER_DRIVE + { + return Err(Error::RequestRejected( + "RustFS advertised an invalid diagnostic archive metadata limit".to_string(), + )); + } + Ok(()) + } + + pub fn timeout(&self) -> Duration { + Duration::from_secs(self.max_duration_secs) + } +} + +/// Opaque RSA private key used only for local archive decryption. +pub struct InspectArchiveKey { + private_key: RsaPrivateKey, + public_key_pem: String, +} + +impl InspectArchiveKey { + /// Generate a fresh 2048-bit ephemeral key. + pub fn generate() -> Result { + let private_key = RsaPrivateKey::new(&mut OsRng, 2048) + .map_err(|_| Error::General("Failed to generate diagnostic archive key".to_string()))?; + Self::from_private_key(private_key) + } + + /// Parse caller-provided PKCS#8 PEM private-key material. + pub fn from_pkcs8_pem(pem: &str) -> Result { + let private_key = RsaPrivateKey::from_pkcs8_pem(pem).map_err(|_| { + Error::InvalidPath( + "Diagnostic archive private key must be valid PKCS#8 RSA PEM".to_string(), + ) + })?; + Self::from_private_key(private_key) + } + + fn from_private_key(private_key: RsaPrivateKey) -> Result { + let bits = private_key.size().saturating_mul(8); + if !(2048..=8192).contains(&bits) { + return Err(Error::InvalidPath( + "Diagnostic archive RSA private key must be between 2048 and 8192 bits".to_string(), + )); + } + let public_key_pem = RsaPublicKey::from(&private_key) + .to_public_key_pem(LineEnding::LF) + .map_err(|_| { + Error::General("Failed to derive diagnostic archive public key".to_string()) + })?; + Ok(Self { + private_key, + public_key_pem, + }) + } + + pub fn public_key_pem(&self) -> &str { + &self.public_key_pem + } +} + +/// Sanitized transport request. It intentionally has no `Debug` implementation. +pub struct InspectArchiveTransportRequest { + pub bucket: String, + pub object: String, + pub public_key_pem: String, + pub max_bytes: usize, + pub timeout: Duration, +} + +/// Encrypted response staged in private temporary storage. +pub struct EncryptedInspectArchive { + pub file: NamedTempFile, + pub bytes: u64, +} + +/// Transport boundary implemented by the RustFS Admin API adapter. +#[async_trait] +pub trait InspectArchiveApi: Send + Sync { + async fn inspect_archive_capability(&self) -> Result; + + async fn download_inspect_archive( + &self, + request: InspectArchiveTransportRequest, + temporary_directory: &Path, + ) -> Result; +} + +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +pub struct InspectArchiveManifest { + pub archive_version: u16, + pub drive_count: usize, + pub artifact_paths: Vec, + pub target_identifiers_included: bool, + pub raw_object_data_included: bool, + pub raw_metadata_included: bool, + #[serde(flatten, default)] + pub extra: BTreeMap, +} + +/// Fully authenticated and structurally validated plaintext archive. +pub struct VerifiedInspectArchive { + file: NamedTempFile, + pub manifest: InspectArchiveManifest, + pub encrypted_bytes: u64, + pub plaintext_bytes: u64, + pub plaintext_sha256: String, +} + +/// Safe metadata returned after atomic publication. +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct PublishedInspectArchive { + pub path: PathBuf, + pub archive_version: u16, + pub drive_count: usize, + pub encrypted_bytes: u64, + pub plaintext_bytes: u64, + pub plaintext_sha256: String, +} + +/// Cooperative cancellation shared with blocking decryption and validation. +#[derive(Clone, Default)] +pub struct InspectArchiveCancellation { + cancelled: Arc, +} + +impl InspectArchiveCancellation { + pub fn cancel(&self) { + self.cancelled.store(true, Ordering::Release); + } + + pub fn is_cancelled(&self) -> bool { + self.cancelled.load(Ordering::Acquire) + } +} + +fn crypto_error(message: &str) -> Error { + Error::General(format!("Diagnostic archive verification failed: {message}")) +} + +fn read_exact(reader: &mut impl Read, bytes: &mut [u8], message: &str) -> Result<()> { + reader.read_exact(bytes).map_err(|_| crypto_error(message)) +} + +fn encryption_aad(record_type: u8, counter: u32) -> [u8; 13] { + let mut aad = [0_u8; 13]; + aad[..8].copy_from_slice(FORMAT_MAGIC); + aad[8] = record_type; + aad[9..].copy_from_slice(&counter.to_be_bytes()); + aad +} + +/// Decrypt, authenticate, and validate an encrypted archive without publishing plaintext. +pub fn decrypt_and_validate_inspect_archive( + encrypted: EncryptedInspectArchive, + key: &InspectArchiveKey, + temporary_directory: &Path, + max_plaintext_bytes: usize, +) -> Result { + decrypt_and_validate_inspect_archive_with_cancel( + encrypted, + key, + temporary_directory, + max_plaintext_bytes, + MAX_INSPECT_ARCHIVE_METADATA_BYTES_PER_DRIVE, + max_plaintext_bytes, + &InspectArchiveCancellation::default(), + ) +} + +pub fn decrypt_and_validate_inspect_archive_with_cancel( + encrypted: EncryptedInspectArchive, + key: &InspectArchiveKey, + temporary_directory: &Path, + max_plaintext_bytes: usize, + max_metadata_bytes_per_drive: usize, + max_unpacked_bytes: usize, + cancellation: &InspectArchiveCancellation, +) -> Result { + let mut reader = + BufReader::new(encrypted.file.reopen().map_err(|_| { + Error::InvalidPath("Failed to open staged diagnostic archive".to_string()) + })?); + let mut fixed_header = [0_u8; 24]; + read_exact(&mut reader, &mut fixed_header, "truncated encrypted header")?; + if &fixed_header[..8] != FORMAT_MAGIC { + return Err(crypto_error("invalid encrypted header")); + } + if u16::from_be_bytes([fixed_header[8], fixed_header[9]]) != INSPECT_ARCHIVE_VERSION { + return Err(crypto_error("unsupported archive version")); + } + let chunk_size = usize::try_from(u32::from_be_bytes( + fixed_header[10..14] + .try_into() + .map_err(|_| crypto_error("invalid chunk size"))?, + )) + .map_err(|_| crypto_error("invalid chunk size"))?; + if chunk_size != ARCHIVE_CHUNK_SIZE { + return Err(crypto_error("unexpected encrypted chunk size")); + } + let wrapped_len = usize::from(u16::from_be_bytes([fixed_header[14], fixed_header[15]])); + if wrapped_len == 0 || wrapped_len > 1024 { + return Err(crypto_error("invalid wrapped key length")); + } + let nonce_prefix: [u8; 8] = fixed_header[16..24] + .try_into() + .map_err(|_| crypto_error("invalid nonce prefix"))?; + let mut wrapped_key = vec![0_u8; wrapped_len]; + read_exact( + &mut reader, + &mut wrapped_key, + "truncated wrapped encryption key", + )?; + let data_key = Zeroizing::new( + key.private_key + .decrypt(Oaep::new::(), &wrapped_key) + .map_err(|_| crypto_error("RSA key unwrap failed"))?, + ); + let cipher = Aes256Gcm::new_from_slice(&data_key) + .map_err(|_| crypto_error("invalid decrypted data key"))?; + + let mut plaintext = NamedTempFile::new_in(temporary_directory).map_err(|_| { + Error::InvalidPath("Failed to create private diagnostic archive staging file".to_string()) + })?; + protect_private_file(plaintext.as_file())?; + let mut digest = Sha256::new(); + let mut plaintext_bytes = 0_usize; + let mut expected_counter = 0_u32; + let final_digest = loop { + if cancellation.is_cancelled() { + return Err(Error::Interrupted( + "Diagnostic archive verification was cancelled".to_string(), + )); + } + let mut record_header = [0_u8; 9]; + read_exact( + &mut reader, + &mut record_header, + "authenticated final record is missing", + )?; + let record_type = record_header[0]; + let counter = u32::from_be_bytes( + record_header[1..5] + .try_into() + .map_err(|_| crypto_error("invalid record counter"))?, + ); + if counter != expected_counter { + return Err(crypto_error("record counter is not contiguous")); + } + let ciphertext_len = usize::try_from(u32::from_be_bytes( + record_header[5..9] + .try_into() + .map_err(|_| crypto_error("invalid record length"))?, + )) + .map_err(|_| crypto_error("invalid record length"))?; + let maximum = match record_type { + RECORD_DATA => ARCHIVE_CHUNK_SIZE + 16, + RECORD_FINAL => 32 + 16, + _ => return Err(crypto_error("unknown encrypted record type")), + }; + if ciphertext_len < 16 || ciphertext_len > maximum { + return Err(crypto_error( + "encrypted record length is outside the contract", + )); + } + let mut ciphertext = vec![0_u8; ciphertext_len]; + read_exact(&mut reader, &mut ciphertext, "truncated encrypted record")?; + let mut nonce = [0_u8; 12]; + nonce[..8].copy_from_slice(&nonce_prefix); + nonce[8..].copy_from_slice(&counter.to_be_bytes()); + let decoded = cipher + .decrypt( + (&nonce).into(), + Payload { + msg: &ciphertext, + aad: &encryption_aad(record_type, counter), + }, + ) + .map_err(|_| crypto_error("record authentication failed"))?; + match record_type { + RECORD_DATA => { + plaintext_bytes = plaintext_bytes + .checked_add(decoded.len()) + .ok_or_else(|| crypto_error("plaintext byte accounting overflow"))?; + if plaintext_bytes > max_plaintext_bytes + || plaintext_bytes > MAX_INSPECT_ARCHIVE_BYTES + { + return Err(crypto_error("plaintext archive exceeds the client limit")); + } + digest.update(&decoded); + plaintext.write_all(&decoded).map_err(|_| { + Error::InvalidPath( + "Failed to write private diagnostic archive staging file".to_string(), + ) + })?; + expected_counter = expected_counter + .checked_add(1) + .ok_or_else(|| crypto_error("record counter exhausted"))?; + } + RECORD_FINAL => { + if decoded.len() != 32 || decoded.as_slice() != digest.clone().finalize().as_slice() + { + return Err(crypto_error("final plaintext digest mismatch")); + } + break decoded; + } + _ => unreachable!(), + } + }; + let mut trailing = [0_u8; 1]; + if reader + .read(&mut trailing) + .map_err(|_| crypto_error("failed to inspect encrypted completion"))? + != 0 + { + return Err(crypto_error("records follow the authenticated completion")); + } + plaintext + .as_file_mut() + .sync_all() + .map_err(|_| Error::InvalidPath("Failed to sync diagnostic archive staging file".into()))?; + plaintext + .as_file_mut() + .seek(SeekFrom::Start(0)) + .map_err(|_| crypto_error("failed to rewind plaintext archive"))?; + let manifest = validate_plaintext_archive( + plaintext.as_file_mut(), + plaintext_bytes, + max_metadata_bytes_per_drive, + max_unpacked_bytes, + cancellation, + )?; + + Ok(VerifiedInspectArchive { + file: plaintext, + manifest, + encrypted_bytes: encrypted.bytes, + plaintext_bytes: u64::try_from(plaintext_bytes) + .map_err(|_| crypto_error("plaintext byte accounting overflow"))?, + plaintext_sha256: hex::encode(final_digest), + }) +} + +fn validate_plaintext_archive( + file: &mut File, + expected_plaintext_bytes: usize, + max_metadata_bytes_per_drive: usize, + max_unpacked_bytes: usize, + cancellation: &InspectArchiveCancellation, +) -> Result { + file.seek(SeekFrom::Start(0)) + .map_err(|_| crypto_error("failed to inspect plaintext archive"))?; + let mut archive = tar::Archive::new(file); + let mut entries = archive + .entries() + .map_err(|_| crypto_error("plaintext tar is malformed"))?; + let Some(first) = entries.next() else { + return Err(crypto_error("plaintext tar is empty")); + }; + let mut first = first.map_err(|_| crypto_error("plaintext tar is malformed"))?; + if !first.header().entry_type().is_file() + || first + .path() + .map_err(|_| crypto_error("manifest path is malformed"))? + .as_ref() + != Path::new("manifest.json") + || first.size() > MAX_MANIFEST_BYTES + { + return Err(crypto_error("manifest entry is invalid")); + } + let mut manifest_bytes = Vec::new(); + first + .read_to_end(&mut manifest_bytes) + .map_err(|_| crypto_error("manifest entry is truncated"))?; + let manifest: InspectArchiveManifest = serde_json::from_slice(&manifest_bytes) + .map_err(|_| crypto_error("manifest JSON is invalid"))?; + validate_manifest(&manifest)?; + let mut unpacked_bytes = manifest_bytes.len(); + if unpacked_bytes > max_unpacked_bytes { + return Err(crypto_error("unpacked archive exceeds the client limit")); + } + + let expected = manifest + .artifact_paths + .iter() + .cloned() + .collect::>(); + let mut observed = BTreeSet::new(); + for entry in entries { + if cancellation.is_cancelled() { + return Err(Error::Interrupted( + "Diagnostic archive verification was cancelled".to_string(), + )); + } + let mut entry = entry.map_err(|_| crypto_error("plaintext tar is malformed"))?; + if !entry.header().entry_type().is_file() + || entry.size() + > u64::try_from(max_metadata_bytes_per_drive) + .map_err(|_| crypto_error("invalid drive metadata limit"))? + { + return Err(crypto_error("drive artifact entry is invalid")); + } + let path = entry + .path() + .map_err(|_| crypto_error("drive artifact path is malformed"))? + .to_string_lossy() + .into_owned(); + if !expected.contains(&path) || !observed.insert(path) { + return Err(crypto_error("plaintext tar contains an unexpected entry")); + } + let mut artifact = Vec::new(); + entry + .read_to_end(&mut artifact) + .map_err(|_| crypto_error("drive artifact entry is truncated"))?; + unpacked_bytes = unpacked_bytes + .checked_add(artifact.len()) + .ok_or_else(|| crypto_error("unpacked byte accounting overflow"))?; + if unpacked_bytes > max_unpacked_bytes { + return Err(crypto_error("unpacked archive exceeds the client limit")); + } + let value: serde_json::Value = serde_json::from_slice(&artifact) + .map_err(|_| crypto_error("drive artifact JSON is invalid"))?; + if !value.is_object() { + return Err(crypto_error("drive artifact JSON is not an object")); + } + } + if observed != expected { + return Err(crypto_error("plaintext tar is missing drive artifacts")); + } + if archive + .into_inner() + .metadata() + .map_err(|_| crypto_error("failed to inspect plaintext archive"))? + .len() + != u64::try_from(expected_plaintext_bytes) + .map_err(|_| crypto_error("plaintext byte accounting overflow"))? + { + return Err(crypto_error( + "plaintext archive length changed during verification", + )); + } + Ok(manifest) +} + +fn validate_manifest(manifest: &InspectArchiveManifest) -> Result<()> { + if manifest.archive_version != INSPECT_ARCHIVE_VERSION + || manifest.target_identifiers_included + || manifest.raw_object_data_included + || manifest.raw_metadata_included + || manifest.drive_count != manifest.artifact_paths.len() + { + return Err(crypto_error("manifest safety contract is invalid")); + } + let expected = (0..manifest.drive_count) + .map(|index| format!("drives/{index:04}.json")) + .collect::>(); + if manifest.artifact_paths != expected { + return Err(crypto_error("manifest artifact paths are invalid")); + } + Ok(()) +} + +fn protect_private_file(file: &File) -> Result<()> { + #[cfg(unix)] + { + use std::os::unix::fs::PermissionsExt; + file.set_permissions(std::fs::Permissions::from_mode(0o600)) + .map_err(|_| { + Error::InvalidPath("Failed to protect diagnostic archive staging file".to_string()) + })?; + } + Ok(()) +} + +/// Reject output directories containing symbolic links or parent traversal. +pub fn validate_inspect_archive_output_directory(directory: &Path) -> Result<()> { + let mut current = if directory.is_absolute() { + PathBuf::new() + } else { + PathBuf::from(".") + }; + let mut normal_depth = 0_usize; + for component in directory.components() { + match component { + Component::Prefix(_) | Component::RootDir => { + current.push(component.as_os_str()); + } + Component::CurDir => continue, + Component::ParentDir => { + return Err(Error::InvalidPath( + "Diagnostic archive output directory cannot contain parent traversal" + .to_string(), + )); + } + Component::Normal(part) => { + normal_depth += 1; + current.push(part); + let metadata = std::fs::symlink_metadata(¤t).map_err(|_| { + Error::InvalidPath( + "Diagnostic archive output directory does not exist or is inaccessible" + .to_string(), + ) + })?; + // macOS exposes system roots such as /var and /tmp as root-level aliases. + let platform_root_alias = directory.is_absolute() && normal_depth == 1; + if metadata.file_type().is_symlink() && !platform_root_alias { + return Err(Error::InvalidPath( + "Diagnostic archive output directory cannot contain symbolic links" + .to_string(), + )); + } + } + } + } + let metadata = std::fs::symlink_metadata(directory).map_err(|_| { + Error::InvalidPath( + "Diagnostic archive output directory does not exist or is inaccessible".to_string(), + ) + })?; + if metadata.file_type().is_symlink() { + return Err(Error::InvalidPath( + "Diagnostic archive output directory cannot be a symbolic link".to_string(), + )); + } + if !metadata.is_dir() { + return Err(Error::InvalidPath( + "Diagnostic archive output directory does not exist".to_string(), + )); + } + Ok(()) +} + +/// Atomically publish a verified archive without replacing an existing destination. +pub fn publish_inspect_archive( + verified: VerifiedInspectArchive, + destination: &Path, +) -> Result { + let parent = destination + .parent() + .filter(|path| !path.as_os_str().is_empty()) + .unwrap_or_else(|| Path::new(".")); + validate_inspect_archive_output_directory(parent)?; + if std::fs::symlink_metadata(destination).is_ok() { + return Err(Error::Conflict( + "Diagnostic archive output already exists".to_string(), + )); + } + let VerifiedInspectArchive { + file, + manifest, + encrypted_bytes, + plaintext_bytes, + plaintext_sha256, + } = verified; + file.persist_noclobber(destination).map_err(|error| { + if error.error.kind() == std::io::ErrorKind::AlreadyExists { + Error::Conflict("Diagnostic archive output already exists".to_string()) + } else { + Error::InvalidPath("Failed to atomically publish diagnostic archive output".to_string()) + } + })?; + #[cfg(unix)] + if File::open(parent) + .and_then(|directory| directory.sync_all()) + .is_err() + { + let _ = std::fs::remove_file(destination); + return Err(Error::InvalidPath( + "Failed to sync diagnostic archive output directory".to_string(), + )); + } + Ok(PublishedInspectArchive { + path: destination.to_path_buf(), + archive_version: manifest.archive_version, + drive_count: manifest.drive_count, + encrypted_bytes, + plaintext_bytes, + plaintext_sha256, + }) +} + +#[cfg(test)] +mod tests { + use super::*; + use aes_gcm::Key; + use rsa::pkcs8::DecodePublicKey; + use tempfile::tempdir; + + fn manifest(raw_metadata: bool) -> InspectArchiveManifest { + InspectArchiveManifest { + archive_version: 1, + drive_count: 1, + artifact_paths: vec!["drives/0000.json".to_string()], + target_identifiers_included: false, + raw_object_data_included: false, + raw_metadata_included: raw_metadata, + extra: BTreeMap::new(), + } + } + + fn tar_fixture(manifest: &InspectArchiveManifest) -> Vec { + let mut output = Vec::new(); + { + let mut archive = tar::Builder::new(&mut output); + for (path, bytes) in [ + ( + "manifest.json", + serde_json::to_vec(manifest).expect("serialize manifest"), + ), + ( + "drives/0000.json", + br#"{"drive_index":0,"status":"ok"}"#.to_vec(), + ), + ] { + let mut header = tar::Header::new_gnu(); + header.set_mode(0o600); + header.set_mtime(0); + header.set_size(bytes.len() as u64); + header.set_cksum(); + archive + .append_data(&mut header, path, bytes.as_slice()) + .expect("append tar entry"); + } + archive.finish().expect("finish tar"); + } + output + } + + fn encrypt_fixture(key: &InspectArchiveKey, plaintext: &[u8]) -> Vec { + let public = + RsaPublicKey::from_public_key_pem(key.public_key_pem()).expect("parse public key"); + let data_key = [7_u8; 32]; + let cipher = Aes256Gcm::new(Key::::from_slice(&data_key)); + let wrapped = public + .encrypt(&mut OsRng, Oaep::new::(), &data_key) + .expect("wrap key"); + let nonce_prefix = [9_u8; 8]; + let mut output = Vec::new(); + output.extend_from_slice(FORMAT_MAGIC); + output.extend_from_slice(&INSPECT_ARCHIVE_VERSION.to_be_bytes()); + output.extend_from_slice(&(ARCHIVE_CHUNK_SIZE as u32).to_be_bytes()); + output.extend_from_slice(&(wrapped.len() as u16).to_be_bytes()); + output.extend_from_slice(&nonce_prefix); + output.extend_from_slice(&wrapped); + + for (counter, chunk) in plaintext.chunks(ARCHIVE_CHUNK_SIZE).enumerate() { + append_encrypted_record( + &mut output, + &cipher, + &nonce_prefix, + RECORD_DATA, + counter as u32, + chunk, + ); + } + append_encrypted_record( + &mut output, + &cipher, + &nonce_prefix, + RECORD_FINAL, + plaintext.len().div_ceil(ARCHIVE_CHUNK_SIZE) as u32, + &Sha256::digest(plaintext), + ); + output + } + + fn append_encrypted_record( + output: &mut Vec, + cipher: &Aes256Gcm, + nonce_prefix: &[u8; 8], + record_type: u8, + counter: u32, + plaintext: &[u8], + ) { + let mut nonce = [0_u8; 12]; + nonce[..8].copy_from_slice(nonce_prefix); + nonce[8..].copy_from_slice(&counter.to_be_bytes()); + let ciphertext = cipher + .encrypt( + (&nonce).into(), + Payload { + msg: plaintext, + aad: &encryption_aad(record_type, counter), + }, + ) + .expect("encrypt record"); + output.push(record_type); + output.extend_from_slice(&counter.to_be_bytes()); + output.extend_from_slice(&(ciphertext.len() as u32).to_be_bytes()); + output.extend_from_slice(&ciphertext); + } + + fn staged(bytes: &[u8], directory: &Path) -> EncryptedInspectArchive { + let mut file = NamedTempFile::new_in(directory).expect("create encrypted staging"); + file.write_all(bytes).expect("write encrypted staging"); + EncryptedInspectArchive { + file, + bytes: bytes.len() as u64, + } + } + + #[test] + fn complete_archive_decrypts_validates_and_publishes_privately() { + let directory = tempdir().expect("temp directory"); + let key = InspectArchiveKey::generate().expect("key"); + let plaintext = tar_fixture(&manifest(false)); + let encrypted = encrypt_fixture(&key, &plaintext); + let verified = decrypt_and_validate_inspect_archive( + staged(&encrypted, directory.path()), + &key, + directory.path(), + MAX_INSPECT_ARCHIVE_BYTES, + ) + .expect("verified archive"); + assert_eq!(verified.manifest.drive_count, 1); + assert_eq!(verified.plaintext_bytes, plaintext.len() as u64); + let destination = directory.path().join("inspect.tar"); + let published = + publish_inspect_archive(verified, &destination).expect("publish verified archive"); + assert_eq!(published.path, destination); + assert_eq!(std::fs::read(&destination).expect("read output"), plaintext); + #[cfg(unix)] + { + use std::os::unix::fs::PermissionsExt; + assert_eq!( + std::fs::metadata(destination) + .expect("metadata") + .permissions() + .mode() + & 0o077, + 0 + ); + } + } + + #[test] + fn corruption_truncation_and_missing_completion_leave_no_plaintext() { + let directory = tempdir().expect("temp directory"); + let key = InspectArchiveKey::generate().expect("key"); + let plaintext = tar_fixture(&manifest(false)); + let encrypted = encrypt_fixture(&key, &plaintext); + let cases = [ + encrypted[..encrypted.len() - 1].to_vec(), + { + let mut corrupt = encrypted.clone(); + let last = corrupt.last_mut().expect("ciphertext"); + *last ^= 1; + corrupt + }, + encrypted[..encrypted.len() - 57].to_vec(), + ]; + for bytes in cases { + let before = std::fs::read_dir(directory.path()) + .expect("read directory") + .count(); + let error = decrypt_and_validate_inspect_archive( + staged(&bytes, directory.path()), + &key, + directory.path(), + MAX_INSPECT_ARCHIVE_BYTES, + ) + .err() + .expect("invalid archive"); + assert!(error.to_string().contains("verification failed")); + let after = std::fs::read_dir(directory.path()) + .expect("read directory") + .count(); + assert_eq!(before, after); + } + } + + #[test] + fn unsafe_manifest_limit_and_cancellation_are_rejected() { + let directory = tempdir().expect("temp directory"); + let key = InspectArchiveKey::generate().expect("key"); + let unsafe_tar = tar_fixture(&manifest(true)); + let error = decrypt_and_validate_inspect_archive( + staged(&encrypt_fixture(&key, &unsafe_tar), directory.path()), + &key, + directory.path(), + MAX_INSPECT_ARCHIVE_BYTES, + ) + .err() + .expect("unsafe manifest"); + assert!(error.to_string().contains("manifest safety")); + + let valid_tar = tar_fixture(&manifest(false)); + let error = decrypt_and_validate_inspect_archive( + staged(&encrypt_fixture(&key, &valid_tar), directory.path()), + &key, + directory.path(), + valid_tar.len() - 1, + ) + .err() + .expect("plaintext limit"); + assert!(error.to_string().contains("client limit")); + + let cancellation = InspectArchiveCancellation::default(); + cancellation.cancel(); + let error = decrypt_and_validate_inspect_archive_with_cancel( + staged(&encrypt_fixture(&key, &valid_tar), directory.path()), + &key, + directory.path(), + MAX_INSPECT_ARCHIVE_BYTES, + MAX_INSPECT_ARCHIVE_METADATA_BYTES_PER_DRIVE, + MAX_INSPECT_ARCHIVE_BYTES, + &cancellation, + ) + .err() + .expect("cancelled verification"); + assert!(matches!(error, Error::Interrupted(_))); + + let error = decrypt_and_validate_inspect_archive_with_cancel( + staged(&encrypt_fixture(&key, &valid_tar), directory.path()), + &key, + directory.path(), + MAX_INSPECT_ARCHIVE_BYTES, + MAX_INSPECT_ARCHIVE_METADATA_BYTES_PER_DRIVE, + 1, + &InspectArchiveCancellation::default(), + ) + .err() + .expect("unpacked limit"); + assert!(error.to_string().contains("unpacked archive")); + } + + #[test] + fn publish_refuses_overwrite_and_removes_staging_file() { + let directory = tempdir().expect("temp directory"); + let key = InspectArchiveKey::generate().expect("key"); + let plaintext = tar_fixture(&manifest(false)); + let verified = decrypt_and_validate_inspect_archive( + staged(&encrypt_fixture(&key, &plaintext), directory.path()), + &key, + directory.path(), + MAX_INSPECT_ARCHIVE_BYTES, + ) + .expect("verified archive"); + let staging = verified.file.path().to_path_buf(); + let destination = directory.path().join("existing.tar"); + std::fs::write(&destination, b"keep").expect("existing output"); + let error = publish_inspect_archive(verified, &destination).expect_err("no overwrite"); + assert!(matches!(error, Error::Conflict(_))); + assert_eq!( + std::fs::read(destination).expect("existing output"), + b"keep" + ); + assert!(!staging.exists()); + } + + #[cfg(unix)] + #[test] + fn output_directory_and_broken_destination_symlinks_are_rejected() { + use std::os::unix::fs::symlink; + + let directory = tempdir().expect("temp directory"); + let real = directory.path().join("real"); + std::fs::create_dir(&real).expect("real directory"); + let child = real.join("child"); + std::fs::create_dir(&child).expect("real child directory"); + let linked = directory.path().join("linked"); + symlink(&real, &linked).expect("directory symlink"); + assert!(validate_inspect_archive_output_directory(&linked).is_err()); + assert!(validate_inspect_archive_output_directory(&linked.join("child")).is_err()); + + let destination = real.join("archive.tar"); + symlink(real.join("missing"), &destination).expect("broken destination symlink"); + let key = InspectArchiveKey::generate().expect("key"); + let plaintext = tar_fixture(&manifest(false)); + let verified = decrypt_and_validate_inspect_archive( + staged(&encrypt_fixture(&key, &plaintext), &real), + &key, + &real, + MAX_INSPECT_ARCHIVE_BYTES, + ) + .expect("verified archive"); + assert!(matches!( + publish_inspect_archive(verified, &destination), + Err(Error::Conflict(_)) + )); + } + + #[test] + fn capability_and_private_key_contracts_fail_closed() { + let supported = InspectArchiveCapabilityContract { + state: super::super::RuntimeCapabilityStatus { + state: super::super::RuntimeCapabilityState::Supported, + reason: None, + extra: BTreeMap::new(), + }, + route: INSPECT_ARCHIVE_ROUTE.to_string(), + archive_version: 1, + content_type: INSPECT_ARCHIVE_CONTENT_TYPE.to_string(), + encryption: INSPECT_ARCHIVE_ENCRYPTION.to_string(), + completion_contract: INSPECT_ARCHIVE_COMPLETION.to_string(), + max_bytes: MAX_INSPECT_ARCHIVE_BYTES, + max_duration_secs: 30, + max_metadata_bytes_per_drive: 4 * 1024 * 1024, + extra: BTreeMap::new(), + }; + supported.validate().expect("supported contract"); + let mut incompatible = supported; + incompatible.encryption = "future".to_string(); + assert!(matches!( + incompatible.validate(), + Err(Error::UnsupportedFeature(_)) + )); + assert!(InspectArchiveKey::from_pkcs8_pem("not-a-key").is_err()); + } +} diff --git a/crates/core/src/admin/mod.rs b/crates/core/src/admin/mod.rs index 8d6ab4b..15511b7 100644 --- a/crates/core/src/admin/mod.rs +++ b/crates/core/src/admin/mod.rs @@ -10,6 +10,7 @@ mod cluster; mod configuration; mod diagnostics; mod iam; +mod inspect_archive; mod kms; mod kms_diagnostic; mod observability; @@ -71,6 +72,16 @@ pub use iam::{ PolicyDetachRequest, PolicyDetachResult, PolicyEntities, PolicyEntitiesQuery, PolicyEntitiesResult, UserPolicyEntities, }; +pub use inspect_archive::{ + EncryptedInspectArchive, INSPECT_ARCHIVE_CAPABILITY, INSPECT_ARCHIVE_COMPLETION, + INSPECT_ARCHIVE_CONTENT_TYPE, INSPECT_ARCHIVE_ENCRYPTION, INSPECT_ARCHIVE_ROUTE, + INSPECT_ARCHIVE_VERSION, InspectArchiveApi, InspectArchiveCancellation, + InspectArchiveCapabilityContract, InspectArchiveKey, InspectArchiveManifest, + InspectArchiveTransportRequest, MAX_INSPECT_ARCHIVE_BYTES, MAX_INSPECT_ARCHIVE_DURATION, + MAX_INSPECT_ARCHIVE_METADATA_BYTES_PER_DRIVE, PublishedInspectArchive, VerifiedInspectArchive, + decrypt_and_validate_inspect_archive, decrypt_and_validate_inspect_archive_with_cancel, + publish_inspect_archive, validate_inspect_archive_output_directory, +}; pub use kms::{ KmsApi, KmsBackendKind, KmsCacheSummary, KmsCancelKeyDeletionResult, KmsConfigSummary, KmsConfigureRequest, KmsCreateKeyRequest, KmsCreateKeyResult, KmsDeleteKeyRequest, diff --git a/crates/s3/src/admin.rs b/crates/s3/src/admin.rs index ee9d554..8067a65 100644 --- a/crates/s3/src/admin.rs +++ b/crates/s3/src/admin.rs @@ -20,28 +20,31 @@ use rc_core::admin::{ ClusterSnapshotDocument, ClusterSnapshotMetadata, ClusterSnapshotSummary, ConfigApi, ConfigDocument, ConfigHelp, ConfigHistoryEntry, ConfigMutationResult, CreateServiceAccountRequest, DecommissionPoolStatus, DecommissionStatus, - DetailedHealthSnapshot, DiagnosticCapability, DiagnosticReadApi, ExtensionsCatalog, Group, - GroupStatus, HealRuntimeState, HealScanMode, HealStartRequest, HealStatus, HealTaskRequest, - IAM_ACCESS_KEYS_BULK_CAPABILITY, IAM_ACCESS_KEYS_BULK_LDAP_CAPABILITY, - IAM_ACCESS_KEYS_BULK_OPENID_CAPABILITY, IAM_POLICY_DETACH_CAPABILITY, - IAM_POLICY_ENTITIES_CAPABILITY, IamArchiveApi, IamArchiveImportResult, IamArchiveImportSection, - IamArchiveInventory, IamArchiveResultEntities, IamMutationApi, IamReadApi, KmsApi, - KmsBackendKind, KmsCacheSummary, KmsCancelKeyDeletionResult, KmsConfigSummary, - KmsConfigureRequest, KmsCreateKeyRequest, KmsCreateKeyResult, KmsDeleteKeyRequest, - KmsDeleteKeyResult, KmsKey, KmsKeyPage, KmsKeyState, KmsKeyUsage, KmsServiceState, KmsStatus, + DetailedHealthSnapshot, DiagnosticCapability, DiagnosticReadApi, EncryptedInspectArchive, + ExtensionsCatalog, Group, GroupStatus, HealRuntimeState, HealScanMode, HealStartRequest, + HealStatus, HealTaskRequest, IAM_ACCESS_KEYS_BULK_CAPABILITY, + IAM_ACCESS_KEYS_BULK_LDAP_CAPABILITY, IAM_ACCESS_KEYS_BULK_OPENID_CAPABILITY, + IAM_POLICY_DETACH_CAPABILITY, IAM_POLICY_ENTITIES_CAPABILITY, INSPECT_ARCHIVE_COMPLETION, + INSPECT_ARCHIVE_CONTENT_TYPE, IamArchiveApi, IamArchiveImportResult, IamArchiveImportSection, + IamArchiveInventory, IamArchiveResultEntities, IamMutationApi, IamReadApi, InspectArchiveApi, + InspectArchiveCapabilityContract, InspectArchiveTransportRequest, KmsApi, KmsBackendKind, + KmsCacheSummary, KmsCancelKeyDeletionResult, KmsConfigSummary, KmsConfigureRequest, + KmsCreateKeyRequest, KmsCreateKeyResult, KmsDeleteKeyRequest, KmsDeleteKeyResult, KmsKey, + KmsKeyPage, KmsKeyState, KmsKeyUsage, KmsServiceState, KmsStatus, MAX_BUCKET_METADATA_ARCHIVE_BYTES, MAX_DIAGNOSTIC_RESPONSE_BYTES, MAX_IAM_ACCESS_KEY_RESULTS, MAX_IAM_ACCESS_KEYS_RESPONSE_BYTES, MAX_IAM_ARCHIVE_BYTES, MAX_IAM_IMPORT_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_REPLICATION_INSPECTION_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, OidcMutationApi, OidcMutationRequest, OidcMutationResult, - OidcProvider, OidcProviderList, OidcReadApi, OidcValidationRequest, OidcValidationResult, - PeerSiteSpec, Policy, PolicyDetachEntity, PolicyDetachRequest, PolicyDetachResult, - PolicyEntitiesQuery, PolicyEntitiesResult, PolicyEntity, PolicyInfo, PoolStatus, PoolTarget, - RealtimeMetrics, RebalanceStartResult, RebalanceStatus, ReplicateEditStatus, ReplicationDiff, + MAX_IAM_POLICY_ENTITIES_RESPONSE_BYTES, MAX_INSPECT_ARCHIVE_BYTES, MAX_METRICS_LINE_BYTES, + MAX_METRICS_RESPONSE_BYTES, MAX_METRICS_SAMPLES, MAX_OIDC_RESPONSE_BYTES, + MAX_REPLICATION_DIFF_RESPONSE_BYTES, MAX_REPLICATION_INSPECTION_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, + OidcMutationApi, OidcMutationRequest, OidcMutationResult, OidcProvider, OidcProviderList, + OidcReadApi, OidcValidationRequest, OidcValidationResult, PeerSiteSpec, Policy, + PolicyDetachEntity, PolicyDetachRequest, PolicyDetachResult, PolicyEntitiesQuery, + PolicyEntitiesResult, PolicyEntity, PolicyInfo, PoolStatus, PoolTarget, RealtimeMetrics, + RebalanceStartResult, RebalanceStatus, ReplicateEditStatus, ReplicationDiff, ReplicationDiffApi, ReplicationInspectionApi, ReplicationMetricScope, ReplicationMetrics, ReplicationMrf, RuntimeCapabilitiesSnapshot, RuntimeCapabilityStatus, ScannerStatus, ServiceAccount, ServiceAccountCreateResponse, ServiceActionResult, SiteRemoveSpec, @@ -50,7 +53,7 @@ use rc_core::admin::{ UpdateServiceAccountRequest, User, UserStatus, }; use rc_core::{Alias, Error, Result}; -use reqwest::header::{CONTENT_TYPE, HOST, HeaderMap, HeaderName, HeaderValue}; +use reqwest::header::{CACHE_CONTROL, CONTENT_TYPE, HOST, HeaderMap, HeaderName, HeaderValue}; use reqwest::{Client, Method, StatusCode}; use serde::{Deserialize, Serialize}; use sha2::{Digest, Sha256}; @@ -67,6 +70,13 @@ struct BoundedJsonResponse { error_mapper: fn(&AdminClient, StatusCode, &str) -> Error, } +#[derive(Serialize)] +struct InspectArchiveWireRequest<'a> { + bucket: &'a str, + object: &'a str, + public_key_pem: &'a str, +} + #[derive(Debug, Serialize)] struct PolicyDetachWireRequest<'a> { policies: &'a [String], @@ -1263,6 +1273,47 @@ impl AdminClient { } } + fn map_inspect_archive_error(&self, status: StatusCode, body: &str) -> Error { + if matches!(status, StatusCode::FORBIDDEN | StatusCode::UNAUTHORIZED) { + return Error::Auth( + "Permission denied while requesting the diagnostic archive".to_string(), + ); + } + let code = parse_admin_error(body).and_then(|error| error.code); + if status == StatusCode::NOT_FOUND + && matches!( + code.as_deref(), + Some("NoSuchBucket" | "NoSuchKey" | "NoSuchObject") + ) + { + return Error::NotFound("Diagnostic archive target was not found".to_string()); + } + match status { + StatusCode::NOT_FOUND + | StatusCode::METHOD_NOT_ALLOWED + | StatusCode::NOT_IMPLEMENTED => Error::UnsupportedFeature( + "Encrypted diagnostic archives are not supported by this RustFS server".to_string(), + ), + StatusCode::BAD_REQUEST | StatusCode::UNPROCESSABLE_ENTITY => Error::InvalidPath( + "RustFS rejected the diagnostic archive target or public key".to_string(), + ), + StatusCode::PAYLOAD_TOO_LARGE => Error::RequestRejected( + "RustFS rejected the diagnostic archive because a server limit was exceeded" + .to_string(), + ), + StatusCode::REQUEST_TIMEOUT | StatusCode::GATEWAY_TIMEOUT => { + Error::Network("Diagnostic archive request timed out".to_string()) + } + StatusCode::TOO_MANY_REQUESTS => Error::RequestRejected( + "RustFS diagnostic archive capacity limit was reached".to_string(), + ), + _ => Error::Network(format!( + "Diagnostic archive request failed with HTTP {}", + status.as_u16() + )), + } + } + fn map_iam_policy_entities_error(&self, status: StatusCode, _body: &str) -> Error { match status { StatusCode::NOT_FOUND @@ -2109,7 +2160,7 @@ impl AdminClient { } fn runtime_summary_entries(runtime: &RuntimeCapabilitiesSnapshot) -> Vec { - [ + let mut entries = [ ("runtime.observability", &runtime.summary.observability), ( "runtime.userspace-profiling", @@ -2125,7 +2176,15 @@ fn runtime_summary_entries(runtime: &RuntimeCapabilitiesSnapshot) -> Vec>(); + if let Some(contract) = &runtime.inspect_archive { + entries.push(CapabilityEntry { + name: DiagnosticCapability::InspectArchive.name().to_string(), + availability: contract.state.availability(), + reason: contract.state.reason.clone(), + }); + } + entries } fn capability_entry(name: &str, status: &RuntimeCapabilityStatus) -> CapabilityEntry { @@ -2276,15 +2335,19 @@ fn add_uniform_diagnostic_capabilities( availability: CapabilityAvailability, reason: &str, ) { - capabilities.extend( - DiagnosticCapability::ALL - .into_iter() - .map(|capability| CapabilityEntry { - name: capability.name().to_string(), - availability, - reason: Some(reason.to_string()), - }), - ); + for capability in DiagnosticCapability::ALL { + if capabilities + .iter() + .any(|entry| entry.name == capability.name()) + { + continue; + } + capabilities.push(CapabilityEntry { + name: capability.name().to_string(), + availability, + reason: Some(reason.to_string()), + }); + } } fn add_beta10_diagnostic_capabilities(capabilities: &mut Vec) { @@ -2330,6 +2393,12 @@ fn add_beta10_diagnostic_capabilities(capabilities: &mut Vec) { "RustFS beta.10 site-replication netperf does not perform a peer network measurement", ), ] { + if capabilities + .iter() + .any(|entry| entry.name == capability.name()) + { + continue; + } capabilities.push(CapabilityEntry { name: capability.name().to_string(), availability, @@ -3698,6 +3767,168 @@ impl DiagnosticReadApi for AdminClient { } } +#[async_trait] +impl InspectArchiveApi for AdminClient { + async fn inspect_archive_capability(&self) -> Result { + let runtime = self + .request_v4::(Method::GET, "/runtime/capabilities") + .await + .map_err(|error| diagnostic_route_error(error, "Diagnostic archive capability"))?; + runtime.inspect_archive.ok_or_else(|| { + Error::UnsupportedFeature( + "RustFS did not advertise an encrypted diagnostic archive contract".to_string(), + ) + }) + } + + async fn download_inspect_archive( + &self, + request: InspectArchiveTransportRequest, + temporary_directory: &std::path::Path, + ) -> Result { + let operation = async { + let body = serde_json::to_vec(&InspectArchiveWireRequest { + bucket: &request.bucket, + object: &request.object, + public_key_pem: &request.public_key_pem, + }) + .map_err(|_| { + Error::General("Failed to encode diagnostic archive request".to_string()) + })?; + let url = self.admin_v4_url("/inspect/archive"); + let headers = self.request_headers(&body)?; + let signed_headers = self + .sign_request(&Method::POST, &url, &headers, &body) + .await?; + let mut builder = self + .http_client + .request(Method::POST, &url) + .header(CONTENT_TYPE, "application/json") + .body(body); + for (name, value) in &signed_headers { + builder = builder.header(name, value); + } + let response = builder + .send() + .await + .map_err(|_| Error::Network("Diagnostic archive request failed".to_string()))?; + let status = response.status(); + if !status.is_success() { + let body = read_bounded_response_body( + response, + 64 * 1024, + "Diagnostic archive error response", + ) + .await?; + return Err(self.map_inspect_archive_error(status, &String::from_utf8_lossy(&body))); + } + let content_type = response + .headers() + .get(CONTENT_TYPE) + .and_then(|value| value.to_str().ok()); + let completion = response + .headers() + .get("x-rustfs-inspect-completion") + .and_then(|value| value.to_str().ok()); + let cache_control = response + .headers() + .get(CACHE_CONTROL) + .and_then(|value| value.to_str().ok()); + if content_type != Some(INSPECT_ARCHIVE_CONTENT_TYPE) + || completion != Some(INSPECT_ARCHIVE_COMPLETION) + || !cache_control.is_some_and(|value| { + value + .split(',') + .any(|directive| directive.trim().eq_ignore_ascii_case("no-store")) + }) + { + return Err(Error::General( + "Diagnostic archive response headers do not match the advertised contract" + .to_string(), + )); + } + if response + .content_length() + .is_some_and(|length| length > request.max_bytes as u64) + { + return Err(Error::RequestRejected( + "Diagnostic archive response exceeds the advertised byte limit".to_string(), + )); + } + + let temporary = tempfile::Builder::new() + .prefix(".rc-inspect-encrypted-") + .tempfile_in(temporary_directory) + .map_err(|_| { + Error::InvalidPath( + "Failed to create private encrypted archive staging file".to_string(), + ) + })?; + #[cfg(unix)] + { + use std::os::unix::fs::PermissionsExt; + temporary + .as_file() + .set_permissions(std::fs::Permissions::from_mode(0o600)) + .map_err(|_| { + Error::InvalidPath( + "Failed to protect encrypted archive staging file".to_string(), + ) + })?; + } + let file = temporary.reopen().map_err(|_| { + Error::InvalidPath( + "Failed to open private encrypted archive staging file".to_string(), + ) + })?; + let mut file = tokio::fs::File::from_std(file); + let mut bytes = 0_usize; + let mut stream = response.bytes_stream(); + while let Some(chunk) = stream.next().await { + let chunk = chunk.map_err(|_| { + Error::Network("Diagnostic archive response stream failed".to_string()) + })?; + bytes = bytes.checked_add(chunk.len()).ok_or_else(|| { + Error::RequestRejected( + "Diagnostic archive response byte accounting overflow".to_string(), + ) + })?; + if bytes > request.max_bytes || bytes > MAX_INSPECT_ARCHIVE_BYTES { + return Err(Error::RequestRejected( + "Diagnostic archive response exceeds the advertised byte limit".to_string(), + )); + } + tokio::io::AsyncWriteExt::write_all(&mut file, &chunk) + .await + .map_err(|_| { + Error::InvalidPath( + "Failed to write encrypted archive staging file".to_string(), + ) + })?; + } + tokio::io::AsyncWriteExt::flush(&mut file) + .await + .map_err(|_| { + Error::InvalidPath("Failed to sync encrypted archive staging file".to_string()) + })?; + file.into_std().await.sync_all().map_err(|_| { + Error::InvalidPath("Failed to sync encrypted archive staging file".to_string()) + })?; + Ok(EncryptedInspectArchive { + file: temporary, + bytes: u64::try_from(bytes).map_err(|_| { + Error::RequestRejected( + "Diagnostic archive response byte accounting overflow".to_string(), + ) + })?, + }) + }; + tokio::time::timeout(request.timeout, operation) + .await + .map_err(|_| Error::Network("Diagnostic archive request timed out".to_string()))? + } +} + #[async_trait] impl ConfigApi for AdminClient { async fn get_config(&self, selector: &str) -> Result { @@ -5012,6 +5243,27 @@ mod tests { (endpoint, handle) } + fn start_admin_captured_raw_response_server( + response: Vec, + ) -> ( + String, + mpsc::Receiver, + thread::JoinHandle<()>, + ) { + let listener = TcpListener::bind("127.0.0.1:0").expect("bind test server"); + let endpoint = format!("http://{}", listener.local_addr().expect("local addr")); + let (sender, receiver) = mpsc::channel(); + let handle = thread::spawn(move || { + let (mut stream, _) = listener.accept().expect("accept request"); + sender + .send(read_admin_request(&mut stream)) + .expect("send captured request"); + let _ = stream.write_all(&response); + }); + + (endpoint, receiver, handle) + } + fn start_admin_owned_test_server( response_status: &str, content_type: &str, @@ -9765,4 +10017,174 @@ mod tests { assert_eq!(request.target, "/rustfs/admin/v3/import-iam-v2"); handle.join().expect("server thread"); } + + fn inspect_archive_request(max_bytes: usize) -> InspectArchiveTransportRequest { + InspectArchiveTransportRequest { + bucket: "diagnostics".to_string(), + object: "node/metadata.json".to_string(), + public_key_pem: "-----BEGIN PUBLIC KEY-----\nopaque\n-----END PUBLIC KEY-----" + .to_string(), + max_bytes, + timeout: Duration::from_secs(2), + } + } + + #[tokio::test] + async fn inspect_archive_capability_uses_exact_v4_contract() { + let status = r#"{"state":"supported"}"#; + let body = format!( + r#"{{ + "summary": {{ + "observability": {status}, + "userspace_profiling": {status}, + "memory_sampling": {status}, + "platform": {status}, + "topology": {status}, + "cluster_snapshot": {status} + }}, + "inspect_archive": {{ + "state": {status}, + "route": "/rustfs/admin/v4/inspect/archive", + "archive_version": 1, + "content_type": "application/vnd.rustfs.inspect-archive.v1", + "encryption": "RSA-OAEP-SHA256+AES-256-GCM-CHUNKED", + "completion_contract": "authenticated-final-record-required", + "max_bytes": 16777216, + "max_duration_secs": 30, + "max_metadata_bytes_per_drive": 1048576 + }}, + "cluster_snapshot_path": "/rustfs/admin/v4/cluster/snapshot", + "cluster_snapshot_summary": null, + "topology_status": {status} + }}"# + ); + let (endpoint, receiver, handle) = + start_admin_owned_test_server("200 OK", "application/json", body); + let contract = admin_client_for_endpoint(&endpoint) + .inspect_archive_capability() + .await + .expect("exact inspect archive contract"); + contract.validate().expect("valid contract"); + let request = receiver.recv().expect("captured request"); + assert_eq!(request.method, "GET"); + assert_eq!(request.target, "/rustfs/admin/v4/runtime/capabilities"); + handle.join().expect("server thread"); + } + + #[tokio::test] + async fn inspect_archive_transport_streams_private_bounded_response() { + let payload = b"encrypted-frame-bytes"; + let response = [ + format!( + "HTTP/1.1 200 OK\r\ncontent-length: {}\r\ncontent-type: {INSPECT_ARCHIVE_CONTENT_TYPE}\r\nx-rustfs-inspect-completion: {INSPECT_ARCHIVE_COMPLETION}\r\ncache-control: no-store\r\nconnection: close\r\n\r\n", + payload.len() + ) + .into_bytes(), + payload.to_vec(), + ] + .concat(); + let (endpoint, receiver, handle) = start_admin_captured_raw_response_server(response); + let directory = tempdir().expect("temporary directory"); + let encrypted = admin_client_for_endpoint(&endpoint) + .download_inspect_archive(inspect_archive_request(1024), directory.path()) + .await + .expect("bounded response"); + assert_eq!( + std::fs::read(encrypted.file.path()).expect("staged file"), + payload + ); + #[cfg(unix)] + { + use std::os::unix::fs::PermissionsExt; + assert_eq!( + encrypted + .file + .as_file() + .metadata() + .expect("metadata") + .permissions() + .mode() + & 0o777, + 0o600 + ); + } + let request = receiver.recv().expect("captured request"); + assert_eq!(request.method, "POST"); + assert_eq!(request.target, "/rustfs/admin/v4/inspect/archive"); + let body: serde_json::Value = serde_json::from_slice(&request.body).expect("JSON body"); + assert_eq!(body["bucket"], "diagnostics"); + assert_eq!(body["object"], "node/metadata.json"); + assert!(body["public_key_pem"].as_str().is_some()); + handle.join().expect("server thread"); + } + + #[tokio::test] + async fn inspect_archive_transport_rejects_bad_headers_limits_and_write_failures() { + let bad_headers = b"HTTP/1.1 200 OK\r\ncontent-length: 1\r\ncontent-type: application/octet-stream\r\nconnection: close\r\n\r\nx".to_vec(); + let (endpoint, handle) = start_admin_raw_response_server(bad_headers); + let directory = tempdir().expect("temporary directory"); + let error = admin_client_for_endpoint(&endpoint) + .download_inspect_archive(inspect_archive_request(1024), directory.path()) + .await + .err() + .expect("contract headers must match"); + assert!(matches!(error, Error::General(_))); + handle.join().expect("server thread"); + + let response = format!( + "HTTP/1.1 200 OK\r\ncontent-length: 2\r\ncontent-type: {INSPECT_ARCHIVE_CONTENT_TYPE}\r\nx-rustfs-inspect-completion: {INSPECT_ARCHIVE_COMPLETION}\r\ncache-control: no-store\r\nconnection: close\r\n\r\nxx" + ) + .into_bytes(); + let (endpoint, handle) = start_admin_raw_response_server(response); + let error = admin_client_for_endpoint(&endpoint) + .download_inspect_archive(inspect_archive_request(1), directory.path()) + .await + .err() + .expect("declared limit"); + assert!(matches!(error, Error::RequestRejected(_))); + handle.join().expect("server thread"); + + let missing = directory.path().join("missing"); + let (endpoint, handle) = start_admin_raw_response_server( + format!( + "HTTP/1.1 200 OK\r\ncontent-length: 0\r\ncontent-type: {INSPECT_ARCHIVE_CONTENT_TYPE}\r\nx-rustfs-inspect-completion: {INSPECT_ARCHIVE_COMPLETION}\r\ncache-control: no-store\r\nconnection: close\r\n\r\n" + ) + .into_bytes(), + ); + let error = admin_client_for_endpoint(&endpoint) + .download_inspect_archive(inspect_archive_request(1024), &missing) + .await + .err() + .expect("staging write failure"); + assert!(matches!(error, Error::InvalidPath(_))); + handle.join().expect("server thread"); + } + + #[tokio::test] + async fn inspect_archive_errors_are_classified_without_echoing_server_secrets() { + for (status, expected) in [ + ("403 Forbidden", "auth"), + ("404 Not Found", "unsupported"), + ("413 Payload Too Large", "limit"), + ("504 Gateway Timeout", "timeout"), + ] { + let leaked = r#"{"message":"secretKey=do-not-echo"}"#; + let (endpoint, _receiver, handle) = start_admin_test_server(status, leaked); + let directory = tempdir().expect("temporary directory"); + let error = admin_client_for_endpoint(&endpoint) + .download_inspect_archive(inspect_archive_request(1024), directory.path()) + .await + .err() + .expect(expected); + assert!(!error.to_string().contains("do-not-echo")); + match expected { + "auth" => assert!(matches!(error, Error::Auth(_))), + "unsupported" => assert!(matches!(error, Error::UnsupportedFeature(_))), + "limit" => assert!(matches!(error, Error::RequestRejected(_))), + "timeout" => assert!(matches!(error, Error::Network(_))), + _ => unreachable!(), + } + handle.join().expect("server thread"); + } + } } diff --git a/deny.toml b/deny.toml new file mode 100644 index 0000000..d42ab0a --- /dev/null +++ b/deny.toml @@ -0,0 +1,7 @@ +# RUSTSEC-2023-0071 affects network-observable RSA private-key operations. +# Diagnostic archive RSA-OAEP decryption is performed only on the operator's +# local machine with an ephemeral or explicitly local private key; rc exposes +# no remote decryption oracle. The advisory currently has no patched release +# and explicitly permits this local-only threat model as its workaround. +[advisories] +ignore = ["RUSTSEC-2023-0071"] diff --git a/schemas/output_v3.json b/schemas/output_v3.json index 9a7ad2f..bf9a442 100644 --- a/schemas/output_v3.json +++ b/schemas/output_v3.json @@ -42,6 +42,7 @@ "health_snapshot", "cluster_snapshot", "extension_catalog", + "diagnostic_archive", "oidc", "iam_policy_entities", "iam_policy_detach", @@ -140,6 +141,26 @@ "external_plugin_flow": { "type": "object" } } }, + "diagnosticArchiveData": { + "type": "object", + "required": [ + "operation", "state", "output", "archive_version", "drive_count", + "encrypted_bytes", "plaintext_bytes", "plaintext_sha256" + ], + "properties": { + "operation": { "const": "retrieve" }, + "state": { "const": "verified" }, + "output": { "type": "string", "minLength": 1 }, + "archive_version": { "const": 1 }, + "drive_count": { "type": "integer", "minimum": 0 }, + "encrypted_bytes": { "type": "integer", "minimum": 0 }, + "plaintext_bytes": { "type": "integer", "minimum": 0 }, + "plaintext_sha256": { + "type": "string", + "pattern": "^[0-9a-f]{64}$" + } + } + }, "objectVersion": { "type": "object", "required": [ @@ -1567,7 +1588,8 @@ "type": "string", "enum": [ "general_error", "usage_error", "network_error", "auth_error", - "not_found", "conflict", "interrupted" + "not_found", "conflict", "interrupted", "invalid_target", + "server_limit", "cryptographic_verification", "local_io", "disabled" ] }, "message": { "type": "string", "minLength": 1 }, @@ -1707,6 +1729,17 @@ } ] }, + { + "allOf": [ + { "$ref": "#/definitions/successEnvelope" }, + { + "properties": { + "type": { "const": "diagnostic_archive" }, + "data": { "$ref": "#/definitions/diagnosticArchiveData" } + } + } + ] + }, { "allOf": [ { "$ref": "#/definitions/successEnvelope" },