server/lib/src/audit/chain.rs
//! **The audit hash chain** — how [`FileAuditLog`](super::FileAuditLog) makes its
//! one append-only file tamper-evident, and how [`verify_file`] proves it.
//!
//! ## The line
//!
//! Every line of `audit.jsonl` is one flat JSON object: the event's fields
//! ([`AuditEvent::to_json_map`](super::AuditEvent::to_json_map)) plus three
//! chain fields.
//!
//! | key | value |
//! |---|---|
//! | `seq` | the line's place in the chain, from 0, checkpoints included |
//! | `prev` | the `hash` of the line before; [`GENESIS`] for the first line |
//! | `hash` | SHA-256 (lowercase hex) of the line's [canonical](canonical) JSON without `hash` |
//!
//! Because `prev` is inside what `hash` covers, each hash commits to the whole
//! history before it. An edited line no longer hashes to its `hash`; a removed
//! or inserted line breaks `seq` and the next line's `prev`.
//!
//! ## The checkpoint
//!
//! Every `checkpoint_every` records (`audit.checkpoint_every`, default
//! [`DEFAULT_CHECKPOINT_EVERY`]) the writer appends a checkpoint line, itself a
//! link of the chain: `kind: "checkpoint"`, `head` (the hash of the line before
//! it), and, when the server has an attestation signer, `envelope` (a DSSE
//! envelope of type [`CHECKPOINT_PAYLOAD_TYPE`] over
//! [`checkpoint_payload`]`(seq, head, ts)`) and `cert` (the signer certificate,
//! base64 DER). The signature is what an attacker who rewrites the whole file
//! consistently cannot reproduce.
//!
//! ## Verify
//!
//! [`verify_file`] walks the file once and stops at the FIRST break, which it
//! names by line number, `seq` and reason (`holger-audit-4`). Given trust roots
//! it also checks every checkpoint signature; without them it reports how many
//! signatures it did not check rather than calling them good.
use crate::lock::LockOrRecover as _;
use std::fs::{File, OpenOptions};
use std::io::{BufRead, BufReader, Read, Seek, SeekFrom, Write};
use std::path::{Path, PathBuf};
use std::sync::{Condvar, Mutex};
use std::time::Duration;
use base64::Engine as _;
use super::{AuditError, AuditEvent};
/// The `prev` of the first line.
pub const GENESIS: &str = "0000000000000000000000000000000000000000000000000000000000000000";
/// Records between two checkpoints when the config names no number.
pub const DEFAULT_CHECKPOINT_EVERY: u64 = 256;
/// The DSSE `payloadType` of a signed checkpoint.
pub const CHECKPOINT_PAYLOAD_TYPE: &str = "application/vnd.holger.audit-checkpoint.v1+json";
/// The `kind` of a checkpoint line.
pub const CHECKPOINT_KIND: &str = "checkpoint";
/// Whether a parsed line is a checkpoint rather than an event.
pub fn is_checkpoint(v: &serde_json::Value) -> bool {
v.get("kind").and_then(|k| k.as_str()) == Some(CHECKPOINT_KIND)
}
/// **The canonical form a hash is taken over**: the object's keys sorted, no
/// whitespace, `hash` left out. Built by hand rather than by `serde_json`'s map
/// order, so it does not depend on whether some crate in the build turned
/// `preserve_order` on.
pub fn canonical(obj: &serde_json::Map<String, serde_json::Value>) -> String {
let mut keys: Vec<&String> = obj.keys().filter(|k| k.as_str() != "hash").collect();
keys.sort();
let mut out = String::from("{");
for (i, k) in keys.iter().enumerate() {
if i > 0 {
out.push(',');
}
out.push_str(&serde_json::Value::String((*k).clone()).to_string());
out.push(':');
out.push_str(&canonical_value(&obj[k.as_str()]));
}
out.push('}');
out
}
fn canonical_value(v: &serde_json::Value) -> String {
match v {
serde_json::Value::Object(m) => canonical(m),
serde_json::Value::Array(a) => {
let parts: Vec<String> = a.iter().map(canonical_value).collect();
format!("[{}]", parts.join(","))
}
other => other.to_string(),
}
}
/// The chain hash of one line's object.
pub fn line_hash(obj: &serde_json::Map<String, serde_json::Value>) -> String {
nornir_hash::sha256_hex(canonical(obj).as_bytes())
}
/// Add `seq`, `prev` and `hash` to `obj` and render the line (with its newline).
pub fn seal(
mut obj: serde_json::Map<String, serde_json::Value>,
seq: u64,
prev: &str,
) -> (String, String) {
obj.insert("seq".into(), seq.into());
obj.insert("prev".into(), prev.into());
obj.remove("hash");
let hash = line_hash(&obj);
obj.insert("hash".into(), hash.clone().into());
let mut line = serde_json::Value::Object(obj).to_string();
line.push('\n');
(line, hash)
}
/// The bytes a checkpoint's DSSE envelope signs.
pub fn checkpoint_payload(seq: u64, head: &str, ts: i64) -> Vec<u8> {
let mut m = serde_json::Map::new();
m.insert("type".into(), "holger.audit.checkpoint/v1".into());
m.insert("seq".into(), seq.into());
m.insert("head".into(), head.into());
m.insert("ts".into(), ts.into());
canonical(&m).into_bytes()
}
/// A signed checkpoint: the envelope and the certificate a verifier needs.
#[derive(Debug, Clone)]
pub struct SignedCheckpoint {
/// The DSSE envelope, JSON.
pub envelope_json: String,
/// The signer certificate, DER.
pub cert_der: Vec<u8>,
}
/// **Whatever signs checkpoints** — the server's attestation signer
/// (`crate::attest::AttestSigner`) implements it in a build with `sign`.
pub trait CheckpointSigner: Send + Sync {
/// Sign `payload` as a DSSE envelope of type `payload_type`.
fn sign_checkpoint(
&self,
payload_type: &str,
payload: &[u8],
) -> Result<SignedCheckpoint, String>;
}
/// A signer that refuses every checkpoint: what a configured signer that could
/// not be loaded becomes, so its checkpoints are refused (`holger-audit-11`)
/// instead of written unsigned.
#[derive(Debug, Clone)]
pub struct RefusingSigner(pub String);
impl CheckpointSigner for RefusingSigner {
fn sign_checkpoint(
&self,
_payload_type: &str,
_payload: &[u8],
) -> Result<SignedCheckpoint, String> {
Err(self.0.clone())
}
}
/// **The signal a follower waits on.** The writer bumps it after each line and
/// never waits; a SIEM exporter sleeps on it with a timeout.
#[derive(Debug, Default)]
pub struct TailSignal {
n: Mutex<u64>,
cv: Condvar,
}
impl TailSignal {
/// One more line was written.
pub fn bump(&self) {
{
let mut g = self.n.lock_or_recover();
*g = g.wrapping_add(1);
}
self.cv.notify_all();
}
/// The current count.
pub fn now(&self) -> u64 {
*self.n.lock_or_recover()
}
/// Wait until the count moves past `seen`, or `timeout`. Returns the count.
pub fn wait_past(&self, seen: u64, timeout: Duration) -> u64 {
let g = self.n.lock_or_recover();
let (g, _) = crate::lock::recover(self.cv.wait_timeout_while(g, timeout, |n| *n == seen));
*g
}
}
/// **The writer's end of the chain**: the open file, the next `seq` and the
/// current head. Every write takes an exclusive `flock` on the file, re-reads the
/// tail when the file grew under it (a second writer in another process, such as
/// a `bundle import`), appends, `fsync`s and lets go: two writers extend one
/// chain rather than forking it, and nothing returns before its bytes are on disk.
#[derive(Debug)]
pub struct ChainWriter {
path: PathBuf,
file: File,
next_seq: u64,
head: String,
len: u64,
since: u64,
synced_next_seq: u64,
}
/// An exclusive `flock` on the chain file for one write, released on drop. It is
/// taken on a duplicate of the writer's descriptor (one open file description, so
/// taking it again inside a held write is no deadlock).
struct WriteLock(File);
impl WriteLock {
fn take(f: &File) -> Result<Self, AuditError> {
let f = f.try_clone()?;
rustix::fs::flock(&f, rustix::fs::FlockOperation::LockExclusive)
.map_err(|e| AuditError::Io(e.into()))?;
Ok(Self(f))
}
}
impl Drop for WriteLock {
fn drop(&mut self) {
let _ = rustix::fs::flock(&self.0, rustix::fs::FlockOperation::Unlock);
}
}
impl ChainWriter {
/// Open `path` for append and pick the chain up from its last line.
pub fn open(path: &Path) -> Result<Self, AuditError> {
let file = OpenOptions::new().create(true).append(true).open(path)?;
let len = file.metadata()?.len();
let (next_seq, head) = read_tail(path)?;
Ok(Self {
path: path.to_path_buf(),
file,
next_seq,
head,
len,
since: 0,
synced_next_seq: next_seq,
})
}
/// The `seq` after the last line this writer has `fsync`ed.
pub fn synced_next_seq(&self) -> u64 {
self.synced_next_seq
}
/// Records written since the last checkpoint.
pub fn since_checkpoint(&self) -> u64 {
self.since
}
fn resync(&mut self) -> Result<(), AuditError> {
let len = self.file.metadata()?.len();
if len != self.len {
let (next_seq, head) = read_tail(&self.path)?;
self.next_seq = next_seq;
self.head = head;
self.len = len;
}
Ok(())
}
fn write_obj(
&mut self,
obj: serde_json::Map<String, serde_json::Value>,
) -> Result<(), AuditError> {
let _lock = WriteLock::take(&self.file)?;
self.resync()?;
let (line, hash) = seal(obj, self.next_seq, &self.head);
self.file.write_all(line.as_bytes())?;
self.file.sync_data()?;
self.next_seq += 1;
self.head = hash;
self.len += line.len() as u64;
self.synced_next_seq = self.next_seq;
Ok(())
}
/// Append one event as the next link.
pub fn append(&mut self, event: &AuditEvent) -> Result<(), AuditError> {
self.write_obj(event.to_json_map())?;
self.since += 1;
Ok(())
}
/// **Append several events as consecutive links, in ONE write** (stinky 30).
///
/// The chain is serial by construction — every link hashes its predecessor —
/// so what a concurrent caller can share is the I/O, not the hashing: one
/// tail check and one `write_all` for the whole run of lines instead of one
/// each. The writer's state moves only after the write succeeded, exactly as
/// for a single [`append`](Self::append).
pub fn append_maps(
&mut self,
objs: Vec<serde_json::Map<String, serde_json::Value>>,
) -> Result<(), AuditError> {
if objs.is_empty() {
return Ok(());
}
let _lock = WriteLock::take(&self.file)?;
self.resync()?;
let n = objs.len() as u64;
let (mut seq, mut head) = (self.next_seq, self.head.clone());
let mut buf = String::new();
for obj in objs {
let (line, hash) = seal(obj, seq, &head);
buf.push_str(&line);
seq += 1;
head = hash;
}
self.file.write_all(buf.as_bytes())?;
// One fsync per batch, before any caller of the batch returns.
self.file.sync_data()?;
self.next_seq = seq;
self.head = head;
self.len += buf.len() as u64;
self.since += n;
self.synced_next_seq = seq;
Ok(())
}
/// Append a checkpoint naming the current head, signed when `signer` is given.
///
/// A checkpoint the given signer cannot sign is REFUSED (`holger-audit-11`):
/// nothing is written, and the caller retries after the next record. Only a
/// writer with NO signer configured writes an `unsigned` checkpoint, and it
/// says so on the line.
pub fn write_checkpoint(
&mut self,
signer: Option<&dyn CheckpointSigner>,
) -> Result<(), AuditError> {
// Held from the tail check to the write: the head the checkpoint names and
// signs is the file's tail when it is written.
let _lock = WriteLock::take(&self.file)?;
self.resync()?;
let ts = super::now_nanos();
let mut m = serde_json::Map::new();
m.insert("kind".into(), CHECKPOINT_KIND.into());
m.insert("ts".into(), ts.into());
m.insert("head".into(), self.head.clone().into());
match signer {
Some(s) => match s.sign_checkpoint(
CHECKPOINT_PAYLOAD_TYPE,
&checkpoint_payload(self.next_seq, &self.head, ts),
) {
Ok(signed) => {
m.insert("envelope".into(), signed.envelope_json.into());
m.insert(
"cert".into(),
base64::engine::general_purpose::STANDARD
.encode(&signed.cert_der)
.into(),
);
}
Err(e) => {
return Err(AuditError::Other(holger_errcode::AUDIT_CHECKPOINT_REFUSED.tag(format!(
"the checkpoint at seq {} could not be signed ({e}); it is refused, never written unsigned, and retried after the next record",
self.next_seq
))));
}
},
None => {
m.insert(
"unsigned".into(),
"no attestation signer is configured".into(),
);
}
}
self.write_obj(m)?;
self.since = 0;
Ok(())
}
/// `fsync` the file.
pub fn sync(&mut self) -> Result<(), AuditError> {
self.file.sync_data()?;
Ok(())
}
}
/// The next `seq` and the head of the chain in `path`. An empty or missing file
/// starts at the genesis. A last line written before the chain existed carries
/// no `hash`: the chain then continues from the hash of that line's raw bytes,
/// and [`verify_file`] reports the unchained lines as the break they are.
pub fn read_tail(path: &Path) -> Result<(u64, String), AuditError> {
let mut f = match File::open(path) {
Ok(f) => f,
Err(e) if e.kind() == std::io::ErrorKind::NotFound => return Ok((0, GENESIS.to_string())),
Err(e) => return Err(e.into()),
};
let len = f.metadata()?.len();
if len == 0 {
return Ok((0, GENESIS.to_string()));
}
// Read backwards in growing windows until the window holds a whole last line.
let mut window: u64 = 64 * 1024;
let last = loop {
let start = len.saturating_sub(window);
f.seek(SeekFrom::Start(start))?;
let mut buf = Vec::with_capacity((len - start) as usize);
Read::by_ref(&mut f)
.take(len - start)
.read_to_end(&mut buf)?;
let text = String::from_utf8_lossy(&buf).into_owned();
let trimmed = text.trim_end_matches(['\n', '\r']);
match trimmed.rfind('\n') {
Some(i) => break trimmed[i + 1..].to_string(),
None if start == 0 => break trimmed.to_string(),
None => window = window.saturating_mul(4),
}
};
let v: serde_json::Value = serde_json::from_str(&last).map_err(|e| {
AuditError::Other(format!(
"the last audit line does not parse, so the chain cannot continue from it: {e}"
))
})?;
match (
v.get("seq").and_then(|s| s.as_u64()),
v.get("hash").and_then(|h| h.as_str()),
) {
(Some(seq), Some(hash)) => Ok((seq + 1, hash.to_string())),
_ => {
let n = BufReader::new(File::open(path)?)
.lines()
.map_while(Result::ok)
.filter(|l| !l.trim().is_empty())
.count();
Ok((n as u64, nornir_hash::sha256_hex(last.as_bytes())))
}
}
}
/// Where the chain broke, and why.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct ChainBreak {
/// 1-based line number in `audit.jsonl`.
pub line: usize,
/// The `seq` the line carries, when it carries one.
pub seq: Option<u64>,
/// What is wrong, in a sentence.
pub why: String,
}
impl std::fmt::Display for ChainBreak {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
let seq = self.seq.map(|s| format!(" (seq {s})")).unwrap_or_default();
f.write_str(
&holger_errcode::AUDIT_CHAIN_BROKEN
.tag(format!("line {}{seq}: {}", self.line, self.why)),
)
}
}
/// What [`verify_file`] found.
#[derive(Debug, Clone, Default, PartialEq, Eq)]
pub struct VerifyReport {
/// Lines that verified before the first break (all of them when unbroken).
pub records: u64,
/// Checkpoint lines among them.
pub checkpoints: u64,
/// Checkpoints whose signature verified against the given roots.
pub signatures_verified: u64,
/// Signed checkpoints whose signature was NOT checked (no roots given, or a
/// build without `sign`).
pub signatures_unchecked: u64,
/// Checkpoints written without a signature.
pub unsigned_checkpoints: u64,
/// The hash of the last line that verified ([`GENESIS`] for an empty log).
pub head: String,
/// The first break, if any.
pub first_break: Option<ChainBreak>,
}
impl VerifyReport {
/// Whether the whole file verified.
pub fn is_intact(&self) -> bool {
self.first_break.is_none()
}
/// One line for an operator.
pub fn summary(&self) -> String {
match &self.first_break {
None => format!(
"audit chain INTACT: {} line(s), {} checkpoint(s) ({} signature(s) verified, {} not checked, {} unsigned), head {}",
self.records, self.checkpoints, self.signatures_verified, self.signatures_unchecked, self.unsigned_checkpoints, self.head
),
Some(b) => format!("audit chain BROKEN after {} good line(s): {b}", self.records),
}
}
}
/// **Verify the whole chain in `<dir>/audit.jsonl`.** `roots_der` are the trust
/// roots checkpoint signatures must chain to; empty = signatures not checked.
pub fn verify_dir(dir: &Path, roots_der: &[Vec<u8>]) -> Result<VerifyReport, AuditError> {
verify_file(&dir.join(super::AUDIT_FILE), roots_der)
}
/// **Verify the whole chain in one file**, naming the first break.
pub fn verify_file(path: &Path, roots_der: &[Vec<u8>]) -> Result<VerifyReport, AuditError> {
let mut rep = VerifyReport {
head: GENESIS.to_string(),
..Default::default()
};
let f = match File::open(path) {
Ok(f) => f,
Err(e) if e.kind() == std::io::ErrorKind::NotFound => return Ok(rep),
Err(e) => return Err(e.into()),
};
let mut want_seq: u64 = 0;
for (i, line) in BufReader::new(f).lines().enumerate() {
let n = i + 1;
let line = line?;
if line.trim().is_empty() {
continue;
}
let brk = |seq: Option<u64>, why: String| ChainBreak { line: n, seq, why };
let obj = match serde_json::from_str::<serde_json::Value>(&line) {
Ok(serde_json::Value::Object(m)) => m,
Ok(_) => {
rep.first_break = Some(brk(None, "the line is not a JSON object".into()));
return Ok(rep);
}
Err(e) => {
rep.first_break = Some(brk(None, format!("the line does not parse ({e})")));
return Ok(rep);
}
};
let seq = obj.get("seq").and_then(|s| s.as_u64());
let (Some(stored), Some(prev)) = (
obj.get("hash").and_then(|h| h.as_str()),
obj.get("prev").and_then(|h| h.as_str()),
) else {
rep.first_break = Some(brk(seq, "the line carries no chain (no `prev`/`hash`): it predates the chain or was written around it".into()));
return Ok(rep);
};
if seq != Some(want_seq) {
rep.first_break = Some(brk(
seq,
format!("seq {} where {want_seq} was expected — a record was removed or inserted before it", seq.map(|s| s.to_string()).unwrap_or_else(|| "missing".into())),
));
return Ok(rep);
}
if prev != rep.head {
rep.first_break = Some(brk(
seq,
format!("it links to {prev}, not to the line before it ({}) — the line before it was rewritten, or a record was removed", rep.head),
));
return Ok(rep);
}
let got = line_hash(&obj);
if got != stored {
rep.first_break = Some(brk(
seq,
format!("its content hashes to {got}, not to its stored hash {stored} — the record was edited"),
));
return Ok(rep);
}
if obj.get("kind").and_then(|k| k.as_str()) == Some(CHECKPOINT_KIND) {
rep.checkpoints += 1;
let head = obj.get("head").and_then(|h| h.as_str()).unwrap_or_default();
if head != prev {
rep.first_break = Some(brk(
seq,
format!(
"the checkpoint names head {head}, but the chain before it ends at {prev}"
),
));
return Ok(rep);
}
match obj.get("envelope").and_then(|e| e.as_str()) {
None => rep.unsigned_checkpoints += 1,
Some(env) => {
let ts = obj.get("ts").and_then(|t| t.as_i64()).unwrap_or(0);
let cert = obj
.get("cert")
.and_then(|c| c.as_str())
.and_then(|c| base64::engine::general_purpose::STANDARD.decode(c).ok())
.unwrap_or_default();
match check_signature(
env,
&cert,
&checkpoint_payload(want_seq, head, ts),
roots_der,
) {
SigCheck::Verified => rep.signatures_verified += 1,
SigCheck::Unchecked => rep.signatures_unchecked += 1,
SigCheck::Bad(why) => {
rep.first_break = Some(brk(
seq,
format!("the checkpoint signature does not verify: {why}"),
));
return Ok(rep);
}
}
}
}
}
rep.records += 1;
rep.head = stored.to_string();
want_seq += 1;
}
Ok(rep)
}
#[cfg_attr(not(feature = "sign"), allow(dead_code))]
enum SigCheck {
Verified,
Unchecked,
Bad(String),
}
#[cfg(feature = "sign")]
fn check_signature(
envelope: &str,
cert_der: &[u8],
want_payload: &[u8],
roots_der: &[Vec<u8>],
) -> SigCheck {
use znippy_common::sign::{dsse_verify, CertStore, DsseEnvelope};
let env = match DsseEnvelope::from_json(envelope) {
Ok(e) => e,
Err(e) => return SigCheck::Bad(format!("the envelope does not parse: {e:#}")),
};
if env.payload_type != CHECKPOINT_PAYLOAD_TYPE || env.payload != want_payload {
return SigCheck::Bad(
"the envelope signs a different checkpoint than the line states".into(),
);
}
if roots_der.is_empty() {
return SigCheck::Unchecked;
}
let store = match CertStore::from_der_certs(roots_der) {
Ok(s) => s,
Err(e) => return SigCheck::Bad(format!("the trust roots do not load: {e:#}")),
};
match dsse_verify(&env, &[cert_der.to_vec()], &store) {
Ok(v) if !v.is_empty() => SigCheck::Verified,
Ok(_) => SigCheck::Bad("no signer verified".into()),
Err(e) => SigCheck::Bad(format!("{e:#}")),
}
}
#[cfg(not(feature = "sign"))]
fn check_signature(
_envelope: &str,
_cert_der: &[u8],
_want_payload: &[u8],
_roots_der: &[Vec<u8>],
) -> SigCheck {
SigCheck::Unchecked
}
#[cfg(test)]
mod tests {
use super::super::{AuditAction, AuditLog, FileAuditLog, AUDIT_FILE};
use super::*;
fn tmp(tag: &str) -> PathBuf {
std::env::temp_dir().join(format!(
"holger-chain-{tag}-{}-{}",
std::process::id(),
super::super::now_nanos()
))
}
fn ev(i: u64) -> AuditEvent {
AuditEvent::new(
format!("u{i}"),
AuditAction::Upload,
"r",
format!("a@{i}"),
"ip",
200,
i,
)
}
#[test]
fn canonical_sorts_keys_and_leaves_out_the_hash() {
let mut m = serde_json::Map::new();
m.insert("b".into(), 1.into());
m.insert("a".into(), "x\"y".into());
m.insert("hash".into(), "ignored".into());
assert_eq!(canonical(&m), r#"{"a":"x\"y","b":1}"#);
}
#[test]
fn a_fresh_log_chains_from_the_genesis_and_verifies() {
let dir = tmp("fresh");
let log = FileAuditLog::new(&dir).unwrap().with_checkpoint_every(3);
for i in 0..7 {
log.record(ev(i)).unwrap();
}
let rep = verify_dir(&dir, &[]).unwrap();
assert!(rep.is_intact(), "{}", rep.summary());
assert_eq!(rep.records, 9, "7 events and 2 checkpoints");
assert_eq!(rep.checkpoints, 2);
assert_eq!(rep.unsigned_checkpoints, 2);
assert_eq!(
FileAuditLog::read_dir(&dir).unwrap().len(),
7,
"checkpoints are not events"
);
let _ = std::fs::remove_dir_all(&dir);
}
#[test]
fn a_reopened_log_continues_the_same_chain() {
let dir = tmp("reopen");
FileAuditLog::new(&dir).unwrap().record(ev(0)).unwrap();
FileAuditLog::new(&dir).unwrap().record(ev(1)).unwrap();
// Two writers on one file, interleaved, still make one chain.
let a = FileAuditLog::new(&dir).unwrap();
let b = FileAuditLog::new(&dir).unwrap();
a.record(ev(2)).unwrap();
b.record(ev(3)).unwrap();
a.record(ev(4)).unwrap();
let rep = verify_dir(&dir, &[]).unwrap();
assert!(rep.is_intact(), "{}", rep.summary());
assert_eq!(rep.records, 5);
let _ = std::fs::remove_dir_all(&dir);
}
#[test]
fn a_removed_record_breaks_seq_at_the_next_line() {
let dir = tmp("removed");
let log = FileAuditLog::new(&dir).unwrap().with_checkpoint_every(0);
for i in 0..4 {
log.record(ev(i)).unwrap();
}
let path = dir.join(AUDIT_FILE);
let text = std::fs::read_to_string(&path).unwrap();
let kept: Vec<&str> = text
.lines()
.enumerate()
.filter(|(i, _)| *i != 1)
.map(|(_, l)| l)
.collect();
std::fs::write(&path, kept.join("\n") + "\n").unwrap();
let rep = verify_dir(&dir, &[]).unwrap();
let b = rep.first_break.expect("broken");
assert_eq!(b.line, 2);
assert_eq!(b.seq, Some(2));
assert!(b.to_string().starts_with("[holger-audit-4]"), "{b}");
let _ = std::fs::remove_dir_all(&dir);
}
}
#[cfg(test)]
mod durability_tests {
use super::super::{AuditAction, AuditLog, FileAuditLog};
use super::*;
use std::sync::Arc;
fn tmp(tag: &str) -> PathBuf {
std::env::temp_dir().join(format!(
"holger-durable-{tag}-{}-{}",
std::process::id(),
super::super::now_nanos()
))
}
fn ev(w: usize, i: u64) -> AuditEvent {
AuditEvent::new(
format!("w{w}"),
AuditAction::Upload,
"r",
format!("a@{i}"),
"ip",
200,
i,
)
}
/// ★ Two writers on one file (two sinks = two open files, as two processes
/// have), writing at once, extend ONE chain: the file lock orders them.
#[test]
fn two_writers_at_once_never_fork_the_chain() {
let dir = tmp("twowriters");
let a = Arc::new(FileAuditLog::new(&dir).unwrap().with_checkpoint_every(16));
let b = Arc::new(FileAuditLog::new(&dir).unwrap().with_checkpoint_every(16));
std::thread::scope(|s| {
for (w, log) in [(0usize, &a), (1, &b), (2, &a), (3, &b)] {
let log = Arc::clone(log);
s.spawn(move || {
for i in 0..150 {
log.record(ev(w, i)).unwrap();
}
});
}
});
let rep = verify_dir(&dir, &[]).unwrap();
assert!(rep.is_intact(), "{}", rep.summary());
assert_eq!(FileAuditLog::read_dir(&dir).unwrap().len(), 600);
let _ = std::fs::remove_dir_all(&dir);
}
/// Every row a caller got back is fsynced.
#[test]
fn a_returned_row_is_synced() {
let dir = tmp("synced");
let log = FileAuditLog::new(&dir).unwrap().with_checkpoint_every(0);
for i in 0..3 {
log.record(ev(0, i)).unwrap();
assert_eq!(log.durable_next_seq(), Some(i + 1));
}
let _ = std::fs::remove_dir_all(&dir);
}
struct Broken;
impl CheckpointSigner for Broken {
fn sign_checkpoint(&self, _: &str, _: &[u8]) -> Result<SignedCheckpoint, String> {
Err("the key is gone".into())
}
}
/// ★ A checkpoint the signer cannot sign is refused, never written unsigned;
/// the records themselves are kept.
#[test]
fn an_unsignable_checkpoint_is_refused_not_written_unsigned() {
let dir = tmp("unsignable");
let log = FileAuditLog::new(&dir)
.unwrap()
.with_checkpoint_every(2)
.with_checkpoint_signer(Some(Arc::new(Broken)));
for i in 0..5 {
log.record(ev(0, i)).unwrap();
}
let e = log.checkpoint_now().unwrap_err().to_string();
assert!(e.contains("[holger-audit-11]"), "{e}");
let rep = verify_dir(&dir, &[]).unwrap();
assert!(rep.is_intact(), "{}", rep.summary());
assert_eq!(
(rep.records, rep.checkpoints, rep.unsigned_checkpoints),
(5, 0, 0),
"{}",
rep.summary()
);
let _ = std::fs::remove_dir_all(&dir);
}
}