crates/gunnar-core/src/account_audit.rs
//! **The account admin's audit log** — every security-relevant event in ONE
//! account, append-only, hash-chained, kept per account, a year by default
//! (owner's enterprise list, 2026-10-05).
//!
//! # Not the control plane's trail
//!
//! [`crate::audit`] is the SERVER's trail: one chain over every decision the
//! gRPC gate makes, read by the operator (`Admin.AuditTrail`). This is the
//! ACCOUNT's: what happened inside `acme` — who logged in, which passkey was
//! added, which repository went public, who approved the operator's look —
//! read ONLY by acme's admins. The operator sees neither its entries nor its
//! names (the operator's aggregates, `Enterprise.Aggregates`); the two logs
//! are separate precisely so that the one an operator reads never has to be
//! filtered to keep that promise.
//!
//! # The shape
//!
//! ```text
//! entry.hash = SHA-256( "gunnar-account-audit-v1" ‖ account ‖ prev ‖ seq ‖ at ‖ kind ‖
//! actor ‖ actor_role ‖ target ‖ detail ‖ outcome ) (each length-prefixed)
//! ```
//!
//! The same construction as [`crate::audit`] — a chain whose tip commits to
//! the whole history, fields length-prefixed so no value can move a boundary
//! — with the ACCOUNT hashed in, so a line copied from one account's log into
//! another's is a broken chain and not a quiet forgery.
//!
//! # Stored as one control document per account
//!
//! `refs/gunnar/account-audit/<account>`, in the control store beside the
//! identity table: replicated to the twin and carried by a restore with
//! everything else, with no new file the twin's snapshot would have to learn
//! about. Entries are bucketed by UTC DAY in a [`ShardedMap`], so an append
//! rewrites one day's bucket and two trie nodes, and the document root holds
//! the chain's head and the list of days that have entries (≤ 366).
//!
//! # Retention, and why it does not break the chain
//!
//! [`DEFAULT_RETENTION_DAYS`]: a year. The append that crosses the line drops
//! every day past it and records the last dropped entry as the ANCHOR — its
//! seq and its hash — so the first kept entry's `prev` is checked against the
//! anchor instead of [`GENESIS`]. A pruned log still verifies end to end.
//!
//! What this does NOT promise: that the operator's disk never held the bytes.
//! The control store keeps every version of every document (its own history),
//! so a pruned day still exists in old blobs on the operator's box until the
//! store itself is compacted. The promise is the API's: nobody but the
//! account's admins is SHOWN the log (`.nornir/account-audit-operator-privacy.md`).
use std::collections::HashMap;
use std::sync::atomic::{AtomicU32, Ordering};
use std::sync::{Arc, Mutex};
use sha2::Digest as _;
use crate::control::ControlStore;
use crate::doc::{impl_doc, DocRef, ShardedMap};
use crate::error::Result;
use crate::flow::{day_of, iso8601_utc, DAY_MS};
/// The prefix every account's log document is named under.
pub const ACCOUNT_AUDIT_REF_PREFIX: &str = "refs/gunnar/account-audit/";
/// **A year**: how long an entry is kept unless the server is told otherwise.
pub const DEFAULT_RETENTION_DAYS: u32 = 365;
/// The `prev` of an account's first entry: 64 zeros, as [`crate::audit::GENESIS`].
pub const GENESIS: &str = crate::audit::GENESIS;
/// Domain separation: a hash from this chain is not a hash from the server's.
const DOMAIN: &str = "gunnar-account-audit-v1";
/// The most entries one query answers, whatever it asks for.
pub const MAX_PAGE: usize = 1000;
/// Did the thing the entry records happen, or was it refused?
#[derive(Clone, Copy, PartialEq, Eq, Debug)]
pub enum Outcome {
/// It happened.
Ok,
/// It was asked for and refused (a login that failed, a name the filter
/// refused, a member trying to make a repository public).
Refused,
}
impl Outcome {
/// `ok` or `refused` — the word the entry carries.
pub fn as_str(self) -> &'static str {
match self {
Outcome::Ok => "ok",
Outcome::Refused => "refused",
}
}
}
/// **One event, as a caller hands it over.** Borrowed: every field is already
/// a `&str` where it happens.
#[derive(Clone, Copy, Debug)]
pub struct AccountEvent<'a> {
/// What happened, `family.verb`: `login`, `credential.added`,
/// `visibility.changed`, `repo.deleted`, `operator_access.approved` …
pub kind: &'a str,
/// The principal that did it; empty for the system (a retention sweep, an
/// expiry).
pub actor: &'a str,
/// What the actor is TO THIS ACCOUNT: `admin`, `member`, `operator`,
/// `machine`, `outsider`, `system`.
pub actor_role: &'a str,
/// What it was done to: a repository path, a principal, a credential's
/// label, a namespace.
pub target: &'a str,
/// One sentence more, or empty.
pub detail: &'a str,
/// Done or refused.
pub outcome: Outcome,
}
/// **One entry, as it is stored and as the admin reads it.**
///
/// Field order is the serialisation order; `deny_unknown_fields` so an entry
/// written by a newer gunnar is refused loudly rather than verified against
/// fields this build cannot see.
#[derive(Clone, PartialEq, Eq, Debug, serde::Serialize, serde::Deserialize)]
#[serde(deny_unknown_fields)]
pub struct AccountEntry {
/// 0 for the account's first entry, then +1 each time, never reused.
pub seq: u64,
/// When, Unix milliseconds (this server's clock).
pub at_unix_ms: i64,
/// [`AccountEvent::kind`].
pub kind: String,
/// [`AccountEvent::actor`].
pub actor: String,
/// [`AccountEvent::actor_role`].
pub actor_role: String,
/// [`AccountEvent::target`].
pub target: String,
/// [`AccountEvent::detail`].
pub detail: String,
/// `ok` or `refused`.
pub outcome: String,
/// The previous entry's hash, the anchor's, or [`GENESIS`].
pub prev: String,
/// This entry's own hash ([`AccountEntry::compute_hash`]).
pub hash: String,
}
impl AccountEntry {
/// The hash this entry SHOULD carry in `account`, from everything else in it.
pub fn compute_hash(&self, account: &str) -> String {
let mut hasher = sha2::Sha256::new();
let mut field = |bytes: &[u8]| {
hasher.update((bytes.len() as u64).to_le_bytes());
hasher.update(bytes);
};
field(DOMAIN.as_bytes());
field(account.as_bytes());
field(self.prev.as_bytes());
field(&self.seq.to_le_bytes());
field(&self.at_unix_ms.to_le_bytes());
field(self.kind.as_bytes());
field(self.actor.as_bytes());
field(self.actor_role.as_bytes());
field(self.target.as_bytes());
field(self.detail.as_bytes());
field(self.outcome.as_bytes());
hex::encode(hasher.finalize())
}
/// Does `kind` (a family like `visibility`, or a whole kind) match this entry?
fn is_kind(&self, kind: &str) -> bool {
self.kind == kind
|| (self.kind.len() > kind.len()
&& self.kind.starts_with(kind)
&& self.kind.as_bytes()[kind.len()] == b'.')
}
}
/// **What the admin asks for**: every filter optional, newest first.
#[derive(Clone, Debug, Default, PartialEq, Eq)]
pub struct AuditQuery {
/// Kinds or families (`login`, `visibility`, `repo.deleted`); empty = all.
pub kinds: Vec<String>,
/// Only this actor; empty = anyone.
pub actor: String,
/// At or after; 0 = from the start.
pub since_unix_ms: i64,
/// At or before; 0 = up to now.
pub until_unix_ms: i64,
/// A substring of the target, the detail, the actor or the kind, any case.
pub text: String,
/// Only entries OLDER than this seq (the previous page's cursor).
pub before_seq: Option<u64>,
/// How many; 0 = 50. Capped at [`MAX_PAGE`].
pub limit: usize,
}
impl AuditQuery {
fn matches(&self, entry: &AccountEntry, text_lower: &str) -> bool {
if !self.kinds.is_empty() && !self.kinds.iter().any(|k| entry.is_kind(k)) {
return false;
}
if !self.actor.is_empty() && entry.actor != self.actor {
return false;
}
if self.since_unix_ms > 0 && entry.at_unix_ms < self.since_unix_ms {
return false;
}
if self.until_unix_ms > 0 && entry.at_unix_ms > self.until_unix_ms {
return false;
}
if !text_lower.is_empty() {
let hit = [&entry.target, &entry.detail, &entry.actor, &entry.kind]
.iter()
.any(|field| field.to_lowercase().contains(text_lower));
if !hit {
return false;
}
}
true
}
}
/// One page of the log.
#[derive(Clone, Debug, Default, PartialEq, Eq)]
pub struct AuditPage {
/// Newest first.
pub entries: Vec<AccountEntry>,
/// Ask again with `before_seq` = this for the next (older) page; `None`
/// when this page was the last.
pub next_before_seq: Option<u64>,
}
/// **Does the chain hold?**
#[derive(Clone, Debug, Default, PartialEq, Eq)]
pub struct ChainReport {
/// The chain verifies from its anchor (or genesis) to its tip.
pub holds: bool,
/// How many entries were checked.
pub entries: u64,
/// The newest entry's hash (the anchor's, or genesis, when there is none).
pub tip_hash: String,
/// The last entry retention dropped, whose hash the first kept entry names.
pub anchor_seq: Option<u64>,
/// The first entry that does not hold, when one does not.
pub broken_at_seq: Option<u64>,
/// Why, in a sentence; empty when it holds.
pub what: String,
}
/// **Verify `entries` (oldest first) as `account`'s chain**, from `anchor`
/// (`(seq, hash)` of the last pruned entry) or from genesis. Pure: the tamper
/// tests drive it with entries they edited.
pub fn verify_entries(
account: &str,
anchor: Option<(u64, &str)>,
entries: &[AccountEntry],
) -> ChainReport {
let (mut want_seq, mut prev) = match anchor {
Some((seq, hash)) => (seq + 1, hash.to_owned()),
None => (0, GENESIS.to_owned()),
};
let mut report = ChainReport {
holds: true,
entries: 0,
tip_hash: prev.clone(),
anchor_seq: anchor.map(|(seq, _)| seq),
broken_at_seq: None,
what: String::new(),
};
for entry in entries {
let what = if entry.seq != want_seq {
Some(format!(
"entry {} is where entry {want_seq} should be: one is missing or out of order",
entry.seq
))
} else if entry.prev != prev {
Some(format!(
"entry {} does not name the entry before it",
entry.seq
))
} else if entry.hash != entry.compute_hash(account) {
Some(format!(
"entry {} was changed after it was written",
entry.seq
))
} else {
None
};
if let Some(what) = what {
report.holds = false;
report.broken_at_seq = Some(entry.seq);
report.what = what;
return report;
}
report.entries += 1;
want_seq = entry.seq + 1;
prev = entry.hash.clone();
report.tip_hash = prev.clone();
}
report
}
/// The chain's head, in the document root.
#[derive(Clone, Debug, Default, serde::Serialize, serde::Deserialize)]
#[serde(deny_unknown_fields)]
struct Head {
/// The seq the next entry gets.
#[serde(default)]
next_seq: u64,
/// The newest entry's hash; empty = none yet (genesis, or the anchor).
#[serde(default)]
tip: String,
/// The last entry retention dropped.
#[serde(default, skip_serializing_if = "Option::is_none")]
anchor_seq: Option<u64>,
#[serde(default, skip_serializing_if = "String::is_empty")]
anchor_hash: String,
/// The UTC days (since the epoch) that hold entries, ascending.
#[serde(default)]
days: Vec<i64>,
}
impl Head {
fn prev(&self) -> String {
if !self.tip.is_empty() {
self.tip.clone()
} else if !self.anchor_hash.is_empty() {
self.anchor_hash.clone()
} else {
GENESIS.to_owned()
}
}
}
#[derive(Clone, Default, serde::Serialize, serde::Deserialize)]
#[serde(deny_unknown_fields)]
struct AccountAuditDoc {
#[serde(default)]
head: Head,
/// Day (zero-padded, so keys sort as days do) → that day's entries, oldest first.
#[serde(default)]
days: ShardedMap<Vec<AccountEntry>>,
}
impl_doc!(AccountAuditDoc { days });
fn day_key(day: i64) -> String {
format!("{day:010}")
}
/// **Every account's log.** Cheap to share; each account's document is opened
/// once and its handle kept, so its cache and its writers' queue are shared.
pub struct AccountAuditLog {
store: Arc<dyn ControlStore>,
docs: Mutex<HashMap<String, DocRef<AccountAuditDoc>>>,
retention_days: AtomicU32,
}
impl std::fmt::Debug for AccountAuditLog {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
f.debug_struct("AccountAuditLog")
.field("retention_days", &self.retention_days())
.finish()
}
}
impl AccountAuditLog {
/// The logs held in `store` (the server's control store).
pub fn new(store: Arc<dyn ControlStore>) -> Result<Self> {
Ok(AccountAuditLog {
store,
docs: Mutex::new(HashMap::new()),
retention_days: AtomicU32::new(DEFAULT_RETENTION_DAYS),
})
}
/// Keep `days` days instead of a year (at least one).
pub fn with_retention_days(self, days: u32) -> Self {
self.retention_days.store(days.max(1), Ordering::Relaxed);
self
}
/// How many days an entry is kept.
pub fn retention_days(&self) -> u32 {
self.retention_days.load(Ordering::Relaxed)
}
fn doc(&self, account: &str) -> Result<DocRef<AccountAuditDoc>> {
crate::sandbox::check_component(account, account)?;
let mut docs = self.docs.lock().unwrap_or_else(|p| p.into_inner());
if let Some(doc) = docs.get(account) {
return Ok(doc.clone());
}
let doc = DocRef::named(
Arc::clone(&self.store),
std::borrow::Cow::Owned(format!("{ACCOUNT_AUDIT_REF_PREFIX}{account}")),
)?;
docs.insert(account.to_owned(), doc.clone());
Ok(doc)
}
/// **Append one event to `account`'s log**, pruning past the retention
/// first. Returns the entry as stored.
pub fn append(
&self,
account: &str,
event: AccountEvent<'_>,
at_unix_ms: i64,
) -> Result<AccountEntry> {
let doc = self.doc(account)?;
let cutoff_day = day_of(at_unix_ms) - i64::from(self.retention_days());
doc.update(|doc| {
// Retention first, so the anchor is set before the new entry names its prev.
Self::drop_before(doc, cutoff_day)?;
let mut entry = AccountEntry {
seq: doc.head.next_seq,
at_unix_ms,
kind: event.kind.to_owned(),
actor: event.actor.to_owned(),
actor_role: event.actor_role.to_owned(),
target: event.target.to_owned(),
detail: event.detail.to_owned(),
outcome: event.outcome.as_str().to_owned(),
prev: doc.head.prev(),
hash: String::new(),
};
entry.hash = entry.compute_hash(account);
// A clock that stepped back files the entry under the newest day
// already held, so the days stay in seq order.
let day = day_of(at_unix_ms).max(doc.head.days.last().copied().unwrap_or(i64::MIN));
let key = day_key(day);
let mut bucket = doc.days.get(&key)?.unwrap_or_default();
bucket.push(entry.clone());
doc.days.insert(key, bucket)?;
if doc.head.days.last() != Some(&day) {
doc.head.days.push(day);
}
doc.head.next_seq += 1;
doc.head.tip = entry.hash.clone();
Ok(entry)
})
}
/// Drop every day before `cutoff_day`, moving the anchor to the last
/// entry dropped. Returns how many entries went.
fn drop_before(doc: &mut AccountAuditDoc, cutoff_day: i64) -> Result<u64> {
let mut dropped_count = 0;
while let Some(&oldest) = doc.head.days.first() {
if oldest >= cutoff_day {
break;
}
if let Some(dropped) = doc.days.remove(&day_key(oldest))? {
dropped_count += dropped.len() as u64;
if let Some(last) = dropped.last() {
doc.head.anchor_seq = Some(last.seq);
doc.head.anchor_hash = last.hash.clone();
if doc.head.tip == last.hash {
doc.head.tip.clear();
}
}
}
doc.head.days.remove(0);
}
Ok(dropped_count)
}
/// **The retention sweep for one account**, at `now`: what an append does
/// first, for an account nothing has happened in for a while. The server
/// runs it over every account once a day. Returns how many entries went;
/// a sweep with nothing to drop writes nothing.
pub fn prune(&self, account: &str, now_unix_ms: i64) -> Result<u64> {
let doc = self.doc(account)?;
let cutoff_day = day_of(now_unix_ms) - i64::from(self.retention_days());
if doc
.load()?
.head
.days
.first()
.is_none_or(|&oldest| oldest >= cutoff_day)
{
return Ok(0);
}
doc.update(|doc| Self::drop_before(doc, cutoff_day))
}
/// **One page of `account`'s log**, newest first.
pub fn query(&self, account: &str, query: &AuditQuery) -> Result<AuditPage> {
let limit = match query.limit {
0 => 50,
n => n.min(MAX_PAGE),
};
let mut page = AuditPage::default();
let mut more = false;
self.walk_newest_first(account, query, &mut |entry| {
if page.entries.len() == limit {
more = true;
return false;
}
page.entries.push(entry);
true
})?;
if more {
page.next_before_seq = page.entries.last().map(|e| e.seq);
}
Ok(page)
}
/// **Every entry the query matches**, newest first, no page (the export).
pub fn all(&self, account: &str, query: &AuditQuery) -> Result<Vec<AccountEntry>> {
let mut out = Vec::new();
self.walk_newest_first(account, query, &mut |entry| {
out.push(entry);
true
})?;
Ok(out)
}
fn walk_newest_first(
&self,
account: &str,
query: &AuditQuery,
take: &mut dyn FnMut(AccountEntry) -> bool,
) -> Result<()> {
let doc = self.doc(account)?.load()?;
let since = query.since_unix_ms;
let text = query.text.to_lowercase();
for &day in doc.head.days.iter().rev() {
if query.until_unix_ms > 0 && day * DAY_MS > query.until_unix_ms {
continue;
}
if (day + 1) * DAY_MS <= since {
break;
}
let Some(bucket) = doc.days.get(&day_key(day))? else {
continue;
};
for entry in bucket.into_iter().rev() {
if query.before_seq.is_some_and(|before| entry.seq >= before) {
continue;
}
if !query.matches(&entry, &text) {
continue;
}
if !take(entry) {
return Ok(());
}
}
}
Ok(())
}
/// **Does `account`'s chain hold**, from its anchor to its tip, and does
/// the tip the head records equal the last entry's hash?
pub fn verify(&self, account: &str) -> Result<ChainReport> {
let doc = self.doc(account)?.load()?;
let mut entries = Vec::new();
for &day in &doc.head.days {
if let Some(bucket) = doc.days.get(&day_key(day))? {
entries.extend(bucket);
}
}
let anchor = doc
.head
.anchor_seq
.map(|seq| (seq, doc.head.anchor_hash.as_str()));
let mut report = verify_entries(account, anchor, &entries);
if report.holds && report.tip_hash != doc.head.prev() {
report.holds = false;
report.what =
"the head names a tip the entries do not end at: entries were cut off".to_owned();
report.broken_at_seq = Some(doc.head.next_seq.saturating_sub(1));
}
Ok(report)
}
/// The tip and the number of entries ever written, without a walk.
pub fn head(&self, account: &str) -> Result<(u64, String)> {
let doc = self.doc(account)?.load()?;
Ok((doc.head.next_seq, doc.head.prev()))
}
}
/// A CSV field: quoted when it must be, and a leading `= + - @` (a formula a
/// spreadsheet would RUN) defused with a `'`.
fn csv_field(out: &mut String, text: &str) {
let defused;
let text = if text.starts_with(['=', '+', '-', '@', '\t', '\r']) {
defused = format!("'{text}");
defused.as_str()
} else {
text
};
if text.contains([',', '"', '\n', '\r']) {
out.push('"');
out.push_str(&text.replace('"', "\"\""));
out.push('"');
} else {
out.push_str(text);
}
}
/// **The export as CSV** (RFC 4180, a header line, UTC timestamps).
pub fn to_csv(entries: &[AccountEntry]) -> String {
let mut out = String::from("seq,at_utc,kind,actor,actor_role,target,detail,outcome,hash\n");
for e in entries {
let fields = [
e.seq.to_string(),
iso8601_utc(e.at_unix_ms),
e.kind.clone(),
e.actor.clone(),
e.actor_role.clone(),
e.target.clone(),
e.detail.clone(),
e.outcome.clone(),
e.hash.clone(),
];
for (i, field) in fields.iter().enumerate() {
if i > 0 {
out.push(',');
}
csv_field(&mut out, field);
}
out.push('\n');
}
out
}
/// **The export as JSON**: an array of entries, each with its UTC time beside
/// the Unix milliseconds.
pub fn to_json(entries: &[AccountEntry]) -> String {
let rows: Vec<serde_json::Value> = entries
.iter()
.map(|e| {
let mut row = serde_json::to_value(e).unwrap_or(serde_json::Value::Null);
if let Some(map) = row.as_object_mut() {
map.insert(
"at_utc".into(),
serde_json::Value::String(iso8601_utc(e.at_unix_ms)),
);
}
row
})
.collect();
serde_json::to_string_pretty(&rows).unwrap_or_else(|_| "[]".to_owned())
}