feat(audit): live-append event for the dashboard audit view
Follow-up to the audit-log backend + surface. Emits a DashboardEvent on
each agent-initiated privileged action so the audit view live-appends off
/dashboard/stream instead of polling.
- new DashboardEvent::AuditEntryAdded { seq, <flattened AuditEntry> } —
serde tag `audit_entry_added`; the AuditEntry fields flatten to the top
level so the wire shape matches one /api/audit-log `entries` row exactly.
- Coordinator::emit_audit_entry helper (stamps seq like the others).
- audit_log::record now returns the canonical inserted AuditEntry (id + ts
assigned) so the streamed event is the same row that was stored — no
drift. Best-effort unchanged (None on a sqlite blip).
- handle_restart_infra records + emits for every attempt (ok/err/denied),
threading the coordinator through.
Tests: kind_tag round-trip now covers the new variant; added a flatten
test pinning the top-level wire shape (kind/seq/id/…/detail, no nesting).
Pairs with iris's audit view (the /dashboard/stream listener half).
This commit is contained in:
parent
9bd51a7440
commit
629f08a113
4 changed files with 122 additions and 19 deletions
|
|
@ -522,7 +522,7 @@ async fn handle_restart_child(coord: &Arc<Coordinator>, agent: &str, name: &str)
|
||||||
// tool. These names are never agent children, so this branch is
|
// tool. These names are never agent children, so this branch is
|
||||||
// disjoint from the child-restart path below.
|
// disjoint from the child-restart path below.
|
||||||
if hive_sh4re::priv_proto::RESTARTABLE_INFRA_CONTAINERS.contains(&name) {
|
if hive_sh4re::priv_proto::RESTARTABLE_INFRA_CONTAINERS.contains(&name) {
|
||||||
return handle_restart_infra(agent, name).await;
|
return handle_restart_infra(coord, agent, name).await;
|
||||||
}
|
}
|
||||||
if let Some(err) = require_child(agent, name, "restart") {
|
if let Some(err) = require_child(agent, name, "restart") {
|
||||||
return err;
|
return err;
|
||||||
|
|
@ -544,13 +544,24 @@ async fn handle_restart_child(coord: &Arc<Coordinator>, agent: &str, name: &str)
|
||||||
/// known to be in `RESTARTABLE_INFRA_CONTAINERS`; this gates on the
|
/// known to be in `RESTARTABLE_INFRA_CONTAINERS`; this gates on the
|
||||||
/// 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(
|
||||||
// Record the attempt in the operator-visible privileged-action audit trail.
|
coord: &Arc<Coordinator>,
|
||||||
// Best-effort: no-op when the global handle isn't installed (early
|
agent: &str,
|
||||||
// startup / tests). `action` is stable so the dashboard can group.
|
container: &str,
|
||||||
|
) -> AgentResponse {
|
||||||
|
// Record the attempt in the operator-visible privileged-action audit
|
||||||
|
// trail, then emit a live `AuditEntryAdded` so the dashboard audit view
|
||||||
|
// appends it off `/dashboard/stream`. Best-effort: `record` returns the
|
||||||
|
// canonical row (or `None` on a sqlite blip), and we stream exactly that
|
||||||
|
// row so the stored + streamed views can't drift. `action` is stable so
|
||||||
|
// the dashboard can group/filter.
|
||||||
let audit = |outcome: crate::audit_log::AuditOutcome, detail: Option<&str>| {
|
let audit = |outcome: crate::audit_log::AuditOutcome, detail: Option<&str>| {
|
||||||
if let Some(log) = crate::audit_log::global() {
|
if let Some(entry) =
|
||||||
log.record(agent, "restart_infra", container, outcome, detail);
|
coord
|
||||||
|
.audit_log
|
||||||
|
.record(agent, "restart_infra", container, outcome, detail)
|
||||||
|
{
|
||||||
|
coord.emit_audit_entry(entry);
|
||||||
}
|
}
|
||||||
};
|
};
|
||||||
if !crate::capabilities::has_cap(agent, hive_sh4re::Capability::InfraAdmin) {
|
if !crate::capabilities::has_cap(agent, hive_sh4re::Capability::InfraAdmin) {
|
||||||
|
|
|
||||||
|
|
@ -133,6 +133,13 @@ impl AuditLog {
|
||||||
/// but never returned, so a transient blip never fails the underlying
|
/// but never returned, so a transient blip never fails the underlying
|
||||||
/// privileged action (the action already happened — losing its audit
|
/// privileged action (the action already happened — losing its audit
|
||||||
/// row is strictly less bad than failing the action retroactively).
|
/// row is strictly less bad than failing the action retroactively).
|
||||||
|
///
|
||||||
|
/// Returns the inserted [`AuditEntry`] (with its assigned id +
|
||||||
|
/// timestamp) on success, or `None` if the insert failed. The
|
||||||
|
/// returned row is the canonical record — callers that also push a
|
||||||
|
/// live event (e.g. the dashboard stream) emit *this* rather than
|
||||||
|
/// re-deriving the fields, so the stored row and the streamed event
|
||||||
|
/// can't drift.
|
||||||
pub fn record(
|
pub fn record(
|
||||||
&self,
|
&self,
|
||||||
agent: &str,
|
agent: &str,
|
||||||
|
|
@ -140,19 +147,31 @@ impl AuditLog {
|
||||||
target: &str,
|
target: &str,
|
||||||
outcome: AuditOutcome,
|
outcome: AuditOutcome,
|
||||||
detail: Option<&str>,
|
detail: Option<&str>,
|
||||||
) {
|
) -> Option<AuditEntry> {
|
||||||
let now = now_secs();
|
let now = now_secs();
|
||||||
let conn = self.conn.lock().unwrap();
|
let conn = self.conn.lock().unwrap();
|
||||||
if let Err(e) = conn.execute(
|
match conn.execute(
|
||||||
"INSERT INTO audit_log (ts_unix, agent, action, target, outcome, detail)
|
"INSERT INTO audit_log (ts_unix, agent, action, target, outcome, detail)
|
||||||
VALUES (?1, ?2, ?3, ?4, ?5, ?6)",
|
VALUES (?1, ?2, ?3, ?4, ?5, ?6)",
|
||||||
params![now, agent, action, target, outcome.as_str(), detail],
|
params![now, agent, action, target, outcome.as_str(), detail],
|
||||||
) {
|
) {
|
||||||
tracing::warn!(
|
Ok(_) => Some(AuditEntry {
|
||||||
%agent, %action, %target,
|
id: conn.last_insert_rowid(),
|
||||||
error = ?e,
|
ts_unix: now,
|
||||||
"audit_log: record failed (dropping entry)"
|
agent: agent.to_owned(),
|
||||||
);
|
action: action.to_owned(),
|
||||||
|
target: target.to_owned(),
|
||||||
|
outcome: outcome.as_str().to_owned(),
|
||||||
|
detail: detail.map(str::to_owned),
|
||||||
|
}),
|
||||||
|
Err(e) => {
|
||||||
|
tracing::warn!(
|
||||||
|
%agent, %action, %target,
|
||||||
|
error = ?e,
|
||||||
|
"audit_log: record failed (dropping entry)"
|
||||||
|
);
|
||||||
|
None
|
||||||
|
}
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|
@ -261,8 +280,15 @@ mod tests {
|
||||||
#[test]
|
#[test]
|
||||||
fn record_and_list_newest_first() {
|
fn record_and_list_newest_first() {
|
||||||
let (_d, db) = tmpdb();
|
let (_d, db) = tmpdb();
|
||||||
db.record("atlas", "restart_infra", "hive-ci", AuditOutcome::Ok, None);
|
// record() returns the canonical inserted row (id + ts assigned).
|
||||||
db.record(
|
let entry = db
|
||||||
|
.record("atlas", "restart_infra", "hive-ci", AuditOutcome::Ok, None)
|
||||||
|
.expect("record returns the inserted entry");
|
||||||
|
assert!(entry.id > 0);
|
||||||
|
assert_eq!(entry.target, "hive-ci");
|
||||||
|
assert_eq!(entry.outcome, "ok");
|
||||||
|
assert!(entry.detail.is_none());
|
||||||
|
let _ = db.record(
|
||||||
"atlas",
|
"atlas",
|
||||||
"restart_infra",
|
"restart_infra",
|
||||||
"hive-gateway",
|
"hive-gateway",
|
||||||
|
|
@ -286,7 +312,7 @@ mod tests {
|
||||||
#[test]
|
#[test]
|
||||||
fn list_clamps_to_500() {
|
fn list_clamps_to_500() {
|
||||||
let (_d, db) = tmpdb();
|
let (_d, db) = tmpdb();
|
||||||
db.record("a", "x", "t", AuditOutcome::Ok, None);
|
let _ = db.record("a", "x", "t", AuditOutcome::Ok, None);
|
||||||
let rows = db.list_recent(999_999).expect("list");
|
let rows = db.list_recent(999_999).expect("list");
|
||||||
assert!(rows.len() <= 500);
|
assert!(rows.len() <= 500);
|
||||||
}
|
}
|
||||||
|
|
@ -294,7 +320,7 @@ mod tests {
|
||||||
#[test]
|
#[test]
|
||||||
fn vacuum_drops_only_old_rows() {
|
fn vacuum_drops_only_old_rows() {
|
||||||
let (_d, db) = tmpdb();
|
let (_d, db) = tmpdb();
|
||||||
db.record("a", "restart_infra", "hive-ci", AuditOutcome::Ok, None);
|
let _ = db.record("a", "restart_infra", "hive-ci", AuditOutcome::Ok, None);
|
||||||
// Backdate it past the retention window.
|
// Backdate it past the retention window.
|
||||||
{
|
{
|
||||||
let conn = db.conn.lock().unwrap();
|
let conn = db.conn.lock().unwrap();
|
||||||
|
|
@ -304,7 +330,7 @@ mod tests {
|
||||||
)
|
)
|
||||||
.unwrap();
|
.unwrap();
|
||||||
}
|
}
|
||||||
db.record("a", "restart_infra", "hive-forge", AuditOutcome::Ok, None);
|
let _ = db.record("a", "restart_infra", "hive-forge", AuditOutcome::Ok, None);
|
||||||
let removed = db.vacuum().expect("vacuum");
|
let removed = db.vacuum().expect("vacuum");
|
||||||
assert_eq!(removed, 1, "only the backdated row should be reaped");
|
assert_eq!(removed, 1, "only the backdated row should be reaped");
|
||||||
let rows = db.list_recent(10).expect("list");
|
let rows = db.list_recent(10).expect("list");
|
||||||
|
|
|
||||||
|
|
@ -692,6 +692,18 @@ impl Coordinator {
|
||||||
self.meta_updates_active.load(Ordering::SeqCst) > 0
|
self.meta_updates_active.load(Ordering::SeqCst) > 0
|
||||||
}
|
}
|
||||||
|
|
||||||
|
/// Emit `AuditEntryAdded` immediately after a privileged-action row
|
||||||
|
/// is recorded, so the dashboard audit view live-appends it off
|
||||||
|
/// `/dashboard/stream`. Pass the [`AuditEntry`](crate::audit_log::AuditEntry)
|
||||||
|
/// returned by `audit_log::record` so the streamed event is the same
|
||||||
|
/// canonical row that was stored.
|
||||||
|
pub fn emit_audit_entry(&self, entry: crate::audit_log::AuditEntry) {
|
||||||
|
self.emit_dashboard_event(DashboardEvent::AuditEntryAdded {
|
||||||
|
seq: self.next_seq(),
|
||||||
|
entry,
|
||||||
|
});
|
||||||
|
}
|
||||||
|
|
||||||
/// Emit `ApprovalAdded` immediately after the row is inserted in
|
/// Emit `ApprovalAdded` immediately after the row is inserted in
|
||||||
/// sqlite. Caller passes the diff text it already computed (or
|
/// sqlite. Caller passes the diff text it already computed (or
|
||||||
/// `None` for spawn approvals which carry no diff).
|
/// `None` for spawn approvals which carry no diff).
|
||||||
|
|
|
||||||
|
|
@ -13,6 +13,17 @@ use crate::rebuild_queue::QueueEntry;
|
||||||
#[derive(Debug, Clone, Serialize)]
|
#[derive(Debug, Clone, Serialize)]
|
||||||
#[serde(rename_all = "snake_case", tag = "kind")]
|
#[serde(rename_all = "snake_case", tag = "kind")]
|
||||||
pub enum DashboardEvent {
|
pub enum DashboardEvent {
|
||||||
|
/// A new agent-initiated privileged action was recorded in the audit
|
||||||
|
/// log. The audit view (`/audit.html`) prepends `entry` live off
|
||||||
|
/// `/dashboard/stream` instead of polling. The `AuditEntry` fields
|
||||||
|
/// are flattened alongside the `kind` tag + `seq`, so the wire shape
|
||||||
|
/// matches one row of the `/api/audit-log` `entries` array exactly
|
||||||
|
/// (`{kind, seq, id, ts_unix, agent, action, target, outcome, detail}`).
|
||||||
|
AuditEntryAdded {
|
||||||
|
seq: u64,
|
||||||
|
#[serde(flatten)]
|
||||||
|
entry: crate::audit_log::AuditEntry,
|
||||||
|
},
|
||||||
/// Broker `Sent` event mirrored onto the dashboard channel.
|
/// Broker `Sent` event mirrored onto the dashboard channel.
|
||||||
/// `file_refs` carries every path-shaped token in `body` that
|
/// `file_refs` carries every path-shaped token in `body` that
|
||||||
/// hive-c0re verified is a regular file under the allow-listed
|
/// hive-c0re verified is a regular file under the allow-listed
|
||||||
|
|
@ -270,6 +281,7 @@ impl DashboardEvent {
|
||||||
DashboardEvent::RemindersChanged { .. } => "reminders_changed",
|
DashboardEvent::RemindersChanged { .. } => "reminders_changed",
|
||||||
DashboardEvent::CapabilitiesChanged { .. } => "capabilities_changed",
|
DashboardEvent::CapabilitiesChanged { .. } => "capabilities_changed",
|
||||||
DashboardEvent::ToolGroupsChanged { .. } => "tool_groups_changed",
|
DashboardEvent::ToolGroupsChanged { .. } => "tool_groups_changed",
|
||||||
|
DashboardEvent::AuditEntryAdded { .. } => "audit_entry_added",
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
@ -408,6 +420,18 @@ mod tests {
|
||||||
descriptions: std::collections::BTreeMap::new(),
|
descriptions: std::collections::BTreeMap::new(),
|
||||||
assignments: std::collections::BTreeMap::new(),
|
assignments: std::collections::BTreeMap::new(),
|
||||||
},
|
},
|
||||||
|
DashboardEvent::AuditEntryAdded {
|
||||||
|
seq: 1,
|
||||||
|
entry: crate::audit_log::AuditEntry {
|
||||||
|
id: 1,
|
||||||
|
ts_unix: 0,
|
||||||
|
agent: "atlas".into(),
|
||||||
|
action: "restart_infra".into(),
|
||||||
|
target: "hive-ci".into(),
|
||||||
|
outcome: "ok".into(),
|
||||||
|
detail: None,
|
||||||
|
},
|
||||||
|
},
|
||||||
];
|
];
|
||||||
for ev in samples {
|
for ev in samples {
|
||||||
let v: serde_json::Value = serde_json::to_value(&ev).expect("serialise");
|
let v: serde_json::Value = serde_json::to_value(&ev).expect("serialise");
|
||||||
|
|
@ -418,4 +442,34 @@ mod tests {
|
||||||
assert_eq!(ev.kind_tag(), serde_kind, "kind_tag() drift on {ev:?}");
|
assert_eq!(ev.kind_tag(), serde_kind, "kind_tag() drift on {ev:?}");
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
/// The flattened `AuditEntry` fields must sit alongside `kind`/`seq`
|
||||||
|
/// at the top level (not nested under `entry`) so the wire shape
|
||||||
|
/// matches one `/api/audit-log` row — the audit view prepends it
|
||||||
|
/// directly.
|
||||||
|
#[test]
|
||||||
|
fn audit_entry_added_flattens_to_top_level() {
|
||||||
|
let ev = DashboardEvent::AuditEntryAdded {
|
||||||
|
seq: 7,
|
||||||
|
entry: crate::audit_log::AuditEntry {
|
||||||
|
id: 42,
|
||||||
|
ts_unix: 1_700_000_000,
|
||||||
|
agent: "atlas".into(),
|
||||||
|
action: "restart_infra".into(),
|
||||||
|
target: "hive-gateway".into(),
|
||||||
|
outcome: "err".into(),
|
||||||
|
detail: Some("denied: missing infra_admin capability".into()),
|
||||||
|
},
|
||||||
|
};
|
||||||
|
let v: serde_json::Value = serde_json::to_value(&ev).expect("serialise");
|
||||||
|
assert_eq!(v["kind"], "audit_entry_added");
|
||||||
|
assert_eq!(v["seq"], 7);
|
||||||
|
assert_eq!(v["id"], 42);
|
||||||
|
assert_eq!(v["agent"], "atlas");
|
||||||
|
assert_eq!(v["target"], "hive-gateway");
|
||||||
|
assert_eq!(v["outcome"], "err");
|
||||||
|
assert_eq!(v["detail"], "denied: missing infra_admin capability");
|
||||||
|
// Not nested — there must be no `entry` sub-object.
|
||||||
|
assert!(v.get("entry").is_none());
|
||||||
|
}
|
||||||
}
|
}
|
||||||
|
|
|
||||||
Loading…
Reference in a new issue