diff --git a/protocol/agent/v1/fixtures/inventory/MANIFEST.sha256 b/protocol/agent/v1/fixtures/inventory/MANIFEST.sha256 index cc0d06f4c..f1aef5b5c 100644 --- a/protocol/agent/v1/fixtures/inventory/MANIFEST.sha256 +++ b/protocol/agent/v1/fixtures/inventory/MANIFEST.sha256 @@ -1,3 +1,4 @@ +58d61bb10b9443eca3c0e9dc340cbe91144eed32a705617cc3406bd825439824 accept-vectors.json d1a73b0a348845bf3ed9fb68301babc5abbc6e72243a610db1cd4794b1328070 canonical-hash.json f1107d3e6accbaee468f1a1fdb79d7103fb2aadafd85c39020fbbca5173b03b4 field-registry.json b3b2e7f761198d4823c94440637a48153437183f4cacec5118570e1920f73b29 old-agent-vectors.json diff --git a/protocol/agent/v1/fixtures/inventory/accept-vectors.json b/protocol/agent/v1/fixtures/inventory/accept-vectors.json new file mode 100644 index 000000000..772d85a91 --- /dev/null +++ b/protocol/agent/v1/fixtures/inventory/accept-vectors.json @@ -0,0 +1,58 @@ +{ + "protocolVersion": "v1", + "fixtureSet": "inventory", + "fixture": "accept-vectors", + "description": "Producer vectors for the negotiated inventory.environment@1 summary. The input names source observations used by the RustFS collector; output is the complete identifier-free payload allowed to leave the deployment.", + "schemaVersion": 1, + "capability": "inventory.environment@1", + "vectors": [ + { + "name": "known two-node deployment", + "input": { + "persistedInventory": { "nodeCount": 2, "driveCount": 8 }, + "sourceObservation": { "osFamily": "LINUX", "filesystemTypes": ["xfs"] } + }, + "expected": { + "decision": "ACCEPT", + "output": { "nodeCount": 2, "driveCount": 8, "osFamily": "LINUX", "filesystemTypes": ["xfs"] } + } + }, + { + "name": "secret-like filesystem observations are not exported", + "input": { + "persistedInventory": { "nodeCount": 2, "driveCount": 8 }, + "sourceObservation": { + "osFamily": "LINUX", + "filesystemTypes": ["xfs"], + "mountPath": "/srv/synthetic-customer-a", + "mountOptions": "rw,password=SYNTHETIC_NOT_A_REAL_SECRET", + "deviceLabel": "synthetic-customer-volume", + "credential": "SYNTHETIC_NOT_A_REAL_CREDENTIAL" + } + }, + "expected": { + "decision": "ACCEPT", + "discarded": ["credential", "deviceLabel", "mountOptions", "mountPath"], + "output": { "nodeCount": 2, "driveCount": 8, "osFamily": "LINUX", "filesystemTypes": ["xfs"] } + } + }, + { + "name": "old schema is rejected before collection", + "input": { "schemaVersion": 0, "capability": "inventory.environment@1" }, + "expected": { "decision": "REJECT", "producerError": "inventory_environment_unsupported_version" } + }, + { + "name": "unknown capability is rejected before collection", + "input": { "schemaVersion": 1, "capability": "inventory.environment@2" }, + "expected": { "decision": "REJECT", "producerError": "inventory_environment_unsupported_capability" } + }, + { + "name": "missing filesystem source is not replaced with an empty success", + "input": { + "persistedInventory": { "nodeCount": 2, "driveCount": 8 }, + "sourceObservation": { "osFamily": "LINUX", "filesystemTypes": [] } + }, + "expected": { "decision": "UNSUPPORTED", "producerError": "inventory_environment_source_unavailable", "output": null } + } + ] +} diff --git a/rustfs/src/config/cli.rs b/rustfs/src/config/cli.rs index 93f5bb738..3589c0b0f 100644 --- a/rustfs/src/config/cli.rs +++ b/rustfs/src/config/cli.rs @@ -127,6 +127,8 @@ pub enum ConnectCommands { Register(ConnectRegisterOpts), /// Import, verify, or inspect a signed Connect service license License(ConnectLicenseOpts), + /// Read the persisted deployment inventory and collect an approved environment summary + Inventory(ConnectInventoryOpts), /// Run an explicitly approved, bounded local performance measurement Performance(ConnectPerformanceOpts), /// Capture a consent-bound local profile and write a signed export @@ -139,6 +141,37 @@ pub enum ConnectCommands { Top(ConnectTopOpts), } +#[derive(Args, Clone)] +pub struct ConnectInventoryOpts { + #[command(subcommand)] + pub command: ConnectInventoryCommands, +} + +#[derive(Subcommand, Clone)] +pub enum ConnectInventoryCommands { + /// Collect the bounded inventory.environment@1 summary + Environment(ConnectEnvironmentInventoryOpts), +} + +#[derive(Args, Clone)] +pub struct ConnectEnvironmentInventoryOpts { + /// Directory containing the persisted Connect inventory + #[arg(long = "state-dir")] + pub state_dir: PathBuf, + /// Negotiated environment schema version + #[arg(long = "schema-version", default_value_t = 1)] + pub schema_version: u16, + /// Negotiated environment capability + #[arg(long, default_value = "inventory.environment@1", value_parser = NonEmptyStringValueParser::new())] + pub capability: String, + /// Maximum collection time in seconds + #[arg(long = "timeout-seconds", default_value_t = 30)] + pub timeout_seconds: u64, + /// Confirm this explicit local L1 inventory operation + #[arg(long = "acknowledge-l1", required = true, action = clap::ArgAction::SetTrue)] + pub acknowledge_l1: bool, +} + #[derive(Args, Clone)] pub struct ConnectTopOpts { #[command(subcommand)] @@ -1057,6 +1090,8 @@ pub enum CommandResult { ConnectRegister(ConnectRegisterOpts), /// Local Connect service-license command ConnectLicense(ConnectLicenseCommands), + /// Explicit local Connect environment inventory command + ConnectEnvironmentInventory(ConnectEnvironmentInventoryOpts), /// Consent-bound local Connect drive performance export ConnectDrivePerformance(ConnectDrivePerformanceOpts), /// Consent-bound client-to-deployment performance export @@ -1248,6 +1283,43 @@ mod tests { assert!(help.to_string().contains("Unix only")); } + #[test] + fn connect_environment_inventory_requires_explicit_l1_acknowledgement() { + let error = Cli::try_parse_from([ + "rustfs", + "connect", + "inventory", + "environment", + "--state-dir", + "/var/lib/rustfs/connect", + ]) + .expect_err("unacknowledged L1 inventory must fail"); + assert_eq!(error.kind(), ErrorKind::MissingRequiredArgument); + assert!(error.to_string().contains("--acknowledge-l1")); + + let cli = Cli::try_parse_from([ + "rustfs", + "connect", + "inventory", + "environment", + "--state-dir", + "/var/lib/rustfs/connect", + "--acknowledge-l1", + ]) + .expect("acknowledged environment inventory parses"); + let Some(Commands::Connect(connect)) = cli.command else { + panic!("connect command expected"); + }; + let ConnectCommands::Inventory(inventory) = connect.command else { + panic!("inventory command expected"); + }; + let ConnectInventoryCommands::Environment(environment) = inventory.command; + assert_eq!(environment.schema_version, 1); + assert_eq!(environment.capability, "inventory.environment@1"); + assert_eq!(environment.timeout_seconds, 30); + assert!(environment.acknowledge_l1); + } + #[test] fn connect_profile_requires_explicit_l3_acknowledgement() { let arguments = [ diff --git a/rustfs/src/config/mod.rs b/rustfs/src/config/mod.rs index c8b472c15..2a4ac6efb 100644 --- a/rustfs/src/config/mod.rs +++ b/rustfs/src/config/mod.rs @@ -54,6 +54,7 @@ pub use cli::{CommandResult, InfoOpts, InfoType}; pub use cli::{ ConnectClientPerformanceOperation, ConnectClientPerformanceOpts, ConnectDrivePerformanceOpts, ConnectPerformanceCommands, }; +pub use cli::{ConnectEnvironmentInventoryOpts, ConnectInventoryCommands}; pub use cli::{ConnectLicenseArtifactOpts, ConnectLicenseCommands, ConnectLicenseScopeOpts}; pub use cli::{ConnectLogsMode, ConnectLogsOpts}; pub use cli::{ConnectProfileOpts, ConnectProfileTool, ConnectThreadProfileScope}; diff --git a/rustfs/src/config/opt.rs b/rustfs/src/config/opt.rs index acc0e86f4..c539a6c37 100644 --- a/rustfs/src/config/opt.rs +++ b/rustfs/src/config/opt.rs @@ -19,8 +19,8 @@ use super::Config; use super::cli::{ - Cli, CommandResult, Commands, ConnectCommands, ConnectPerformanceCommands, ServerOpts, default_server_opts, - preprocess_args_for_legacy, + Cli, CommandResult, Commands, ConnectCommands, ConnectInventoryCommands, ConnectPerformanceCommands, ServerOpts, + default_server_opts, preprocess_args_for_legacy, }; use crate::apply_external_env_compat; use CommandResult::Server; @@ -143,6 +143,9 @@ impl Opt { Some(Commands::Connect(opts)) => match opts.command { ConnectCommands::Register(opts) => Ok(CommandResult::ConnectRegister(opts)), ConnectCommands::License(opts) => Ok(CommandResult::ConnectLicense(opts.command)), + ConnectCommands::Inventory(opts) => match opts.command { + ConnectInventoryCommands::Environment(opts) => Ok(CommandResult::ConnectEnvironmentInventory(opts)), + }, ConnectCommands::Performance(opts) => match opts.command { ConnectPerformanceCommands::Client(opts) => Ok(CommandResult::ConnectClientPerformance(*opts)), ConnectPerformanceCommands::Drive(opts) => Ok(CommandResult::ConnectDrivePerformance(*opts)), diff --git a/rustfs/src/startup_entrypoint.rs b/rustfs/src/startup_entrypoint.rs index 89ebda2eb..d35309b29 100644 --- a/rustfs/src/startup_entrypoint.rs +++ b/rustfs/src/startup_entrypoint.rs @@ -15,9 +15,9 @@ use crate::{ config::{ CommandResult, Config, ConnectClientPerformanceOperation, ConnectClientPerformanceOpts, ConnectDrivePerformanceOpts, - ConnectLicenseCommands, ConnectLicenseScopeOpts, ConnectLogsMode, ConnectLogsOpts, ConnectProfileOpts, - ConnectProfileTool, ConnectTelemetryArtifactOpts, ConnectTelemetryCommands, ConnectThreadProfileScope, - ConnectTopCommands, Opt, + ConnectEnvironmentInventoryOpts, ConnectLicenseCommands, ConnectLicenseScopeOpts, ConnectLogsMode, ConnectLogsOpts, + ConnectProfileOpts, ConnectProfileTool, ConnectTelemetryArtifactOpts, ConnectTelemetryCommands, + ConnectThreadProfileScope, ConnectTopCommands, Opt, }, startup_lifecycle::{StartupRuntimeLifecycle, run_startup_runtime_lifecycle}, startup_preflight::{StartupServerPreflightError, bootstrap_external_prefix_compat, init_startup_server_preflight}, @@ -137,6 +137,7 @@ async fn async_main() -> Result<()> { return Ok(()); } CommandResult::ConnectLicense(command) => return execute_connect_license(command), + CommandResult::ConnectEnvironmentInventory(options) => return execute_connect_environment_inventory(options).await, CommandResult::ConnectClientPerformance(options) => return execute_connect_client_performance(options).await, CommandResult::ConnectDrivePerformance(options) => return execute_connect_drive_performance(options).await, CommandResult::ConnectProfile(options) => return execute_connect_profile(options).await, @@ -172,6 +173,41 @@ async fn async_main() -> Result<()> { } } +async fn execute_connect_environment_inventory(options: ConnectEnvironmentInventoryOpts) -> Result<()> { + use crate::connect::environment::collect_environment; + use crate::connect::inventory::InventoryStateStore; + use crate::connect::{EnvironmentCollectionRequest, EnvironmentError}; + + let request = EnvironmentCollectionRequest::negotiate( + options.schema_version, + &options.capability, + Duration::from_secs(options.timeout_seconds), + ) + .map_err(Error::other)?; + let store = InventoryStateStore::from_state_root(&options.state_dir).map_err(Error::other)?; + let persisted = tokio::task::spawn_blocking(move || store.read_latest(chrono::Utc::now())) + .await + .map_err(Error::other)? + .map_err(Error::other)?; + let cancel = CancellationToken::new(); + let collection = collect_environment(&persisted.snapshot, request, &cancel); + tokio::pin!(collection); + let inventory = tokio::select! { + biased; + signal = tokio::signal::ctrl_c() => { + signal.map_err(Error::other)?; + cancel.cancel(); + collection.await.map_err(Error::other)? + } + result = collection.as_mut() => result.map_err(|error| match error { + EnvironmentError::Cancelled => Error::other("inventory environment collection cancelled"), + error => Error::other(error), + })?, + }; + println!("{}", serde_json::to_string(&inventory).map_err(Error::other)?); + Ok(()) +} + async fn execute_connect_logs(options: ConnectLogsOpts) -> Result<()> { use crate::connect::{ CaptureMode, IdentityStore, LocalLogConsent, LogCaptureRequest, LogProvenance, export_logs, save_signed_log_export, diff --git a/rustfs/tests/connect_environment.rs b/rustfs/tests/connect_environment.rs index 2a2751f1f..b5eee66d9 100644 --- a/rustfs/tests/connect_environment.rs +++ b/rustfs/tests/connect_environment.rs @@ -14,6 +14,17 @@ use std::time::Duration; +#[cfg(target_os = "linux")] +use std::fs; +#[cfg(target_os = "linux")] +use std::os::unix::fs::PermissionsExt as _; +#[cfg(target_os = "linux")] +use std::process::Command; + +#[cfg(target_os = "linux")] +use rustfs::connect::{ + CredentialStore, HeartbeatConfig, IdentityStore, InventorySchedule, InventoryStatus, spawn_inventory_runtime, +}; use rustfs::connect::{ ENVIRONMENT_CAPABILITY, ENVIRONMENT_SCHEMA_VERSION, EnvironmentCollectionRequest, EnvironmentError, EnvironmentFilesystemType, InventorySnapshot, MAX_ENVIRONMENT_DURATION, collect_environment, @@ -99,3 +110,75 @@ async fn environment_collection_emits_only_the_closed_identifier_free_schema() { println!("inventory.environment actual output: {encoded}"); assert!(matches!(serialized, Value::Object(_))); } + +#[cfg(target_os = "linux")] +#[tokio::test] +async fn production_binary_collects_environment_from_persisted_inventory() { + let temp = tempfile::tempdir_in(env!("CARGO_MANIFEST_DIR")).expect("safe tempdir"); + let state = temp.path().join("state"); + fs::create_dir(&state).expect("state root"); + fs::set_permissions(&state, fs::Permissions::from_mode(0o700)).expect("private state root"); + let config = HeartbeatConfig::new( + "", + Vec::new(), + IdentityStore::new(state.join("identity")), + CredentialStore::new(state.join("credential")), + state.join("heartbeat/state.json"), + ); + let shutdown = CancellationToken::new(); + let runtime = spawn_inventory_runtime(Some(config), InventorySchedule::default(), &shutdown, || { + std::future::ready(Ok(inventory())) + }) + .expect("state-only inventory runtime") + .expect("configured inventory runtime"); + let mut status = runtime.status(); + tokio::time::timeout(Duration::from_secs(5), async { + loop { + if matches!(&*status.borrow(), InventoryStatus::Unchanged { .. }) { + break; + } + status.changed().await.expect("inventory runtime remains active"); + } + }) + .await + .expect("persisted inventory timeout"); + runtime.shutdown().await; + + let output = Command::new(env!("CARGO_BIN_EXE_rustfs")) + .args([ + "connect", + "inventory", + "environment", + "--state-dir", + state.to_str().expect("UTF-8 state path"), + "--acknowledge-l1", + ]) + .output() + .expect("run production RustFS binary"); + assert!(output.status.success(), "stderr: {}", String::from_utf8_lossy(&output.stderr)); + let actual: Value = serde_json::from_slice(&output.stdout).expect("environment JSON output"); + assert_eq!(actual["nodeCount"], 2); + assert_eq!(actual["driveCount"], 8); + assert_eq!(actual.as_object().expect("environment object").len(), 4); + + for (argument, value, expected) in [ + ("--schema-version", "0", "inventory_environment_unsupported_version"), + ("--capability", "inventory.environment@2", "inventory_environment_unsupported_capability"), + ] { + let rejected = Command::new(env!("CARGO_BIN_EXE_rustfs")) + .args([ + "connect", + "inventory", + "environment", + "--state-dir", + state.to_str().expect("UTF-8 state path"), + "--acknowledge-l1", + argument, + value, + ]) + .output() + .expect("run incompatible production command"); + assert!(!rejected.status.success()); + assert!(String::from_utf8_lossy(&rejected.stderr).contains(expected)); + } +}