Add the audit writer with its startup check
Implemented-By: OpenCode session (model recorded in docs/implementer-log.md)
This commit is contained in:
@@ -0,0 +1,423 @@
|
||||
//! 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<ChainFailure>),
|
||||
/// 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<io::Error> 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 `<YYYY-MM-DD>.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<Path>, 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<Vec<String>, AuditError> {
|
||||
let mut names: Vec<String> = 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<ChainReport, AuditError> {
|
||||
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<ChainReport, AuditError> {
|
||||
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<String>, 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<String>,
|
||||
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<Opened, AuditError> {
|
||||
// 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<u64, AuditError> {
|
||||
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<Box<ChainFailure>>,
|
||||
}
|
||||
Reference in New Issue
Block a user