//! The audit writer: a hash-chained log of decisions, approvals and results, one JSON record per //! line. The chain is the log's integrity: every record carries the hash of the line before it, so //! any change is seen. Opening the log verifies it (a short check for an ordinary start, the whole //! log when a break is to be accepted), recovers a torn tail, or accepts a break in an older file //! when told to, and only then hands back a writer that appends the next record. use std::fs::{self, TryLockError}; use std::io; use std::io::Write; use std::os::unix::fs::{DirBuilderExt, OpenOptionsExt}; use std::path::{Path, PathBuf}; use proto::{ AuditEvent, AuditRecord, ChainFailure, ChainReport, ChainVerifier, Hash32, Timestamp, sha256, }; /// The notice appended to the log when a torn final line is recovered, as the RUNBOOK entry names. pub const RECOVERED_NOTICE: &str = "audit: recovered a torn final line\nsee docs/runbook.md#audit-recovered"; /// Where the owner looks when the audit log cannot be read, written or locked. const RUNBOOK_UNAVAILABLE: &str = "see docs/runbook.md#audit-unavailable"; /// The RUNBOOK entry a second writer points to. const RUNBOOK_ALREADY_RUNNING: &str = "see docs/runbook.md#brokerd-already-running"; /// The RUNBOOK entry a broken chain points to. const RUNBOOK_BROKEN: &str = "see docs/runbook.md#audit-chain-broken"; /// Why opening or writing the audit log can fail. #[derive(Debug)] pub enum AuditError { /// The log is already locked by another running brokerd. Locked, /// A record in the log does not chain: the log is damaged. Broken(Box), /// Asked to accept a break, but the log verifies: there is none. NothingToAccept, /// The log could not be read, written or locked. Io { what: String, source: io::Error }, /// A write failed; the writer has stopped and must be restarted. Stopped, } impl std::fmt::Display for AuditError { fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { match self { AuditError::Locked => { write!(f, "brokerd is already running\n{RUNBOOK_ALREADY_RUNNING}") } AuditError::Broken(failure) => write!( f, "{}:{}: {}\n{RUNBOOK_BROKEN}", failure.file, failure.line, failure.what ), AuditError::NothingToAccept => { write!(f, "nothing to accept: the audit log verifies") } AuditError::Io { what, source } => { write!(f, "audit: {what}: {source}\n{RUNBOOK_UNAVAILABLE}") } AuditError::Stopped => write!( f, "audit: an earlier write failed; restart brokerd\n{RUNBOOK_UNAVAILABLE}" ), } } } impl std::error::Error for AuditError { fn source(&self) -> Option<&(dyn std::error::Error + 'static)> { match self { AuditError::Io { source, .. } => Some(source), _ => None, } } } impl From for AuditError { fn from(err: io::Error) -> Self { AuditError::Io { what: "audit log".to_string(), source: err, } } } /// A name is a log file when it is `.jsonl`: 16 chars, `-` at 4 and 7, `.jsonl` at 10 /// to 15, digits elsewhere. fn is_log_name(name: &str) -> bool { let b = name.as_bytes(); if b.len() != 16 { return false; } b[4] == b'-' && b[7] == b'-' && &b[10..16] == b".jsonl" && b[0..4].iter().all(|c| c.is_ascii_digit()) && b[5..6].iter().all(|c| c.is_ascii_digit()) && b[8..10].iter().all(|c| c.is_ascii_digit()) } /// The log file name for a record's time: the day of the timestamp, `.jsonl`. fn day_name(time: Timestamp) -> String { let when = time.to_rfc3339(); format!("{}.jsonl", &when[..10]) } /// An I/O error as an audit error, the path in what. fn io(what: &str, path: impl AsRef, source: io::Error) -> AuditError { let path = path.as_ref(); AuditError::Io { what: format!("{what} {}", path.display()), source, } } /// The bytes after the last newline, once one trailing newline is removed. None if nothing is left. fn last_line(bytes: &[u8]) -> Option<&[u8]> { let bytes = match bytes.last() { Some(b'\n') => &bytes[..bytes.len() - 1], _ => bytes, }; let line = match bytes.iter().rposition(|b| *b == b'\n') { Some(pos) => &bytes[pos + 1..], None => bytes, }; if line.is_empty() { None } else { Some(line) } } /// Every log file in `dir`, by name. An unreadable directory or entry is an I/O error. fn log_files(dir: &Path) -> Result, AuditError> { let mut names: Vec = Vec::new(); let mut entries = fs::read_dir(dir).map_err(|e| io("read audit log", dir, e))?; loop { let entry = match entries.next() { Some(entry) => entry, None => break, }; let entry = entry.map_err(|e| io("read audit log", dir, e))?; let name = entry.file_name(); let name = name.to_string_lossy(); if is_log_name(&name) { names.push(name.into_owned()); } } names.sort(); Ok(names) } /// Verify the whole log, from the first record. pub fn verify_dir(dir: &Path) -> Result { let files = log_files(dir)?; let mut verifier = ChainVerifier::new(); for file in files { let path = dir.join(&file); let bytes = fs::read(&path).map_err(|e| io("read audit log", &path, e))?; verifier.feed(&file, &bytes); } Ok(verifier.finish()) } /// Verify the latest file only, resuming from the last line of the file before it, when that line /// is a record. Otherwise the whole log is verified. fn short_check(dir: &Path) -> Result { let files = log_files(dir)?; if files.len() < 2 { return verify_dir(dir); } let latest = files.last().unwrap(); let before = &files[files.len() - 2]; let path_before = dir.join(before); let bytes = fs::read(&path_before).map_err(|e| io("read audit log", &path_before, e))?; let line = match last_line(&bytes) { Some(line) => line, None => return verify_dir(dir), }; let record: AuditRecord = match serde_json::from_slice(line) { Ok(record) => record, Err(_) => return verify_dir(dir), }; let next_seq = match record.seq.checked_add(1) { Some(seq) => seq, None => return verify_dir(dir), }; let prev = match sha256(line) { Ok(hash) => hash, Err(_) => return verify_dir(dir), }; let mut verifier = ChainVerifier::resume(next_seq, prev); let path_latest = dir.join(latest); let latest_bytes = fs::read(&path_latest).map_err(|e| io("read audit log", &path_latest, e))?; verifier.feed(latest, &latest_bytes); Ok(verifier.finish()) } /// Append one record: serialise it, open the target file, write the line, and return the seq and /// the hash of the line it wrote. A fresh file is created mode 0600 and the directory synced. fn write_record( dir: &Path, target: &str, newline_first: bool, seq: u64, prev: Hash32, event: AuditEvent, time: Timestamp, ) -> Result<(u64, Hash32), AuditError> { let record = AuditRecord { seq, time, prev, event, }; let json = serde_json::to_string(&record).map_err(io::Error::other)?; let path = dir.join(target); let mut file = fs::OpenOptions::new() .append(true) .create(true) .mode(0o600) .open(&path) .map_err(|e| io("open audit log", &path, e))?; if newline_first { file.write_all(b"\n").map_err(io::Error::other)?; } file.write_all(json.as_bytes()).map_err(io::Error::other)?; file.write_all(b"\n").map_err(io::Error::other)?; file.sync_all().map_err(io::Error::other)?; fs::File::open(dir).map_err(io::Error::other)?.sync_all()?; let hash = sha256(json.as_bytes()).map_err(io::Error::other)?; Ok((seq, hash)) } /// The last record's file name: today's, or the latest if that is already current. fn target_for(latest: &Option, time: Timestamp) -> String { let candidate = day_name(time); match latest { Some(current) if candidate <= *current => current.clone(), _ => candidate, } } /// A held directory lock: the open `.lock` file, released when it drops. Kept in the `Writer` so /// the lock lasts as long as the writer; `fs::File` is not `Debug`, so this wraps it. struct Lock(fs::File); impl std::fmt::Debug for Lock { fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { f.debug_tuple("Lock").field(&"open").finish() } } /// The audit log's writer. Holds the directory lock for its whole life, so a second writer is /// refused. #[derive(Debug)] pub struct Writer { lock_file: Lock, path: PathBuf, latest: Option, prev: Hash32, next_seq: u64, stopped: bool, } impl Writer { /// Open the log at `dir`: create it, lock it, verify it, and recover or accept as needed. pub fn open(dir: &Path, accept_break: bool) -> Result { // 1. The directory, mode 0700. Create it if missing; if it already exists, that is fine. if let Err(e) = fs::DirBuilder::new().mode(0o700).create(dir) && e.kind() != io::ErrorKind::AlreadyExists { return Err(io("create audit log", dir, e)); } // 2. The lock, mode 0600: try_lock refuses a second writer while this one lives. let lock = Lock( fs::OpenOptions::new() .write(true) .create(true) .truncate(false) .mode(0o600) .open(dir.join(".lock")) .map_err(|e| io("open audit lock", dir.join(".lock"), e))?, ); match lock.0.try_lock() { Ok(()) => (), Err(TryLockError::WouldBlock) => return Err(AuditError::Locked), Err(TryLockError::Error(e)) => return Err(io("open audit lock", dir.join(".lock"), e)), } // 3. Verify the log. let report = if accept_break { verify_dir(dir)? } else { short_check(dir)? }; // The latest log file before any recovery or break, for the target of either. let latest = log_files(dir) .ok() .and_then(|files| files.into_iter().last()); let now = Timestamp::now(); let recovered = report.torn_tail.is_some(); // 4-9. A broken chain, nothing to accept, or a torn tail: write the recovery record, if any, // and set the chain state for the next append. let (prev, next_seq) = if let Some(failure) = report.failure.as_ref() { if !accept_break { return Err(AuditError::Broken(Box::new(failure.clone()))); } let (seq, hash) = write_record( dir, latest.as_deref().unwrap_or(&day_name(now)), failure.tail_torn, failure.break_seq, failure.break_prev, AuditEvent::AcceptedBreak { file: failure.file.clone(), line: failure.line, last_good: failure.last_good, }, now, )?; (hash, seq.checked_add(1).unwrap_or(report.next_seq)) } else if accept_break { return Err(AuditError::NothingToAccept); } else if let Some(torn) = report.torn_tail.as_ref() { let (seq, hash) = write_record( dir, latest.as_deref().unwrap_or(&day_name(now)), !torn.has_newline, torn.recovery_seq, torn.recovery_prev, AuditEvent::Recovery { torn_bytes: torn.bytes, torn_sha256: torn.sha256, }, now, )?; (hash, seq.checked_add(1).unwrap_or(report.next_seq)) } else { let prev = match report.head { Some(head) => head, None => { let files = log_files(dir)?; if files.is_empty() { Hash32::ZERO } else { let before = &files[files.len() - 2]; let path = dir.join(before); let bytes = fs::read(&path).map_err(|e| io("read audit log", &path, e))?; match last_line(&bytes) { Some(line) => sha256(line).map_err(io::Error::other)?, None => Hash32::ZERO, } } } }; (prev, report.next_seq) }; // 10. Hand back the writer. Ok(Opened { writer: Writer { lock_file: lock, path: dir.to_path_buf(), latest, prev, next_seq, stopped: false, }, recovered, accepted: report.failure.map(Box::new), }) } /// Append a record at `time`, in the right day's file. The seq it writes is returned. pub fn append(&mut self, time: Timestamp, event: AuditEvent) -> Result { if self.stopped { return Err(AuditError::Stopped); } let target = target_for(&self.latest, time); let (seq, hash) = match write_record( &self.path, &target, false, self.next_seq, self.prev, event, time, ) { Ok(r) => r, Err(e) => { self.stopped = true; return Err(e); } }; self.prev = hash; self.next_seq = seq.checked_add(1).unwrap_or(self.next_seq); if self.latest.as_deref() != Some(target.as_str()) { self.latest = Some(target); } Ok(seq) } /// The seq the next append writes. pub fn next_seq(&self) -> u64 { self.next_seq } } impl Drop for Writer { fn drop(&mut self) { // This field holds the .lock handle open, so the lock lasts as long as the writer. let _ = &self.lock_file; let _ = fs::remove_file(self.path.join(".lock")); } } /// What `open` hands back: the writer plus what it recovered or accepted. #[derive(Debug)] pub struct Opened { pub writer: Writer, pub recovered: bool, pub accepted: Option>, }