diff --git a/Cargo.lock b/Cargo.lock index 2930621..b494c44 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -2429,6 +2429,7 @@ dependencies = [ "thiserror 1.0.69", "tokio", "tracing", + "uuid", ] [[package]] diff --git a/sentinel-capabilities/Cargo.toml b/sentinel-capabilities/Cargo.toml index c6f7176..fc490c5 100644 --- a/sentinel-capabilities/Cargo.toml +++ b/sentinel-capabilities/Cargo.toml @@ -19,3 +19,4 @@ chrono = { workspace = true } tokio = { workspace = true } tempfile = { workspace = true } mockall = { workspace = true } +uuid = { workspace = true } diff --git a/sentinel-capabilities/src/network.rs b/sentinel-capabilities/src/network.rs index ea75a93..0375bea 100644 --- a/sentinel-capabilities/src/network.rs +++ b/sentinel-capabilities/src/network.rs @@ -246,7 +246,9 @@ impl Capability for NetworkInterfaces { mod tests { use super::*; use std::collections::HashMap; + use std::sync::Mutex; use sentinel_exec::{CommandExecutorTrait, CommandOutput}; + use uuid::Uuid; struct DummyExecutor; #[async_trait::async_trait] @@ -266,6 +268,118 @@ mod tests { Arc::new(DummyExecutor) } + /// Mock executor that records each invocation and returns canned stdout + /// per program (always exit 0). Lets us drive `invoke()` without the OS. + struct MockExecutor { + calls: Mutex)>>, + } + + impl MockExecutor { + fn new() -> Arc { + Arc::new(Self { calls: Mutex::new(Vec::new()) }) + } + } + + #[async_trait::async_trait] + impl CommandExecutorTrait for MockExecutor { + async fn run( + &self, + program: &str, + args: &[&str], + _env: &HashMap, + _max_output_bytes: usize, + ) -> Result { + self.calls + .lock() + .unwrap() + .push((program.to_string(), args.iter().map(|s| s.to_string()).collect())); + + let stdout = match program { + // `which ` succeeds → primary tool (ss / ip) is selected. + "which" => "/usr/bin/tool\n", + "ss" => { + "Netid State Recv-Q Send-Q Local Address:Port Peer Address:Port\n\ + tcp LISTEN 0 128 0.0.0.0:22 0.0.0.0:*\n\ + tcp ESTAB 0 0 192.168.1.1:22 10.0.0.1:54321\n" + } + "netstat" => "Proto Recv-Q Send-Q Local Address Foreign Address State\n", + "ip" => { + "1: lo: mtu 65536 state UNKNOWN\n \ + inet 127.0.0.1/8 scope host lo\n\ + 2: eth0: mtu 1500 state UP\n \ + inet 192.168.1.10/24 brd 192.168.1.255 scope global eth0\n" + } + "ifconfig" => "", + _ => "", + }; + + Ok(CommandOutput { + exit_code: Some(0), + stdout: stdout.to_string(), + stderr: String::new(), + truncated: false, + }) + } + } + + fn ctx() -> ExecutionContext { + ExecutionContext::new(Uuid::new_v4(), "localhost") + } + + #[tokio::test] + async fn network_connections_invoke_parses_ss_output() { + let exec = MockExecutor::new(); + let cap = NetworkConnections::new(Arc::clone(&exec) as Arc); + + let result = cap.invoke(json!({}), &ctx()).await; + assert!(result.is_success()); + if let CapabilityResult::Success { output } = result { + assert_eq!(output["tool"], "ss"); + // Two data lines (header skipped) → two connections. + assert_eq!(output["count"], 2); + assert_eq!(output["connections"][0]["proto"], "tcp"); + assert_eq!(output["connections"][0]["state"], "LISTEN"); + } + + // Verify it actually shelled out to `ss -tuln` (no shell, explicit args). + let calls = exec.calls.lock().unwrap(); + assert!(calls.iter().any(|(p, a)| p == "ss" && a == &["-tuln"])); + } + + #[tokio::test] + async fn network_connections_invoke_state_filter() { + let exec = MockExecutor::new(); + let cap = NetworkConnections::new(exec as Arc); + + let result = cap.invoke(json!({ "state": "LISTEN" }), &ctx()).await; + assert!(result.is_success()); + if let CapabilityResult::Success { output } = result { + // Only the LISTEN line survives the filter. + assert_eq!(output["count"], 1); + assert_eq!(output["connections"][0]["state"], "LISTEN"); + } + } + + #[tokio::test] + async fn network_interfaces_invoke_parses_ip_output() { + let exec = MockExecutor::new(); + let cap = NetworkInterfaces::new(Arc::clone(&exec) as Arc); + + let result = cap.invoke(json!({}), &ctx()).await; + assert!(result.is_success()); + if let CapabilityResult::Success { output } = result { + assert_eq!(output["tool"], "ip"); + let ifaces = output["interfaces"].as_array().unwrap(); + assert_eq!(ifaces.len(), 2); + assert_eq!(ifaces[0]["name"], "lo"); + assert_eq!(ifaces[1]["name"], "eth0"); + assert_eq!(ifaces[1]["state"], "UP"); + } + + let calls = exec.calls.lock().unwrap(); + assert!(calls.iter().any(|(p, a)| p == "ip" && a == &["addr"])); + } + #[test] fn network_connections_no_args() { let cap = NetworkConnections::new(make_executor()); diff --git a/sentinel-capabilities/src/packages.rs b/sentinel-capabilities/src/packages.rs index a8dfef2..34be747 100644 --- a/sentinel-capabilities/src/packages.rs +++ b/sentinel-capabilities/src/packages.rs @@ -323,7 +323,9 @@ impl Capability for PackageUpgrade { mod tests { use super::*; use std::collections::HashMap; + use std::sync::Mutex; use sentinel_exec::{CommandExecutorTrait, CommandOutput}; + use uuid::Uuid; struct DummyExecutor; #[async_trait::async_trait] @@ -343,6 +345,132 @@ mod tests { Arc::new(DummyExecutor) } + /// Mock executor: records calls and returns canned stdout (exit 0). `which` + /// succeeds for every candidate, so `apt` (first probed) is selected. + struct MockExecutor { + calls: Mutex)>>, + } + + impl MockExecutor { + fn new() -> Arc { + Arc::new(Self { calls: Mutex::new(Vec::new()) }) + } + } + + #[async_trait::async_trait] + impl CommandExecutorTrait for MockExecutor { + async fn run( + &self, + program: &str, + args: &[&str], + _env: &HashMap, + _max_output_bytes: usize, + ) -> Result { + self.calls + .lock() + .unwrap() + .push((program.to_string(), args.iter().map(|s| s.to_string()).collect())); + + let stdout = match program { + "which" => "/usr/bin/apt\n", + "apt" if args.contains(&"list") => { + "Listing... Done\n\ + nginx/stable 1.18.0 amd64 [installed]\n\ + openssl/stable 1.1.1 amd64 [installed]\n" + } + "apt" => "Reading package lists... Done\nUpgrading nginx (1.18.0 -> 1.18.1)\n", + _ => "", + }; + + Ok(CommandOutput { + exit_code: Some(0), + stdout: stdout.to_string(), + stderr: String::new(), + truncated: false, + }) + } + } + + fn ctx() -> ExecutionContext { + ExecutionContext::new(Uuid::new_v4(), "localhost") + } + + #[tokio::test] + async fn package_list_invoke_uses_apt() { + let exec = MockExecutor::new(); + let cap = PackageList::new(Arc::clone(&exec) as Arc); + + let result = cap.invoke(json!({}), &ctx()).await; + assert!(result.is_success()); + if let CapabilityResult::Success { output } = result { + assert_eq!(output["package_manager"], "apt"); + // Three stdout lines → three "packages" (no header skipping here). + assert_eq!(output["count"], 3); + } + + let calls = exec.calls.lock().unwrap(); + assert!(calls.iter().any(|(p, a)| p == "apt" && a == &["list", "--installed"])); + } + + #[tokio::test] + async fn package_list_invoke_filter() { + let exec = MockExecutor::new(); + let cap = PackageList::new(exec as Arc); + + let result = cap.invoke(json!({ "filter": "nginx" }), &ctx()).await; + assert!(result.is_success()); + if let CapabilityResult::Success { output } = result { + // Only the nginx line matches the filter. + assert_eq!(output["count"], 1); + } + } + + #[tokio::test] + async fn package_upgrade_all_invoke() { + let exec = MockExecutor::new(); + let cap = PackageUpgrade::new(Arc::clone(&exec) as Arc); + + let result = cap.invoke(json!({ "all": true }), &ctx()).await; + assert!(result.is_success()); + if let CapabilityResult::Success { output } = result { + assert_eq!(output["package_manager"], "apt"); + assert_eq!(output["success"], true); + } + + let calls = exec.calls.lock().unwrap(); + assert!(calls.iter().any(|(p, a)| p == "apt" && a == &["upgrade", "-y"])); + } + + #[tokio::test] + async fn package_upgrade_specific_invoke() { + let exec = MockExecutor::new(); + let cap = PackageUpgrade::new(Arc::clone(&exec) as Arc); + + let result = cap.invoke(json!({ "packages": ["nginx"] }), &ctx()).await; + assert!(result.is_success()); + + // No shell — explicit `apt install --only-upgrade -y nginx`. + let calls = exec.calls.lock().unwrap(); + assert!(calls + .iter() + .any(|(p, a)| p == "apt" && a == &["install", "--only-upgrade", "-y", "nginx"])); + } + + #[tokio::test] + async fn package_upgrade_inverse_returns_failure() { + let cap = PackageUpgrade::new(make_executor()); + let result = cap.invoke_inverse(json!({ "all": true }), &ctx()).await; + // has_inverse is true, but auto-rollback is intentionally unsupported: + // it returns a non-recoverable failure with manual guidance. + match result { + Some(CapabilityResult::Failure { error, recoverable }) => { + assert!(!recoverable); + assert!(error.contains("rollback") || error.contains("downgrade")); + } + other => panic!("expected Some(Failure), got {other:?}"), + } + } + #[test] fn package_list_no_filter() { let cap = PackageList::new(make_executor()); diff --git a/sentinel-fleet/src/fleet_exec.rs b/sentinel-fleet/src/fleet_exec.rs new file mode 100644 index 0000000..97a2bda --- /dev/null +++ b/sentinel-fleet/src/fleet_exec.rs @@ -0,0 +1,439 @@ +//! SSH-based multi-host orchestration. +//! +//! This module provides a lightweight, agentless way to run a single +//! capability across many hosts in parallel over SSH, aggregating the +//! per-host [`CapabilityResult`]s. +//! +//! The real transport ([`SshHostExecutor`]) shells out to the system `ssh` +//! binary via `tokio::process` — no native SSH library is required. All +//! orchestration logic (parallel fan-out, result aggregation, error +//! isolation) is decoupled from the transport behind the [`HostExecutor`] +//! trait, so it can be exercised in tests with a mock executor. + +use std::collections::HashMap; +use std::sync::Arc; +use std::time::Duration; + +use async_trait::async_trait; +use serde_json::{json, Value}; +use tokio::process::Command; +use tokio::task::JoinSet; +use tracing::{debug, warn}; + +use sentinel_core::{CapabilityResult, ExecutionContext}; + +/// Connection details for a single fleet host. +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct HostConfig { + /// Hostname or IP address to connect to. + pub hostname: String, + /// SSH login user. + pub user: String, + /// SSH port (defaults to 22). + pub port: u16, + /// Optional path to a private key (`ssh -i`). + pub key_path: Option, +} + +impl HostConfig { + /// Construct a host config with port 22 and no explicit key. + pub fn new(hostname: impl Into, user: impl Into) -> Self { + Self { + hostname: hostname.into(), + user: user.into(), + port: 22, + key_path: None, + } + } + + /// Parse a host spec of the form `[user@]hostname[:port]`. + /// + /// When the user is omitted it defaults to `root`; when the port is + /// omitted it defaults to `22`. + pub fn parse(spec: &str) -> Self { + let (user, rest) = match spec.split_once('@') { + Some((u, r)) => (u.to_string(), r), + None => ("root".to_string(), spec), + }; + let (hostname, port) = match rest.rsplit_once(':') { + Some((h, p)) => (h.to_string(), p.parse::().unwrap_or(22)), + None => (rest.to_string(), 22), + }; + Self { + hostname, + user, + port, + key_path: None, + } + } +} + +/// A fleet — an ordered collection of [`HostConfig`]s. +#[derive(Debug, Clone, Default)] +pub struct FleetConfig { + /// Hosts the capability will run against. + pub hosts: Vec, +} + +impl FleetConfig { + /// Build a config from a list of `[user@]host[:port]` specs. + pub fn from_specs(specs: I) -> Self + where + I: IntoIterator, + S: AsRef, + { + Self { + hosts: specs + .into_iter() + .map(|s| HostConfig::parse(s.as_ref())) + .collect(), + } + } + + /// Number of hosts in the fleet. + pub fn len(&self) -> usize { + self.hosts.len() + } + + /// Whether the fleet has no hosts. + pub fn is_empty(&self) -> bool { + self.hosts.is_empty() + } +} + +/// The aggregated outcome of a fleet execution: `hostname → result`. +pub type FleetResult = HashMap; + +/// Transport abstraction for executing a capability on a single host. +/// +/// The production implementation is [`SshHostExecutor`]; tests provide a mock. +#[async_trait] +pub trait HostExecutor: Send + Sync { + /// Execute capability `cap_id` with `args` on `host`, returning its result. + /// + /// Implementations MUST NOT panic on remote failure — a transport or + /// remote error should be returned as a [`CapabilityResult::Failure`] so + /// that one bad host never aborts the rest of the fleet. + async fn execute( + &self, + host: &HostConfig, + cap_id: &str, + args: &HashMap, + ctx: &ExecutionContext, + ) -> CapabilityResult; +} + +/// Production [`HostExecutor`] that runs the remote Sentinel agent over SSH. +/// +/// For each host it invokes: +/// +/// ```text +/// ssh -o BatchMode=yes -o ConnectTimeout=10 -p [-i ] \ +/// @ " agent-exec ''" +/// ``` +/// +/// A zero exit status whose stdout parses as a [`CapabilityResult`] is +/// returned verbatim; otherwise stdout is wrapped as a success payload. A +/// non-zero status (or a spawn/timeout failure) becomes a recoverable +/// [`CapabilityResult::Failure`]. +pub struct SshHostExecutor { + /// Name of the Sentinel binary on the remote host. + remote_binary: String, +} + +impl SshHostExecutor { + /// Create an executor that invokes `remote_binary` on each host. + pub fn new(remote_binary: impl Into) -> Self { + Self { + remote_binary: remote_binary.into(), + } + } +} + +impl Default for SshHostExecutor { + fn default() -> Self { + Self::new("sentinel") + } +} + +/// Build the `ssh` argument vector for a host and remote command. +/// +/// Pulled out as a pure function so the argv construction can be unit-tested +/// without invoking `ssh`. +fn build_ssh_args(host: &HostConfig, remote_cmd: &str) -> Vec { + let mut args = vec![ + "-o".to_string(), + "BatchMode=yes".to_string(), + "-o".to_string(), + "StrictHostKeyChecking=accept-new".to_string(), + "-o".to_string(), + "ConnectTimeout=10".to_string(), + "-p".to_string(), + host.port.to_string(), + ]; + if let Some(key) = &host.key_path { + args.push("-i".to_string()); + args.push(key.clone()); + } + args.push(format!("{}@{}", host.user, host.hostname)); + args.push(remote_cmd.to_string()); + args +} + +/// Single-quote a string for safe inclusion in a remote shell command. +fn shell_quote(s: &str) -> String { + // Close the quote, emit an escaped quote, reopen — the POSIX-safe idiom. + format!("'{}'", s.replace('\'', "'\\''")) +} + +#[async_trait] +impl HostExecutor for SshHostExecutor { + async fn execute( + &self, + host: &HostConfig, + cap_id: &str, + args: &HashMap, + ctx: &ExecutionContext, + ) -> CapabilityResult { + let args_json = serde_json::to_string(args).unwrap_or_else(|_| "{}".to_string()); + let remote_cmd = format!( + "{} agent-exec {} {}", + self.remote_binary, + cap_id, + shell_quote(&args_json) + ); + let ssh_args = build_ssh_args(host, &remote_cmd); + + debug!(host = %host.hostname, %cap_id, "fleet: ssh exec"); + + let timeout = Duration::from_millis(ctx.timeout_ms.max(1)); + let run = Command::new("ssh").args(&ssh_args).output(); + + let output = match tokio::time::timeout(timeout, run).await { + Ok(Ok(o)) => o, + Ok(Err(e)) => { + return CapabilityResult::failure( + format!("ssh to {} failed to spawn: {e}", host.hostname), + true, + ) + } + Err(_) => { + return CapabilityResult::failure( + format!("ssh to {} timed out after {} ms", host.hostname, ctx.timeout_ms), + true, + ) + } + }; + + if output.status.success() { + let stdout = String::from_utf8_lossy(&output.stdout).trim().to_string(); + // Prefer a structured remote result; otherwise wrap raw stdout. + match serde_json::from_str::(&stdout) { + Ok(result) => result, + Err(_) => CapabilityResult::success(json!({ + "host": host.hostname, + "stdout": stdout, + })), + } + } else { + let stderr = String::from_utf8_lossy(&output.stderr).trim().to_string(); + CapabilityResult::failure( + format!("remote execution on {} failed: {stderr}", host.hostname), + true, + ) + } + } +} + +/// Run `cap_id` on every host in `config` in parallel over SSH and aggregate +/// the results by hostname. +/// +/// This is the production entry point; it uses [`SshHostExecutor`]. Use +/// [`execute_on_fleet_with`] to inject a custom (e.g. mock) transport. +pub async fn execute_on_fleet( + config: &FleetConfig, + cap_id: &str, + args: &HashMap, + ctx: &ExecutionContext, +) -> FleetResult { + execute_on_fleet_with(Arc::new(SshHostExecutor::default()), config, cap_id, args, ctx).await +} + +/// Like [`execute_on_fleet`] but with a caller-supplied [`HostExecutor`]. +/// +/// Each host runs on its own task; a failure (or panic) on one host never +/// prevents the others from completing. +pub async fn execute_on_fleet_with( + executor: Arc, + config: &FleetConfig, + cap_id: &str, + args: &HashMap, + ctx: &ExecutionContext, +) -> FleetResult { + let mut set: JoinSet<(String, CapabilityResult)> = JoinSet::new(); + + for host in &config.hosts { + let executor = Arc::clone(&executor); + let host = host.clone(); + let cap_id = cap_id.to_string(); + let args = args.clone(); + let ctx = ctx.clone(); + set.spawn(async move { + let result = executor.execute(&host, &cap_id, &args, &ctx).await; + (host.hostname, result) + }); + } + + let mut results = FleetResult::new(); + while let Some(joined) = set.join_next().await { + match joined { + Ok((hostname, result)) => { + results.insert(hostname, result); + } + Err(e) => { + // A task panicked — record it rather than dropping the host. + warn!(error = %e, "fleet: host task panicked"); + } + } + } + results +} + +// ── Tests ─────────────────────────────────────────────────────────────────── + +#[cfg(test)] +mod tests { + use super::*; + use std::sync::atomic::{AtomicUsize, Ordering}; + use uuid::Uuid; + + fn ctx() -> ExecutionContext { + ExecutionContext::new(Uuid::new_v4(), "fleet") + } + + fn fleet(specs: &[&str]) -> FleetConfig { + FleetConfig::from_specs(specs.iter().copied()) + } + + /// Mock transport: returns a success echoing the host, and counts calls. + struct EchoExecutor { + calls: Arc, + } + + #[async_trait] + impl HostExecutor for EchoExecutor { + async fn execute( + &self, + host: &HostConfig, + cap_id: &str, + _args: &HashMap, + _ctx: &ExecutionContext, + ) -> CapabilityResult { + self.calls.fetch_add(1, Ordering::SeqCst); + CapabilityResult::success(json!({ + "host": host.hostname, + "cap": cap_id, + })) + } + } + + /// Mock transport that fails for one specific host, succeeds for the rest. + struct FlakyExecutor { + fail_host: String, + } + + #[async_trait] + impl HostExecutor for FlakyExecutor { + async fn execute( + &self, + host: &HostConfig, + _cap_id: &str, + _args: &HashMap, + _ctx: &ExecutionContext, + ) -> CapabilityResult { + if host.hostname == self.fail_host { + CapabilityResult::failure(format!("{} is unreachable", host.hostname), true) + } else { + CapabilityResult::success(json!({ "host": host.hostname })) + } + } + } + + #[test] + fn host_config_parse_variants() { + let h = HostConfig::parse("web-01"); + assert_eq!(h.user, "root"); + assert_eq!(h.hostname, "web-01"); + assert_eq!(h.port, 22); + + let h = HostConfig::parse("deploy@db-02:2222"); + assert_eq!(h.user, "deploy"); + assert_eq!(h.hostname, "db-02"); + assert_eq!(h.port, 2222); + } + + #[test] + fn ssh_args_include_port_key_and_target() { + let mut host = HostConfig::new("h1", "ops"); + host.port = 2200; + host.key_path = Some("/keys/id_ed25519".to_string()); + let args = build_ssh_args(&host, "sentinel agent-exec disk_usage '{}'"); + assert!(args.windows(2).any(|w| w == ["-p", "2200"])); + assert!(args.windows(2).any(|w| w == ["-i", "/keys/id_ed25519"])); + assert_eq!(args[args.len() - 2], "ops@h1"); + assert_eq!(args[args.len() - 1], "sentinel agent-exec disk_usage '{}'"); + } + + #[tokio::test] + async fn single_host_round_trip() { + let calls = Arc::new(AtomicUsize::new(0)); + let exec = Arc::new(EchoExecutor { calls: Arc::clone(&calls) }); + let config = fleet(&["app-01"]); + + let results = + execute_on_fleet_with(exec, &config, "system_metrics", &HashMap::new(), &ctx()).await; + + assert_eq!(results.len(), 1); + assert_eq!(calls.load(Ordering::SeqCst), 1); + let r = &results["app-01"]; + assert!(r.is_success()); + if let CapabilityResult::Success { output } = r { + assert_eq!(output["host"], "app-01"); + assert_eq!(output["cap"], "system_metrics"); + } + } + + #[tokio::test] + async fn multi_host_parallel_all_succeed() { + let calls = Arc::new(AtomicUsize::new(0)); + let exec = Arc::new(EchoExecutor { calls: Arc::clone(&calls) }); + let config = fleet(&["h1", "h2", "h3", "h4", "h5"]); + + let results = + execute_on_fleet_with(exec, &config, "process_list", &HashMap::new(), &ctx()).await; + + assert_eq!(results.len(), 5); + assert_eq!(calls.load(Ordering::SeqCst), 5); + for host in ["h1", "h2", "h3", "h4", "h5"] { + assert!(results[host].is_success(), "{host} should have succeeded"); + } + } + + #[tokio::test] + async fn error_isolation_one_host_fails() { + let exec = Arc::new(FlakyExecutor { fail_host: "bad-host".to_string() }); + let config = fleet(&["good-1", "bad-host", "good-2"]); + + let results = + execute_on_fleet_with(exec, &config, "disk_usage", &HashMap::new(), &ctx()).await; + + // Every host is represented; the failure is isolated to one entry. + assert_eq!(results.len(), 3); + assert!(results["good-1"].is_success()); + assert!(results["good-2"].is_success()); + assert!(results["bad-host"].is_failure()); + if let CapabilityResult::Failure { error, .. } = &results["bad-host"] { + assert!(error.contains("unreachable")); + } + } +} diff --git a/sentinel-fleet/src/lib.rs b/sentinel-fleet/src/lib.rs index 4be85be..a904c26 100644 --- a/sentinel-fleet/src/lib.rs +++ b/sentinel-fleet/src/lib.rs @@ -1,12 +1,17 @@ pub mod agent_client; pub mod controller; pub mod error; +pub mod fleet_exec; pub mod host; pub mod staged_rollout; pub mod tls; pub mod topology; pub use error::FleetError; +pub use fleet_exec::{ + execute_on_fleet, execute_on_fleet_with, FleetConfig, FleetResult, HostConfig, HostExecutor, + SshHostExecutor, +}; pub use host::{Host, HostId, HostStatus}; pub use staged_rollout::{CanaryConfig, RolloutStage, RolloutStatus, StagedRollout}; pub use tls::FleetTls; diff --git a/sentinel-policy/src/evaluator.rs b/sentinel-policy/src/evaluator.rs index c0ffc2d..a085cc8 100644 --- a/sentinel-policy/src/evaluator.rs +++ b/sentinel-policy/src/evaluator.rs @@ -177,6 +177,14 @@ impl PolicyEvaluator { &self.kill_switch } + /// Return the evaluator's rules, sorted ascending by priority. + /// + /// Useful for introspection — e.g. rendering the active policy in a CLI or + /// TUI — without exposing mutable access to the internal vector. + pub fn rules(&self) -> &[PolicyRule] { + &self.rules + } + // ── Private helpers ─────────────────────────────────────────────────────── fn check_kill_switch(&self, req: &PolicyRequest) -> Option { @@ -469,6 +477,19 @@ mod tests { assert!(not_found.is_none()); } + #[test] + fn rules_getter_returns_sorted_rules() { + let ks = KillSwitch::new(); + // Pass rules out of priority order; constructor sorts them ascending. + let evaluator = + PolicyEvaluator::new(vec![allow_all_low_rule(), deny_critical_rule()], ks, vec![]); + let rules = evaluator.rules(); + assert_eq!(rules.len(), 2); + // deny-critical has priority 10, allow-low has priority 100. + assert_eq!(rules[0].id, "deny-critical"); + assert_eq!(rules[1].id, "allow-low"); + } + #[test] fn get_applicable_rules() { let ks = KillSwitch::new(); diff --git a/sentinel-tui/src/app.rs b/sentinel-tui/src/app.rs index f6bf879..5b038bc 100644 --- a/sentinel-tui/src/app.rs +++ b/sentinel-tui/src/app.rs @@ -1,4 +1,5 @@ use serde::{Deserialize, Serialize}; +use tokio::sync::{mpsc, oneshot}; use uuid::Uuid; use sentinel_core::{RiskTier, SessionPhase}; @@ -98,6 +99,40 @@ pub enum ApprovalDecision { Reject { reason: String }, } +// ── Interactive per-step approval ───────────────────────────────────────────── + +/// The operator's answer to a blocking per-step approval prompt. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum ApprovalOutcome { + /// Approve this step — the agent may proceed. + Approve, + /// Abort — the agent must not run this step (and should stop the plan). + Abort, +} + +/// A request from the agent for interactive approval of a single step. +/// +/// When the policy engine returns `RequiresApproval`, the agent sends one of +/// these on the approval channel instead of blocking on stdin. The TUI +/// surfaces it as a modal and replies with an [`ApprovalOutcome`] on +/// `responder`. +#[derive(Debug)] +pub struct ApprovalRequest { + /// The step awaiting approval (capability, risk, args, description). + pub step: PlanStep, + /// One-shot channel the TUI uses to send the operator's decision back. + pub responder: oneshot::Sender, +} + +/// High-level interaction state of the TUI. +#[derive(Debug)] +pub enum AppState { + /// Normal browsing/editing — tabs and inputs are active. + Normal, + /// A modal is blocking input, awaiting approval of the contained step. + ApprovingPlan(PlanStep), +} + /// Severity level for TUI log entries. #[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)] pub enum LogLevel { @@ -317,6 +352,12 @@ pub struct App { pub status_message: Option, /// Cursor position within the goal input field. pub input_cursor: usize, + /// Current interaction state — drives modal/blocking input handling. + pub state: AppState, + /// Channel responder for the in-flight approval modal, if any. + approval_responder: Option>, + /// Channel on which the agent emits approval requests for the TUI to poll. + approval_rx: Option>, } impl Default for App { @@ -336,9 +377,92 @@ impl App { should_quit: false, status_message: None, input_cursor: 0, + state: AppState::Normal, + approval_responder: None, + approval_rx: None, } } + // ── Interactive approval ────────────────────────────────────────────────── + + /// Attach the channel on which the agent will send approval requests. + pub fn set_approval_channel(&mut self, rx: mpsc::Receiver) { + self.approval_rx = Some(rx); + } + + /// Returns `true` when a blocking approval modal is active. + pub fn is_approving(&self) -> bool { + matches!(self.state, AppState::ApprovingPlan(_)) + } + + /// The step currently awaiting approval, if any. + pub fn pending_approval_step(&self) -> Option<&PlanStep> { + match &self.state { + AppState::ApprovingPlan(step) => Some(step), + AppState::Normal => None, + } + } + + /// Enter the approval modal for `step`, replying on `responder`. + /// + /// Used by [`poll_approval`](Self::poll_approval); also exposed for direct + /// wiring and tests. + pub fn begin_approval( + &mut self, + step: PlanStep, + responder: oneshot::Sender, + ) { + self.status_message = Some(format!( + "Approval required for '{}' (risk {:?}). Press y to approve, n/Esc to abort.", + step.capability_id, step.risk_tier + )); + self.state = AppState::ApprovingPlan(step); + self.approval_responder = Some(responder); + } + + /// Non-blocking check for an incoming approval request. When one arrives + /// and no modal is already active, enter the approval modal state. + /// + /// Call this on every tick so requests surface promptly. + pub fn poll_approval(&mut self) { + if self.is_approving() { + return; + } + let Some(rx) = self.approval_rx.as_mut() else { + return; + }; + if let Ok(req) = rx.try_recv() { + self.begin_approval(req.step, req.responder); + } + } + + /// Resolve the active approval modal, replying to the agent and returning + /// to [`AppState::Normal`]. A no-op if no modal is active. + pub fn submit_approval(&mut self, outcome: ApprovalOutcome) { + if !self.is_approving() { + return; + } + if let Some(responder) = self.approval_responder.take() { + // The agent may have given up waiting; ignore a closed channel. + let _ = responder.send(outcome); + } + if let Some(s) = &mut self.session { + match outcome { + ApprovalOutcome::Approve => { + s.log(LogLevel::Info, "Step approved by operator.") + } + ApprovalOutcome::Abort => { + s.log(LogLevel::Warn, "Step aborted by operator.") + } + } + } + self.status_message = Some(match outcome { + ApprovalOutcome::Approve => "Step approved.".to_string(), + ApprovalOutcome::Abort => "Plan aborted by operator.".to_string(), + }); + self.state = AppState::Normal; + } + // ── Navigation ──────────────────────────────────────────────────────────── pub fn next_tab(&mut self) { @@ -705,4 +829,74 @@ mod tests { pv.toggle_expanded(); assert!(!pv.steps[0].expanded); } + + // ── Interactive approval ────────────────────────────────────────────────── + + fn approval_step() -> PlanStep { + PlanStep::new( + "Restart nginx", + "service_restart", + serde_json::json!({ "service": "nginx" }), + RiskTier::High, + ) + } + + #[tokio::test] + async fn poll_approval_enters_modal_state() { + let mut app = App::new(); + assert!(!app.is_approving()); + + let (tx, rx) = mpsc::channel::(4); + app.set_approval_channel(rx); + + // Agent emits an approval request. + let (responder, _resp_rx) = oneshot::channel(); + tx.send(ApprovalRequest { + step: approval_step(), + responder, + }) + .await + .unwrap(); + + app.poll_approval(); + + assert!(app.is_approving()); + let step = app.pending_approval_step().expect("step present"); + assert_eq!(step.capability_id, "service_restart"); + assert_eq!(step.risk_tier, RiskTier::High); + } + + #[tokio::test] + async fn submit_approval_approve_sends_outcome_and_resets() { + let mut app = App::new(); + let (responder, resp_rx) = oneshot::channel(); + app.begin_approval(approval_step(), responder); + assert!(app.is_approving()); + + app.submit_approval(ApprovalOutcome::Approve); + + assert!(!app.is_approving()); + assert!(matches!(app.state, AppState::Normal)); + assert_eq!(resp_rx.await.unwrap(), ApprovalOutcome::Approve); + } + + #[tokio::test] + async fn submit_approval_abort_sends_outcome() { + let mut app = App::new(); + let (responder, resp_rx) = oneshot::channel(); + app.begin_approval(approval_step(), responder); + + app.submit_approval(ApprovalOutcome::Abort); + + assert!(!app.is_approving()); + assert_eq!(resp_rx.await.unwrap(), ApprovalOutcome::Abort); + } + + #[test] + fn submit_approval_is_noop_when_not_approving() { + let mut app = App::new(); + // Should not panic or change state. + app.submit_approval(ApprovalOutcome::Approve); + assert!(matches!(app.state, AppState::Normal)); + } } diff --git a/sentinel-tui/src/event_handler.rs b/sentinel-tui/src/event_handler.rs index 85029dc..78b5b35 100644 --- a/sentinel-tui/src/event_handler.rs +++ b/sentinel-tui/src/event_handler.rs @@ -1,7 +1,7 @@ use crossterm::event::{KeyCode, KeyEvent, KeyModifiers}; use uuid::Uuid; -use crate::app::{App, SessionUpdate, StepStatus, Tab}; +use crate::app::{App, ApprovalOutcome, SessionUpdate, StepStatus, Tab}; /// Events that the TUI event loop dispatches. #[derive(Debug, Clone)] @@ -26,11 +26,14 @@ pub async fn handle_events( app: &mut App, event: AppEvent, ) -> Result<(), anyhow::Error> { + // Surface any pending approval request from the agent before handling the + // event, so a freshly-arrived request blocks input on this same pass. + app.poll_approval(); + match event { AppEvent::Key(key) => handle_key(app, key), AppEvent::Tick => { - // Periodic tick — nothing to do for now. Could be used to refresh - // external state or animate progress indicators. + // Periodic tick — nothing to do beyond the approval poll above. } AppEvent::SessionUpdate(update) => { app.apply_session_update(update); @@ -47,6 +50,20 @@ pub async fn handle_events( /// Handle a raw keyboard event. fn handle_key(app: &mut App, key: KeyEvent) { + // A blocking approval modal owns all input while active: only y/n/Esc act. + if app.is_approving() { + match key.code { + KeyCode::Char('y') | KeyCode::Char('Y') => { + app.submit_approval(ApprovalOutcome::Approve); + } + KeyCode::Char('n') | KeyCode::Char('N') | KeyCode::Esc => { + app.submit_approval(ApprovalOutcome::Abort); + } + _ => {} + } + return; + } + // Ctrl-C or Ctrl-Q always quits, regardless of tab. if key.modifiers.contains(KeyModifiers::CONTROL) { match key.code { @@ -285,6 +302,57 @@ mod tests { assert_eq!(app.log_scroll, 2); } + // ── Interactive approval modal ──────────────────────────────────────────── + + fn approving_app() -> (App, tokio::sync::oneshot::Receiver) { + use crate::app::PlanStep; + use sentinel_core::RiskTier; + let mut app = App::new(); + let (tx, rx) = tokio::sync::oneshot::channel(); + let step = PlanStep::new( + "Restart nginx", + "service_restart", + serde_json::json!({ "service": "nginx" }), + RiskTier::High, + ); + app.begin_approval(step, tx); + (app, rx) + } + + #[test] + fn y_key_approves_during_modal() { + let (mut app, mut rx) = approving_app(); + handle_key(&mut app, key(KeyCode::Char('y'))); + assert!(!app.is_approving()); + assert_eq!(rx.try_recv().unwrap(), ApprovalOutcome::Approve); + } + + #[test] + fn n_key_aborts_during_modal() { + let (mut app, mut rx) = approving_app(); + handle_key(&mut app, key(KeyCode::Char('n'))); + assert!(!app.is_approving()); + assert_eq!(rx.try_recv().unwrap(), ApprovalOutcome::Abort); + } + + #[test] + fn esc_aborts_during_modal() { + let (mut app, mut rx) = approving_app(); + handle_key(&mut app, key(KeyCode::Esc)); + assert!(!app.is_approving()); + assert_eq!(rx.try_recv().unwrap(), ApprovalOutcome::Abort); + } + + #[test] + fn other_key_does_nothing_during_modal() { + let (mut app, _rx) = approving_app(); + handle_key(&mut app, key(KeyCode::Char('x'))); + // Still blocking on approval; modal not dismissed. + assert!(app.is_approving()); + handle_key(&mut app, key(KeyCode::Tab)); + assert!(app.is_approving()); + } + // ── Async handle_events ─────────────────────────────────────────────────── #[tokio::test] diff --git a/sentinel-tui/src/main.rs b/sentinel-tui/src/main.rs index f76fae4..d5b835a 100644 --- a/sentinel-tui/src/main.rs +++ b/sentinel-tui/src/main.rs @@ -17,9 +17,10 @@ use sentinel_agent_llm::{ }; use sentinel_audit::AuditLog; use sentinel_capabilities::all_capabilities; -use sentinel_core::ApprovalDecision; +use sentinel_core::{ApprovalDecision, CapabilityResult, ExecutionContext}; use sentinel_exec::RealCommandExecutor; -use sentinel_policy::default_policy; +use sentinel_fleet::{execute_on_fleet, FleetConfig}; +use sentinel_policy::{default_policy, RuleCondition}; use tokio::sync::Mutex; use uuid::Uuid; @@ -81,6 +82,21 @@ enum Commands { VerifyAudit { path: std::path::PathBuf, }, + /// Run a capability across multiple hosts in parallel over SSH + Fleet { + /// Operational goal / label for this fleet run + #[arg(help = "Goal or label for this fleet run")] + goal: String, + /// Comma-separated host specs: [user@]host[:port] + #[arg(long, value_delimiter = ',')] + hosts: Vec, + /// Capability id to run on every host + #[arg(long, default_value = "system_metrics")] + capability: String, + /// JSON arguments object for the capability + #[arg(long, default_value = "{}")] + args: String, + }, /// Launch the interactive TUI Tui { /// Target host @@ -104,6 +120,12 @@ async fn main() -> Result<()> { Commands::Capabilities => list_capabilities(), Commands::Policy => show_policy(), Commands::VerifyAudit { path } => verify_audit(&path)?, + Commands::Fleet { + goal, + hosts, + capability, + args, + } => run_fleet(goal, hosts, capability, args).await?, Commands::Run { goal, host, @@ -312,6 +334,68 @@ async fn run_agent( Ok(()) } +/// Run a capability across multiple hosts in parallel over SSH and print +/// per-host results. +async fn run_fleet( + goal: String, + hosts: Vec, + capability: String, + args: String, +) -> Result<()> { + if hosts.is_empty() { + return Err(anyhow::anyhow!( + "no hosts specified; pass --hosts host1,host2[,...]" + )); + } + + let parsed_args: std::collections::HashMap = + serde_json::from_str(&args) + .map_err(|e| anyhow::anyhow!("--args must be a JSON object: {e}"))?; + + let config = FleetConfig::from_specs(&hosts); + let session_id = Uuid::new_v4(); + let ctx = ExecutionContext::new(session_id, "fleet"); + + println!("Fleet run: {goal}"); + println!("Capability : {capability}"); + println!("Hosts ({}) : {}", config.len(), hosts.join(", ")); + println!(); + + let results = execute_on_fleet(&config, &capability, &parsed_args, &ctx).await; + + let mut hostnames: Vec<&String> = results.keys().collect(); + hostnames.sort(); + + let mut ok = 0usize; + let mut failed = 0usize; + for hostname in hostnames { + match &results[hostname] { + CapabilityResult::Success { output } => { + ok += 1; + println!("✔ {hostname}: success"); + if let Ok(pretty) = serde_json::to_string(output) { + println!(" {pretty}"); + } + } + CapabilityResult::Failure { error, .. } => { + failed += 1; + println!("x {hostname}: FAILED — {error}"); + } + CapabilityResult::DryRun { predicted_effect } => { + ok += 1; + println!("• {hostname}: dry-run"); + if let Ok(pretty) = serde_json::to_string(predicted_effect) { + println!(" {pretty}"); + } + } + } + } + + println!(); + println!("Fleet summary: {ok} succeeded, {failed} failed across {} host(s).", config.len()); + Ok(()) +} + fn list_capabilities() { let executor = Arc::new(RealCommandExecutor); let caps = all_capabilities(executor); @@ -330,19 +414,82 @@ fn list_capabilities() { } fn show_policy() { - let _evaluator = default_policy(); - println!("Default Sentinel policy (deny-by-default):"); - println!("{:-<60}", ""); - println!(" 10 deny-critical Deny ALL Critical-risk actions"); - println!(" 20 deny-high-mutating Deny High-risk mutating actions"); - println!(" 50 require-approval-high High-risk actions require approval"); - println!( - " 100 require-approval-medium Mutating Medium-risk requires approval" - ); + let evaluator = default_policy(); + let rules = evaluator.rules(); + println!( - " 200 allow-read-low-medium Allow read-only Low/Medium without approval" + "Default Sentinel policy (deny-by-default) — {} rule(s):", + rules.len() ); - println!(" 300 allow-low-risk Allow all Low-risk actions"); + println!("{:-<78}", ""); + println!(" {:<5} {:<33} {:<16} Conditions", "Prio", "Rule ID", "Effect"); + println!("{:-<78}", ""); + + for rule in rules { + let conditions = if rule.conditions.is_empty() { + "".to_string() + } else { + rule.conditions + .iter() + .map(describe_condition) + .collect::>() + .join(" AND ") + }; + let effect = if rule.enabled { + format!("{:?}", rule.effect) + } else { + format!("{:?} (disabled)", rule.effect) + }; + println!( + " {:<5} {:<33} {:<16} {}", + rule.priority, rule.id, effect, conditions + ); + println!(" {}", rule.description); + } + + println!("{:-<78}", ""); + println!("Rules are evaluated in ascending priority order; the first match wins."); + println!("Any request not matched by an Allow/AuditOnly rule is denied by default."); +} + +/// Render a [`RuleCondition`] as a concise, human-readable predicate string. +fn describe_condition(cond: &RuleCondition) -> String { + match cond { + RuleCondition::CapabilityId { matches } => format!("capability_id == \"{matches}\""), + RuleCondition::CapabilityIdIn { ids } => { + format!("capability_id in [{}]", ids.join(", ")) + } + RuleCondition::RiskTierAtLeast { tier } => format!("risk >= {tier:?}"), + RuleCondition::RiskTierExactly { tier } => format!("risk == {tier:?}"), + RuleCondition::TargetHost { pattern } => format!("host matches \"{pattern}\""), + RuleCondition::ArgValueContains { path, value } => { + format!("args.{path} contains \"{value}\"") + } + RuleCondition::TimeWindow { + start_hour, + end_hour, + days, + } => format!("time in [{start_hour:02}:00, {end_hour:02}:00) days={days:?}"), + RuleCondition::CapabilityKindIs { kind } => format!("kind == {kind:?}"), + RuleCondition::SessionPhase { phase } => format!("phase == \"{phase}\""), + RuleCondition::Not { condition } => format!("NOT ({})", describe_condition(condition)), + RuleCondition::And { conditions } => format!( + "({})", + conditions + .iter() + .map(describe_condition) + .collect::>() + .join(" AND ") + ), + RuleCondition::Or { conditions } => format!( + "({})", + conditions + .iter() + .map(describe_condition) + .collect::>() + .join(" OR ") + ), + } } fn verify_audit(path: &std::path::Path) -> Result<()> { diff --git a/sentinel-tui/src/ui.rs b/sentinel-tui/src/ui.rs index fda8ac8..aee3442 100644 --- a/sentinel-tui/src/ui.rs +++ b/sentinel-tui/src/ui.rs @@ -3,14 +3,14 @@ use ratatui::{ style::{Color, Modifier, Style}, text::{Line, Span}, widgets::{ - Block, Borders, Cell, Gauge, List, ListItem, Paragraph, Row, Table, Tabs, + Block, Borders, Cell, Clear, Gauge, List, ListItem, Paragraph, Row, Table, Tabs, Wrap, }, Frame, }; use sentinel_core::RiskTier; -use crate::app::{App, LogLevel, Session, StepStatus, Tab}; +use crate::app::{App, LogLevel, PlanStep, Session, StepStatus, Tab}; /// Entry point for rendering the entire TUI. pub fn draw(frame: &mut Frame, app: &App) { @@ -36,6 +36,99 @@ pub fn draw(frame: &mut Frame, app: &App) { } render_status_bar(frame, chunks[3], app); + + // A pending approval renders as a blocking modal over everything else. + if let Some(step) = app.pending_approval_step() { + render_approval_modal(frame, frame.area(), step); + } +} + +/// Compute a `Rect` centered within `area`, sized as a percentage of it. +fn centered_rect(percent_x: u16, percent_y: u16, area: Rect) -> Rect { + let vertical = Layout::default() + .direction(Direction::Vertical) + .constraints([ + Constraint::Percentage((100 - percent_y) / 2), + Constraint::Percentage(percent_y), + Constraint::Percentage((100 - percent_y) / 2), + ]) + .split(area); + + Layout::default() + .direction(Direction::Horizontal) + .constraints([ + Constraint::Percentage((100 - percent_x) / 2), + Constraint::Percentage(percent_x), + Constraint::Percentage((100 - percent_x) / 2), + ]) + .split(vertical[1])[1] +} + +/// Render the blocking approval modal for a step requiring operator approval. +fn render_approval_modal(frame: &mut Frame, area: Rect, step: &PlanStep) { + let popup = centered_rect(60, 50, area); + + // Clear whatever is underneath so the modal stands alone. + frame.render_widget(Clear, popup); + + let args = serde_json::to_string_pretty(&step.args) + .unwrap_or_else(|_| step.args.to_string()); + + let mut lines = vec![ + Line::from(Span::styled( + "This step requires your approval before it can run.", + Style::default().fg(Color::Yellow), + )), + Line::from(""), + Line::from(vec![ + Span::raw("Capability : "), + Span::styled( + step.capability_id.clone(), + Style::default().fg(Color::Cyan).add_modifier(Modifier::BOLD), + ), + ]), + Line::from(vec![ + Span::raw("Risk tier : "), + Span::styled( + step.risk_tier.to_string(), + Style::default() + .fg(risk_color(step.risk_tier)) + .add_modifier(Modifier::BOLD), + ), + ]), + Line::from(vec![ + Span::raw("Action : "), + Span::raw(step.description.clone()), + ]), + Line::from(""), + Line::from(Span::styled("Arguments:", Style::default().fg(Color::Gray))), + ]; + for arg_line in args.lines() { + lines.push(Line::from(Span::styled( + format!(" {arg_line}"), + Style::default().fg(Color::White), + ))); + } + lines.push(Line::from("")); + lines.push(Line::from(Span::styled( + "Approve? [y] yes [n / Esc] abort", + Style::default().fg(Color::White).add_modifier(Modifier::BOLD), + ))); + + let modal = Paragraph::new(lines) + .block( + Block::default() + .title(" Step Approval Required ") + .borders(Borders::ALL) + .border_style( + Style::default() + .fg(Color::Yellow) + .add_modifier(Modifier::BOLD), + ), + ) + .wrap(Wrap { trim: false }); + + frame.render_widget(modal, popup); } fn render_header(frame: &mut Frame, area: Rect, app: &App) {