1use serde::{Deserialize, Serialize};
2use sha2::{Digest, Sha256};
3use std::fs::{self, OpenOptions};
4use std::io::{BufRead, Read, Write};
5use std::path::{Path, PathBuf};
6
7mod checkpoint;
8mod segments;
9mod v2;
10
11pub use v2::{
12 AuditDecision, ENTRY_VERSION, HMAC_PREFIX, SessionFacts, hmac_hex, keyed_digest,
13 keyed_input_digest, load_or_create_hmac_key,
14};
15
16#[derive(Debug, Clone, Serialize, Deserialize)]
19pub struct AuditEntry {
20 pub timestamp: String,
21 pub agent_id: String,
22 pub tool: String,
23 pub action: Option<String>,
24 pub input_hash: String,
25 pub output_tokens: u32,
26 pub role: String,
27 pub event_type: AuditEventType,
28 pub prev_hash: String,
29 pub entry_hash: String,
30 #[serde(skip_serializing_if = "Option::is_none", default)]
33 pub signature: Option<String>,
34 #[serde(skip_serializing_if = "Option::is_none", default)]
37 pub v: Option<u64>,
38 #[serde(skip_serializing_if = "Option::is_none", default)]
40 pub session_id: Option<String>,
41 #[serde(skip_serializing_if = "Option::is_none", default)]
43 pub seq: Option<u64>,
44 #[serde(skip_serializing_if = "Option::is_none", default)]
45 pub instance_id: Option<String>,
46 #[serde(skip_serializing_if = "Option::is_none", default)]
47 pub policy_digest: Option<String>,
48 #[serde(skip_serializing_if = "Option::is_none", default)]
51 pub decision: Option<String>,
52 #[serde(skip_serializing_if = "Option::is_none", default)]
53 pub output_digest: Option<String>,
54 #[serde(skip_serializing_if = "Option::is_none", default)]
56 pub v2_hash: Option<String>,
57}
58
59#[derive(Debug, Clone, PartialEq, Eq)]
60pub enum AuditEventType {
61 ToolCall,
62 ToolDenied,
63 PathJailViolation,
64 BudgetExceeded,
65 CrossProjectAccess,
66 RateLimited,
67 SecurityViolation,
68 RoleChanged,
69 SecretDetected,
70 AgentRegistered,
72 AgentSuspended,
73 AgentResumed,
74 AgentDecommissioned,
75 AgentDriftDetected,
79 AgentDriftAcknowledged,
83 Unknown(String),
86}
87
88impl AuditEventType {
89 fn wire_label(&self) -> &str {
90 match self {
91 Self::ToolCall => "tool_call",
92 Self::ToolDenied => "tool_denied",
93 Self::PathJailViolation => "path_jail_violation",
94 Self::BudgetExceeded => "budget_exceeded",
95 Self::CrossProjectAccess => "cross_project_access",
96 Self::RateLimited => "rate_limited",
97 Self::SecurityViolation => "security_violation",
98 Self::RoleChanged => "role_changed",
99 Self::SecretDetected => "secret_detected",
100 Self::AgentRegistered => "agent_registered",
101 Self::AgentSuspended => "agent_suspended",
102 Self::AgentResumed => "agent_resumed",
103 Self::AgentDecommissioned => "agent_decommissioned",
104 Self::AgentDriftDetected => "agent_drift_detected",
105 Self::AgentDriftAcknowledged => "agent_drift_acknowledged",
106 Self::Unknown(label) => label,
107 }
108 }
109}
110
111impl Serialize for AuditEventType {
112 fn serialize<S>(&self, serializer: S) -> Result<S::Ok, S::Error>
113 where
114 S: serde::Serializer,
115 {
116 if let Self::Unknown(label) = self {
117 if !valid_unknown_event_label_for_write(label) {
118 return Err(serde::ser::Error::custom(format!(
119 "unknown event_type is not a safe additive label: {label}"
120 )));
121 }
122 }
123 serializer.serialize_str(self.wire_label())
124 }
125}
126
127impl<'de> Deserialize<'de> for AuditEventType {
128 fn deserialize<D>(deserializer: D) -> Result<Self, D::Error>
129 where
130 D: serde::Deserializer<'de>,
131 {
132 let label = String::deserialize(deserializer)?;
133 let event_type = match label.as_str() {
134 "tool_call" => Self::ToolCall,
135 "tool_denied" => Self::ToolDenied,
136 "path_jail_violation" => Self::PathJailViolation,
137 "budget_exceeded" => Self::BudgetExceeded,
138 "cross_project_access" => Self::CrossProjectAccess,
139 "rate_limited" => Self::RateLimited,
140 "security_violation" => Self::SecurityViolation,
141 "role_changed" => Self::RoleChanged,
142 "secret_detected" => Self::SecretDetected,
143 "agent_registered" => Self::AgentRegistered,
144 "agent_suspended" => Self::AgentSuspended,
145 "agent_resumed" => Self::AgentResumed,
146 "agent_decommissioned" => Self::AgentDecommissioned,
147 "agent_drift_detected" => Self::AgentDriftDetected,
148 "agent_drift_acknowledged" => Self::AgentDriftAcknowledged,
149 _ if valid_extension_event_label(&label) => Self::Unknown(label),
150 _ => {
151 return Err(serde::de::Error::custom(
152 "unknown event_type must use the x- extension namespace",
153 ));
154 }
155 };
156 Ok(event_type)
157 }
158}
159
160fn valid_extension_event_label(label: &str) -> bool {
161 const MAX_EVENT_LABEL_BYTES: usize = 64;
162 if label.len() > MAX_EVENT_LABEL_BYTES {
163 return false;
164 }
165 let Some(rest) = label.strip_prefix("x-") else {
166 return false;
167 };
168 let mut chars = rest.chars();
169 chars.next().is_some_and(|c| c.is_ascii_lowercase())
170 && chars.all(|c| c.is_ascii_lowercase() || c.is_ascii_digit() || c == '_')
171}
172
173fn valid_unknown_event_label_for_write(label: &str) -> bool {
174 valid_extension_event_label(label)
175}
176
177pub struct AuditEntryData {
178 pub agent_id: String,
179 pub tool: String,
180 pub action: Option<String>,
181 pub input_hash: String,
182 pub output_tokens: u32,
183 pub role: String,
184 pub event_type: AuditEventType,
185}
186
187pub struct ChainVerifyResult {
188 pub total_entries: usize,
189 pub valid: bool,
190 pub first_invalid_at: Option<usize>,
191 pub source_present: bool,
195 pub unknown_event_types: usize,
202}
203
204fn trail_path() -> Option<PathBuf> {
205 let dir = crate::core::data_dir::lean_ctx_data_dir().ok()?;
206 let audit_dir = dir.join("audit");
207 fs::create_dir_all(&audit_dir).ok()?;
208 Some(audit_dir.join("trail.jsonl"))
209}
210
211fn read_tail(file: &fs::File) -> Result<Tail, String> {
221 Ok(Tail::after(read_tail_entry(file)?.as_ref()))
222}
223
224#[derive(Clone)]
227struct Tail {
228 prev_hash: String,
229 head: String,
230}
231
232impl Tail {
233 fn after(entry: Option<&serde_json::Value>) -> Self {
234 let prev_hash = entry
235 .and_then(|entry| entry.get("entry_hash")?.as_str())
236 .unwrap_or("genesis")
237 .to_string();
238 let head = entry
239 .and_then(|entry| v2::head_of(entry, v2::version_of(entry)?))
240 .unwrap_or_else(|| prev_hash.clone());
241 Self { prev_hash, head }
242 }
243}
244
245fn read_tail_entry(file: &fs::File) -> Result<Option<serde_json::Value>, String> {
248 use std::io::{Read, Seek, SeekFrom};
249 const TAIL: i64 = 64 * 1024;
250
251 let mut f = file;
252 let len = f
253 .seek(SeekFrom::End(0))
254 .map_err(|e| format!("audit tail seek: {e}"))?;
255 if len == 0 {
256 return Ok(None);
257 }
258 let start = if (len as i64) > TAIL {
259 -TAIL
260 } else {
261 -(len as i64)
262 };
263 f.seek(SeekFrom::End(start))
264 .map_err(|e| format!("audit tail seek: {e}"))?;
265 let mut buf = String::new();
266 f.read_to_string(&mut buf)
267 .map_err(|e| format!("audit tail read: {e}"))?;
268 for line in buf.lines().rev() {
271 if line.trim().is_empty() {
272 continue;
273 }
274 let mut last: Option<serde_json::Value> = None;
275 for parsed in serde_json::Deserializer::from_str(line).into_iter::<serde_json::Value>() {
276 let v = parsed.map_err(|e| format!("audit tail parse: {e}"))?;
277 if v.get("entry_hash").and_then(|h| h.as_str()).is_some() {
278 last = Some(v);
279 }
280 }
281 if last.is_some() {
282 return Ok(last);
283 }
284 }
285 Err("audit tail: no readable chain head".to_string())
288}
289
290fn compute_entry_hash(prev_hash: &str, data_json: &str) -> String {
291 let mut hasher = Sha256::new();
292 hasher.update(prev_hash.as_bytes());
293 hasher.update(data_json.as_bytes());
294 crate::core::agent_identity::hex_encode(&hasher.finalize())
295}
296
297pub fn record(data: AuditEntryData) {
302 let _ = try_record(data);
303}
304
305const LOCK_TIMEOUT: std::time::Duration = std::time::Duration::from_millis(750);
310const LOCK_RETRY_INTERVAL: std::time::Duration = std::time::Duration::from_millis(25);
311
312fn lock_trail(file: &fs::File) -> Result<(), String> {
314 lock_trail_with(file, fs2::FileExt::try_lock_exclusive)
315}
316
317fn lock_trail_shared(file: &fs::File) -> Result<(), String> {
324 lock_trail_with(file, fs2::FileExt::try_lock_shared)
325}
326
327fn release_read_lock(file: &fs::File) {
341 #[cfg(not(windows))]
342 let _ = fs2::FileExt::unlock(file);
343 #[cfg(windows)]
344 let _ = file;
345}
346
347fn still_holds_prefix(file: &fs::File, len: u64) -> bool {
350 file.metadata().is_ok_and(|meta| meta.len() >= len)
351}
352
353fn lock_trail_with(
354 file: &fs::File,
355 try_lock: fn(&fs::File) -> std::io::Result<()>,
356) -> Result<(), String> {
357 use std::time::Instant;
358
359 let deadline = Instant::now() + LOCK_TIMEOUT;
360 loop {
361 match try_lock(file) {
362 Ok(()) => return Ok(()),
363 Err(error) if super::file_lock::is_contended(&error) => {
364 if Instant::now() >= deadline {
365 return Err(format!(
366 "audit trail lock timed out after {}ms; another append is still active",
367 LOCK_TIMEOUT.as_millis()
368 ));
369 }
370 std::thread::sleep(LOCK_RETRY_INTERVAL);
371 }
372 Err(error) => return Err(format!("audit trail lock: {error}")),
373 }
374 }
375}
376
377pub fn try_record(data: AuditEntryData) -> Result<AuditEntry, String> {
388 let path = trail_path().ok_or_else(|| "audit trail unavailable".to_string())?;
389 try_record_at(&path, data)
390}
391
392pub fn record_at(path: &Path, data: AuditEntryData) {
396 let _ = try_record_at(path, data);
397}
398
399fn try_record_at(path: &Path, data: AuditEntryData) -> Result<AuditEntry, String> {
400 try_record_at_time(path, Record { data, facts: None }, chrono::Utc::now())
401}
402
403pub fn record_v2(data: AuditEntryData, facts: SessionFacts) {
405 let _ = try_record_v2(data, facts);
406}
407
408pub fn try_record_v2(data: AuditEntryData, facts: SessionFacts) -> Result<AuditEntry, String> {
412 let path = trail_path().ok_or_else(|| "audit trail unavailable".to_string())?;
413 try_record_v2_at(&path, data, facts)
414}
415
416pub fn try_record_v2_at(
418 path: &Path,
419 data: AuditEntryData,
420 facts: SessionFacts,
421) -> Result<AuditEntry, String> {
422 if facts.session_id.is_empty() {
423 return Err("audit v2: empty session id".to_string());
424 }
425 let record = Record {
426 data,
427 facts: Some(facts),
428 };
429 try_record_at_time(path, record, chrono::Utc::now())
430}
431
432struct Record {
434 data: AuditEntryData,
435 facts: Option<SessionFacts>,
436}
437
438fn try_record_at_time(
440 path: &Path,
441 record: Record,
442 now: chrono::DateTime<chrono::Utc>,
443) -> Result<AuditEntry, String> {
444 let mut record = record;
446 for _ in 0..SNAPSHOT_ATTEMPTS {
447 let mut file = open_active_locked(path)?;
448 let appended = append_or_seal(path, &mut file, record, now);
449 let _ = fs2::FileExt::unlock(&file);
450 match appended? {
451 Appended::Done(entry) => return Ok(entry),
452 Appended::Moved(back) => record = back,
453 }
454 }
455 Err("audit trail: the active file kept moving".to_string())
456}
457
458enum Appended {
460 Done(AuditEntry),
461 Moved(Record),
463}
464
465fn open_active_locked(path: &Path) -> Result<fs::File, String> {
470 for _ in 0..SNAPSHOT_ATTEMPTS {
471 let file = OpenOptions::new()
472 .create(true)
473 .append(true)
474 .read(true)
475 .open(path)
476 .map_err(|e| format!("audit trail open: {e}"))?;
477 lock_trail(&file)?;
478 if is_current(path, &file) {
479 return Ok(file);
480 }
481 let _ = fs2::FileExt::unlock(&file);
482 }
483 Err("audit trail: the active file kept moving".to_string())
484}
485
486fn is_current(path: &Path, file: &fs::File) -> bool {
488 #[cfg(unix)]
489 {
490 use std::os::unix::fs::MetadataExt;
491 match (fs::metadata(path), file.metadata()) {
492 (Ok(named), Ok(open)) => (named.dev(), named.ino()) == (open.dev(), open.ino()),
493 _ => false,
494 }
495 }
496 #[cfg(not(unix))]
499 {
500 let _ = (path, file);
501 true
502 }
503}
504
505fn append_or_seal(
508 path: &Path,
509 file: &mut fs::File,
510 record: Record,
511 now: chrono::DateTime<chrono::Utc>,
512) -> Result<Appended, String> {
513 let Some(last) = read_tail_entry(file)? else {
514 let tail = match segments::sealed(path).last() {
516 Some((_, segment)) => fs::File::open(segment)
517 .map_err(|e| format!("audit segment open: {e}"))
518 .and_then(|segment| read_tail(&segment))?,
519 None => Tail::after(None),
520 };
521 return append_locked(file, record, tail).map(Appended::Done);
522 };
523 let last_tail = Tail::after(Some(&last));
524 let next_seq = segments::sealed(path).last().map_or(0, |(seq, _)| seq + 1);
525 let sealed_seq = if last.get("event_type").and_then(serde_json::Value::as_str)
526 == Some(segments::SEAL_EVENT)
527 {
528 segments::sealed_seq(last.get("action").and_then(serde_json::Value::as_str))
529 } else {
530 None
531 };
532 let len = file
533 .metadata()
534 .map_err(|e| format!("audit stat: {e}"))?
535 .len();
536 let last_at = last.get("timestamp").and_then(serde_json::Value::as_str);
537 let seq = match sealed_seq {
538 Some(seq) if seq == next_seq => seq,
540 Some(seq) if seq < next_seq => return Ok(Appended::Moved(record)),
542 _ if segments::due(len, last_at, now)
543 && !segments::backing_off()
544 && verified_once(path, file) =>
545 {
546 let seal = Record {
547 data: seal_data(next_seq),
548 facts: None,
549 };
550 append_locked(file, seal, last_tail)?;
551 next_seq
552 }
553 _ => return append_locked(file, record, last_tail).map(Appended::Done),
554 };
555 match segments::retire(path, seq) {
556 Ok(()) => Ok(Appended::Moved(record)),
557 Err(error) => {
558 tracing::warn!("audit trail not sealed: {error}");
561 segments::back_off();
562 let tail = read_tail(file)?;
563 append_locked(file, record, tail).map(Appended::Done)
564 }
565 }
566}
567
568fn verified_once(path: &Path, file: &fs::File) -> bool {
572 let Some(store) = checkpoint::Store::for_trail(path) else {
573 return false;
574 };
575 checkpoint::FileStamp::of(file).is_some_and(|stamp| store.holds_prefix_of(file, &stamp))
576}
577
578fn seal_data(seq: u64) -> AuditEntryData {
579 AuditEntryData {
580 agent_id: "lean-ctx".to_string(),
581 tool: "audit_trail".to_string(),
582 action: Some(segments::seal_action(seq)),
583 input_hash: String::new(),
584 output_tokens: 0,
585 role: "system".to_string(),
586 event_type: AuditEventType::Unknown(segments::SEAL_EVENT.to_string()),
587 }
588}
589
590pub fn is_seal(event_type: &AuditEventType) -> bool {
593 matches!(event_type, AuditEventType::Unknown(label) if label == segments::SEAL_EVENT)
594}
595
596fn append_locked(file: &mut fs::File, record: Record, tail: Tail) -> Result<AuditEntry, String> {
599 use std::io::{Seek, SeekFrom};
600
601 let Record { data, facts } = record;
602 let Tail { prev_hash, head } = tail;
603
604 let authenticated_drift = matches!(
605 &data.event_type,
606 AuditEventType::AgentDriftDetected | AuditEventType::AgentDriftAcknowledged
607 ) || is_seal(&data.event_type);
608
609 let partial = serde_json::json!({
610 "agent_id": data.agent_id,
611 "tool": data.tool,
612 "action": data.action,
613 "input_hash": data.input_hash,
614 "output_tokens": data.output_tokens,
615 "role": data.role,
616 "event_type": data.event_type,
617 });
618 let data_json = serde_json::to_string(&partial).map_err(|e| format!("audit serialize: {e}"))?;
619 let entry_hash = compute_entry_hash(&prev_hash, &data_json);
620
621 let signature_result =
622 crate::core::agent_identity::sign_bytes("lean-ctx", entry_hash.as_bytes())
623 .map(|sig| crate::core::agent_identity::hex_encode(&sig));
624 let signature = if authenticated_drift {
625 Some(signature_result.map_err(|e| format!("audit signature: {e}"))?)
626 } else {
627 signature_result.ok()
628 };
629
630 let facts = facts.map(|facts| {
631 let seq = v2::next_seq(&facts.session_id);
632 (facts, seq)
633 });
634 let (session_id, seq, instance_id, policy_digest, decision, output_digest) = match facts {
635 Some((facts, seq)) => (
636 Some(facts.session_id),
637 Some(seq),
638 facts.instance_id,
639 facts.policy_digest,
640 Some(facts.decision.as_str().to_string()),
641 facts.output_digest,
642 ),
643 None => (None, None, None, None, None, None),
644 };
645 let mut entry = AuditEntry {
646 timestamp: chrono::Utc::now().to_rfc3339(),
647 agent_id: data.agent_id,
648 tool: data.tool,
649 action: data.action,
650 input_hash: data.input_hash,
651 output_tokens: data.output_tokens,
652 role: data.role,
653 event_type: data.event_type,
654 prev_hash,
655 entry_hash,
656 signature,
657 v: Some(ENTRY_VERSION),
658 session_id,
659 seq,
660 instance_id,
661 policy_digest,
662 decision,
663 output_digest,
664 v2_hash: None,
665 };
666 let raw = serde_json::to_value(&entry).map_err(|e| format!("audit serialize: {e}"))?;
667 let fields = v2::v2_fields(&raw).ok_or_else(|| "audit v2: malformed facts".to_string())?;
668 entry.v2_hash = Some(v2::compute_v2_hash(&head, &fields));
669
670 let mut line = serde_json::to_string(&entry).map_err(|e| format!("audit serialize: {e}"))?;
671 line.push('\n');
672 let rollback_len = file
676 .seek(SeekFrom::End(0))
677 .map_err(|e| format!("audit seek: {e}"))?;
678 if let Err(e) = file
679 .write_all(line.as_bytes())
680 .and_then(|()| file.flush())
681 .and_then(|()| file.sync_data())
682 {
683 let _ = file.set_len(rollback_len);
684 return Err(format!("audit append: {e}"));
685 }
686 Ok(entry)
687}
688
689pub fn default_trail_path() -> Option<PathBuf> {
691 trail_path()
692}
693
694pub fn load_recent(limit: usize) -> Vec<AuditEntry> {
696 let Some(path) = trail_path() else {
697 return Vec::new();
698 };
699 load_recent_at(&path, limit)
700}
701
702const TAIL_BLOCK: u64 = 256 * 1024;
704
705pub fn load_recent_at(path: &Path, limit: usize) -> Vec<AuditEntry> {
709 let mut entries: Vec<AuditEntry> = Vec::new();
710 for file in segments::chain(path).iter().rev() {
711 if entries.len() >= limit {
712 break;
713 }
714 let mut older = load_recent_in(file, limit - entries.len());
715 older.append(&mut entries);
716 entries = older;
717 }
718 entries
719}
720
721fn load_recent_in(path: &Path, limit: usize) -> Vec<AuditEntry> {
722 use std::io::{Seek, SeekFrom};
723
724 if limit == 0 {
725 return Vec::new();
726 }
727 let Some((file, len)) = open_for_listing(path) else {
728 return Vec::new();
729 };
730 let mut start = len;
731 let mut blocks: Vec<Vec<u8>> = Vec::new();
732 let mut newlines = 0usize;
733 while start > 0 && newlines <= limit {
734 let from = start.saturating_sub(TAIL_BLOCK);
735 let mut block = Vec::new();
736 let read = (&file)
737 .seek(SeekFrom::Start(from))
738 .and_then(|_| (&file).take(start - from).read_to_end(&mut block));
739 if read.is_err() {
740 break;
741 }
742 newlines += bytecount::count(&block, b'\n');
743 blocks.push(block);
744 start = from;
745 }
746 let mut tail: Vec<u8> = blocks.into_iter().rev().flatten().collect();
747 if start > 0 {
749 let cut = tail
750 .iter()
751 .position(|&byte| byte == b'\n')
752 .map_or(tail.len(), |at| at + 1);
753 tail.drain(..cut);
754 }
755 let entries = parse_entries(std::io::BufReader::new(tail.as_slice()));
756 let skip = entries.len().saturating_sub(limit);
757 entries.into_iter().skip(skip).collect()
758}
759
760pub fn load_all_at(path: &Path) -> Vec<AuditEntry> {
762 let mut entries = Vec::new();
763 for file in segments::chain(path) {
764 if let Some((file, len)) = open_for_listing(&file) {
765 entries.extend(parse_entries(std::io::BufReader::new(file.take(len))));
766 }
767 }
768 entries
769}
770
771pub fn trail_files_since(
777 trail: &Path,
778 since: Option<chrono::DateTime<chrono::Utc>>,
779) -> Vec<PathBuf> {
780 let mut files: Vec<PathBuf> = segments::sealed(trail)
781 .into_iter()
782 .map(|(_, path)| path)
783 .filter(|path| {
784 since.is_none_or(|since| {
785 fs::File::open(path)
786 .ok()
787 .and_then(|file| read_tail_entry(&file).ok().flatten())
788 .and_then(|last| entry_time(&last))
789 .is_none_or(|sealed_at| sealed_at >= since)
790 })
791 })
792 .collect();
793 if trail.exists() {
794 files.push(trail.to_path_buf());
795 }
796 files
797}
798
799pub fn read_trail_since(
804 trail: &Path,
805 since: Option<chrono::DateTime<chrono::Utc>>,
806) -> std::io::Result<Option<String>> {
807 for _ in 0..SNAPSHOT_ATTEMPTS {
808 let before = segments::sealed(trail);
809 let files = trail_files_since(trail, since);
810 if files.is_empty() {
811 return Ok(None);
812 }
813 let mut text = String::new();
814 for file in files {
815 match fs::read_to_string(&file) {
816 Ok(part) => {
817 text.push_str(&part);
818 if !text.is_empty() && !text.ends_with('\n') {
819 text.push('\n');
820 }
821 }
822 Err(error) if error.kind() == std::io::ErrorKind::NotFound => {}
823 Err(error) => return Err(error),
824 }
825 }
826 if segments::sealed(trail) == before {
827 return Ok(Some(text));
828 }
829 }
830 Err(std::io::Error::other(
831 "the audit trail kept being sealed while it was read",
832 ))
833}
834
835const SNAPSHOT_ATTEMPTS: usize = 3;
837
838fn entry_time(entry: &serde_json::Value) -> Option<chrono::DateTime<chrono::Utc>> {
839 let stamp = entry.get("timestamp")?.as_str()?;
840 Some(chrono::DateTime::parse_from_rfc3339(stamp).ok()?.to_utc())
841}
842
843fn open_for_listing(path: &Path) -> Option<(fs::File, u64)> {
846 let file = fs::File::open(path).ok()?;
847 let locked = lock_trail_shared(&file);
850 if let Err(error) = &locked {
851 tracing::debug!("audit listing reads unlocked and may end early: {error}");
852 }
853 let len = file.metadata().ok()?.len();
854 if locked.is_ok() {
855 release_read_lock(&file);
856 }
857 Some((file, len))
858}
859
860fn parse_entries(reader: impl BufRead) -> Vec<AuditEntry> {
861 let mut entries = Vec::new();
862 for line in reader.lines().map_while(Result::ok) {
863 entries.extend(
864 serde_json::Deserializer::from_str(&line)
865 .into_iter::<AuditEntry>()
866 .map_while(Result::ok),
867 );
868 }
869 entries
870}
871
872pub fn latest_unacknowledged_drift(agent_id: &str) -> Result<Option<(String, String)>, String> {
877 let (chain, pending) = verify_chain_with_drift(Some(agent_id));
878 if !chain.source_present {
879 return Err("audit trail unavailable".to_string());
880 }
881 if !chain.valid {
882 return Err(format!(
883 "audit chain invalid at entry {}",
884 chain.first_invalid_at.unwrap_or(chain.total_entries)
885 ));
886 }
887 if chain.total_entries == 0 {
888 return Err("audit trail unavailable".to_string());
894 }
895
896 Ok(pending)
897}
898
899fn hashed_fields(value: &serde_json::Value) -> Option<serde_json::Value> {
906 let get = |key: &str| value.get(key).cloned();
907 Some(serde_json::json!({
908 "agent_id": get("agent_id")?,
909 "tool": get("tool")?,
910 "action": value.get("action").cloned().unwrap_or(serde_json::Value::Null),
911 "input_hash": get("input_hash")?,
912 "output_tokens": get("output_tokens")?,
913 "role": get("role")?,
914 "event_type": get("event_type")?,
915 }))
916}
917
918fn invalid_at(total: usize, unknown: usize) -> ChainVerifyResult {
919 ChainVerifyResult {
920 total_entries: total,
921 valid: false,
922 first_invalid_at: Some(total),
923 source_present: true,
924 unknown_event_types: unknown,
925 }
926}
927
928fn unavailable() -> ChainVerifyResult {
929 ChainVerifyResult {
930 total_entries: 0,
931 valid: false,
932 first_invalid_at: Some(0),
933 source_present: false,
934 unknown_event_types: 0,
935 }
936}
937
938fn verify_chain_with_drift(
947 agent_id: Option<&str>,
948) -> (ChainVerifyResult, Option<(String, String)>) {
949 let Some(path) = trail_path() else {
950 return (unavailable(), None);
951 };
952 verify_chain_with_drift_at(&path, agent_id)
953}
954
955fn verify_chain_with_drift_at(
956 path: &Path,
957 agent_id: Option<&str>,
958) -> (ChainVerifyResult, Option<(String, String)>) {
959 verify_chain_mode(path, agent_id, false)
960}
961
962#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
965pub(super) struct ChainState {
966 prev_hash: String,
967 head: Option<String>,
970 total: usize,
971 unknown: usize,
972 pending: std::collections::BTreeMap<String, (String, String)>,
976 tally: SignedTallyEvidence,
978 security: SignedTallyEvidence,
981 sealed: Option<u64>,
983}
984
985impl ChainState {
986 fn genesis() -> Self {
987 Self {
988 prev_hash: "genesis".to_string(),
989 head: None,
990 total: 0,
991 unknown: 0,
992 pending: std::collections::BTreeMap::new(),
993 tally: SignedTallyEvidence::default(),
994 security: SignedTallyEvidence::default(),
995 sealed: None,
996 }
997 }
998
999 fn invalid(&self) -> (ChainVerifyResult, Option<(String, String)>) {
1000 (invalid_at(self.total, self.unknown), None)
1001 }
1002
1003 fn valid(&self, agent_id: Option<&str>) -> (ChainVerifyResult, Option<(String, String)>) {
1004 (
1005 ChainVerifyResult {
1006 total_entries: self.total,
1007 valid: true,
1008 first_invalid_at: None,
1009 source_present: true,
1010 unknown_event_types: self.unknown,
1011 },
1012 agent_id.and_then(|id| self.pending.get(id).cloned()),
1013 )
1014 }
1015
1016 fn advance(
1024 &mut self,
1025 value: serde_json::Value,
1026 key: &mut Option<Result<ed25519_dalek::VerifyingKey, ()>>,
1027 ) -> Result<(), ()> {
1028 let (Some(entry_prev), Some(entry_hash)) = (
1029 value.get("prev_hash").and_then(serde_json::Value::as_str),
1030 value.get("entry_hash").and_then(serde_json::Value::as_str),
1031 ) else {
1032 return Err(());
1033 };
1034 let partial = hashed_fields(&value).ok_or(())?;
1035 if entry_prev != self.prev_hash {
1036 return Err(());
1037 }
1038 let data_json = serde_json::to_string(&partial).unwrap_or_default();
1039 if entry_hash != compute_entry_hash(&self.prev_hash, &data_json) {
1040 return Err(());
1041 }
1042 let entry_hash = entry_hash.to_string();
1043 let version = v2::version_of(&value).ok_or(())?;
1047 let newer_version = match version {
1048 0 => return Err(()),
1049 1 => false,
1050 v2::ENTRY_VERSION => {
1051 let fields = v2::v2_fields(&value).ok_or(())?;
1052 let claimed = value.get("v2_hash").and_then(serde_json::Value::as_str);
1053 if claimed != Some(v2::compute_v2_hash(self.head(), &fields).as_str()) {
1054 return Err(());
1055 }
1056 false
1057 }
1058 _ => true,
1059 };
1060 let head = v2::head_of(&value, version).ok_or(())?;
1061 let entry = serde_json::from_value::<AuditEntry>(value).map_err(|_| ())?;
1062 let seal = is_seal(&entry.event_type);
1063 if (matches!(&entry.event_type, AuditEventType::Unknown(_)) && !seal) || newer_version {
1064 self.unknown += 1;
1065 }
1066 self.sealed = if seal {
1067 Some(segments::sealed_seq(entry.action.as_deref()).ok_or(())?)
1068 } else {
1069 None
1070 };
1071 let signature_required = seal
1072 || matches!(
1073 &entry.event_type,
1074 AuditEventType::AgentDriftDetected | AuditEventType::AgentDriftAcknowledged
1075 );
1076 if signature_required && entry.signature.is_none() {
1077 return Err(());
1078 }
1079 if let Some(signature) = entry.signature.as_deref() {
1080 let signature = crate::core::agent_identity::hex_decode(signature).map_err(|_| ())?;
1081 let Ok(public_key) = key
1082 .get_or_insert_with(|| {
1083 crate::core::agent_identity::load_public_key("lean-ctx").map_err(|_| ())
1084 })
1085 .as_ref()
1086 else {
1087 return Err(());
1088 };
1089 if !crate::core::agent_identity::verify_signature(
1090 public_key.as_bytes(),
1091 entry.entry_hash.as_bytes(),
1092 &signature,
1093 ) {
1094 return Err(());
1095 }
1096 }
1097 if let Some((kind, n, _)) = entry
1098 .action
1099 .as_deref()
1100 .and_then(crate::core::security_events::parse_action)
1101 {
1102 self.security.record(kind, n, &entry.entry_hash);
1103 if kind.in_signed_tally() {
1104 self.tally.record(kind, n, &entry.entry_hash);
1105 }
1106 }
1107 match entry.event_type {
1108 AuditEventType::AgentDriftDetected => {
1109 if let Some(evidence) = entry.action {
1110 self.pending.insert(
1111 entry.agent_id,
1112 (
1113 entry.timestamp,
1114 format!("{evidence} observation={}", entry.entry_hash),
1115 ),
1116 );
1117 }
1118 }
1119 AuditEventType::AgentDriftAcknowledged
1120 if self
1121 .pending
1122 .get(&entry.agent_id)
1123 .is_some_and(|(_, evidence)| entry.action.as_ref() == Some(evidence)) =>
1124 {
1125 self.pending.remove(&entry.agent_id);
1126 }
1127 _ => {}
1128 }
1129 self.head = (head != entry_hash).then_some(head);
1130 self.prev_hash = entry_hash;
1131 self.total += 1;
1132 Ok(())
1133 }
1134
1135 fn head(&self) -> &str {
1137 self.head.as_deref().unwrap_or(&self.prev_hash)
1138 }
1139
1140 fn verify_stream(&mut self, reader: &mut impl BufRead) -> Result<(u64, bool), ()> {
1144 let mut consumed = 0u64;
1145 let mut ends_on_newline = true;
1146 let mut buf = Vec::new();
1147 let mut key = None;
1148 loop {
1149 buf.clear();
1150 let read = reader.read_until(b'\n', &mut buf).map_err(|_| ())?;
1151 if read == 0 {
1152 return Ok((consumed, ends_on_newline));
1153 }
1154 consumed += read as u64;
1155 ends_on_newline = buf.last() == Some(&b'\n');
1156 if ends_on_newline {
1157 buf.pop();
1158 if buf.last() == Some(&b'\r') {
1159 buf.pop();
1160 }
1161 }
1162 let line = std::str::from_utf8(&buf).map_err(|_| ())?;
1163 let mut saw_value = false;
1164 for parsed in serde_json::Deserializer::from_str(line).into_iter::<serde_json::Value>()
1165 {
1166 saw_value = true;
1167 self.advance(parsed.map_err(|_| ())?, &mut key)?;
1168 }
1169 if !saw_value {
1170 return Err(());
1171 }
1172 }
1173 }
1174}
1175
1176fn verify_chain_mode(
1182 path: &Path,
1183 agent_id: Option<&str>,
1184 force_full: bool,
1185) -> (ChainVerifyResult, Option<(String, String)>) {
1186 verify_chain_mode_at(path, agent_id, force_full, chrono::Utc::now().timestamp())
1187}
1188
1189fn verify_chain_mode_at(
1191 path: &Path,
1192 agent_id: Option<&str>,
1193 force_full: bool,
1194 now: i64,
1195) -> (ChainVerifyResult, Option<(String, String)>) {
1196 match verify_state_at(path, force_full, now) {
1197 Verified::Missing => (
1198 ChainVerifyResult {
1199 total_entries: 0,
1200 valid: true,
1201 first_invalid_at: None,
1202 source_present: false,
1203 unknown_event_types: 0,
1204 },
1205 None,
1206 ),
1207 Verified::Unavailable => (unavailable(), None),
1208 Verified::Invalid(state) => state.invalid(),
1209 Verified::Valid(state) => state.valid(agent_id),
1210 }
1211}
1212
1213enum Verified {
1214 Missing,
1216 Unavailable,
1218 Invalid(ChainState),
1220 Valid(ChainState),
1221}
1222
1223fn verify_state_at(path: &Path, force_full: bool, now: i64) -> Verified {
1228 for _ in 0..SNAPSHOT_ATTEMPTS {
1229 let sealed = segments::sealed(path);
1230 let active = match fs::File::open(path) {
1231 Ok(file) => Some(file),
1232 Err(error) if error.kind() == std::io::ErrorKind::NotFound => None,
1233 Err(_) => return Verified::Unavailable,
1234 };
1235 if segments::sealed(path) == sealed {
1236 return verify_snapshot(path, &sealed, active, force_full, now);
1237 }
1238 }
1239 Verified::Unavailable
1240}
1241
1242fn verify_snapshot(
1243 path: &Path,
1244 sealed: &[(u64, PathBuf)],
1245 active: Option<fs::File>,
1246 force_full: bool,
1247 now: i64,
1248) -> Verified {
1249 if sealed.is_empty() && active.is_none() {
1250 return Verified::Missing;
1251 }
1252 let day = usize::try_from(now.div_euclid(24 * 60 * 60)).unwrap_or(0);
1255 let sample = day.checked_rem(sealed.len());
1256 let mut state = ChainState::genesis();
1257 for (index, (seq, segment)) in sealed.iter().enumerate() {
1258 let Ok(file) = fs::File::open(segment) else {
1259 return Verified::Unavailable;
1260 };
1261 let max_age = (Some(index) == sample).then_some(checkpoint::FULL_REVERIFY_SECS);
1262 state = match verify_file(segment, &file, &state, force_full, max_age, now) {
1263 Verified::Valid(state) if state.sealed == Some(*seq) => state,
1265 Verified::Valid(state) => return Verified::Invalid(state),
1266 other => return other,
1267 };
1268 }
1269 match active {
1270 Some(file) => verify_file(
1271 path,
1272 &file,
1273 &state,
1274 force_full,
1275 Some(checkpoint::FULL_REVERIFY_SECS),
1276 now,
1277 ),
1278 None => Verified::Valid(state),
1279 }
1280}
1281
1282fn verify_file(
1286 path: &Path,
1287 file: &fs::File,
1288 start: &ChainState,
1289 force_full: bool,
1290 max_age: Option<i64>,
1291 now: i64,
1292) -> Verified {
1293 use std::io::{Seek, SeekFrom};
1294
1295 let mut cursor = file;
1296 if lock_trail_shared(file).is_err() {
1302 return Verified::Unavailable;
1303 }
1304 let store = checkpoint::Store::for_trail(path);
1305 let stamp = checkpoint::FileStamp::of(file);
1306 let Some(len) = stamp
1307 .as_ref()
1308 .map(|stamp| stamp.size)
1309 .or_else(|| file.metadata().ok().map(|meta| meta.len()))
1310 else {
1311 return Verified::Unavailable;
1312 };
1313 release_read_lock(file);
1314
1315 if !force_full {
1316 if let (Some(store), Some(stamp)) = (&store, &stamp) {
1317 if let Some(resumed) = store.resume(file, stamp, start, max_age, now) {
1318 let mut state = resumed.state;
1319 if resumed.offset == stamp.size {
1320 return Verified::Valid(state);
1321 }
1322 if cursor.seek(SeekFrom::Start(resumed.offset)).is_ok() {
1323 let suffix = state.verify_stream(&mut std::io::BufReader::new(
1324 file.take(len - resumed.offset),
1325 ));
1326 if !still_holds_prefix(file, len) {
1327 return Verified::Unavailable;
1328 }
1329 if let Ok((read, clean)) = suffix {
1330 if clean && resumed.offset + read == stamp.size {
1331 store.save(file, stamp, start, &state, resumed.full_verified_at);
1332 }
1333 return Verified::Valid(state);
1334 }
1335 }
1336 }
1339 }
1340 }
1341
1342 if cursor.seek(SeekFrom::Start(0)).is_err() {
1343 return Verified::Unavailable;
1344 }
1345 let mut state = start.clone();
1346 let pass = state.verify_stream(&mut std::io::BufReader::new(file.take(len)));
1347 if !still_holds_prefix(file, len) {
1349 return Verified::Unavailable;
1350 }
1351 let Ok((read, clean)) = pass else {
1352 if let Some(store) = &store {
1353 store.discard();
1354 }
1355 return Verified::Invalid(state);
1356 };
1357 if let (Some(store), Some(stamp)) = (&store, &stamp) {
1358 if clean && read == stamp.size {
1359 store.save(file, stamp, start, &state, now);
1360 }
1361 }
1362 Verified::Valid(state)
1363}
1364
1365#[derive(Debug, Clone, Default, PartialEq, Eq, Serialize, Deserialize)]
1369pub struct SignedTallyEvidence {
1370 pub counts: crate::core::security_events::SecurityCounts,
1371 pub entries: u64,
1372 pub first_hash: Option<String>,
1373 pub last_hash: Option<String>,
1374}
1375
1376impl SignedTallyEvidence {
1377 fn record(&mut self, kind: crate::core::security_events::SecurityKind, n: u64, hash: &str) {
1378 self.counts.add(kind, n);
1379 self.entries += 1;
1380 self.first_hash.get_or_insert_with(|| hash.to_string());
1381 self.last_hash = Some(hash.to_string());
1382 }
1383}
1384
1385pub fn verified_signed_tally_at(path: &Path) -> Option<SignedTallyEvidence> {
1389 match verify_state_at(path, false, chrono::Utc::now().timestamp()) {
1390 Verified::Missing => Some(SignedTallyEvidence::default()),
1391 Verified::Valid(state) => Some(state.tally),
1392 Verified::Unavailable | Verified::Invalid(_) => None,
1393 }
1394}
1395
1396pub fn verified_security_at(path: &Path) -> (ChainVerifyResult, Option<SignedTallyEvidence>) {
1402 match verify_state_at(path, false, chrono::Utc::now().timestamp()) {
1403 Verified::Missing => (
1404 ChainVerifyResult {
1405 total_entries: 0,
1406 valid: true,
1407 first_invalid_at: None,
1408 source_present: false,
1409 unknown_event_types: 0,
1410 },
1411 Some(SignedTallyEvidence::default()),
1412 ),
1413 Verified::Unavailable => (unavailable(), None),
1414 Verified::Invalid(state) => (state.invalid().0, None),
1415 Verified::Valid(state) => (state.valid(None).0, Some(state.security)),
1416 }
1417}
1418
1419pub fn verify_chain() -> ChainVerifyResult {
1422 verify_chain_with_drift(None).0
1423}
1424
1425pub fn verify_chain_full() -> ChainVerifyResult {
1428 let Some(path) = trail_path() else {
1429 return unavailable();
1430 };
1431 verify_chain_mode(&path, None, true).0
1432}
1433
1434pub fn verify_chain_at(path: &Path) -> ChainVerifyResult {
1437 verify_chain_with_drift_at(path, None).0
1438}
1439
1440pub fn hash_input(args: &serde_json::Map<String, serde_json::Value>) -> String {
1441 let serialized = serde_json::to_string(args).unwrap_or_default();
1442 let mut hasher = Sha256::new();
1443 hasher.update(serialized.as_bytes());
1444 crate::core::agent_identity::hex_encode(&hasher.finalize())
1445}
1446
1447#[cfg(test)]
1448#[path = "audit_trail_tests.rs"]
1449mod tests;