1use std::fs::OpenOptions;
30use std::io::BufWriter;
31use std::io::Write as _;
32use std::os::unix::fs::OpenOptionsExt as _;
33use std::path::Path;
34use std::sync::atomic::{AtomicBool, Ordering};
35use std::sync::mpsc::{Receiver, SyncSender, TrySendError, sync_channel};
36use std::sync::{Arc, Mutex};
37use std::thread::JoinHandle;
38use std::time::Duration;
39use std::time::SystemTime;
40
41use serde_json::json;
42use tracing::{error, info};
43
44use crate::decision::{DecisionRecord, Outcome, op_token};
45
46const DEFAULT_QUEUE_CAPACITY: usize = 1024;
47const REOPEN_POLL: Duration = Duration::from_secs(1);
48
49#[derive(Debug)]
56pub struct AuditLog {
57 sender: Mutex<Option<SyncSender<AuditCommand>>>,
58 reopen_requested: Arc<AtomicBool>,
59 worker: Mutex<Option<JoinHandle<()>>>,
60}
61
62impl AuditLog {
63 pub fn open(path: &Path) -> std::io::Result<Self> {
75 Self::open_with_capacity(path, DEFAULT_QUEUE_CAPACITY)
76 }
77
78 fn open_with_capacity(path: &Path, capacity: usize) -> std::io::Result<Self> {
79 let file = open_append(path)?;
80 let (sender, receiver) = sync_channel(capacity);
81 let reopen_requested = Arc::new(AtomicBool::new(false));
82 let worker_reopen_requested = Arc::clone(&reopen_requested);
83 let worker_path = path.to_path_buf();
84 let worker = std::thread::Builder::new()
85 .name("basil-audit-writer".to_string())
86 .spawn(move || {
87 run_writer(
88 &worker_path,
89 BufWriter::new(file),
90 &receiver,
91 &worker_reopen_requested,
92 );
93 })?;
94 Ok(Self {
95 sender: Mutex::new(Some(sender)),
96 reopen_requested,
97 worker: Mutex::new(Some(worker)),
98 })
99 }
100
101 pub fn append(&self, record: &DecisionRecord) {
108 let line = serialize_line(record);
109 self.try_send(AuditCommand::Line(line));
110 }
111
112 pub fn append_value(&self, value: &serde_json::Value) {
120 self.try_send(AuditCommand::Line(value.to_string()));
121 }
122
123 pub fn append_reload(
139 &self,
140 previous_generation: u64,
141 new_generation: u64,
142 outcome: &str,
143 reason: &str,
144 actor: ReloadActor,
145 ) {
146 let line =
147 serialize_reload_line(previous_generation, new_generation, outcome, reason, actor);
148 self.try_send(AuditCommand::Line(line));
149 }
150
151 pub fn request_reopen(&self) {
156 self.reopen_requested.store(true, Ordering::Release);
157 self.try_send(AuditCommand::Wake);
158 }
159
160 fn try_send(&self, cmd: AuditCommand) {
161 let sender = {
162 let guard = match self.sender.lock() {
163 Ok(g) => g,
164 Err(poisoned) => poisoned.into_inner(),
165 };
166 guard.as_ref().cloned()
167 };
168 let Some(sender) = sender else {
169 error!("audit writer is stopped; dropping audit command");
170 return;
171 };
172 let result = sender.try_send(cmd);
173 match result {
174 Ok(()) => {}
175 Err(TrySendError::Full(AuditCommand::Line(_))) => {
176 error!("audit log queue full; dropping audit line (decision is in tracing)");
177 }
178 Err(TrySendError::Full(_cmd)) => {
179 error!("audit log queue full; writer will observe pending reopen later");
180 }
181 Err(TrySendError::Disconnected(_cmd)) => {
182 error!("audit writer is stopped; dropping audit command");
183 }
184 }
185 }
186
187 #[cfg(test)]
188 fn reopen_for_test(&self) {
189 let (sender, receiver) = sync_channel(0);
190 self.reopen_requested.store(true, Ordering::Release);
191 self.try_send(AuditCommand::ReopenAck(sender));
192 receiver
193 .recv()
194 .expect("audit writer should acknowledge reopen");
195 }
196}
197
198impl Drop for AuditLog {
199 fn drop(&mut self) {
200 if let Ok(mut sender) = self.sender.lock() {
201 sender.take();
202 }
203 if let Ok(mut worker) = self.worker.lock()
204 && let Some(handle) = worker.take()
205 && handle.join().is_err()
206 {
207 error!("audit writer thread panicked during shutdown");
208 }
209 }
210}
211
212#[derive(Debug)]
213enum AuditCommand {
214 Line(String),
215 Wake,
216 #[cfg(test)]
217 ReopenAck(SyncSender<()>),
218}
219
220fn run_writer(
221 path: &Path,
222 mut writer: BufWriter<std::fs::File>,
223 receiver: &Receiver<AuditCommand>,
224 reopen_requested: &AtomicBool,
225) {
226 loop {
227 maybe_reopen(path, &mut writer, reopen_requested);
228 let cmd = match receiver.recv_timeout(REOPEN_POLL) {
229 Ok(cmd) => cmd,
230 Err(std::sync::mpsc::RecvTimeoutError::Timeout) => continue,
231 Err(std::sync::mpsc::RecvTimeoutError::Disconnected) => break,
232 };
233 match cmd {
234 AuditCommand::Line(line) => {
235 maybe_reopen(path, &mut writer, reopen_requested);
236 if let Err(e) = write_line(&mut writer, &line) {
237 error!(error = %e, "failed to append audit log line; continuing (decision is in tracing)");
238 }
239 }
240 AuditCommand::Wake => {}
241 #[cfg(test)]
242 AuditCommand::ReopenAck(sender) => {
243 maybe_reopen(path, &mut writer, reopen_requested);
244 let _ = sender.send(());
245 }
246 }
247 }
248 if let Err(e) = writer.flush() {
249 error!(error = %e, "failed to flush audit log during shutdown");
250 }
251}
252
253fn maybe_reopen(path: &Path, writer: &mut BufWriter<std::fs::File>, reopen_requested: &AtomicBool) {
254 if !reopen_requested.swap(false, Ordering::AcqRel) {
255 return;
256 }
257 if let Err(e) = writer.flush() {
258 error!(error = %e, "failed to flush audit log before reopen; continuing");
259 }
260 match open_append(path) {
261 Ok(file) => {
262 *writer = BufWriter::new(file);
263 info!(path = %path.display(), "audit log reopened");
264 }
265 Err(e) => {
266 reopen_requested.store(true, Ordering::Release);
267 error!(error = %e, path = %path.display(), "failed to reopen audit log; keeping existing handle");
268 }
269 }
270}
271
272fn open_append(path: &Path) -> std::io::Result<std::fs::File> {
273 OpenOptions::new()
274 .create(true)
275 .append(true)
276 .mode(0o600)
277 .open(path)
278}
279
280fn write_line<W: std::io::Write>(w: &mut W, line: &str) -> std::io::Result<()> {
285 w.write_all(line.as_bytes())?;
286 w.write_all(b"\n")?;
287 w.flush()
289}
290
291fn serialize_line(record: &DecisionRecord) -> String {
301 let obj = json!({
302 "event_kind": "basil.audit.authz",
303 "event_version": 2,
304 "occurred_at": timestamp(),
305 "generation": record.generation,
306 "op": op_token(record.op),
307 "target_kind": "catalog_key",
308 "target_id": record.key,
309 "actor_kind": record.actor_kind,
310 "actor_id": record.actor_id,
311 "authenticated_by": record.authenticated_by,
312 "presenter_kind": record.presenter_kind,
313 "presenter_id": record.presenter_id,
314 "decision": outcome_token(record.outcome),
315 "outcome": outcome_token(record.outcome),
316 "reason": record.reason,
317 });
318 serde_json::to_string(&obj).unwrap_or_else(|_| {
319 format!(
322 "{{\"event_kind\":\"basil.audit.authz\",\"event_version\":2,\"op\":\"{}\",\"target_kind\":\"catalog_key\",\"target_id\":\"{}\",\"actor_kind\":\"{}\",\"actor_id\":\"{}\",\"outcome\":\"{}\"}}",
323 op_token(record.op),
324 record.key,
325 record.actor_kind,
326 record.actor_id,
327 outcome_token(record.outcome),
328 )
329 })
330}
331
332#[derive(Clone, Copy, Debug, PartialEq, Eq)]
341pub enum ReloadActor {
342 Sighup,
344 Caller(u32),
347}
348
349impl ReloadActor {
350 fn kind_and_id(self) -> (&'static str, String) {
354 match self {
355 Self::Sighup => ("signal", "SIGHUP".to_string()),
356 Self::Caller(uid) => ("unix_uid", uid.to_string()),
357 }
358 }
359}
360
361fn serialize_reload_line(
372 previous_generation: u64,
373 new_generation: u64,
374 outcome: &str,
375 reason: &str,
376 actor: ReloadActor,
377) -> String {
378 let (actor_kind, actor_id) = actor.kind_and_id();
379 let obj = json!({
380 "event": {
381 "kind": "basil.audit.reload",
382 "version": 1,
383 },
384 "occurred_at": timestamp(),
385 "actor": {
386 "kind": actor_kind,
387 "id": actor_id,
388 },
389 "previous_generation": previous_generation,
390 "generation": new_generation,
391 "outcome": outcome,
392 "reason": reason,
393 });
394 serde_json::to_string(&obj).unwrap_or_else(|_| {
395 format!(
396 "{{\"event\":{{\"kind\":\"basil.audit.reload\",\"version\":1}},\"previous_generation\":{previous_generation},\"generation\":{new_generation},\"outcome\":\"{outcome}\"}}"
397 )
398 })
399}
400
401pub(crate) fn timestamp() -> String {
408 let now = SystemTime::now();
409 match now.duration_since(SystemTime::UNIX_EPOCH) {
410 Ok(dur) => format_rfc3339(dur.as_secs()),
411 Err(e) => {
412 let secs = e.duration().as_secs();
414 format!("-{secs}")
415 }
416 }
417}
418
419fn format_rfc3339(secs: u64) -> String {
425 let days = secs / 86_400;
426 let secs_of_day = secs % 86_400;
427 let (hour, min, sec) = (
428 secs_of_day / 3600,
429 (secs_of_day % 3600) / 60,
430 secs_of_day % 60,
431 );
432
433 let z = days + 719_468;
435 let era = z / 146_097;
436 let doe = z % 146_097; let yoe = (doe - doe / 1460 + doe / 36_524 - doe / 146_096) / 365; let y = yoe + era * 400;
439 let doy = doe - (365 * yoe + yoe / 4 - yoe / 100); let mp = (5 * doy + 2) / 153; let day = doy - (153 * mp + 2) / 5 + 1; let month = if mp < 10 { mp + 3 } else { mp - 9 }; let year = if month <= 2 { y + 1 } else { y };
444
445 format!("{year:04}-{month:02}-{day:02}T{hour:02}:{min:02}:{sec:02}Z")
446}
447
448const fn outcome_token(outcome: Outcome) -> &'static str {
450 outcome.as_str()
451}
452
453#[cfg(test)]
454mod tests {
455 use super::*;
456 use crate::catalog::policy::Op;
457 use std::io;
458
459 fn allow_record() -> DecisionRecord {
461 DecisionRecord {
462 generation: 1,
463 op: Op::Sign,
464 key: "nats.account".to_string(),
465 actor_kind: "subject".to_string(),
466 actor_id: "svc.nats".to_string(),
467 authenticated_by: vec!["unix_peercred:svc.nats".to_string()],
468 presenter_kind: "unix_peercred".to_string(),
469 presenter_id: "svc-nats(9002)".to_string(),
470 outcome: Outcome::Allow,
471 reason: "subject:svc.nats".to_string(),
472 }
473 }
474
475 fn deny_record() -> DecisionRecord {
476 DecisionRecord {
477 generation: 1,
478 op: Op::Decrypt,
479 key: "secret.payload".to_string(),
480 actor_kind: "subject".to_string(),
481 actor_id: "svc.api".to_string(),
482 authenticated_by: vec!["unix_peercred:svc.api".to_string()],
483 presenter_kind: "unix_peercred".to_string(),
484 presenter_id: "svc-api(7777)".to_string(),
485 outcome: Outcome::Deny,
486 reason: "not_permitted".to_string(),
487 }
488 }
489
490 #[test]
491 fn serialized_line_is_valid_json_with_expected_fields() {
492 let line = serialize_line(&allow_record());
493 assert!(!line.contains('\n'), "a line must be single-line JSONL");
494 let v: serde_json::Value = serde_json::from_str(&line).expect("audit line must be JSON");
495 assert_eq!(v["event_kind"], "basil.audit.authz");
496 assert_eq!(v["event_version"], 2);
497 assert_eq!(v["generation"], 1);
498 assert_eq!(v["op"], "sign");
499 assert_eq!(v["target_kind"], "catalog_key");
500 assert_eq!(v["target_id"], "nats.account");
501 assert_eq!(v["actor_kind"], "subject");
502 assert_eq!(v["actor_id"], "svc.nats");
503 assert_eq!(v["authenticated_by"][0], "unix_peercred:svc.nats");
504 assert_eq!(v["presenter_kind"], "unix_peercred");
505 assert_eq!(v["presenter_id"], "svc-nats(9002)");
506 assert_eq!(v["decision"], "allow");
507 assert_eq!(v["outcome"], "allow");
508 assert_eq!(v["reason"], "subject:svc.nats");
509 let ts = v["occurred_at"]
511 .as_str()
512 .expect("timestamp must be a string");
513 assert!(ts.ends_with('Z'), "ts is RFC3339 UTC: {ts}");
514 assert!(ts.contains('T'), "ts is RFC3339 datetime: {ts}");
515 }
516
517 #[test]
518 fn reload_line_is_valid_json_with_expected_fields() {
519 let applied = serialize_reload_line(1, 2, "applied", "signal", ReloadActor::Sighup);
520 assert!(!applied.contains('\n'), "a line must be single-line JSONL");
521 let v: serde_json::Value =
522 serde_json::from_str(&applied).expect("reload line must be JSON");
523 assert_eq!(v["event"]["kind"], "basil.audit.reload");
524 assert_eq!(v["event"]["version"], 1);
525 assert_eq!(v["actor"]["kind"], "signal");
526 assert_eq!(v["actor"]["id"], "SIGHUP");
527 assert_eq!(v["previous_generation"], 1);
528 assert_eq!(v["generation"], 2);
529 assert_eq!(v["outcome"], "applied");
530 assert_eq!(v["reason"], "signal");
531
532 let rejected =
534 serialize_reload_line(5, 5, "rejected", "validation_failed", ReloadActor::Sighup);
535 let r: serde_json::Value =
536 serde_json::from_str(&rejected).expect("reject line must be JSON");
537 assert_eq!(r["previous_generation"], 5);
538 assert_eq!(r["generation"], 5);
539 assert_eq!(r["outcome"], "rejected");
540 assert_eq!(r["reason"], "validation_failed");
541 }
542
543 #[test]
548 fn reload_actor_distinguishes_signal_from_rpc_caller() {
549 let rpc = serialize_reload_line(1, 2, "applied", "admin_rpc", ReloadActor::Caller(4242));
550 let v: serde_json::Value =
551 serde_json::from_str(&rpc).expect("rpc reload line must be JSON");
552 assert_eq!(v["actor"]["kind"], "unix_uid");
553 assert_eq!(v["actor"]["id"], "4242");
554 let sig = serialize_reload_line(1, 2, "applied", "admin_rpc", ReloadActor::Sighup);
556 let s: serde_json::Value = serde_json::from_str(&sig).expect("sighup line must be JSON");
557 assert_eq!(v["event"], s["event"]);
558 assert_eq!(v["previous_generation"], s["previous_generation"]);
559 assert_eq!(v["generation"], s["generation"]);
560 assert_eq!(v["outcome"], s["outcome"]);
561 assert_eq!(v["reason"], s["reason"]);
562 assert_ne!(v["actor"], s["actor"], "only the actor differs");
563 }
564
565 #[test]
566 fn deny_line_is_shaped_and_carries_reason() {
567 let line = serialize_line(&deny_record());
568 let v: serde_json::Value = serde_json::from_str(&line).expect("audit line must be JSON");
569 assert_eq!(v["op"], "decrypt");
570 assert_eq!(v["outcome"], "deny");
571 assert_eq!(v["reason"], "not_permitted");
572 assert_eq!(v["actor_kind"], "subject");
573 assert_eq!(v["actor_id"], "svc.api");
574 assert_eq!(v["presenter_id"], "svc-api(7777)");
575 }
576
577 #[test]
578 fn appended_file_has_one_parseable_jsonl_line_per_record() {
579 let dir = std::env::temp_dir();
580 let path = dir.join(format!(
581 "vault-yul-async-audit-{}.jsonl",
582 std::process::id()
583 ));
584 let _ = std::fs::remove_file(&path);
585
586 let log = AuditLog::open(&path).expect("open temp audit log");
587 log.append(&allow_record());
588 log.append(&deny_record());
589 drop(log); let body = std::fs::read_to_string(&path).expect("read back audit file");
592 let lines: Vec<&str> = body.lines().collect();
593 assert_eq!(lines.len(), 2, "one JSONL line per appended record");
594
595 let recs: Vec<serde_json::Value> = lines
596 .iter()
597 .map(|l| serde_json::from_str(l).expect("each line is JSON"))
598 .collect();
599 assert_eq!(recs[0]["event_kind"], "basil.audit.authz");
600 assert_eq!(recs[0]["op"], "sign");
601 assert_eq!(recs[0]["outcome"], "allow");
602 assert_eq!(recs[1]["op"], "decrypt");
603 assert_eq!(recs[1]["outcome"], "deny");
604
605 assert!(body.contains("svc-nats(9002)"));
609 assert!(!body.contains("BEGIN"), "no PEM/key material in audit log");
610
611 let _ = std::fs::remove_file(&path);
612 }
613
614 #[test]
615 fn reopen_writes_new_records_to_new_path_inode() {
616 let dir = std::env::temp_dir();
617 let path = dir.join(format!(
618 "vault-xrk-reopen-audit-{}.jsonl",
619 std::process::id()
620 ));
621 let rotated = dir.join(format!(
622 "vault-xrk-reopen-audit-{}.jsonl.1",
623 std::process::id()
624 ));
625 let _ = std::fs::remove_file(&path);
626 let _ = std::fs::remove_file(&rotated);
627
628 let log = AuditLog::open(&path).expect("open temp audit log");
629 log.append(&allow_record());
630 log.reopen_for_test();
631
632 std::fs::rename(&path, &rotated).expect("simulate logrotate rename");
633 log.reopen_for_test();
634 log.append(&deny_record());
635 drop(log);
636
637 let old_body = std::fs::read_to_string(&rotated).expect("read rotated audit file");
638 let new_body = std::fs::read_to_string(&path).expect("read reopened audit file");
639 assert_eq!(old_body.lines().count(), 1);
640 assert!(old_body.contains("\"outcome\":\"allow\""));
641 assert_eq!(new_body.lines().count(), 1);
642 assert!(new_body.contains("\"outcome\":\"deny\""));
643
644 let _ = std::fs::remove_file(&path);
645 let _ = std::fs::remove_file(&rotated);
646 }
647
648 #[test]
649 fn bounded_queue_drops_instead_of_blocking_request_path() {
650 let dir = std::env::temp_dir();
651 let path = dir.join(format!(
652 "vault-yul-bounded-audit-{}.jsonl",
653 std::process::id()
654 ));
655 let _ = std::fs::remove_file(&path);
656
657 let log = AuditLog::open_with_capacity(&path, 1).expect("open temp audit log");
658 for _ in 0..128 {
659 log.append(&allow_record());
660 }
661 drop(log);
662
663 let body = std::fs::read_to_string(&path).expect("read back audit file");
664 assert!(
665 !body.is_empty(),
666 "bounded writer should persist the lines it accepted"
667 );
668 for line in body.lines() {
669 let value: serde_json::Value =
670 serde_json::from_str(line).expect("accepted audit line remains JSON");
671 assert_eq!(value["op"], "sign");
672 assert_eq!(value["target_id"], "nats.account");
673 }
674
675 let _ = std::fs::remove_file(&path);
676 }
677
678 #[test]
679 fn append_is_owner_only_0600() {
680 use std::os::unix::fs::PermissionsExt as _;
681 let dir = std::env::temp_dir();
682 let path = dir.join(format!("vault-vq5-mode-{}.jsonl", std::process::id()));
683 let _ = std::fs::remove_file(&path);
684
685 let log = AuditLog::open(&path).expect("open temp audit log");
686 log.append(&allow_record());
687 drop(log);
688
689 let mode = std::fs::metadata(&path)
690 .expect("stat audit file")
691 .permissions()
692 .mode()
693 & 0o777;
694 assert_eq!(mode, 0o600, "audit file is owner-only");
695
696 let _ = std::fs::remove_file(&path);
697 }
698
699 struct FailingWriter;
702
703 impl io::Write for FailingWriter {
704 fn write(&mut self, _buf: &[u8]) -> io::Result<usize> {
705 Err(io::Error::other("disk go boom"))
706 }
707 fn flush(&mut self) -> io::Result<()> {
708 Err(io::Error::other("disk go boom"))
709 }
710 }
711
712 #[test]
713 fn a_failing_sink_returns_error_not_panic() {
714 let line = serialize_line(&allow_record());
719 let mut bad = FailingWriter;
720 let err = write_line(&mut bad, &line);
721 assert!(err.is_err(), "a bad sink yields an error, not a panic");
722 }
723
724 #[test]
725 fn rfc3339_formats_known_epochs() {
726 assert_eq!(format_rfc3339(0), "1970-01-01T00:00:00Z");
728 assert_eq!(format_rfc3339(1_609_459_200), "2021-01-01T00:00:00Z");
730 assert_eq!(format_rfc3339(1_582_979_696), "2020-02-29T12:34:56Z");
732 }
733}