diff --git a/rustfs/src/connect/config.rs b/rustfs/src/connect/config.rs index 391d4ef45..8ecc6f04b 100644 --- a/rustfs/src/connect/config.rs +++ b/rustfs/src/connect/config.rs @@ -14,6 +14,7 @@ use std::env; use std::ffi::OsString; +#[cfg(target_os = "linux")] use std::fs; use std::path::PathBuf; use std::time::Duration; @@ -63,16 +64,37 @@ impl HeartbeatConfig { credential_store: CredentialStore, state_path: impl Into, ) -> Self { + let state_path = state_path.into(); Self { endpoint: endpoint.into(), root_ca_pem: root_ca_pem.into(), identity_store, credential_store, - state_path: state_path.into(), + state_path, schedule: HeartbeatSchedule::default(), } } + #[cfg(any(target_os = "linux", test))] + pub(crate) fn state_only(state_root: PathBuf) -> Self { + Self { + endpoint: String::new(), + root_ca_pem: Vec::new(), + identity_store: IdentityStore::new(state_root.join("identity")), + credential_store: CredentialStore::new(state_root.join("credential")), + state_path: state_root.join("heartbeat/state.json"), + schedule: HeartbeatSchedule::default(), + } + } + + pub(crate) fn transport_enabled(&self) -> bool { + !self.endpoint.is_empty() + } + + pub(crate) fn state_root(&self) -> Option<&std::path::Path> { + self.state_path.parent().and_then(std::path::Path::parent) + } + pub fn from_env() -> Result, HeartbeatConfigError> { Self::from_env_values( env::var_os(ENV_CONNECT_ENDPOINT), @@ -90,19 +112,33 @@ impl HeartbeatConfig { if !configured { return Ok(None); } - let (Some(endpoint), Some(root_ca_file), Some(state_dir)) = (endpoint, root_ca_file, state_dir) else { + let Some(state_dir) = state_dir else { return Err(HeartbeatConfigError::Partial); }; - let endpoint = endpoint.into_string().map_err(|_| HeartbeatConfigError::EndpointEncoding)?; - let root_ca_file = PathBuf::from(root_ca_file); let state_dir = PathBuf::from(state_dir); - if endpoint.is_empty() || root_ca_file.as_os_str().is_empty() || state_dir.as_os_str().is_empty() { + if state_dir.as_os_str().is_empty() || endpoint.is_some() != root_ca_file.is_some() { return Err(HeartbeatConfigError::Partial); } + #[cfg(not(target_os = "linux"))] + return Err(HeartbeatConfigError::PlatformSecurity); + #[cfg(target_os = "linux")] + let (Some(endpoint), Some(root_ca_file)) = (endpoint, root_ca_file) else { + return Ok(Some(Self::state_only(state_dir))); + }; + #[cfg(target_os = "linux")] + let endpoint = endpoint.into_string().map_err(|_| HeartbeatConfigError::EndpointEncoding)?; + #[cfg(target_os = "linux")] + let root_ca_file = PathBuf::from(root_ca_file); + #[cfg(target_os = "linux")] + if endpoint.is_empty() || root_ca_file.as_os_str().is_empty() { + return Err(HeartbeatConfigError::Partial); + } + #[cfg(target_os = "linux")] let root_ca_pem = fs::read(&root_ca_file).map_err(|source| HeartbeatConfigError::RootCertificate { path: root_ca_file, source, })?; + #[cfg(target_os = "linux")] Ok(Some(Self::new( endpoint, root_ca_pem, @@ -116,17 +152,19 @@ impl HeartbeatConfig { #[derive(Debug, thiserror::Error)] pub enum HeartbeatConfigError { #[error( - "Connect heartbeat configuration requires RUSTFS_CONNECT_ENDPOINT, RUSTFS_CONNECT_ROOT_CA_FILE, and RUSTFS_CONNECT_STATE_DIR" + "Connect requires RUSTFS_CONNECT_STATE_DIR and either both or neither of RUSTFS_CONNECT_ENDPOINT and RUSTFS_CONNECT_ROOT_CA_FILE" )] Partial, #[error("RUSTFS_CONNECT_ENDPOINT is not valid UTF-8")] EndpointEncoding, - #[error("failed to read the Connect root CA at {path}: {source}")] + #[error("Connect root CA could not be read")] RootCertificate { path: PathBuf, #[source] source: std::io::Error, }, + #[error("Connect inventory persistence requires Linux filesystem security guarantees")] + PlatformSecurity, } #[cfg(test)] @@ -149,9 +187,34 @@ mod tests { HeartbeatConfig::from_env_values(Some(OsString::from("https://connect.example/agent/")), None, None), Err(HeartbeatConfigError::Partial) )); + assert!(matches!( + HeartbeatConfig::from_env_values( + Some(OsString::from("https://connect.example/agent/")), + Some(OsString::from("root.pem")), + None, + ), + Err(HeartbeatConfigError::Partial) + )); + assert!(matches!( + HeartbeatConfig::from_env_values(None, Some(OsString::from("root.pem")), Some(OsString::from("state"))), + Err(HeartbeatConfigError::Partial) + )); } #[test] + #[cfg(target_os = "linux")] + fn state_directory_alone_enables_local_inventory_without_transport() { + let state = tempfile::tempdir().expect("tempdir").keep(); + let config = HeartbeatConfig::from_env_values(None, None, Some(state.clone().into_os_string())) + .expect("state-only config") + .expect("enabled config"); + + assert_eq!(config.state_root(), Some(state.as_path())); + assert!(!config.transport_enabled()); + } + + #[test] + #[cfg(target_os = "linux")] fn complete_environment_builds_the_durable_paths() { let temp = tempfile::tempdir().expect("tempdir"); let root = temp.path().join("root.pem"); @@ -168,6 +231,24 @@ mod tests { assert_eq!(config.endpoint, "https://connect.example/agent/"); assert_eq!(config.root_ca_pem, b"root certificate"); assert_eq!(config.state_path, state.join("heartbeat/state.json")); + assert_eq!(config.state_root(), Some(state.as_path())); assert!(!state.exists(), "parsing configuration must not create state"); } + + #[test] + #[cfg(not(target_os = "linux"))] + fn configured_inventory_fails_without_linux_filesystem_guarantees() { + assert!(matches!( + HeartbeatConfig::from_env_values(None, None, Some(OsString::from("state"))), + Err(HeartbeatConfigError::PlatformSecurity) + )); + assert!(matches!( + HeartbeatConfig::from_env_values( + Some(OsString::from("https://connect.example/agent/")), + Some(OsString::from("missing-root.pem")), + Some(OsString::from("state")), + ), + Err(HeartbeatConfigError::PlatformSecurity) + )); + } } diff --git a/rustfs/src/connect/inventory.rs b/rustfs/src/connect/inventory.rs index 5a5943d2f..6238a581a 100644 --- a/rustfs/src/connect/inventory.rs +++ b/rustfs/src/connect/inventory.rs @@ -14,8 +14,16 @@ use std::collections::BTreeSet; use std::fs; -use std::io::{self, Write as _}; -use std::path::{Path, PathBuf}; +#[cfg(any(target_os = "linux", test))] +use std::io; +#[cfg(target_os = "linux")] +use std::io::Read as _; +#[cfg(target_os = "linux")] +use std::io::Write as _; +use std::path::Path; +#[cfg(target_os = "linux")] +use std::sync::Arc; +#[cfg(target_os = "linux")] use std::sync::atomic::{AtomicU64, Ordering}; use std::time::Duration; @@ -37,10 +45,34 @@ const RUSTFS_VERSION: &str = concat!( const HASH_PREFIX: &[u8] = b"rustfs-connect/agent/v1/inventory-snapshot\n"; const MAX_SEQUENCE: u64 = 9_007_199_254_740_991; const MAX_SAFE_INTEGER: u64 = 9_007_199_254_740_991; -#[cfg(unix)] +const ENVELOPE_FORMAT_VERSION: &str = "v1"; +const ENVELOPE_HASH_PREFIX: &[u8] = b"rustfs-connect-inventory-envelope-v1"; +#[cfg(target_os = "linux")] +const MAX_PERSISTED_BYTES: usize = 16 * 1024; +#[allow(dead_code)] // Kept for the crate-private stopped-server reader consumed by R06. +const MAX_FUTURE_SKEW: Duration = Duration::from_secs(5 * 60); +#[cfg(target_os = "linux")] const FILE_MODE: u32 = 0o600; +#[cfg(target_os = "linux")] static STAGING_SEQUENCE: AtomicU64 = AtomicU64::new(0); +#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] +#[serde(deny_unknown_fields, rename_all = "camelCase")] +struct InventoryEnvelope { + format_version: String, + captured_at: String, + snapshot: InventorySnapshot, + envelope_hash: String, +} + +#[derive(Debug, PartialEq, Eq)] +#[allow(dead_code)] // This is the intentionally narrow handoff to R06. +pub(crate) struct PersistedInventory { + pub(crate) snapshot: InventorySnapshot, + pub(crate) captured_at: String, + pub(crate) age: Duration, +} + #[derive(Clone, Copy, Debug, PartialEq, Eq)] pub struct InventorySchedule { pub cadence: Duration, @@ -257,6 +289,7 @@ impl PendingInventory { } } + #[cfg(target_os = "linux")] fn is_valid(&self) -> bool { self.protocol_version == PROTOCOL_VERSION && self.sequence <= MAX_SEQUENCE @@ -268,13 +301,17 @@ impl PendingInventory { fn content_hash(&self) -> Result { self.snapshot.content_hash() } + + pub(crate) fn snapshot(&self) -> &InventorySnapshot { + &self.snapshot + } } pub(crate) enum InventoryDelivery { Accepted { content_hash: String, received_at: String }, Retry { retry_after: Option }, - AuthenticationStopped { status: u16, reason: Option }, - Rejected { status: u16, reason: Option }, + AuthenticationStopped { status: u16 }, + Rejected { status: u16 }, } pub(crate) struct InventorySender { @@ -318,17 +355,50 @@ impl InventorySender { }) } TelemetryDelivery::Retry { retry_after } => Ok(InventoryDelivery::Retry { retry_after }), - TelemetryDelivery::AuthenticationStopped { status, reason } => { - Ok(InventoryDelivery::AuthenticationStopped { status, reason }) - } - TelemetryDelivery::Rejected { status, reason } => Ok(InventoryDelivery::Rejected { status, reason }), + TelemetryDelivery::AuthenticationStopped { status, .. } => Ok(InventoryDelivery::AuthenticationStopped { status }), + TelemetryDelivery::Rejected { status, .. } => Ok(InventoryDelivery::Rejected { status }), } } } #[derive(Clone)] pub(crate) struct InventoryStateStore { - path: PathBuf, + #[cfg(target_os = "linux")] + directory: Arc, + #[cfg(target_os = "linux")] + state_root: Arc, +} + +#[cfg(target_os = "linux")] +struct StateRootAnchor { + root: fs::File, + components: Vec<(std::ffi::OsString, fs::File)>, +} + +#[cfg(target_os = "linux")] +impl StateRootAnchor { + fn state_root(&self) -> Result<&fs::File, InventoryError> { + self.components + .last() + .map(|(_, directory)| directory) + .ok_or(InventoryError::StatePath) + } + + fn validate(&self) -> Result<(), InventoryError> { + validate_directory(&self.root, false)?; + let mut current = None; + for (index, (component, expected)) in self.components.iter().enumerate() { + let parent = current.as_ref().unwrap_or(&self.root); + let resolved = open_directory_component_at(parent, component)?; + let dedicated = index + 1 == self.components.len(); + validate_directory(&resolved, dedicated)?; + if file_identity(expected)? != file_identity(&resolved)? { + return Err(InventoryError::PersistenceSecurity); + } + current = Some(resolved); + } + Ok(()) + } } #[derive(Default, Serialize, Deserialize)] @@ -340,49 +410,63 @@ struct InventoryState { } impl InventoryStateStore { - pub(crate) fn from_heartbeat_path(path: &Path) -> Result { - let root = path.parent().and_then(Path::parent).ok_or(InventoryError::StatePath)?; - Ok(Self { - path: root.join("inventory/state.json"), - }) + pub(crate) fn from_state_root(path: &Path) -> Result { + #[cfg(not(target_os = "linux"))] + { + let _ = path; + Err(InventoryError::PlatformSecurity) + } + #[cfg(target_os = "linux")] + { + let (state_root, directory) = open_inventory_directory(path)?; + Ok(Self { + directory: Arc::new(directory), + state_root: Arc::new(state_root), + }) + } } pub(crate) fn try_runtime_lock(&self) -> Result { - let directory = parent(&self.path)?; - prepare_inventory_directory(directory)?; - let name = filename(&self.path)?; - let path = directory.join(format!(".{name}.lock")); - let mut options = fs::OpenOptions::new(); - options.create(true).truncate(false).read(true).write(true); - #[cfg(unix)] + #[cfg(not(target_os = "linux"))] { - use std::os::unix::fs::OpenOptionsExt as _; - options.mode(FILE_MODE); + Err(InventoryError::PlatformSecurity) + } + #[cfg(target_os = "linux")] + { + self.validate_anchor()?; + let lock = open_file_at(&self.directory, ".state.json.lock", true, true)?; + validate_regular_file(&lock)?; + lock.try_lock().map_err(|_| InventoryError::AlreadyRunning)?; + self.validate_anchor()?; + Ok(lock) } - let lock = options.open(&path).map_err(|source| state_io(&path, source))?; - check_mode(&path)?; - lock.try_lock().map_err(|_| InventoryError::AlreadyRunning)?; - Ok(lock) } - pub(crate) async fn pending(&self) -> Result, InventoryError> { + pub(crate) async fn pending(&self) -> Result, InventoryError> { let store = self.clone(); tokio::task::spawn_blocking(move || { - let state = store.read()?; + let (state, persisted_at) = store.read_with_persisted_at()?; if state.pending.is_none() && state.next_sequence > MAX_SEQUENCE { return Err(InventoryError::SequenceExhausted); } - Ok(state.pending) + state + .pending + .map(|pending| { + persisted_at + .map(|persisted_at| (pending, persisted_at)) + .ok_or(InventoryError::StateCorrupt) + }) + .transpose() }) .await - .map_err(|source| state_io(&self.path, io::Error::other(source)))? + .map_err(|_| InventoryError::StateIo)? } pub(crate) async fn prepare(&self, snapshot: InventorySnapshot) -> Result, InventoryError> { let store = self.clone(); tokio::task::spawn_blocking(move || store.prepare_sync(snapshot)) .await - .map_err(|source| state_io(&self.path, io::Error::other(source)))? + .map_err(|_| InventoryError::StateIo)? } pub(crate) async fn mark_accepted(&self, accepted: &PendingInventory) -> Result<(), InventoryError> { @@ -390,7 +474,94 @@ impl InventoryStateStore { let accepted = accepted.clone(); tokio::task::spawn_blocking(move || store.mark_accepted_sync(&accepted)) .await - .map_err(|source| state_io(&self.path, io::Error::other(source)))? + .map_err(|_| InventoryError::StateIo)? + } + + pub(crate) async fn publish_latest( + &self, + snapshot: InventorySnapshot, + captured_at: String, + shutdown: tokio_util::sync::CancellationToken, + ) -> Result<(), InventoryError> { + let store = self.clone(); + tokio::task::spawn_blocking(move || store.publish_latest_sync(snapshot, captured_at, &shutdown)) + .await + .map_err(|_| InventoryError::StateIo)? + } + + pub(crate) async fn ensure_latest( + &self, + snapshot: InventorySnapshot, + captured_at: String, + shutdown: tokio_util::sync::CancellationToken, + ) -> Result<(), InventoryError> { + let store = self.clone(); + tokio::task::spawn_blocking(move || match store.read_latest(chrono::Utc::now()) { + Ok(_) => Ok(()), + Err(InventoryError::StateMissing) => store.publish_latest_sync(snapshot, captured_at, &shutdown), + Err(error) => Err(error), + }) + .await + .map_err(|_| InventoryError::StateIo)? + } + + #[allow(dead_code)] // R06 reads this after the server has stopped. + pub(crate) fn read_latest(&self, now: chrono::DateTime) -> Result { + self.read_latest_inner(now, || {}) + } + + #[cfg(all(test, target_os = "linux"))] + fn read_latest_after_open( + &self, + now: chrono::DateTime, + after_open: impl FnOnce(), + ) -> Result { + self.read_latest_inner(now, after_open) + } + + fn read_latest_inner( + &self, + now: chrono::DateTime, + after_open: impl FnOnce(), + ) -> Result { + #[cfg(not(target_os = "linux"))] + { + let _ = (now, after_open); + Err(InventoryError::PlatformSecurity) + } + #[cfg(target_os = "linux")] + { + self.validate_anchor()?; + let mut file = open_file_at(&self.directory, "latest.json", false, false)?; + validate_regular_file(&file)?; + let before = file_identity(&file)?; + after_open(); + let bytes = read_bounded(&mut file)?; + validate_regular_file(&file)?; + if before != file_identity(&file)? { + return Err(InventoryError::PersistenceSecurity); + } + let current = open_file_at(&self.directory, "latest.json", false, false)?; + validate_regular_file(¤t)?; + if before != file_identity(¤t)? { + return Err(InventoryError::PersistenceSecurity); + } + self.validate_anchor()?; + decode_envelope(&bytes, now) + } + } + + #[cfg(target_os = "linux")] + fn validate_anchor(&self) -> Result<(), InventoryError> { + self.state_root.validate()?; + let state_root = self.state_root.state_root()?; + validate_directory(&self.directory, true)?; + let current = open_directory_at(state_root, "inventory")?; + validate_directory(¤t, true)?; + if file_identity(&self.directory)? != file_identity(¤t)? { + return Err(InventoryError::PersistenceSecurity); + } + Ok(()) } fn prepare_sync(&self, snapshot: InventorySnapshot) -> Result, InventoryError> { @@ -424,46 +595,223 @@ impl InventoryStateStore { } fn read(&self) -> Result { - let bytes = match fs::read(&self.path) { - Ok(bytes) => bytes, - Err(source) if source.kind() == io::ErrorKind::NotFound => return Ok(InventoryState::default()), - Err(source) => return Err(state_io(&self.path, source)), - }; - check_mode(&self.path)?; - let state: InventoryState = serde_json::from_slice(&bytes).map_err(|source| InventoryError::StateInvalid { - path: self.path.clone(), - source, - })?; - let last_hash_valid = state.last_accepted_content_hash.as_deref().is_none_or(valid_content_hash); - let pending_valid = state.pending.as_ref().is_none_or(|pending| { - pending.sequence == state.next_sequence - && pending.is_valid() - && pending - .content_hash() - .is_ok_and(|hash| state.last_accepted_content_hash.as_deref() != Some(&hash)) - }); - if state.next_sequence > MAX_SEQUENCE + 1 || !last_hash_valid || !pending_valid { - return Err(InventoryError::StateCorrupt { path: self.path.clone() }); + self.read_with_persisted_at().map(|(state, _)| state) + } + + fn read_with_persisted_at(&self) -> Result<(InventoryState, Option), InventoryError> { + #[cfg(not(target_os = "linux"))] + { + Err(InventoryError::PlatformSecurity) + } + #[cfg(target_os = "linux")] + { + self.validate_anchor()?; + let mut file = match open_file_at(&self.directory, "state.json", false, false) { + Ok(file) => file, + Err(InventoryError::StateMissing) => return Ok((InventoryState::default(), None)), + Err(error) => return Err(error), + }; + validate_regular_file(&file)?; + let bytes = read_bounded(&mut file)?; + let modified = file + .metadata() + .and_then(|metadata| metadata.modified()) + .map_err(|_| InventoryError::StateIo)?; + self.validate_anchor()?; + let state: InventoryState = serde_json::from_slice(&bytes).map_err(|_| InventoryError::StateInvalid)?; + let last_hash_valid = state.last_accepted_content_hash.as_deref().is_none_or(valid_content_hash); + let pending_valid = state.pending.as_ref().is_none_or(|pending| { + pending.sequence == state.next_sequence + && pending.is_valid() + && pending + .content_hash() + .is_ok_and(|hash| state.last_accepted_content_hash.as_deref() != Some(&hash)) + }); + if state.next_sequence > MAX_SEQUENCE + 1 || !last_hash_valid || !pending_valid { + return Err(InventoryError::StateCorrupt); + } + let modified = chrono::DateTime::::from(modified); + if modified > chrono::Utc::now() + MAX_FUTURE_SKEW { + return Err(InventoryError::StateCorrupt); + } + let persisted_at = modified.format("%Y-%m-%dT%H:%M:%SZ").to_string(); + Ok((state, Some(persisted_at))) } - Ok(state) } fn write(&self, state: &InventoryState) -> Result<(), InventoryError> { - let bytes = serde_json::to_vec(state).map_err(|source| InventoryError::StateInvalid { - path: self.path.clone(), - source, - })?; - let directory = parent(&self.path)?; - prepare_inventory_directory(directory)?; - let temp = stage(directory, &self.path, &bytes)?; - let result = fs::rename(&temp, &self.path) - .map_err(|source| state_io(&self.path, source)) - .and_then(|()| fsync_dir(directory).map_err(|source| state_io(directory, source))); - if result.is_err() { - let _ = fs::remove_file(temp); - } - result + let bytes = serde_json::to_vec(state).map_err(|_| InventoryError::StateInvalid)?; + self.replace_file("state.json", &bytes, || false) } + + fn publish_latest_sync( + &self, + snapshot: InventorySnapshot, + captured_at: String, + shutdown: &tokio_util::sync::CancellationToken, + ) -> Result<(), InventoryError> { + snapshot.validate()?; + let bytes = encode_envelope(snapshot, captured_at)?; + self.replace_file("latest.json", &bytes, || shutdown.is_cancelled()) + } + + fn replace_file(&self, destination: &str, bytes: &[u8], cancelled: impl FnOnce() -> bool) -> Result<(), InventoryError> { + self.replace_file_inner( + destination, + bytes, + cancelled, + #[cfg(all(test, target_os = "linux"))] + None, + #[cfg(all(test, target_os = "linux"))] + || {}, + ) + } + + fn replace_file_inner( + &self, + destination: &str, + bytes: &[u8], + cancelled: impl FnOnce() -> bool, + #[cfg(all(test, target_os = "linux"))] fault: Option, + #[cfg(all(test, target_os = "linux"))] before_commit: impl FnOnce(), + ) -> Result<(), InventoryError> { + #[cfg(not(target_os = "linux"))] + { + let _ = (destination, bytes, cancelled); + Err(InventoryError::PlatformSecurity) + } + #[cfg(target_os = "linux")] + { + self.validate_anchor()?; + match open_file_at(&self.directory, destination, false, false) { + Ok(existing) => validate_regular_file(&existing)?, + Err(InventoryError::StateMissing) => {} + Err(error) => return Err(error), + } + let (temp_name, mut temp) = stage_at(&self.directory, destination)?; + #[cfg(test)] + let injected_write = matches!(fault.as_ref(), Some(PersistFault::Write)); + #[cfg(not(test))] + let injected_write = false; + let staged = if injected_write { + Err(InventoryError::StateIo) + } else { + temp.write_all(bytes).map_err(|_| InventoryError::StateIo) + } + .and_then(|()| { + #[cfg(test)] + if matches!(fault.as_ref(), Some(PersistFault::TempSync)) { + return Err(InventoryError::StateIo); + } + temp.sync_all().map_err(|_| InventoryError::StateIo) + }); + if staged.is_err() || cancelled() { + let _ = unlink_at(&self.directory, &temp_name); + return staged.and(Err(InventoryError::Cancelled)); + } + if let Err(error) = validate_regular_file(&temp) { + let _ = unlink_at(&self.directory, &temp_name); + return Err(error); + } + #[cfg(test)] + before_commit(); + if let Err(error) = self.validate_anchor() { + let _ = unlink_at(&self.directory, &temp_name); + return Err(error); + } + #[cfg(test)] + if let Some(PersistFault::CancelDuringCommit(token)) = fault.as_ref() { + token.cancel(); + } + #[cfg(test)] + let rename_failed = matches!(fault.as_ref(), Some(PersistFault::Rename)); + #[cfg(not(test))] + let rename_failed = false; + if rename_failed || rename_at(&self.directory, &temp_name, destination).is_err() { + let _ = unlink_at(&self.directory, &temp_name); + return Err(InventoryError::StateIo); + } + #[cfg(test)] + if matches!(fault.as_ref(), Some(PersistFault::DirectorySync)) { + return Err(InventoryError::DurabilityAfterCommit); + } + self.directory.sync_all().map_err(|_| InventoryError::DurabilityAfterCommit)?; + self.validate_anchor() + } + } +} + +#[cfg(all(test, target_os = "linux"))] +enum PersistFault { + Write, + TempSync, + Rename, + DirectorySync, + CancelDuringCommit(tokio_util::sync::CancellationToken), +} + +fn encode_envelope(snapshot: InventorySnapshot, captured_at: String) -> Result, InventoryError> { + if !is_exact_utc_seconds(&captured_at) { + return Err(InventoryError::EnvelopeTimestamp); + } + let envelope_hash = envelope_hash(ENVELOPE_FORMAT_VERSION, &captured_at, &snapshot)?; + serde_json::to_vec(&InventoryEnvelope { + format_version: ENVELOPE_FORMAT_VERSION.to_owned(), + captured_at, + snapshot, + envelope_hash, + }) + .map_err(|_| InventoryError::StateInvalid) +} + +fn envelope_hash(format_version: &str, captured_at: &str, snapshot: &InventorySnapshot) -> Result { + #[derive(Serialize)] + #[serde(rename_all = "camelCase")] + struct Canonical<'a> { + format_version: &'a str, + captured_at: &'a str, + snapshot: &'a InventorySnapshot, + } + let canonical = serde_json::to_vec(&Canonical { + format_version, + captured_at, + snapshot, + }) + .map_err(|_| InventoryError::StateInvalid)?; + let mut digest = Sha256::new(); + digest.update(ENVELOPE_HASH_PREFIX); + digest.update([0]); + digest.update(canonical); + Ok(hex_simd::encode_to_string(digest.finalize(), hex_simd::AsciiCase::Lower)) +} + +#[allow(dead_code)] // Used by the crate-private stopped-server reader. +fn decode_envelope(bytes: &[u8], now: chrono::DateTime) -> Result { + let envelope: InventoryEnvelope = serde_json::from_slice(bytes).map_err(|_| InventoryError::EnvelopeInvalid)?; + if envelope.format_version != ENVELOPE_FORMAT_VERSION { + return Err(InventoryError::EnvelopeVersion); + } + if !is_exact_utc_seconds(&envelope.captured_at) { + return Err(InventoryError::EnvelopeTimestamp); + } + envelope.snapshot.validate()?; + let expected = envelope_hash(&envelope.format_version, &envelope.captured_at, &envelope.snapshot)?; + if envelope.envelope_hash != expected || !valid_content_hash(&envelope.envelope_hash) { + return Err(InventoryError::EnvelopeHash); + } + let captured_at = chrono::DateTime::parse_from_rfc3339(&envelope.captured_at) + .map_err(|_| InventoryError::EnvelopeTimestamp)? + .with_timezone(&chrono::Utc); + let future = captured_at.signed_duration_since(now); + if future > chrono::Duration::from_std(MAX_FUTURE_SKEW).map_err(|_| InventoryError::EnvelopeTimestamp)? { + return Err(InventoryError::EnvelopeFuture); + } + let age = now.signed_duration_since(captured_at).to_std().unwrap_or_default(); + Ok(PersistedInventory { + snapshot: envelope.snapshot, + captured_at: envelope.captured_at, + age, + }) } fn valid_content_hash(value: &str) -> bool { @@ -473,277 +821,975 @@ fn valid_content_hash(value: &str) -> bool { .all(|byte| byte.is_ascii_digit() || (b'a'..=b'f').contains(&byte)) } -fn parent(path: &Path) -> Result<&Path, InventoryError> { - path.parent() - .ok_or_else(|| state_io(path, io::Error::new(io::ErrorKind::InvalidInput, "state path has no parent"))) -} - -fn filename(path: &Path) -> Result<&str, InventoryError> { - path.file_name() - .and_then(|name| name.to_str()) - .ok_or_else(|| state_io(path, io::Error::new(io::ErrorKind::InvalidInput, "state filename is invalid"))) -} - -fn prepare_inventory_directory(directory: &Path) -> Result<(), InventoryError> { - prepare_inventory_directory_with(directory, create_inventory_directory, fsync_dir) -} - -fn prepare_inventory_directory_with( - directory: &Path, - create: impl FnOnce(&Path) -> io::Result<()>, - mut sync: impl FnMut(&Path) -> io::Result<()>, -) -> Result<(), InventoryError> { - let root = parent(directory)?; - let root_metadata = fs::symlink_metadata(root).map_err(|source| state_io(root, source))?; - if root_metadata.file_type().is_symlink() || !root_metadata.is_dir() { - return Err(state_io( - root, - io::Error::new(io::ErrorKind::InvalidInput, "inventory state root is not a directory"), - )); +#[cfg(target_os = "linux")] +fn read_bounded(file: &mut fs::File) -> Result, InventoryError> { + let mut bytes = Vec::new(); + file.take((MAX_PERSISTED_BYTES + 1) as u64) + .read_to_end(&mut bytes) + .map_err(|_| InventoryError::StateIo)?; + if bytes.len() > MAX_PERSISTED_BYTES { + return Err(InventoryError::StateOversize); } - match create(directory) { - Ok(()) => {} - Err(source) if source.kind() == io::ErrorKind::AlreadyExists => { - let metadata = fs::symlink_metadata(directory).map_err(|source| state_io(directory, source))?; - if metadata.file_type().is_symlink() || !metadata.is_dir() { - return Err(state_io( - directory, - io::Error::new(io::ErrorKind::InvalidInput, "inventory state path is not a directory"), - )); - } + Ok(bytes) +} + +#[cfg(target_os = "linux")] +// SAFETY: libc path operations use validated directory descriptors and checked C strings; returned descriptors become owned files. +#[allow(unsafe_code)] +fn open_inventory_directory(path: &Path) -> Result<(StateRootAnchor, fs::File), InventoryError> { + use std::os::fd::AsRawFd as _; + use std::path::Component; + + let absolute = if path.is_absolute() { + path.to_path_buf() + } else { + std::env::current_dir().map_err(|_| InventoryError::StateIo)?.join(path) + }; + let root = fs::File::open("/").map_err(|_| InventoryError::StateIo)?; + validate_directory(&root, false)?; + let components = absolute.components().collect::>(); + let names = components + .iter() + .filter_map(|component| match component { + Component::Normal(name) => Some(name.to_os_string()), + Component::RootDir => None, + _ => Some(std::ffi::OsString::new()), + }) + .collect::>(); + if names.iter().any(|name| name.is_empty()) || names.is_empty() { + return Err(InventoryError::StatePath); + } + let component_count = names.len(); + let mut state_root = StateRootAnchor { + root, + components: Vec::with_capacity(component_count), + }; + for (index, name) in names.into_iter().enumerate() { + let parent = state_root + .components + .last() + .map(|(_, directory)| directory) + .unwrap_or(&state_root.root); + let directory = open_directory_component_at(parent, &name)?; + validate_directory(&directory, index + 1 == component_count)?; + state_root.components.push((name, directory)); + } + + let directory = state_root.state_root()?; + let inventory = c_name("inventory")?; + // SAFETY: descriptor and C string are valid; mode is applied only if the directory is created. + if unsafe { libc::mkdirat(directory.as_raw_fd(), inventory.as_ptr(), 0o700) } != 0 { + let error = io::Error::last_os_error(); + if error.kind() != io::ErrorKind::AlreadyExists { + return Err(InventoryError::StateIo); } - Err(source) => return Err(state_io(directory, source)), } - sync(directory).map_err(|source| state_io(directory, source))?; - sync(root).map_err(|source| state_io(root, source)) + let child = open_directory_at(directory, "inventory")?; + validate_directory(&child, true)?; + sync_inventory_anchor(&child, directory)?; + Ok((state_root, child)) } -fn create_inventory_directory(directory: &Path) -> io::Result<()> { - let mut builder = fs::DirBuilder::new(); - #[cfg(unix)] - { - use std::os::unix::fs::DirBuilderExt as _; - builder.mode(0o700); - } - builder.create(directory) +#[cfg(target_os = "linux")] +fn sync_inventory_anchor(inventory: &fs::File, state_root: &fs::File) -> Result<(), InventoryError> { + sync_inventory_anchor_with(|inventory_target| { + if inventory_target { + inventory.sync_all() + } else { + state_root.sync_all() + } + }) } -fn stage(directory: &Path, destination: &Path, bytes: &[u8]) -> Result { - let name = filename(destination)?; +#[cfg(any(target_os = "linux", test))] +fn sync_inventory_anchor_with(mut sync: impl FnMut(bool) -> io::Result<()>) -> Result<(), InventoryError> { + sync(true).map_err(|_| InventoryError::StateIo)?; + sync(false).map_err(|_| InventoryError::StateIo) +} + +#[cfg(target_os = "linux")] +fn open_directory_at(parent: &fs::File, name: &str) -> Result { + open_directory_component_at(parent, std::ffi::OsStr::new(name)) +} + +#[cfg(target_os = "linux")] +#[allow(unsafe_code)] +fn open_directory_component_at(parent: &fs::File, name: &std::ffi::OsStr) -> Result { + use std::os::fd::{AsRawFd as _, FromRawFd as _}; + use std::os::unix::ffi::OsStrExt as _; + let name = std::ffi::CString::new(name.as_bytes()).map_err(|_| InventoryError::StatePath)?; + // SAFETY: the parent descriptor and C string are valid; ownership of a successful descriptor is transferred. + let fd = unsafe { + libc::openat( + parent.as_raw_fd(), + name.as_ptr(), + libc::O_RDONLY | libc::O_CLOEXEC | libc::O_NOFOLLOW | libc::O_DIRECTORY, + ) + }; + if fd < 0 { + return Err(InventoryError::PersistenceSecurity); + } + // SAFETY: openat returned a new owned descriptor. + Ok(unsafe { fs::File::from_raw_fd(fd) }) +} + +#[cfg(target_os = "linux")] +fn validate_directory(directory: &fs::File, dedicated: bool) -> Result<(), InventoryError> { + use std::os::unix::fs::{MetadataExt as _, PermissionsExt as _}; + let metadata = directory.metadata().map_err(|_| InventoryError::StateIo)?; + let mode = metadata.permissions().mode() & 0o7777; + let uid = process_uid(); + if !metadata.is_dir() || !unix_directory_is_trusted(metadata.uid(), mode, uid, dedicated) { + return Err(InventoryError::PersistenceSecurity); + } + Ok(()) +} + +#[cfg(target_os = "linux")] +fn unix_directory_is_trusted(owner: u32, mode: u32, process: u32, dedicated: bool) -> bool { + let trusted_owner = owner == process || (!dedicated && owner == 0); + let trusted_mode = if dedicated { mode == 0o700 } else { mode & 0o7022 == 0 }; + trusted_owner && trusted_mode +} + +#[cfg(target_os = "linux")] +// SAFETY: openat receives a live directory descriptor and checked C string; a successful descriptor becomes an owned file. +#[allow(unsafe_code)] +fn open_file_at(directory: &fs::File, name: &str, create: bool, write: bool) -> Result { + use std::os::fd::{AsRawFd as _, FromRawFd as _}; + let name = c_name(name)?; + let mut flags = libc::O_CLOEXEC | libc::O_NOFOLLOW | libc::O_NONBLOCK; + flags |= if write { libc::O_RDWR } else { libc::O_RDONLY }; + if create { + flags |= libc::O_CREAT; + } + // SAFETY: the directory descriptor and C string are valid; ownership of a successful descriptor is transferred. + let fd = unsafe { libc::openat(directory.as_raw_fd(), name.as_ptr(), flags, FILE_MODE) }; + if fd < 0 { + let error = io::Error::last_os_error(); + return if error.kind() == io::ErrorKind::NotFound { + Err(InventoryError::StateMissing) + } else { + Err(InventoryError::PersistenceSecurity) + }; + } + // SAFETY: openat returned a new owned descriptor. + Ok(unsafe { fs::File::from_raw_fd(fd) }) +} + +#[cfg(target_os = "linux")] +// SAFETY: openat receives a live directory descriptor and checked C string; a successful descriptor becomes an owned file. +#[allow(unsafe_code)] +fn stage_at(directory: &fs::File, destination: &str) -> Result<(String, fs::File), InventoryError> { + use std::os::fd::{AsRawFd as _, FromRawFd as _}; loop { - let path = directory.join(format!( - ".{name}.{}.{}.tmp", + let name = format!( + ".{destination}.{}.{}.tmp", std::process::id(), STAGING_SEQUENCE.fetch_add(1, Ordering::Relaxed) - )); - let mut options = fs::OpenOptions::new(); - options.write(true).create_new(true); - #[cfg(unix)] - { - use std::os::unix::fs::OpenOptionsExt as _; - options.mode(FILE_MODE); - } - let mut file = match options.open(&path) { - Ok(file) => file, - Err(source) if source.kind() == io::ErrorKind::AlreadyExists => continue, - Err(source) => return Err(state_io(&path, source)), + ); + let c_name = c_name(&name)?; + // SAFETY: the directory descriptor and C string are valid; ownership of a successful descriptor is transferred. + let fd = unsafe { + libc::openat( + directory.as_raw_fd(), + c_name.as_ptr(), + libc::O_WRONLY | libc::O_CREAT | libc::O_EXCL | libc::O_CLOEXEC | libc::O_NOFOLLOW, + FILE_MODE, + ) }; - if let Err(source) = file.write_all(bytes).and_then(|()| file.sync_all()) { - let _ = fs::remove_file(&path); - return Err(state_io(&path, source)); + if fd >= 0 { + // SAFETY: openat returned a new owned descriptor. + let file = unsafe { fs::File::from_raw_fd(fd) }; + if let Err(error) = validate_regular_file(&file) { + let _ = unlink_at(directory, &name); + return Err(error); + } + return Ok((name, file)); + } + if io::Error::last_os_error().kind() != io::ErrorKind::AlreadyExists { + return Err(InventoryError::StateIo); } - return Ok(path); } } -fn state_io(path: &Path, source: io::Error) -> InventoryError { - InventoryError::StateIo { - path: path.to_path_buf(), - source, +#[cfg(target_os = "linux")] +#[allow(unsafe_code)] +fn rename_at(directory: &fs::File, source: &str, destination: &str) -> io::Result<()> { + use std::os::fd::AsRawFd as _; + let source = std::ffi::CString::new(source).map_err(|_| io::Error::from(io::ErrorKind::InvalidInput))?; + let destination = std::ffi::CString::new(destination).map_err(|_| io::Error::from(io::ErrorKind::InvalidInput))?; + // SAFETY: both names are valid C strings and both directory descriptors remain open. + if unsafe { libc::renameat(directory.as_raw_fd(), source.as_ptr(), directory.as_raw_fd(), destination.as_ptr()) } == 0 { + Ok(()) + } else { + Err(io::Error::last_os_error()) } } -#[cfg(unix)] -fn check_mode(path: &Path) -> Result<(), InventoryError> { - use std::os::unix::fs::PermissionsExt as _; +#[cfg(target_os = "linux")] +#[allow(unsafe_code)] +fn unlink_at(directory: &fs::File, name: &str) -> io::Result<()> { + use std::os::fd::AsRawFd as _; + let name = std::ffi::CString::new(name).map_err(|_| io::Error::from(io::ErrorKind::InvalidInput))?; + // SAFETY: the name is a valid C string and the directory descriptor remains open. + if unsafe { libc::unlinkat(directory.as_raw_fd(), name.as_ptr(), 0) } == 0 { + Ok(()) + } else { + Err(io::Error::last_os_error()) + } +} - let mode = fs::metadata(path) - .map_err(|source| state_io(path, source))? - .permissions() - .mode() - & 0o7777; - if mode != FILE_MODE { - return Err(InventoryError::StatePermissions { - path: path.to_path_buf(), - mode, - expected: FILE_MODE, - }); +#[cfg(target_os = "linux")] +fn validate_regular_file(file: &fs::File) -> Result<(), InventoryError> { + use std::os::unix::fs::{MetadataExt as _, PermissionsExt as _}; + let metadata = file.metadata().map_err(|_| InventoryError::StateIo)?; + if !metadata.is_file() + || !unix_regular_file_is_secure(metadata.uid(), metadata.permissions().mode() & 0o7777, metadata.nlink(), process_uid()) + { + return Err(InventoryError::PersistenceSecurity); } Ok(()) } -#[cfg(not(unix))] -fn check_mode(_path: &Path) -> Result<(), InventoryError> { - Ok(()) +#[cfg(target_os = "linux")] +fn unix_regular_file_is_secure(owner: u32, mode: u32, links: u64, process: u32) -> bool { + owner == process && mode == FILE_MODE && links == 1 } -fn fsync_dir(directory: &Path) -> io::Result<()> { - #[cfg(unix)] - fs::File::open(directory)?.sync_all()?; - #[cfg(not(unix))] - let _ = directory; - Ok(()) +#[cfg(target_os = "linux")] +#[allow(dead_code)] // Used by the crate-private stopped-server reader. +fn file_identity(file: &fs::File) -> Result<(u64, u64, u64), InventoryError> { + use std::os::unix::fs::MetadataExt as _; + let metadata = file.metadata().map_err(|_| InventoryError::StateIo)?; + Ok((metadata.dev(), metadata.ino(), metadata.nlink())) +} + +#[cfg(target_os = "linux")] +fn c_name(name: &str) -> Result { + std::ffi::CString::new(name).map_err(|_| InventoryError::StatePath) +} + +#[cfg(target_os = "linux")] +#[allow(unsafe_code)] +fn process_uid() -> u32 { + // SAFETY: geteuid has no pointer arguments or caller preconditions. + unsafe { libc::geteuid() } } #[derive(Debug, thiserror::Error)] pub enum InventoryError { - #[error("the RustFS inventory version is outside protocol bounds")] + #[error("connect_inventory_snapshot_version")] RustfsVersion, - #[error("the RustFS inventory operating-system version is outside protocol bounds")] + #[error("connect_inventory_snapshot_os_version")] OsVersion, - #[error("the RustFS inventory node count is outside protocol bounds")] + #[error("connect_inventory_snapshot_node_count")] NodeCount, - #[error("the RustFS inventory drive count is outside protocol bounds")] + #[error("connect_inventory_snapshot_drive_count")] DriveCount, - #[error("the RustFS inventory capacity is outside protocol bounds")] + #[error("connect_inventory_snapshot_capacity")] Capacity, - #[error("the RustFS inventory coarse flags are not canonical")] + #[error("connect_inventory_snapshot_flags")] CoarseFlags, - #[error("the RustFS inventory snapshot is incomplete: observed {observed} of {expected} configured drives")] + #[error("connect_inventory_snapshot_incomplete")] SnapshotIncomplete { expected: usize, observed: usize }, - #[error("the Connect inventory schedule is invalid")] + #[error("connect_inventory_schedule")] Schedule, - #[error("the Connect inventory sequence is exhausted")] + #[error("connect_inventory_sequence_exhausted")] SequenceExhausted, - #[error("a Connect inventory runtime already owns this state")] + #[error("connect_inventory_already_running")] AlreadyRunning, - #[error("the persisted Connect inventory changed while delivery was in flight")] + #[error("connect_inventory_state_conflict")] StateConflict, - #[error("the Connect inventory state path is invalid")] + #[error("connect_inventory_state_path")] StatePath, - #[error("Connect inventory state I/O failed at {path}: {source}")] - StateIo { - path: PathBuf, - #[source] - source: io::Error, - }, - #[error("Connect inventory state at {path} is invalid: {source}")] - StateInvalid { - path: PathBuf, - #[source] - source: serde_json::Error, - }, - #[error("Connect inventory state at {path} violates the protocol invariants")] - StateCorrupt { path: PathBuf }, - #[cfg(unix)] - #[error("Connect inventory state at {path} has mode {mode:o}, expected {expected:o}")] - StatePermissions { path: PathBuf, mode: u32, expected: u32 }, - #[error("Connect returned an invalid inventory response")] + #[error("connect_inventory_state_missing")] + StateMissing, + #[error("connect_inventory_state_io")] + StateIo, + #[error("connect_inventory_state_invalid")] + StateInvalid, + #[error("connect_inventory_state_corrupt")] + StateCorrupt, + #[error("connect_inventory_state_oversize")] + StateOversize, + #[error("connect_inventory_persistence_security")] + PersistenceSecurity, + #[error("connect_inventory_platform_security")] + PlatformSecurity, + #[error("connect_inventory_cancelled")] + Cancelled, + #[error("connect_inventory_durability_after_commit")] + DurabilityAfterCommit, + #[error("connect_inventory_envelope_invalid")] + EnvelopeInvalid, + #[error("connect_inventory_envelope_version")] + EnvelopeVersion, + #[error("connect_inventory_envelope_timestamp")] + EnvelopeTimestamp, + #[error("connect_inventory_envelope_future")] + EnvelopeFuture, + #[error("connect_inventory_envelope_hash")] + EnvelopeHash, + #[error("connect_inventory_response")] Response, - #[error(transparent)] + #[error("connect_inventory_json")] Json(#[from] serde_json::Error), - #[error("Connect inventory delivery failed: {0}")] - Telemetry(String), + #[error("connect_inventory_telemetry")] + Telemetry, } impl From for InventoryError { - fn from(error: TelemetryError) -> Self { - Self::Telemetry(error.to_string()) + fn from(_error: TelemetryError) -> Self { + Self::Telemetry } } #[cfg(test)] mod tests { - use std::cell::RefCell; - use std::rc::Rc; - use super::*; - fn create_directory(path: &Path) -> io::Result<()> { - fs::create_dir(path) + fn safe_tempdir() -> tempfile::TempDir { + let temp = tempfile::tempdir_in(env!("CARGO_MANIFEST_DIR")).expect("safe temporary directory"); + #[cfg(target_os = "linux")] + { + use std::os::unix::fs::PermissionsExt as _; + fs::set_permissions(temp.path(), fs::Permissions::from_mode(0o700)).expect("private temporary directory"); + } + temp + } + + #[cfg(not(target_os = "linux"))] + #[test] + fn persistence_fails_closed_before_accessing_state() { + let temp = safe_tempdir(); + let state = temp.path().join("state-must-not-be-created"); + + assert!(matches!( + InventoryStateStore::from_state_root(&state), + Err(InventoryError::PlatformSecurity) + )); + assert!(!state.exists()); + } + + #[cfg(target_os = "linux")] + #[allow(unsafe_code)] + fn make_fifo(path: &Path) { + use std::os::unix::ffi::OsStrExt as _; + + let path = std::ffi::CString::new(path.as_os_str().as_bytes()).expect("FIFO path"); + // SAFETY: the path is a valid C string and mkfifo does not retain it. + assert_eq!(unsafe { libc::mkfifo(path.as_ptr(), FILE_MODE as libc::mode_t) }, 0, "create FIFO"); + } + + fn snapshot() -> InventorySnapshot { + InventorySnapshot::new("1.2.3", None, 2, 4, 1_000, 400, [InventoryFlag::DriveOffline]).expect("snapshot") } #[test] - fn inventory_directory_creation_is_synced_before_state_can_be_committed() { - let temp = tempfile::tempdir().expect("tempdir"); - let root = temp.path().join("root"); - fs::create_dir(&root).expect("state root"); - let directory = root.join("inventory"); - let events = Rc::new(RefCell::new(Vec::new())); - let create_events = events.clone(); - let sync_events = events.clone(); - - prepare_inventory_directory_with( - &directory, - move |path| { - create_events.borrow_mut().push(format!("mkdir:{}", path.display())); - fs::create_dir(path) - }, - move |path| { - sync_events.borrow_mut().push(format!("sync:{}", path.display())); - Ok(()) - }, - ) - .expect("durable directory"); + fn envelope_has_stable_canonical_bytes_and_hash() { + let bytes = encode_envelope(snapshot(), "2026-08-23T01:02:03Z".to_owned()).expect("envelope"); assert_eq!( - events.borrow().as_slice(), - [ - format!("mkdir:{}", directory.display()), - format!("sync:{}", directory.display()), - format!("sync:{}", root.display()), - ] + String::from_utf8(bytes).expect("JSON"), + r#"{"formatVersion":"v1","capturedAt":"2026-08-23T01:02:03Z","snapshot":{"rustfsVersion":"1.2.3","osVersion":null,"nodeCount":2,"driveCount":4,"capacityTotalBytes":1000,"capacityUsedBytes":400,"coarseFlags":["drive.offline"]},"envelopeHash":"fb927e66c9635e0020b97993c868636bff1f8ddefe01b49345286de6239865e3"}"# ); } #[test] - fn inventory_directory_sync_failure_prevents_state_commit_and_is_retried() { - let temp = tempfile::tempdir().expect("tempdir"); - let root = temp.path().join("root"); - fs::create_dir(&root).expect("state root"); - let directory = root.join("inventory"); - let state = directory.join("state.json"); - let error = prepare_inventory_directory_with(&directory, create_directory, |path| { - if path == directory { - Err(io::Error::other("injected leaf sync failure")) - } else { - Ok(()) - } - }) - .expect_err("sync failure"); - assert!(matches!(error, InventoryError::StateIo { path, .. } if path == directory)); - assert!(!state.exists()); - - let error = prepare_inventory_directory_with(&directory, create_directory, |path| { - if path == root { - Err(io::Error::other("injected parent sync failure")) - } else { - Ok(()) - } - }) - .expect_err("parent sync failure"); - assert!(matches!(error, InventoryError::StateIo { path, .. } if path == root)); - assert!(!state.exists()); - - let mut synced = Vec::new(); - prepare_inventory_directory_with(&directory, create_directory, |path| { - synced.push(path.to_path_buf()); - Ok(()) - }) - .expect("retry must sync an already-created leaf"); - assert_eq!(synced, [directory, root]); + fn telemetry_failures_are_normalized_before_runtime_status() { + assert_eq!(InventoryError::from(TelemetryError::Endpoint).to_string(), "connect_inventory_telemetry"); } #[test] - fn inventory_directory_requires_its_fixed_root_to_exist() { - let temp = tempfile::tempdir().expect("tempdir"); - let root = temp.path().join("missing-root"); - let directory = root.join("inventory"); + fn inventory_anchor_syncs_child_then_state_root_and_retries_failures() { + let mut calls = Vec::new(); + let error = sync_inventory_anchor_with(|inventory_target| { + calls.push(inventory_target); + if !inventory_target { + Err(io::Error::other("injected state-root sync failure")) + } else { + Ok(()) + } + }) + .expect_err("state-root sync failure"); + assert!(matches!(error, InventoryError::StateIo)); + assert_eq!(calls, [true, false]); + + calls.clear(); + sync_inventory_anchor_with(|inventory_target| { + calls.push(inventory_target); + Ok(()) + }) + .expect("retry syncs the whole anchor"); + assert_eq!(calls, [true, false]); + } + + #[test] + fn reader_rejects_tampering_unknown_members_and_future_time() { + let now = chrono::DateTime::parse_from_rfc3339("2026-08-23T01:02:03Z") + .expect("time") + .with_timezone(&chrono::Utc); + let valid = encode_envelope(snapshot(), "2026-08-23T01:02:03Z".to_owned()).expect("envelope"); + assert_eq!(decode_envelope(&valid, now).expect("valid envelope").age, Duration::ZERO); + + let mut hash_tampered: serde_json::Value = serde_json::from_slice(&valid).expect("JSON"); + hash_tampered["snapshot"]["nodeCount"] = 3.into(); + assert!(matches!( + decode_envelope(&serde_json::to_vec(&hash_tampered).expect("JSON"), now), + Err(InventoryError::EnvelopeHash) + )); + + let mut timestamp_tampered: serde_json::Value = serde_json::from_slice(&valid).expect("JSON"); + timestamp_tampered["capturedAt"] = "2026-08-23T01:02:04Z".into(); + assert!(matches!( + decode_envelope(&serde_json::to_vec(×tamp_tampered).expect("JSON"), now), + Err(InventoryError::EnvelopeHash) + )); + + let mut unknown: serde_json::Value = serde_json::from_slice(&valid).expect("JSON"); + unknown["extra"] = true.into(); + assert!(matches!( + decode_envelope(&serde_json::to_vec(&unknown).expect("JSON"), now), + Err(InventoryError::EnvelopeInvalid) + )); + + let mut unknown_snapshot: serde_json::Value = serde_json::from_slice(&valid).expect("JSON"); + unknown_snapshot["snapshot"]["hostname"] = "secret.invalid".into(); + assert!(matches!( + decode_envelope(&serde_json::to_vec(&unknown_snapshot).expect("JSON"), now), + Err(InventoryError::EnvelopeInvalid) + )); + + let mut version: serde_json::Value = serde_json::from_slice(&valid).expect("JSON"); + version["formatVersion"] = "v2".into(); + assert!(matches!( + decode_envelope(&serde_json::to_vec(&version).expect("JSON"), now), + Err(InventoryError::EnvelopeVersion) + )); + + let mut timestamp: serde_json::Value = serde_json::from_slice(&valid).expect("JSON"); + timestamp["capturedAt"] = "2026-08-23T01:02:03.000Z".into(); + assert!(matches!( + decode_envelope(&serde_json::to_vec(×tamp).expect("JSON"), now), + Err(InventoryError::EnvelopeTimestamp) + )); + + let future = encode_envelope(snapshot(), "2026-08-23T01:07:04Z".to_owned()).expect("envelope"); + assert!(matches!(decode_envelope(&future, now), Err(InventoryError::EnvelopeFuture))); + let skew_boundary = encode_envelope(snapshot(), "2026-08-23T01:07:03Z".to_owned()).expect("envelope"); + assert_eq!(decode_envelope(&skew_boundary, now).expect("five-minute skew").age, Duration::ZERO); + let old = encode_envelope(snapshot(), "2026-08-22T01:02:03Z".to_owned()).expect("envelope"); + assert_eq!(decode_envelope(&old, now).expect("old envelope").age, Duration::from_secs(24 * 60 * 60)); + } + + #[cfg(target_os = "linux")] + #[test] + fn cancelled_publish_keeps_last_good_and_removes_staging_file() { + let temp = safe_tempdir(); + let store = InventoryStateStore::from_state_root(temp.path()).expect("store"); + let last_good = encode_envelope(snapshot(), "2026-08-23T01:02:03Z".to_owned()).expect("last good"); + store.replace_file("latest.json", &last_good, || false).expect("seed latest"); + let token = tokio_util::sync::CancellationToken::new(); + token.cancel(); + let replacement = InventorySnapshot::new("1.2.3", None, 2, 4, 1_001, 401, []).expect("replacement"); + let error = store + .publish_latest_sync(replacement, "2026-08-23T02:02:03Z".to_owned(), &token) + .expect_err("cancelled before commit"); + + assert!(matches!(error, InventoryError::Cancelled)); + assert_eq!(fs::read(temp.path().join("inventory/latest.json")).expect("last good"), last_good); + let entries = fs::read_dir(temp.path().join("inventory")) + .expect("inventory directory") + .map(|entry| entry.expect("entry").file_name()) + .collect::>(); + assert_eq!(entries, vec![std::ffi::OsString::from("latest.json")]); + } + + #[cfg(target_os = "linux")] + #[test] + fn reader_bounds_the_opened_latest_file_and_reports_missing_state() { + use std::os::unix::fs::PermissionsExt as _; + + let temp = safe_tempdir(); + let store = InventoryStateStore::from_state_root(temp.path()).expect("store"); + let now = chrono::DateTime::parse_from_rfc3339("2026-08-23T01:02:03Z") + .expect("time") + .with_timezone(&chrono::Utc); + assert!(matches!(store.read_latest(now), Err(InventoryError::StateMissing))); + + let latest = temp.path().join("inventory/latest.json"); + fs::write(&latest, vec![b'x'; MAX_PERSISTED_BYTES]).expect("bounded latest"); + fs::set_permissions(&latest, fs::Permissions::from_mode(0o600)).expect("mode"); + assert!(matches!(store.read_latest(now), Err(InventoryError::EnvelopeInvalid))); + fs::write(&latest, vec![b'x'; MAX_PERSISTED_BYTES + 1]).expect("oversized latest"); + assert!(matches!(store.read_latest(now), Err(InventoryError::StateOversize))); + fs::write(&latest, b"not-json").expect("corrupt latest"); + assert!(matches!(store.read_latest(now), Err(InventoryError::EnvelopeInvalid))); + } + + #[cfg(target_os = "linux")] + #[test] + fn fifo_children_are_rejected_without_blocking_readers_or_writers() { + let latest_temp = safe_tempdir(); + let latest_store = InventoryStateStore::from_state_root(latest_temp.path()).expect("latest store"); + let latest = latest_temp.path().join("inventory/latest.json"); + make_fifo(&latest); + let now = chrono::DateTime::parse_from_rfc3339("2026-08-23T01:02:03Z") + .expect("time") + .with_timezone(&chrono::Utc); + assert!(matches!(latest_store.read_latest(now), Err(InventoryError::PersistenceSecurity))); + assert!(matches!( + latest_store.publish_latest_sync( + snapshot(), + "2026-08-23T01:02:03Z".to_owned(), + &tokio_util::sync::CancellationToken::new() + ), + Err(InventoryError::PersistenceSecurity) + )); + + let state_temp = safe_tempdir(); + let state_store = InventoryStateStore::from_state_root(state_temp.path()).expect("state store"); + let state = state_temp.path().join("inventory/state.json"); + make_fifo(&state); + assert!(matches!(state_store.read(), Err(InventoryError::PersistenceSecurity))); + assert!(matches!( + state_store.write(&InventoryState::default()), + Err(InventoryError::PersistenceSecurity) + )); + } + + #[cfg(target_os = "linux")] + #[test] + fn reader_rejects_insecure_file_modes_and_hardlinks() { + use std::os::unix::fs::{PermissionsExt as _, symlink}; + + let temp = safe_tempdir(); + let store = InventoryStateStore::from_state_root(temp.path()).expect("store"); + store + .publish_latest_sync(snapshot(), "2026-08-23T01:02:03Z".to_owned(), &tokio_util::sync::CancellationToken::new()) + .expect("publish"); + let latest = temp.path().join("inventory/latest.json"); + fs::set_permissions(&latest, fs::Permissions::from_mode(0o4600)).expect("special-bit mode"); + let now = chrono::DateTime::parse_from_rfc3339("2026-08-23T01:02:03Z") + .expect("time") + .with_timezone(&chrono::Utc); + assert!(matches!(store.read_latest(now), Err(InventoryError::PersistenceSecurity))); + + fs::set_permissions(&latest, fs::Permissions::from_mode(0o644)).expect("mode"); + assert!(matches!(store.read_latest(now), Err(InventoryError::PersistenceSecurity))); + + fs::set_permissions(&latest, fs::Permissions::from_mode(0o600)).expect("mode"); + fs::hard_link(&latest, temp.path().join("inventory/second-link")).expect("hard link"); + assert!(matches!(store.read_latest(now), Err(InventoryError::PersistenceSecurity))); + + fs::remove_file(&latest).expect("remove latest link"); + symlink(temp.path().join("inventory/second-link"), &latest).expect("latest symlink"); + assert!(matches!(store.read_latest(now), Err(InventoryError::PersistenceSecurity))); + } + + #[cfg(target_os = "linux")] + #[test] + fn store_rejects_unsafe_ancestors_and_symlinked_inventory_directory() { + use std::os::unix::fs::{PermissionsExt as _, symlink}; + + let temp = safe_tempdir(); + let unsafe_parent = temp.path().join("unsafe"); + fs::create_dir(&unsafe_parent).expect("unsafe parent"); + fs::set_permissions(&unsafe_parent, fs::Permissions::from_mode(0o777)).expect("unsafe mode"); + let state = unsafe_parent.join("state"); + fs::create_dir(&state).expect("state root"); + fs::set_permissions(&state, fs::Permissions::from_mode(0o700)).expect("state mode"); + assert!(matches!( + InventoryStateStore::from_state_root(&state), + Err(InventoryError::PersistenceSecurity) + )); + + let safe_state = temp.path().join("safe-state"); + fs::create_dir(&safe_state).expect("safe state root"); + fs::set_permissions(&safe_state, fs::Permissions::from_mode(0o1700)).expect("special-bit state mode"); + assert!(matches!( + InventoryStateStore::from_state_root(&safe_state), + Err(InventoryError::PersistenceSecurity) + )); + fs::set_permissions(&safe_state, fs::Permissions::from_mode(0o700)).expect("state mode"); + symlink(temp.path(), safe_state.join("inventory")).expect("inventory symlink"); + assert!(matches!( + InventoryStateStore::from_state_root(&safe_state), + Err(InventoryError::PersistenceSecurity) + )); + } + + #[cfg(target_os = "linux")] + #[test] + fn unix_persistence_policy_rejects_wrong_owners_modes_and_link_counts() { + let uid = 501; + assert!(unix_directory_is_trusted(uid, 0o700, uid, true)); + assert!(!unix_directory_is_trusted(uid + 1, 0o700, uid, true)); + assert!(!unix_directory_is_trusted(uid, 0o755, uid, true)); + assert!(!unix_directory_is_trusted(uid, 0o1700, uid, true)); + assert!(unix_directory_is_trusted(0, 0o755, uid, false)); + assert!(!unix_directory_is_trusted(0, 0o777, uid, false)); + assert!(!unix_directory_is_trusted(0, 0o1755, uid, false)); + + assert!(unix_regular_file_is_secure(uid, 0o600, 1, uid)); + assert!(!unix_regular_file_is_secure(uid + 1, 0o600, 1, uid)); + assert!(!unix_regular_file_is_secure(uid, 0o644, 1, uid)); + assert!(!unix_regular_file_is_secure(uid, 0o4600, 1, uid)); + assert!(!unix_regular_file_is_secure(uid, 0o600, 2, uid)); + } + + #[cfg(target_os = "linux")] + #[test] + fn reader_rejects_a_real_wrong_owner_when_chown_is_permitted() { + use std::os::unix::fs::chown; + + let temp = safe_tempdir(); + let store = InventoryStateStore::from_state_root(temp.path()).expect("store"); + store + .publish_latest_sync(snapshot(), "2026-08-23T01:02:03Z".to_owned(), &tokio_util::sync::CancellationToken::new()) + .expect("publish"); + let latest = temp.path().join("inventory/latest.json"); + let wrong_uid = process_uid().checked_add(1).unwrap_or_else(|| process_uid() - 1); + if let Err(error) = chown(&latest, Some(wrong_uid), None) { + assert_eq!(error.kind(), io::ErrorKind::PermissionDenied, "unexpected chown failure"); + return; + } + let now = chrono::DateTime::parse_from_rfc3339("2026-08-23T01:02:03Z") + .expect("time") + .with_timezone(&chrono::Utc); + assert!(matches!(store.read_latest(now), Err(InventoryError::PersistenceSecurity))); + } + + #[cfg(target_os = "linux")] + #[test] + fn directory_component_exchange_is_rejected_by_the_open_anchor() { + use std::os::unix::fs::PermissionsExt as _; + + let temp = safe_tempdir(); + let store = InventoryStateStore::from_state_root(temp.path()).expect("store"); + store + .publish_latest_sync(snapshot(), "2026-08-23T01:02:03Z".to_owned(), &tokio_util::sync::CancellationToken::new()) + .expect("publish"); + fs::rename(temp.path().join("inventory"), temp.path().join("original-inventory")).expect("exchange original"); + fs::create_dir(temp.path().join("inventory")).expect("replacement inventory"); + fs::set_permissions(temp.path().join("inventory"), fs::Permissions::from_mode(0o700)).expect("replacement mode"); + + let now = chrono::DateTime::parse_from_rfc3339("2026-08-23T01:02:03Z") + .expect("time") + .with_timezone(&chrono::Utc); + assert!(matches!(store.read_latest(now), Err(InventoryError::PersistenceSecurity))); + assert!(matches!( + store.publish_latest_sync(snapshot(), "2026-08-23T02:02:03Z".to_owned(), &tokio_util::sync::CancellationToken::new()), + Err(InventoryError::PersistenceSecurity) + )); + assert!(!temp.path().join("inventory/latest.json").exists()); + } + + #[cfg(target_os = "linux")] + #[test] + fn state_root_and_ancestor_exchanges_are_rejected_by_the_path_anchor() { + use std::os::unix::fs::PermissionsExt as _; + + for exchange_ancestor in [false, true] { + let temp = safe_tempdir(); + let ancestor = temp.path().join("anchor"); + let state_root = ancestor.join("state"); + fs::create_dir_all(&state_root).expect("state root"); + fs::set_permissions(&ancestor, fs::Permissions::from_mode(0o700)).expect("ancestor mode"); + fs::set_permissions(&state_root, fs::Permissions::from_mode(0o700)).expect("state mode"); + let store = InventoryStateStore::from_state_root(&state_root).expect("store"); + store + .publish_latest_sync(snapshot(), "2026-08-23T01:02:03Z".to_owned(), &tokio_util::sync::CancellationToken::new()) + .expect("publish"); + + let exchanged = if exchange_ancestor { &ancestor } else { &state_root }; + fs::rename(exchanged, temp.path().join("original")).expect("exchange original directory"); + fs::create_dir_all(&state_root).expect("replacement state root"); + fs::set_permissions(&ancestor, fs::Permissions::from_mode(0o700)).expect("replacement ancestor mode"); + fs::set_permissions(&state_root, fs::Permissions::from_mode(0o700)).expect("replacement state mode"); + + let now = chrono::DateTime::parse_from_rfc3339("2026-08-23T01:02:03Z") + .expect("time") + .with_timezone(&chrono::Utc); + assert!(matches!(store.read_latest(now), Err(InventoryError::PersistenceSecurity))); + assert!(matches!( + store.publish_latest_sync( + snapshot(), + "2026-08-23T02:02:03Z".to_owned(), + &tokio_util::sync::CancellationToken::new() + ), + Err(InventoryError::PersistenceSecurity) + )); + assert!(!state_root.join("inventory/latest.json").exists()); + } + } + + #[cfg(target_os = "linux")] + #[test] + fn fresh_state_concurrency_creates_one_anchor_and_allows_one_runtime_owner() { + let temp = safe_tempdir(); + let state_root = Arc::new(temp.path().to_path_buf()); + let start = Arc::new(std::sync::Barrier::new(3)); + let release = Arc::new(std::sync::Barrier::new(3)); + let (result_tx, result_rx) = std::sync::mpsc::channel(); + let mut threads = Vec::new(); + for _ in 0..2 { + let state_root = state_root.clone(); + let start = start.clone(); + let release = release.clone(); + let result_tx = result_tx.clone(); + threads.push(std::thread::spawn(move || { + start.wait(); + let result = InventoryStateStore::from_state_root(&state_root).and_then(|store| store.try_runtime_lock()); + result_tx + .send(result.as_ref().map(|_| ()).map_err(|error| error.to_string())) + .expect("result"); + release.wait(); + result + })); + } + start.wait(); + let results = [ + result_rx.recv().expect("first result"), + result_rx.recv().expect("second result"), + ]; + assert_eq!(results.iter().filter(|result| result.is_ok()).count(), 1); + assert_eq!( + results + .iter() + .filter(|result| { matches!(result, Err(error) if error.as_str() == "connect_inventory_already_running") }) + .count(), + 1 + ); + release.wait(); + for thread in threads { + let _ = thread.join().expect("thread"); + } + assert!(state_root.join("inventory/.state.json.lock").is_file()); + assert!(!state_root.join("inventory/.state.lock").exists()); + } + + #[cfg(target_os = "linux")] + fn exchange_path_component(temp: &Path, state_root: &Path, component: &str) { + use std::os::unix::fs::PermissionsExt as _; + + let ancestor = state_root.parent().expect("state ancestor"); + let inventory = state_root.join("inventory"); + let exchanged = match component { + "ancestor" => ancestor, + "state-root" => state_root, + "inventory" => &inventory, + _ => unreachable!(), + }; + fs::rename(exchanged, temp.join(format!("original-{component}"))).expect("exchange path component"); + fs::create_dir_all(&inventory).expect("replacement inventory path"); + for directory in [ancestor, state_root, inventory.as_path()] { + fs::set_permissions(directory, fs::Permissions::from_mode(0o700)).expect("replacement directory mode"); + } + } + + #[cfg(target_os = "linux")] + #[test] + fn path_exchanges_during_reads_and_writes_fail_closed() { + for operation in ["read", "write"] { + for component in ["ancestor", "state-root", "inventory"] { + let temp = safe_tempdir(); + let ancestor = temp.path().join("anchor"); + let state_root = ancestor.join("state"); + fs::create_dir_all(&state_root).expect("state root"); + use std::os::unix::fs::PermissionsExt as _; + fs::set_permissions(&ancestor, fs::Permissions::from_mode(0o700)).expect("ancestor mode"); + fs::set_permissions(&state_root, fs::Permissions::from_mode(0o700)).expect("state mode"); + let store = InventoryStateStore::from_state_root(&state_root).expect("store"); + let captured_at = "2026-08-23T01:02:03Z"; + store + .publish_latest_sync(snapshot(), captured_at.to_owned(), &tokio_util::sync::CancellationToken::new()) + .expect("seed latest"); + + let error = if operation == "read" { + let now = chrono::DateTime::parse_from_rfc3339(captured_at) + .expect("time") + .with_timezone(&chrono::Utc); + store + .read_latest_after_open(now, || exchange_path_component(temp.path(), &state_root, component)) + .expect_err("path exchange during read") + } else { + let replacement = encode_envelope(snapshot(), "2026-08-23T02:02:03Z".to_owned()).expect("replacement"); + store + .replace_file_inner( + "latest.json", + &replacement, + || false, + None, + || exchange_path_component(temp.path(), &state_root, component), + ) + .expect_err("path exchange before commit") + }; + + assert!(matches!(error, InventoryError::PersistenceSecurity)); + assert!(!state_root.join("inventory/latest.json").exists()); + } + } + } + + #[cfg(target_os = "linux")] + #[tokio::test] + async fn legacy_pending_does_not_replace_a_newer_local_snapshot() { + let temp = safe_tempdir(); + let store = InventoryStateStore::from_state_root(temp.path()).expect("store"); + let current = snapshot(); + store + .publish_latest_sync( + current.clone(), + "2026-08-23T01:02:03Z".to_owned(), + &tokio_util::sync::CancellationToken::new(), + ) + .expect("current latest"); + let legacy = InventorySnapshot::new("1.2.3", None, 2, 4, 900, 300, []).expect("legacy snapshot"); + + store + .ensure_latest(legacy, "2026-08-23T02:02:03Z".to_owned(), tokio_util::sync::CancellationToken::new()) + .await + .expect("existing latest remains authoritative"); + + let now = chrono::DateTime::parse_from_rfc3339("2026-08-23T02:02:03Z") + .expect("time") + .with_timezone(&chrono::Utc); + let persisted = store.read_latest(now).expect("latest"); + assert_eq!(persisted.snapshot, current); + assert_eq!(persisted.captured_at, "2026-08-23T01:02:03Z"); + assert_eq!(persisted.age, Duration::from_secs(60 * 60)); + } + + #[cfg(target_os = "linux")] + #[test] + fn precommit_failures_preserve_last_good_and_postcommit_sync_failure_keeps_new_file() { + let temp = safe_tempdir(); + let store = InventoryStateStore::from_state_root(temp.path()).expect("store"); + let old = encode_envelope(snapshot(), "2026-08-23T01:02:03Z".to_owned()).expect("old envelope"); + store.replace_file("latest.json", &old, || false).expect("seed latest"); + let new_snapshot = InventorySnapshot::new("1.2.3", None, 2, 4, 1_001, 401, []).expect("new snapshot"); + let new = encode_envelope(new_snapshot, "2026-08-23T02:02:03Z".to_owned()).expect("new envelope"); + + for fault in [PersistFault::Write, PersistFault::TempSync, PersistFault::Rename] { + assert!(matches!( + store.replace_file_inner("latest.json", &new, || false, Some(fault), || {}), + Err(InventoryError::StateIo) + )); + assert_eq!(fs::read(temp.path().join("inventory/latest.json")).expect("last good"), old); + assert_eq!( + fs::read_dir(temp.path().join("inventory")) + .expect("inventory directory") + .filter_map(Result::ok) + .count(), + 1, + "failed staging must be removed" + ); + } assert!(matches!( - prepare_inventory_directory_with(&directory, create_directory, |_| Ok(())), - Err(InventoryError::StateIo { path, .. }) if path == root + store.replace_file_inner("latest.json", &new, || false, Some(PersistFault::DirectorySync), || {}), + Err(InventoryError::DurabilityAfterCommit) )); - assert!(!directory.exists()); + assert_eq!(fs::read(temp.path().join("inventory/latest.json")).expect("committed latest"), new); + } + + #[cfg(target_os = "linux")] + #[test] + fn cancellation_during_commit_does_not_interrupt_rename_or_sync() { + let temp = safe_tempdir(); + let store = InventoryStateStore::from_state_root(temp.path()).expect("store"); + let replacement = encode_envelope(snapshot(), "2026-08-23T01:02:03Z".to_owned()).expect("envelope"); + let cancellation = tokio_util::sync::CancellationToken::new(); + + store + .replace_file_inner( + "latest.json", + &replacement, + || cancellation.is_cancelled(), + Some(PersistFault::CancelDuringCommit(cancellation.clone())), + || {}, + ) + .expect("commit ignores cancellation after its cancellation gate"); + + assert!(cancellation.is_cancelled()); + assert_eq!( + fs::read(temp.path().join("inventory/latest.json")).expect("committed latest"), + replacement + ); + } + + #[cfg(target_os = "linux")] + #[test] + fn reader_rejects_replacement_of_the_file_it_opened() { + let temp = safe_tempdir(); + let store = InventoryStateStore::from_state_root(temp.path()).expect("store"); + let captured_at = "2026-08-23T01:02:03Z"; + let first = encode_envelope(snapshot(), captured_at.to_owned()).expect("first envelope"); + let second_snapshot = InventorySnapshot::new("1.2.3", None, 2, 4, 1_001, 401, []).expect("second snapshot"); + let second = encode_envelope(second_snapshot.clone(), captured_at.to_owned()).expect("second envelope"); + store.replace_file("latest.json", &first, || false).expect("seed latest"); + let now = chrono::DateTime::parse_from_rfc3339(captured_at) + .expect("time") + .with_timezone(&chrono::Utc); + + assert!(matches!( + store.read_latest_after_open(now, || { + store + .replace_file("latest.json", &second, || false) + .expect("replace opened file"); + }), + Err(InventoryError::PersistenceSecurity) + )); + assert_eq!(store.read_latest(now).expect("replacement").snapshot, second_snapshot); + } + + #[cfg(target_os = "linux")] + #[test] + fn concurrent_reader_observes_only_complete_old_or_new_envelopes() { + let temp = safe_tempdir(); + let store = InventoryStateStore::from_state_root(temp.path()).expect("store"); + let first = snapshot(); + let second = InventorySnapshot::new("1.2.3", None, 2, 4, 1_001, 401, []).expect("second snapshot"); + let captured_at = "2026-08-23T01:02:03Z"; + store + .replace_file( + "latest.json", + &encode_envelope(first.clone(), captured_at.to_owned()).expect("first envelope"), + || false, + ) + .expect("seed latest"); + let writer = store.clone(); + let first_writer = first.clone(); + let second_writer = second.clone(); + let start = Arc::new(std::sync::Barrier::new(2)); + let writer_start = start.clone(); + let thread = std::thread::spawn(move || { + writer_start.wait(); + for index in 0..100 { + let snapshot = if index % 2 == 0 { + first_writer.clone() + } else { + second_writer.clone() + }; + let bytes = encode_envelope(snapshot, captured_at.to_owned()).expect("envelope"); + writer.replace_file("latest.json", &bytes, || false).expect("atomic replace"); + } + }); + let now = chrono::DateTime::parse_from_rfc3339(captured_at) + .expect("time") + .with_timezone(&chrono::Utc); + start.wait(); + for _ in 0..100 { + match store.read_latest(now) { + Ok(observed) => assert!(observed.snapshot == first || observed.snapshot == second), + Err(InventoryError::PersistenceSecurity) => {} + Err(error) => panic!("reader observed neither a complete envelope nor a replacement: {error}"), + } + } + thread.join().expect("writer"); + let observed = store.read_latest(now).expect("stable final envelope").snapshot; + assert!(observed == first || observed == second); } } diff --git a/rustfs/src/connect/runtime.rs b/rustfs/src/connect/runtime.rs index eb7652eb1..7f2b60625 100644 --- a/rustfs/src/connect/runtime.rs +++ b/rustfs/src/connect/runtime.rs @@ -32,7 +32,6 @@ pub struct HeartbeatRuntime { shutdown: CancellationToken, status: watch::Receiver, task: Option>, - inventory: Option, } impl HeartbeatRuntime { @@ -40,22 +39,11 @@ impl HeartbeatRuntime { self.status.clone() } - pub(crate) fn with_inventory(mut self, inventory: Option) -> Self { - self.inventory = inventory; - self - } - pub async fn shutdown(mut self) { self.shutdown.cancel(); - if let Some(inventory) = self.inventory.as_ref() { - inventory.shutdown.cancel(); - } if let Some(task) = self.task.take() { let _ = task.await; } - if let Some(inventory) = self.inventory.take() { - inventory.shutdown().await; - } } } @@ -90,6 +78,20 @@ impl Drop for InventoryRuntime { } } +pub(crate) async fn shutdown_connect_runtimes(heartbeat: Option, inventory: Option) { + let heartbeat = async move { + if let Some(runtime) = heartbeat { + runtime.shutdown().await; + } + }; + let inventory = async move { + if let Some(runtime) = inventory { + runtime.shutdown().await; + } + }; + tokio::join!(heartbeat, inventory); +} + pub fn spawn_heartbeat_runtime( config: Option, parent_shutdown: &CancellationToken, @@ -101,6 +103,9 @@ where let Some(config) = config else { return Ok(None); }; + if !config.transport_enabled() { + return Ok(None); + } let sender = HeartbeatSender::new(config.clone())?; let store = HeartbeatStateStore::new(config.state_path.clone()); let lock = store.try_runtime_lock()?; @@ -163,7 +168,6 @@ where shutdown, status: status_rx, task: Some(task), - inventory: None, })) } @@ -184,9 +188,14 @@ where return Err(InventoryError::Schedule); } let retry_schedule = config.schedule; - let store = InventoryStateStore::from_heartbeat_path(&config.state_path)?; + let state_root = config.state_root().ok_or(InventoryError::StatePath)?; + let store = InventoryStateStore::from_state_root(state_root)?; let lock = store.try_runtime_lock()?; - let sender = InventorySender::new(config)?; + let sender = if config.transport_enabled() { + Some(InventorySender::new(config)?) + } else { + None + }; let shutdown = parent_shutdown.child_token(); let task_shutdown = shutdown.clone(); let (status_tx, status_rx) = watch::channel(InventoryStatus::Starting); @@ -197,8 +206,19 @@ where if task_shutdown.is_cancelled() { break; } - let pending = match store.pending().await { - Ok(Some(pending)) => pending, + let pending = match if sender.is_some() { store.pending().await } else { Ok(None) } { + Ok(Some((pending, captured_at))) => { + if let Err(error) = store + .ensure_latest(pending.snapshot().clone(), captured_at, task_shutdown.clone()) + .await + { + if matches!(error, InventoryError::Cancelled) && task_shutdown.is_cancelled() { + break; + } + return failed_inventory(&status_tx, error); + } + pending + } Ok(None) => { let snapshot = match cancellable(&task_shutdown, sample()).await { Some(Ok(snapshot)) => snapshot, @@ -218,6 +238,27 @@ where Ok(content_hash) => content_hash, Err(error) => return failed_inventory(&status_tx, error), }; + let captured_at = Utc::now().format("%Y-%m-%dT%H:%M:%SZ").to_string(); + if let Err(error) = store + .publish_latest(snapshot.clone(), captured_at, task_shutdown.clone()) + .await + { + if matches!(error, InventoryError::Cancelled) && task_shutdown.is_cancelled() { + break; + } + return failed_inventory(&status_tx, error); + } + if task_shutdown.is_cancelled() { + break; + } + if sender.is_none() { + backoff = retry_schedule.initial_backoff; + let _ = status_tx.send(InventoryStatus::Unchanged { content_hash }); + if sleep_or_cancel(&task_shutdown, schedule.cadence.saturating_add(jitter(schedule.jitter))).await { + break; + } + continue; + } match store.prepare(snapshot).await { Ok(Some(pending)) => pending, Ok(None) => { @@ -233,7 +274,15 @@ where } Err(error) => return failed_inventory(&status_tx, error), }; - let delivery = match cancellable(&task_shutdown, sender.send(&pending)).await { + let delivery = match cancellable( + &task_shutdown, + sender + .as_ref() + .expect("sender exists when delivery state is prepared") + .send(&pending), + ) + .await + { Some(Ok(delivery)) => delivery, Some(Err(error)) => return failed_inventory(&status_tx, error), None => break, @@ -261,14 +310,13 @@ where let _ = status_tx.send(InventoryStatus::BackingOff { delay }); delay } - InventoryDelivery::AuthenticationStopped { status, reason } => { - let _ = status_tx.send(InventoryStatus::AuthenticationStopped { status, reason }); + InventoryDelivery::AuthenticationStopped { status } => { + let _ = status_tx.send(InventoryStatus::AuthenticationStopped { status, reason: None }); return; } - InventoryDelivery::Rejected { status, reason } => { - let suffix = reason.map_or_else(String::new, |reason| format!("; reason={reason}")); + InventoryDelivery::Rejected { status } => { let _ = status_tx.send(InventoryStatus::Failed { - reason: format!("Connect rejected inventory with HTTP {status}{suffix}"), + reason: format!("connect_inventory_rejected_http_{status}"), }); return; } @@ -292,6 +340,46 @@ fn failed(status: &watch::Sender, error: HeartbeatError) { }); } +pub(crate) fn heartbeat_failure_reason(error: &HeartbeatError) -> &'static str { + use super::registration::CredentialValidationError; + + match error { + HeartbeatError::Endpoint => "connect_heartbeat_endpoint", + HeartbeatError::RootCertificate => "connect_heartbeat_root_certificate", + HeartbeatError::Schedule => "connect_heartbeat_schedule", + HeartbeatError::NotRegistered => "connect_heartbeat_not_registered", + HeartbeatError::IdentityMissing => "connect_heartbeat_identity_missing", + HeartbeatError::IdentityCertificate => "connect_heartbeat_identity_certificate", + HeartbeatError::CredentialName => "connect_heartbeat_credential_name", + HeartbeatError::CredentialExpired => "connect_heartbeat_credential_expired", + HeartbeatError::NodeSummary => "connect_heartbeat_node_summary", + HeartbeatError::SequenceExhausted => "connect_heartbeat_sequence_exhausted", + HeartbeatError::AlreadyRunning => "connect_heartbeat_already_running", + HeartbeatError::StateConflict => "connect_heartbeat_state_conflict", + HeartbeatError::StateIo { .. } => "connect_heartbeat_state_io", + HeartbeatError::StateInvalid { .. } => "connect_heartbeat_state_invalid", + HeartbeatError::StateCorrupt { .. } => "connect_heartbeat_state_corrupt", + #[cfg(unix)] + HeartbeatError::StatePermissions { .. } => "connect_heartbeat_state_permissions", + HeartbeatError::ResponseTooLarge => "connect_heartbeat_response_too_large", + HeartbeatError::Response => "connect_heartbeat_response", + HeartbeatError::Url(_) => "connect_heartbeat_url", + HeartbeatError::Transport(_) => "connect_heartbeat_transport", + HeartbeatError::Identity(_) => "connect_heartbeat_identity", + HeartbeatError::IdentityStore(_) => "connect_heartbeat_identity_store", + HeartbeatError::CredentialStore(_) => "connect_heartbeat_credential_store", + HeartbeatError::CredentialValidation(error) => match error { + CredentialValidationError::Certificate => "connect_heartbeat_credential_certificate", + CredentialValidationError::Chain => "connect_heartbeat_credential_chain", + CredentialValidationError::Identity => "connect_heartbeat_credential_identity", + CredentialValidationError::Key => "connect_heartbeat_credential_key", + CredentialValidationError::Validity => "connect_heartbeat_credential_validity", + CredentialValidationError::CertificateRequest => "connect_heartbeat_credential_request", + CredentialValidationError::RotationTranscript => "connect_heartbeat_credential_rotation_transcript", + }, + } +} + fn failed_inventory(status: &watch::Sender, error: InventoryError) { let _ = status.send(InventoryStatus::Failed { reason: error.to_string(), @@ -327,7 +415,51 @@ mod tests { use super::*; #[tokio::test] - async fn heartbeat_shutdown_cancels_inventory_before_waiting_for_heartbeat() { + async fn state_only_configuration_does_not_start_heartbeat() { + let temp = tempfile::tempdir_in(env!("CARGO_MANIFEST_DIR")).expect("safe temporary directory"); + let config = HeartbeatConfig::state_only(temp.path().to_path_buf()); + let shutdown = CancellationToken::new(); + + assert!( + spawn_heartbeat_runtime(Some(config), &shutdown, || { CoarseNodeSummary::new(1, 0, 0).expect("summary") }) + .expect("disabled transport") + .is_none() + ); + } + + #[test] + fn runtimes_expose_only_stable_machine_reasons() { + let error = HeartbeatError::StateIo { + path: std::path::PathBuf::from("/private/connect/state.json"), + source: std::io::Error::other("transport.internal"), + }; + assert_eq!(heartbeat_failure_reason(&error), "connect_heartbeat_state_io"); + assert_eq!( + heartbeat_failure_reason(&HeartbeatError::StateCorrupt { + path: std::path::PathBuf::from("/private/connect/state.json"), + }), + "connect_heartbeat_state_corrupt" + ); + assert_eq!( + heartbeat_failure_reason(&HeartbeatError::CredentialExpired), + "connect_heartbeat_credential_expired" + ); + assert_eq!( + heartbeat_failure_reason(&HeartbeatError::CredentialValidation( + super::super::registration::CredentialValidationError::Identity + )), + "connect_heartbeat_credential_identity" + ); + assert_eq!( + heartbeat_failure_reason(&HeartbeatError::CredentialValidation( + super::super::registration::CredentialValidationError::Key + )), + "connect_heartbeat_credential_key" + ); + } + + #[tokio::test] + async fn unified_shutdown_cancels_inventory_before_waiting_for_heartbeat() { let heartbeat_shutdown = CancellationToken::new(); let inventory_shutdown = CancellationToken::new(); let task_inventory_shutdown = inventory_shutdown.clone(); @@ -342,18 +474,18 @@ mod tests { task_inventory_shutdown.cancelled().await; let _ = inventory_stopped.send(()); }); - let runtime = HeartbeatRuntime { + let heartbeat = HeartbeatRuntime { shutdown: heartbeat_shutdown, status: heartbeat_status, task: Some(heartbeat_task), - inventory: Some(InventoryRuntime { - shutdown: inventory_shutdown, - status: inventory_status, - task: Some(inventory_task), - }), + }; + let inventory = InventoryRuntime { + shutdown: inventory_shutdown, + status: inventory_status, + task: Some(inventory_task), }; - let shutdown = tokio::spawn(runtime.shutdown()); + let shutdown = tokio::spawn(shutdown_connect_runtimes(Some(heartbeat), Some(inventory))); tokio::time::timeout(Duration::from_millis(250), stopped) .await .expect("inventory cancellation must not wait for heartbeat") diff --git a/rustfs/src/startup_lifecycle.rs b/rustfs/src/startup_lifecycle.rs index ad5681366..5586511c1 100644 --- a/rustfs/src/startup_lifecycle.rs +++ b/rustfs/src/startup_lifecycle.rs @@ -14,6 +14,7 @@ use crate::storage_api::startup::lifecycle::ECStore; use crate::{ + connect::runtime::shutdown_connect_runtimes, server::{ServiceStateManager, ShutdownHandle, start_persisted_event_notifier_reconciler, wait_for_shutdown}, startup_iam::{IamBootstrapDisposition, publish_ready_for_iam_bootstrap}, startup_runtime_sources, @@ -129,6 +130,7 @@ pub(crate) async fn run_startup_runtime_lifecycle(lifecycle: StartupRuntimeLifec let StartupServiceRuntime { optional_runtimes, heartbeat, + inventory, iam_bootstrap, enable_scanner, } = service_runtime; @@ -163,9 +165,7 @@ pub(crate) async fn run_startup_runtime_lifecycle(lifecycle: StartupRuntimeLifec shutdown_token, ) .await; - if let Some(heartbeat) = heartbeat { - heartbeat.shutdown().await; - } + shutdown_connect_runtimes(heartbeat, inventory).await; if let Err(err) = event_notifier_reconciler.await { tracing::warn!( target: "rustfs::main::run", diff --git a/rustfs/src/startup_services.rs b/rustfs/src/startup_services.rs index f8c509d71..8a0292615 100644 --- a/rustfs/src/startup_services.rs +++ b/rustfs/src/startup_services.rs @@ -17,8 +17,9 @@ use crate::storage_api::startup::services::{ECStore, EndpointServerPools, Server use crate::{ config::Config, connect::{ - CoarseNodeSummary, HeartbeatConfig, HeartbeatRuntime, InventoryError, InventoryFlag, InventoryRuntime, InventorySchedule, - InventorySnapshot, spawn_heartbeat_runtime, spawn_inventory_runtime, + CoarseNodeSummary, HeartbeatConfig, HeartbeatError, HeartbeatRuntime, InventoryError, InventoryFlag, InventoryRuntime, + InventorySchedule, InventorySnapshot, runtime::heartbeat_failure_reason, spawn_heartbeat_runtime, + spawn_inventory_runtime, }, init::{init_buffer_profile_system, init_kms_system}, server::ServiceStateManager, @@ -40,6 +41,7 @@ use tokio_util::sync::CancellationToken; pub(crate) struct StartupServiceRuntime { pub(crate) optional_runtimes: OptionalRuntimeServices, pub(crate) heartbeat: Option, + pub(crate) inventory: Option, pub(crate) iam_bootstrap: IamBootstrapDisposition, pub(crate) enable_scanner: bool, } @@ -104,11 +106,11 @@ pub(crate) async fn init_startup_runtime_services( init_observability_runtime(store.clone(), ctx.clone()).await; let heartbeat = start_heartbeat_runtime(heartbeat_config.clone(), heartbeat_nodes, &ctx)?; let inventory = start_inventory_runtime(heartbeat_config, heartbeat_nodes, inventory_drives, store, &ctx)?; - let heartbeat = heartbeat.map(|heartbeat| heartbeat.with_inventory(inventory)); Ok(StartupServiceRuntime { optional_runtimes, heartbeat, + inventory, iam_bootstrap, enable_scanner, }) @@ -122,11 +124,18 @@ fn start_heartbeat_runtime( let Some(config) = config else { return Ok(None); }; + if !config.transport_enabled() { + return Ok(None); + } let summary = u16::try_from(node_count.unwrap_or_default()) .ok() .and_then(|total| CoarseNodeSummary::new(total, 0, 0).ok()) .ok_or_else(|| std::io::Error::other("Connect heartbeat node count is outside protocol bounds"))?; - spawn_heartbeat_runtime(Some(config), shutdown, move || summary).map_err(std::io::Error::other) + spawn_heartbeat_runtime(Some(config), shutdown, move || summary).map_err(startup_heartbeat_error) +} + +fn startup_heartbeat_error(error: HeartbeatError) -> std::io::Error { + std::io::Error::other(heartbeat_failure_reason(&error)) } fn start_inventory_runtime( @@ -318,6 +327,16 @@ fn aggregate_inventory_capacity( mod tests { use super::*; + #[test] + fn heartbeat_startup_errors_expose_only_stable_codes() { + let error = startup_heartbeat_error(HeartbeatError::StateIo { + path: std::path::PathBuf::from("/private/connect/canary/state.json"), + source: std::io::Error::other("private-source-canary"), + }); + + assert_eq!(error.to_string(), "connect_heartbeat_state_io"); + } + fn disk(state: &str, runtime_state: Option<&str>, disk_index: i32) -> rustfs_madmin::Disk { rustfs_madmin::Disk { state: state.to_owned(), diff --git a/rustfs/tests/connect_inventory.rs b/rustfs/tests/connect_inventory.rs index 3aa28f3ca..0261320fc 100644 --- a/rustfs/tests/connect_inventory.rs +++ b/rustfs/tests/connect_inventory.rs @@ -12,6 +12,8 @@ // See the License for the specific language governing permissions and // limitations under the License. +#![cfg(target_os = "linux")] + use std::collections::VecDeque; use std::fs; use std::sync::atomic::{AtomicUsize, Ordering}; @@ -46,6 +48,10 @@ const CLUSTER_UID: &str = "0198f4b0-2b00-7d20-9e31-3f4a5b6c7d81"; const DEVICE_UID: &str = "0198f4b0-3c00-7e30-8f41-4a5b6c7d8e92"; const SNAPSHOT_UID: &str = "0198f4b0-4d00-7f40-9051-5b6c7d8e9fa3"; +fn safe_tempdir() -> tempfile::TempDir { + tempfile::tempdir_in(env!("CARGO_MANIFEST_DIR")).expect("safe temporary directory") +} + struct TestPki { root_params: CertificateParams, root_key: KeyPair, @@ -244,6 +250,7 @@ fn config(temp: &tempfile::TempDir, pki: &TestPki, server: &TestServer) -> Heart if let Err(error) = fs::create_dir(temp.path().join("private-config-secret")) { assert_eq!(error.kind(), std::io::ErrorKind::AlreadyExists, "Connect state root"); } + private_directory_mode(&temp.path().join("private-config-secret")); HeartbeatConfig { endpoint: server.endpoint.clone(), root_ca_pem: pki.root_pem.as_bytes().to_vec(), @@ -394,6 +401,38 @@ fn connect_inventory_bounds_fail_instead_of_truncating_or_inventing_values() { )); } +#[cfg(target_os = "linux")] +#[tokio::test] +async fn connect_inventory_state_only_persists_without_constructing_transport() { + let temp = safe_tempdir(); + let state = temp.path().join("state"); + fs::create_dir(&state).expect("state root"); + private_directory_mode(&state); + 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), schedule(), &shutdown, || std::future::ready(Ok(snapshot()))) + .expect("state-only inventory") + .expect("configured inventory"); + let mut status = runtime.status(); + + assert!(matches!( + wait_for(&mut status, |status| matches!(status, InventoryStatus::Unchanged { .. })).await, + InventoryStatus::Unchanged { .. } + )); + let envelope: Value = serde_json::from_slice(&fs::read(state.join("inventory/latest.json")).expect("latest inventory")) + .expect("latest envelope"); + assert_eq!(envelope["formatVersion"], "v1"); + assert_eq!(envelope["snapshot"], serde_json::to_value(snapshot()).expect("snapshot JSON")); + assert_eq!(envelope.as_object().expect("envelope object").len(), 4); + runtime.shutdown().await; +} + #[tokio::test] async fn connect_inventory_rejects_deserialized_invalid_snapshots_before_state_or_network() { let pki = TestPki::new(); @@ -409,7 +448,7 @@ async fn connect_inventory_rejects_deserialized_invalid_snapshots_before_state_o "coarseFlags": [] })) .expect("serde should not bypass the runtime validation boundary"); - let temp = tempfile::tempdir().expect("tempdir"); + let temp = safe_tempdir(); let shutdown = CancellationToken::new(); let runtime = spawn_inventory_runtime(Some(config(&temp, &pki, &server)), schedule(), &shutdown, move || { std::future::ready(Ok(invalid.clone())) @@ -420,7 +459,7 @@ async fn connect_inventory_rejects_deserialized_invalid_snapshots_before_state_o assert!(matches!( wait_for(&mut status, |status| matches!(status, InventoryStatus::Failed { .. })).await, - InventoryStatus::Failed { reason } if reason.contains("version is outside protocol bounds") + InventoryStatus::Failed { reason } if reason == "connect_inventory_snapshot_version" )); assert!(!temp.path().join("private-config-secret/inventory/state.json").exists()); runtime.shutdown().await; @@ -433,7 +472,7 @@ async fn connect_inventory_restart_replays_the_pending_request_and_then_skips_un let pki = TestPki::new(); let content_hash = snapshot().content_hash().expect("content hash"); let first_server = server(&pki, vec![Reply::error(StatusCode::SERVICE_UNAVAILABLE, "UNAVAILABLE")]).await; - let temp = tempfile::tempdir().expect("tempdir"); + let temp = safe_tempdir(); let shutdown = CancellationToken::new(); let samples = Arc::new(AtomicUsize::new(0)); let sampled = samples.clone(); @@ -451,7 +490,40 @@ async fn connect_inventory_restart_replays_the_pending_request_and_then_skips_un )); assert_eq!(samples.load(Ordering::Relaxed), 1); let original = first_server.seen.lock().expect("seen lock")[0].clone(); + let latest: Value = serde_json::from_slice( + &fs::read(temp.path().join("private-config-secret/inventory/latest.json")).expect("latest inventory"), + ) + .expect("latest envelope"); + assert_eq!(latest["snapshot"], serde_json::to_value(snapshot()).expect("snapshot JSON")); + for field in [ + "rustfsVersion", + "osVersion", + "nodeCount", + "driveCount", + "capacityTotalBytes", + "capacityUsedBytes", + "coarseFlags", + ] { + assert_eq!(latest["snapshot"][field], original[field]); + } runtime.shutdown().await; + let state = temp.path().join("private-config-secret/inventory/state.json"); + let legacy_persisted_at = std::time::SystemTime::now() - Duration::from_secs(60 * 60); + fs::File::options() + .write(true) + .open(&state) + .expect("pending state") + .set_times(std::fs::FileTimes::new().set_modified(legacy_persisted_at)) + .expect("legacy pending timestamp"); + let legacy_persisted_at = chrono::DateTime::::from( + fs::metadata(&state) + .and_then(|metadata| metadata.modified()) + .expect("persisted pending timestamp"), + ) + .format("%Y-%m-%dT%H:%M:%SZ") + .to_string(); + fs::remove_file(temp.path().join("private-config-secret/inventory/latest.json")) + .expect("simulate a pending snapshot created before local persistence"); let mut limited = Reply::error(StatusCode::TOO_MANY_REQUESTS, "RATE_LIMITED"); limited.retry_after = Some("0"); @@ -473,10 +545,17 @@ async fn connect_inventory_restart_replays_the_pending_request_and_then_skips_un if accepted == content_hash && received_at == "2026-08-22T01:02:03Z" )); assert_eq!(restart_samples.load(Ordering::Relaxed), 0); + let restored_latest: Value = serde_json::from_slice( + &fs::read(temp.path().join("private-config-secret/inventory/latest.json")).expect("restored latest inventory"), + ) + .expect("restored latest envelope"); + assert_eq!(restored_latest["snapshot"], serde_json::to_value(snapshot()).expect("snapshot JSON")); + assert_eq!(restored_latest["capturedAt"], legacy_persisted_at); let delivered = restart_server.seen.lock().expect("seen lock").clone(); assert_eq!(delivered, vec![original.clone(), original.clone()]); assert_eq!(original["sequence"], 0); let encoded = serde_json::to_string(&original).expect("request JSON"); + let persisted = serde_json::to_string(&latest).expect("persisted JSON"); for forbidden in [ "private-config-secret", "BEGIN CERTIFICATE", @@ -486,9 +565,17 @@ async fn connect_inventory_restart_replays_the_pending_request_and_then_skips_un "path", ] { assert!(!encoded.contains(forbidden), "request exposed {forbidden}"); + assert!(!persisted.contains(forbidden), "persisted inventory exposed {forbidden}"); } assert_eq!(original.as_object().expect("request object").len(), 10); restart.shutdown().await; + #[cfg(target_os = "linux")] + let latest_before_unchanged = { + use std::os::unix::fs::MetadataExt as _; + fs::metadata(temp.path().join("private-config-secret/inventory/latest.json")) + .expect("latest metadata") + .ino() + }; let unchanged_samples = Arc::new(AtomicUsize::new(0)); let sampled = unchanged_samples.clone(); @@ -505,6 +592,17 @@ async fn connect_inventory_restart_replays_the_pending_request_and_then_skips_un )); assert_eq!(unchanged_samples.load(Ordering::Relaxed), 1); assert_eq!(restart_server.seen.lock().expect("seen lock").len(), 2); + #[cfg(target_os = "linux")] + { + use std::os::unix::fs::MetadataExt as _; + assert_ne!( + fs::metadata(temp.path().join("private-config-secret/inventory/latest.json")) + .expect("refreshed latest metadata") + .ino(), + latest_before_unchanged, + "a complete unchanged sample must refresh the local envelope" + ); + } unchanged.shutdown().await; } @@ -512,7 +610,7 @@ async fn connect_inventory_restart_replays_the_pending_request_and_then_skips_un async fn connect_inventory_disconnect_retries_without_resampling() { let pki = TestPki::new(); let unavailable = server(&pki, Vec::new()).await; - let temp = tempfile::tempdir().expect("tempdir"); + let temp = safe_tempdir(); let config = config(&temp, &pki, &unavailable); drop(unavailable); let shutdown = CancellationToken::new(); @@ -542,7 +640,7 @@ async fn connect_inventory_retries_an_incomplete_sample_before_delivery() { let pki = TestPki::new(); let content_hash = snapshot().content_hash().expect("content hash"); let server = server(&pki, vec![Reply::ok(&content_hash)]).await; - let temp = tempfile::tempdir().expect("tempdir"); + let temp = safe_tempdir(); let shutdown = CancellationToken::new(); let samples = Arc::new(AtomicUsize::new(0)); let sampled = samples.clone(); @@ -561,6 +659,11 @@ async fn connect_inventory_retries_an_incomplete_sample_before_delivery() { .expect("configured inventory"); let mut status = runtime.status(); + assert!(matches!( + wait_for(&mut status, |status| matches!(status, InventoryStatus::BackingOff { .. })).await, + InventoryStatus::BackingOff { .. } + )); + assert!(!temp.path().join("private-config-secret/inventory/latest.json").exists()); assert!(matches!( wait_for(&mut status, |status| matches!(status, InventoryStatus::Online { .. })).await, InventoryStatus::Online { content_hash: accepted, .. } if accepted == content_hash @@ -575,7 +678,7 @@ async fn connect_inventory_unchanged_sample_resets_incomplete_backoff() { let pki = TestPki::new(); let content_hash = snapshot().content_hash().expect("content hash"); let server = server(&pki, vec![Reply::ok(&content_hash)]).await; - let temp = tempfile::tempdir().expect("tempdir"); + let temp = safe_tempdir(); let shutdown = CancellationToken::new(); let config = config(&temp, &pki, &server); let seed = spawn_inventory_runtime(Some(config.clone()), schedule(), &shutdown, || std::future::ready(Ok(snapshot()))) @@ -645,7 +748,7 @@ async fn connect_inventory_unchanged_sample_resets_incomplete_backoff() { async fn connect_inventory_revoked_device_stops_without_retrying() { let pki = TestPki::new(); let server = server(&pki, vec![Reply::error(StatusCode::UNAUTHORIZED, "DEVICE_REVOKED")]).await; - let temp = tempfile::tempdir().expect("tempdir"); + let temp = safe_tempdir(); let shutdown = CancellationToken::new(); let runtime = spawn_inventory_runtime(Some(config(&temp, &pki, &server)), schedule(), &shutdown, || { std::future::ready(Ok(snapshot())) @@ -656,7 +759,10 @@ async fn connect_inventory_revoked_device_stops_without_retrying() { assert!(matches!( wait_for(&mut status, |status| matches!(status, InventoryStatus::AuthenticationStopped { .. })).await, - InventoryStatus::AuthenticationStopped { status: 401, reason: Some(reason) } if reason == "DEVICE_REVOKED" + InventoryStatus::AuthenticationStopped { + status: 401, + reason: None + } )); assert_eq!(server.seen.lock().expect("seen lock").len(), 1); runtime.shutdown().await; @@ -666,10 +772,11 @@ async fn connect_inventory_revoked_device_stops_without_retrying() { async fn connect_inventory_sequence_overflow_fails_before_sampling_or_network_delivery() { let pki = TestPki::new(); let server = server(&pki, Vec::new()).await; - let temp = tempfile::tempdir().expect("tempdir"); + let temp = safe_tempdir(); let config = config(&temp, &pki, &server); let state = temp.path().join("private-config-secret/inventory/state.json"); fs::create_dir_all(state.parent().expect("state directory")).expect("create state directory"); + private_directory_mode(state.parent().expect("state directory")); fs::write( &state, br#"{"nextSequence":9007199254740992,"pending":null,"lastAcceptedContentHash":null}"#, @@ -689,7 +796,7 @@ async fn connect_inventory_sequence_overflow_fails_before_sampling_or_network_de assert!(matches!( wait_for(&mut status, |status| matches!(status, InventoryStatus::Failed { .. })).await, - InventoryStatus::Failed { reason } if reason.contains("sequence is exhausted") + InventoryStatus::Failed { reason } if reason == "connect_inventory_sequence_exhausted" )); assert_eq!(samples.load(Ordering::Relaxed), 0); assert!(server.seen.lock().expect("seen lock").is_empty()); @@ -702,10 +809,11 @@ async fn connect_inventory_rejects_noncanonical_persisted_snapshots_before_deliv let server = server(&pki, Vec::new()).await; for invalid_case in ["flags", "os-version"] { - let temp = tempfile::tempdir().expect("tempdir"); + let temp = safe_tempdir(); let config = config(&temp, &pki, &server); let state = temp.path().join("private-config-secret/inventory/state.json"); fs::create_dir_all(state.parent().expect("state directory")).expect("create state directory"); + private_directory_mode(state.parent().expect("state directory")); let mut pending = json!({ "protocolVersion": "v1", "requestId": "00000000-0000-4000-8000-000000000001", @@ -753,7 +861,7 @@ async fn connect_inventory_rejects_noncanonical_persisted_snapshots_before_deliv matches!(status, InventoryStatus::Failed { .. } | InventoryStatus::BackingOff { .. }) }) .await, - InventoryStatus::Failed { reason } if reason.contains("violates the protocol invariants") + InventoryStatus::Failed { reason } if reason == "connect_inventory_state_corrupt" )); assert_eq!(samples.load(Ordering::Relaxed), 0); runtime.shutdown().await; @@ -762,11 +870,20 @@ async fn connect_inventory_rejects_noncanonical_persisted_snapshots_before_deliv assert!(server.seen.lock().expect("seen lock").is_empty()); } -#[cfg(unix)] +#[cfg(target_os = "linux")] fn private_mode(path: &std::path::Path) { use std::os::unix::fs::PermissionsExt as _; fs::set_permissions(path, fs::Permissions::from_mode(0o600)).expect("private permissions"); } -#[cfg(not(unix))] +#[cfg(target_os = "linux")] +fn private_directory_mode(path: &std::path::Path) { + use std::os::unix::fs::PermissionsExt as _; + fs::set_permissions(path, fs::Permissions::from_mode(0o700)).expect("private directory permissions"); +} + +#[cfg(not(target_os = "linux"))] fn private_mode(_path: &std::path::Path) {} + +#[cfg(not(target_os = "linux"))] +fn private_directory_mode(_path: &std::path::Path) {}