From 2bd83a5276d5e809664048ad06c6c9b7f7c9427d Mon Sep 17 00:00:00 2001 From: Zhengchao An Date: Tue, 25 Aug 2026 21:37:44 +0800 Subject: [PATCH] feat(connect): build signed offline bundles (#6579) * feat(connect): build signed offline bundles * fix(connect): validate offline bundle inputs * test(connect): use the target architecture * chore(connect): scope the unsafe allowance --- rustfs/src/bin/rustfs-cli.rs | 253 ++++- rustfs/src/connect/offline/bundle_writer.rs | 965 ++++++++++++++++++++ rustfs/src/connect/offline/manifest.rs | 147 +++ rustfs/src/connect/offline/mod.rs | 14 +- rustfs/tests/connect_offline_bundle.rs | 454 +++++++++ 5 files changed, 1821 insertions(+), 12 deletions(-) create mode 100644 rustfs/src/connect/offline/bundle_writer.rs create mode 100644 rustfs/src/connect/offline/manifest.rs create mode 100644 rustfs/tests/connect_offline_bundle.rs diff --git a/rustfs/src/bin/rustfs-cli.rs b/rustfs/src/bin/rustfs-cli.rs index 3d1073100..ce92d6c19 100644 --- a/rustfs/src/bin/rustfs-cli.rs +++ b/rustfs/src/bin/rustfs-cli.rs @@ -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 --output [--key-dir ] + rustfs-cli connect offline bundle --state-dir --device-name --output [--key-dir ] 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 = 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 = None; + let mut device_name: Option = None; + let mut output_path: Option = None; + let mut key_directory: Option = None; + + let mut index = 0; + while index < arguments.len() { + let flag = arguments[index].as_str(); + let take_value = |name: &str| -> Result { + 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( + cancel: &CancellationToken, + mut signal: Pin<&mut S>, + mut writer: tokio::task::JoinHandle>, +) -> Result +where + S: Future> + ?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, tokio::task::JoinError>, +) -> Result { + result + .map_err(|error| format!("offline bundle writer task failed: {error}"))? + .map_err(|error| error.to_string()) +} + +fn run_offline_runtime( + runtime: tokio::runtime::Runtime, + operation: impl Future>, +) -> 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 = None; let mut output_path: Option = 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()); + } +} diff --git a/rustfs/src/connect/offline/bundle_writer.rs b/rustfs/src/connect/offline/bundle_writer.rs new file mode 100644 index 000000000..877c1bfc7 --- /dev/null +++ b/rustfs/src/connect/offline/bundle_writer.rs @@ -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 { + #[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 { + 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::>(); + 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::(&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::().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 { + serde_json::from_str::(&entry.canonical_json) + .ok()? + .get(field)? + .as_u64() +} + +#[cfg(target_os = "linux")] +fn sign(key: &DeviceIdentity, manifest: &[u8]) -> Result { + 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(¤t_directory)?; + if file_identity(¤t_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 { + 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 { + 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 { + 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 { + 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::::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 { + 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), + } + + thread_local! { + static NEXT: RefCell> = 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 { + 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); + } +} diff --git a/rustfs/src/connect/offline/manifest.rs b/rustfs/src/connect/offline/manifest.rs new file mode 100644 index 000000000..e8014aba5 --- /dev/null +++ b/rustfs/src/connect/offline/manifest.rs @@ -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 { + if !is_uuid_v7(bundle_uid) { + return None; + } + + let parts = device_name.split('/').collect::>(); + 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 + }) +} diff --git a/rustfs/src/connect/offline/mod.rs b/rustfs/src/connect/offline/mod.rs index 37a7f3b84..6b4d06d57 100644 --- a/rustfs/src/connect/offline/mod.rs +++ b/rustfs/src/connect/offline/mod.rs @@ -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; diff --git a/rustfs/tests/connect_offline_bundle.rs b/rustfs/tests/connect_offline_bundle.rs new file mode 100644 index 000000000..d98608fdc --- /dev/null +++ b/rustfs/tests/connect_offline_bundle.rs @@ -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 { + 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, path: &str) -> Vec { + 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) { + 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() +}