diff --git a/crates/hound-api/src/lib.rs b/crates/hound-api/src/lib.rs index 3edb824..6bd08a2 100644 --- a/crates/hound-api/src/lib.rs +++ b/crates/hound-api/src/lib.rs @@ -393,6 +393,14 @@ pub struct Settings { #[serde(default = "default_true")] pub confirm_quit: bool, + // Fleet console + /// URL of the fleet console (e.g. "https://fleet.houndav.com"). Empty = disabled. + #[serde(default)] + pub fleet_url: Option, + /// One-time enrollment token. Consumed on first boot, then ignored. + #[serde(default)] + pub fleet_token: Option, + // Global /// Master switch — when true, realtime is suspended and the tray is gray. pub paused: bool, @@ -451,6 +459,8 @@ impl Default for Settings { rootkit_enabled: true, notify_desktop: true, auto_update_signatures: true, + fleet_url: None, + fleet_token: None, paused: false, } } diff --git a/crates/houndd/src/engine.rs b/crates/houndd/src/engine.rs index 130fc08..8d45f12 100644 --- a/crates/houndd/src/engine.rs +++ b/crates/houndd/src/engine.rs @@ -87,6 +87,11 @@ pub trait ScanEngine: Send + Sync { fn cached_verdict(&self, _md: &std::fs::Metadata) -> Option> { None } + + /// `(entries, hits, misses)` from the verdict cache, for fleet telemetry. + fn verdict_cache_stats(&self) -> (usize, u64, u64) { + (0, 0, 0) + } } /// The ClamAV-backed engine: `clamscan` + `freshclam` over their diff --git a/crates/houndd/src/fleet.rs b/crates/houndd/src/fleet.rs new file mode 100644 index 0000000..20e1dc5 --- /dev/null +++ b/crates/houndd/src/fleet.rs @@ -0,0 +1,442 @@ +//! Fleet console connectivity — enrollment + periodic heartbeat. +//! +//! When `fleet_url` is set in settings the daemon registers with the +//! fleet console once (enrollment), then sends a heartbeat every 30 s +//! carrying status metrics and buffered events. The fleet console +//! responds with policy-sync data the agent can act on in future phases. +//! +//! All outbound HTTP is blocking via `ureq`, matching the rest of the +//! daemon. The fleet thread runs alongside the scheduler with its own +//! sleep loop so heartbeat cadence is independent of definition checks. + +use anyhow::{bail, Context, Result}; +use serde::{Deserialize, Serialize}; +use std::path::PathBuf; +use std::sync::{Arc, Mutex}; + +use crate::DaemonState; + +const HEARTBEAT_INTERVAL_SECS: u64 = 30; +const ENROLL_RETRY_SECS: u64 = 60; + +// ── Wire types (match console-types JSON format) ────────────────────── + +#[derive(Serialize)] +struct EnrollRequest { + token: String, + hostname: String, + os: String, + arch: String, + machine_id: String, +} + +#[derive(Deserialize)] +struct EnrollResponse { + agent_id: String, + policy_group: String, + policy: Option, + policy_version: Option, +} + +#[derive(Serialize)] +struct HeartbeatRequest { + agent_id: String, + hostname: String, + os: String, + arch: String, + machine_id: String, + state: String, + engine: String, + rules_version: String, + gate_armed: bool, + cpu_percent: f64, + rss_mb: f64, + gate_latency_us: i64, + verdicts_cached: i64, + execs_per_sec: f64, + uptime_s: i64, + policy_group: String, + policy_version: i64, + events: Vec, +} + +#[derive(Serialize)] +struct HeartbeatEvent { + #[serde(rename = "type")] + event_type: String, + severity: String, + timestamp: String, + path: String, + sha256: String, + rule: String, + verdict: String, + detail: serde_json::Value, +} + +#[derive(Deserialize)] +struct HeartbeatResponse { + ack: bool, + policy_changed: bool, + policy: Option, + policy_version: Option, +} + +// ── Persistent fleet identity ───────────────────────────────────────── + +#[derive(Serialize, Deserialize, Default)] +struct FleetIdentity { + agent_id: String, + policy_group: String, + policy_version: i64, +} + +fn identity_path() -> PathBuf { + let data = std::env::var("XDG_DATA_HOME") + .ok() + .filter(|s| !s.is_empty()) + .map(PathBuf::from) + .unwrap_or_else(|| { + let home = std::env::var("HOME").unwrap_or_else(|_| "/root".into()); + PathBuf::from(home).join(".local/share") + }); + data.join("hound").join("fleet.json") +} + +fn load_identity() -> Option { + let path = identity_path(); + let data = std::fs::read_to_string(&path).ok()?; + serde_json::from_str(&data).ok() +} + +fn save_identity(id: &FleetIdentity) -> Result<()> { + let path = identity_path(); + if let Some(parent) = path.parent() { + std::fs::create_dir_all(parent) + .with_context(|| format!("creating {}", parent.display()))?; + } + let json = serde_json::to_string_pretty(id)?; + std::fs::write(&path, json).with_context(|| format!("writing {}", path.display()))?; + Ok(()) +} + +// ── System metrics ──────────────────────────────────────────────────── + +fn hostname() -> String { + std::fs::read_to_string("/etc/hostname") + .unwrap_or_default() + .trim() + .to_string() +} + +fn machine_id() -> String { + std::fs::read_to_string("/etc/machine-id") + .unwrap_or_default() + .trim() + .to_string() +} + +fn os_name() -> String { + std::fs::read_to_string("/etc/os-release") + .ok() + .and_then(|c| { + c.lines() + .find(|l| l.starts_with("PRETTY_NAME=")) + .map(|l| l.trim_start_matches("PRETTY_NAME=").trim_matches('"').to_string()) + }) + .unwrap_or_else(|| "Linux".into()) +} + +fn arch() -> &'static str { + std::env::consts::ARCH +} + +fn self_rss_mb() -> f64 { + std::fs::read_to_string("/proc/self/status") + .ok() + .and_then(|s| { + s.lines() + .find(|l| l.starts_with("VmRSS:")) + .and_then(|l| { + l.split_whitespace() + .nth(1) + .and_then(|v| v.parse::().ok()) + }) + }) + .map(|kb| kb / 1024.0) + .unwrap_or(0.0) +} + +fn self_cpu_percent() -> f64 { + // Snapshot approach: read /proc/self/stat once per heartbeat. + // For a 30s interval this gives a rough average. A proper per-interval + // delta requires holding state between calls, which we skip for v1. + 0.0 +} + +// ── Fleet thread ────────────────────────────────────────────────────── + +/// Shared handle to the fleet thread's state, used by the status RPC. +#[derive(Clone)] +pub struct FleetHandle { + inner: Arc>, +} + +struct FleetState { + connected: bool, + agent_id: String, + last_heartbeat: Option, + last_error: String, +} + +impl FleetHandle { + fn new() -> Self { + Self { + inner: Arc::new(Mutex::new(FleetState { + connected: false, + agent_id: String::new(), + last_heartbeat: None, + last_error: String::new(), + })), + } + } + + pub fn is_connected(&self) -> bool { + self.inner.lock().unwrap().connected + } + + pub fn agent_id(&self) -> String { + self.inner.lock().unwrap().agent_id.clone() + } + + pub fn last_error(&self) -> String { + self.inner.lock().unwrap().last_error.clone() + } +} + +/// Start the fleet connectivity thread. Returns a handle for status queries. +pub fn start(state: DaemonState) -> FleetHandle { + let handle = FleetHandle::new(); + let h = handle.clone(); + + std::thread::spawn(move || { + // Wait a bit after boot so the network stack is up. + std::thread::sleep(std::time::Duration::from_secs(10)); + fleet_loop(state, h); + }); + + handle +} + +fn fleet_loop(state: DaemonState, handle: FleetHandle) { + let boot = std::time::Instant::now(); + let cached_hostname = hostname(); + let cached_os = os_name(); + let cached_machine_id = machine_id(); + + loop { + let settings = state.settings.get(); + let fleet_url = settings.fleet_url.as_deref().unwrap_or(""); + if fleet_url.is_empty() { + std::thread::sleep(std::time::Duration::from_secs(HEARTBEAT_INTERVAL_SECS)); + continue; + } + let fleet_url = fleet_url.trim_end_matches('/'); + + // Ensure we have an agent_id (enroll if needed) + let identity = match ensure_enrolled(fleet_url, &settings, &cached_hostname, &cached_os, &cached_machine_id) { + Ok(id) => id, + Err(e) => { + let msg = format!("{e:#}"); + eprintln!("fleet: enrollment failed: {msg}"); + { + let mut h = handle.inner.lock().unwrap(); + h.connected = false; + h.last_error = msg; + } + std::thread::sleep(std::time::Duration::from_secs(ENROLL_RETRY_SECS)); + continue; + } + }; + + { + let mut h = handle.inner.lock().unwrap(); + h.agent_id = identity.agent_id.clone(); + } + + // Gather events to send + let daemon_events = state.events.list(50); + let heartbeat_events: Vec = daemon_events + .iter() + .map(|e| HeartbeatEvent { + event_type: map_event_kind(&e.kind), + severity: e.severity.clone(), + timestamp: e.ts.clone(), + path: String::new(), + sha256: String::new(), + rule: String::new(), + verdict: String::new(), + detail: serde_json::json!({ "message": e.message }), + }) + .collect(); + + // Gather metrics + let gate_armed = state.gate.is_some(); + let (gate_allowed, gate_denied, _gate_timed_out) = state + .gate + .as_ref() + .map(|g| g.responder().counters()) + .unwrap_or((0, 0, 0)); + + let engine = crate::engine::engine(); + let (verdicts_cached, _, _) = engine.verdict_cache_stats(); + + let uptime_s = boot.elapsed().as_secs() as i64; + + let daemon_state = if state.settings.get().paused { + "paused" + } else if gate_armed { + "protected" + } else { + "protected" + }; + + let req = HeartbeatRequest { + agent_id: identity.agent_id.clone(), + hostname: cached_hostname.clone(), + os: cached_os.clone(), + arch: arch().to_string(), + machine_id: cached_machine_id.clone(), + state: daemon_state.to_string(), + engine: engine.name().to_string(), + rules_version: state.defs.current().version.clone(), + gate_armed, + cpu_percent: self_cpu_percent(), + rss_mb: self_rss_mb(), + gate_latency_us: 0, // TODO: expose avg latency from gate + verdicts_cached: verdicts_cached as i64, + execs_per_sec: (gate_allowed + gate_denied) as f64 / uptime_s.max(1) as f64, + uptime_s, + policy_group: identity.policy_group.clone(), + policy_version: identity.policy_version, + events: heartbeat_events, + }; + + match send_heartbeat(fleet_url, &req) { + Ok(resp) => { + let mut h = handle.inner.lock().unwrap(); + h.connected = true; + h.last_heartbeat = Some(std::time::Instant::now()); + h.last_error.clear(); + + if resp.policy_changed { + if let (Some(_policy), Some(ver)) = (&resp.policy, resp.policy_version) { + let mut id = identity.clone(); + id.policy_version = ver; + let _ = save_identity(&id); + } + } + } + Err(e) => { + let msg = format!("{e:#}"); + eprintln!("fleet: heartbeat failed: {msg}"); + let mut h = handle.inner.lock().unwrap(); + h.connected = false; + h.last_error = msg; + } + } + + std::thread::sleep(std::time::Duration::from_secs(HEARTBEAT_INTERVAL_SECS)); + } +} + +fn ensure_enrolled( + fleet_url: &str, + settings: &hound_api::Settings, + hostname: &str, + os: &str, + machine_id: &str, +) -> Result { + if let Some(id) = load_identity() { + if !id.agent_id.is_empty() { + return Ok(id); + } + } + + let token = settings + .fleet_token + .as_deref() + .filter(|t| !t.is_empty()) + .context("fleet_url set but no fleet_token for enrollment")?; + + let req = EnrollRequest { + token: token.to_string(), + hostname: hostname.to_string(), + os: os.to_string(), + arch: arch().to_string(), + machine_id: machine_id.to_string(), + }; + + let url = format!("{fleet_url}/agent/v1/enroll"); + let body = serde_json::to_string(&req)?; + let resp = ureq::post(&url) + .set("Content-Type", "application/json") + .send_string(&body) + .context("enrolling with fleet console")?; + + let resp_body = resp.into_string().context("reading enrollment response")?; + let resp: EnrollResponse = serde_json::from_str(&resp_body) + .context("parsing enrollment response")?; + + let identity = FleetIdentity { + agent_id: resp.agent_id, + policy_group: resp.policy_group, + policy_version: resp.policy_version.unwrap_or(0), + }; + save_identity(&identity)?; + eprintln!("fleet: enrolled as {}", identity.agent_id); + + Ok(identity) +} + +fn send_heartbeat(fleet_url: &str, req: &HeartbeatRequest) -> Result { + let url = format!("{fleet_url}/agent/v1/heartbeat"); + let body = serde_json::to_string(req)?; + let resp = ureq::post(&url) + .set("Content-Type", "application/json") + .send_string(&body) + .context("sending heartbeat")?; + + let resp_body = resp.into_string().context("reading heartbeat response")?; + let resp: HeartbeatResponse = serde_json::from_str(&resp_body) + .context("parsing heartbeat response")?; + if !resp.ack { + bail!("heartbeat not acknowledged"); + } + Ok(resp) +} + +/// Map houndd event kinds to the fleet console's EventType enum values. +fn map_event_kind(kind: &str) -> String { + match kind { + "gate" => "gate_verdict", + "quarantine" | "restore" => "threat_found", + "rootkit" | "persistence" => "persistence_change", + "scan" | "threat" => "scan_complete", + "update" => "rule_update", + "realtime" => "threat_found", + "supply" => "threat_found", + "info" => "agent_started", + _ => "agent_started", + } + .to_string() +} + +impl Clone for FleetIdentity { + fn clone(&self) -> Self { + Self { + agent_id: self.agent_id.clone(), + policy_group: self.policy_group.clone(), + policy_version: self.policy_version, + } + } +} diff --git a/crates/houndd/src/main.rs b/crates/houndd/src/main.rs index 40356c0..d98c6ef 100644 --- a/crates/houndd/src/main.rs +++ b/crates/houndd/src/main.rs @@ -46,6 +46,7 @@ mod defs; mod engine; mod events; mod fanotify; +mod fleet; mod license; mod native; mod peer; @@ -88,6 +89,7 @@ struct DaemonState { gate: Option>, gate_detail: std::sync::Arc, gate_paths: Vec, + fleet: Option, } fn main() -> Result<()> { @@ -109,7 +111,7 @@ fn main() -> Result<()> { // is the old behaviour and safe. open_socket_to_hound_group(&sock_path); - let state = DaemonState::boot(); + let mut state = DaemonState::boot(); // Announce any blindness at startup rather than letting it be inferred // from wrong answers later. Every serious bug found in desktop testing @@ -124,6 +126,15 @@ fn main() -> Result<()> { start_scheduler(state.clone()); + // Start fleet connectivity if configured + if state.settings.get().fleet_url.as_deref().unwrap_or("").is_empty() { + eprintln!("fleet: disabled (no fleet_url in settings)"); + } else { + let handle = fleet::start(state.clone()); + state.fleet = Some(handle); + eprintln!("fleet: thread started"); + } + eprintln!( "houndd {DAEMON_VERSION} listening on {sock} [engine: {}] (Ctrl-C to stop)", engine::engine().name() @@ -232,6 +243,7 @@ impl DaemonState { gate, gate_detail: std::sync::Arc::new(gate_detail), gate_paths, + fleet: None, } } diff --git a/crates/houndd/src/native.rs b/crates/houndd/src/native.rs index b780e6e..8d80899 100644 --- a/crates/houndd/src/native.rs +++ b/crates/houndd/src/native.rs @@ -176,6 +176,10 @@ impl ScanEngine for HoundEngine { .map(|v| v.map(|name| name.to_string())) } + fn verdict_cache_stats(&self) -> (usize, u64, u64) { + self.cache.stats() + } + fn scan_bytes(&self, bytes: &[u8]) -> Option { let set = self.rules.current(); let mut scanner = yara_x::Scanner::new(&set.rules);