cheers_server/
audit.rs

1//! Centralized audit table — the [`AuditStore`] trait.
2//!
3//! Cheers's audit table is the durable, queryable copy of every MCP-mediated
4//! action kamaji observes on its host. Kamaji retains a local JSONL as
5//! the source of truth and forwards batches here via `POST /audit/ingest`;
6//! cheers's responsibility ends at "accepted and durable on cheers's side"
7//! (see `.yah/docs/working/mcp-auth-and-ownership.md` §Audit ingest).
8//!
9//! Append-only. There is no in-place edit or delete — the only writes are
10//! [`AuditStore::insert_batch`]. Reads (paged by `on_behalf_of`) land in F14
11//! atop this same trait.
12//!
13//! Record shape per W159 §Audit journal: `{ at, sub, act, camp_id, aud,
14//! method, scope, result, request_id }`. The wire and storage shapes match
15//! one-to-one — cheers doesn't repackage what kamaji sends, it just
16//! durably appends. The only field cheers contributes is the row `id` and
17//! the `ingested_at` server timestamp.
18
19use std::collections::VecDeque;
20use std::sync::{Arc, Mutex};
21
22use async_trait::async_trait;
23use serde::{Deserialize, Serialize};
24
25use cheers_core::{Actor, PrincipalId, Scope, StoreError};
26
27/// Why an [`AuditRecord`] failed validation before reaching the store.
28///
29/// The HTTP layer surfaces these as `400 audit_invalid`. They are the
30/// "forbidden shape" the F13 verify item calls out — a malformed batch is
31/// rejected wholesale so kamaji's bounded-backoff retry sees a clean
32/// 4xx, not a partial commit.
33#[derive(Debug, Clone, PartialEq, Eq, thiserror::Error)]
34pub enum AuditValidationError {
35    #[error("audit record: `at` must be > 0; got {0}")]
36    NonPositiveAt(i64),
37    #[error("audit record: `aud` must not be empty")]
38    EmptyAud,
39    #[error("audit record: `method` must not be empty")]
40    EmptyMethod,
41    #[error("audit record: `result` must not be empty")]
42    EmptyResult,
43    #[error("audit record: `request_id` must not be empty")]
44    EmptyRequestId,
45}
46
47/// One audit record — verbatim with W159 §Audit journal.
48///
49/// Required fields: `at`, `sub`, `aud`, `method`, `result`, `request_id`.
50/// `scope` MAY be empty (e.g. a call rejected before scope-matching). `act`
51/// and `camp_id` are set when an agent acted on the subject's behalf and/or
52/// the call was scoped to a camp; both `None` for a direct user-principal
53/// action with no camp context.
54///
55/// Mirrors the conditional-claim shape of [`McpClaims`](cheers_core::McpClaims)
56/// so a verified token's metadata can flow into the record without lossy
57/// repackaging.
58#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
59#[non_exhaustive]
60pub struct AuditRecord {
61    /// Unix-seconds timestamp the action happened (kamaji's local clock).
62    pub at: i64,
63    /// Acting principal — the `sub` claim from the verified token.
64    pub sub: PrincipalId,
65    #[serde(default, skip_serializing_if = "Option::is_none")]
66    pub act: Option<Actor>,
67    #[serde(default, skip_serializing_if = "Option::is_none")]
68    pub camp_id: Option<String>,
69    /// Target resource URI (the verified token's `aud`).
70    pub aud: String,
71    /// Free-form method identifier (`POST /cloud/deploy`, `mcp.tools/call`, …)
72    /// — kamaji supplies a stable string per call shape.
73    pub method: String,
74    /// Scope list the call presented. Stored as a typed `Vec<Scope>` so a
75    /// future wildcard or unknown-scope ingest is rejected at parse time
76    /// (per composition rule 1 — no wildcards on the wire).
77    #[serde(default)]
78    pub scope: Vec<Scope>,
79    /// Outcome string kamaji produced (`allow` | `deny` | `error` is the
80    /// usual taxonomy, but cheers does not constrain it — see audit-reader
81    /// docs in F14 for the canonical vocabulary). Cheers is durable storage,
82    /// not the source of result semantics.
83    pub result: String,
84    /// Correlator into kamaji's local JSONL — same id appears in both
85    /// places so an operator can match a cheers row to kamaji's source.
86    pub request_id: String,
87}
88
89impl AuditRecord {
90    /// Construct + validate. Returns the same error variants the wire layer
91    /// surfaces as `400 audit_invalid`.
92    #[allow(clippy::too_many_arguments)]
93    pub fn new(
94        at: i64,
95        sub: PrincipalId,
96        act: Option<Actor>,
97        camp_id: Option<String>,
98        aud: impl Into<String>,
99        method: impl Into<String>,
100        scope: Vec<Scope>,
101        result: impl Into<String>,
102        request_id: impl Into<String>,
103    ) -> Result<Self, AuditValidationError> {
104        let rec = Self {
105            at,
106            sub,
107            act,
108            camp_id,
109            aud: aud.into(),
110            method: method.into(),
111            scope,
112            result: result.into(),
113            request_id: request_id.into(),
114        };
115        rec.validate()?;
116        Ok(rec)
117    }
118
119    /// Re-check the wire-shape invariants. Called from [`Self::new`]; the
120    /// HTTP layer calls it again on records that arrived via `Deserialize`
121    /// so a deserialized record can't side-step the constructor.
122    pub fn validate(&self) -> Result<(), AuditValidationError> {
123        if self.at <= 0 {
124            return Err(AuditValidationError::NonPositiveAt(self.at));
125        }
126        if self.aud.is_empty() {
127            return Err(AuditValidationError::EmptyAud);
128        }
129        if self.method.is_empty() {
130            return Err(AuditValidationError::EmptyMethod);
131        }
132        if self.result.is_empty() {
133            return Err(AuditValidationError::EmptyResult);
134        }
135        if self.request_id.is_empty() {
136            return Err(AuditValidationError::EmptyRequestId);
137        }
138        Ok(())
139    }
140}
141
142/// One row in the audit table — an [`AuditRecord`] plus the bits cheers
143/// contributes (`id` and `ingested_at`).
144#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
145#[non_exhaustive]
146pub struct AuditRow {
147    pub id: String,
148    pub record: AuditRecord,
149    /// Unix-seconds timestamp cheers received and durably appended the row.
150    /// Distinct from `record.at` (kamaji's clock) — having both lets an
151    /// operator see ingest latency without joining clocks at query time.
152    pub ingested_at: i64,
153}
154
155impl AuditRow {
156    pub fn new(id: String, record: AuditRecord, ingested_at: i64) -> Self {
157        Self {
158            id,
159            record,
160            ingested_at,
161        }
162    }
163}
164
165/// Append-only persistence for audit records.
166///
167/// Only one mutation method — [`insert_batch`](Self::insert_batch). A batch
168/// is rejected wholesale if any record fails [`AuditRecord::validate`]; no
169/// partial commits, so kamaji's retry-with-backoff sees a clean 4xx.
170///
171/// Read APIs (paged by `on_behalf_of` + filters) land alongside F14.
172#[async_trait]
173pub trait AuditStore: Send + Sync {
174    /// Insert a batch of records atomically. The impl assigns each row a
175    /// fresh opaque id and stamps `ingested_at`. Returns the resulting
176    /// [`AuditRow`]s in the same order as the input.
177    ///
178    /// Validation is the caller's responsibility — pass already-validated
179    /// [`AuditRecord`]s (the HTTP layer runs [`AuditRecord::validate`] over
180    /// the batch before calling here). The trait does not re-validate; it
181    /// just durably appends.
182    async fn insert_batch(
183        &self,
184        records: &[AuditRecord],
185        ingested_at: i64,
186    ) -> Result<Vec<AuditRow>, StoreError>;
187}
188
189/// In-memory [`AuditStore`] for tests and single-node bootstrapping. Cheap
190/// to `clone` — shares one backing queue. Records keep insertion order.
191#[derive(Default, Clone)]
192pub struct MemoryAuditStore {
193    inner: Arc<Mutex<MemoryAuditInner>>,
194}
195
196#[derive(Default)]
197struct MemoryAuditInner {
198    next_seq: u64,
199    rows: VecDeque<AuditRow>,
200}
201
202impl MemoryAuditStore {
203    pub fn new() -> Self {
204        Self::default()
205    }
206
207    /// Snapshot of every row inserted so far, in insertion order. Test-only
208    /// accessor — `AuditStore` proper exposes reads through F14's surface.
209    pub fn snapshot(&self) -> Vec<AuditRow> {
210        self.inner
211            .lock()
212            .expect("audit store mutex poisoned")
213            .rows
214            .iter()
215            .cloned()
216            .collect()
217    }
218}
219
220impl std::fmt::Debug for MemoryAuditStore {
221    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
222        f.debug_struct("MemoryAuditStore").finish_non_exhaustive()
223    }
224}
225
226#[async_trait]
227impl AuditStore for MemoryAuditStore {
228    async fn insert_batch(
229        &self,
230        records: &[AuditRecord],
231        ingested_at: i64,
232    ) -> Result<Vec<AuditRow>, StoreError> {
233        let mut g = self.inner.lock().expect("audit store mutex poisoned");
234        let mut out = Vec::with_capacity(records.len());
235        for rec in records {
236            g.next_seq += 1;
237            let id = format!("audit-{:016x}", g.next_seq);
238            let row = AuditRow::new(id, rec.clone(), ingested_at);
239            g.rows.push_back(row.clone());
240            out.push(row);
241        }
242        Ok(out)
243    }
244}
245
246#[cfg(test)]
247mod tests {
248    use super::*;
249    use cheers_core::PrincipalId;
250    use pollster::block_on;
251
252    fn rec(method: &str, request_id: &str) -> AuditRecord {
253        AuditRecord::new(
254            1_700_000_000,
255            PrincipalId::user("alice"),
256            None,
257            Some("camp-a".into()),
258            "https://kamaji.example",
259            method,
260            vec![Scope::CloudDeploy],
261            "allow",
262            request_id,
263        )
264        .unwrap()
265    }
266
267    #[test]
268    fn new_accepts_well_formed_record() {
269        let r = rec("POST /cloud/deploy", "req-1");
270        assert_eq!(r.sub, PrincipalId::user("alice"));
271        assert_eq!(r.method, "POST /cloud/deploy");
272        assert_eq!(r.result, "allow");
273        assert_eq!(r.scope, vec![Scope::CloudDeploy]);
274    }
275
276    #[test]
277    fn validate_rejects_each_empty_field() {
278        // at <= 0
279        let mut r = rec("m", "rid");
280        r.at = 0;
281        assert_eq!(r.validate(), Err(AuditValidationError::NonPositiveAt(0)));
282        r.at = -1;
283        assert_eq!(r.validate(), Err(AuditValidationError::NonPositiveAt(-1)));
284
285        // empty aud
286        let mut r = rec("m", "rid");
287        r.aud = String::new();
288        assert_eq!(r.validate(), Err(AuditValidationError::EmptyAud));
289
290        // empty method
291        let mut r = rec("m", "rid");
292        r.method = String::new();
293        assert_eq!(r.validate(), Err(AuditValidationError::EmptyMethod));
294
295        // empty result
296        let mut r = rec("m", "rid");
297        r.result = String::new();
298        assert_eq!(r.validate(), Err(AuditValidationError::EmptyResult));
299
300        // empty request_id
301        let mut r = rec("m", "rid");
302        r.request_id = String::new();
303        assert_eq!(r.validate(), Err(AuditValidationError::EmptyRequestId));
304    }
305
306    #[test]
307    fn validate_allows_empty_scope_list() {
308        // A call rejected before scope-matching legitimately has no scope.
309        let mut r = rec("m", "rid");
310        r.scope.clear();
311        r.validate().expect("empty scope is allowed");
312    }
313
314    #[test]
315    fn memory_store_insert_batch_preserves_order_and_stamps_metadata() {
316        let store = MemoryAuditStore::new();
317        let batch = vec![
318            rec("m1", "rid-1"),
319            rec("m2", "rid-2"),
320            rec("m3", "rid-3"),
321        ];
322        let rows = block_on(store.insert_batch(&batch, 5_000)).unwrap();
323        assert_eq!(rows.len(), 3);
324        assert_eq!(rows[0].record.method, "m1");
325        assert_eq!(rows[1].record.method, "m2");
326        assert_eq!(rows[2].record.method, "m3");
327        for row in &rows {
328            assert_eq!(row.ingested_at, 5_000);
329            assert!(!row.id.is_empty());
330        }
331        // Ids are unique.
332        let ids: std::collections::HashSet<_> = rows.iter().map(|r| &r.id).collect();
333        assert_eq!(ids.len(), 3);
334
335        // Snapshot reflects insertion order across batches.
336        let more = vec![rec("m4", "rid-4")];
337        block_on(store.insert_batch(&more, 6_000)).unwrap();
338        let snap = store.snapshot();
339        assert_eq!(snap.len(), 4);
340        assert_eq!(snap[3].record.method, "m4");
341        assert_eq!(snap[3].ingested_at, 6_000);
342    }
343
344    #[test]
345    fn memory_store_empty_batch_is_a_noop() {
346        let store = MemoryAuditStore::new();
347        let rows = block_on(store.insert_batch(&[], 1_000)).unwrap();
348        assert!(rows.is_empty());
349        assert!(store.snapshot().is_empty());
350    }
351
352    #[test]
353    fn audit_record_round_trips_through_serde() {
354        let r = rec("POST /x", "rid");
355        let json = serde_json::to_string(&r).unwrap();
356        let back: AuditRecord = serde_json::from_str(&json).unwrap();
357        assert_eq!(back, r);
358    }
359
360    #[test]
361    fn audit_record_serde_omits_optional_none_fields() {
362        let r = rec("m", "rid");
363        let json = serde_json::to_string(&r).unwrap();
364        // act is None for this fixture; camp_id is Some — only act should be omitted.
365        assert!(!json.contains("\"act\""), "act:None must be skipped: {json}");
366        assert!(json.contains("\"camp_id\""), "camp_id:Some(_) must serialize: {json}");
367    }
368
369    #[test]
370    fn trait_is_dyn_compatible() {
371        fn _u(_: &dyn AuditStore) {}
372    }
373}
cheers_server/audit.rs

cheers_server/
audit.rs
