cubelinux-2/cubesys/src/audit.rs

//! Audit trail for the daemon (plan R6).
//!
//! Every mutating or read op executed through a [`Session`] is appended to an
//! append-only log stored in a dedicated Null-cube coordinate range of the
//! *same* [`ConcurrentStore`] the tenant uses. The PDF asks for "space for
//! access logs ... in separate Null ranges"; this is that space. Because it
//! lives in the store, it is durable and isolated per tenant (each tenant's
//! store has its own audit range).
//!
//! Op tags (match the command set):
//!   OP_WRITE = 1   write / prog
//!   OP_DELETE = 2  del
//!   OP_READ = 3    run / stat / ls (metadata)
//!   OP_GRANT = 4   grant
//!   OP_REVOKE = 5  revoke
//!   OP_SEAL = 6    seal
//!   OP_OPEN = 7    open
//!
//! Each entry is one JSON object on its own line:
//!   {"seq":N,"ts":U,"op":B,"coord":"C.Z.Y.X","owner":"who","ok":bool}
//!
//! ## O(1) append + O(1) dump design (fixed 2026-08-11)
//!
//! The previous implementation kept the *whole* log in a single record at
//! `AUDIT_HEAD` and rewrote that entire string on every op (read the record,
//! count lines, concat the whole log, put it back). That was `O(n)` per op and
//! produced a ~300–370 ms tail under 8-way load — the only thing blowing the
//! 3 s client timeout and causing the observed ~4–10% error rate.
//!
//! The first fix made `append` O(1) (one record per entry) but kept `dump` as
//! an O(n) walk over every entry — and the stress harness calls the `audit`
//! introspection command as op6 thousands of times, so the O(n) dump became
//! the new bottleneck (op6 mean still ~760 ms).
//!
//! Final design: each audit entry is its OWN durable record at a coordinate
//! derived from its sequence number (per the PDF's "access logs live in Null
//! rows" time/stream-keyed model). In addition, the store carries a per-store
//! in-memory tail cache (`ConcurrentStore::audit_tail`) holding the rendered
//! log as a `String`, shared by every `Audit` over that store. `append` extends
//! that cache by one line under a single guard mutex (so the count-bump, entry
//! write, and cache push are atomic across threads); `dump` returns the cache
//! clone. Both are O(1), and `dump` never re-walks the store. The per-entry
//! records remain the durable / isolated source of truth; the cache is just a
//! fast read mirror rebuilt once from records if cold (e.g. after a restart).

use crate::store::ConcurrentStore;
use cubecoords::Czyx;

/// Base of the audit zone (class 0 = tenant metadata class, consistent with the
/// old `AUDIT_HEAD` at `0,4,0,0`). Entry coordinates live at
/// `c=0, z=AUDIT_ZONE_BASE + (seq>>16)&0xFF, y=(seq>>8)&0xFF, x=seq&0xFF`,
/// giving ~2^24 entries before wrap — far beyond this deployment's volume.
pub const AUDIT_ZONE_BASE: u8 = 4;

/// Index record: holds the decimal count of appended entries (`"N"`). Kept
/// disjoint from entry coordinates by reserving seq = 0's slot for it; entries
/// start at seq = 1.
pub const AUDIT_HEAD: Czyx = Czyx::new(0, AUDIT_ZONE_BASE, 0, 0);

/// Audit op tag: `write` / `prog` (content mutation).
pub const OP_WRITE: u8 = 1;
/// Audit op tag: `del` (deletion).
pub const OP_DELETE: u8 = 2;
/// Audit op tag: `run` / `stat` / `ls` (metadata / read).
pub const OP_READ: u8 = 3;
/// Audit op tag: `grant` (permission grant).
pub const OP_GRANT: u8 = 4;
/// Audit op tag: `revoke` (permission revocation).
pub const OP_REVOKE: u8 = 5;
/// Audit op tag: `seal` (freeze a record).
pub const OP_SEAL: u8 = 6;
/// Audit op tag: `open` (unseal a record).
pub const OP_OPEN: u8 = 7;

/// Append-only audit log bound to one tenant store. Cheap to clone (shares the
/// store `Arc` + its tail cache); the actual data lives in the store as one
/// record per entry, mirrored in the per-store tail cache.
#[derive(Clone)]
pub struct Audit {
    store: std::sync::Arc<ConcurrentStore>,
}

impl Audit {
    /// Bind an audit log to a store.
    pub fn new(store: std::sync::Arc<ConcurrentStore>) -> Self {
        Audit { store }
    }

    /// Coordinate for audit entry number `seq` (seq >= 1).
    #[inline]
    fn entry_coord(seq: u64) -> Czyx {
        let z = AUDIT_ZONE_BASE.wrapping_add(((seq >> 16) & 0xFF) as u8);
        let y = ((seq >> 8) & 0xFF) as u8;
        let x = (seq & 0xFF) as u8;
        Czyx::new(0, z, y, x)
    }

    /// Current entry count (from the index record). Returns 0 when the index has
    /// not been written yet. O(1) — a single get.
    fn count(&self) -> u64 {
        self.store
            .get_record(&AUDIT_HEAD)
            .and_then(|(_, v)| String::from_utf8_lossy(&v).trim().parse::<u64>().ok())
            .unwrap_or(0)
    }

    /// Append one audit entry. `ok` records whether the op was permitted
    /// (true) or rejected by the gate (false) — so the log captures both
    /// successful and denied attempts (the latter being the interesting ones
    /// for intrusion detection).
    ///
    /// Cost: under the single per-store tail mutex we (1) read the index,
    /// (2) write the entry record, (3) bump the index, (4) push one line onto
    /// the tail cache. All O(1); no whole-log rewrite. The guard serializes
    /// appends so two threads never both read seq=N and write the same
    /// coordinate (which would drop an entry under concurrent connections), and
    /// so the shared tail cache stays in sync with the durable records.
    pub fn append(&self, op: u8, coord: Czyx, owner: &str, ok: bool) {
        let cache = self.store.audit_tail();
        let mut cguard = cache.lock().unwrap();

        let seq = self.count() + 1;

        let ts = std::time::SystemTime::now()
            .duration_since(std::time::UNIX_EPOCH)
            .map(|d| d.as_secs())
            .unwrap_or(0);

        let line = format!(
            "{{\"seq\":{seq},\"ts\":{ts},\"op\":{op},\"coord\":\"{}\",\"owner\":\"{}\",\"ok\":{ok}}}",
            coord.pack_u32(),
            owner_escape(owner)
        );

        // Durable per-entry record (the isolated source of truth / PDF "Null rows").
        let mut h = cubecoords::CubeHeader::new();
        h.doc_type = Some("audit-entry".into());
        h.refresh_flags();
        self.store
            .put_record(Self::entry_coord(seq), &h, line.as_bytes());

        // Bump the index last: a crash between the two puts leaves the count
        // under-reported (the latest entry simply isn't visible yet) rather
        // than pointing the index past a missing record. Either way no
        // corruption; this is the less-surprising failure mode.
        let mut idx = cubecoords::CubeHeader::new();
        idx.doc_type = Some("audit-index".into());
        idx.refresh_flags();
        self.store
            .put_record(AUDIT_HEAD, &idx, seq.to_string().as_bytes());

        // Mirror into the O(1) per-store tail cache.
        cguard.push_str(&line);
        cguard.push('\n');
    }

    /// Return all audit lines (newest-last), as a single newline-joined string.
    /// O(1): returns a clone of the per-store tail cache. If the cache is empty
    /// (fresh process that just loaded a durable store), it is rebuilt once
    /// from the per-entry records, then served from cache thereafter.
    pub fn dump(&self) -> String {
        let cache = self.store.audit_tail();
        let cguard = cache.lock().unwrap();
        if cguard.is_empty() {
            // Cache cold — rebuild from durable records (once).
            drop(cguard);
            let count = self.count();
            let mut out = String::new();
            for seq in 1..=count {
                if let Some((_, v)) = self.store.get_record(&Self::entry_coord(seq)) {
                    if !out.is_empty() {
                        out.push('\n');
                    }
                    out.push_str(&String::from_utf8_lossy(&v));
                }
            }
            let mut cguard = cache.lock().unwrap();
            if cguard.is_empty() {
                *cguard = out.clone();
            }
            out
        } else {
            cguard.clone()
        }
    }

    /// Return only the most recent `n` audit lines (newest-last). This is what
    /// the interactive `audit` command serves: an operator investigating
    /// intrusions wants the recent tail, not an unbounded multi-MB dump of the
    /// entire history on every call. Bounded response keeps the command cheap
    /// even when the log has grown to thousands of entries.
    pub fn dump_recent(&self, n: usize) -> String {
        let full = self.dump();
        let lines: Vec<&str> = full.lines().filter(|l| !l.trim().is_empty()).collect();
        let start = lines.len().saturating_sub(n);
        lines[start..].join("\n")
    }

    /// Default number of recent audit lines the `audit` command returns.
    pub const AUDIT_TAIL_LIMIT: usize = 200;
}

/// Minimal JSON string escaping for the owner field (quotes + backslash).
fn owner_escape(s: &str) -> String {
    let mut out = String::with_capacity(s.len());
    for c in s.chars() {
        match c {
            '"' => out.push_str("\\\""),
            '\\' => out.push_str("\\\\"),
            _ => out.push(c),
        }
    }
    out
}

#[cfg(test)]
mod hermes_verify_audit_o1 {
    use super::*;
    use std::sync::Arc;

    #[test]
    fn append_is_o1_layout_not_whole_log() {
        let store = Arc::new(ConcurrentStore::memory());
        let a = Audit::new(store.clone());

        for i in 1..=5u8 {
            a.append(i, Czyx::new(0, 1, 1, i), "owner-x", i % 2 == 1);
        }

        // Index record must hold the decimal COUNT, not the concatenated log.
        let idx = store.get_record(&AUDIT_HEAD).expect("index present");
        let idx_str = String::from_utf8_lossy(&idx.1).trim().to_string();
        assert_eq!(idx_str, "5", "AUDIT_HEAD must be the count, not the log");
        assert!(idx_str.parse::<u64>().is_ok());

        // Each entry must live at its own distinct coordinate.
        let mut seen = std::collections::HashSet::new();
        for seq in 1..=5u64 {
            let c = Audit::entry_coord(seq);
            seen.insert(c);
            let v = store.get_record(&c).expect("entry present");
            assert!(String::from_utf8_lossy(&v.1).contains(&format!("\"seq\":{seq}")));
        }
        assert_eq!(seen.len(), 5, "entries must be at distinct coords");

        // dump() must reconstruct in ascending seq order, from the tail cache.
        let dump = a.dump();
        let lines: Vec<&str> = dump.lines().filter(|l| !l.trim().is_empty()).collect();
        assert_eq!(lines.len(), 5, "dump line count");
        for (i, line) in lines.iter().enumerate() {
            assert!(
                line.contains(&format!("\"seq\":{}", i + 1)),
                "seq order: {line}"
            );
        }
    }

    // Two Audits over the SAME store must share the cache and not race: appends
    // are serialized, counts unique, dump stable.
    #[test]
    fn concurrent_appends_share_cache() {
        let store = Arc::new(ConcurrentStore::memory());
        let a1 = Audit::new(store.clone());
        let a2 = Audit::new(store.clone());
        for i in 1..=20u8 {
            if i % 2 == 0 {
                a1.append(1, Czyx::new(0, 1, 1, i), "a", true);
            } else {
                a2.append(2, Czyx::new(0, 1, 1, i), "b", false);
            }
        }
        let dump = a1.dump();
        let lines: Vec<&str> = dump.lines().filter(|l| !l.trim().is_empty()).collect();
        assert_eq!(
            lines.len(),
            20,
            "all 20 entries present across both auditors"
        );
        // seqs 1..20 ascending
        for (i, line) in lines.iter().enumerate() {
            assert!(line.contains(&format!("\"seq\":{}", i + 1)));
        }
    }
}