diff --git a/rustfs/src/connect/heartbeat.rs b/rustfs/src/connect/heartbeat.rs index 2eeb8c135..1a95d0d6f 100644 --- a/rustfs/src/connect/heartbeat.rs +++ b/rustfs/src/connect/heartbeat.rs @@ -19,23 +19,19 @@ use std::sync::atomic::{AtomicU64, Ordering}; use std::time::Duration; use chrono::{DateTime, SecondsFormat, Utc}; -use reqwest::{Client, StatusCode, Url, header}; -use rustls::RootCertStore; -use rustls::pki_types::{CertificateDer, pem::PemObject as _}; use serde::{Deserialize, Serialize}; use uuid::Uuid; -use zeroize::Zeroizing; use super::config::HeartbeatConfig; -use super::credential_store::{CredentialStoreError, DeviceCredential}; +use super::credential_store::CredentialStoreError; use super::identity::IdentityError; use super::identity_store::StoreError; -use super::registration::{CredentialValidationError, validate_stored_credential}; +use super::registration::CredentialValidationError; +use super::telemetry::{TelemetryDelivery, TelemetryError, TelemetryTransport, is_exact_utc_seconds}; const PROTOCOL_VERSION: &str = "v1"; const AGENT_VERSION: &str = concat!("rustfs-agent/", env!("CARGO_PKG_VERSION")); const MAX_SEQUENCE: u64 = 9_007_199_254_740_991; -const MAX_RESPONSE_BYTES: usize = 64 * 1024; #[cfg(unix)] const FILE_MODE: u32 = 0o600; static STAGING_SEQUENCE: AtomicU64 = AtomicU64::new(0); @@ -122,133 +118,39 @@ pub(crate) enum Delivery { } pub(crate) struct HeartbeatSender { - endpoint: Url, - root_store: RootCertStore, - roots: Vec>, - config: HeartbeatConfig, + transport: TelemetryTransport, } impl HeartbeatSender { pub(crate) fn new(config: HeartbeatConfig) -> Result { - let mut endpoint = Url::parse(&config.endpoint).map_err(|_| HeartbeatError::Endpoint)?; - if endpoint.scheme() != "https" - || endpoint.cannot_be_a_base() - || !endpoint.username().is_empty() - || endpoint.password().is_some() - || endpoint.query().is_some() - || endpoint.fragment().is_some() - { - return Err(HeartbeatError::Endpoint); - } - if !endpoint.path().ends_with('/') { - endpoint.set_path(&format!("{}/", endpoint.path())); - } - let roots = CertificateDer::pem_slice_iter(&config.root_ca_pem) - .collect::, _>>() - .map_err(|_| HeartbeatError::RootCertificate)?; - if roots.is_empty() { - return Err(HeartbeatError::RootCertificate); - } - let mut root_store = RootCertStore::empty(); - let (accepted, rejected) = root_store.add_parsable_certificates(roots.clone()); - if accepted != roots.len() || rejected != 0 { - return Err(HeartbeatError::RootCertificate); - } let schedule = config.schedule; - if schedule.cadence.is_zero() - || schedule.timeout.is_zero() - || schedule.timeout > Duration::from_secs(5) - || schedule.initial_backoff.is_zero() - || schedule.max_backoff < schedule.initial_backoff - || schedule.max_backoff > Duration::from_secs(5 * 60) - || schedule.jitter > schedule.cadence - { + if schedule.cadence.is_zero() || schedule.jitter > schedule.cadence { return Err(HeartbeatError::Schedule); } Ok(Self { - endpoint, - root_store, - roots, - config, + transport: TelemetryTransport::new(config)?, }) } pub(crate) async fn send(&self, heartbeat: &PendingHeartbeat) -> Result { - let (cluster_uid, client) = { - let _lock = self.config.credential_store.lock().await?; - let credential = self.config.credential_store.load()?.ok_or(HeartbeatError::NotRegistered)?; - let identity = self.config.identity_store.load()?.ok_or(HeartbeatError::IdentityMissing)?; - validate_stored_credential(&credential, &identity, &self.root_store, &self.roots)?; - let now = Utc::now().timestamp(); - if now < credential.not_before_unix || now >= credential.not_after_unix { - return Err(HeartbeatError::CredentialExpired); + match self.transport.post("heartbeats", heartbeat).await? { + TelemetryDelivery::Accepted { body, .. } => { + let accepted: HeartbeatResponse = serde_json::from_slice(&body).map_err(|_| HeartbeatError::Response)?; + if accepted.accepted_version != PROTOCOL_VERSION + || accepted.capability_hints.len() > 32 + || accepted.capability_hints.iter().any(|hint| hint.len() > 32) + || !is_exact_utc_seconds(&accepted.server_time) + { + return Err(HeartbeatError::Response); + } + Ok(Delivery::Accepted { + server_time: accepted.server_time, + }) } - let cluster_uid = cluster_uid(&credential)?.to_owned(); - let client = self.client(&credential, &identity.to_pkcs8_pem()?)?; - (cluster_uid, client) - }; - let url = self.endpoint.join(&format!("clusters/{cluster_uid}/heartbeats"))?; - let response = match client.post(url).json(heartbeat).send().await { - Ok(response) => response, - Err(error) if error.is_timeout() || error.is_connect() || error.is_request() => { - return Ok(Delivery::Retry { retry_after: None }); - } - Err(error) => return Err(error.into()), - }; - let status = response.status(); - if status == StatusCode::TOO_MANY_REQUESTS { - return Ok(Delivery::Retry { - retry_after: retry_after(response.headers(), Utc::now(), self.config.schedule.max_backoff), - }); + TelemetryDelivery::Retry { retry_after } => Ok(Delivery::Retry { retry_after }), + TelemetryDelivery::AuthenticationStopped { status, reason } => Ok(Delivery::AuthenticationStopped { status, reason }), + TelemetryDelivery::Rejected { status, reason } => Ok(Delivery::Rejected { status, reason }), } - if status == StatusCode::REQUEST_TIMEOUT || status.is_server_error() { - return Ok(Delivery::Retry { retry_after: None }); - } - if matches!(status, StatusCode::UNAUTHORIZED | StatusCode::FORBIDDEN) { - return Ok(Delivery::AuthenticationStopped { - status: status.as_u16(), - reason: response_reason(response).await, - }); - } - if status != StatusCode::OK { - return Ok(Delivery::Rejected { - status: status.as_u16(), - reason: response_reason(response).await, - }); - } - let accepted: HeartbeatResponse = - serde_json::from_slice(&bounded_body(response).await?).map_err(|_| HeartbeatError::Response)?; - if accepted.accepted_version != PROTOCOL_VERSION - || accepted.capability_hints.len() > 32 - || accepted.capability_hints.iter().any(|hint| hint.len() > 32) - || !is_exact_utc_seconds(&accepted.server_time) - { - return Err(HeartbeatError::Response); - } - Ok(Delivery::Accepted { - server_time: accepted.server_time, - }) - } - - fn client(&self, credential: &DeviceCredential, key: &Zeroizing) -> Result { - let mut pem = Zeroizing::new(Vec::with_capacity(credential.certificate_chain.len() + key.len() + 1)); - pem.extend_from_slice(credential.certificate_chain.as_bytes()); - pem.push(b'\n'); - pem.extend_from_slice(key.as_bytes()); - let identity = reqwest::Identity::from_pem(&pem).map_err(|_| HeartbeatError::IdentityCertificate)?; - let roots = self - .roots - .iter() - .map(|root| reqwest::Certificate::from_der(root.as_ref())) - .collect::, _>>()?; - Client::builder() - .https_only(true) - .redirect(reqwest::redirect::Policy::none()) - .timeout(self.config.schedule.timeout) - .tls_certs_only(roots) - .identity(identity) - .build() - .map_err(Into::into) } } @@ -378,73 +280,6 @@ impl HeartbeatStateStore { } } -fn cluster_uid(credential: &DeviceCredential) -> Result<&str, HeartbeatError> { - let mut parts = credential.name.split('/'); - let valid = parts.next() == Some("organizations"); - let organization_uid = parts.next(); - let valid = valid && parts.next() == Some("clusters"); - let cluster_uid = parts.next(); - let valid = valid && parts.next() == Some("clusterDevices"); - let device_uid = parts.next(); - if !valid - || organization_uid.is_none_or(str::is_empty) - || cluster_uid.is_none_or(str::is_empty) - || device_uid != Some(credential.uid.as_str()) - || parts.next().is_some() - { - return Err(HeartbeatError::CredentialName); - } - cluster_uid.ok_or(HeartbeatError::CredentialName) -} - -fn retry_after(headers: &header::HeaderMap, now: DateTime, maximum: Duration) -> Option { - let value = headers.get(header::RETRY_AFTER)?.to_str().ok()?; - let delay = value.parse::().ok().map(Duration::from_secs).or_else(|| { - DateTime::parse_from_rfc2822(value) - .ok() - .and_then(|at| (at.with_timezone(&Utc) - now).to_std().ok()) - })?; - Some(delay.min(maximum)) -} - -fn is_exact_utc_seconds(value: &str) -> bool { - DateTime::parse_from_rfc3339(value).is_ok_and(|time| { - time.offset().local_minus_utc() == 0 - && value.ends_with('Z') - && time.with_timezone(&Utc).to_rfc3339_opts(SecondsFormat::Secs, true) == value - }) -} - -async fn response_reason(response: reqwest::Response) -> Option { - #[derive(Deserialize)] - struct Envelope { - #[serde(default)] - details: Vec, - } - #[derive(Deserialize)] - struct Detail { - #[serde(default)] - reason: String, - } - - serde_json::from_slice::(&bounded_body(response).await.ok()?) - .ok()? - .details - .into_iter() - .find_map(|detail| (!detail.reason.is_empty()).then_some(detail.reason)) -} - -async fn bounded_body(mut response: reqwest::Response) -> Result, HeartbeatError> { - let mut body = Vec::new(); - while let Some(chunk) = response.chunk().await? { - if body.len().saturating_add(chunk.len()) > MAX_RESPONSE_BYTES { - return Err(HeartbeatError::ResponseTooLarge); - } - body.extend_from_slice(&chunk); - } - Ok(body) -} - fn parent(path: &Path) -> Result<&Path, HeartbeatError> { path.parent() .ok_or_else(|| state_io(path, io::Error::new(io::ErrorKind::InvalidInput, "state path has no parent"))) @@ -583,3 +418,25 @@ pub enum HeartbeatError { #[error(transparent)] CredentialValidation(#[from] CredentialValidationError), } + +impl From for HeartbeatError { + fn from(error: TelemetryError) -> Self { + match error { + TelemetryError::Endpoint => Self::Endpoint, + TelemetryError::RootCertificate => Self::RootCertificate, + TelemetryError::Schedule => Self::Schedule, + TelemetryError::NotRegistered => Self::NotRegistered, + TelemetryError::IdentityMissing => Self::IdentityMissing, + TelemetryError::IdentityCertificate => Self::IdentityCertificate, + TelemetryError::CredentialName => Self::CredentialName, + TelemetryError::CredentialExpired => Self::CredentialExpired, + TelemetryError::ResponseTooLarge => Self::ResponseTooLarge, + TelemetryError::Url(error) => Self::Url(error), + TelemetryError::Transport(error) => Self::Transport(error), + TelemetryError::Identity(error) => Self::Identity(error), + TelemetryError::IdentityStore(error) => Self::IdentityStore(error), + TelemetryError::CredentialStore(error) => Self::CredentialStore(error), + TelemetryError::CredentialValidation(error) => Self::CredentialValidation(error), + } + } +} diff --git a/rustfs/src/connect/inventory.rs b/rustfs/src/connect/inventory.rs new file mode 100644 index 000000000..40c4f6cc5 --- /dev/null +++ b/rustfs/src/connect/inventory.rs @@ -0,0 +1,592 @@ +// Copyright 2024 RustFS Team +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +use std::collections::BTreeSet; +use std::fs; +use std::io::{self, Write as _}; +use std::path::{Path, PathBuf}; +use std::sync::atomic::{AtomicU64, Ordering}; +use std::time::Duration; + +use serde::{Deserialize, Serialize}; +use sha2::{Digest as _, Sha256}; +use uuid::Uuid; + +use super::config::HeartbeatConfig; +use super::telemetry::{TelemetryDelivery, TelemetryError, TelemetryTransport, is_exact_utc_seconds}; + +const PROTOCOL_VERSION: &str = "v1"; +const RUSTFS_VERSION: &str = concat!( + env!("CARGO_PKG_VERSION_MAJOR"), + ".", + env!("CARGO_PKG_VERSION_MINOR"), + ".", + env!("CARGO_PKG_VERSION_PATCH") +); +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 FILE_MODE: u32 = 0o600; +static STAGING_SEQUENCE: AtomicU64 = AtomicU64::new(0); + +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +pub struct InventorySchedule { + pub cadence: Duration, + pub jitter: Duration, +} + +impl Default for InventorySchedule { + fn default() -> Self { + Self { + cadence: Duration::from_secs(6 * 60 * 60), + jitter: Duration::from_secs(30 * 60), + } + } +} + +#[derive(Clone, Debug, PartialEq, Eq)] +pub enum InventoryStatus { + Starting, + Unchanged { content_hash: String }, + Online { content_hash: String, received_at: String }, + BackingOff { delay: Duration }, + AuthenticationStopped { status: u16, reason: Option }, + Failed { reason: String }, + Stopped, +} + +#[derive(Clone, Copy, Debug, PartialEq, Eq, PartialOrd, Ord, Serialize, Deserialize)] +pub enum InventoryFlag { + #[serde(rename = "capacity.critical")] + CapacityCritical, + #[serde(rename = "capacity.warning")] + CapacityWarning, + #[serde(rename = "clock.skew")] + ClockSkew, + #[serde(rename = "cluster.degraded")] + ClusterDegraded, + #[serde(rename = "cluster.healing")] + ClusterHealing, + #[serde(rename = "cluster.readonly")] + ClusterReadonly, + #[serde(rename = "drive.offline")] + DriveOffline, + #[serde(rename = "node.offline")] + NodeOffline, +} + +#[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize, Deserialize)] +#[serde(rename_all = "lowercase")] +pub enum OperatingSystemFamily { + Linux, + Darwin, + Windows, + Freebsd, + Other, +} + +#[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize, Deserialize)] +#[serde(deny_unknown_fields, rename_all = "camelCase")] +pub struct InventoryOsVersion { + family: OperatingSystemFamily, + major: u16, + minor: u16, +} + +impl InventoryOsVersion { + pub fn new(family: OperatingSystemFamily, major: u16, minor: u16) -> Result { + if major > 9999 || minor > 9999 { + return Err(InventoryError::OsVersion); + } + Ok(Self { family, major, minor }) + } +} + +#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] +#[serde(deny_unknown_fields, rename_all = "camelCase")] +pub struct InventorySnapshot { + rustfs_version: String, + os_version: Option, + node_count: u16, + drive_count: u32, + capacity_total_bytes: u64, + capacity_used_bytes: u64, + coarse_flags: Vec, +} + +impl InventorySnapshot { + pub fn current( + node_count: usize, + drive_count: usize, + capacity_total_bytes: u64, + capacity_free_bytes: u64, + coarse_flags: impl IntoIterator, + ) -> Result { + let capacity_used_bytes = capacity_total_bytes + .checked_sub(capacity_free_bytes) + .ok_or(InventoryError::Capacity)?; + Self::new( + RUSTFS_VERSION, + None, + node_count, + drive_count, + capacity_total_bytes, + capacity_used_bytes, + coarse_flags, + ) + } + + pub fn new( + rustfs_version: impl Into, + os_version: Option, + node_count: usize, + drive_count: usize, + capacity_total_bytes: u64, + capacity_used_bytes: u64, + coarse_flags: impl IntoIterator, + ) -> Result { + let snapshot = Self { + rustfs_version: rustfs_version.into(), + os_version, + node_count: u16::try_from(node_count).map_err(|_| InventoryError::NodeCount)?, + drive_count: u32::try_from(drive_count).map_err(|_| InventoryError::DriveCount)?, + capacity_total_bytes, + capacity_used_bytes, + coarse_flags: coarse_flags.into_iter().collect::>().into_iter().collect(), + }; + snapshot.validate()?; + Ok(snapshot) + } + + pub fn content_hash(&self) -> Result { + #[derive(Serialize)] + #[serde(rename_all = "camelCase")] + struct Canonical<'a> { + capacity_total_bytes: u64, + capacity_used_bytes: u64, + coarse_flags: &'a [InventoryFlag], + drive_count: u32, + node_count: u16, + os_version: Option, + rustfs_version: &'a str, + } + + let canonical = serde_json::to_vec(&Canonical { + capacity_total_bytes: self.capacity_total_bytes, + capacity_used_bytes: self.capacity_used_bytes, + coarse_flags: &self.coarse_flags, + drive_count: self.drive_count, + node_count: self.node_count, + os_version: self.os_version, + rustfs_version: &self.rustfs_version, + })?; + let mut digest = Sha256::new(); + digest.update(HASH_PREFIX); + digest.update(canonical); + Ok(hex_simd::encode_to_string(digest.finalize(), hex_simd::AsciiCase::Lower)) + } + + fn validate(&self) -> Result<(), InventoryError> { + if !valid_version(&self.rustfs_version) { + return Err(InventoryError::RustfsVersion); + } + if self.node_count == 0 || self.node_count > 4096 { + return Err(InventoryError::NodeCount); + } + if self.drive_count > 1_048_576 { + return Err(InventoryError::DriveCount); + } + if self.capacity_total_bytes > MAX_SAFE_INTEGER || self.capacity_used_bytes > self.capacity_total_bytes { + return Err(InventoryError::Capacity); + } + Ok(()) + } +} + +fn valid_version(version: &str) -> bool { + let components = version.split('.').collect::>(); + components.len() == 3 + && components.iter().all(|component| { + !component.is_empty() + && component.len() <= 4 + && (component == &"0" || !component.starts_with('0')) + && component.parse::().is_ok_and(|value| value <= 9999) + }) +} + +#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] +#[serde(deny_unknown_fields, rename_all = "camelCase")] +pub(crate) struct PendingInventory { + protocol_version: String, + request_id: String, + sequence: u64, + #[serde(flatten)] + snapshot: InventorySnapshot, +} + +impl PendingInventory { + fn new(snapshot: InventorySnapshot, sequence: u64) -> Self { + Self { + protocol_version: PROTOCOL_VERSION.to_owned(), + request_id: Uuid::new_v4().to_string(), + sequence, + snapshot, + } + } + + fn is_valid(&self) -> bool { + self.protocol_version == PROTOCOL_VERSION + && self.sequence <= MAX_SEQUENCE + && self.snapshot.validate().is_ok() + && Uuid::parse_str(&self.request_id) + .is_ok_and(|request_id| request_id.get_version_num() == 4 && request_id.to_string() == self.request_id) + } + + fn content_hash(&self) -> Result { + self.snapshot.content_hash() + } +} + +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 }, +} + +pub(crate) struct InventorySender { + transport: TelemetryTransport, +} + +impl InventorySender { + pub(crate) fn new(config: HeartbeatConfig) -> Result { + Ok(Self { + transport: TelemetryTransport::new(config)?, + }) + } + + pub(crate) async fn send(&self, inventory: &PendingInventory) -> Result { + match self.transport.post("inventorySnapshots", inventory).await? { + TelemetryDelivery::Accepted { cluster_name, body } => { + #[derive(Deserialize)] + #[serde(rename_all = "camelCase")] + struct InventoryResponse { + name: String, + uid: String, + content_hash: String, + received_at: String, + } + + let accepted: InventoryResponse = serde_json::from_slice(&body).map_err(|_| InventoryError::Response)?; + let uid = Uuid::parse_str(&accepted.uid).map_err(|_| InventoryError::Response)?; + let content_hash = inventory.content_hash()?; + if uid.get_version_num() != 7 + || uid.to_string() != accepted.uid + || accepted.name != format!("{cluster_name}/inventorySnapshots/{}", accepted.uid) + || accepted.content_hash != content_hash + || !is_exact_utc_seconds(&accepted.received_at) + { + return Err(InventoryError::Response); + } + Ok(InventoryDelivery::Accepted { + content_hash, + received_at: accepted.received_at, + }) + } + 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 }), + } + } +} + +#[derive(Clone)] +pub(crate) struct InventoryStateStore { + path: PathBuf, +} + +#[derive(Default, Serialize, Deserialize)] +#[serde(deny_unknown_fields, rename_all = "camelCase")] +struct InventoryState { + next_sequence: u64, + pending: Option, + last_accepted_content_hash: Option, +} + +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 try_runtime_lock(&self) -> Result { + let directory = parent(&self.path)?; + fs::create_dir_all(directory).map_err(|source| state_io(directory, source))?; + 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)] + { + use std::os::unix::fs::OpenOptionsExt as _; + options.mode(FILE_MODE); + } + 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> { + let store = self.clone(); + tokio::task::spawn_blocking(move || { + let state = store.read()?; + if state.pending.is_none() && state.next_sequence > MAX_SEQUENCE { + return Err(InventoryError::SequenceExhausted); + } + Ok(state.pending) + }) + .await + .map_err(|source| state_io(&self.path, io::Error::other(source)))? + } + + 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)))? + } + + pub(crate) async fn mark_accepted(&self, accepted: &PendingInventory) -> Result<(), InventoryError> { + let store = self.clone(); + 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)))? + } + + fn prepare_sync(&self, snapshot: InventorySnapshot) -> Result, InventoryError> { + let mut state = self.read()?; + if state.pending.is_some() { + return Ok(state.pending); + } + let content_hash = snapshot.content_hash()?; + if state.last_accepted_content_hash.as_deref() == Some(&content_hash) { + return Ok(None); + } + if state.next_sequence > MAX_SEQUENCE { + return Err(InventoryError::SequenceExhausted); + } + let pending = PendingInventory::new(snapshot, state.next_sequence); + state.pending = Some(pending.clone()); + self.write(&state)?; + Ok(Some(pending)) + } + + fn mark_accepted_sync(&self, accepted: &PendingInventory) -> Result<(), InventoryError> { + let mut state = self.read()?; + if state.pending.as_ref() != Some(accepted) { + return Err(InventoryError::StateConflict); + } + state.next_sequence = accepted.sequence.checked_add(1).ok_or(InventoryError::SequenceExhausted)?; + state.last_accepted_content_hash = Some(accepted.content_hash()?); + state.pending = None; + self.write(&state) + } + + 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() }); + } + 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)?; + fs::create_dir_all(directory).map_err(|source| state_io(directory, source))?; + 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 + } +} + +fn valid_content_hash(value: &str) -> bool { + value.len() == 64 + && value + .bytes() + .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 stage(directory: &Path, destination: &Path, bytes: &[u8]) -> Result { + let name = filename(destination)?; + loop { + let path = directory.join(format!( + ".{name}.{}.{}.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)), + }; + if let Err(source) = file.write_all(bytes).and_then(|()| file.sync_all()) { + let _ = fs::remove_file(&path); + return Err(state_io(&path, source)); + } + return Ok(path); + } +} + +fn state_io(path: &Path, source: io::Error) -> InventoryError { + InventoryError::StateIo { + path: path.to_path_buf(), + source, + } +} + +#[cfg(unix)] +fn check_mode(path: &Path) -> Result<(), InventoryError> { + use std::os::unix::fs::PermissionsExt as _; + + 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, + }); + } + Ok(()) +} + +#[cfg(not(unix))] +fn check_mode(_path: &Path) -> Result<(), InventoryError> { + Ok(()) +} + +fn fsync_dir(directory: &Path) -> io::Result<()> { + #[cfg(unix)] + fs::File::open(directory)?.sync_all()?; + #[cfg(not(unix))] + let _ = directory; + Ok(()) +} + +#[derive(Debug, thiserror::Error)] +pub enum InventoryError { + #[error("the RustFS inventory version is outside protocol bounds")] + RustfsVersion, + #[error("the RustFS inventory operating-system version is outside protocol bounds")] + OsVersion, + #[error("the RustFS inventory node count is outside protocol bounds")] + NodeCount, + #[error("the RustFS inventory drive count is outside protocol bounds")] + DriveCount, + #[error("the RustFS inventory capacity is outside protocol bounds")] + Capacity, + #[error("the Connect inventory schedule is invalid")] + Schedule, + #[error("the Connect inventory sequence is exhausted")] + SequenceExhausted, + #[error("a Connect inventory runtime already owns this state")] + AlreadyRunning, + #[error("the persisted Connect inventory changed while delivery was in flight")] + StateConflict, + #[error("the Connect inventory state path is invalid")] + 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")] + Response, + #[error(transparent)] + Json(#[from] serde_json::Error), + #[error("Connect inventory delivery failed: {0}")] + Telemetry(String), +} + +impl From for InventoryError { + fn from(error: TelemetryError) -> Self { + Self::Telemetry(error.to_string()) + } +} diff --git a/rustfs/src/connect/mod.rs b/rustfs/src/connect/mod.rs index 3ce2ad975..50d6927b4 100644 --- a/rustfs/src/connect/mod.rs +++ b/rustfs/src/connect/mod.rs @@ -31,9 +31,11 @@ pub mod credential_store; pub mod heartbeat; pub mod identity; pub mod identity_store; +pub mod inventory; pub mod offline; pub mod registration; pub mod runtime; +mod telemetry; pub use client::{ClientError, ConnectClient, ConnectConfig}; pub use config::{HeartbeatConfig, HeartbeatConfigError, HeartbeatSchedule}; @@ -41,6 +43,10 @@ pub use credential_store::{CredentialStore, DeviceCredential}; pub use heartbeat::{CoarseNodeSummary, HeartbeatError, HeartbeatStatus}; pub use identity::{DeviceIdentity, IdentityError, RegistrationProof, RegistrationTranscript}; pub use identity_store::{IdentityStore, StoreError}; +pub use inventory::{ + InventoryError, InventoryFlag, InventoryOsVersion, InventorySchedule, InventorySnapshot, InventoryStatus, + OperatingSystemFamily, +}; pub use offline::{EnrollmentError, OfflineEnrollment, OfflineKeyStore, VerifiedChallenge}; pub use registration::{RegistrationToken, TokenError}; -pub use runtime::{HeartbeatRuntime, spawn_heartbeat_runtime}; +pub use runtime::{HeartbeatRuntime, InventoryRuntime, spawn_heartbeat_runtime, spawn_inventory_runtime}; diff --git a/rustfs/src/connect/runtime.rs b/rustfs/src/connect/runtime.rs index f0d919a07..0f8882bca 100644 --- a/rustfs/src/connect/runtime.rs +++ b/rustfs/src/connect/runtime.rs @@ -23,11 +23,16 @@ use tokio_util::sync::CancellationToken; use super::config::HeartbeatConfig; use super::heartbeat::{CoarseNodeSummary, Delivery, HeartbeatError, HeartbeatSender, HeartbeatStateStore, HeartbeatStatus}; +use super::inventory::{ + InventoryDelivery, InventoryError, InventorySchedule, InventorySender, InventorySnapshot, InventoryStateStore, + InventoryStatus, +}; pub struct HeartbeatRuntime { shutdown: CancellationToken, status: watch::Receiver, task: Option>, + inventory: Option, } impl HeartbeatRuntime { @@ -35,6 +40,39 @@ 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(task) = self.task.take() { + let _ = task.await; + } + if let Some(inventory) = self.inventory.take() { + inventory.shutdown().await; + } + } +} + +impl Drop for HeartbeatRuntime { + fn drop(&mut self) { + self.shutdown.cancel(); + } +} + +pub struct InventoryRuntime { + shutdown: CancellationToken, + status: watch::Receiver, + task: Option>, +} + +impl InventoryRuntime { + pub fn status(&self) -> watch::Receiver { + self.status.clone() + } + pub async fn shutdown(mut self) { self.shutdown.cancel(); if let Some(task) = self.task.take() { @@ -43,7 +81,7 @@ impl HeartbeatRuntime { } } -impl Drop for HeartbeatRuntime { +impl Drop for InventoryRuntime { fn drop(&mut self) { self.shutdown.cancel(); } @@ -122,6 +160,116 @@ where shutdown, status: status_rx, task: Some(task), + inventory: None, + })) +} + +pub fn spawn_inventory_runtime( + config: Option, + schedule: InventorySchedule, + parent_shutdown: &CancellationToken, + sample: F, +) -> Result, InventoryError> +where + F: Fn() -> Fut + Send + Sync + 'static, + Fut: Future> + Send + 'static, +{ + let Some(config) = config else { + return Ok(None); + }; + if schedule.cadence.is_zero() || schedule.jitter > schedule.cadence { + return Err(InventoryError::Schedule); + } + let retry_schedule = config.schedule; + let store = InventoryStateStore::from_heartbeat_path(&config.state_path)?; + let lock = store.try_runtime_lock()?; + let sender = InventorySender::new(config)?; + let shutdown = parent_shutdown.child_token(); + let task_shutdown = shutdown.clone(); + let (status_tx, status_rx) = watch::channel(InventoryStatus::Starting); + let task = tokio::spawn(async move { + let _lock = lock; + let mut backoff = retry_schedule.initial_backoff; + loop { + if task_shutdown.is_cancelled() { + break; + } + let pending = match store.pending().await { + Ok(Some(pending)) => pending, + Ok(None) => { + let snapshot = match cancellable(&task_shutdown, sample()).await { + Some(Ok(snapshot)) => snapshot, + Some(Err(error)) => return failed_inventory(&status_tx, error), + None => break, + }; + let content_hash = match snapshot.content_hash() { + Ok(content_hash) => content_hash, + Err(error) => return failed_inventory(&status_tx, error), + }; + match store.prepare(snapshot).await { + Ok(Some(pending)) => pending, + Ok(None) => { + let _ = status_tx.send(InventoryStatus::Unchanged { content_hash }); + if sleep_or_cancel(&task_shutdown, schedule.cadence.saturating_add(jitter(schedule.jitter))).await { + break; + } + continue; + } + Err(error) => return failed_inventory(&status_tx, error), + } + } + Err(error) => return failed_inventory(&status_tx, error), + }; + let delivery = match cancellable(&task_shutdown, sender.send(&pending)).await { + Some(Ok(delivery)) => delivery, + Some(Err(error)) => return failed_inventory(&status_tx, error), + None => break, + }; + let delay = match delivery { + InventoryDelivery::Accepted { + content_hash, + received_at, + } => { + if let Err(error) = store.mark_accepted(&pending).await { + return failed_inventory(&status_tx, error); + } + backoff = retry_schedule.initial_backoff; + let _ = status_tx.send(InventoryStatus::Online { + content_hash, + received_at, + }); + schedule.cadence.saturating_add(jitter(schedule.jitter)) + } + InventoryDelivery::Retry { retry_after } => { + let delay = retry_after + .unwrap_or(backoff) + .clamp(retry_schedule.initial_backoff, retry_schedule.max_backoff); + backoff = backoff.saturating_mul(2).min(retry_schedule.max_backoff); + let _ = status_tx.send(InventoryStatus::BackingOff { delay }); + delay + } + InventoryDelivery::AuthenticationStopped { status, reason } => { + let _ = status_tx.send(InventoryStatus::AuthenticationStopped { status, reason }); + return; + } + InventoryDelivery::Rejected { status, reason } => { + let suffix = reason.map_or_else(String::new, |reason| format!("; reason={reason}")); + let _ = status_tx.send(InventoryStatus::Failed { + reason: format!("Connect rejected inventory with HTTP {status}{suffix}"), + }); + return; + } + }; + if sleep_or_cancel(&task_shutdown, delay).await { + break; + } + } + let _ = status_tx.send(InventoryStatus::Stopped); + }); + Ok(Some(InventoryRuntime { + shutdown, + status: status_rx, + task: Some(task), })) } @@ -131,6 +279,12 @@ fn failed(status: &watch::Sender, error: HeartbeatError) { }); } +fn failed_inventory(status: &watch::Sender, error: InventoryError) { + let _ = status.send(InventoryStatus::Failed { + reason: error.to_string(), + }); +} + fn jitter(maximum: Duration) -> Duration { if maximum.is_zero() { Duration::ZERO diff --git a/rustfs/src/connect/telemetry.rs b/rustfs/src/connect/telemetry.rs new file mode 100644 index 000000000..bcee54320 --- /dev/null +++ b/rustfs/src/connect/telemetry.rs @@ -0,0 +1,265 @@ +// Copyright 2024 RustFS Team +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +use std::time::Duration; + +use chrono::{DateTime, SecondsFormat, Utc}; +use reqwest::{Client, StatusCode, Url, header}; +use rustls::RootCertStore; +use rustls::pki_types::{CertificateDer, pem::PemObject as _}; +use serde::{Deserialize, Serialize}; +use zeroize::Zeroizing; + +use super::config::HeartbeatConfig; +use super::credential_store::{CredentialStoreError, DeviceCredential}; +use super::identity::IdentityError; +use super::identity_store::StoreError; +use super::registration::{CredentialValidationError, validate_stored_credential}; + +const MAX_RESPONSE_BYTES: usize = 64 * 1024; + +pub(crate) enum TelemetryDelivery { + Accepted { cluster_name: String, body: Vec }, + Retry { retry_after: Option }, + AuthenticationStopped { status: u16, reason: Option }, + Rejected { status: u16, reason: Option }, +} + +pub(crate) struct TelemetryTransport { + endpoint: Url, + root_store: RootCertStore, + roots: Vec>, + config: HeartbeatConfig, +} + +impl TelemetryTransport { + pub(crate) fn new(config: HeartbeatConfig) -> Result { + let mut endpoint = Url::parse(&config.endpoint).map_err(|_| TelemetryError::Endpoint)?; + if endpoint.scheme() != "https" + || endpoint.cannot_be_a_base() + || !endpoint.username().is_empty() + || endpoint.password().is_some() + || endpoint.query().is_some() + || endpoint.fragment().is_some() + { + return Err(TelemetryError::Endpoint); + } + if !endpoint.path().ends_with('/') { + endpoint.set_path(&format!("{}/", endpoint.path())); + } + let roots = CertificateDer::pem_slice_iter(&config.root_ca_pem) + .collect::, _>>() + .map_err(|_| TelemetryError::RootCertificate)?; + if roots.is_empty() { + return Err(TelemetryError::RootCertificate); + } + let mut root_store = RootCertStore::empty(); + let (accepted, rejected) = root_store.add_parsable_certificates(roots.clone()); + if accepted != roots.len() || rejected != 0 { + return Err(TelemetryError::RootCertificate); + } + let schedule = config.schedule; + if schedule.timeout.is_zero() + || schedule.timeout > Duration::from_secs(5) + || schedule.initial_backoff.is_zero() + || schedule.max_backoff < schedule.initial_backoff + || schedule.max_backoff > Duration::from_secs(5 * 60) + { + return Err(TelemetryError::Schedule); + } + Ok(Self { + endpoint, + root_store, + roots, + config, + }) + } + + pub(crate) async fn post(&self, collection: &str, value: &T) -> Result { + let (cluster_name, cluster_uid, client) = self.authenticated_client().await?; + let url = self.endpoint.join(&format!("clusters/{cluster_uid}/{collection}"))?; + let response = match client.post(url).json(value).send().await { + Ok(response) => response, + Err(error) if error.is_timeout() || error.is_connect() || error.is_request() => { + return Ok(TelemetryDelivery::Retry { retry_after: None }); + } + Err(error) => return Err(error.into()), + }; + let status = response.status(); + if status == StatusCode::TOO_MANY_REQUESTS { + return Ok(TelemetryDelivery::Retry { + retry_after: retry_after(response.headers(), Utc::now(), self.config.schedule.max_backoff), + }); + } + if status == StatusCode::REQUEST_TIMEOUT || status.is_server_error() { + return Ok(TelemetryDelivery::Retry { retry_after: None }); + } + if matches!(status, StatusCode::UNAUTHORIZED | StatusCode::FORBIDDEN) { + return Ok(TelemetryDelivery::AuthenticationStopped { + status: status.as_u16(), + reason: response_reason(response).await, + }); + } + if status != StatusCode::OK { + return Ok(TelemetryDelivery::Rejected { + status: status.as_u16(), + reason: response_reason(response).await, + }); + } + Ok(TelemetryDelivery::Accepted { + cluster_name, + body: bounded_body(response).await?, + }) + } + + async fn authenticated_client(&self) -> Result<(String, String, Client), TelemetryError> { + let _lock = self.config.credential_store.lock().await?; + let credential = self.config.credential_store.load()?.ok_or(TelemetryError::NotRegistered)?; + let identity = self.config.identity_store.load()?.ok_or(TelemetryError::IdentityMissing)?; + validate_stored_credential(&credential, &identity, &self.root_store, &self.roots)?; + let now = Utc::now().timestamp(); + if now < credential.not_before_unix || now >= credential.not_after_unix { + return Err(TelemetryError::CredentialExpired); + } + let (organization_uid, cluster_uid) = credential_parent(&credential)?; + let cluster_name = format!("organizations/{organization_uid}/clusters/{cluster_uid}"); + let client = self.client(&credential, &identity.to_pkcs8_pem()?)?; + Ok((cluster_name, cluster_uid.to_owned(), client)) + } + + fn client(&self, credential: &DeviceCredential, key: &Zeroizing) -> Result { + let mut pem = Zeroizing::new(Vec::with_capacity(credential.certificate_chain.len() + key.len() + 1)); + pem.extend_from_slice(credential.certificate_chain.as_bytes()); + pem.push(b'\n'); + pem.extend_from_slice(key.as_bytes()); + let identity = reqwest::Identity::from_pem(&pem).map_err(|_| TelemetryError::IdentityCertificate)?; + let roots = self + .roots + .iter() + .map(|root| reqwest::Certificate::from_der(root.as_ref())) + .collect::, _>>()?; + Client::builder() + .https_only(true) + .redirect(reqwest::redirect::Policy::none()) + .timeout(self.config.schedule.timeout) + .tls_certs_only(roots) + .identity(identity) + .build() + .map_err(Into::into) + } +} + +fn credential_parent(credential: &DeviceCredential) -> Result<(&str, &str), TelemetryError> { + let mut parts = credential.name.split('/'); + let valid = parts.next() == Some("organizations"); + let organization_uid = parts.next(); + let valid = valid && parts.next() == Some("clusters"); + let cluster_uid = parts.next(); + let valid = valid && parts.next() == Some("clusterDevices"); + let device_uid = parts.next(); + if !valid + || organization_uid.is_none_or(str::is_empty) + || cluster_uid.is_none_or(str::is_empty) + || device_uid != Some(credential.uid.as_str()) + || parts.next().is_some() + { + return Err(TelemetryError::CredentialName); + } + Ok(( + organization_uid.ok_or(TelemetryError::CredentialName)?, + cluster_uid.ok_or(TelemetryError::CredentialName)?, + )) +} + +fn retry_after(headers: &header::HeaderMap, now: DateTime, maximum: Duration) -> Option { + let value = headers.get(header::RETRY_AFTER)?.to_str().ok()?; + let delay = value.parse::().ok().map(Duration::from_secs).or_else(|| { + DateTime::parse_from_rfc2822(value) + .ok() + .and_then(|at| (at.with_timezone(&Utc) - now).to_std().ok()) + })?; + Some(delay.min(maximum)) +} + +pub(crate) fn is_exact_utc_seconds(value: &str) -> bool { + DateTime::parse_from_rfc3339(value).is_ok_and(|time| { + time.offset().local_minus_utc() == 0 + && value.ends_with('Z') + && time.with_timezone(&Utc).to_rfc3339_opts(SecondsFormat::Secs, true) == value + }) +} + +async fn response_reason(response: reqwest::Response) -> Option { + #[derive(Deserialize)] + struct Envelope { + #[serde(default)] + details: Vec, + } + #[derive(Deserialize)] + struct Detail { + #[serde(default)] + reason: String, + } + + serde_json::from_slice::(&bounded_body(response).await.ok()?) + .ok()? + .details + .into_iter() + .find_map(|detail| (!detail.reason.is_empty()).then_some(detail.reason)) +} + +async fn bounded_body(mut response: reqwest::Response) -> Result, TelemetryError> { + let mut body = Vec::new(); + while let Some(chunk) = response.chunk().await? { + if body.len().saturating_add(chunk.len()) > MAX_RESPONSE_BYTES { + return Err(TelemetryError::ResponseTooLarge); + } + body.extend_from_slice(&chunk); + } + Ok(body) +} + +#[derive(Debug, thiserror::Error)] +pub(crate) enum TelemetryError { + #[error("Connect telemetry endpoint must be an HTTPS base URL without credentials, query, or fragment")] + Endpoint, + #[error("Connect telemetry root CA configuration is invalid")] + RootCertificate, + #[error("Connect telemetry retry schedule is invalid")] + Schedule, + #[error("RustFS is not registered with Connect")] + NotRegistered, + #[error("the Connect device private key is missing")] + IdentityMissing, + #[error("the stored Connect certificate and device private key cannot form a TLS identity")] + IdentityCertificate, + #[error("the stored Connect credential name is invalid")] + CredentialName, + #[error("the stored Connect device certificate is not currently valid")] + CredentialExpired, + #[error("Connect telemetry response exceeded 64 KiB")] + ResponseTooLarge, + #[error(transparent)] + Url(#[from] url::ParseError), + #[error(transparent)] + Transport(#[from] reqwest::Error), + #[error(transparent)] + Identity(#[from] IdentityError), + #[error(transparent)] + IdentityStore(#[from] StoreError), + #[error(transparent)] + CredentialStore(#[from] CredentialStoreError), + #[error(transparent)] + CredentialValidation(#[from] CredentialValidationError), +} diff --git a/rustfs/src/startup_services.rs b/rustfs/src/startup_services.rs index c07db0f67..51ecc0b2a 100644 --- a/rustfs/src/startup_services.rs +++ b/rustfs/src/startup_services.rs @@ -13,10 +13,13 @@ // limitations under the License. use crate::site_replication_reconcile::spawn_site_replication_reconcile_task; -use crate::storage_api::startup::services::{ECStore, EndpointServerPools, ServerContextSlot}; +use crate::storage_api::startup::services::{ECStore, EndpointServerPools, ServerContextSlot, StorageAdminApi}; use crate::{ config::Config, - connect::{CoarseNodeSummary, HeartbeatConfig, HeartbeatRuntime, spawn_heartbeat_runtime}, + connect::{ + CoarseNodeSummary, HeartbeatConfig, HeartbeatRuntime, InventoryFlag, InventoryRuntime, InventorySchedule, + InventorySnapshot, spawn_heartbeat_runtime, spawn_inventory_runtime, + }, init::{init_buffer_profile_system, init_kms_system}, server::ServiceStateManager, startup_audit::init_audit_runtime, @@ -96,7 +99,9 @@ pub(crate) async fn init_startup_runtime_services( init_notification_runtime(endpoint_pools, buckets).await?; let enable_scanner = init_background_service_runtime(store.clone()).await?; init_observability_runtime(store.clone(), ctx.clone()).await; - let heartbeat = start_heartbeat_runtime(heartbeat_config, heartbeat_nodes, &ctx)?; + let heartbeat = start_heartbeat_runtime(heartbeat_config.clone(), heartbeat_nodes, &ctx)?; + let inventory = start_inventory_runtime(heartbeat_config, heartbeat_nodes, store, &ctx)?; + let heartbeat = heartbeat.map(|heartbeat| heartbeat.with_inventory(inventory)); Ok(StartupServiceRuntime { optional_runtimes, @@ -120,3 +125,32 @@ fn start_heartbeat_runtime( .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) } + +fn start_inventory_runtime( + config: Option, + node_count: Option, + store: Arc, + shutdown: &CancellationToken, +) -> Result> { + let Some(config) = config else { + return Ok(None); + }; + let node_count = node_count.unwrap_or_default(); + spawn_inventory_runtime(Some(config), InventorySchedule::default(), shutdown, move || { + let store = store.clone(); + async move { + let info = StorageAdminApi::storage_info(store.as_ref()).await; + let total = crate::app::storage_api::capacity::get_total_usable_capacity(&info.disks, &info) as u64; + let free = crate::app::storage_api::capacity::get_total_usable_capacity_free(&info.disks, &info) as u64; + let mut flags = Vec::with_capacity(3); + if info.disks.iter().any(|disk| disk.state == rustfs_madmin::ITEM_OFFLINE) { + flags.extend([InventoryFlag::ClusterDegraded, InventoryFlag::DriveOffline]); + } + if info.disks.iter().any(|disk| disk.healing) { + flags.push(InventoryFlag::ClusterHealing); + } + InventorySnapshot::current(node_count, info.disks.len(), total, free, flags) + } + }) + .map_err(std::io::Error::other) +} diff --git a/rustfs/src/storage_api.rs b/rustfs/src/storage_api.rs index dea2923ac..8b1ff9f23 100644 --- a/rustfs/src/storage_api.rs +++ b/rustfs/src/storage_api.rs @@ -279,6 +279,7 @@ pub(crate) mod startup { } pub(crate) mod services { + pub(crate) use super::super::storage_contracts::StorageAdminApi; pub(crate) use crate::storage::storage_api::{ECStore, EndpointServerPools, ServerContextSlot}; } diff --git a/rustfs/tests/connect_inventory.rs b/rustfs/tests/connect_inventory.rs new file mode 100644 index 000000000..494a2f005 --- /dev/null +++ b/rustfs/tests/connect_inventory.rs @@ -0,0 +1,565 @@ +// Copyright 2024 RustFS Team +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +use std::collections::VecDeque; +use std::fs; +use std::sync::atomic::{AtomicUsize, Ordering}; +use std::sync::{Arc, Mutex}; +use std::time::Duration; + +use bytes::Bytes; +use http_body_util::{BodyExt as _, Full}; +use hyper::service::service_fn; +use hyper::{Request, Response, StatusCode}; +use hyper_util::rt::TokioIo; +use rcgen::{ + BasicConstraints, CertificateParams, DistinguishedName, DnType, ExtendedKeyUsagePurpose, IsCa, Issuer, KeyPair, + KeyUsagePurpose, SanType, +}; +use rustfs::connect::{ + CredentialStore, DeviceCredential, HeartbeatConfig, HeartbeatSchedule, IdentityStore, InventoryFlag, InventoryOsVersion, + InventorySchedule, InventorySnapshot, InventoryStatus, OperatingSystemFamily, spawn_inventory_runtime, +}; +use rustls::RootCertStore; +use rustls::pki_types::{CertificateDer, PrivateKeyDer, PrivatePkcs8KeyDer}; +use rustls::server::WebPkiClientVerifier; +use serde_json::{Value, json}; +use time::OffsetDateTime; +use tokio::net::TcpListener; +use tokio::sync::watch; +use tokio_rustls::TlsAcceptor; +use tokio_util::sync::CancellationToken; + +const ORGANIZATION_UID: &str = "0198f4b0-1a00-7c10-8d21-2e3f4a5b6c70"; +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"; + +struct TestPki { + root_params: CertificateParams, + root_key: KeyPair, + root_der: CertificateDer<'static>, + root_pem: String, + server_der: CertificateDer<'static>, + server_key: PrivatePkcs8KeyDer<'static>, +} + +impl TestPki { + fn new() -> Self { + let now = OffsetDateTime::now_utc(); + let root_key = KeyPair::generate().expect("generate root key"); + let mut root_params = CertificateParams::default(); + root_params.is_ca = IsCa::Ca(BasicConstraints::Unconstrained); + root_params.not_before = now - time::Duration::days(30); + root_params.not_after = now + time::Duration::days(30); + root_params.key_usages = vec![KeyUsagePurpose::KeyCertSign, KeyUsagePurpose::DigitalSignature]; + let root = root_params.self_signed(&root_key).expect("sign root"); + + let server_key = KeyPair::generate().expect("generate server key"); + let mut server_params = CertificateParams::default(); + server_params.not_before = now - time::Duration::hours(1); + server_params.not_after = now + time::Duration::days(2); + server_params + .subject_alt_names + .push(SanType::DnsName("localhost".try_into().expect("valid DNS name"))); + server_params.extended_key_usages = vec![ExtendedKeyUsagePurpose::ServerAuth]; + let server = server_params + .signed_by(&server_key, &Issuer::from_params(&root_params, &root_key)) + .expect("sign server certificate"); + Self { + root_params, + root_key, + root_der: root.der().clone(), + root_pem: root.pem(), + server_der: server.der().clone(), + server_key: PrivatePkcs8KeyDer::from(server_key.serialize_der()), + } + } + + fn server_config(&self) -> rustls::ServerConfig { + let mut roots = RootCertStore::empty(); + roots.add(self.root_der.clone()).expect("add client root"); + let verifier = WebPkiClientVerifier::builder(Arc::new(roots)) + .build() + .expect("client verifier"); + rustls::ServerConfig::builder() + .with_client_cert_verifier(verifier) + .with_single_cert(vec![self.server_der.clone()], PrivateKeyDer::Pkcs8(self.server_key.clone_key())) + .expect("server TLS") + } + + fn stores(&self, temp: &tempfile::TempDir) -> (IdentityStore, CredentialStore) { + let identity_store = IdentityStore::new(temp.path().join("identity")); + let identity = identity_store.load_or_create().expect("create identity"); + let private_key = PrivatePkcs8KeyDer::from(identity.to_pkcs8_der().expect("serialize key").to_vec()); + let device_key = KeyPair::from_pkcs8_der_and_sign_algo(&private_key, &rcgen::PKCS_ECDSA_P256_SHA256).expect("device key"); + let now = OffsetDateTime::now_utc(); + let mut params = CertificateParams::default(); + params.not_before = now - time::Duration::hours(1); + params.not_after = now + time::Duration::hours(23); + params.serial_number = Some(vec![1; 16].into()); + params.key_usages = vec![KeyUsagePurpose::DigitalSignature]; + params.extended_key_usages = vec![ExtendedKeyUsagePurpose::ClientAuth]; + params.distinguished_name = DistinguishedName::new(); + params.distinguished_name.push(DnType::CommonName, DEVICE_UID); + params.subject_alt_names.push(SanType::URI( + format!("urn:rustfs:connect:device:{DEVICE_UID}") + .try_into() + .expect("device URI"), + )); + let certificate = params + .signed_by(&device_key, &Issuer::from_params(&self.root_params, &self.root_key)) + .expect("device certificate"); + let cluster = format!("organizations/{ORGANIZATION_UID}/clusters/{CLUSTER_UID}"); + let credential = DeviceCredential { + name: format!("{cluster}/clusterDevices/{DEVICE_UID}"), + uid: DEVICE_UID.to_owned(), + protocol_version: "v1".to_owned(), + key_id: format!("x509-{}", "01".repeat(16)), + certificate_serial: "01".repeat(16), + certificate: certificate.pem(), + certificate_chain: certificate.pem(), + not_before_unix: (now - time::Duration::hours(1)).unix_timestamp(), + not_after_unix: (now + time::Duration::hours(23)).unix_timestamp(), + }; + let directory = temp.path().join("credential"); + fs::create_dir_all(&directory).expect("credential directory"); + let path = directory.join("device.crt.json"); + fs::write(&path, serde_json::to_vec(&credential).expect("credential JSON")).expect("write credential"); + private_mode(&path); + (identity_store, CredentialStore::new(directory)) + } +} + +#[derive(Clone)] +struct Reply { + status: StatusCode, + body: Value, + retry_after: Option<&'static str>, +} + +impl Reply { + fn ok(content_hash: &str) -> Self { + Self { + status: StatusCode::OK, + body: json!({ + "name": format!("organizations/{ORGANIZATION_UID}/clusters/{CLUSTER_UID}/inventorySnapshots/{SNAPSHOT_UID}"), + "uid": SNAPSHOT_UID, + "contentHash": content_hash, + "receivedAt": "2026-08-22T01:02:03Z", + "futureField": true + }), + retry_after: None, + } + } + + fn error(status: StatusCode, reason: &str) -> Self { + Self { + status, + body: json!({"details": [{"reason": reason}]}), + retry_after: None, + } + } +} + +struct TestServer { + endpoint: String, + seen: Arc>>, + task: tokio::task::JoinHandle<()>, +} + +impl Drop for TestServer { + fn drop(&mut self) { + self.task.abort(); + } +} + +async fn server(pki: &TestPki, replies: Vec) -> TestServer { + let listener = TcpListener::bind("127.0.0.1:0").await.expect("bind server"); + let address = listener.local_addr().expect("server address"); + let acceptor = TlsAcceptor::from(Arc::new(pki.server_config())); + let replies = Arc::new(Mutex::new(VecDeque::from(replies))); + let seen = Arc::new(Mutex::new(Vec::new())); + let captured = seen.clone(); + let task = tokio::spawn(async move { + while let Ok((stream, _)) = listener.accept().await { + let acceptor = acceptor.clone(); + let replies = replies.clone(); + let seen = captured.clone(); + tokio::spawn(async move { + let Ok(stream) = acceptor.accept(stream).await else { return }; + let service = service_fn(move |request: Request| { + let replies = replies.clone(); + let seen = seen.clone(); + async move { + assert_eq!(request.uri().path(), format!("/agent/clusters/{CLUSTER_UID}/inventorySnapshots")); + let body = request.into_body().collect().await.expect("request body").to_bytes(); + seen.lock() + .expect("seen lock") + .push(serde_json::from_slice(&body).expect("request JSON")); + let reply = replies + .lock() + .expect("reply lock") + .pop_front() + .unwrap_or_else(|| Reply::error(StatusCode::SERVICE_UNAVAILABLE, "UNAVAILABLE")); + let mut builder = Response::builder() + .status(reply.status) + .header("content-type", "application/json"); + if let Some(value) = reply.retry_after { + builder = builder.header("retry-after", value); + } + Ok::<_, hyper::Error>( + builder + .body(Full::new(Bytes::from(serde_json::to_vec(&reply.body).expect("reply JSON")))) + .expect("reply"), + ) + } + }); + let _ = hyper::server::conn::http1::Builder::new() + .serve_connection(TokioIo::new(stream), service) + .await; + }); + } + }); + TestServer { + endpoint: format!("https://localhost:{}/agent/", address.port()), + seen, + task, + } +} + +fn config(temp: &tempfile::TempDir, pki: &TestPki, server: &TestServer) -> HeartbeatConfig { + let (identity_store, credential_store) = pki.stores(temp); + HeartbeatConfig { + endpoint: server.endpoint.clone(), + root_ca_pem: pki.root_pem.as_bytes().to_vec(), + identity_store, + credential_store, + state_path: temp.path().join("private-config-secret/heartbeat/state.json"), + schedule: HeartbeatSchedule { + cadence: Duration::from_secs(30), + jitter: Duration::ZERO, + timeout: Duration::from_millis(200), + initial_backoff: Duration::from_millis(20), + max_backoff: Duration::from_millis(80), + }, + } +} + +fn schedule() -> InventorySchedule { + InventorySchedule { + cadence: Duration::from_secs(60), + jitter: Duration::ZERO, + } +} + +fn snapshot() -> InventorySnapshot { + InventorySnapshot::new( + "1.4.2", + Some(InventoryOsVersion::new(OperatingSystemFamily::Linux, 6, 8).expect("valid operating-system version")), + 8, + 96, + 1_099_511_627_776, + 412_316_860_416, + [InventoryFlag::ClusterDegraded, InventoryFlag::DriveOffline], + ) + .expect("valid inventory") +} + +fn collect_strings(value: &Value, strings: &mut Vec) { + match value { + Value::String(value) => strings.push(value.clone()), + Value::Array(values) => values.iter().for_each(|value| collect_strings(value, strings)), + Value::Object(values) => values.values().for_each(|value| collect_strings(value, strings)), + _ => {} + } +} + +async fn wait_for( + status: &mut watch::Receiver, + predicate: impl Fn(&InventoryStatus) -> bool, +) -> InventoryStatus { + tokio::time::timeout(Duration::from_secs(3), async { + loop { + let current = status.borrow_and_update().clone(); + if predicate(¤t) { + return current; + } + status.changed().await.expect("status channel"); + } + }) + .await + .expect("inventory status timeout") +} + +#[test] +fn connect_inventory_frozen_vector_has_the_exact_canonical_hash_and_no_open_ended_fields() { + let fixtures: Value = serde_json::from_str(include_str!("../../protocol/agent/v1/fixtures/inventory/valid-vectors.json")) + .expect("valid fixture JSON"); + let expected = &fixtures["vectors"][0]["expected"]; + let snapshot = snapshot(); + + assert_eq!(snapshot.content_hash().expect("content hash"), expected["contentHash"]); + assert_eq!(InventorySchedule::default().cadence, Duration::from_secs(6 * 60 * 60)); + assert_eq!(InventorySchedule::default().jitter, Duration::from_secs(30 * 60)); + let encoded = serde_json::to_value(snapshot).expect("snapshot JSON"); + assert_eq!( + encoded, + json!({ + "rustfsVersion": "1.4.2", + "osVersion": {"family": "linux", "major": 6, "minor": 8}, + "nodeCount": 8, + "driveCount": 96, + "capacityTotalBytes": 1099511627776_u64, + "capacityUsedBytes": 412316860416_u64, + "coarseFlags": ["cluster.degraded", "drive.offline"] + }) + ); + + let fixtures: Value = + serde_json::from_str(include_str!("../../protocol/agent/v1/fixtures/inventory/secret-like-vectors.json")) + .expect("valid secret-like fixture JSON"); + let known_fields = [ + "protocolVersion", + "rustfsVersion", + "osVersion", + "nodeCount", + "driveCount", + "capacityTotalBytes", + "capacityUsedBytes", + "coarseFlags", + ]; + let known_flags = ["cluster.degraded", "drive.offline"]; + let mut excluded = Vec::new(); + for vector in fixtures["vectors"].as_array().expect("fixture vectors") { + let input = vector["input"].as_object().expect("fixture input"); + for (name, value) in input { + if !known_fields.contains(&name.as_str()) { + collect_strings(value, &mut excluded); + } + } + for (name, value) in input["osVersion"].as_object().expect("fixture OS version") { + if !["family", "major", "minor"].contains(&name.as_str()) { + collect_strings(value, &mut excluded); + } + } + for flag in input["coarseFlags"].as_array().expect("fixture coarse flags") { + let flag = flag.as_str().expect("fixture coarse flag"); + if !known_flags.contains(&flag) { + excluded.push(flag.to_owned()); + } + } + } + let encoded = serde_json::to_string(&encoded).expect("encoded snapshot"); + for value in excluded { + assert!(!encoded.contains(&value), "snapshot exposed fixture value {value}"); + } +} + +#[test] +fn connect_inventory_bounds_fail_instead_of_truncating_or_inventing_values() { + assert!(matches!( + InventorySnapshot::current(0, 0, 0, 0, []), + Err(rustfs::connect::InventoryError::NodeCount) + )); + assert!(matches!( + InventorySnapshot::current(1, 1_048_577, 0, 0, []), + Err(rustfs::connect::InventoryError::DriveCount) + )); + assert!(matches!( + InventorySnapshot::current(1, 0, 9_007_199_254_740_992, 0, []), + Err(rustfs::connect::InventoryError::Capacity) + )); + assert!(matches!( + InventorySnapshot::current(1, 0, 10, 11, []), + Err(rustfs::connect::InventoryError::Capacity) + )); + assert!(matches!( + InventorySnapshot::new("1.0.0-private.1", None, 1, 0, 0, 0, []), + Err(rustfs::connect::InventoryError::RustfsVersion) + )); +} + +#[tokio::test] +async fn connect_inventory_restart_replays_the_pending_request_and_then_skips_unchanged_inventory() { + 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 shutdown = CancellationToken::new(); + let samples = Arc::new(AtomicUsize::new(0)); + let sampled = samples.clone(); + let runtime = spawn_inventory_runtime(Some(config(&temp, &pki, &first_server)), schedule(), &shutdown, move || { + sampled.fetch_add(1, Ordering::Relaxed); + std::future::ready(Ok(snapshot())) + }) + .expect("start inventory") + .expect("configured inventory"); + let mut status = runtime.status(); + + assert!(matches!( + wait_for(&mut status, |status| matches!(status, InventoryStatus::BackingOff { .. })).await, + InventoryStatus::BackingOff { delay } if delay == Duration::from_millis(20) + )); + assert_eq!(samples.load(Ordering::Relaxed), 1); + let original = first_server.seen.lock().expect("seen lock")[0].clone(); + runtime.shutdown().await; + + let mut limited = Reply::error(StatusCode::TOO_MANY_REQUESTS, "RATE_LIMITED"); + limited.retry_after = Some("0"); + let restart_server = server(&pki, vec![limited, Reply::ok(&content_hash)]).await; + let restart_config = config(&temp, &pki, &restart_server); + let restart_samples = Arc::new(AtomicUsize::new(0)); + let sampled = restart_samples.clone(); + let restart = spawn_inventory_runtime(Some(restart_config.clone()), schedule(), &shutdown, move || { + sampled.fetch_add(1, Ordering::Relaxed); + std::future::ready(Ok(snapshot())) + }) + .expect("restart inventory") + .expect("configured inventory"); + let mut status = restart.status(); + + assert!(matches!( + wait_for(&mut status, |status| matches!(status, InventoryStatus::Online { .. })).await, + InventoryStatus::Online { content_hash: accepted, received_at } + if accepted == content_hash && received_at == "2026-08-22T01:02:03Z" + )); + assert_eq!(restart_samples.load(Ordering::Relaxed), 0); + 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"); + for forbidden in [ + "private-config-secret", + "BEGIN CERTIFICATE", + "AKIAIOSFODNN7EXAMPLE", + "bucket", + "object", + "path", + ] { + assert!(!encoded.contains(forbidden), "request exposed {forbidden}"); + } + assert_eq!(original.as_object().expect("request object").len(), 10); + restart.shutdown().await; + + let unchanged_samples = Arc::new(AtomicUsize::new(0)); + let sampled = unchanged_samples.clone(); + let unchanged = spawn_inventory_runtime(Some(restart_config), schedule(), &shutdown, move || { + sampled.fetch_add(1, Ordering::Relaxed); + std::future::ready(Ok(snapshot())) + }) + .expect("restart inventory") + .expect("configured inventory"); + let mut unchanged_status = unchanged.status(); + assert!(matches!( + wait_for(&mut unchanged_status, |status| matches!(status, InventoryStatus::Unchanged { .. })).await, + InventoryStatus::Unchanged { content_hash: unchanged } if unchanged == content_hash + )); + assert_eq!(unchanged_samples.load(Ordering::Relaxed), 1); + assert_eq!(restart_server.seen.lock().expect("seen lock").len(), 2); + unchanged.shutdown().await; +} + +#[tokio::test] +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 config = config(&temp, &pki, &unavailable); + drop(unavailable); + let shutdown = CancellationToken::new(); + let samples = Arc::new(AtomicUsize::new(0)); + let sampled = samples.clone(); + let runtime = spawn_inventory_runtime(Some(config), schedule(), &shutdown, move || { + sampled.fetch_add(1, Ordering::Relaxed); + std::future::ready(Ok(snapshot())) + }) + .expect("start inventory") + .expect("configured inventory"); + let mut status = runtime.status(); + + assert!(matches!( + wait_for(&mut status, |status| { + matches!(status, InventoryStatus::BackingOff { delay } if *delay == Duration::from_millis(40)) + }) + .await, + InventoryStatus::BackingOff { delay } if delay == Duration::from_millis(40) + )); + assert_eq!(samples.load(Ordering::Relaxed), 1); + runtime.shutdown().await; +} + +#[tokio::test] +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 shutdown = CancellationToken::new(); + let runtime = spawn_inventory_runtime(Some(config(&temp, &pki, &server)), schedule(), &shutdown, || { + std::future::ready(Ok(snapshot())) + }) + .expect("start inventory") + .expect("configured inventory"); + let mut status = runtime.status(); + + assert!(matches!( + wait_for(&mut status, |status| matches!(status, InventoryStatus::AuthenticationStopped { .. })).await, + InventoryStatus::AuthenticationStopped { status: 401, reason: Some(reason) } if reason == "DEVICE_REVOKED" + )); + assert_eq!(server.seen.lock().expect("seen lock").len(), 1); + runtime.shutdown().await; +} + +#[tokio::test] +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 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"); + fs::write( + &state, + br#"{"nextSequence":9007199254740992,"pending":null,"lastAcceptedContentHash":null}"#, + ) + .expect("write state"); + private_mode(&state); + let samples = Arc::new(AtomicUsize::new(0)); + let sampled = samples.clone(); + let shutdown = CancellationToken::new(); + let runtime = spawn_inventory_runtime(Some(config), schedule(), &shutdown, move || { + sampled.fetch_add(1, Ordering::Relaxed); + std::future::ready(Ok(snapshot())) + }) + .expect("start inventory") + .expect("configured inventory"); + let mut status = runtime.status(); + + assert!(matches!( + wait_for(&mut status, |status| matches!(status, InventoryStatus::Failed { .. })).await, + InventoryStatus::Failed { reason } if reason.contains("sequence is exhausted") + )); + assert_eq!(samples.load(Ordering::Relaxed), 0); + assert!(server.seen.lock().expect("seen lock").is_empty()); + runtime.shutdown().await; +} + +#[cfg(unix)] +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))] +fn private_mode(_path: &std::path::Path) {}