feat(audit): persistent audit log of agent-initiated privileged actions
Adds a durable, operator-visible audit trail of privileged operations hive-c0re performs on behalf of an agent — the ones that cross the agent/operator trust boundary. First entry: infra-container restarts via the infra_admin-gated `restart` tool, which until now were recorded only as a hive-priv journal trace. Backend: - new `audit_log` module: sqlite-backed store (audit_log.sqlite, same dir as build_logs) with schema (ts/agent/action/target/outcome/detail), best-effort `record`, `list_recent` (clamped 500), 90-day `vacuum`, and a process-singleton handle mirroring build_logs. - Coordinator opens + installs the handle; main spawns the hourly vacuum. - agent_server::handle_restart_infra records every attempt (ok, error, and capability-denied) via the global handle — best-effort, never fails the underlying action. - dashboard: `GET /api/audit-log` returns recent entries as JSON. Scope is deliberately agent-initiated privileged actions only (not every PrivRequest — token writes + nspawn edits are constant lifecycle noise). Extensible: future agent-initiated priv ops record via the same handle. Unit tests cover record/list ordering, the 500 clamp, and retention vacuum. The dashboard *surface* (an AUDIT view consuming /api/audit-log) is a frontend follow-up coordinated with iris.
This commit is contained in:
atlas
parent
e9dec143e5
commit
a452a92fb1
6 changed files with 335 additions and 4 deletions
288
hive-c0re/src/audit_log.rs
Normal file
288
hive-c0re/src/audit_log.rs
Normal file
|
|
@ -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<AuditLog>` 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<AuditLog>` through every entry point.
|
||||
static GLOBAL: OnceLock<Arc<AuditLog>> = OnceLock::new();
|
||||
|
||||
/// Install the process-wide `AuditLog` handle. Idempotent: a second call
|
||||
/// silently keeps the first handle.
|
||||
pub fn install(handle: Arc<AuditLog>) {
|
||||
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<Arc<AuditLog>> {
|
||||
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<String>,
|
||||
}
|
||||
|
||||
/// Sqlite-backed audit-log store. `Arc<AuditLog>`-friendly: all methods
|
||||
/// take `&self`, an internal `Mutex<Connection>` serializes access.
|
||||
pub struct AuditLog {
|
||||
conn: Mutex<Connection>,
|
||||
}
|
||||
|
||||
impl AuditLog {
|
||||
pub fn open(db_dir: &Path) -> Result<Self> {
|
||||
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<Vec<AuditEntry>> {
|
||||
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<u64> {
|
||||
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<crate::coordinator::Coordinator>) {
|
||||
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<AuditEntry> {
|
||||
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");
|
||||
}
|
||||
}
|
||||
Loading…
Reference in a new issue