fleet: add heartbeat client for fleet console connectivity

New fleet.rs module in houndd: enrollment with token validation,
periodic 30s heartbeat POST carrying agent status, metrics, and
buffered events. Identity persisted to ~/.local/share/hound/fleet.json.

Settings: fleet_url + fleet_token fields (both optional, default off).
Engine trait: verdict_cache_stats() for fleet telemetry.

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>

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<String>,
+    /// One-time enrollment token. Consumed on first boot, then ignored.
+    #[serde(default)]
+    pub fleet_token: Option<String>,
+
     // 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,
         }
     }

crates/houndd/src/engine.rs

@@ -87,6 +87,11 @@ pub trait ScanEngine: Send + Sync {
     fn cached_verdict(&self, _md: &std::fs::Metadata) -> Option<Option<String>> {
         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

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<serde_json::Value>,
+    policy_version: Option<i64>,
+}
+
+#[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<HeartbeatEvent>,
+}
+
+#[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<serde_json::Value>,
+    policy_version: Option<i64>,
+}
+
+// ── 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<FleetIdentity> {
+    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::<f64>().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<Mutex<FleetState>>,
+}
+
+struct FleetState {
+    connected: bool,
+    agent_id: String,
+    last_heartbeat: Option<std::time::Instant>,
+    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<HeartbeatEvent> = 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<FleetIdentity> {
+    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<HeartbeatResponse> {
+    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,
+        }
+    }
+}

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<std::sync::Arc<fanotify::Gate>>,
     gate_detail: std::sync::Arc<String>,
     gate_paths: Vec<String>,
+    fleet: Option<fleet::FleetHandle>,
 }
 
 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,
         }
     }
 

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<String> {
         let set = self.rules.current();
         let mut scanner = yara_x::Scanner::new(&set.rules);