⎇ f4eb7dd94404c2fadd8099c61ca11d2860bcff7a
crates/ente-brain/src/audit.rs
Ñawpa willay
Pi imata llamkarqan
ch'uya
Capability quotas, CAS gc, patrones burst/silence, snapshot incremental
1//! Audit log: cada acción mutadora del cerebro deja una entry inmutable
2//! con su predecesor encadenado por SHA256 (estilo Merkle). Verificable a
3//! posteriori sin confianza en quien escribe.
4//!
5//! Los entries viven en memoria. Para persistencia, `flush_to_cas()` los
6//! escribe al content-addressable store y devuelve el SHA del head, que
7//! puede guardarse en un archivo de "head pointer" (fuera de scope aquí).
8
9use crate::crystallize::Crystal;
10use serde::{Deserialize, Serialize};
11use std::collections::VecDeque;
12use ulid::Ulid;
13
14#[derive(Debug, Clone, Serialize, Deserialize)]
15pub struct AuditEntry {
16 /// Sequence number monotónico desde el inicio del log.
17 pub seq: u64,
18 /// Wall-clock al insertar.
19 pub timestamp_ms: u64,
20 /// SHA256 del entry anterior. None para el primer entry.
21 pub prev_sha: Option<[u8; 32]>,
22 /// SHA256 de este entry (auto-calculado al construir).
23 pub sha: [u8; 32],
24 /// Acción registrada.
25 pub action: AuditAction,
26}
27
28/// Sin `#[serde(tag)]`: bincode requiere external tagging (default serde
29/// para enums) para no usar `deserialize_any`. JSON sigue legible.
30#[derive(Debug, Clone, Serialize, Deserialize)]
31pub enum AuditAction {
32 PromoteCrystal { rule_id: Ulid, crystal: Crystal },
33 RemoveRule { rule_id: Ulid },
34 LoadRulesFile { path: String, count: usize },
35}
36
37pub struct AuditLog {
38 entries: VecDeque<AuditEntry>,
39 next_seq: u64,
40 /// Cap del log en memoria. Entries más viejos se descartan tras flush.
41 cap: usize,
42 /// Total acumulado de entries flusheadas a CAS.
43 flushed_count: u64,
44 /// SHA del último entry persistido a CAS — el "head pointer" del log.
45 last_flushed_sha: Option<[u8; 32]>,
46 /// Path opcional donde escribir el head pointer tras cada flush.
47 head_pointer_path: Option<std::path::PathBuf>,
48 /// Subscribers a entries en tiempo real. Cada `append` empuja a todos.
49 /// Subscribers cuyo receiver se dropeó se purgan en el siguiente push.
50 subscribers: Vec<tokio::sync::mpsc::UnboundedSender<AuditEntry>>,
51 /// Wall-clock del último flush exitoso a CAS. None si aún no se flush.
52 last_flush_at_ms: Option<u64>,
53}
54
55impl AuditLog {
56 pub fn new() -> Self {
57 Self::with_cap(512)
58 }
59
60 pub fn with_cap(cap: usize) -> Self {
61 Self {
62 entries: VecDeque::new(),
63 next_seq: 0,
64 cap,
65 flushed_count: 0,
66 last_flushed_sha: None,
67 head_pointer_path: None,
68 subscribers: Vec::new(),
69 last_flush_at_ms: None,
70 }
71 }
72
73 /// Registra un nuevo subscriber. El receiver recibe cada `AuditEntry`
74 /// futuro hasta que el receiver se dropee (subscriber se purga al
75 /// siguiente `append`).
76 pub fn subscribe(&mut self) -> tokio::sync::mpsc::UnboundedReceiver<AuditEntry> {
77 let (tx, rx) = tokio::sync::mpsc::unbounded_channel();
78 self.subscribers.push(tx);
79 rx
80 }
81
82 pub fn subscriber_count(&self) -> usize { self.subscribers.len() }
83
84 pub fn with_head_pointer(mut self, path: std::path::PathBuf) -> Self {
85 self.head_pointer_path = Some(path);
86 self
87 }
88
89 /// Apendea una acción. Calcula el SHA encadenado contra el último entry.
90 pub fn append(&mut self, action: AuditAction) -> AuditEntry {
91 let prev_sha = self.entries.back().map(|e| e.sha);
92 let timestamp_ms = now_ms();
93 let seq = self.next_seq;
94 self.next_seq += 1;
95
96 // Pre-construct con sha en cero, luego calcular sha sobre el
97 // serializado canónico, luego sobreescribir el campo.
98 let mut entry = AuditEntry {
99 seq, timestamp_ms, prev_sha, sha: [0u8; 32], action,
100 };
101 entry.sha = compute_sha(&entry);
102
103 if self.entries.len() >= self.cap {
104 self.entries.pop_front();
105 }
106 self.entries.push_back(entry.clone());
107 // Empujar a subscribers, purgando los muertos in-place.
108 self.subscribers.retain(|tx| tx.send(entry.clone()).is_ok());
109 entry
110 }
111
112 pub fn recent(&self, limit: usize) -> impl Iterator<Item = &AuditEntry> {
113 let n = if limit == 0 { self.entries.len() } else { limit.min(self.entries.len()) };
114 self.entries.iter().skip(self.entries.len() - n)
115 }
116
117 pub fn len(&self) -> usize { self.entries.len() }
118 pub fn is_empty(&self) -> bool { self.entries.is_empty() }
119
120 pub fn head_sha(&self) -> Option<[u8; 32]> {
121 self.entries.back().map(|e| e.sha)
122 }
123
124 /// Persiste el entry pasado al CAS y devuelve su SHA. Pensado para
125 /// snapshots externos — el log en memoria sigue intacto.
126 pub fn persist_to_cas(entry: &AuditEntry) -> anyhow::Result<[u8; 32]> {
127 let bytes = serde_json::to_vec(entry)?;
128 let sha = ente_cas::store(&bytes)?;
129 Ok(sha)
130 }
131
132 /// Persiste TODOS los entries actuales al CAS y actualiza el head pointer.
133 /// Idempotente: re-flushar dos veces da los mismos SHAs (CAS dedup).
134 /// Devuelve cuántas entries se flushearon en esta pasada.
135 ///
136 /// Forma canónica: serializamos `entry` con `sha = [0; 32]` (formato
137 /// pre-hash). El CAS computa sha256 sobre esos bytes y devuelve un SHA
138 /// que por construcción coincide con `entry.sha` calculado al append.
139 pub fn flush_to_cas(&mut self) -> anyhow::Result<usize> {
140 let mut written = 0;
141 let mut last_sha = self.last_flushed_sha;
142 for entry in &self.entries {
143 if entry.seq < self.flushed_count { continue; }
144 let bytes = canonical_bytes(entry);
145 let sha = ente_cas::store(&bytes)?;
146 debug_assert_eq!(sha, entry.sha,
147 "CAS sha != entry.sha — fórmula canónica rota");
148 last_sha = Some(sha);
149 written += 1;
150 }
151 self.flushed_count += written as u64;
152 self.last_flushed_sha = last_sha;
153 if written > 0 {
154 self.last_flush_at_ms = Some(now_ms());
155 }
156 // Persistir head pointer si está configurado.
157 if let (Some(path), Some(sha)) = (&self.head_pointer_path, last_sha) {
158 let pointer = AuditHeadPointer {
159 last_seq: self.next_seq.saturating_sub(1),
160 last_sha: sha,
161 flushed_count: self.flushed_count,
162 timestamp_ms: now_ms(),
163 };
164 let json = serde_json::to_vec_pretty(&pointer)?;
165 // Escritura atómica: tmp + rename
166 let tmp = path.with_extension("tmp");
167 if let Some(parent) = path.parent() { let _ = std::fs::create_dir_all(parent); }
168 std::fs::write(&tmp, json)?;
169 std::fs::rename(&tmp, path)?;
170 }
171 Ok(written)
172 }
173
174 pub fn flushed_count(&self) -> u64 { self.flushed_count }
175 pub fn last_flushed_sha(&self) -> Option<[u8; 32]> { self.last_flushed_sha }
176 pub fn last_flush_at_ms(&self) -> Option<u64> { self.last_flush_at_ms }
177
178 /// Segundos transcurridos desde el último flush. None si nunca se flush.
179 pub fn last_flush_age_secs(&self) -> Option<f64> {
180 let then = self.last_flush_at_ms?;
181 let now = now_ms();
182 Some((now.saturating_sub(then)) as f64 / 1000.0)
183 }
184}
185
186/// Pointer al head del audit log — escrito atómicamente en disco tras cada
187/// flush. Permite verificar la integridad del log sin escanearlo entero:
188/// el cliente lee el head, recupera el blob desde CAS, valida `prev_sha`
189/// recursivamente hasta el genesis.
190#[derive(Debug, Clone, Serialize, Deserialize)]
191pub struct AuditHeadPointer {
192 pub last_seq: u64,
193 pub last_sha: [u8; 32],
194 pub flushed_count: u64,
195 pub timestamp_ms: u64,
196}
197
198/// Reporte de un replay: número de actions aplicadas + reglas finales.
199#[derive(Debug, Clone, Serialize, Deserialize)]
200pub struct ReplayReport {
201 pub applied: u64,
202 pub final_rule_count: usize,
203 pub error: Option<String>,
204}
205
206/// Reporte de verificación de la cadena audit.
207#[derive(Debug, Clone, Serialize, Deserialize)]
208pub struct VerificationReport {
209 /// Cuántas entries se recorrieron y verificaron exitosamente.
210 pub verified: u64,
211 /// Si hubo error, el seq donde se detectó.
212 pub broken_at_seq: Option<u64>,
213 /// Detalles del error si hubo.
214 pub error: Option<String>,
215 /// SHA del genesis (primer entry; prev_sha = None).
216 pub genesis_sha: Option<[u8; 32]>,
217}
218
219/// Recorre la cadena del audit log desde `start_sha` hacia atrás vía `prev_sha`
220/// hasta el genesis. Para cada entry valida:
221/// 1. CAS contiene un blob bajo ese SHA
222/// 2. sha256(blob) == SHA esperado (defensa contra tampering del CAS)
223/// 3. El blob deserializa a AuditEntry con sha=[0;32] (forma canónica)
224///
225/// Devuelve un VerificationReport con el conteo, posibles errores y
226/// el SHA del genesis (útil para clientes que quieren cachearlo).
227pub fn verify_chain_from_cas(start_sha: [u8; 32]) -> VerificationReport {
228 let mut current = Some(start_sha);
229 let mut verified = 0u64;
230 let mut last_seen: Option<AuditEntry> = None;
231
232 while let Some(sha) = current {
233 let path = ente_cas::cas_root().join(ente_cas::hex(&sha));
234 let bytes = match std::fs::read(&path) {
235 Ok(b) => b,
236 Err(e) => return VerificationReport {
237 verified,
238 broken_at_seq: last_seen.as_ref().map(|e| e.seq),
239 error: Some(format!("CAS read {}: {e}", path.display())),
240 genesis_sha: None,
241 },
242 };
243 // Verificación 1: el blob hashea a la SHA esperada (CAS contract).
244 let actual = ente_cas::sha256_of(&bytes);
245 if actual != sha {
246 return VerificationReport {
247 verified,
248 broken_at_seq: last_seen.as_ref().map(|e| e.seq),
249 error: Some(format!(
250 "CAS tamper en {}: expected {} got {}",
251 path.display(), ente_cas::hex(&sha), ente_cas::hex(&actual)
252 )),
253 genesis_sha: None,
254 };
255 }
256 // Verificación 2: deserialize. El blob canónico tiene sha=[0;32].
257 let mut entry: AuditEntry = match serde_json::from_slice(&bytes) {
258 Ok(e) => e,
259 Err(e) => return VerificationReport {
260 verified,
261 broken_at_seq: last_seen.as_ref().map(|e| e.seq),
262 error: Some(format!("deserialize: {e}")),
263 genesis_sha: None,
264 },
265 };
266 // Re-poblar el sha en el entry para reportar coherentemente.
267 entry.sha = sha;
268 verified += 1;
269
270 let prev = entry.prev_sha;
271 last_seen = Some(entry);
272 current = prev;
273 }
274
275 VerificationReport {
276 verified,
277 broken_at_seq: None,
278 error: None,
279 genesis_sha: last_seen.as_ref().map(|e| e.sha),
280 }
281}
282
283/// Devuelve el set de SHAs alcanzables desde `start_sha` siguiendo
284/// `prev_sha` hasta el genesis. Usado por el GC del CAS para construir
285/// las "raíces vivas" del audit log.
286pub fn reachable_from_head(start_sha: [u8; 32]) -> std::collections::HashSet<[u8; 32]> {
287 let mut set = std::collections::HashSet::new();
288 let mut current = Some(start_sha);
289 while let Some(sha) = current {
290 if !set.insert(sha) { break; } // ciclo (no debería pasar) — corta
291 let path = ente_cas::cas_root().join(ente_cas::hex(&sha));
292 let bytes = match std::fs::read(&path) { Ok(b) => b, Err(_) => break };
293 let entry: AuditEntry = match serde_json::from_slice(&bytes) {
294 Ok(e) => e, Err(_) => break,
295 };
296 current = entry.prev_sha;
297 }
298 set
299}
300
301/// Recorre la cadena entera (head→genesis) y reconstruye la lista de
302/// actions en orden cronológico (oldest first). Útil tanto para replay
303/// como para auditoría retrospectiva.
304pub fn collect_chain_from_cas(start_sha: [u8; 32]) -> anyhow::Result<Vec<AuditEntry>> {
305 let mut entries = Vec::new();
306 let mut current = Some(start_sha);
307 while let Some(sha) = current {
308 let path = ente_cas::cas_root().join(ente_cas::hex(&sha));
309 let bytes = std::fs::read(&path)?;
310 let mut entry: AuditEntry = serde_json::from_slice(&bytes)?;
311 entry.sha = sha;
312 let prev = entry.prev_sha;
313 entries.push(entry);
314 current = prev;
315 }
316 // entries está en orden head→genesis. Reverse para chronological.
317 entries.reverse();
318 Ok(entries)
319}
320
321/// Aplica las actions de la cadena en orden cronológico contra un engine
322/// fresco. PromoteCrystal → insert. RemoveRule → remove. LoadRulesFile →
323/// log informativo (los archivos pueden no existir en el ambiente actual).
324pub fn replay_chain(
325 start_sha: [u8; 32],
326 engine: &mut crate::engine::RuleEngine,
327) -> ReplayReport {
328 let entries = match collect_chain_from_cas(start_sha) {
329 Ok(es) => es,
330 Err(e) => return ReplayReport {
331 applied: 0, final_rule_count: engine.len(),
332 error: Some(format!("collect chain: {e}")),
333 },
334 };
335 let mut applied = 0u64;
336 for entry in &entries {
337 match &entry.action {
338 AuditAction::PromoteCrystal { rule_id, crystal } => {
339 let mut rule = crate::crystallize::crystal_to_rule(crystal);
340 rule.id = *rule_id; // preservar identidad histórica
341 engine.insert(rule);
342 }
343 AuditAction::RemoveRule { rule_id } => {
344 engine.remove(*rule_id);
345 }
346 AuditAction::LoadRulesFile { path: _, count: _ } => {
347 // Los archivos referenciados por path pueden haber cambiado
348 // o no existir. Log y skip — el replay sólo reconstruye
349 // promotes/removes que tienen estado en CAS.
350 }
351 }
352 applied += 1;
353 }
354 ReplayReport {
355 applied,
356 final_rule_count: engine.len(),
357 error: None,
358 }
359}
360
361impl Default for AuditLog {
362 fn default() -> Self { Self::new() }
363}
364
365fn now_ms() -> u64 {
366 std::time::SystemTime::now()
367 .duration_since(std::time::UNIX_EPOCH)
368 .map(|d| d.as_millis() as u64)
369 .unwrap_or(0)
370}
371
372/// SHA256 sobre el entry en forma canónica (sha=[0;32]). Hash y CAS storage
373/// ven los mismos bytes, así que `ente_cas::store(canonical)` devuelve el
374/// mismo SHA que `compute_sha(entry)`.
375fn compute_sha(entry: &AuditEntry) -> [u8; 32] {
376 let bytes = canonical_bytes(entry);
377 ente_cas::sha256_of(&bytes)
378}
379
380/// Forma canónica: el entry serializado JSON con `sha = [0; 32]`.
381/// JSON sin pretty-print es determinístico para nuestros tipos.
382fn canonical_bytes(entry: &AuditEntry) -> Vec<u8> {
383 let canonical = AuditEntry {
384 sha: [0u8; 32],
385 ..entry.clone()
386 };
387 serde_json::to_vec(&canonical).unwrap_or_default()
388}
389
390#[cfg(test)]
391mod tests {
392 use super::*;
393
394 #[test]
395 fn chain_links_consecutive_entries() {
396 let mut log = AuditLog::new();
397 let e1 = log.append(AuditAction::RemoveRule { rule_id: Ulid::new() });
398 let e2 = log.append(AuditAction::RemoveRule { rule_id: Ulid::new() });
399 assert!(e1.prev_sha.is_none());
400 assert_eq!(e2.prev_sha, Some(e1.sha));
401 assert_ne!(e1.sha, e2.sha);
402 }
403
404 #[test]
405 fn seq_monotonic() {
406 let mut log = AuditLog::new();
407 let e1 = log.append(AuditAction::RemoveRule { rule_id: Ulid::new() });
408 let e2 = log.append(AuditAction::RemoveRule { rule_id: Ulid::new() });
409 assert_eq!(e2.seq, e1.seq + 1);
410 }
411
412 #[test]
413 fn cap_evicts_oldest() {
414 let mut log = AuditLog::with_cap(3);
415 for _ in 0..5 {
416 log.append(AuditAction::RemoveRule { rule_id: Ulid::new() });
417 }
418 assert_eq!(log.len(), 3);
419 // El primer seq superviviente debe ser 2.
420 assert_eq!(log.recent(0).next().unwrap().seq, 2);
421 }
422
423 // ---------- Tests de integración con CAS real (en directorio temporal) ----------
424
425 use crate::engine::RuleEngine;
426 use std::sync::Mutex;
427
428 /// Lock para serializar tests que mutan ENTE_CAS_ROOT (test threads
429 /// comparten env vars). Sin esto, dos tests en paralelo pisan el path.
430 static CAS_TEST_LOCK: Mutex<()> = Mutex::new(());
431
432 fn with_temp_cas<F: FnOnce()>(f: F) {
433 let _guard = CAS_TEST_LOCK.lock().unwrap();
434 let dir = std::env::temp_dir().join(format!("ente-cas-test-{}", Ulid::new()));
435 std::env::set_var("ENTE_CAS_ROOT", &dir);
436 let _cleanup = scopeguard(&dir);
437 f();
438 }
439
440 fn scopeguard(dir: &std::path::Path) -> impl Drop + '_ {
441 struct G<'a>(&'a std::path::Path);
442 impl<'a> Drop for G<'a> {
443 fn drop(&mut self) {
444 std::env::remove_var("ENTE_CAS_ROOT");
445 let _ = std::fs::remove_dir_all(self.0);
446 }
447 }
448 G(dir)
449 }
450
451 fn dummy_crystal(ant: EventKind, con: EventKind) -> Crystal {
452 Crystal {
453 antecedent: ant,
454 consequent: con,
455 conditional_prob: 0.9,
456 pmi: 1.5,
457 support: 7,
458 gap_stats: None,
459 }
460 }
461
462 use crate::rules::EventKind;
463
464 #[test]
465 fn flush_round_trip_preserves_chain() {
466 with_temp_cas(|| {
467 let mut log = AuditLog::new();
468 let id1 = Ulid::new();
469 let id2 = Ulid::new();
470 log.append(AuditAction::PromoteCrystal {
471 rule_id: id1,
472 crystal: dummy_crystal(EventKind::EnteSpawned, EventKind::EnteDied),
473 });
474 log.append(AuditAction::PromoteCrystal {
475 rule_id: id2,
476 crystal: dummy_crystal(EventKind::BusAnnounce, EventKind::BusInvoke),
477 });
478 log.append(AuditAction::RemoveRule { rule_id: id1 });
479
480 assert_eq!(log.flush_to_cas().unwrap(), 3);
481 let head = log.last_flushed_sha().expect("head set");
482 let report = verify_chain_from_cas(head);
483 assert!(report.error.is_none(), "verification failed: {:?}", report.error);
484 assert_eq!(report.verified, 3);
485 });
486 }
487
488 #[test]
489 fn replay_reconstructs_engine_state() {
490 with_temp_cas(|| {
491 let mut log = AuditLog::new();
492 let id1: Ulid = "01KQR3000000000000000000A1".parse().unwrap();
493 let id2: Ulid = "01KQR3000000000000000000A2".parse().unwrap();
494 let id3: Ulid = "01KQR3000000000000000000A3".parse().unwrap();
495 log.append(AuditAction::PromoteCrystal {
496 rule_id: id1,
497 crystal: dummy_crystal(EventKind::EnteSpawned, EventKind::EnteDied),
498 });
499 log.append(AuditAction::PromoteCrystal {
500 rule_id: id2,
501 crystal: dummy_crystal(EventKind::BusAnnounce, EventKind::BusInvoke),
502 });
503 log.append(AuditAction::PromoteCrystal {
504 rule_id: id3,
505 crystal: dummy_crystal(EventKind::DeviceAdded, EventKind::DeviceRemoved),
506 });
507 log.append(AuditAction::RemoveRule { rule_id: id2 });
508 log.flush_to_cas().unwrap();
509 let head = log.last_flushed_sha().unwrap();
510
511 let mut engine = RuleEngine::empty();
512 let rep = replay_chain(head, &mut engine);
513 assert!(rep.error.is_none(), "replay error: {:?}", rep.error);
514 assert_eq!(rep.applied, 4);
515 assert_eq!(engine.len(), 2, "id2 should be removed, id1 + id3 remain");
516 // Ulids preservados
517 let ids: Vec<Ulid> = engine.rules().map(|r| r.id).collect();
518 assert!(ids.contains(&id1));
519 assert!(!ids.contains(&id2));
520 assert!(ids.contains(&id3));
521 });
522 }
523
524 #[test]
525 fn replay_after_eviction_still_works() {
526 with_temp_cas(|| {
527 // Cap pequeño: la mayoría de entries se evictan de memoria pero
528 // siguen en CAS. Replay debe poder reconstruir desde CAS solo.
529 let mut log = AuditLog::with_cap(2);
530 let mut ids = Vec::new();
531 for _ in 0..6 {
532 let id = Ulid::new();
533 ids.push(id);
534 log.append(AuditAction::PromoteCrystal {
535 rule_id: id,
536 crystal: dummy_crystal(EventKind::EnteSpawned, EventKind::EnteDied),
537 });
538 log.flush_to_cas().unwrap();
539 }
540 assert_eq!(log.len(), 2, "cap eviction limita memoria");
541 let head = log.last_flushed_sha().unwrap();
542
543 let mut engine = RuleEngine::empty();
544 let rep = replay_chain(head, &mut engine);
545 assert!(rep.error.is_none());
546 assert_eq!(rep.applied, 6);
547 assert_eq!(engine.len(), 6);
548 });
549 }
550}