feat(connect): collect bounded offline diagnostics (#6450)

* feat(connect): collect bounded offline diagnostics

* fix(connect): tighten offline diagnostic boundaries

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

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",

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<Arc<Semaphore>> = 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<Self, Self::Error> {
+        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::<Result<BTreeSet<_>, 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<String>,
+    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::<BTreeSet<_>>()
+            .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<SystemSnapshot, CollectorError> {
+    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<Vec<ManifestEntry>, 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);
+    }
+}

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<Self, CollectorError> {
+        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,
+        })
+    }
+}

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};

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<Vec<Regex>> = 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<Self, Self::Error> {
+        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<String, Value>,
+    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<String, Value>) -> Result<RedactionResult, RedactionError> {
+    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<RedactionResult, RedactionError> {
+    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<String, Value>,
+    depth: usize,
+    nodes: &mut usize,
+    counts: &mut RedactionCounts,
+) -> Result<(bool, Map<String, Value>), 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<Value>,
+    key: &str,
+    depth: usize,
+    nodes: &mut usize,
+    counts: &mut RedactionCounts,
+) -> Result<Vec<Value>, 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
+}

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::<String>();
+    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<Value> {
+    fixture_json(name)["vectors"].as_array().expect("fixture vectors").clone()
+}
+
+fn object(value: &Value) -> Map<String, Value> {
+    value.as_object().expect("fixture document is an object").clone()
+}
+
+fn redact_document(source: RedactionSource, document: &Map<String, Value>) -> 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)
+    ));
+}