gsadmin/rustfs
1
0
Fork 0
mirror of https://github.com/rustfs/rustfs.git synced 2026-08-27 15:37:02 +00:00
Code Issues Packages Projects Releases Wiki Activity

feat(connect): build signed offline bundles (#6579)

Browse Source 
* feat(connect): build signed offline bundles

* fix(connect): validate offline bundle inputs

* test(connect): use the target architecture

* chore(connect): scope the unsafe allowance
This commit is contained in:
Zhengchao An
2026-08-25 21:37:44 +08:00
committed by GitHub
parent 6a99edab50
commit 2bd83a5276
5 changed files with 1821 additions and 12 deletions
Download Patch File Download Diff File Expand all files Collapse all files
rustfs/src/bin/rustfs-cli.rs
+246 -7
View File
@@ -18,37 +18,46 @@
//! documented entry point for offline tooling such as `inspect bucket-meta`.
use std::fs;
use std::future::Future;
use std::io::{Read as _, Write as _};
use std::path::{Path, PathBuf};
use std::pin::Pin;
use std::process::ExitCode;
use std::time::{SystemTime, UNIX_EPOCH};
use std::time::{Duration, SystemTime, UNIX_EPOCH};
use rustfs::connect::offline::{OfflineEnrollment, OfflineKeyStore};
use rustfs::connect::offline::{
BundleContext, BundleError, BundleReceipt, OfflineEnrollment, OfflineKeyStore, collect_offline_diagnostics,
write_offline_bundle,
};
use tokio_util::sync::CancellationToken;
/// Owner read/write only. The response names the key being enrolled and the
/// challenge it answers; neither belongs to anyone else on the machine.
#[cfg(unix)]
const RESPONSE_MODE: u32 = 0o600;
const OFFLINE_RUNTIME_SHUTDOWN_TIMEOUT: Duration = Duration::from_millis(100);
const USAGE: &str = "\
Usage: rustfs-cli connect offline enroll --challenge <path|-> --output <path> [--key-dir <path>]
rustfs-cli connect offline bundle --state-dir <path> --device-name <name> --output <path> [--key-dir <path>]
Answers a Connect offline enrolment challenge without a network. Reads the
challenge from a file or from stdin when the path is `-`, verifies it against the
enrolment root compiled into this binary, mints the key being enrolled on first
use, and writes the signed response.
Builds a deterministic support bundle from the stopped runtime's persisted
inventory and bounded host diagnostics. The bundle command never uploads.
No secret is ever accepted on the command line.
";
fn main() -> ExitCode {
let arguments: Vec<String> = std::env::args().skip(1).collect();
// Offline enrolment is handled before the server dispatcher is reached, and
// the reason is the surface's whole point: `run_process` builds a Tokio
// runtime and enters the server's async main. An air-gapped enrolment must
// not start a runtime, a task, or anything that could open a socket, so the
// two paths cannot share an entry.
// Offline operations are handled before the server dispatcher so they can
// never enter the networked server runtime. Bundle collection creates only
// a current-thread runtime for its timeout and SIGINT cancellation.
if matches!(
arguments.first().map(String::as_str),
Some("connect") if matches!(arguments.get(1).map(String::as_str), Some("offline"))
@@ -70,11 +79,182 @@ fn main() -> ExitCode {
fn run_offline(arguments: &[String]) -> Result<(), String> {
match arguments.first().map(String::as_str) {
Some("enroll") => enroll(&arguments[1..]),
Some("bundle") => bundle(&arguments[1..]),
Some(other) => Err(format!("unknown offline subcommand `{other}`\n\n{USAGE}")),
None => Err(format!("missing offline subcommand\n\n{USAGE}")),
}
}
fn bundle(arguments: &[String]) -> Result<(), String> {
if !cfg!(target_os = "linux") {
return Err(BundleError::UnsupportedPlatform.to_string());
}
let mut state_directory: Option<String> = None;
let mut device_name: Option<String> = None;
let mut output_path: Option<String> = None;
let mut key_directory: Option<String> = None;
let mut index = 0;
while index < arguments.len() {
let flag = arguments[index].as_str();
let take_value = |name: &str| -> Result<String, String> {
arguments
.get(index + 1)
.cloned()
.ok_or_else(|| format!("`{name}` needs a value\n\n{USAGE}"))
};
match flag {
"--state-dir" => state_directory = Some(take_value("--state-dir")?),
"--device-name" => device_name = Some(take_value("--device-name")?),
"--output" => output_path = Some(take_value("--output")?),
"--key-dir" => key_directory = Some(take_value("--key-dir")?),
"-h" | "--help" => {
println!("{USAGE}");
return Ok(());
}
other => return Err(format!("unknown option `{other}`\n\n{USAGE}")),
}
index += 2;
}
let state_directory = state_directory.ok_or_else(|| format!("`--state-dir` is required\n\n{USAGE}"))?;
let device_name = device_name.ok_or_else(|| format!("`--device-name` is required\n\n{USAGE}"))?;
let output_path = output_path.ok_or_else(|| format!("`--output` is required\n\n{USAGE}"))?;
let key_directory = key_directory.unwrap_or_else(|| ".".to_owned());
let key = OfflineKeyStore::new(&key_directory)
.load()
.map_err(|error| error.to_string())?
.ok_or_else(|| "offline enrollment key is missing; run `connect offline enroll` first".to_owned())?;
let cancel = CancellationToken::new();
let runtime = tokio::runtime::Builder::new_current_thread()
.enable_time()
.enable_io()
.build()
.map_err(|error| format!("cannot create the offline collector runtime: {error}"))?;
let output = PathBuf::from(&output_path);
let operation = async move {
let signal = async {
tokio::signal::ctrl_c()
.await
.map_err(|error| format!("cannot listen for SIGINT: {error}"))
};
tokio::pin!(signal);
let diagnostics = tokio::select! {
biased;
signal = signal.as_mut() => {
cancel.cancel();
signal?;
return Err("offline bundle production was cancelled".to_owned());
}
result = collect_offline_diagnostics(Path::new(&state_directory), &cancel) => {
result.map_err(|error| error.to_string())?
}
};
let elapsed = SystemTime::now()
.duration_since(UNIX_EPOCH)
.map_err(|_| "the system clock is before the Unix epoch".to_owned())?;
let produced_at_unix =
i64::try_from(elapsed.as_secs()).map_err(|_| "the system clock is outside the supported range".to_owned())?;
let timestamp_millis =
u64::try_from(elapsed.as_millis()).map_err(|_| "the system clock is outside the supported range".to_owned())?;
let mut nonce = [0u8; 32];
getrandom(&mut nonce)?;
let mut uuid_random = [0u8; 10];
getrandom(&mut uuid_random)?;
let bundle_uid = uuid::Builder::from_unix_timestamp_millis(timestamp_millis, &uuid_random)
.into_uuid()
.to_string();
let inventory_captured_at = diagnostics.inventory_captured_at;
let inventory_age = diagnostics.inventory_age;
let writer_cancel = cancel.clone();
let writer = tokio::task::spawn_blocking(move || {
write_offline_bundle(
&output,
&BundleContext {
bundle_uid,
device_name,
nonce,
produced_at_unix,
},
&diagnostics.entries,
&key,
&writer_cancel,
)
});
let receipt = await_offline_writer(&cancel, signal.as_mut(), writer).await?;
Ok((receipt, inventory_captured_at, inventory_age))
};
let (runtime, (receipt, inventory_captured_at, inventory_age)) = run_offline_runtime(runtime, operation)?;
drop(runtime);
println!("Bundle: {}", receipt.bundle_uid);
println!("Device: {}", receipt.device_name);
println!("Inventory: {} ({}s old)", inventory_captured_at, inventory_age.as_secs());
println!("L0: {} entries, {} bytes", receipt.l0_count, receipt.l0_bytes);
println!("L1: {} entries, {} bytes", receipt.l1_count, receipt.l1_bytes);
println!(
"Redaction: {} ({})",
rustfs::connect::offline::redaction::REDACTION_VERSION,
rustfs::connect::offline::redaction::RULESET_HASH
);
println!("Archive: {} bytes, sha256 {}", receipt.archive_size_bytes, receipt.archive_sha256);
println!("Output: {output_path}");
println!("Upload: not performed");
Ok(())
}
async fn await_offline_writer<S>(
cancel: &CancellationToken,
mut signal: Pin<&mut S>,
mut writer: tokio::task::JoinHandle<Result<BundleReceipt, BundleError>>,
) -> Result<BundleReceipt, String>
where
S: Future<Output = Result<(), String>> + ?Sized,
{
tokio::select! {
biased;
signal = signal.as_mut() => {
cancel.cancel();
let writer = finish_offline_writer(writer.await);
match writer {
Ok(receipt) => Ok(receipt),
Err(error) => {
signal?;
Err(error)
}
}
}
result = &mut writer => finish_offline_writer(result),
}
}
fn finish_offline_writer(
result: Result<Result<BundleReceipt, BundleError>, tokio::task::JoinError>,
) -> Result<BundleReceipt, String> {
result
.map_err(|error| format!("offline bundle writer task failed: {error}"))?
.map_err(|error| error.to_string())
}
fn run_offline_runtime<T>(
runtime: tokio::runtime::Runtime,
operation: impl Future<Output = Result<T, String>>,
) -> Result<(tokio::runtime::Runtime, T), String> {
match runtime.block_on(operation) {
Ok(value) => Ok((runtime, value)),
Err(error) => {
runtime.shutdown_timeout(OFFLINE_RUNTIME_SHUTDOWN_TIMEOUT);
Err(error)
}
}
}
fn enroll(arguments: &[String]) -> Result<(), String> {
let mut challenge_path: Option<String> = None;
let mut output_path: Option<String> = None;
@@ -233,3 +413,62 @@ fn getrandom(buffer: &mut [u8]) -> Result<(), String> {
.try_fill_bytes(buffer)
.map_err(|error| format!("the operating system random source failed: {error}"))
}
#[cfg(test)]
mod tests {
use std::time::Instant;
use super::*;
#[test]
fn connect_offline_bundle_cli_shutdown_is_bounded_with_a_stuck_blocking_collector() {
let runtime = tokio::runtime::Builder::new_current_thread().build().expect("test runtime");
let operation = async {
let (started, observed) = tokio::sync::oneshot::channel();
tokio::task::spawn_blocking(move || {
started.send(()).expect("report blocking task");
std::thread::sleep(Duration::from_secs(5));
});
observed.await.expect("blocking task started");
Err::<(), _>("collector cancelled".to_owned())
};
let started_at = Instant::now();
assert!(run_offline_runtime(runtime, operation).is_err());
assert!(started_at.elapsed() < Duration::from_secs(1));
}
#[test]
fn connect_offline_bundle_cli_preserves_success_when_signal_and_committed_writer_are_ready() {
let runtime = tokio::runtime::Builder::new_current_thread().build().expect("test runtime");
let cancel = CancellationToken::new();
let temp = tempfile::tempdir().expect("output tempdir");
let output = temp.path().join("bundle.zip");
let writer_output = output.clone();
let signal = std::future::ready(Ok(()));
tokio::pin!(signal);
let writer = runtime.spawn(async move {
fs::write(writer_output, b"published bundle").expect("publish test bundle");
Ok(BundleReceipt {
bundle_uid: "0198f3a1-8000-7e50-8f61-4a5b6c7d8e94".to_owned(),
device_name: "organizations/0198f3a1-4c00-7a10-8b21-0c1d2e3f4a50/clusters/0198f3a1-5d00-7b20-9c31-1d2e3f4a5b61/clusterDevices/0198f3a1-6e00-7c30-ad41-2e3f4a5b6c72".to_owned(),
archive_size_bytes: 1,
archive_sha256: "00".repeat(32),
l0_count: 6,
l0_bytes: 1,
l1_count: 6,
l1_bytes: 1,
})
});
runtime.block_on(async {
while !writer.is_finished() {
tokio::task::yield_now().await;
}
});
assert_eq!(fs::read(&output).expect("read committed output"), b"published bundle");
let result = runtime.block_on(await_offline_writer(&cancel, signal.as_mut(), writer));
assert!(result.is_ok());
assert!(cancel.is_cancelled());
}
}
rustfs/src/connect/offline/bundle_writer.rs
+965
View File
@@ -0,0 +1,965 @@
// Copyright 2024 RustFS Team
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
// http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.
//! Deterministic, bounded support-bundle archive production.
#[cfg(target_os = "linux")]
use std::ffi::{CStr, CString};
#[cfg(target_os = "linux")]
use std::fs::{self, File};
#[cfg(target_os = "linux")]
use std::io::{Read as _, Seek as _, SeekFrom, Write as _};
use std::path::Path;
#[cfg(target_os = "linux")]
use std::path::PathBuf;
#[cfg(target_os = "linux")]
use base64::{Engine as _, engine::general_purpose::URL_SAFE_NO_PAD};
#[cfg(target_os = "linux")]
use p256::ecdsa::{Signature, SigningKey, signature::Signer as _};
#[cfg(target_os = "linux")]
use p256::pkcs8::DecodePrivateKey as _;
#[cfg(target_os = "linux")]
use rand::{TryRng as _, rngs::SysRng};
#[cfg(target_os = "linux")]
use sha2::{Digest as _, Sha256};
use thiserror::Error;
#[cfg(target_os = "linux")]
use time::{OffsetDateTime, format_description::well_known::Rfc3339};
use tokio_util::sync::CancellationToken;
#[cfg(target_os = "linux")]
use zip::{CompressionMethod, ZipWriter, write::SimpleFileOptions};
#[cfg(target_os = "linux")]
use super::collectors::{DataClassification, OfflineCollector};
#[cfg(target_os = "linux")]
use super::manifest::{
BundleIdentity, BundleManifest, BundleManifestEntry, BundleSignature, DOMAIN_TAG, SIGNATURE_FILE, SIGNED_FILE,
};
use super::manifest_entry::ManifestEntry;
#[cfg(target_os = "linux")]
use super::redaction::{REDACTION_VERSION, RULESET_HASH, RedactionSource, redact_json};
use crate::connect::identity::DeviceIdentity;
#[cfg(target_os = "linux")]
const ENTRY_TYPE: &str = "offline-diagnostic";
#[cfg(target_os = "linux")]
const ENTRY_LIMIT: usize = 16 * 1024;
#[cfg(target_os = "linux")]
const MAX_SAFE_INTEGER: u64 = 9_007_199_254_740_991;
#[cfg(target_os = "linux")]
const MANIFEST_LIMIT: usize = 1024 * 1024;
#[cfg(target_os = "linux")]
const SIGNATURE_LIMIT: usize = 4 * 1024;
#[cfg(target_os = "linux")]
const ARCHIVE_LIMIT: u64 = 256 * 1024 * 1024;
#[cfg(target_os = "linux")]
const OUTPUT_MODE: u32 = 0o600;
#[cfg(target_os = "linux")]
const ENTRY_SPECS: [(OfflineCollector, &str); 12] = [
(OfflineCollector::RustfsVersion, "offline/rustfs-version.json"),
(OfflineCollector::NodeCount, "offline/node-count.json"),
(OfflineCollector::DriveCount, "offline/drive-count.json"),
(OfflineCollector::CapacityUsedBytes, "offline/capacity-used-bytes.json"),
(OfflineCollector::CapacityTotalBytes, "offline/capacity-total-bytes.json"),
(OfflineCollector::CoarseHealthFlags, "offline/coarse-health-flags.json"),
(OfflineCollector::OsSummary, "offline/os-summary.json"),
(OfflineCollector::KernelSummary, "offline/kernel-summary.json"),
(OfflineCollector::CpuSummary, "offline/cpu-summary.json"),
(OfflineCollector::MemorySummary, "offline/memory-summary.json"),
(OfflineCollector::FilesystemSummary, "offline/filesystem-summary.json"),
(OfflineCollector::NetworkSummary, "offline/network-summary.json"),
];
#[derive(Clone, Debug, PartialEq, Eq)]
pub struct BundleContext {
pub bundle_uid: String,
pub device_name: String,
pub nonce: [u8; 32],
pub produced_at_unix: i64,
}
#[derive(Clone, Debug, PartialEq, Eq)]
pub struct BundleReceipt {
pub bundle_uid: String,
pub device_name: String,
pub archive_size_bytes: u64,
pub archive_sha256: String,
pub l0_count: usize,
pub l0_bytes: u64,
pub l1_count: usize,
pub l1_bytes: u64,
}
#[derive(Debug, Error)]
pub enum BundleError {
#[error("offline bundle production was cancelled")]
Cancelled,
#[error("bundleUid or deviceName is not a canonical UUIDv7 resource name")]
InvalidIdentity,
#[error("offline diagnostic entries do not match the fixed bundle schema")]
InvalidEntries,
#[error("offline bundle metadata is not representable")]
InvalidMetadata,
#[error("offline bundle exceeds its {kind} size limit")]
TooLarge { kind: &'static str },
#[error("the device key cannot sign the offline bundle")]
Signing,
#[error("the operating system random source failed")]
Random,
#[error("offline bundles require Linux file security semantics")]
UnsupportedPlatform,
#[error("the offline bundle output directory is not private and stable")]
UnsafeOutput,
#[error("cannot write the offline bundle: {0}")]
Io(#[from] std::io::Error),
#[error("cannot encode the offline bundle archive: {0}")]
Zip(#[from] zip::result::ZipError),
#[error("the bundle was published, but its directory could not be made durable: {0}")]
DurabilityAfterCommit(std::io::Error),
}
pub fn write_offline_bundle(
output: &Path,
context: &BundleContext,
entries: &[ManifestEntry],
key: &DeviceIdentity,
cancel: &CancellationToken,
) -> Result<BundleReceipt, BundleError> {
#[cfg(target_os = "linux")]
{
write_offline_bundle_unix(output, context, entries, key, cancel)
}
#[cfg(not(target_os = "linux"))]
{
let _ = (output, context, entries, key, cancel);
Err(BundleError::UnsupportedPlatform)
}
}
#[cfg(target_os = "linux")]
fn write_offline_bundle_unix(
output: &Path,
context: &BundleContext,
entries: &[ManifestEntry],
key: &DeviceIdentity,
cancel: &CancellationToken,
) -> Result<BundleReceipt, BundleError> {
check_cancel(cancel)?;
let identity = BundleIdentity::parse(&context.bundle_uid, &context.device_name).ok_or(BundleError::InvalidIdentity)?;
validate_entries(entries)?;
let produced_at = OffsetDateTime::from_unix_timestamp(context.produced_at_unix).map_err(|_| BundleError::InvalidMetadata)?;
let produced_at = produced_at.format(&Rfc3339).map_err(|_| BundleError::InvalidMetadata)?;
let nonce = URL_SAFE_NO_PAD.encode(context.nonce);
let device_key_id = hex_lower(&Sha256::digest(key.public_key_der()));
let manifest_entries = entries
.iter()
.zip(ENTRY_SPECS)
.map(|(entry, (_, path))| BundleManifestEntry {
path,
entry_type: ENTRY_TYPE,
size_bytes: entry.canonical_json.len() as u64,
sha256: hex_lower(&Sha256::digest(entry.canonical_json.as_bytes())),
classification: entry.classification,
})
.collect::<Vec<_>>();
let manifest = BundleManifest::new(&identity, &device_key_id, &nonce, &produced_at, &manifest_entries);
let manifest_bytes = serde_json::to_vec(&manifest).map_err(|_| BundleError::InvalidMetadata)?;
if manifest_bytes.len() > MANIFEST_LIMIT {
return Err(BundleError::TooLarge { kind: "manifest" });
}
let signature = sign(key, &manifest_bytes)?;
let signature_bytes =
serde_json::to_vec(&BundleSignature::new(&device_key_id, &signature)).map_err(|_| BundleError::InvalidMetadata)?;
if signature_bytes.len() > SIGNATURE_LIMIT {
return Err(BundleError::TooLarge { kind: "signature" });
}
let (mut temporary, file) = TemporaryBundle::create(output)?;
let options = SimpleFileOptions::DEFAULT
.compression_method(CompressionMethod::Stored)
.system(zip::System::Unix)
.unix_permissions(OUTPUT_MODE);
let mut archive = ZipWriter::new(file);
#[cfg(test)]
test_support::at(test_support::Stage::EntryWrite)?;
for ((entry, (_, path)), manifest_entry) in entries.iter().zip(ENTRY_SPECS).zip(&manifest_entries) {
check_cancel(cancel)?;
debug_assert_eq!(entry.canonical_json.len() as u64, manifest_entry.size_bytes);
archive.start_file(path, options)?;
archive.write_all(entry.canonical_json.as_bytes())?;
}
check_cancel(cancel)?;
archive.start_file(SIGNED_FILE, options)?;
archive.write_all(&manifest_bytes)?;
archive.start_file(SIGNATURE_FILE, options)?;
archive.write_all(&signature_bytes)?;
let mut file = archive.finish()?;
#[cfg(test)]
test_support::at(test_support::Stage::FileSync)?;
file.sync_all()?;
let archive_size_bytes = file.metadata()?.len();
if archive_size_bytes > ARCHIVE_LIMIT {
return Err(BundleError::TooLarge { kind: "archive" });
}
file.seek(SeekFrom::Start(0))?;
let mut digest = Sha256::new();
let mut buffer = [0u8; 64 * 1024];
loop {
let read = file.read(&mut buffer)?;
if read == 0 {
break;
}
digest.update(&buffer[..read]);
}
#[cfg(test)]
test_support::at(test_support::Stage::BeforePublish)?;
check_cancel(cancel)?;
drop(file);
temporary.publish()?;
let mut receipt = BundleReceipt {
bundle_uid: identity.bundle_uid,
device_name: identity.device_name,
archive_size_bytes,
archive_sha256: hex_lower(&digest.finalize()),
l0_count: 0,
l0_bytes: 0,
l1_count: 0,
l1_bytes: 0,
};
for entry in &manifest_entries {
match entry.classification {
DataClassification::L0 => {
receipt.l0_count += 1;
receipt.l0_bytes += entry.size_bytes;
}
DataClassification::L1 => {
receipt.l1_count += 1;
receipt.l1_bytes += entry.size_bytes;
}
}
}
Ok(receipt)
}
#[cfg(target_os = "linux")]
fn validate_entries(entries: &[ManifestEntry]) -> Result<(), BundleError> {
if entries.len() != ENTRY_SPECS.len() {
return Err(BundleError::InvalidEntries);
}
for (entry, (collector, _)) in entries.iter().zip(ENTRY_SPECS) {
if entry.canonical_json.len() > ENTRY_LIMIT
|| entry.field_id != collector.field_id()
|| entry.classification != collector.classification()
|| entry.redaction_version != REDACTION_VERSION
|| entry.ruleset_hash != RULESET_HASH
{
return Err(BundleError::InvalidEntries);
}
let Ok(value) = serde_json::from_str::<serde_json::Value>(&entry.canonical_json) else {
return Err(BundleError::InvalidEntries);
};
let Some(object) = value.as_object() else {
return Err(BundleError::InvalidEntries);
};
let Some(payload) = object.get(collector.field_name()) else {
return Err(BundleError::InvalidEntries);
};
let Ok(redacted) = redact_json(RedactionSource::OfflineDiagnostic, entry.canonical_json.as_bytes()) else {
return Err(BundleError::InvalidEntries);
};
if redacted.canonical_json != entry.canonical_json
|| redacted.redaction_version != REDACTION_VERSION
|| redacted.ruleset_hash != RULESET_HASH
|| object.len() != 1
|| !valid_payload(collector, payload)
|| serde_json::to_string(&value).ok().as_deref() != Some(entry.canonical_json.as_str())
{
return Err(BundleError::InvalidEntries);
}
}
let used = entry_u64(&entries[3], "capacityUsedBytes").ok_or(BundleError::InvalidEntries)?;
let total = entry_u64(&entries[4], "capacityTotalBytes").ok_or(BundleError::InvalidEntries)?;
if used > total {
return Err(BundleError::InvalidEntries);
}
Ok(())
}
#[cfg(target_os = "linux")]
fn valid_payload(collector: OfflineCollector, value: &serde_json::Value) -> bool {
match collector {
OfflineCollector::RustfsVersion => value.as_str().is_some_and(valid_rustfs_version),
OfflineCollector::NodeCount => value.as_u64().is_some_and(|count| (1..=4096).contains(&count)),
OfflineCollector::DriveCount => value.as_u64().is_some_and(|count| count <= 1_048_576),
OfflineCollector::CapacityUsedBytes | OfflineCollector::CapacityTotalBytes => {
value.as_u64().is_some_and(|bytes| bytes <= MAX_SAFE_INTEGER)
}
OfflineCollector::CoarseHealthFlags => value.as_array().is_some_and(|flags| {
ordered_known_strings(
flags,
&[
"capacity.critical",
"capacity.warning",
"clock.skew",
"cluster.degraded",
"cluster.healing",
"cluster.readonly",
"drive.offline",
"node.offline",
],
)
}),
OfflineCollector::OsSummary | OfflineCollector::KernelSummary => value.is_string(),
OfflineCollector::CpuSummary => value.as_object().is_some_and(|object| {
object.len() == 2
&& object.get("architecture").and_then(serde_json::Value::as_str) == Some(std::env::consts::ARCH)
&& object.get("cores").and_then(serde_json::Value::as_u64).is_some()
}),
OfflineCollector::MemorySummary => value.as_object().is_some_and(|object| {
object.len() == 2
&& object.get("totalBytes").and_then(serde_json::Value::as_u64).is_some()
&& object.get("underPressure").and_then(serde_json::Value::as_bool).is_some()
}),
OfflineCollector::FilesystemSummary => value.as_array().is_some_and(ordered_strings),
OfflineCollector::NetworkSummary => value.as_object().is_some_and(|object| {
object.len() == 2
&& object.get("bondCount").and_then(serde_json::Value::as_u64).is_some()
&& object.get("interfaceCount").and_then(serde_json::Value::as_u64).is_some()
}),
}
}
#[cfg(target_os = "linux")]
fn valid_rustfs_version(version: &str) -> bool {
let mut components = version.split('.');
(0..3).all(|_| {
components.next().is_some_and(|component| {
!component.is_empty()
&& component.len() <= 4
&& (component == "0" || !component.starts_with('0'))
&& component.parse::<u16>().is_ok_and(|value| value <= 9999)
})
}) && components.next().is_none()
}
#[cfg(target_os = "linux")]
fn ordered_strings(values: &[serde_json::Value]) -> bool {
let mut previous = None;
for value in values {
let Some(value) = value.as_str() else {
return false;
};
if previous.is_some_and(|previous| previous >= value) {
return false;
}
previous = Some(value);
}
true
}
#[cfg(target_os = "linux")]
fn ordered_known_strings(values: &[serde_json::Value], allowed: &[&str]) -> bool {
ordered_strings(values)
&& values
.iter()
.all(|value| value.as_str().is_some_and(|value| allowed.binary_search(&value).is_ok()))
}
#[cfg(target_os = "linux")]
fn entry_u64(entry: &ManifestEntry, field: &str) -> Option<u64> {
serde_json::from_str::<serde_json::Value>(&entry.canonical_json)
.ok()?
.get(field)?
.as_u64()
}
#[cfg(target_os = "linux")]
fn sign(key: &DeviceIdentity, manifest: &[u8]) -> Result<String, BundleError> {
let pkcs8 = key.to_pkcs8_der().map_err(|_| BundleError::Signing)?;
let signing_key = SigningKey::from_pkcs8_der(pkcs8.as_slice()).map_err(|_| BundleError::Signing)?;
let mut input = Vec::with_capacity(DOMAIN_TAG.len() + 1 + manifest.len());
input.extend_from_slice(DOMAIN_TAG.as_bytes());
input.push(0);
input.extend_from_slice(manifest);
let signature: Signature = signing_key.sign(&input);
Ok(URL_SAFE_NO_PAD.encode(signature.normalize_s().to_bytes()))
}
#[cfg(target_os = "linux")]
fn check_cancel(cancel: &CancellationToken) -> Result<(), BundleError> {
if cancel.is_cancelled() {
Err(BundleError::Cancelled)
} else {
Ok(())
}
}
#[cfg(target_os = "linux")]
fn hex_lower(bytes: &[u8]) -> String {
use std::fmt::Write as _;
let mut output = String::with_capacity(bytes.len() * 2);
for byte in bytes {
write!(&mut output, "{byte:02x}").expect("writing to a string cannot fail");
}
output
}
#[cfg(target_os = "linux")]
struct TemporaryBundle {
directory: File,
directory_identity: (u64, u64),
parent_path: PathBuf,
temporary_name: CString,
output_name: CString,
identity: (u64, u64),
published: bool,
}
#[cfg(target_os = "linux")]
impl TemporaryBundle {
fn create(output: &Path) -> Result<(Self, File), BundleError> {
let parent = output
.parent()
.filter(|path| !path.as_os_str().is_empty())
.unwrap_or(Path::new("."));
let directory = open_directory(parent)?;
validate_directory(&directory)?;
let directory_identity = file_identity(&directory)?;
let output_name = c_name(
output
.file_name()
.ok_or_else(|| std::io::Error::from(std::io::ErrorKind::InvalidInput))?,
)?;
for _ in 0..16 {
let mut random = [0u8; 16];
SysRng.try_fill_bytes(&mut random).map_err(|_| BundleError::Random)?;
let temporary_name = CString::new(format!(".bundle.{}.tmp", hex_lower(&random)))
.map_err(|_| std::io::Error::from(std::io::ErrorKind::InvalidInput))?;
match create_file_at(&directory, &temporary_name) {
Ok(file) => {
use std::os::unix::fs::PermissionsExt as _;
let mut created = CreatedFile::new(&directory, &temporary_name, &file);
#[cfg(test)]
test_support::at(test_support::Stage::Identity)?;
let identity = file_identity(&file)?;
created.disarm();
drop(created);
let temporary = Self {
directory,
directory_identity,
parent_path: parent.to_owned(),
temporary_name,
output_name,
identity,
published: false,
};
#[cfg(test)]
test_support::at(test_support::Stage::Permissions)?;
file.set_permissions(fs::Permissions::from_mode(OUTPUT_MODE))?;
if validate_regular_file(&file)? != identity {
return Err(BundleError::UnsafeOutput);
}
return Ok((temporary, file));
}
Err(error) if error.kind() == std::io::ErrorKind::AlreadyExists => continue,
Err(error) => return Err(error.into()),
}
}
Err(std::io::Error::new(std::io::ErrorKind::AlreadyExists, "cannot allocate a unique temporary bundle").into())
}
fn publish(&mut self) -> Result<(), BundleError> {
validate_directory(&self.directory)?;
self.validate_parent_path()?;
let staged = open_file_at(&self.directory, &self.temporary_name)?;
if validate_regular_file(&staged)? != self.identity {
return Err(BundleError::UnsafeOutput);
}
#[cfg(test)]
test_support::at(test_support::Stage::Rename)?;
rename_at(&self.directory, &self.temporary_name, &self.output_name)?;
self.published = true;
let published = open_file_at(&self.directory, &self.output_name).map_err(BundleError::DurabilityAfterCommit)?;
if validate_regular_file(&published).map_err(|error| match error {
BundleError::Io(error) => BundleError::DurabilityAfterCommit(error),
_ => BundleError::DurabilityAfterCommit(std::io::Error::other("published bundle identity is unsafe")),
})? != self.identity
{
return Err(BundleError::DurabilityAfterCommit(std::io::Error::other(
"published bundle identity changed",
)));
}
#[cfg(test)]
test_support::at(test_support::Stage::DirectorySync).map_err(BundleError::DurabilityAfterCommit)?;
self.directory.sync_all().map_err(BundleError::DurabilityAfterCommit)?;
self.validate_parent_path().map_err(|error| {
BundleError::DurabilityAfterCommit(std::io::Error::other(format!(
"offline bundle output directory changed after publication: {error}"
)))
})?;
Ok(())
}
fn validate_parent_path(&self) -> Result<(), BundleError> {
let current_directory = open_directory(&self.parent_path)?;
validate_directory(&current_directory)?;
if file_identity(&current_directory)? != self.directory_identity {
return Err(BundleError::UnsafeOutput);
}
Ok(())
}
}
#[cfg(target_os = "linux")]
impl Drop for TemporaryBundle {
fn drop(&mut self) {
if !self.published
&& path_identity_at(&self.directory, &self.temporary_name).is_ok_and(|identity| identity == self.identity)
{
let _ = unlink_at(&self.directory, &self.temporary_name);
}
}
}
#[cfg(target_os = "linux")]
struct CreatedFile<'a> {
directory: &'a File,
name: &'a CStr,
file: &'a File,
armed: bool,
}
#[cfg(target_os = "linux")]
impl<'a> CreatedFile<'a> {
fn new(directory: &'a File, name: &'a CStr, file: &'a File) -> Self {
Self {
directory,
name,
file,
armed: true,
}
}
fn disarm(&mut self) {
self.armed = false;
}
}
#[cfg(target_os = "linux")]
impl Drop for CreatedFile<'_> {
fn drop(&mut self) {
if self.armed
&& file_identity(self.file)
.is_ok_and(|expected| path_identity_at(self.directory, self.name).is_ok_and(|identity| identity == expected))
{
let _ = unlink_at(self.directory, self.name);
}
}
}
#[cfg(target_os = "linux")]
fn open_directory(path: &Path) -> Result<File, BundleError> {
use std::os::fd::AsRawFd as _;
use std::path::Component;
let root = if path.is_absolute() { c"/" } else { c"." };
let mut directory = open_directory_at(libc::AT_FDCWD, root)?;
for component in path.components() {
match component {
Component::RootDir | Component::CurDir => {}
Component::Normal(name) => {
let name = c_name(name)?;
directory = open_directory_at(directory.as_raw_fd(), &name)?;
}
Component::ParentDir | Component::Prefix(_) => return Err(BundleError::UnsafeOutput),
}
}
Ok(directory)
}
#[cfg(target_os = "linux")]
#[allow(unsafe_code)]
fn open_directory_at(parent: std::os::fd::RawFd, name: &CStr) -> std::io::Result<File> {
use std::os::fd::FromRawFd as _;
// SAFETY: the parent descriptor and C string are live; a successful descriptor is transferred to File.
let descriptor = unsafe {
libc::openat(
parent,
name.as_ptr(),
libc::O_RDONLY | libc::O_CLOEXEC | libc::O_NOFOLLOW | libc::O_DIRECTORY,
)
};
if descriptor < 0 {
return Err(std::io::Error::last_os_error());
}
// SAFETY: openat returned a new owned descriptor.
Ok(unsafe { File::from_raw_fd(descriptor) })
}
#[cfg(target_os = "linux")]
fn validate_directory(directory: &File) -> Result<(), BundleError> {
use std::os::unix::fs::{MetadataExt as _, PermissionsExt as _};
let metadata = directory.metadata()?;
let mode = metadata.permissions().mode() & 0o7777;
if !metadata.is_dir() || metadata.uid() != process_uid() || mode & 0o022 != 0 {
return Err(BundleError::UnsafeOutput);
}
Ok(())
}
#[cfg(target_os = "linux")]
#[allow(unsafe_code)]
fn create_file_at(directory: &File, name: &CStr) -> std::io::Result<File> {
use std::os::fd::{AsRawFd as _, FromRawFd as _};
// SAFETY: the directory descriptor and C string are live; a successful descriptor is transferred to File.
let descriptor = unsafe {
libc::openat(
directory.as_raw_fd(),
name.as_ptr(),
libc::O_RDWR | libc::O_CREAT | libc::O_EXCL | libc::O_CLOEXEC | libc::O_NOFOLLOW,
OUTPUT_MODE,
)
};
if descriptor < 0 {
return Err(std::io::Error::last_os_error());
}
// SAFETY: openat returned a new owned descriptor.
Ok(unsafe { File::from_raw_fd(descriptor) })
}
#[cfg(target_os = "linux")]
#[allow(unsafe_code)]
fn open_file_at(directory: &File, name: &CStr) -> std::io::Result<File> {
use std::os::fd::{AsRawFd as _, FromRawFd as _};
// SAFETY: the directory descriptor and C string are live; a successful descriptor is transferred to File.
let descriptor = unsafe {
libc::openat(
directory.as_raw_fd(),
name.as_ptr(),
libc::O_RDONLY | libc::O_CLOEXEC | libc::O_NOFOLLOW | libc::O_NONBLOCK,
)
};
if descriptor < 0 {
return Err(std::io::Error::last_os_error());
}
// SAFETY: openat returned a new owned descriptor.
Ok(unsafe { File::from_raw_fd(descriptor) })
}
#[cfg(target_os = "linux")]
fn validate_regular_file(file: &File) -> Result<(u64, u64), BundleError> {
use std::os::unix::fs::{MetadataExt as _, PermissionsExt as _};
let metadata = file.metadata()?;
if !metadata.is_file()
|| metadata.uid() != process_uid()
|| metadata.permissions().mode() & 0o7777 != OUTPUT_MODE
|| metadata.nlink() != 1
{
return Err(BundleError::UnsafeOutput);
}
Ok((metadata.dev(), metadata.ino()))
}
#[cfg(target_os = "linux")]
fn file_identity(file: &File) -> Result<(u64, u64), BundleError> {
use std::os::unix::fs::MetadataExt as _;
let metadata = file.metadata()?;
Ok((metadata.dev(), metadata.ino()))
}
#[cfg(target_os = "linux")]
#[allow(unsafe_code)]
fn path_identity_at(directory: &File, name: &CStr) -> std::io::Result<(u64, u64)> {
use std::mem::MaybeUninit;
use std::os::fd::AsRawFd as _;
let mut metadata = MaybeUninit::<libc::stat>::uninit();
// SAFETY: the directory descriptor and C string remain live, and metadata points to writable storage.
if unsafe { libc::fstatat(directory.as_raw_fd(), name.as_ptr(), metadata.as_mut_ptr(), libc::AT_SYMLINK_NOFOLLOW) } != 0 {
return Err(std::io::Error::last_os_error());
}
// SAFETY: fstatat initialized metadata after returning success.
let metadata = unsafe { metadata.assume_init() };
if metadata.st_mode & libc::S_IFMT != libc::S_IFREG {
return Err(std::io::Error::other("staged bundle is not a regular file"));
}
Ok((metadata.st_dev, metadata.st_ino))
}
#[cfg(target_os = "linux")]
#[allow(unsafe_code)]
fn rename_at(directory: &File, source: &CStr, destination: &CStr) -> std::io::Result<()> {
use std::os::fd::AsRawFd as _;
// SAFETY: both C strings and the directory descriptor remain live for the call.
if unsafe { libc::renameat(directory.as_raw_fd(), source.as_ptr(), directory.as_raw_fd(), destination.as_ptr()) } == 0 {
Ok(())
} else {
Err(std::io::Error::last_os_error())
}
}
#[cfg(target_os = "linux")]
#[allow(unsafe_code)]
fn unlink_at(directory: &File, name: &CStr) -> std::io::Result<()> {
use std::os::fd::AsRawFd as _;
// SAFETY: the C string and directory descriptor remain live for the call.
if unsafe { libc::unlinkat(directory.as_raw_fd(), name.as_ptr(), 0) } == 0 {
Ok(())
} else {
Err(std::io::Error::last_os_error())
}
}
#[cfg(target_os = "linux")]
fn c_name(name: &std::ffi::OsStr) -> Result<CString, BundleError> {
use std::os::unix::ffi::OsStrExt as _;
CString::new(name.as_bytes())
.map_err(|_| std::io::Error::from(std::io::ErrorKind::InvalidInput))
.map_err(Into::into)
}
#[cfg(target_os = "linux")]
#[allow(unsafe_code)]
fn process_uid() -> u32 {
// SAFETY: geteuid has no pointer arguments or caller preconditions.
unsafe { libc::geteuid() }
}
#[cfg(all(test, target_os = "linux"))]
mod test_support {
use std::cell::RefCell;
use std::io;
use tokio_util::sync::CancellationToken;
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub(super) enum Stage {
Identity,
Permissions,
EntryWrite,
FileSync,
BeforePublish,
Rename,
DirectorySync,
}
pub(super) enum Action {
Error(i32),
Cancel(CancellationToken),
Run(Box<dyn FnOnce()>),
}
thread_local! {
static NEXT: RefCell<Option<(Stage, Action)>> = const { RefCell::new(None) };
}
pub(super) fn set(stage: Stage, action: Action) {
NEXT.with(|next| {
assert!(next.borrow_mut().replace((stage, action)).is_none());
});
}
pub(super) fn at(stage: Stage) -> io::Result<()> {
let action = NEXT.with(|next| {
let mut next = next.borrow_mut();
next.as_ref()
.is_some_and(|(expected, _)| *expected == stage)
.then(|| next.take().expect("fault action"))
});
match action {
Some((_, Action::Error(code))) => Err(io::Error::from_raw_os_error(code)),
Some((_, Action::Cancel(cancel))) => {
cancel.cancel();
Ok(())
}
Some((_, Action::Run(action))) => {
action();
Ok(())
}
None => Ok(()),
}
}
}
#[cfg(all(test, target_os = "linux"))]
mod tests {
use serde_json::{Map, json};
use super::test_support::{self, Action, Stage};
use super::*;
fn context() -> BundleContext {
BundleContext {
bundle_uid: "0198f3a1-8000-7e50-8f61-4a5b6c7d8e94".to_owned(),
device_name: "organizations/0198f3a1-4c00-7a10-8b21-0c1d2e3f4a50/clusters/0198f3a1-5d00-7b20-9c31-1d2e3f4a5b61/clusterDevices/0198f3a1-6e00-7c30-ad41-2e3f4a5b6c72".to_owned(),
nonce: [7; 32],
produced_at_unix: 1_777_860_000,
}
}
fn entries() -> Vec<ManifestEntry> {
ENTRY_SPECS
.iter()
.map(|(collector, _)| {
let mut value = Map::new();
value.insert(
collector.field_name().to_owned(),
match collector {
OfflineCollector::RustfsVersion => json!("1.4.2"),
OfflineCollector::NodeCount => json!(2),
OfflineCollector::DriveCount => json!(3),
OfflineCollector::CapacityUsedBytes => json!(1500),
OfflineCollector::CapacityTotalBytes => json!(6000),
OfflineCollector::CoarseHealthFlags => json!(["cluster.degraded", "drive.offline"]),
OfflineCollector::OsSummary => json!("Linux"),
OfflineCollector::KernelSummary => json!("6.8.0"),
OfflineCollector::CpuSummary => json!({"architecture": std::env::consts::ARCH, "cores": 8}),
OfflineCollector::MemorySummary => json!({"totalBytes": 17179869184_u64, "underPressure": false}),
OfflineCollector::FilesystemSummary => json!(["ext4", "xfs"]),
OfflineCollector::NetworkSummary => json!({"bondCount": 1, "interfaceCount": 4}),
},
);
ManifestEntry {
field_id: collector.field_id(),
classification: collector.classification(),
canonical_json: serde_json::to_string(&value).expect("test entry"),
redaction_version: REDACTION_VERSION,
ruleset_hash: RULESET_HASH,
redacted_count: 0,
}
})
.collect()
}
fn temp_residue(directory: &Path) -> usize {
fs::read_dir(directory)
.expect("read test directory")
.filter_map(Result::ok)
.filter(|entry| entry.file_name().to_string_lossy().starts_with(".bundle."))
.count()
}
#[test]
fn connect_offline_bundle_faults_preserve_the_commit_boundary() {
let temp = tempfile::tempdir().expect("bundle tempdir");
let output = temp.path().join("bundle.zip");
let key = DeviceIdentity::generate();
for stage in [
Stage::Identity,
Stage::Permissions,
Stage::EntryWrite,
Stage::FileSync,
Stage::Rename,
] {
fs::write(&output, b"old bundle").expect("old output");
test_support::set(stage, Action::Error(libc::ENOSPC));
assert!(matches!(
write_offline_bundle(&output, &context(), &entries(), &key, &CancellationToken::new()),
Err(BundleError::Io(_))
));
assert_eq!(fs::read(&output).expect("preserved output"), b"old bundle");
assert_eq!(temp_residue(temp.path()), 0);
}
fs::write(&output, b"old bundle").expect("old output");
let cancel = CancellationToken::new();
test_support::set(Stage::BeforePublish, Action::Cancel(cancel.clone()));
assert!(matches!(
write_offline_bundle(&output, &context(), &entries(), &key, &cancel),
Err(BundleError::Cancelled)
));
assert_eq!(fs::read(&output).expect("preserved output"), b"old bundle");
assert_eq!(temp_residue(temp.path()), 0);
let original = temp.path().join("original");
let moved = temp.path().join("moved");
fs::create_dir(&original).expect("original output directory");
let swapped_output = original.join("bundle.zip");
let swap_original = original.clone();
let swap_moved = moved.clone();
test_support::set(
Stage::BeforePublish,
Action::Run(Box::new(move || {
fs::rename(&swap_original, &swap_moved).expect("move anchored directory");
fs::create_dir(&swap_original).expect("replacement directory");
})),
);
assert!(matches!(
write_offline_bundle(&swapped_output, &context(), &entries(), &key, &CancellationToken::new()),
Err(BundleError::UnsafeOutput)
));
assert!(!swapped_output.exists());
assert_eq!(temp_residue(&moved), 0);
use std::os::unix::fs::symlink;
let target = temp.path().join("ancestor-target");
let private = target.join("private");
fs::create_dir_all(&private).expect("symlink target directory");
let link = temp.path().join("ancestor-link");
symlink(&target, &link).expect("ancestor symlink");
let escaped_output = link.join("private/bundle.zip");
assert!(write_offline_bundle(&escaped_output, &context(), &entries(), &key, &CancellationToken::new()).is_err());
assert!(!private.join("bundle.zip").exists());
let original = temp.path().join("publish-original");
let moved = temp.path().join("publish-moved");
fs::create_dir(&original).expect("original publish directory");
let swapped_output = original.join("bundle.zip");
let swap_original = original.clone();
let swap_moved = moved.clone();
test_support::set(
Stage::Rename,
Action::Run(Box::new(move || {
fs::rename(&swap_original, &swap_moved).expect("move directory during publication");
fs::create_dir(&swap_original).expect("replacement publish directory");
})),
);
assert!(matches!(
write_offline_bundle(&swapped_output, &context(), &entries(), &key, &CancellationToken::new()),
Err(BundleError::DurabilityAfterCommit(_))
));
assert!(!swapped_output.exists());
assert!(
fs::read(moved.join("bundle.zip"))
.expect("committed bundle")
.starts_with(b"PK")
);
assert_eq!(temp_residue(&moved), 0);
test_support::set(Stage::DirectorySync, Action::Error(libc::EIO));
assert!(matches!(
write_offline_bundle(&output, &context(), &entries(), &key, &CancellationToken::new()),
Err(BundleError::DurabilityAfterCommit(_))
));
assert!(fs::read(&output).expect("published output").starts_with(b"PK"));
assert_eq!(temp_residue(temp.path()), 0);
}
}
rustfs/src/connect/offline/manifest.rs
+147
View File
@@ -0,0 +1,147 @@
// Copyright 2024 RustFS Team
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
// http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.
//! Exact support-bundle manifest and detached-signature documents.
use serde::Serialize;
use uuid::{Uuid, Variant, Version};
use super::collectors::DataClassification;
use super::redaction::{REDACTION_VERSION, RULESET_HASH};
pub(super) const FORMAT_VERSION: &str = "rustfs.connect.support.bundleManifest/1";
pub(super) const PROTOCOL_VERSION: &str = "v1";
pub(super) const CLASSIFICATION_REGISTRY_VERSION: u8 = 1;
pub(super) const SIGNATURE_ALGORITHM: &str = "ES256";
pub(super) const SIGNED_FILE: &str = "manifest.json";
pub(super) const SIGNATURE_FILE: &str = "manifest.sig";
pub(super) const DOMAIN_TAG: &str = "rustfs-support-bundle-v1";
#[derive(Debug, Serialize)]
#[serde(rename_all = "camelCase")]
pub(super) struct BundleManifest<'a> {
format_version: &'static str,
protocol_version: &'static str,
bundle_uid: &'a str,
organization_name: &'a str,
cluster_name: &'a str,
device_name: &'a str,
device_key_id: &'a str,
nonce: &'a str,
produced_at: &'a str,
redaction_version: &'static str,
ruleset_hash: &'static str,
classification_registry_version: u8,
entries: &'a [BundleManifestEntry],
}
impl<'a> BundleManifest<'a> {
pub(super) fn new(
identity: &'a BundleIdentity,
device_key_id: &'a str,
nonce: &'a str,
produced_at: &'a str,
entries: &'a [BundleManifestEntry],
) -> Self {
Self {
format_version: FORMAT_VERSION,
protocol_version: PROTOCOL_VERSION,
bundle_uid: &identity.bundle_uid,
organization_name: &identity.organization_name,
cluster_name: &identity.cluster_name,
device_name: &identity.device_name,
device_key_id,
nonce,
produced_at,
redaction_version: REDACTION_VERSION,
ruleset_hash: RULESET_HASH,
classification_registry_version: CLASSIFICATION_REGISTRY_VERSION,
entries,
}
}
}
#[derive(Debug, Serialize)]
#[serde(rename_all = "camelCase")]
pub(super) struct BundleManifestEntry {
pub(super) path: &'static str,
#[serde(rename = "type")]
pub(super) entry_type: &'static str,
pub(super) size_bytes: u64,
pub(super) sha256: String,
pub(super) classification: DataClassification,
}
#[derive(Debug, Serialize)]
#[serde(rename_all = "camelCase")]
pub(super) struct BundleSignature<'a> {
algorithm: &'static str,
key_id: &'a str,
value: &'a str,
signed_file: &'static str,
domain_separation_tag: &'static str,
}
impl<'a> BundleSignature<'a> {
pub(super) fn new(key_id: &'a str, value: &'a str) -> Self {
Self {
algorithm: SIGNATURE_ALGORITHM,
key_id,
value,
signed_file: SIGNED_FILE,
domain_separation_tag: DOMAIN_TAG,
}
}
}
#[derive(Clone, Debug, PartialEq, Eq)]
pub(super) struct BundleIdentity {
pub(super) bundle_uid: String,
pub(super) organization_name: String,
pub(super) cluster_name: String,
pub(super) device_name: String,
}
impl BundleIdentity {
pub(super) fn parse(bundle_uid: &str, device_name: &str) -> Option<Self> {
if !is_uuid_v7(bundle_uid) {
return None;
}
let parts = device_name.split('/').collect::<Vec<_>>();
if parts.len() != 6
|| parts[0] != "organizations"
|| parts[2] != "clusters"
|| parts[4] != "clusterDevices"
|| !is_uuid_v7(parts[1])
|| !is_uuid_v7(parts[3])
|| !is_uuid_v7(parts[5])
{
return None;
}
Some(Self {
bundle_uid: bundle_uid.to_owned(),
organization_name: parts[..2].join("/"),
cluster_name: parts[..4].join("/"),
device_name: device_name.to_owned(),
})
}
}
fn is_uuid_v7(value: &str) -> bool {
Uuid::parse_str(value).is_ok_and(|uuid| {
uuid.get_version() == Some(Version::SortRand) && uuid.get_variant() == Variant::RFC4122 && uuid.to_string() == value
})
}
rustfs/src/connect/offline/mod.rs
+9 -5
View File
@@ -12,7 +12,7 @@
// See the License for the specific language governing permissions and
// limitations under the License.
//! Offline enrolment: joining a Connect tenant without a network.
//! Air-gapped Connect enrolment and support-bundle production.
//!
//! An air-gapped cluster cannot perform the registration exchange, so an
//! operator carries a signed challenge in and a signed response out. The device
@@ -20,20 +20,24 @@
//! whose fingerprint is compiled into this binary, minting the key being
//! enrolled, and signing the response.
//!
//! Nothing here opens a socket. That is the point of the surface, and it is
//! asserted rather than assumed: the enrolment path takes bytes and returns
//! bytes.
//! The same enrolled key signs a deterministic bundle of the stopped runtime's
//! persisted inventory and bounded host diagnostics. Nothing here opens a
//! socket; upload is a separate, operator-controlled step.
//!
//! The trust model, the signing convention, and every rejection reason are
//! frozen by `protocol/agent/v1/fixtures/offline-enrollment/` and by
//! frozen by `protocol/agent/v1/fixtures/{offline-enrollment,bundle}/` and by
//! `docs/adr/0009-offline-signing.md` on the Connect side.
pub mod bundle_writer;
pub mod collectors;
pub mod enrollment;
pub mod key_store;
#[cfg(target_os = "linux")]
mod manifest;
pub mod manifest_entry;
pub mod redaction;
pub use bundle_writer::{BundleContext, BundleError, BundleReceipt, write_offline_bundle};
pub use collectors::{CollectorError, OfflineCollector, OfflineDiagnostics, collect_offline_diagnostics};
pub use enrollment::{EnrollmentError, OfflineEnrollment, VerifiedChallenge};
pub use key_store::OfflineKeyStore;
rustfs/tests/connect_offline_bundle.rs
+454
View File
@@ -0,0 +1,454 @@
// Copyright 2024 RustFS Team
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
// http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.
#![cfg(target_os = "linux")]
//! Deterministic archive, signature, cleanup, and CLI coverage for R07.
use std::fs::{self, File};
use std::io::Read as _;
use std::process::Command;
#[cfg(target_os = "linux")]
use std::time::Duration;
use base64::{Engine as _, engine::general_purpose::URL_SAFE_NO_PAD};
use p256::ecdsa::{Signature, VerifyingKey, signature::Verifier as _};
use p256::pkcs8::DecodePublicKey as _;
use rustfs::connect::DeviceIdentity;
use rustfs::connect::offline::collectors::DataClassification;
use rustfs::connect::offline::redaction::{REDACTION_VERSION, RULESET_HASH};
use rustfs::connect::offline::{BundleContext, BundleError, ManifestEntry, OfflineKeyStore, write_offline_bundle};
#[cfg(target_os = "linux")]
use rustfs::connect::{
CredentialStore, HeartbeatConfig, IdentityStore, InventoryFlag, InventorySchedule, InventorySnapshot, InventoryStatus,
spawn_inventory_runtime,
};
use serde_json::Value;
use sha2::{Digest as _, Sha256};
#[cfg(target_os = "linux")]
use tokio::sync::watch;
use tokio_util::sync::CancellationToken;
use zip::{CompressionMethod, ZipArchive};
const BUNDLE_UID: &str = "0198f3a1-8000-7e50-8f61-4a5b6c7d8e94";
const DEVICE_NAME: &str = "organizations/0198f3a1-4c00-7a10-8b21-0c1d2e3f4a50/clusters/0198f3a1-5d00-7b20-9c31-1d2e3f4a5b61/clusterDevices/0198f3a1-6e00-7c30-ad41-2e3f4a5b6c72";
const PATHS: [&str; 14] = [
"offline/rustfs-version.json",
"offline/node-count.json",
"offline/drive-count.json",
"offline/capacity-used-bytes.json",
"offline/capacity-total-bytes.json",
"offline/coarse-health-flags.json",
"offline/os-summary.json",
"offline/kernel-summary.json",
"offline/cpu-summary.json",
"offline/memory-summary.json",
"offline/filesystem-summary.json",
"offline/network-summary.json",
"manifest.json",
"manifest.sig",
];
fn context() -> BundleContext {
BundleContext {
bundle_uid: BUNDLE_UID.to_owned(),
device_name: DEVICE_NAME.to_owned(),
nonce: [0x2a; 32],
produced_at_unix: 1_777_860_000,
}
}
fn entries() -> Vec<ManifestEntry> {
let cpu_summary = format!(r#"{{"cpuSummary":{{"architecture":"{}","cores":8}}}}"#, std::env::consts::ARCH);
let values = [
("offline.rustfsVersion", DataClassification::L0, r#"{"rustfsVersion":"1.4.2"}"#),
("offline.nodeCount", DataClassification::L0, r#"{"nodeCount":2}"#),
("offline.driveCount", DataClassification::L0, r#"{"driveCount":3}"#),
("offline.capacityUsedBytes", DataClassification::L0, r#"{"capacityUsedBytes":1500}"#),
("offline.capacityTotalBytes", DataClassification::L0, r#"{"capacityTotalBytes":6000}"#),
(
"offline.coarseHealthFlags",
DataClassification::L0,
r#"{"coarseHealthFlags":["cluster.degraded","drive.offline"]}"#,
),
("offline.osSummary", DataClassification::L1, r#"{"osSummary":"Linux"}"#),
("offline.kernelSummary", DataClassification::L1, r#"{"kernelSummary":"6.8.0"}"#),
("offline.cpuSummary", DataClassification::L1, cpu_summary.as_str()),
(
"offline.memorySummary",
DataClassification::L1,
r#"{"memorySummary":{"totalBytes":17179869184,"underPressure":false}}"#,
),
(
"offline.filesystemSummary",
DataClassification::L1,
r#"{"filesystemSummary":["ext4","xfs"]}"#,
),
(
"offline.networkSummary",
DataClassification::L1,
r#"{"networkSummary":{"bondCount":1,"interfaceCount":4}}"#,
),
];
values
.into_iter()
.map(|(field_id, classification, canonical_json)| ManifestEntry {
field_id,
classification,
canonical_json: canonical_json.to_owned(),
redaction_version: REDACTION_VERSION,
ruleset_hash: RULESET_HASH,
redacted_count: 0,
})
.collect()
}
fn read_member(archive: &mut ZipArchive<File>, path: &str) -> Vec<u8> {
let mut file = archive.by_name(path).unwrap_or_else(|error| panic!("read {path}: {error}"));
let mut bytes = Vec::new();
file.read_to_end(&mut bytes)
.unwrap_or_else(|error| panic!("read {path} bytes: {error}"));
bytes
}
#[test]
fn connect_offline_bundle_is_deterministic_bounded_and_signed_over_exact_manifest_bytes() {
let temp = tempfile::tempdir().expect("bundle tempdir");
let first = temp.path().join("first.zip");
let second = temp.path().join("second.zip");
let key = DeviceIdentity::generate();
let cancel = CancellationToken::new();
let first_receipt = write_offline_bundle(&first, &context(), &entries(), &key, &cancel).expect("first bundle");
let second_receipt = write_offline_bundle(&second, &context(), &entries(), &key, &cancel).expect("second bundle");
let first_bytes = fs::read(&first).expect("read first bundle");
assert_eq!(first_bytes, fs::read(&second).expect("read second bundle"));
let private_key = key.to_pkcs8_der().expect("encode test key");
assert!(
!first_bytes
.windows(private_key.len())
.any(|window| window == private_key.as_slice())
);
assert_eq!(first_receipt, second_receipt);
assert_eq!(first_receipt.archive_size_bytes, first_bytes.len() as u64);
assert_eq!(first_receipt.archive_sha256, hex_lower(&Sha256::digest(&first_bytes)));
assert_eq!((first_receipt.l0_count, first_receipt.l1_count), (6, 6));
#[cfg(unix)]
{
use std::os::unix::fs::PermissionsExt as _;
assert_eq!(fs::metadata(&first).expect("bundle metadata").permissions().mode() & 0o777, 0o600);
}
let mut archive = ZipArchive::new(File::open(&first).expect("open bundle")).expect("parse bundle");
assert_eq!(archive.len(), PATHS.len());
for (index, expected_path) in PATHS.into_iter().enumerate() {
let file = archive.by_index(index).expect("archive member");
assert_eq!(file.name(), expected_path);
assert_eq!(file.compression(), CompressionMethod::Stored);
assert_eq!(file.last_modified(), Some(zip::DateTime::DEFAULT));
assert_eq!(file.unix_mode().map(|mode| mode & 0o777), Some(0o600));
assert!(file.is_file());
assert!(!file.is_symlink());
}
let manifest_bytes = read_member(&mut archive, "manifest.json");
let signature_document: Value = serde_json::from_slice(&read_member(&mut archive, "manifest.sig")).expect("signature json");
assert_eq!(signature_document["algorithm"], "ES256");
assert_eq!(signature_document["signedFile"], "manifest.json");
assert_eq!(signature_document["domainSeparationTag"], "rustfs-support-bundle-v1");
let signature_bytes: [u8; 64] = URL_SAFE_NO_PAD
.decode(signature_document["value"].as_str().expect("signature value"))
.expect("signature base64url")
.try_into()
.expect("fixed-width signature");
let signature = Signature::from_slice(&signature_bytes).expect("P-256 signature");
assert_eq!(signature, signature.normalize_s(), "signature must be low-S");
let verifying_key = VerifyingKey::from_public_key_der(&key.public_key_der()).expect("device public key");
let mut signature_input = b"rustfs-support-bundle-v1\0".to_vec();
signature_input.extend_from_slice(&manifest_bytes);
verifying_key
.verify(&signature_input, &signature)
.expect("manifest signature");
signature_input
.last_mut()
.map(|byte| *byte ^= 1)
.expect("manifest is non-empty");
assert!(verifying_key.verify(&signature_input, &signature).is_err());
let manifest: Value = serde_json::from_slice(&manifest_bytes).expect("manifest json");
assert_eq!(manifest["formatVersion"], "rustfs.connect.support.bundleManifest/1");
assert_eq!(manifest["protocolVersion"], "v1");
assert_eq!(manifest["bundleUid"], BUNDLE_UID);
assert_eq!(manifest["organizationName"], "organizations/0198f3a1-4c00-7a10-8b21-0c1d2e3f4a50");
assert_eq!(
manifest["clusterName"],
"organizations/0198f3a1-4c00-7a10-8b21-0c1d2e3f4a50/clusters/0198f3a1-5d00-7b20-9c31-1d2e3f4a5b61"
);
assert_eq!(manifest["deviceName"], DEVICE_NAME);
assert_eq!(manifest["nonce"], URL_SAFE_NO_PAD.encode([0x2a; 32]));
assert_eq!(manifest["producedAt"], "2026-05-04T02:00:00Z");
assert_eq!(manifest["redactionVersion"], REDACTION_VERSION);
assert_eq!(manifest["rulesetHash"], RULESET_HASH);
assert_eq!(manifest["classificationRegistryVersion"], 1);
assert_eq!(manifest["deviceKeyId"], hex_lower(&Sha256::digest(key.public_key_der())));
for (index, entry) in manifest["entries"].as_array().expect("manifest entries").iter().enumerate() {
let path = entry["path"].as_str().expect("entry path");
assert_eq!(path, PATHS[index]);
let bytes = read_member(&mut archive, path);
assert_eq!(entry["type"], "offline-diagnostic");
assert_eq!(entry["sizeBytes"], bytes.len() as u64);
assert_eq!(entry["sha256"], hex_lower(&Sha256::digest(&bytes)));
assert_eq!(entry["classification"], if index < 6 { "L0" } else { "L1" });
}
}
#[test]
fn connect_offline_bundle_rejects_schema_drift_and_removes_every_temporary_file() {
let temp = tempfile::tempdir().expect("bundle tempdir");
let key = DeviceIdentity::generate();
let active = CancellationToken::new();
let mut invalid = Vec::new();
let mut missing = entries();
missing.pop();
invalid.push(missing);
let mut duplicate = entries();
duplicate[1] = duplicate[0].clone();
invalid.push(duplicate);
let mut wrong_order = entries();
wrong_order.swap(0, 1);
invalid.push(wrong_order);
let mut wrong_classification = entries();
wrong_classification[0].classification = DataClassification::L1;
invalid.push(wrong_classification);
let mut wrong_redaction = entries();
wrong_redaction[0].redaction_version = "rustfs.connect.redaction.v2";
invalid.push(wrong_redaction);
let mut wrong_hash = entries();
wrong_hash[0].ruleset_hash = "foreign";
invalid.push(wrong_hash);
let mut oversized = entries();
oversized[0].canonical_json = format!(r#"{{"rustfsVersion":"{}"}}"#, "x".repeat(16 * 1024));
invalid.push(oversized);
let mut noncanonical = entries();
noncanonical[0].canonical_json = r#"{"rustfsVersion":"1.4.2" }"#.to_owned();
invalid.push(noncanonical);
let mut wrong_payload = entries();
wrong_payload[0].canonical_json = r#"{"nodeCount":2}"#.to_owned();
invalid.push(wrong_payload);
let mut wrong_payload_type = entries();
wrong_payload_type[1].canonical_json = r#"{"nodeCount":"customer-alpha"}"#.to_owned();
invalid.push(wrong_payload_type);
let mut out_of_range = entries();
out_of_range[1].canonical_json = r#"{"nodeCount":0}"#.to_owned();
invalid.push(out_of_range);
let mut impossible_capacity = entries();
impossible_capacity[3].canonical_json = r#"{"capacityUsedBytes":6001}"#.to_owned();
invalid.push(impossible_capacity);
let mut unknown_health_flag = entries();
unknown_health_flag[5].canonical_json = r#"{"coarseHealthFlags":["customer.alpha"]}"#.to_owned();
invalid.push(unknown_health_flag);
let mut unknown_payload_field = entries();
unknown_payload_field[8].canonical_json = format!(
r#"{{"cpuSummary":{{"architecture":"{}","cores":8,"customerName":"Acme"}}}}"#,
std::env::consts::ARCH
);
invalid.push(unknown_payload_field);
let mut secret_canary = entries();
secret_canary[6].canonical_json = r#"{"osSummary":"AKIAIOSFODNN7EXAMPLE"}"#.to_owned();
invalid.push(secret_canary);
for (index, entries) in invalid.iter().enumerate() {
let output = temp.path().join(format!("invalid-{index}.zip"));
assert!(matches!(
write_offline_bundle(&output, &context(), entries, &key, &active),
Err(BundleError::InvalidEntries)
));
assert!(!output.exists());
}
for (index, invalid_context) in [
BundleContext {
bundle_uid: "0198f3a1-8000-7e50-cf61-4a5b6c7d8e94".to_owned(),
..context()
},
BundleContext {
device_name: DEVICE_NAME.replace("8b21", "cb21"),
..context()
},
]
.iter()
.enumerate()
{
let output = temp.path().join(format!("invalid-identity-{index}.zip"));
assert!(matches!(
write_offline_bundle(&output, invalid_context, &entries(), &key, &active),
Err(BundleError::InvalidIdentity)
));
assert!(!output.exists());
}
let cancelled = CancellationToken::new();
cancelled.cancel();
let output = temp.path().join("cancelled.zip");
assert!(matches!(
write_offline_bundle(&output, &context(), &entries(), &key, &cancelled),
Err(BundleError::Cancelled)
));
assert!(!output.exists());
let output_directory = temp.path().join("cannot-replace-directory");
fs::create_dir(&output_directory).expect("output directory");
assert!(write_offline_bundle(&output_directory, &context(), &entries(), &key, &active).is_err());
assert!(output_directory.is_dir());
let residue = fs::read_dir(temp.path())
.expect("read tempdir")
.filter_map(Result::ok)
.filter(|entry| entry.file_name().to_string_lossy().ends_with(".tmp"))
.count();
assert_eq!(residue, 0);
}
#[test]
fn connect_offline_bundle_accepts_an_already_redacted_entry_with_its_original_count() {
let temp = tempfile::tempdir().expect("bundle tempdir");
let output = temp.path().join("redacted.zip");
let mut entries = entries();
entries[6].canonical_json = r#"{"osSummary":"[REDACTED]"}"#.to_owned();
entries[6].redacted_count = 1;
write_offline_bundle(&output, &context(), &entries, &DeviceIdentity::generate(), &CancellationToken::new())
.expect("bundle with an already redacted entry");
assert!(output.exists());
}
#[test]
fn connect_offline_bundle_cli_requires_an_existing_offline_key() {
let temp = tempfile::tempdir().expect("CLI tempdir");
let output = temp.path().join("bundle.zip");
let key_directory = temp.path().join("keys");
let result = Command::new(env!("CARGO_BIN_EXE_rustfs-cli"))
.args([
"connect",
"offline",
"bundle",
"--state-dir",
temp.path().to_string_lossy().as_ref(),
"--device-name",
DEVICE_NAME,
"--output",
output.to_string_lossy().as_ref(),
"--key-dir",
key_directory.to_string_lossy().as_ref(),
])
.output()
.expect("run rustfs-cli");
assert!(!result.status.success());
assert!(String::from_utf8_lossy(&result.stderr).contains("offline enrollment key is missing"));
assert!(!OfflineKeyStore::new(&key_directory).key_path().exists());
assert!(!output.exists());
}
#[cfg(target_os = "linux")]
async fn wait_for_inventory(status: &mut watch::Receiver<InventoryStatus>) {
tokio::time::timeout(Duration::from_secs(3), async {
loop {
if matches!(status.borrow_and_update().clone(), InventoryStatus::Unchanged { .. }) {
return;
}
status.changed().await.expect("inventory status channel");
}
})
.await
.expect("inventory persistence timeout");
}
#[cfg(target_os = "linux")]
#[tokio::test]
async fn connect_offline_bundle_cli_builds_from_stopped_persisted_inventory_without_uploading() {
use std::os::unix::fs::PermissionsExt as _;
let temp = tempfile::tempdir_in(env!("CARGO_MANIFEST_DIR")).expect("CLI tempdir");
let state = temp.path().join("state");
let keys = temp.path().join("keys");
let output = temp.path().join("bundle.zip");
fs::create_dir(&state).expect("state root");
fs::set_permissions(&state, fs::Permissions::from_mode(0o700)).expect("state permissions");
let config = HeartbeatConfig::new(
"",
Vec::new(),
IdentityStore::new(state.join("identity")),
CredentialStore::new(state.join("credential")),
state.join("heartbeat/state.json"),
);
let shutdown = CancellationToken::new();
let runtime = spawn_inventory_runtime(
Some(config),
InventorySchedule {
cadence: Duration::from_secs(60),
jitter: Duration::ZERO,
},
&shutdown,
|| {
std::future::ready(InventorySnapshot::new(
"1.4.2",
None,
2,
3,
6_000,
1_500,
[InventoryFlag::ClusterDegraded, InventoryFlag::DriveOffline],
))
},
)
.expect("state-only inventory")
.expect("configured inventory");
let mut status = runtime.status();
wait_for_inventory(&mut status).await;
runtime.shutdown().await;
OfflineKeyStore::new(&keys).load_or_create().expect("offline key");
let result = Command::new(env!("CARGO_BIN_EXE_rustfs-cli"))
.args([
"connect",
"offline",
"bundle",
"--state-dir",
state.to_string_lossy().as_ref(),
"--device-name",
DEVICE_NAME,
"--output",
output.to_string_lossy().as_ref(),
"--key-dir",
keys.to_string_lossy().as_ref(),
])
.output()
.expect("run rustfs-cli");
assert!(result.status.success(), "{}", String::from_utf8_lossy(&result.stderr));
let stdout = String::from_utf8_lossy(&result.stdout);
assert!(stdout.contains("L0: 6 entries"));
assert!(stdout.contains("L1: 6 entries"));
assert!(stdout.contains("Upload: not performed"));
let mut archive = ZipArchive::new(File::open(&output).expect("open CLI bundle")).expect("parse CLI bundle");
let manifest: Value = serde_json::from_slice(&read_member(&mut archive, "manifest.json")).expect("manifest json");
assert_eq!(manifest["deviceName"], DEVICE_NAME);
assert_eq!(manifest["entries"].as_array().expect("manifest entries").len(), 12);
}
fn hex_lower(bytes: &[u8]) -> String {
bytes.iter().map(|byte| format!("{byte:02x}")).collect()
}