gunnar_core/

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())
}