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