From 2788ef7229d0d39cb2aae050ab70f60c204bcb76 Mon Sep 17 00:00:00 2001 From: Zhengchao An Date: Sun, 23 Aug 2026 21:32:43 +0800 Subject: [PATCH] feat(connect): collect bounded offline diagnostics (#6450) * feat(connect): collect bounded offline diagnostics * fix(connect): tighten offline diagnostic boundaries --- .../v1/fixtures/redaction/MANIFEST.sha256 | 2 +- .../v1/fixtures/redaction/secret-vectors.json | 20 + rustfs/src/connect/offline/collectors.rs | 417 ++++++++++++++++++ rustfs/src/connect/offline/manifest_entry.rs | 74 ++++ rustfs/src/connect/offline/mod.rs | 6 + rustfs/src/connect/offline/redaction.rs | 362 +++++++++++++++ rustfs/tests/connect_offline_collectors.rs | 290 ++++++++++++ 7 files changed, 1170 insertions(+), 1 deletion(-) create mode 100644 rustfs/src/connect/offline/collectors.rs create mode 100644 rustfs/src/connect/offline/manifest_entry.rs create mode 100644 rustfs/src/connect/offline/redaction.rs create mode 100644 rustfs/tests/connect_offline_collectors.rs diff --git a/protocol/agent/v1/fixtures/redaction/MANIFEST.sha256 b/protocol/agent/v1/fixtures/redaction/MANIFEST.sha256 index d3c9cafca..99fbcf2b0 100644 --- a/protocol/agent/v1/fixtures/redaction/MANIFEST.sha256 +++ b/protocol/agent/v1/fixtures/redaction/MANIFEST.sha256 @@ -1,4 +1,4 @@ 38e793226476bdb5f74c704c23ccc0e9ec09d51be3b51791e8b2cdfbf27a5c02 allowed-vectors.json 014b06540e664f6e38f0174746a418069d2677a2a1a4ef96c17f16dec7e47886 rejection-vectors.json fdf1d8f4c7ed6f96026e86c7d89f56e5e08269c0b7a49a1e3c3880e5d023c600 ruleset.json -5e349d5121037a09a9b1009b08feac9f5fb4b14cd6548419b88bd9f032929b55 secret-vectors.json +7a4d297943e84b4cea457677762cf3d0d5c2cba35a4aac496befa30ee522b3c0 secret-vectors.json diff --git a/protocol/agent/v1/fixtures/redaction/secret-vectors.json b/protocol/agent/v1/fixtures/redaction/secret-vectors.json index ad467dc5f..f7f1a525c 100644 --- a/protocol/agent/v1/fixtures/redaction/secret-vectors.json +++ b/protocol/agent/v1/fixtures/redaction/secret-vectors.json @@ -28,6 +28,26 @@ }, "expectedRedactedCount": 1 }, + { + "name": "an S3 access key id adjacent to a Unicode letter", + "source": "offline-diagnostic", + "valueRule": "AWS_ACCESS_KEY_ID", + "ruleSubject": "éAKIAIOSFODNN7EXAMPLE", + "document": { + "osSummary": "éAKIAIOSFODNN7EXAMPLE", + "rustfsVersion": "1.19.4" + }, + "secretLiterals": [ + "AKIAIOSFODNN7EXAMPLE" + ], + "expectedCanonicalJson": "{\"osSummary\":\"[REDACTED]\",\"rustfsVersion\":\"1.19.4\"}", + "expectedCounts": { + "droppedField": 0, + "redactedValue": 1, + "redactedOversizeValue": 0 + }, + "expectedRedactedCount": 1 + }, { "name": "an S3 secret access key standing alone in an offline OS summary", "source": "offline-diagnostic", diff --git a/rustfs/src/connect/offline/collectors.rs b/rustfs/src/connect/offline/collectors.rs new file mode 100644 index 000000000..336c455ac --- /dev/null +++ b/rustfs/src/connect/offline/collectors.rs @@ -0,0 +1,417 @@ +// 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. + +//! Fixed Q07 L0/L1 collectors for an operator-triggered offline diagnostic. + +use std::collections::BTreeSet; +use std::sync::{Arc, LazyLock}; +use std::time::Duration; + +use rustfs_madmin::{ITEM_OFFLINE, StorageInfo}; +use serde::Serialize; +use serde_json::{Value, json}; +use sysinfo::{Disks, Networks, RefreshKind, System}; +use thiserror::Error; +use tokio::sync::Semaphore; +use tokio_util::sync::CancellationToken; +use url::Url; + +use super::manifest_entry::ManifestEntry; +use super::redaction::RedactionError; + +const COLLECT_TIMEOUT: Duration = Duration::from_secs(2); +const MAX_ENTRY_BYTES: usize = 16 * 1024; +const MAX_DRIVES: usize = 4_096; +static SYSTEM_SCAN_PERMIT: LazyLock> = LazyLock::new(|| Arc::new(Semaphore::new(1))); + +#[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize)] +pub enum DataClassification { + L0, + L1, +} + +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +pub enum OfflineCollector { + RustfsVersion, + NodeCount, + DriveCount, + CapacityUsedBytes, + CapacityTotalBytes, + CoarseHealthFlags, + OsSummary, + KernelSummary, + CpuSummary, + MemorySummary, + FilesystemSummary, + NetworkSummary, +} + +const COLLECTORS: [OfflineCollector; 12] = [ + OfflineCollector::RustfsVersion, + OfflineCollector::NodeCount, + OfflineCollector::DriveCount, + OfflineCollector::CapacityUsedBytes, + OfflineCollector::CapacityTotalBytes, + OfflineCollector::CoarseHealthFlags, + OfflineCollector::OsSummary, + OfflineCollector::KernelSummary, + OfflineCollector::CpuSummary, + OfflineCollector::MemorySummary, + OfflineCollector::FilesystemSummary, + OfflineCollector::NetworkSummary, +]; + +impl OfflineCollector { + pub const fn field_id(self) -> &'static str { + match self { + Self::RustfsVersion => "offline.rustfsVersion", + Self::NodeCount => "offline.nodeCount", + Self::DriveCount => "offline.driveCount", + Self::CapacityUsedBytes => "offline.capacityUsedBytes", + Self::CapacityTotalBytes => "offline.capacityTotalBytes", + Self::CoarseHealthFlags => "offline.coarseHealthFlags", + Self::OsSummary => "offline.osSummary", + Self::KernelSummary => "offline.kernelSummary", + Self::CpuSummary => "offline.cpuSummary", + Self::MemorySummary => "offline.memorySummary", + Self::FilesystemSummary => "offline.filesystemSummary", + Self::NetworkSummary => "offline.networkSummary", + } + } + + pub const fn classification(self) -> DataClassification { + match self { + Self::RustfsVersion + | Self::NodeCount + | Self::DriveCount + | Self::CapacityUsedBytes + | Self::CapacityTotalBytes + | Self::CoarseHealthFlags => DataClassification::L0, + Self::OsSummary + | Self::KernelSummary + | Self::CpuSummary + | Self::MemorySummary + | Self::FilesystemSummary + | Self::NetworkSummary => DataClassification::L1, + } + } + + pub const fn max_entry_bytes(self) -> usize { + MAX_ENTRY_BYTES + } + + pub(crate) const fn timeout(self) -> Duration { + COLLECT_TIMEOUT + } + + pub(crate) fn field_name(self) -> &'static str { + self.field_id().split_once('.').expect("collector field ids are frozen").1 + } + + fn value(self, storage: &StorageSnapshot, system: &SystemSnapshot) -> Value { + match self { + Self::RustfsVersion => json!(env!("CARGO_PKG_VERSION")), + Self::NodeCount => json!(storage.node_count), + Self::DriveCount => json!(storage.drive_count), + Self::CapacityUsedBytes => json!(storage.capacity_used_bytes), + Self::CapacityTotalBytes => json!(storage.capacity_total_bytes), + Self::CoarseHealthFlags => json!({ + "degraded": storage.degraded, + "healing": storage.healing, + "offlineDrives": storage.offline_drives, + "scanning": storage.scanning, + }), + Self::OsSummary => json!(system.os_summary), + Self::KernelSummary => json!(system.kernel_summary), + Self::CpuSummary => json!({ "architecture": system.architecture, "cores": system.cores }), + Self::MemorySummary => json!({ + "totalBytes": system.total_memory_bytes, + "underPressure": system.under_memory_pressure, + }), + Self::FilesystemSummary => json!(system.filesystem_types), + Self::NetworkSummary => json!({ + "bondCount": system.bond_count, + "interfaceCount": system.interface_count, + }), + } + } +} + +#[derive(Debug, Error)] +pub enum CollectorError { + #[error("offline diagnostic collection was cancelled")] + Cancelled, + #[error("offline diagnostic collection exceeded its 2 second budget")] + TimedOut, + #[error("offline diagnostic collector task failed")] + TaskFailed, + #[error("offline diagnostic storage topology exceeds its 4096 drive budget")] + StorageTopologyTooLarge, + #[error("offline diagnostic storage topology contains an invalid endpoint")] + InvalidStorageEndpoint, + #[error("offline diagnostic field {field_id} exceeds its {limit} byte entry budget")] + EntryTooLarge { field_id: &'static str, limit: usize }, + #[error("offline diagnostic entry is not representable as JSON")] + NotRepresentable, + #[error(transparent)] + Redaction(#[from] RedactionError), +} + +#[derive(Debug)] +struct StorageSnapshot { + node_count: usize, + drive_count: usize, + capacity_used_bytes: u64, + capacity_total_bytes: u64, + offline_drives: usize, + degraded: bool, + healing: bool, + scanning: bool, +} + +impl TryFrom<&StorageInfo> for StorageSnapshot { + type Error = CollectorError; + + fn try_from(info: &StorageInfo) -> Result { + if info.disks.len() > MAX_DRIVES { + return Err(CollectorError::StorageTopologyTooLarge); + } + let node_count = info + .disks + .iter() + .map(|disk| { + let endpoint = Url::parse(&disk.endpoint).map_err(|_| CollectorError::InvalidStorageEndpoint)?; + let host = endpoint.host_str().ok_or(CollectorError::InvalidStorageEndpoint)?; + Ok((host.to_owned(), endpoint.port_or_known_default())) + }) + .collect::, CollectorError>>()? + .len(); + let offline_drives = info.disks.iter().filter(|disk| disk.state == ITEM_OFFLINE).count(); + Ok(Self { + node_count, + drive_count: info.disks.len(), + capacity_used_bytes: info + .disks + .iter() + .fold(0_u64, |total, disk| total.saturating_add(disk.used_space)), + capacity_total_bytes: info + .disks + .iter() + .fold(0_u64, |total, disk| total.saturating_add(disk.total_space)), + offline_drives, + degraded: info + .disks + .iter() + .any(|disk| !matches!(disk.state.as_str(), "ok" | "unformatted" | rustfs_madmin::ITEM_ONLINE)), + healing: info.disks.iter().any(|disk| disk.healing), + scanning: info.disks.iter().any(|disk| disk.scanning), + }) + } +} + +#[derive(Debug)] +struct SystemSnapshot { + os_summary: String, + kernel_summary: String, + architecture: &'static str, + cores: usize, + total_memory_bytes: u64, + under_memory_pressure: bool, + filesystem_types: Vec, + interface_count: usize, + bond_count: usize, +} + +impl SystemSnapshot { + fn collect() -> Self { + #[cfg(test)] + let _scan = test_support::ScanGuard::start(); + + // Do not enumerate processes: Q07 allows only CPU and memory summaries. + let system = System::new_with_specifics(RefreshKind::everything().without_processes()); + let total_memory_bytes = system.total_memory(); + let available_memory = system.available_memory(); + let filesystem_types = Disks::new_with_refreshed_list() + .iter() + .map(|disk| disk.file_system().to_string_lossy().into_owned()) + .collect::>() + .into_iter() + .collect(); + let networks = Networks::new_with_refreshed_list(); + Self { + os_summary: System::long_os_version().unwrap_or_else(|| "unknown".to_owned()), + kernel_summary: System::kernel_long_version(), + architecture: std::env::consts::ARCH, + cores: system.cpus().len(), + total_memory_bytes, + under_memory_pressure: total_memory_bytes != 0 && available_memory.saturating_mul(10) < total_memory_bytes, + filesystem_types, + interface_count: networks.len(), + bond_count: networks.keys().filter(|name| name.starts_with("bond")).count(), + } + } +} + +async fn collect_system_snapshot(cancel: &CancellationToken) -> Result { + let deadline = tokio::time::Instant::now() + COLLECT_TIMEOUT; + let permit = tokio::select! { + biased; + () = cancel.cancelled() => return Err(CollectorError::Cancelled), + result = tokio::time::timeout_at(deadline, SYSTEM_SCAN_PERMIT.clone().acquire_owned()) => { + match result { + Ok(Ok(permit)) => permit, + Ok(Err(_)) => return Err(CollectorError::TaskFailed), + Err(_) => return Err(CollectorError::TimedOut), + } + } + }; + let task = tokio::task::spawn_blocking(move || { + let _permit = permit; + SystemSnapshot::collect() + }); + tokio::select! { + biased; + () = cancel.cancelled() => Err(CollectorError::Cancelled), + result = tokio::time::timeout_at(deadline, task) => { + match result { + Ok(Ok(snapshot)) => Ok(snapshot), + Ok(Err(_)) => Err(CollectorError::TaskFailed), + Err(_) => Err(CollectorError::TimedOut), + } + } + } +} + +/// Collect all and only the Q07 offline L0/L1 fields, with one independently +/// bounded and redacted manifest entry per field. +pub async fn collect_offline_diagnostics( + storage_info: &StorageInfo, + cancel: &CancellationToken, +) -> Result, CollectorError> { + if cancel.is_cancelled() { + return Err(CollectorError::Cancelled); + } + let storage = StorageSnapshot::try_from(storage_info)?; + let system = collect_system_snapshot(cancel).await?; + + let mut entries = Vec::with_capacity(COLLECTORS.len()); + for collector in COLLECTORS { + entries.push(ManifestEntry::from_value(collector, collector.value(&storage, &system), cancel)?); + } + Ok(entries) +} + +#[cfg(test)] +mod test_support { + use std::sync::atomic::{AtomicU64, AtomicUsize, Ordering}; + use std::time::Duration; + + pub(super) static DELAY_MILLIS: AtomicU64 = AtomicU64::new(0); + pub(super) static ACTIVE: AtomicUsize = AtomicUsize::new(0); + pub(super) static MAX_ACTIVE: AtomicUsize = AtomicUsize::new(0); + + pub(super) struct ScanGuard; + + impl ScanGuard { + pub(super) fn start() -> Self { + let active = ACTIVE.fetch_add(1, Ordering::SeqCst) + 1; + MAX_ACTIVE.fetch_max(active, Ordering::SeqCst); + let delay = DELAY_MILLIS.load(Ordering::SeqCst); + if delay != 0 { + std::thread::sleep(Duration::from_millis(delay)); + } + Self + } + } + + impl Drop for ScanGuard { + fn drop(&mut self) { + ACTIVE.fetch_sub(1, Ordering::SeqCst); + } + } +} + +#[cfg(test)] +mod tests { + use std::sync::atomic::Ordering; + + use rustfs_madmin::StorageInfo; + + use super::*; + + async fn wait_for_active(expected: usize) { + tokio::time::timeout(Duration::from_secs(1), async { + while test_support::ACTIVE.load(Ordering::SeqCst) != expected { + tokio::time::sleep(Duration::from_millis(5)).await; + } + }) + .await + .expect("system scan reaches expected state"); + } + + #[tokio::test] + async fn connect_offline_collectors_timeout_and_cancel_never_overlap_system_scans() { + test_support::MAX_ACTIVE.store(0, Ordering::SeqCst); + test_support::DELAY_MILLIS.store((COLLECT_TIMEOUT + Duration::from_millis(200)).as_millis() as u64, Ordering::SeqCst); + + let first_cancel = CancellationToken::new(); + assert!(matches!( + collect_offline_diagnostics(&StorageInfo::default(), &first_cancel).await, + Err(CollectorError::TimedOut) + )); + assert_eq!(test_support::ACTIVE.load(Ordering::SeqCst), 1, "timed-out blocking scan remains active"); + + let second_cancel = CancellationToken::new(); + let second = tokio::spawn({ + let second_cancel = second_cancel.clone(); + async move { collect_offline_diagnostics(&StorageInfo::default(), &second_cancel).await } + }); + tokio::time::sleep(Duration::from_millis(50)).await; + assert_eq!( + test_support::MAX_ACTIVE.load(Ordering::SeqCst), + 1, + "a timed-out scan keeps the single-flight permit" + ); + second_cancel.cancel(); + assert!(matches!(second.await.expect("second collector task"), Err(CollectorError::Cancelled))); + wait_for_active(0).await; + + test_support::MAX_ACTIVE.store(0, Ordering::SeqCst); + test_support::DELAY_MILLIS.store(250, Ordering::SeqCst); + let third_cancel = CancellationToken::new(); + let third = tokio::spawn({ + let third_cancel = third_cancel.clone(); + async move { collect_offline_diagnostics(&StorageInfo::default(), &third_cancel).await } + }); + wait_for_active(1).await; + third_cancel.cancel(); + assert!(matches!(third.await.expect("third collector task"), Err(CollectorError::Cancelled))); + + let fourth_cancel = CancellationToken::new(); + let fourth = tokio::spawn({ + let fourth_cancel = fourth_cancel.clone(); + async move { collect_offline_diagnostics(&StorageInfo::default(), &fourth_cancel).await } + }); + tokio::time::sleep(Duration::from_millis(50)).await; + assert_eq!( + test_support::MAX_ACTIVE.load(Ordering::SeqCst), + 1, + "a cancelled scan keeps the single-flight permit" + ); + fourth_cancel.cancel(); + assert!(matches!(fourth.await.expect("fourth collector task"), Err(CollectorError::Cancelled))); + wait_for_active(0).await; + test_support::DELAY_MILLIS.store(0, Ordering::SeqCst); + } +} diff --git a/rustfs/src/connect/offline/manifest_entry.rs b/rustfs/src/connect/offline/manifest_entry.rs new file mode 100644 index 000000000..bae49c5f3 --- /dev/null +++ b/rustfs/src/connect/offline/manifest_entry.rs @@ -0,0 +1,74 @@ +// 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. + +//! A bounded, already-redacted offline diagnostic manifest entry. + +use std::time::Instant; + +use serde::Serialize; +use serde_json::{Map, Value}; +use tokio_util::sync::CancellationToken; + +use super::collectors::{CollectorError, DataClassification, OfflineCollector}; +use super::redaction::{REDACTION_VERSION, RULESET_HASH, RedactionSource, redact}; + +#[derive(Clone, Debug, PartialEq, Serialize)] +#[serde(rename_all = "camelCase")] +pub struct ManifestEntry { + pub field_id: &'static str, + pub classification: DataClassification, + pub canonical_json: String, + pub redaction_version: &'static str, + pub ruleset_hash: &'static str, + pub redacted_count: usize, +} + +impl ManifestEntry { + pub(super) fn from_value( + collector: OfflineCollector, + value: Value, + cancel: &CancellationToken, + ) -> Result { + let started = Instant::now(); + if cancel.is_cancelled() { + return Err(CollectorError::Cancelled); + } + let mut document = Map::new(); + document.insert(collector.field_name().to_owned(), value); + let input_size = serde_json::to_vec(&document) + .map_err(|_| CollectorError::NotRepresentable)? + .len(); + if input_size > collector.max_entry_bytes() { + return Err(CollectorError::EntryTooLarge { + field_id: collector.field_id(), + limit: collector.max_entry_bytes(), + }); + } + let result = redact(RedactionSource::OfflineDiagnostic, &document)?; + if started.elapsed() > collector.timeout() { + return Err(CollectorError::TimedOut); + } + if cancel.is_cancelled() { + return Err(CollectorError::Cancelled); + } + Ok(Self { + field_id: collector.field_id(), + classification: collector.classification(), + canonical_json: result.canonical_json, + redaction_version: REDACTION_VERSION, + ruleset_hash: RULESET_HASH, + redacted_count: result.redacted_count, + }) + } +} diff --git a/rustfs/src/connect/offline/mod.rs b/rustfs/src/connect/offline/mod.rs index c5364cb0b..10b1cd6fa 100644 --- a/rustfs/src/connect/offline/mod.rs +++ b/rustfs/src/connect/offline/mod.rs @@ -28,8 +28,14 @@ //! frozen by `protocol/agent/v1/fixtures/offline-enrollment/` and by //! `docs/adr/0009-offline-signing.md` on the Connect side. +pub mod collectors; pub mod enrollment; pub mod key_store; +pub mod manifest_entry; +pub mod redaction; +pub use collectors::{CollectorError, OfflineCollector, collect_offline_diagnostics}; pub use enrollment::{EnrollmentError, OfflineEnrollment, VerifiedChallenge}; pub use key_store::OfflineKeyStore; +pub use manifest_entry::ManifestEntry; +pub use redaction::{RedactionError, RedactionResult, RedactionSource, redact_json}; diff --git a/rustfs/src/connect/offline/redaction.rs b/rustfs/src/connect/offline/redaction.rs new file mode 100644 index 000000000..03c4a2691 --- /dev/null +++ b/rustfs/src/connect/offline/redaction.rs @@ -0,0 +1,362 @@ +// Copyright 2024 RustFS Team +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +//! Deterministic redaction shared with Connect's frozen D05 fixture contract. + +use std::sync::LazyLock; + +use regex::Regex; +use serde::Serialize; +use serde_json::{Map, Value}; +use thiserror::Error; + +pub const REDACTION_VERSION: &str = "rustfs.connect.redaction.v1"; +pub const RULESET_HASH: &str = "b37436d8e72515394a122d633865b1dc028d4ece349352a0a3a23f52ca4285f3"; +const PLACEHOLDER: &str = "[REDACTED]"; +const MAX_INPUT_BYTES: usize = 262_144; +const MAX_DEPTH: usize = 8; +const MAX_NODES: usize = 4_096; +const MAX_VALUE_BYTES: usize = 4_096; + +const HEARTBEAT_FIELDS: &[&str] = &[ + "agentVersion", + "capabilities", + "clientTime", + "coarseNodeSummary", + "protocolVersion", + "sequence", +]; +const INVENTORY_FIELDS: &[&str] = &[ + "capacityTotalBytes", + "capacityUsedBytes", + "coarseFlags", + "driveCount", + "nodeCount", + "osVersion", + "rustfsVersion", +]; +const OFFLINE_FIELDS: &[&str] = &[ + "capacityTotalBytes", + "capacityUsedBytes", + "coarseHealthFlags", + "cpuSummary", + "driveCount", + "filesystemSummary", + "kernelSummary", + "memorySummary", + "networkSummary", + "nodeCount", + "osSummary", + "rustfsVersion", +]; + +const KEY_RULES: &[&str] = &[ + "accesskey", + "accesskeyid", + "apikey", + "apitoken", + "authorization", + "bearertoken", + "cookie", + "credential", + "credentials", + "csrftoken", + "kmskey", + "kmskeyid", + "kmsmasterkey", + "kmssecret", + "passphrase", + "passwd", + "password", + "privatekey", + "pwd", + "refreshtoken", + "registrationtoken", + "secret", + "secretaccesskey", + "secretkey", + "sessioncookie", + "sessionid", + "sessiontoken", + "signingkey", + "token", + "xsrftoken", +]; + +static VALUE_RULES: LazyLock> = LazyLock::new(|| { + [ + r"(?-u:\b)(?:A3T[A-Z0-9]{2}|ABIA|ACCA|AKIA|ASIA)[A-Z0-9]{16}(?-u:\b)", + r"(?i:(?-u:\b)bearer\s{1,8}[A-Za-z0-9\-._~+/]{8,4096}={0,2})", + r#"(?i:(?-u:\b)[a-z0-9_.-]{0,24}(?:access[_.-]?key(?:[_.-]?id)?|api[_.-]?key|credentials?|passphrase|secret(?:[_.-]?key)?|token)(?-u:\b)\s{0,8}[:=]\s{0,8}["']?[A-Za-z0-9\-._~+/=]{8,4096})"#, + r"(?-u:\b)eyJ[A-Za-z0-9_-]{4,4096}\.[A-Za-z0-9_-]{4,4096}\.[A-Za-z0-9_-]{4,4096}", + r"(?i:(?-u:\b)(?:passwd|password|pwd)(?-u:\b)\s{0,8}[:=]\s{0,8}\S)", + r"-----BEGIN [A-Z0-9 ]{0,32}PRIVATE KEY(?: BLOCK)?-----", + r#"(?i:(?-u:\b)(?:csrf[_.-]?token|jsessionid|phpsessid|sess|session|sid|xsrf[_.-]?token)(?:[_.-]?id)?(?-u:\b)\s{0,8}[:=]\s{0,8}["']?[A-Za-z0-9%\-._~+/]{12,4096})"#, + r"(?-u:\b)[a-zA-Z][a-zA-Z0-9+.\-]{0,31}://[^\s/@:]{1,256}(?::[^\s/@]{0,256})?@", + ] + .into_iter() + .map(|pattern| Regex::new(pattern).expect("the frozen redaction patterns are valid")) + .collect() +}); + +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +pub enum RedactionSource { + Heartbeat, + Inventory, + OfflineDiagnostic, +} + +impl TryFrom<&str> for RedactionSource { + type Error = RedactionError; + + fn try_from(value: &str) -> Result { + match value { + "heartbeat" => Ok(Self::Heartbeat), + "inventory" => Ok(Self::Inventory), + "offline-diagnostic" => Ok(Self::OfflineDiagnostic), + _ => Err(RedactionError::UnknownSurface), + } + } +} + +impl RedactionSource { + fn allows(self, field: &str) -> bool { + match self { + Self::Heartbeat => HEARTBEAT_FIELDS.contains(&field), + Self::Inventory => INVENTORY_FIELDS.contains(&field), + Self::OfflineDiagnostic => OFFLINE_FIELDS.contains(&field), + } + } +} + +#[derive(Clone, Debug, Default, PartialEq, Eq, Serialize)] +#[serde(rename_all = "camelCase")] +pub struct RedactionCounts { + pub dropped_field: usize, + pub redacted_value: usize, + pub redacted_oversize_value: usize, +} + +#[derive(Clone, Debug, PartialEq, Serialize)] +#[serde(rename_all = "camelCase")] +pub struct RedactionResult { + pub document: Map, + pub canonical_json: String, + pub redaction_version: &'static str, + pub ruleset_hash: &'static str, + pub redacted_count: usize, + pub counts: RedactionCounts, +} + +#[derive(Clone, Copy, Debug, Error, PartialEq, Eq)] +pub enum RedactionError { + #[error("Redaction refused the document: it names no registered collection surface.")] + UnknownSurface, + #[error("Redaction refused the document: its size in bytes exceeds the frozen budget of 262144.")] + InputTooLarge, + #[error("Redaction refused the document: its nesting depth exceeds the frozen budget of 8.")] + TooDeep, + #[error("Redaction refused the document: its node count exceeds the frozen budget of 4096.")] + TooManyNodes, + #[error("Redaction refused the document: it is not representable as JSON.")] + NotRepresentable, +} + +#[must_use] +pub(super) fn redact(source: RedactionSource, document: &Map) -> Result { + let encoded = serde_json::to_vec(document).map_err(|_| RedactionError::NotRepresentable)?; + if encoded.len() > MAX_INPUT_BYTES { + return Err(RedactionError::InputTooLarge); + } + + let mut counts = RedactionCounts::default(); + let allowed = document + .iter() + .filter_map(|(key, value)| { + if source.allows(key) { + Some((key.clone(), value.clone())) + } else { + counts.dropped_field += 1; + None + } + }) + .collect(); + let mut nodes = 0; + let (_, redacted) = walk_map(allowed, 0, &mut nodes, &mut counts)?; + let canonical_json = serde_json::to_string(&redacted).map_err(|_| RedactionError::NotRepresentable)?; + let redacted_count = counts.dropped_field + counts.redacted_value + counts.redacted_oversize_value; + + Ok(RedactionResult { + document: redacted, + canonical_json, + redaction_version: REDACTION_VERSION, + ruleset_hash: RULESET_HASH, + redacted_count, + counts, + }) +} + +/// Redact a JSON object received at a protocol boundary. Invalid JSON and +/// non-object JSON are refused without including any input bytes in the error. +pub fn redact_json(source: RedactionSource, encoded: &[u8]) -> Result { + if encoded.len() > MAX_INPUT_BYTES { + return Err(RedactionError::InputTooLarge); + } + let Value::Object(document) = serde_json::from_slice(encoded).map_err(|_| RedactionError::NotRepresentable)? else { + return Err(RedactionError::NotRepresentable); + }; + redact(source, &document) +} + +fn walk_map( + map: Map, + depth: usize, + nodes: &mut usize, + counts: &mut RedactionCounts, +) -> Result<(bool, Map), RedactionError> { + check_depth(depth)?; + let mut out = Map::new(); + for (key, value) in map { + count_node(nodes)?; + if !is_ascii_key(&key) { + counts.dropped_field += 1; + continue; + } + match value { + Value::Object(nested) => { + let (keep, nested) = walk_map(nested, depth + 1, nodes, counts)?; + if keep { + out.insert(key, Value::Object(nested)); + } else { + counts.dropped_field += 1; + } + } + Value::Array(list) => { + out.insert(key.clone(), Value::Array(walk_list(list, &key, depth + 1, nodes, counts)?)); + } + scalar => { + out.insert(key.clone(), scrub_value(&key, scalar, counts)); + } + } + } + out.sort_keys(); + Ok((!out.is_empty(), out)) +} + +fn walk_list( + list: Vec, + key: &str, + depth: usize, + nodes: &mut usize, + counts: &mut RedactionCounts, +) -> Result, RedactionError> { + check_depth(depth)?; + let mut out = Vec::with_capacity(list.len()); + for value in list { + count_node(nodes)?; + match value { + Value::Object(nested) => { + let (keep, nested) = walk_map(nested, depth + 1, nodes, counts)?; + if keep { + out.push(Value::Object(nested)); + } else { + counts.dropped_field += 1; + } + } + Value::Array(nested) => out.push(Value::Array(walk_list(nested, key, depth + 1, nodes, counts)?)), + scalar => out.push(scrub_value(key, scalar, counts)), + } + } + Ok(out) +} + +fn check_depth(depth: usize) -> Result<(), RedactionError> { + if depth > MAX_DEPTH { + Err(RedactionError::TooDeep) + } else { + Ok(()) + } +} + +fn count_node(nodes: &mut usize) -> Result<(), RedactionError> { + *nodes += 1; + if *nodes > MAX_NODES { + Err(RedactionError::TooManyNodes) + } else { + Ok(()) + } +} + +fn scrub_value(key: &str, value: Value, counts: &mut RedactionCounts) -> Value { + let Value::String(text) = value else { + return value; + }; + if text.len() > MAX_VALUE_BYTES { + counts.redacted_oversize_value += 1; + return Value::String(PLACEHOLDER.to_owned()); + } + if redacts_key(key) || matches_value(&text) { + counts.redacted_value += 1; + return Value::String(PLACEHOLDER.to_owned()); + } + Value::String(text) +} + +fn is_ascii_key(key: &str) -> bool { + (1..=64).contains(&key.len()) + && key.is_ascii() + && key.as_bytes()[0].is_ascii_alphanumeric() + && key.as_bytes()[1..] + .iter() + .all(|byte| byte.is_ascii_alphanumeric() || matches!(byte, b'_' | b'-' | b'.')) +} + +fn redacts_key(key: &str) -> bool { + let normalized: String = key + .bytes() + .filter(|byte| !matches!(byte, b'_' | b'-' | b'.')) + .map(|byte| byte.to_ascii_lowercase() as char) + .collect(); + KEY_RULES.contains(&normalized.as_str()) +} + +fn matches_value(value: &str) -> bool { + matches_aws_secret_access_key(value) || VALUE_RULES.iter().any(|rule| rule.is_match(value)) +} + +fn matches_aws_secret_access_key(value: &str) -> bool { + let bytes = value.as_bytes(); + let is_secret_char = |byte: u8| byte.is_ascii_alphanumeric() || matches!(byte, b'+' | b'/'); + let mut start = 0; + while start < bytes.len() { + while start < bytes.len() && !is_secret_char(bytes[start]) { + start += 1; + } + let mut end = start; + while end < bytes.len() && is_secret_char(bytes[end]) { + end += 1; + } + let candidate = &bytes[start..end]; + if candidate.len() == 40 + && bytes.get(end) != Some(&b'=') + && candidate.iter().any(u8::is_ascii_lowercase) + && candidate.iter().any(u8::is_ascii_uppercase) + && candidate.iter().any(u8::is_ascii_digit) + { + return true; + } + start = end.saturating_add(1); + } + false +} diff --git a/rustfs/tests/connect_offline_collectors.rs b/rustfs/tests/connect_offline_collectors.rs new file mode 100644 index 000000000..bf706baf1 --- /dev/null +++ b/rustfs/tests/connect_offline_collectors.rs @@ -0,0 +1,290 @@ +// 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. + +//! Offline collector and frozen redaction conformance tests. + +use std::fs; +use std::path::PathBuf; + +use rustfs::connect::offline::{CollectorError, RedactionSource, collect_offline_diagnostics, redact_json}; +use rustfs_madmin::{Disk, ITEM_OFFLINE, StorageInfo}; +use serde_json::{Map, Value, json}; +use sha2::{Digest as _, Sha256}; +use tokio_util::sync::CancellationToken; + +fn fixture_dir() -> PathBuf { + PathBuf::from(env!("CARGO_MANIFEST_DIR")).join("../protocol/agent/v1/fixtures/redaction") +} + +fn fixture_json(name: &str) -> Value { + let manifest = fs::read_to_string(fixture_dir().join("MANIFEST.sha256")).expect("read fixture manifest"); + let expected = manifest + .lines() + .find_map(|line| { + let (digest, file) = line.split_once(" ")?; + (file == name).then_some(digest) + }) + .unwrap_or_else(|| panic!("{name} is listed in the fixture manifest")); + let bytes = fs::read(fixture_dir().join(name)).unwrap_or_else(|error| panic!("read {name}: {error}")); + let actual = Sha256::digest(&bytes) + .iter() + .map(|byte| format!("{byte:02x}")) + .collect::(); + assert_eq!(actual, expected, "{name} matches the frozen manifest"); + serde_json::from_slice(&bytes).unwrap_or_else(|error| panic!("parse {name}: {error}")) +} + +fn vectors(name: &str) -> Vec { + fixture_json(name)["vectors"].as_array().expect("fixture vectors").clone() +} + +fn object(value: &Value) -> Map { + value.as_object().expect("fixture document is an object").clone() +} + +fn redact_document(source: RedactionSource, document: &Map) -> rustfs::connect::offline::RedactionResult { + let encoded = serde_json::to_vec(document).expect("fixture document is representable"); + redact_json(source, &encoded).expect("fixture document must be accepted") +} + +#[test] +fn connect_offline_collectors_match_every_allowed_and_secret_redaction_vector() { + let ruleset = fixture_json("ruleset.json"); + for fixture in ["allowed-vectors.json", "secret-vectors.json"] { + for vector in vectors(fixture) { + let name = vector["name"].as_str().expect("vector name"); + let source = RedactionSource::try_from(vector["source"].as_str().expect("vector source")) + .unwrap_or_else(|error| panic!("{name}: {error}")); + let result = redact_document(source, &object(&vector["document"])); + assert_eq!(result.redaction_version, ruleset["redactionVersion"], "{name}: redaction version"); + assert_eq!(result.ruleset_hash, ruleset["rulesetHash"], "{name}: ruleset hash"); + assert_eq!(result.canonical_json, vector["expectedCanonicalJson"], "{name}: canonical bytes"); + if let Some(expected) = vector["expectedRedactedCount"].as_u64() { + assert_eq!(result.redacted_count as u64, expected, "{name}: redacted count"); + assert_eq!(result.counts.dropped_field as u64, vector["expectedCounts"]["droppedField"], "{name}"); + assert_eq!(result.counts.redacted_value as u64, vector["expectedCounts"]["redactedValue"], "{name}"); + assert_eq!( + result.counts.redacted_oversize_value as u64, vector["expectedCounts"]["redactedOversizeValue"], + "{name}" + ); + } else { + assert_eq!(result.redacted_count, 0, "{name}: allowed vectors must not be changed"); + } + if let Some(secrets) = vector["secretLiterals"].as_array() { + for secret in secrets { + assert!( + !result.canonical_json.contains(secret.as_str().expect("secret literal")), + "{name}: a secret survived redaction" + ); + } + } + } + } +} + +#[test] +fn connect_offline_collectors_enforce_the_frozen_rejection_budgets() { + for vector in vectors("rejection-vectors.json") { + let name = vector["name"].as_str().expect("vector name"); + let expected = vector["expected"]["message"].as_str(); + let source = match RedactionSource::try_from(vector["source"].as_str().expect("source")) { + Ok(source) => source, + Err(error) => { + assert_eq!(error.to_string(), expected.expect("refusal message"), "{name}"); + continue; + } + }; + let build = &vector["build"]; + let field = build["field"].as_str().unwrap_or("rustfsVersion"); + let document = match build["kind"].as_str().expect("builder kind") { + "literal" => object(&build["document"]), + "bulkStrings" => { + let entries = build["entries"].as_u64().expect("entries") as usize; + let bytes = build["valueBytes"].as_u64().expect("value bytes") as usize; + let nested = (0..entries) + .map(|index| (format!("f{index}"), json!("a".repeat(bytes)))) + .collect(); + Map::from_iter([(field.to_owned(), Value::Object(nested))]) + } + "nestedDepth" => { + let mut nested = json!({ "leaf": 1 }); + for _ in 0..build["depth"].as_u64().expect("depth") { + nested = json!({ "nested": nested }); + } + Map::from_iter([(field.to_owned(), nested)]) + } + "listNodes" => { + let count = build["count"].as_u64().expect("count") as usize; + Map::from_iter([(field.to_owned(), Value::Array(vec![json!(1); count]))]) + } + "unrepresentable" => { + let encoded = format!(r#"{{"{field}":NaN}}"#); + assert_eq!( + redact_json(source, encoded.as_bytes()).expect_err(name).to_string(), + expected.expect("refusal message"), + "{name}" + ); + continue; + } + kind => panic!("unknown fixture builder {kind}"), + }; + match expected { + Some(message) => { + let encoded = serde_json::to_vec(&document).expect("rejection fixture document is representable"); + assert_eq!(redact_json(source, &encoded).expect_err(name).to_string(), message, "{name}") + } + None => assert_eq!( + redact_document(source, &document).redacted_count, + vector["expected"]["redactedCount"], + "{name}" + ), + } + } +} + +#[test] +fn connect_offline_collectors_reject_oversize_raw_input_before_parsing() { + let invalid = vec![b'!'; 262_145]; + assert_eq!( + redact_json(RedactionSource::OfflineDiagnostic, &invalid) + .expect_err("oversize raw input") + .to_string(), + "Redaction refused the document: its size in bytes exceeds the frozen budget of 262144." + ); +} + +#[tokio::test] +async fn connect_offline_collectors_emit_only_fixed_redacted_entries_and_honor_cancellation() { + let storage = StorageInfo { + disks: vec![ + Disk { + endpoint: "https://node-a.private.example:9000/data-a".to_owned(), + drive_path: "/secret/customer/path-a".to_owned(), + uuid: "private-drive-a".to_owned(), + state: "ok".to_owned(), + total_space: 1_000, + used_space: 400, + ..Disk::default() + }, + Disk { + endpoint: "https://node-a.private.example:9000/data-b".to_owned(), + drive_path: "/secret/customer/path-b".to_owned(), + uuid: "private-drive-b".to_owned(), + state: "unformatted".to_owned(), + total_space: 2_000, + used_space: 500, + ..Disk::default() + }, + Disk { + endpoint: "https://node-b.private.example:9000/data-c".to_owned(), + drive_path: "/secret/customer/path-c".to_owned(), + uuid: "private-drive-c".to_owned(), + state: ITEM_OFFLINE.to_owned(), + total_space: 3_000, + used_space: 600, + healing: true, + ..Disk::default() + }, + ], + ..StorageInfo::default() + }; + let cancel = CancellationToken::new(); + let entries = collect_offline_diagnostics(&storage, &cancel) + .await + .expect("collect fixed offline entries"); + assert_eq!(entries.len(), 12); + let encoded = serde_json::to_string(&entries).expect("manifest entries serialize"); + for forbidden in [ + "node-a.private.example", + "node-b.private.example", + "/secret/customer/path-a", + "/secret/customer/path-b", + "/secret/customer/path-c", + "private-drive-a", + "private-drive-b", + "private-drive-c", + ] { + assert!(!encoded.contains(forbidden), "private storage metadata must not leave the collector"); + } + assert!(entries.iter().all(|entry| entry.field_id.starts_with("offline."))); + assert!(entries.iter().all(|entry| entry.canonical_json.len() <= 16 * 1024)); + let canonical = |field_id| { + entries + .iter() + .find(|entry| entry.field_id == field_id) + .unwrap_or_else(|| panic!("missing {field_id}")) + .canonical_json + .as_str() + }; + assert_eq!(canonical("offline.nodeCount"), r#"{"nodeCount":2}"#); + assert_eq!(canonical("offline.driveCount"), r#"{"driveCount":3}"#); + assert_eq!(canonical("offline.capacityUsedBytes"), r#"{"capacityUsedBytes":1500}"#); + assert_eq!(canonical("offline.capacityTotalBytes"), r#"{"capacityTotalBytes":6000}"#); + assert_eq!( + canonical("offline.coarseHealthFlags"), + r#"{"coarseHealthFlags":{"degraded":true,"healing":true,"offlineDrives":1,"scanning":false}}"# + ); + + let healthy = StorageInfo { + disks: ["ok", "unformatted", "online"] + .into_iter() + .enumerate() + .map(|(index, state)| Disk { + endpoint: format!("https://healthy.example:9000/data-{index}"), + state: state.to_owned(), + ..Disk::default() + }) + .collect(), + ..StorageInfo::default() + }; + let healthy_entries = collect_offline_diagnostics(&healthy, &CancellationToken::new()) + .await + .expect("collect healthy storage summary"); + assert_eq!( + healthy_entries + .iter() + .find(|entry| entry.field_id == "offline.coarseHealthFlags") + .expect("healthy coarse health entry") + .canonical_json, + r#"{"coarseHealthFlags":{"degraded":false,"healing":false,"offlineDrives":0,"scanning":false}}"# + ); + + cancel.cancel(); + assert!(matches!( + collect_offline_diagnostics(&storage, &cancel).await, + Err(CollectorError::Cancelled) + )); + + let oversized = StorageInfo { + disks: vec![Disk::default(); 4_097], + ..StorageInfo::default() + }; + let active = CancellationToken::new(); + assert!(matches!( + collect_offline_diagnostics(&oversized, &active).await, + Err(CollectorError::StorageTopologyTooLarge) + )); + + let invalid_endpoint = StorageInfo { + disks: vec![Disk { + endpoint: "not-an-endpoint".to_owned(), + ..Disk::default() + }], + ..StorageInfo::default() + }; + assert!(matches!( + collect_offline_diagnostics(&invalid_endpoint, &active).await, + Err(CollectorError::InvalidStorageEndpoint) + )); +}