vinzenz/hyperhive
vinzenz/hyperhive
Watch
0
0
Fork
You've already forked hyperhive
 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions 2

Compare commits

base: vinzenz:79fa4f97b7989a0daea44a2f82530a0d78bd219b
Branches Tags
vinzenz:main
..
compare: vinzenz:e9dec143e5626e9327f449a65fb815621080fa0a
Branches Tags
vinzenz:main
6 changed files with 4 additions and 370 deletions
Download patch file Download diff file Expand all files Collapse all files

25
hive-c0re/src/agent_server.rs
View file

@ -545,20 +545,8 @@ async fn handle_restart_child(coord: &Arc<Coordinator>, agent: &str, name: &str)
/// capability and routes the systemctl restart through hive-priv (which /// capability and routes the systemctl restart through hive-priv (which
/// re-validates the name root-side). Direct, not approval-gated. /// re-validates the name root-side). Direct, not approval-gated.
async fn handle_restart_infra(agent: &str, container: &str) -> AgentResponse { 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) { if !crate::capabilities::has_cap(agent, hive_sh4re::Capability::InfraAdmin) {
tracing::warn!(%agent, %container, "agent: infra restart denied (no infra_admin capability)"); 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 { return AgentResponse::Err {
message: format!( message: format!(
"restarting infra container `{container}` requires the `infra_admin` capability" "restarting infra container `{container}` requires the `infra_admin` capability"
@ -567,15 +555,10 @@ async fn handle_restart_infra(agent: &str, container: &str) -> AgentResponse {
} }
tracing::info!(%agent, %container, "agent: restart infra container"); tracing::info!(%agent, %container, "agent: restart infra container");
match crate::priv_client::restart_infra_container(container).await { match crate::priv_client::restart_infra_container(container).await {
Ok(()) => { Ok(()) => AgentResponse::Ok,
audit(crate::audit_log::AuditOutcome::Ok, None); Err(e) => AgentResponse::Err {
AgentResponse::Ok message: format!("{e:#}"),
} },
Err(e) => {
let msg = format!("{e:#}");
audit(crate::audit_log::AuditOutcome::Err, Some(&msg));
AgentResponse::Err { message: msg }
}
} }
} }

314
hive-c0re/src/audit_log.rs
View file

@ -1,314 +0,0 @@
//! 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 {
/// Open (creating if absent) the `audit_log.sqlite` store under
/// `db_dir` and apply the schema. `db_dir` is shared with
/// `build_logs` (the broker db's parent directory).
///
/// # Errors
/// Returns an error if the directory can't be created, the sqlite
/// file can't be opened, or applying the schema fails.
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.
///
/// # Errors
/// Returns an error if the query fails to prepare or a row fails to
/// deserialize.
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)
}
/// Total row count, regardless of the `list_recent` clamp. Lets the
/// dashboard show "latest N of TOTAL" instead of silently capping.
///
/// # Errors
/// Returns an error if the `COUNT(*)` query fails.
pub fn count_total(&self) -> Result<i64> {
let conn = self.conn.lock().unwrap();
let n: i64 = conn.query_row("SELECT COUNT(*) FROM audit_log", [], |r| r.get(0))?;
Ok(n)
}
/// Drop rows older than the retention window. Returns the number of
/// rows deleted. Called from the hourly vacuum loop.
///
/// # Errors
/// Returns an error if the `DELETE` query fails.
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");
assert_eq!(db.count_total().expect("count"), 2);
}
#[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");
}
}

11
hive-c0re/src/coordinator.rs
View file

@ -47,9 +47,6 @@ pub struct Coordinator {
/// `get_full` for the per-card chip + side-panel viewer. See /// `get_full` for the per-card chip + side-panel viewer. See
/// `build_logs.rs` for retention. /// `build_logs.rs` for retention.
pub build_logs: Arc<crate::build_logs::BuildLogs>, pub build_logs: Arc<crate::build_logs::BuildLogs>,
/// Audit trail of agent-initiated privileged actions (infra restart,
/// …). See `audit_log.rs`. Same dir as `build_logs`.
pub audit_log: Arc<crate::audit_log::AuditLog>,
/// URL of the hyperhive flake (no fragment). Inlined into per-agent /// URL of the hyperhive flake (no fragment). Inlined into per-agent
/// `flake.nix` files as `inputs.hyperhive.url`. /// `flake.nix` files as `inputs.hyperhive.url`.
pub hyperhive_flake: String, pub hyperhive_flake: String,
@ -421,13 +418,6 @@ impl Coordinator {
// to thread an `Arc<BuildLogs>` through every public entry // to thread an `Arc<BuildLogs>` through every public entry
// point in the lifecycle surface. // point in the lifecycle surface.
crate::build_logs::install(build_logs.clone()); 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<AuditLog>` 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 (dashboard_events, _) = broadcast::channel(DASHBOARD_CHANNEL);
let (shutdown_tx, _) = watch::channel(false); let (shutdown_tx, _) = watch::channel(false);
Ok(Self { Ok(Self {
@ -436,7 +426,6 @@ impl Coordinator {
questions: Arc::new(questions), questions: Arc::new(questions),
scheduled_prompts: Arc::new(scheduled_prompts), scheduled_prompts: Arc::new(scheduled_prompts),
build_logs, build_logs,
audit_log,
hyperhive_flake, hyperhive_flake,
nixpkgs_flake, nixpkgs_flake,
nixpkgs_unstable_flake, nixpkgs_unstable_flake,

20
hive-c0re/src/dashboard.rs
View file

@ -104,7 +104,6 @@ pub async fn serve(port: u16, coord: Arc<Coordinator>) -> Result<()> {
.route("/api/operator-inbox", get(api_operator_inbox)) .route("/api/operator-inbox", get(api_operator_inbox))
.route("/api/stats-hive", get(api_stats_hive)) .route("/api/stats-hive", get(api_stats_hive))
.route("/api/container-resources", get(api_container_resources)) .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", get(build_logs::get_build_logs_all))
.route( .route(
"/api/build-logs/{agent}", "/api/build-logs/{agent}",
@ -1313,25 +1312,6 @@ async fn api_container_resources() -> Response {
axum::Json(crate::container_stats::gather().await).into_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
/// `{ "entries": [AuditEntry…], "total": N }` so the UI can show
/// "latest 500 of N" rather than silently capping. `ts_unix` is in
/// **seconds**.
async fn api_audit_log(State(state): State<AppState>) -> Response {
const LIMIT: usize = 500;
let entries = match state.coord.audit_log.list_recent(LIMIT) {
Ok(rows) => rows,
Err(e) => return error_response(&format!("audit-log: {e:#}")),
};
let total = match state.coord.audit_log.count_total() {
Ok(n) => n,
Err(e) => return error_response(&format!("audit-log count: {e:#}")),
};
axum::Json(serde_json::json!({ "entries": entries, "total": total })).into_response()
}
/// Validate that a path-param agent name conforms to the hyperhive /// Validate that a path-param agent name conforms to the hyperhive
/// naming whitelist: 1-63 chars of `[a-z0-9_-]`. Rejects empty, /// naming whitelist: 1-63 chars of `[a-z0-9_-]`. Rejects empty,
/// uppercase, slashes, dots, and any non-ASCII (incl. unicode /// uppercase, slashes, dots, and any non-ASCII (incl. unicode

1
hive-c0re/src/lib.rs
View file

@ -17,7 +17,6 @@ pub mod agent_ports;
pub mod agent_server; pub mod agent_server;
pub mod agent_sockets; pub mod agent_sockets;
pub mod approvals; pub mod approvals;
pub mod audit_log;
pub mod auto_update; pub mod auto_update;
pub mod bash_tasks_vacuum; pub mod bash_tasks_vacuum;
pub mod broker; pub mod broker;

3
hive-c0re/src/main.rs
View file

@ -397,9 +397,6 @@ async fn cmd_serve(
// build_logs.sqlite vacuum: c0re-side (single db). Failures kept // build_logs.sqlite vacuum: c0re-side (single db). Failures kept
// 30d, successes 24h — see `build_logs::vacuum` for the rule. // 30d, successes 24h — see `build_logs::vacuum` for the rule.
hive_c0re::build_logs::spawn_vacuum(&coord); 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 // Container crash watcher: emits HelperEvent::ContainerCrash
// when a previously-running container goes away without an // when a previously-running container goes away without an
// operator-initiated transient state. // operator-initiated transient state.