diff --git a/hive-c0re/src/agent_server.rs b/hive-c0re/src/agent_server.rs index 86a405b8..0e037388 100644 --- a/hive-c0re/src/agent_server.rs +++ b/hive-c0re/src/agent_server.rs @@ -545,8 +545,20 @@ async fn handle_restart_child(coord: &Arc, agent: &str, name: &str) /// capability and routes the systemctl restart through hive-priv (which /// re-validates the name root-side). Direct, not approval-gated. async fn handle_restart_infra(agent: &str, container: &str) -> AgentResponse { + // Record the attempt in the operator-visible privileged-action audit trail. + // Best-effort: no-op when the global handle isn't installed (early + // startup / tests). `action` is stable so the dashboard can group. + let audit = |outcome: crate::audit_log::AuditOutcome, detail: Option<&str>| { + if let Some(log) = crate::audit_log::global() { + log.record(agent, "restart_infra", container, outcome, detail); + } + }; if !crate::capabilities::has_cap(agent, hive_sh4re::Capability::InfraAdmin) { tracing::warn!(%agent, %container, "agent: infra restart denied (no infra_admin capability)"); + audit( + crate::audit_log::AuditOutcome::Err, + Some("denied: missing infra_admin capability"), + ); return AgentResponse::Err { message: format!( "restarting infra container `{container}` requires the `infra_admin` capability" @@ -555,10 +567,15 @@ async fn handle_restart_infra(agent: &str, container: &str) -> AgentResponse { } tracing::info!(%agent, %container, "agent: restart infra container"); match crate::priv_client::restart_infra_container(container).await { - Ok(()) => AgentResponse::Ok, - Err(e) => AgentResponse::Err { - message: format!("{e:#}"), - }, + Ok(()) => { + audit(crate::audit_log::AuditOutcome::Ok, None); + AgentResponse::Ok + } + Err(e) => { + let msg = format!("{e:#}"); + audit(crate::audit_log::AuditOutcome::Err, Some(&msg)); + AgentResponse::Err { message: msg } + } } } diff --git a/hive-c0re/src/audit_log.rs b/hive-c0re/src/audit_log.rs new file mode 100644 index 00000000..9fd4382c --- /dev/null +++ b/hive-c0re/src/audit_log.rs @@ -0,0 +1,288 @@ +//! Sqlite-backed audit trail of agent-initiated privileged actions. +//! +//! Surfaces, durably and operator-visibly, the privileged operations +//! hive-c0re performs *on behalf of an agent* — the ones that cross the +//! agent/operator trust boundary and so warrant a who/what/when record +//! beyond hive-priv's low-level journal trace. First entry: infra +//! container restarts via the `infra_admin`-gated `restart` tool (the +//! follow-up audit trail for that capability). +//! +//! Deliberately scoped to *agent-initiated* privileged actions. The bulk +//! of `PrivRequest` traffic (token writes, nspawn-flag edits) fires +//! constantly during normal lifecycle and is hive-c0re's own bookkeeping, +//! not an agent crossing the boundary — logging all of it would drown the +//! signal the operator actually wants. +//! +//! Same process-singleton handle pattern as `build_logs`: installed once +//! at `Coordinator::open`, fetched via [`global`] so the recording sites +//! (e.g. `agent_server::handle_restart_infra`) don't have to thread an +//! `Arc` through every call path. Recording is best-effort: a +//! sqlite blip must never fail the underlying privileged action. + +use std::path::Path; +use std::sync::{Arc, Mutex, OnceLock}; +use std::time::{SystemTime, UNIX_EPOCH}; + +use anyhow::{Context, Result}; +use rusqlite::{Connection, params}; +use serde::Serialize; + +/// Process-singleton handle, set once at coordinator startup. Mirrors +/// `build_logs::GLOBAL` — lets recording sites write without threading an +/// `Arc` through every entry point. +static GLOBAL: OnceLock> = OnceLock::new(); + +/// Install the process-wide `AuditLog` handle. Idempotent: a second call +/// silently keeps the first handle. +pub fn install(handle: Arc) { + let _ = GLOBAL.set(handle); +} + +/// Fetch the process-wide handle, or `None` if `install` hasn't run yet +/// (early startup, or unit tests). Callers must gracefully no-op on `None`. +#[must_use] +pub fn global() -> Option> { + GLOBAL.get().cloned() +} + +/// Retain audit rows for 90 days. Longer than build-log retention — this +/// is a security/accountability record, not debug noise; the operator may +/// want to review "who restarted what" well after the fact. +const KEEP_SECS: i64 = 90 * 24 * 3600; + +const SCHEMA: &str = " +CREATE TABLE IF NOT EXISTS audit_log ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + ts_unix INTEGER NOT NULL, + agent TEXT NOT NULL, + action TEXT NOT NULL, + target TEXT NOT NULL, + outcome TEXT NOT NULL, + detail TEXT +); +CREATE INDEX IF NOT EXISTS idx_audit_log_ts ON audit_log (ts_unix DESC); +"; + +/// Outcome of a recorded privileged action. Stored as the literal string +/// in the `outcome` column. +#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize)] +#[serde(rename_all = "snake_case")] +pub enum AuditOutcome { + /// The privileged action succeeded. + Ok, + /// The privileged action was attempted but failed (e.g. the + /// underlying systemctl call errored). Denied-by-capability attempts + /// are recorded too — see the recording site. + Err, +} + +impl AuditOutcome { + fn as_str(self) -> &'static str { + match self { + Self::Ok => "ok", + Self::Err => "err", + } + } +} + +/// One audit row as returned to the dashboard. +#[derive(Debug, Clone, Serialize)] +pub struct AuditEntry { + pub id: i64, + pub ts_unix: i64, + /// Agent on whose behalf the action was taken. + pub agent: String, + /// What was done (e.g. `restart_infra`). + pub action: String, + /// What it acted on (e.g. `hive-ci`). + pub target: String, + /// `"ok"` | `"err"`. + pub outcome: String, + /// Optional free-text detail (e.g. the error message on failure). + pub detail: Option, +} + +/// Sqlite-backed audit-log store. `Arc`-friendly: all methods +/// take `&self`, an internal `Mutex` serializes access. +pub struct AuditLog { + conn: Mutex, +} + +impl AuditLog { + pub fn open(db_dir: &Path) -> Result { + std::fs::create_dir_all(db_dir) + .with_context(|| format!("create audit_log db parent {}", db_dir.display()))?; + let path = db_dir.join("audit_log.sqlite"); + let conn = Connection::open(&path) + .with_context(|| format!("open audit_log db {}", path.display()))?; + conn.execute_batch(SCHEMA) + .context("apply audit_log schema")?; + Ok(Self { + conn: Mutex::new(conn), + }) + } + + /// Record one privileged action. Best-effort: a sqlite error is logged + /// but never returned, so a transient blip never fails the underlying + /// privileged action (the action already happened — losing its audit + /// row is strictly less bad than failing the action retroactively). + pub fn record( + &self, + agent: &str, + action: &str, + target: &str, + outcome: AuditOutcome, + detail: Option<&str>, + ) { + let now = now_secs(); + let conn = self.conn.lock().unwrap(); + if let Err(e) = conn.execute( + "INSERT INTO audit_log (ts_unix, agent, action, target, outcome, detail) + VALUES (?1, ?2, ?3, ?4, ?5, ?6)", + params![now, agent, action, target, outcome.as_str(), detail], + ) { + tracing::warn!( + %agent, %action, %target, + error = ?e, + "audit_log: record failed (dropping entry)" + ); + } + } + + /// Return the most recent `limit` rows, newest first. Limit is + /// hard-clamped to 500 to bound the worst-case payload. + pub fn list_recent(&self, limit: usize) -> Result> { + let limit = limit.min(500); + let conn = self.conn.lock().unwrap(); + let mut stmt = conn.prepare( + "SELECT id, ts_unix, agent, action, target, outcome, detail + FROM audit_log + ORDER BY ts_unix DESC, id DESC + LIMIT ?1", + )?; + let rows = stmt.query_map(params![i64::try_from(limit).unwrap_or(500)], row_to_entry)?; + let mut out = Vec::new(); + for r in rows { + out.push(r?); + } + Ok(out) + } + + /// Drop rows older than the retention window. Returns the number of + /// rows deleted. Called from the hourly vacuum loop. + pub fn vacuum(&self) -> Result { + let cutoff = now_secs() - KEEP_SECS; + let conn = self.conn.lock().unwrap(); + let removed = conn.execute("DELETE FROM audit_log WHERE ts_unix < ?1", params![cutoff])?; + Ok(u64::try_from(removed).unwrap_or(0)) + } +} + +/// Spawn the hourly retention sweep. Mirrors `build_logs::spawn_vacuum` +/// in cadence + shutdown handling. +pub fn spawn_vacuum(coord: &Arc) { + use std::time::Duration; + let audit = coord.audit_log.clone(); + let mut shutdown = coord.shutdown_rx(); + let interval = Duration::from_hours(1); + tokio::spawn(async move { + loop { + match audit.vacuum() { + Ok(0) => {} + Ok(n) => tracing::info!(removed = n, "audit_log vacuum"), + Err(e) => tracing::warn!(error = ?e, "audit_log vacuum failed"), + } + tokio::select! { + () = tokio::time::sleep(interval) => {} + _ = shutdown.changed() => { + tracing::info!("audit_log vacuum: shutdown signal received"); + break; + } + } + } + }); +} + +fn row_to_entry(r: &rusqlite::Row) -> rusqlite::Result { + Ok(AuditEntry { + id: r.get(0)?, + ts_unix: r.get(1)?, + agent: r.get(2)?, + action: r.get(3)?, + target: r.get(4)?, + outcome: r.get(5)?, + detail: r.get(6)?, + }) +} + +fn now_secs() -> i64 { + SystemTime::now() + .duration_since(UNIX_EPOCH) + .ok() + .and_then(|d| i64::try_from(d.as_secs()).ok()) + .unwrap_or(0) +} + +#[cfg(test)] +mod tests { + use super::*; + + fn tmpdb() -> (tempfile::TempDir, AuditLog) { + let dir = tempfile::tempdir().expect("tempdir"); + let db = AuditLog::open(dir.path()).expect("open"); + (dir, db) + } + + #[test] + fn record_and_list_newest_first() { + let (_d, db) = tmpdb(); + db.record("atlas", "restart_infra", "hive-ci", AuditOutcome::Ok, None); + db.record( + "atlas", + "restart_infra", + "hive-gateway", + AuditOutcome::Err, + Some("systemctl failed"), + ); + let rows = db.list_recent(10).expect("list"); + assert_eq!(rows.len(), 2); + // Newest first: the gateway/err row was inserted last. + assert_eq!(rows[0].target, "hive-gateway"); + assert_eq!(rows[0].outcome, "err"); + assert_eq!(rows[0].detail.as_deref(), Some("systemctl failed")); + assert_eq!(rows[1].target, "hive-ci"); + assert_eq!(rows[1].outcome, "ok"); + assert!(rows[1].detail.is_none()); + assert_eq!(rows[0].agent, "atlas"); + assert_eq!(rows[0].action, "restart_infra"); + } + + #[test] + fn list_clamps_to_500() { + let (_d, db) = tmpdb(); + db.record("a", "x", "t", AuditOutcome::Ok, None); + let rows = db.list_recent(999_999).expect("list"); + assert!(rows.len() <= 500); + } + + #[test] + fn vacuum_drops_only_old_rows() { + let (_d, db) = tmpdb(); + db.record("a", "restart_infra", "hive-ci", AuditOutcome::Ok, None); + // Backdate it past the retention window. + { + let conn = db.conn.lock().unwrap(); + conn.execute( + "UPDATE audit_log SET ts_unix = ?1", + params![now_secs() - KEEP_SECS - 60], + ) + .unwrap(); + } + db.record("a", "restart_infra", "hive-forge", AuditOutcome::Ok, None); + let removed = db.vacuum().expect("vacuum"); + assert_eq!(removed, 1, "only the backdated row should be reaped"); + let rows = db.list_recent(10).expect("list"); + assert_eq!(rows.len(), 1); + assert_eq!(rows[0].target, "hive-forge"); + } +} diff --git a/hive-c0re/src/coordinator.rs b/hive-c0re/src/coordinator.rs index ff79f063..56f50373 100644 --- a/hive-c0re/src/coordinator.rs +++ b/hive-c0re/src/coordinator.rs @@ -47,6 +47,9 @@ pub struct Coordinator { /// `get_full` for the per-card chip + side-panel viewer. See /// `build_logs.rs` for retention. pub build_logs: Arc, + /// Audit trail of agent-initiated privileged actions (infra restart, + /// …). See `audit_log.rs`. Same dir as `build_logs`. + pub audit_log: Arc, /// URL of the hyperhive flake (no fragment). Inlined into per-agent /// `flake.nix` files as `inputs.hyperhive.url`. pub hyperhive_flake: String, @@ -418,6 +421,13 @@ impl Coordinator { // to thread an `Arc` through every public entry // point in the lifecycle surface. crate::build_logs::install(build_logs.clone()); + // Audit log shares the same db dir; install its process-wide + // handle so privileged-action recording sites (e.g. + // `agent_server::handle_restart_infra`) write without threading an + // `Arc` through the agent-request surface. + let audit_log = + Arc::new(crate::audit_log::AuditLog::open(build_logs_dir).context("open audit_log")?); + crate::audit_log::install(audit_log.clone()); let (dashboard_events, _) = broadcast::channel(DASHBOARD_CHANNEL); let (shutdown_tx, _) = watch::channel(false); Ok(Self { @@ -426,6 +436,7 @@ impl Coordinator { questions: Arc::new(questions), scheduled_prompts: Arc::new(scheduled_prompts), build_logs, + audit_log, hyperhive_flake, nixpkgs_flake, nixpkgs_unstable_flake, diff --git a/hive-c0re/src/dashboard.rs b/hive-c0re/src/dashboard.rs index 05e8b43f..30a889ee 100644 --- a/hive-c0re/src/dashboard.rs +++ b/hive-c0re/src/dashboard.rs @@ -104,6 +104,7 @@ pub async fn serve(port: u16, coord: Arc) -> Result<()> { .route("/api/operator-inbox", get(api_operator_inbox)) .route("/api/stats-hive", get(api_stats_hive)) .route("/api/container-resources", get(api_container_resources)) + .route("/api/audit-log", get(api_audit_log)) .route("/api/build-logs", get(build_logs::get_build_logs_all)) .route( "/api/build-logs/{agent}", @@ -1312,6 +1313,16 @@ async fn api_container_resources() -> Response { axum::Json(crate::container_stats::gather().await).into_response() } +/// `GET /api/audit-log` — most-recent agent-initiated privileged-action +/// audit entries, newest first (server-clamped to 500). Backs the +/// operator dashboard's audit view. Returns `Vec` JSON. +async fn api_audit_log(State(state): State) -> Response { + match state.coord.audit_log.list_recent(500) { + Ok(rows) => axum::Json(rows).into_response(), + Err(e) => error_response(&format!("audit-log: {e:#}")), + } +} + /// Validate that a path-param agent name conforms to the hyperhive /// naming whitelist: 1-63 chars of `[a-z0-9_-]`. Rejects empty, /// uppercase, slashes, dots, and any non-ASCII (incl. unicode diff --git a/hive-c0re/src/lib.rs b/hive-c0re/src/lib.rs index 85009735..4d624832 100644 --- a/hive-c0re/src/lib.rs +++ b/hive-c0re/src/lib.rs @@ -17,6 +17,7 @@ pub mod agent_ports; pub mod agent_server; pub mod agent_sockets; pub mod approvals; +pub mod audit_log; pub mod auto_update; pub mod bash_tasks_vacuum; pub mod broker; diff --git a/hive-c0re/src/main.rs b/hive-c0re/src/main.rs index 60413952..8d9efacd 100644 --- a/hive-c0re/src/main.rs +++ b/hive-c0re/src/main.rs @@ -397,6 +397,9 @@ async fn cmd_serve( // build_logs.sqlite vacuum: c0re-side (single db). Failures kept // 30d, successes 24h — see `build_logs::vacuum` for the rule. hive_c0re::build_logs::spawn_vacuum(&coord); + // audit_log.sqlite vacuum: agent-initiated privileged-action trail, + // 90d retention — see `audit_log::vacuum`. + hive_c0re::audit_log::spawn_vacuum(&coord); // Container crash watcher: emits HelperEvent::ContainerCrash // when a previously-running container goes away without an // operator-initiated transient state.