1use std::collections::HashMap;
20use std::sync::atomic::{AtomicU64, Ordering};
21use std::sync::{Arc, Mutex, MutexGuard};
22use std::time::Instant;
23
24use serde::{Deserialize, Serialize};
25use serde_json::Value;
26
27use super::StoreData;
28use super::call_log::{self, CallOutcome, CallRecord, ModelUsage, Payload, Side};
29use super::security::{AuditDecision, CheckOutcome, SecurityCheck};
30
31#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
34#[serde(rename_all = "kebab-case")]
35pub enum DecisionAction {
36 Allow,
37 Deny,
38 AskHuman,
39}
40
41#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
43pub struct RuleCitation {
44 pub caller: String,
45 pub index: usize,
46 pub name: Option<String>,
47}
48
49#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
50#[serde(rename_all = "kebab-case", tag = "kind")]
51pub enum DecisionCause {
52 Rule(RuleCitation),
54 Default { caller_block: bool },
56 RuntimeInvariant { reason: String },
60 Unexplained,
62}
63
64#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
65#[serde(rename_all = "kebab-case", tag = "kind", content = "variable")]
66pub enum FailureReasonRecord {
67 FieldMissing,
68 VariableNotBound(String),
69 NotSatisfied,
70}
71
72#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
74pub struct FailureRecord {
75 pub comparison: String,
76 pub actual: Option<Value>,
77 pub expected: Option<String>,
78 pub reason: FailureReasonRecord,
79 pub negated: bool,
80}
81
82#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
84pub struct NearMissRecord {
85 pub rule: RuleCitation,
86 pub filter: String,
87 pub failures: Vec<FailureRecord>,
88}
89
90#[derive(Debug, Clone, PartialEq, Serialize)]
92pub struct DecisionExplanation {
93 pub action: DecisionAction,
94 pub cause: DecisionCause,
95 pub near_misses: Vec<NearMissRecord>,
96}
97
98#[derive(Debug, Clone, Copy, PartialEq, Eq, Default, Serialize, Deserialize)]
100#[serde(rename_all = "kebab-case", tag = "kind")]
101pub enum EntryPath {
102 #[default]
104 GatedOp,
105 PackageCheck,
107 RedirectHop {
109 parent_call_index: u64,
110 index: u32,
111 },
112 Git,
113 FileTool,
115}
116
117#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
119pub struct SourceLine {
120 pub line: u32,
121 pub column: Option<u32>,
122}
123
124#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
127pub struct PackageLine {
128 pub package: String,
129 pub module: String,
132 #[serde(default, skip_serializing_if = "Option::is_none")]
134 pub function: Option<String>,
135 pub line: u32,
136 #[serde(default, skip_serializing_if = "Option::is_none")]
137 pub column: Option<u32>,
138}
139
140#[derive(Debug, Clone, Default, PartialEq, Eq)]
143pub struct CallOrigin {
144 pub line: Option<SourceLine>,
145 pub package: Option<PackageLine>,
146}
147
148#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize)]
150pub struct CallTicket {
151 pub call_index: u64,
153 pub seq: u64,
156 pub at_micros: u64,
158 pub line: Option<SourceLine>,
159}
160
161#[derive(Debug, Clone, Copy, Default)]
163pub struct CallSite {
164 pub ticket: Option<CallTicket>,
165 pub entry: EntryPath,
166}
167
168impl CallSite {
169 pub fn new(ticket: Option<CallTicket>, entry: EntryPath) -> Self {
170 Self { ticket, entry }
171 }
172}
173
174pub trait DecisionRecorder: Send + Sync {
177 fn begin_call(&self, caller: &str, capability: &str, origin: CallOrigin) -> CallTicket;
180 fn mark_last_filtered(&self);
182 fn wants_line(&self) -> bool;
188 fn note_line_capture(&self, frames: usize);
191 fn enter_host_call(&self) -> u64;
194 fn exit_host_call(&self, marker: u64, returned: bool);
197 fn wants_payload(&self, _call_index: u64) -> bool {
201 true
202 }
203 fn call_payload(&self, call_index: u64, side: Side, payload: Payload<'_>);
205 fn call_usage(&self, call_index: u64, usage: ModelUsage);
207}
208
209pub trait RecordObserver: Send + Sync {
216 fn call_started(&self, _call: &CallRecord) {}
217 fn decision(&self, _record: &DecisionRecord) {}
218 fn call_finished(&self, _call: &CallRecord) {}
222 fn console_line(&self, _line: &ConsoleLine<'_>) {}
225}
226
227#[derive(Debug, Clone, Copy)]
229pub struct ConsoleLine<'a> {
230 pub text: &'a str,
231 pub written_micros: u64,
233 pub after_call: Option<u64>,
235 pub truncated: bool,
237}
238
239#[derive(Debug, Clone, Serialize, Deserialize)]
240pub struct DecisionRecord {
241 pub call_index: u64,
242 pub seq: u64,
243 pub at_micros: u64,
244 pub caller: String,
245 pub capability: String,
246 pub context: Value,
248 pub context_truncated: bool,
249 pub context_digest: u64,
251 pub allowed: bool,
252 pub action: DecisionAction,
253 pub cause: DecisionCause,
254 pub near_misses: Vec<NearMissRecord>,
255 pub source: String,
257 pub rule: Option<usize>,
258 pub reason: Option<String>,
259 pub entry_path: EntryPath,
260 pub line: Option<SourceLine>,
261 pub filtered: bool,
263 pub payload_dropped: bool,
265}
266
267#[derive(Debug, Clone)]
268pub struct DecisionLogConfig {
269 pub max_decisions: usize,
271 pub max_context_value_bytes: usize,
273 pub max_recorder_bytes: u64,
276 pub max_line_capture_frames: u64,
290 pub max_calls: usize,
292 pub max_payload_bytes: usize,
295}
296
297impl Default for DecisionLogConfig {
298 fn default() -> Self {
299 Self {
300 max_decisions: 10_000,
301 max_context_value_bytes: 1024,
302 max_recorder_bytes: 16 * 1024 * 1024,
303 max_line_capture_frames: 1_000_000,
304 max_calls: 10_000,
305 max_payload_bytes: 1024 * 1024,
306 }
307 }
308}
309
310#[derive(Debug, Clone, Default)]
312pub struct DecisionLogOutput {
313 pub records: Vec<DecisionRecord>,
314 pub truncated: bool,
318 pub dropped: u64,
320 pub line_frames: u64,
322 pub calls: Vec<CallRecord>,
324 pub calls_dropped: u64,
326}
327
328const MAX_TRACKED_PAIRS: usize = 4096;
331const PAIR_OVERHEAD_BYTES: u64 = 96;
332const RECORD_BASE_BYTES: u64 = std::mem::size_of::<DecisionRecord>() as u64;
333const CALL_BASE_BYTES: u64 = std::mem::size_of::<CallRecord>() as u64;
334const RECORD_RESERVE_DIVISOR: u64 = 4;
339const MAX_CONTEXT_DEPTH: usize = 8;
340const MAX_CONTEXT_ENTRIES: usize = 64;
341
342#[derive(Default)]
343struct PairState {
344 count: u64,
345 last: Option<CallTicket>,
346}
347
348#[derive(Default)]
349struct LogState {
350 records: Vec<DecisionRecord>,
351 pairs: HashMap<(String, String), PairState>,
352 next_call_index: u64,
353 last_call_index: Option<u64>,
354 truncated: bool,
355 dropped: u64,
356 charged: u64,
357 line_frames: u64,
358 lines_exhausted: bool,
360 bytes_exhausted: bool,
363 calls: Vec<CallRecord>,
364 open_calls: Vec<(u64, usize)>,
366 calls_dropped: u64,
367}
368
369pub struct DecisionLog {
371 state: Mutex<LogState>,
372 config: DecisionLogConfig,
373 started: Instant,
374 clock: AtomicU64,
375 observer: Option<Arc<dyn RecordObserver>>,
376}
377
378impl DecisionLog {
379 pub fn install(data: &mut StoreData, config: DecisionLogConfig) -> Arc<Self> {
386 Self::install_observed(data, config, None)
387 }
388
389 pub fn install_observed(
391 data: &mut StoreData,
392 config: DecisionLogConfig,
393 observer: Option<Arc<dyn RecordObserver>>,
394 ) -> Arc<Self> {
395 let log = Self::new(config, observer);
396 data.security_check = log.wrap(data.security_check.clone());
397 log
398 }
399
400 pub fn new(config: DecisionLogConfig, observer: Option<Arc<dyn RecordObserver>>) -> Arc<Self> {
403 Arc::new(Self {
404 state: Mutex::new(LogState::default()),
405 config,
406 started: Instant::now(),
407 clock: AtomicU64::new(0),
408 observer,
409 })
410 }
411
412 pub fn wrap(self: &Arc<Self>, inner: Arc<dyn SecurityCheck>) -> Arc<dyn SecurityCheck> {
415 Arc::new(RecordingCheck {
416 inner,
417 log: self.clone(),
418 })
419 }
420
421 pub fn finish(&self) -> DecisionLogOutput {
425 let mut state = self.lock();
426 let records = std::mem::take(&mut state.records);
427 for (_, position) in std::mem::take(&mut state.open_calls) {
428 if let Some(call) = state.calls.get_mut(position) {
429 call.outcome = Some(CallOutcome::Unfinished);
430 }
431 }
432 let output = DecisionLogOutput {
433 records,
434 truncated: state.truncated,
435 dropped: state.dropped,
436 line_frames: state.line_frames,
437 calls: std::mem::take(&mut state.calls),
438 calls_dropped: state.calls_dropped,
439 };
440 state.charged = 0;
441 state.pairs.clear();
442 output
443 }
444
445 pub fn console_line(&self, text: &str, truncated: bool) {
448 let Some(observer) = &self.observer else {
449 return;
450 };
451 let after_call = self.lock().last_call_index;
452 observer.console_line(&ConsoleLine {
453 text,
454 written_micros: self.now_micros(),
455 after_call,
456 truncated,
457 });
458 }
459
460 fn lock(&self) -> MutexGuard<'_, LogState> {
461 self.state
467 .lock()
468 .unwrap_or_else(std::sync::PoisonError::into_inner)
469 }
470
471 fn now_micros(&self) -> u64 {
472 let elapsed = u64::try_from(self.started.elapsed().as_micros()).unwrap_or(u64::MAX);
473 let previous = self.clock.fetch_max(elapsed, Ordering::Relaxed);
474 previous.max(elapsed)
475 }
476
477 fn charge(&self, state: &mut LogState, bytes: u64) -> bool {
478 self.charge_within(state, bytes, self.config.max_recorder_bytes)
479 }
480
481 fn charge_within(&self, state: &mut LogState, bytes: u64, limit: u64) -> bool {
483 let next = state.charged.saturating_add(bytes);
484 if next > limit {
485 return false;
486 }
487 state.charged = next;
488 true
489 }
490
491 fn payload_limit(&self) -> u64 {
494 let max = self.config.max_recorder_bytes;
495 max.saturating_sub(max / RECORD_RESERVE_DIVISOR)
496 }
497
498 fn can_keep_another(&self, state: &LogState) -> bool {
500 state.records.len() < self.config.max_decisions
501 && state.charged.saturating_add(RECORD_BASE_BYTES) <= self.config.max_recorder_bytes
502 }
503
504 fn begin(
505 &self,
506 state: &mut LogState,
507 caller: &str,
508 capability: &str,
509 line: Option<SourceLine>,
510 ) -> CallTicket {
511 let call_index = state.next_call_index;
512 state.next_call_index = state.next_call_index.saturating_add(1);
513 state.last_call_index = Some(call_index);
514 let key = (caller.to_owned(), capability.to_owned());
515 if !state.pairs.contains_key(&key) {
516 let cost = (caller.len() + capability.len()) as u64 + PAIR_OVERHEAD_BYTES;
517 if state.pairs.len() >= MAX_TRACKED_PAIRS || !self.charge(state, cost) {
518 state.truncated = true;
519 return CallTicket {
520 call_index,
521 seq: 0,
522 at_micros: self.now_micros(),
523 line,
524 };
525 }
526 }
527 let pair = state.pairs.entry(key).or_default();
528 pair.count = pair.count.saturating_add(1);
529 let ticket = CallTicket {
530 call_index,
531 seq: pair.count,
532 at_micros: self.now_micros(),
533 line,
534 };
535 pair.last = Some(ticket);
536 ticket
537 }
538
539 fn record(&self, decision: &AuditDecision<'_>) {
541 let mut state = self.lock();
542 let ticket = match decision.site.ticket {
543 Some(ticket) => ticket,
544 None => self.begin(&mut state, decision.caller, decision.capability, None),
546 };
547 if state.records.len() >= self.config.max_decisions || state.records.try_reserve(1).is_err()
548 {
549 state.truncated = true;
550 state.dropped = state.dropped.saturating_add(1);
551 return;
552 }
553 let mut record = self.build(decision, ticket);
554 let minimal = RECORD_BASE_BYTES
555 + (record.caller.len()
556 + record.capability.len()
557 + record.source.len()
558 + record.reason.as_ref().map_or(0, String::len)) as u64;
559 let payload = payload_bytes(&record);
560 if !self.charge(&mut state, minimal.saturating_add(payload)) {
561 record.context = Value::Null;
563 record.near_misses = Vec::new();
564 record.payload_dropped = true;
565 record.context_truncated = true;
566 state.truncated = true;
567 if !self.charge(&mut state, minimal) {
568 state.bytes_exhausted = true;
569 state.dropped = state.dropped.saturating_add(1);
570 return;
571 }
572 }
573 let observed = self.observer.as_ref().map(|_| record.clone());
574 state.records.push(record);
575 drop(state);
576 if let (Some(observer), Some(record)) = (&self.observer, observed) {
577 observer.decision(&record);
578 }
579 }
580
581 fn open_call(
584 &self,
585 state: &mut LogState,
586 caller: &str,
587 capability: &str,
588 ticket: CallTicket,
589 package_line: Option<PackageLine>,
590 ) -> Option<CallRecord> {
591 let package_bytes = package_line.as_ref().map_or(0, |at| {
592 at.package.len() + at.module.len() + at.function.as_ref().map_or(0, String::len)
593 });
594 let cost = CALL_BASE_BYTES.saturating_add(
595 caller
596 .len()
597 .saturating_add(capability.len())
598 .saturating_add(package_bytes) as u64,
599 );
600 if state.calls.len() >= self.config.max_calls
601 || state.calls.try_reserve(1).is_err()
602 || state.open_calls.try_reserve(1).is_err()
603 || !self.charge(state, cost)
604 {
605 state.truncated = true;
606 state.calls_dropped = state.calls_dropped.saturating_add(1);
607 return None;
608 }
609 let call = CallRecord {
610 call_index: ticket.call_index,
611 caller: caller.to_owned(),
612 capability: capability.to_owned(),
613 started_micros: ticket.at_micros,
614 ended_micros: None,
615 outcome: None,
616 line: ticket.line,
617 package_line,
618 request: None,
619 response: None,
620 usage: None,
621 };
622 let observed = self.observer.as_ref().map(|_| call.clone());
623 state
624 .open_calls
625 .push((ticket.call_index, state.calls.len()));
626 state.calls.push(call);
627 observed
628 }
629
630 fn call_mut(state: &mut LogState, call_index: u64) -> Option<&mut CallRecord> {
631 let position = state
632 .open_calls
633 .iter()
634 .rev()
635 .find(|(index, _)| *index == call_index)
636 .map(|(_, position)| *position)?;
637 state.calls.get_mut(position)
638 }
639
640 fn build(&self, decision: &AuditDecision<'_>, ticket: CallTicket) -> DecisionRecord {
641 let digest = digest_of(decision.context);
642 let (context, mut context_truncated) = if decision.capability == "secrets.get" {
643 let name = decision.context.get("name").cloned().unwrap_or(Value::Null);
645 cap_context(
646 &serde_json::json!({ "name": name }),
647 self.config.max_context_value_bytes,
648 )
649 } else {
650 cap_context(decision.context, self.config.max_context_value_bytes)
651 };
652 let (action, cause, near_misses) = match decision.explanation {
653 Some(explanation) => (
654 enforced_action(decision.allowed, Some(explanation.action)),
655 explanation.cause.clone(),
656 self.capped_near_misses(&explanation.near_misses, &mut context_truncated),
657 ),
658 None => (
659 enforced_action(decision.allowed, None),
660 fallback_cause(decision),
661 Vec::new(),
662 ),
663 };
664 DecisionRecord {
665 call_index: ticket.call_index,
666 seq: ticket.seq,
667 at_micros: ticket.at_micros,
668 caller: decision.caller.to_owned(),
669 capability: decision.capability.to_owned(),
670 context,
671 context_truncated,
672 context_digest: digest,
673 allowed: decision.allowed,
674 action,
675 cause,
676 near_misses,
677 source: decision.source.to_owned(),
678 rule: decision.rule,
679 reason: decision.reason.map(str::to_owned),
680 entry_path: decision.site.entry,
681 line: ticket.line,
682 filtered: false,
683 payload_dropped: false,
684 }
685 }
686
687 fn capped_near_misses(
689 &self,
690 near_misses: &[NearMissRecord],
691 truncated: &mut bool,
692 ) -> Vec<NearMissRecord> {
693 let max = self.config.max_context_value_bytes;
694 near_misses
695 .iter()
696 .map(|miss| NearMissRecord {
697 rule: miss.rule.clone(),
698 filter: miss.filter.clone(),
699 failures: miss
700 .failures
701 .iter()
702 .map(|failure| FailureRecord {
703 actual: failure
704 .actual
705 .as_ref()
706 .map(|actual| cap_value(actual, max, 0, truncated)),
707 ..failure.clone()
708 })
709 .collect(),
710 })
711 .collect()
712 }
713}
714
715impl DecisionRecorder for DecisionLog {
716 fn begin_call(&self, caller: &str, capability: &str, origin: CallOrigin) -> CallTicket {
717 let mut state = self.lock();
718 let ticket = self.begin(&mut state, caller, capability, origin.line);
719 let opened = self.open_call(&mut state, caller, capability, ticket, origin.package);
720 drop(state);
721 if let (Some(observer), Some(call)) = (&self.observer, opened) {
722 observer.call_started(&call);
723 }
724 ticket
725 }
726
727 fn enter_host_call(&self) -> u64 {
728 self.lock().next_call_index
729 }
730
731 fn exit_host_call(&self, marker: u64, returned: bool) {
732 let mut state = self.lock();
733 let Some(first) = state
734 .open_calls
735 .iter()
736 .position(|(index, _)| *index >= marker)
737 else {
738 return;
739 };
740 let ended = self.now_micros();
741 let closing: Vec<_> = state.open_calls.drain(first..).collect();
742 let mut finished = Vec::new();
743 for (_, position) in closing {
744 if let Some(call) = state.calls.get_mut(position) {
745 call.ended_micros = Some(ended);
746 call.outcome = Some(if returned {
747 CallOutcome::Returned
748 } else {
749 CallOutcome::Failed
750 });
751 if self.observer.is_some() {
752 finished.push(call.without_bodies());
754 }
755 }
756 }
757 drop(state);
758 if let Some(observer) = &self.observer {
759 for call in &finished {
760 observer.call_finished(call);
761 }
762 }
763 }
764
765 fn wants_payload(&self, call_index: u64) -> bool {
766 let mut state = self.lock();
767 Self::call_mut(&mut state, call_index).is_some()
770 && state.charged.saturating_add(call_log::DIGEST_ONLY_COST)
771 <= self.config.max_recorder_bytes
772 }
773
774 fn call_payload(&self, call_index: u64, side: Side, payload: Payload<'_>) {
775 let room = {
776 let mut state = self.lock();
777 if Self::call_mut(&mut state, call_index).is_none() {
780 return;
781 }
782 self.payload_limit().saturating_sub(state.charged)
783 };
784 let record = call_log::capture(
785 &payload,
786 self.config.max_context_value_bytes,
787 room,
788 self.config.max_payload_bytes,
789 );
790 let mut state = self.lock();
791 if Self::call_mut(&mut state, call_index).is_none() {
793 return;
794 }
795 let cost = call_log::payload_cost(&record);
796 let record = if self.charge_within(&mut state, cost, self.payload_limit()) {
797 record
798 } else {
799 state.truncated = true;
801 let mut kept = record.digest_only();
802 kept.truncated = true;
803 if !self.charge(&mut state, call_log::payload_cost(&kept)) {
804 return;
805 }
806 kept
807 };
808 let Some(call) = Self::call_mut(&mut state, call_index) else {
809 return;
810 };
811 match side {
812 Side::Request => call.request = Some(Box::new(record)),
813 Side::Response => call.response = Some(Box::new(record)),
814 }
815 }
816
817 fn call_usage(&self, call_index: u64, usage: ModelUsage) {
818 let mut state = self.lock();
819 if let Some(call) = Self::call_mut(&mut state, call_index) {
820 call.usage = Some(usage);
821 }
822 }
823
824 fn wants_line(&self) -> bool {
825 let state = self.lock();
826 !state.lines_exhausted && !state.bytes_exhausted && self.can_keep_another(&state)
827 }
828
829 fn note_line_capture(&self, frames: usize) {
830 let mut state = self.lock();
831 state.line_frames = state.line_frames.saturating_add(frames as u64);
832 if state.line_frames > self.config.max_line_capture_frames {
833 state.lines_exhausted = true;
834 state.truncated = true;
835 }
836 }
837
838 fn mark_last_filtered(&self) {
839 let mut state = self.lock();
840 let Some(last) = state.last_call_index else {
841 return;
842 };
843 if let Some(record) = state
844 .records
845 .iter_mut()
846 .rev()
847 .find(|record| record.call_index == last)
848 {
849 record.filtered = true;
850 }
851 }
852}
853
854fn enforced_action(allowed: bool, explained: Option<DecisionAction>) -> DecisionAction {
857 match (allowed, explained) {
858 (true, _) => DecisionAction::Allow,
859 (false, Some(DecisionAction::AskHuman)) => DecisionAction::AskHuman,
860 (false, _) => DecisionAction::Deny,
861 }
862}
863
864fn fallback_cause(decision: &AuditDecision<'_>) -> DecisionCause {
865 if decision.source == "policy" {
866 return DecisionCause::Unexplained;
867 }
868 DecisionCause::RuntimeInvariant {
869 reason: decision.reason.unwrap_or_default().to_owned(),
870 }
871}
872
873fn digest_of(context: &Value) -> u64 {
874 use std::hash::{Hash, Hasher};
875 let Ok(text) = serde_json::to_string(context) else {
876 return 0;
877 };
878 let mut hasher = std::collections::hash_map::DefaultHasher::new();
880 text.hash(&mut hasher);
881 hasher.finish()
882}
883
884fn payload_bytes(record: &DecisionRecord) -> u64 {
885 let context = serde_json::to_string(&record.context).map_or(0, |text| text.len());
886 let misses = serde_json::to_string(&record.near_misses).map_or(0, |text| text.len());
887 (context.saturating_add(misses) as u64).saturating_mul(2)
889}
890
891pub(crate) fn cap_context(context: &Value, max_string_bytes: usize) -> (Value, bool) {
894 let mut truncated = false;
895 let capped = cap_value(context, max_string_bytes, 0, &mut truncated);
896 (capped, truncated)
897}
898
899fn cap_value(value: &Value, max_string: usize, depth: usize, truncated: &mut bool) -> Value {
900 match value {
901 Value::String(text) => Value::String(cap_text(text, max_string, truncated)),
902 Value::Array(_) | Value::Object(_) if depth >= MAX_CONTEXT_DEPTH => {
903 *truncated = true;
904 Value::String("…[truncated, nested too deep]".to_owned())
905 }
906 Value::Array(items) => {
907 if items.len() > MAX_CONTEXT_ENTRIES {
908 *truncated = true;
909 }
910 Value::Array(
911 items
912 .iter()
913 .take(MAX_CONTEXT_ENTRIES)
914 .map(|item| cap_value(item, max_string, depth + 1, truncated))
915 .collect(),
916 )
917 }
918 Value::Object(fields) => {
919 if fields.len() > MAX_CONTEXT_ENTRIES {
920 *truncated = true;
921 }
922 Value::Object(
923 fields
924 .iter()
925 .take(MAX_CONTEXT_ENTRIES)
926 .enumerate()
927 .map(|(position, (key, item))| {
928 (
929 capped_key(key, position, max_string, truncated),
930 cap_value(item, max_string, depth + 1, truncated),
931 )
932 })
933 .collect(),
934 )
935 }
936 other => other.clone(),
937 }
938}
939
940fn capped_key(key: &str, position: usize, max: usize, truncated: &mut bool) -> String {
943 let capped = cap_text(key, max, truncated);
944 if capped.len() == key.len() {
945 return capped;
946 }
947 format!("{capped} #{position}")
948}
949
950pub(crate) fn cap_text(text: &str, max: usize, truncated: &mut bool) -> String {
952 if text.len() <= max {
953 return text.to_owned();
954 }
955 *truncated = true;
956 let mut end = max;
957 while end > 0 && !text.is_char_boundary(end) {
958 end -= 1;
959 }
960 let kept = text.get(..end).unwrap_or_default();
961 format!(
962 "{kept}…[truncated, {} of {} bytes kept]",
963 kept.len(),
964 text.len()
965 )
966}
967
968struct RecordingCheck {
969 inner: Arc<dyn SecurityCheck>,
970 log: Arc<DecisionLog>,
971}
972
973impl SecurityCheck for RecordingCheck {
974 fn audit(&self, decision: AuditDecision<'_>) {
975 self.inner.audit(decision);
976 self.log.record(&decision);
977 }
978
979 fn audit_context<'a>(
980 &self,
981 capability: &str,
982 context: &'a Value,
983 cwd: &str,
984 ) -> std::borrow::Cow<'a, Value> {
985 self.inner.audit_context(capability, context, cwd)
986 }
987
988 fn check_with_cwd(
989 &self,
990 caller: &str,
991 capability: &str,
992 context: &Value,
993 cwd: &str,
994 ) -> CheckOutcome {
995 self.inner.check_with_cwd(caller, capability, context, cwd)
996 }
997
998 fn check(&self, caller: &str, capability: &str, context: &Value) -> CheckOutcome {
999 self.inner.check(caller, capability, context)
1000 }
1001
1002 fn recorder(&self) -> Option<&dyn DecisionRecorder> {
1003 Some(self.log.as_ref())
1004 }
1005
1006 fn explain(
1007 &self,
1008 caller: &str,
1009 capability: &str,
1010 context: &Value,
1011 cwd: &str,
1012 ) -> Option<DecisionExplanation> {
1013 self.inner.explain(caller, capability, context, cwd)
1014 }
1015}
1016
1017#[cfg(test)]
1018mod tests {
1019 use super::*;
1020 use crate::runtime::Vfs;
1021
1022 fn log_with(config: DecisionLogConfig) -> (StoreData, Arc<DecisionLog>) {
1023 let mut data = StoreData::with_vfs_and_cap(Vfs::none(), 1 << 20);
1024 let log = DecisionLog::install(&mut data, config);
1025 (data, log)
1026 }
1027
1028 fn audit_denied(data: &StoreData, ticket: CallTicket) {
1029 let context = serde_json::json!({});
1030 data.security_check.audit(
1031 AuditDecision::new("main", "fs.read", &context, false, "policy", None, None)
1032 .with_site(CallSite::new(Some(ticket), EntryPath::GatedOp)),
1033 );
1034 }
1035
1036 #[test]
1037 fn a_call_still_open_at_finish_is_unfinished() {
1038 let (data, log) = log_with(DecisionLogConfig::default());
1039 let recorder = data.security_check.recorder().unwrap();
1040 recorder.enter_host_call();
1041 recorder.begin_call("main", "http.get", CallOrigin::default());
1042 let output = log.finish();
1043 let [call] = output.calls.as_slice() else {
1044 panic!("one call: {output:?}");
1045 };
1046 assert_eq!(call.outcome, Some(CallOutcome::Unfinished));
1047 assert_eq!(call.ended_micros, None);
1048 }
1049
1050 #[test]
1051 fn payload_copies_leave_the_decision_reserve_free() {
1052 let max = 64 * 1024;
1053 let (data, log) = log_with(DecisionLogConfig {
1054 max_recorder_bytes: max,
1055 ..DecisionLogConfig::default()
1056 });
1057 let recorder = data.security_check.recorder().unwrap();
1058 let body = vec![b'x'; 20 * 1024];
1059 for _ in 0..6 {
1060 let ticket = recorder.begin_call("main", "http.get", CallOrigin::default());
1061 recorder.call_payload(
1062 ticket.call_index,
1063 Side::Response,
1064 Payload::meta(serde_json::Value::Null).with_body(&body),
1065 );
1066 }
1067 assert!(
1068 log.lock().charged <= max - max / RECORD_RESERVE_DIVISOR + 6 * 256,
1069 "bodies stay out of the reserve; only digests may enter it"
1070 );
1071 let output = log.finish();
1072 let responses: Vec<_> = output
1073 .calls
1074 .iter()
1075 .filter_map(|call| call.response.as_deref())
1076 .collect();
1077 assert_eq!(responses.len(), 6, "every payload keeps a record");
1078 assert!(responses.iter().any(|payload| payload.body.is_some()));
1079 let cut = responses
1080 .iter()
1081 .find(|payload| payload.body.is_none())
1082 .expect("a payload past the reserve keeps no body");
1083 assert!(cut.truncated && cut.bytes == body.len() as u64 && !cut.digest.is_empty());
1084 assert!(output.truncated);
1085 }
1086
1087 #[test]
1088 fn digest_only_records_do_not_spend_the_reserve_on_masked_names() {
1089 let max = 64 * 1024;
1090 let (data, log) = log_with(DecisionLogConfig {
1091 max_recorder_bytes: max,
1092 ..DecisionLogConfig::default()
1093 });
1094 let recorder = data.security_check.recorder().unwrap();
1095 let body = vec![b'x'; 20 * 1024];
1096 for _ in 0..6 {
1097 let ticket = recorder.begin_call("main", "http.get", CallOrigin::default());
1098 recorder.call_payload(
1099 ticket.call_index,
1100 Side::Response,
1101 Payload::meta(serde_json::Value::Null).with_body(&body),
1102 );
1103 }
1104 let before = log.lock().charged;
1105 let names: Vec<String> = (0..64)
1106 .map(|i| format!("x-token-{i}-{}", "n".repeat(900)))
1107 .collect();
1108 let ticket = recorder.begin_call("main", "http.get", CallOrigin::default());
1109 recorder.call_payload(
1110 ticket.call_index,
1111 Side::Response,
1112 Payload::meta(serde_json::Value::Null)
1113 .with_body(&body)
1114 .with_masked(names),
1115 );
1116 let spent = log.lock().charged - before;
1117 assert!(
1118 spent < 512,
1119 "a payload past the limit enters the reserve as a digest and a size: {spent}"
1120 );
1121 audit_denied(
1122 &data,
1123 recorder.begin_call("main", "fs.read", CallOrigin::default()),
1124 );
1125 let output = log.finish();
1126 assert_eq!(output.records.len(), 1, "a later denial is still recorded");
1127 let last = output
1128 .calls
1129 .iter()
1130 .rev()
1131 .find_map(|call| call.response.as_deref());
1132 assert!(last.is_some_and(|payload| payload.masked_headers.is_empty()));
1133 }
1134
1135 #[test]
1136 fn a_body_is_sized_to_the_room_its_record_leaves() {
1137 let max = 16 * 1024;
1138 let (data, log) = log_with(DecisionLogConfig {
1139 max_recorder_bytes: max,
1140 ..DecisionLogConfig::default()
1141 });
1142 let recorder = data.security_check.recorder().unwrap();
1143 let body = vec![b'x'; 64 * 1024];
1144 let meta = serde_json::json!({ "url": "https://x.test/" });
1145 let ticket = recorder.begin_call("main", "http.get", CallOrigin::default());
1146 recorder.call_payload(
1147 ticket.call_index,
1148 Side::Response,
1149 Payload::meta(meta.clone()).with_body(&body),
1150 );
1151 let charged = log.lock().charged;
1152 assert!(
1153 charged <= log.payload_limit(),
1154 "the copy fits what it is charged"
1155 );
1156 let output = log.finish();
1157 let payload = output.calls[0]
1158 .response
1159 .as_deref()
1160 .expect("a response record");
1161 let kept = match &payload.body {
1162 Some(call_log::BodyCopy::Text(text)) => text.len(),
1163 other => panic!("a cut text body is kept as text: {other:?}"),
1164 };
1165 assert!(payload.truncated && kept > 0 && kept < body.len());
1166 assert_eq!(payload.bytes, body.len() as u64);
1167 let (data, log) = log_with(DecisionLogConfig {
1169 max_recorder_bytes: max,
1170 ..DecisionLogConfig::default()
1171 });
1172 let recorder = data.security_check.recorder().unwrap();
1173 let binary = vec![0xffu8; 64 * 1024];
1174 let ticket = recorder.begin_call("main", "http.get", CallOrigin::default());
1175 recorder.call_payload(
1176 ticket.call_index,
1177 Side::Response,
1178 Payload::meta(meta).with_body(&binary),
1179 );
1180 let output = log.finish();
1181 let payload = output.calls[0]
1182 .response
1183 .as_deref()
1184 .expect("a response record");
1185 let Some(call_log::BodyCopy::Base64(text)) = &payload.body else {
1186 panic!("a binary body is kept as base64: {payload:?}");
1187 };
1188 assert!(payload.truncated);
1189 assert!(text.len() as u64 <= max - max / RECORD_RESERVE_DIVISOR);
1190 }
1191
1192 #[test]
1193 fn a_payload_is_wanted_exactly_while_its_digest_only_record_fits() {
1194 let max = 4096;
1195 let (data, log) = log_with(DecisionLogConfig {
1196 max_recorder_bytes: max,
1197 ..DecisionLogConfig::default()
1198 });
1199 let recorder = data.security_check.recorder().unwrap();
1200 recorder.enter_host_call();
1201 let ticket = recorder.begin_call("main", "http.get", CallOrigin::default());
1202 log.lock().charged = max - call_log::DIGEST_ONLY_COST;
1203 assert!(recorder.wants_payload(ticket.call_index));
1204 log.lock().charged = max - call_log::DIGEST_ONLY_COST + 1;
1205 assert!(!recorder.wants_payload(ticket.call_index));
1206 }
1207
1208 #[test]
1209 fn a_body_with_no_room_is_not_copied() {
1210 let (data, log) = log_with(DecisionLogConfig {
1211 max_recorder_bytes: 1024,
1212 ..DecisionLogConfig::default()
1213 });
1214 let recorder = data.security_check.recorder().unwrap();
1215 let ticket = recorder.begin_call("main", "http.get", CallOrigin::default());
1216 let meta = serde_json::json!({ "url": "u".repeat(400) });
1217 recorder.call_payload(
1218 ticket.call_index,
1219 Side::Response,
1220 Payload::meta(meta).with_body(&[b'x'; 4096]),
1221 );
1222 let output = log.finish();
1223 let payload = output.calls[0]
1224 .response
1225 .as_deref()
1226 .expect("a response record");
1227 assert!(payload.body.is_none() && payload.truncated);
1228 assert_eq!(payload.bytes, 4096);
1229 }
1230
1231 #[test]
1232 fn a_payload_is_wanted_while_its_call_is_open_and_the_budget_lasts() {
1233 let (data, _log) = log_with(DecisionLogConfig {
1234 max_recorder_bytes: 4096,
1235 ..DecisionLogConfig::default()
1236 });
1237 let recorder = data.security_check.recorder().unwrap();
1238 recorder.enter_host_call();
1239 let ticket = recorder.begin_call("main", "http.get", CallOrigin::default());
1240 assert!(recorder.wants_payload(ticket.call_index));
1241 assert!(
1242 !recorder.wants_payload(ticket.call_index + 1),
1243 "no such call"
1244 );
1245 recorder.exit_host_call(0, true);
1246 assert!(!recorder.wants_payload(ticket.call_index), "the call ended");
1247 }
1248
1249 #[test]
1250 fn a_full_log_stops_asking_for_lines() {
1251 let (data, log) = log_with(DecisionLogConfig {
1252 max_decisions: 2,
1253 ..DecisionLogConfig::default()
1254 });
1255 let recorder = data.security_check.recorder().unwrap();
1256 for _ in 0..2 {
1257 assert!(recorder.wants_line());
1258 audit_denied(
1259 &data,
1260 recorder.begin_call("main", "fs.read", CallOrigin::default()),
1261 );
1262 }
1263 assert!(!recorder.wants_line(), "the cap is reached");
1264 audit_denied(
1265 &data,
1266 recorder.begin_call("main", "fs.read", CallOrigin::default()),
1267 );
1268 assert!(!recorder.wants_line(), "still off once the log is full");
1269 assert!(log.finish().truncated);
1270 }
1271
1272 #[test]
1273 fn a_byte_budget_payload_drop_keeps_asking_for_lines() {
1274 let (data, log) = log_with(DecisionLogConfig {
1275 max_recorder_bytes: 4 * RECORD_BASE_BYTES,
1276 ..DecisionLogConfig::default()
1277 });
1278 let recorder = data.security_check.recorder().unwrap();
1279 let context = serde_json::json!({ "path": "p".repeat(900) });
1280 let ticket = recorder.begin_call("main", "fs.read", CallOrigin::default());
1281 data.security_check.audit(
1282 AuditDecision::new("main", "fs.read", &context, true, "policy", None, None)
1283 .with_site(CallSite::new(Some(ticket), EntryPath::GatedOp)),
1284 );
1285 assert!(recorder.wants_line(), "a minimal record still fits");
1286 let output = log.finish();
1287 assert!(output.truncated);
1288 assert!(output.records.iter().any(|r| r.payload_dropped));
1289 }
1290
1291 #[test]
1292 fn filling_the_pair_cap_keeps_asking_for_lines() {
1293 let (data, log) = log_with(DecisionLogConfig::default());
1294 let recorder = data.security_check.recorder().unwrap();
1295 for index in 0..=MAX_TRACKED_PAIRS {
1296 recorder.begin_call("main", &format!("cap.{index}"), CallOrigin::default());
1297 }
1298 assert!(recorder.wants_line());
1299 assert!(log.finish().truncated, "the pair cap truncates");
1300 }
1301
1302 #[test]
1303 fn lines_stop_once_the_byte_budget_cannot_fit_a_record() {
1304 let (data, log) = log_with(DecisionLogConfig {
1305 max_recorder_bytes: 3 * RECORD_BASE_BYTES,
1306 ..DecisionLogConfig::default()
1307 });
1308 let recorder = data.security_check.recorder().unwrap();
1309 let mut captures = 0;
1310 for _ in 0..20 {
1311 if recorder.wants_line() {
1312 captures += 1;
1313 recorder.note_line_capture(5);
1314 }
1315 audit_denied(
1316 &data,
1317 recorder.begin_call("main", "fs.read", CallOrigin::default()),
1318 );
1319 }
1320 let output = log.finish();
1321 assert!(output.dropped > 0, "records were dropped whole");
1322 assert!(
1323 captures < 20 && output.line_frames == 5 * captures,
1324 "capture stopped: {captures} captures"
1325 );
1326 assert!(
1327 captures <= output.records.len() as u64 + 1,
1328 "no captures beyond the first dropped record: {captures} for {} kept",
1329 output.records.len()
1330 );
1331 }
1332
1333 #[test]
1334 fn a_budget_below_one_bare_record_wants_no_line() {
1335 for max_recorder_bytes in [0, RECORD_BASE_BYTES - 1] {
1336 let (data, _log) = log_with(DecisionLogConfig {
1337 max_recorder_bytes,
1338 ..DecisionLogConfig::default()
1339 });
1340 let recorder = data.security_check.recorder().unwrap();
1341 assert!(!recorder.wants_line(), "budget {max_recorder_bytes}");
1342 }
1343 }
1344
1345 #[test]
1346 fn a_whole_record_drop_stops_lines_though_a_bare_record_would_fit() {
1347 let (data, log) = log_with(DecisionLogConfig {
1348 max_recorder_bytes: PAIR_OVERHEAD_BYTES
1351 + "main".len() as u64
1352 + "fs.read".len() as u64
1353 + RECORD_BASE_BYTES
1354 + 2,
1355 max_calls: 0,
1356 ..DecisionLogConfig::default()
1357 });
1358 let recorder = data.security_check.recorder().unwrap();
1359 assert!(recorder.wants_line(), "a bare record fits before any call");
1360 let ticket = recorder.begin_call("main", "fs.read", CallOrigin::default());
1361 assert!(
1362 recorder.wants_line(),
1363 "a bare record still fits after the call"
1364 );
1365 audit_denied(&data, ticket);
1366 assert!(!recorder.wants_line(), "the drop turns lines off");
1367 let output = log.finish();
1368 assert_eq!(output.dropped, 1);
1369 assert!(output.records.is_empty());
1370 }
1371
1372 #[test]
1373 fn the_line_capture_frame_budget_truncates_the_log_once_exceeded() {
1374 let (data, log) = log_with(DecisionLogConfig {
1375 max_line_capture_frames: 10,
1376 ..DecisionLogConfig::default()
1377 });
1378 let recorder = data.security_check.recorder().unwrap();
1379 recorder.note_line_capture(10);
1380 assert!(recorder.wants_line(), "at the budget is still within it");
1381 recorder.note_line_capture(1);
1382 assert!(!recorder.wants_line());
1383 assert!(log.finish().truncated);
1384 }
1385
1386 #[test]
1387 fn a_secret_read_keeps_the_key_name_and_nothing_else() {
1388 let (data, log) = log_with(DecisionLogConfig::default());
1389 let context = serde_json::json!({ "name": "api-key", "value": "s3cret" });
1390 data.security_check.audit(AuditDecision::new(
1391 "main",
1392 "secrets.get",
1393 &context,
1394 false,
1395 "invariant",
1396 None,
1397 Some("refused"),
1398 ));
1399 let output = log.finish();
1400 let [record] = output.records.as_slice() else {
1401 panic!("one record");
1402 };
1403 assert_eq!(record.context, serde_json::json!({ "name": "api-key" }));
1404 assert!(!format!("{record:?}").contains("s3cret"));
1405 }
1406
1407 #[test]
1408 fn an_exhausted_recorder_budget_keeps_the_verdict_and_drops_the_payload() {
1409 let (data, log) = log_with(DecisionLogConfig {
1411 max_recorder_bytes: 2048,
1412 ..DecisionLogConfig::default()
1413 });
1414 let context = serde_json::json!({ "path": "p".repeat(900) });
1415 let recorder = data.security_check.recorder().unwrap();
1416 let ticket = recorder.begin_call("main", "fs.read", CallOrigin::default());
1417 data.security_check.audit(
1418 AuditDecision::new("main", "fs.read", &context, true, "policy", None, None)
1419 .with_site(CallSite::new(Some(ticket), EntryPath::GatedOp)),
1420 );
1421 let output = log.finish();
1422 assert!(output.truncated);
1423 let [record] = output.records.as_slice() else {
1424 panic!("the verdict is kept: {output:?}");
1425 };
1426 assert!(record.allowed && record.payload_dropped);
1427 assert_eq!(record.context, serde_json::Value::Null);
1428 assert_ne!(record.context_digest, 0, "the digest survives");
1429 assert_eq!(
1430 data.tenant_limits.host_attached_bytes(),
1431 0,
1432 "the run's memory budget is not touched"
1433 );
1434 }
1435
1436 #[test]
1437 fn caps_mark_long_strings_deep_nesting_and_wide_containers() {
1438 let long = serde_json::json!({ "a": "é".repeat(100) });
1439 let (capped, truncated) = cap_context(&long, 11);
1440 assert!(truncated);
1441 let text = capped["a"].as_str().unwrap();
1442 assert!(
1443 text.starts_with("éééé") && text.contains("truncated"),
1444 "{text}"
1445 );
1446
1447 let mut nested = serde_json::json!(1);
1448 for _ in 0..20 {
1449 nested = serde_json::json!([nested]);
1450 }
1451 assert!(cap_context(&nested, 100).1);
1452
1453 let wide = Value::Array((0..200).map(Value::from).collect());
1454 let (capped, truncated) = cap_context(&wide, 100);
1455 assert!(truncated);
1456 assert_eq!(capped.as_array().unwrap().len(), MAX_CONTEXT_ENTRIES);
1457
1458 let small = serde_json::json!({ "host": "x", "n": [1, 2] });
1459 assert_eq!(cap_context(&small, 100), (small.clone(), false));
1460 }
1461
1462 #[test]
1463 fn finish_refunds_the_recorders_budget() {
1464 let (data, log) = log_with(DecisionLogConfig {
1465 max_recorder_bytes: 4096,
1466 ..DecisionLogConfig::default()
1467 });
1468 let context = serde_json::json!({ "path": "p".repeat(300) });
1469 for _ in 0..2 {
1470 data.security_check.audit(AuditDecision::new(
1471 "main", "fs.read", &context, true, "policy", None, None,
1472 ));
1473 let output = log.finish();
1474 assert!(!output.truncated && output.records.len() == 1, "{output:?}");
1475 }
1476 assert_eq!(data.tenant_limits.host_attached_bytes(), 0);
1477 }
1478
1479 #[test]
1480 fn long_keys_sharing_a_prefix_both_survive_capping() {
1481 let long = "k".repeat(100);
1482 let value = serde_json::json!({ format!("{long}a"): 1, format!("{long}b"): 2 });
1483 let mut truncated = false;
1484 let Value::Object(capped) = cap_value(&value, 16, 0, &mut truncated) else {
1485 panic!("expected an object");
1486 };
1487 assert!(truncated);
1488 assert_eq!(capped.len(), 2, "{capped:?}");
1489 let mut values: Vec<_> = capped.values().cloned().collect();
1490 values.sort_by_key(Value::as_i64);
1491 assert_eq!(values, vec![serde_json::json!(1), serde_json::json!(2)]);
1492 }
1493
1494 #[test]
1495 fn near_miss_actual_values_and_keys_are_capped_like_the_context() {
1496 let (data, log) = log_with(DecisionLogConfig {
1497 max_context_value_bytes: 16,
1498 ..DecisionLogConfig::default()
1499 });
1500 let explanation = DecisionExplanation {
1501 action: DecisionAction::Deny,
1502 cause: DecisionCause::Default { caller_block: true },
1503 near_misses: vec![NearMissRecord {
1504 rule: RuleCitation {
1505 caller: "main".into(),
1506 index: 0,
1507 name: None,
1508 },
1509 filter: "f".into(),
1510 failures: vec![FailureRecord {
1511 comparison: "c".into(),
1512 actual: Some(serde_json::json!({ "k".repeat(100): "v".repeat(100) })),
1513 expected: None,
1514 reason: FailureReasonRecord::NotSatisfied,
1515 negated: false,
1516 }],
1517 }],
1518 };
1519 let context = serde_json::json!({});
1520 data.security_check.audit(
1521 AuditDecision::new("main", "fs.read", &context, false, "policy", None, None)
1522 .with_explanation(Some(&explanation)),
1523 );
1524 let output = log.finish();
1525 let [record] = output.records.as_slice() else {
1526 panic!("one record");
1527 };
1528 assert!(record.context_truncated);
1529 let actual = record.near_misses[0].failures[0].actual.as_ref().unwrap();
1530 assert!(
1531 serde_json::to_string(actual).unwrap().len() < 150,
1532 "{actual}"
1533 );
1534 }
1535}