server/lib/src/audit/mod.rs
//! Append-only audit log.
//!
//! Records **who** did **what** to **which artifact** from **where**. The
//! [`AuditLog`] trait is the swap point: the server holds one
//! `Arc<dyn AuditLog>` and the default wiring installs [`FileAuditLog`], which
//! appends one JSON line per event to `<dir>/audit.jsonl` (opened `O_APPEND`,
//! never rewritten). [`MemoryAuditLog`] is a test buffer and [`NoopAuditLog`]
//! records nothing.
//!
//! Two modes, chosen in RON under `audit: ( mode: ... )`:
//!
//! * [`AuditMode::Everything`] — one row per call, written before `record`
//! returns. The default.
//! * [`AuditMode::Off`] — nothing is recorded, so there is no evidence at all.
//! Readers refuse (`holger-audit-8`) rather than answer "none".
//!
//! ## Tamper evidence (lane S8)
//!
//! Every line [`FileAuditLog`] writes is a link of a **hash chain**
//! ([`chain`]): it carries its sequence number `seq`, the `hash` of the line
//! before it as `prev`, and its own `hash` (SHA-256 over the line's canonical
//! JSON without `hash`). Every `checkpoint_every` records a **checkpoint** line
//! names the chain head so far, signed into a DSSE envelope by the server's
//! attestation signer when one is configured. [`chain::verify_file`] walks the
//! whole file and names the FIRST line that breaks (`holger-audit-4`): an edited
//! record no longer hashes to its `hash`, a removed or inserted one breaks `seq`
//! and `prev`. Anyone who can write the directory can still rewrite the WHOLE
//! file consistently; the signed checkpoints, and the SIEM export
//! ([`siem`]) that ships every line to a store the server cannot reach back
//! into, are what catch that.
//!
//! ## The per-artifact trail
//!
//! An event that is part of an artifact's life carries the artifact's `sha256`
//! and a [`trail::TrailKind`] (imported, hash verified, scanned, policy passed or
//! denied, approved, sealed, exported, imported on the airgap side).
//! [`trail::artifact_trail`] reads them back for one digest, in log order. There
//! is one log: the trail is a view of it, not a second file.
//!
//! ## The Arrow index, the queries and the counters (lane HOL-AUDIT)
//!
//! Beside the chain, `<dir>/index/` holds an Arrow IPC copy of the records
//! ([`arrow_ipc`]): derived, caught up from the chain on every read and rebuilt
//! from it when it is missing or no longer the chain's prefix. [`query`] filters
//! and pages it (actor, artifact, action, time range, source address),
//! [`aggregate`] counts it (per repository, action, day, actor, client) and
//! [`export`] hands it out as an Arrow IPC file. The flow picture's fetches,
//! bytes sent and pullers are aggregates of it (`crate::flow`).
//!
//! ## Durability
//!
//! * Every batch of rows is `fsync`ed before any caller of the batch returns:
//! concurrent callers share one append and one `fsync` (flat combining), so the
//! window in which a returned row can be lost is zero.
//! * Every write takes an exclusive `flock` on `audit.jsonl` around the tail
//! check, the append and the `fsync`, so two processes (the server and a
//! `bundle import`) extend one chain and never fork it.
//! * A checkpoint that cannot be signed by the configured signer is REFUSED
//! (`holger-audit-11`) and retried after the next record; it is never written
//! unsigned.
//! * At startup the server walks the whole chain (`Holger::audit_log`); a break
//! refuses the boot in the critical profile and is recorded (and logged) in the
//! standard one, and an audit directory that cannot be opened refuses the boot
//! in every profile (`holger-audit-12`) instead of running a no-op logger.
use crate::lock::LockOrRecover as _;
use std::fmt;
use std::fs::File;
use std::io::{BufRead, BufReader};
use std::path::{Path, PathBuf};
use std::sync::{Arc, Mutex};
use std::time::{SystemTime, UNIX_EPOCH};
pub mod aggregate;
pub mod arrow_ipc;
pub mod chain;
pub mod export;
pub mod query;
pub mod siem;
pub mod trail;
pub use trail::TrailKind;
/// The file [`FileAuditLog`] appends to, under its directory.
pub const AUDIT_FILE: &str = "audit.jsonl";
/// How much the audit log records — the operator's choice, made in RON under
/// `audit: ( mode: ... )`.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Default, serde::Serialize, serde::Deserialize)]
pub enum AuditMode {
/// Record nothing. Every surface that reads the log refuses rather than
/// answering "none".
Off,
/// One durable row per call — the complete record, and the default.
#[default]
Everything,
}
impl AuditMode {
/// Stable lowercase wire string (RON value, proto field, log line).
pub fn as_str(&self) -> &'static str {
match self {
AuditMode::Off => "off",
AuditMode::Everything => "everything",
}
}
/// Parse a wire string. An unknown value is REFUSED: a mode comes from the
/// operator's config file, and running a different amount of auditing than
/// the file asks for is the failure this enum exists to prevent.
pub fn from_wire(s: &str) -> Option<Self> {
match s {
"off" => Some(AuditMode::Off),
"everything" => Some(AuditMode::Everything),
_ => None,
}
}
/// One sentence an operator can be shown next to a result.
pub fn caveat(&self) -> &'static str {
match self {
AuditMode::Off => {
"audit is OFF: this server records nothing, so an empty answer means \
NOT RECORDED, never 'it did not happen'"
}
AuditMode::Everything => "audit records every call",
}
}
}
impl fmt::Display for AuditMode {
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
f.write_str(self.as_str())
}
}
/// The operation an audit record describes.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum AuditAction {
/// Artifact downloaded / fetched.
Download,
/// Artifact uploaded / stored.
Upload,
/// Artifact deleted.
Delete,
/// Repository / artifact listing.
List,
/// DEV → PROD promotion.
Promote,
/// A signed entitlement fact was written (or refused) for a repository.
Entitlement,
/// A repository was PURGED — every artifact deleted by one admin call.
Purge,
/// An artifact was scanned and its evidence recorded (`crate::scanner`).
Scan,
/// A repository token was minted, revoked or refused at mint (`crate::repo_token`).
Token,
/// Anything not covered above.
Other,
}
impl AuditAction {
/// Stable lowercase wire string (what gets written to the log).
pub fn as_str(&self) -> &'static str {
match self {
AuditAction::Download => "download",
AuditAction::Upload => "upload",
AuditAction::Delete => "delete",
AuditAction::List => "list",
AuditAction::Promote => "promote",
AuditAction::Entitlement => "entitlement",
AuditAction::Purge => "purge",
AuditAction::Scan => "scan",
AuditAction::Token => "token",
AuditAction::Other => "other",
}
}
/// Inverse of [`AuditAction::as_str`]; unknown strings map to `Other` so a
/// reader never fails on a value written by a newer server.
pub fn from_wire(s: &str) -> Self {
match s {
"download" => AuditAction::Download,
"upload" => AuditAction::Upload,
"delete" => AuditAction::Delete,
"list" => AuditAction::List,
"promote" => AuditAction::Promote,
"entitlement" => AuditAction::Entitlement,
"purge" => AuditAction::Purge,
"scan" => AuditAction::Scan,
"token" => AuditAction::Token,
_ => AuditAction::Other,
}
}
}
/// One immutable audit record: time, identity, action, the artifact + repo it
/// touched, the client source address, the result status and the byte count.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct AuditEvent {
/// Event time, nanoseconds since the Unix epoch (UTC).
pub ts_nanos: i64,
/// Authenticated principal; `"anonymous"` for open reads.
pub ident: String,
/// The operation performed.
pub action: AuditAction,
/// Repository name the request targeted.
pub repo: String,
/// Artifact identifier; empty when the request had no specific artifact.
pub artifact: String,
/// Client source address (`ip:port`); empty if the transport exposed none.
pub source_ip: String,
/// Result status — an HTTP-like code.
pub status: u16,
/// Bytes transferred (0 when not applicable).
pub bytes: u64,
/// Free-form detail — an error message, content type, etc.
pub detail: String,
/// The lowercase-hex SHA-256 of the artifact's bytes when the event is
/// about one artifact's content; empty otherwise. The trail key.
pub sha256: String,
/// The step of the artifact's trail this event records, if it is one.
pub trail: Option<TrailKind>,
}
impl AuditEvent {
/// Build an event stamped with the current wall-clock time and no detail.
#[allow(clippy::too_many_arguments)]
pub fn new(
ident: impl Into<String>,
action: AuditAction,
repo: impl Into<String>,
artifact: impl Into<String>,
source_ip: impl Into<String>,
status: u16,
bytes: u64,
) -> Self {
Self {
ts_nanos: now_nanos(),
ident: ident.into(),
action,
repo: repo.into(),
artifact: artifact.into(),
source_ip: source_ip.into(),
status,
bytes,
detail: String::new(),
sha256: String::new(),
trail: None,
}
}
/// Mark the event as one step of the trail of the artifact whose bytes hash
/// to `sha256` (builder style).
pub fn on_trail(mut self, kind: TrailKind, sha256: impl Into<String>) -> Self {
self.trail = Some(kind);
self.sha256 = sha256.into();
self
}
/// Attach a detail string (builder style).
pub fn with_detail(mut self, detail: impl Into<String>) -> Self {
self.detail = detail.into();
self
}
/// The event's fields as a flat JSON object — the body of one log line
/// before the chain fields are added, and what every SIEM format renders.
pub fn to_json_map(&self) -> serde_json::Map<String, serde_json::Value> {
let mut m = serde_json::Map::new();
m.insert("ts".into(), self.ts_nanos.into());
m.insert("ident".into(), self.ident.clone().into());
m.insert("action".into(), self.action.as_str().into());
m.insert("repo".into(), self.repo.clone().into());
m.insert("artifact".into(), self.artifact.clone().into());
m.insert("ip".into(), self.source_ip.clone().into());
m.insert("status".into(), self.status.into());
m.insert("bytes".into(), self.bytes.into());
m.insert("detail".into(), self.detail.clone().into());
if !self.sha256.is_empty() {
m.insert("sha256".into(), self.sha256.clone().into());
}
if let Some(t) = self.trail {
m.insert("trail".into(), t.as_str().into());
}
m
}
/// Read an event back from a parsed log line.
pub fn from_json_value(v: &serde_json::Value) -> Self {
let s = |k: &str| {
v.get(k)
.and_then(|x| x.as_str())
.unwrap_or_default()
.to_string()
};
let n = |k: &str| v.get(k).and_then(|x| x.as_u64()).unwrap_or(0);
Self {
ts_nanos: v.get("ts").and_then(|x| x.as_i64()).unwrap_or(0),
ident: s("ident"),
action: AuditAction::from_wire(&s("action")),
repo: s("repo"),
artifact: s("artifact"),
source_ip: s("ip"),
status: n("status") as u16,
bytes: n("bytes"),
detail: s("detail"),
sha256: s("sha256"),
trail: v
.get("trail")
.and_then(|x| x.as_str())
.and_then(TrailKind::from_wire),
}
}
}
/// **The client of a recorded `source_ip`, without its port**: `10.0.0.1:5` and
/// `[::1]:5` become `10.0.0.1` and `::1`; an address without a port is itself.
pub fn client_host(source_ip: &str) -> String {
let s = source_ip.trim();
if let Some(rest) = s.strip_prefix('[') {
return rest.split(']').next().unwrap_or(rest).to_string();
}
match s.rsplit_once(':') {
Some((h, p)) if !h.contains(':') && p.chars().all(|c| c.is_ascii_digit()) => h.to_string(),
_ => s.to_string(),
}
}
/// Current wall-clock time in nanoseconds since the Unix epoch.
pub fn now_nanos() -> i64 {
SystemTime::now()
.duration_since(UNIX_EPOCH)
.map(|d| d.as_nanos() as i64)
.unwrap_or(0)
}
/// Where a sink's records can be read back from.
#[derive(Debug, Clone)]
pub enum QuerySource {
/// An owned snapshot (the in-memory sink).
Snapshot(Vec<AuditEvent>),
/// A [`FileAuditLog`] directory; read it with [`FileAuditLog::read_dir`].
Directory(PathBuf),
}
impl QuerySource {
/// Every event the source holds, in write order.
pub fn events(&self) -> Result<Vec<AuditEvent>, AuditError> {
match self {
QuerySource::Snapshot(e) => Ok(e.clone()),
QuerySource::Directory(d) => FileAuditLog::read_dir(d),
}
}
/// **Every record as Arrow batches** — the index of a directory, caught up
/// with its chain first ([`arrow_ipc::AuditIndex::table`]); a snapshot's
/// records, numbered in order.
pub fn table(&self) -> Result<arrow_ipc::AuditTable, AuditError> {
match self {
QuerySource::Snapshot(_) => arrow_ipc::AuditTable::of_records(&self.records()?),
QuerySource::Directory(d) => arrow_ipc::AuditIndex::open(d).table(),
}
}
/// Every record the source holds with its place in the chain. A snapshot
/// keeps no chain: its records are numbered in order and carry no hash.
pub fn records(&self) -> Result<Vec<StoredRecord>, AuditError> {
match self {
QuerySource::Snapshot(e) => Ok(e
.iter()
.enumerate()
.map(|(i, ev)| StoredRecord {
seq: i as u64,
hash: String::new(),
event: ev.clone(),
})
.collect()),
QuerySource::Directory(d) => FileAuditLog::read_records(d),
}
}
}
/// Pluggable audit sink. Implementations MUST be append-only and thread-safe.
pub trait AuditLog: Send + Sync {
/// Append one event. Durable (written to the file) by the time it returns.
fn record(&self, event: AuditEvent) -> Result<(), AuditError>;
/// Flush any buffered state to durable storage. Default: no-op.
fn flush(&self) -> Result<(), AuditError> {
Ok(())
}
/// **Give up every file this sink holds open** — the data set is about to
/// be unmounted for a growth (`crate::grow`). Until [`reacquire_files`]
/// (Self::reacquire_files) a `record` is REFUSED, never buffered.
fn release_files(&self) -> Result<(), AuditError> {
Ok(())
}
/// Take the files back after the growth.
fn reacquire_files(&self) -> Result<(), AuditError> {
Ok(())
}
/// Where this sink's records can be read back from, if anywhere. `None` makes
/// a reader REFUSE (`holger-audit-8`) instead of answering an empty page.
fn query_source(&self) -> Option<QuerySource> {
None
}
/// How much this sink records — see [`AuditMode`].
fn mode(&self) -> AuditMode {
AuditMode::Everything
}
/// The signal a SIEM exporter follows this sink by, if it keeps a file.
fn tail_signal(&self) -> Option<Arc<chain::TailSignal>> {
None
}
}
/// The default audit backend ([`FileAuditLog`]) under `dir`.
pub fn default_audit_log(dir: impl AsRef<Path>) -> Result<Arc<dyn AuditLog>, AuditError> {
Ok(Arc::new(FileAuditLog::new(dir)?))
}
/// The sink an [`AuditMode`] asks for. `Off` does not create `dir`: "off" must
/// not leave a directory behind that looks like a log.
pub fn audit_log_for_mode(
mode: AuditMode,
dir: impl AsRef<Path>,
) -> Result<Arc<dyn AuditLog>, AuditError> {
match mode {
AuditMode::Off => Ok(Arc::new(NoopAuditLog)),
AuditMode::Everything => default_audit_log(dir),
}
}
/// A row handed in, with the slot its caller waits on.
type PendingRow = (serde_json::Map<String, serde_json::Value>, Arc<RowSlot>);
/// **The one durable sink**: hash-chained JSON lines appended to
/// `<dir>/audit.jsonl` (see [`chain`] for the line format).
pub struct FileAuditLog {
dir: PathBuf,
file: Mutex<Option<chain::ChainWriter>>,
/// Rows handed in but not yet written, each with the slot its caller waits
/// on. Whoever holds `file` next writes ALL of them in one append (flat
/// combining, stinky 30): concurrent callers share one write instead of
/// queueing a write each behind one process-wide mutex.
pending: Mutex<Vec<PendingRow>>,
checkpoint_every: u64,
signer: Option<Arc<dyn chain::CheckpointSigner>>,
tail: Arc<chain::TailSignal>,
}
impl fmt::Debug for FileAuditLog {
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
f.debug_struct("FileAuditLog")
.field("dir", &self.dir)
.field("checkpoint_every", &self.checkpoint_every)
.field("signed_checkpoints", &self.signer.is_some())
.finish()
}
}
impl FileAuditLog {
/// Open (creating) `<dir>/audit.jsonl` for append, picking the chain up
/// from its last line.
pub fn new(dir: impl AsRef<Path>) -> Result<Self, AuditError> {
let dir = dir.as_ref().to_path_buf();
std::fs::create_dir_all(&dir)?;
let w = chain::ChainWriter::open(&dir.join(AUDIT_FILE))?;
Ok(Self {
dir,
file: Mutex::new(Some(w)),
pending: Mutex::new(Vec::new()),
checkpoint_every: chain::DEFAULT_CHECKPOINT_EVERY,
signer: None,
tail: Arc::new(chain::TailSignal::default()),
})
}
/// Write a checkpoint line after every `n` records (`0` = never).
pub fn with_checkpoint_every(mut self, n: u64) -> Self {
self.checkpoint_every = n;
self
}
/// Sign every checkpoint with `signer` (the server's attestation signer).
pub fn with_checkpoint_signer(
mut self,
signer: Option<Arc<dyn chain::CheckpointSigner>>,
) -> Self {
self.signer = signer;
self
}
/// The directory this sink writes under.
pub fn dir(&self) -> &Path {
&self.dir
}
/// The `seq` the next line gets, counting only lines this sink has `fsync`ed:
/// every line below it is on disk. `None` while the file is released.
pub fn durable_next_seq(&self) -> Option<u64> {
self.file
.lock_or_recover()
.as_ref()
.map(|w| w.synced_next_seq())
}
/// Write a checkpoint line now, whatever the counter says.
pub fn checkpoint_now(&self) -> Result<(), AuditError> {
let mut g = self.file.lock_or_recover();
let Some(w) = g.as_mut() else {
return Err(AuditError::Other(holger_errcode::GROW_STORES_CLOSED.tag(
"the audit file is closed while the data set is unmounted for a growth",
)));
};
w.write_checkpoint(self.signer.as_deref())?;
drop(g);
self.tail.bump();
Ok(())
}
/// Every event in `dir`, in write order. A missing file is an empty log.
/// Checkpoint lines are not events and are skipped.
pub fn read_dir(dir: impl AsRef<Path>) -> Result<Vec<AuditEvent>, AuditError> {
Ok(Self::read_records(dir)?
.into_iter()
.map(|r| r.event)
.collect())
}
/// Every record in `dir` with its place in the chain, in write order.
pub fn read_records(dir: impl AsRef<Path>) -> Result<Vec<StoredRecord>, AuditError> {
let path = dir.as_ref().join(AUDIT_FILE);
let f = match File::open(&path) {
Ok(f) => f,
Err(e) if e.kind() == std::io::ErrorKind::NotFound => return Ok(Vec::new()),
Err(e) => return Err(e.into()),
};
let mut out = Vec::new();
for line in BufReader::new(f).lines() {
let line = line?;
if line.trim().is_empty() {
continue;
}
let v: serde_json::Value = serde_json::from_str(&line)
.map_err(|e| AuditError::Other(format!("bad audit line: {e}")))?;
if chain::is_checkpoint(&v) {
continue;
}
out.push(StoredRecord {
seq: v
.get("seq")
.and_then(|x| x.as_u64())
.unwrap_or(out.len() as u64),
hash: v
.get("hash")
.and_then(|x| x.as_str())
.unwrap_or_default()
.to_string(),
event: AuditEvent::from_json_value(&v),
});
}
Ok(out)
}
}
/// One record as the log holds it: its place in the chain and the event.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct StoredRecord {
/// The record's sequence number in the chain (checkpoints count too).
pub seq: u64,
/// The record's chain hash; empty for a sink that keeps no chain.
pub hash: String,
/// The event.
pub event: AuditEvent,
}
/// Where a combined writer leaves one caller's answer.
#[derive(Default)]
struct RowSlot(Mutex<Option<Result<(), String>>>);
impl RowSlot {
fn take(&self) -> Option<Result<(), AuditError>> {
self.0
.lock_or_recover()
.take()
.map(|r| r.map_err(AuditError::Other))
}
}
impl AuditLog for FileAuditLog {
/// The row is built OUTSIDE every lock, queued, and written by whoever holds
/// the file next — this caller, or one ahead of it that took the whole queue.
/// It is durable (appended AND `fsync`ed, under the file's `flock`) by the time
/// this returns: a caller only returns once a writer has set its slot, and a
/// writer sets the slots after the batch's `fsync`.
fn record(&self, event: AuditEvent) -> Result<(), AuditError> {
let slot = Arc::new(RowSlot::default());
let obj = event.to_json_map();
self.pending
.lock_or_recover()
.push((obj, Arc::clone(&slot)));
let mut g = self.file.lock_or_recover();
if let Some(done) = slot.take() {
// A writer ahead of this caller wrote the row (and bumped the tail).
return done;
}
let batch = std::mem::take(&mut *self.pending.lock_or_recover());
let (objs, slots): (Vec<_>, Vec<_>) = batch.into_iter().unzip();
let outcome: Result<(), String> = match g.as_mut() {
None => Err(holger_errcode::GROW_STORES_CLOSED.tag(
"the audit file is closed while the data set is unmounted for a growth — this row is refused, not buffered",
)),
Some(w) => match w.append_maps(objs) {
Ok(()) => {
if self.checkpoint_every > 0 && w.since_checkpoint() >= self.checkpoint_every {
// The records above are durable; a checkpoint that cannot be
// written (or signed: it is refused, never written unsigned)
// is logged and retried after the next record, never a
// refusal of them.
if let Err(e) = w.write_checkpoint(self.signer.as_deref()) {
log::error!("audit: the checkpoint line was not written ({e}); it is retried after the next record");
}
}
Ok(())
}
Err(e) => Err(e.to_string()),
},
};
for s in &slots {
*s.0.lock_or_recover() = Some(outcome.clone());
}
drop(g);
if outcome.is_ok() {
self.tail.bump();
}
slot.take().unwrap_or(Ok(()))
}
fn flush(&self) -> Result<(), AuditError> {
let mut g = self.file.lock_or_recover();
if let Some(w) = g.as_mut() {
w.sync()?;
}
Ok(())
}
fn release_files(&self) -> Result<(), AuditError> {
let mut g = self.file.lock_or_recover();
if let Some(mut w) = g.take() {
w.sync()?;
}
Ok(())
}
fn reacquire_files(&self) -> Result<(), AuditError> {
let mut g = self.file.lock_or_recover();
if g.is_none() {
*g = Some(chain::ChainWriter::open(&self.dir.join(AUDIT_FILE))?);
}
Ok(())
}
fn query_source(&self) -> Option<QuerySource> {
Some(QuerySource::Directory(self.dir.clone()))
}
fn tail_signal(&self) -> Option<Arc<chain::TailSignal>> {
Some(Arc::clone(&self.tail))
}
}
/// A disabled audit log — records nothing.
#[derive(Debug, Default, Clone, Copy)]
pub struct NoopAuditLog;
impl AuditLog for NoopAuditLog {
fn record(&self, _event: AuditEvent) -> Result<(), AuditError> {
Ok(())
}
fn mode(&self) -> AuditMode {
AuditMode::Off
}
}
/// In-memory audit log — keeps every event in a `Vec`. For tests.
#[derive(Debug, Default)]
pub struct MemoryAuditLog {
events: Mutex<Vec<AuditEvent>>,
}
impl MemoryAuditLog {
/// Empty log.
pub fn new() -> Self {
Self::default()
}
/// Snapshot of every recorded event, in insertion order.
pub fn events(&self) -> Vec<AuditEvent> {
self.events.lock_or_recover().clone()
}
}
impl AuditLog for MemoryAuditLog {
fn record(&self, event: AuditEvent) -> Result<(), AuditError> {
self.events.lock_or_recover().push(event);
Ok(())
}
fn query_source(&self) -> Option<QuerySource> {
Some(QuerySource::Snapshot(self.events()))
}
}
/// Errors from an audit backend.
#[derive(Debug, thiserror::Error)]
pub enum AuditError {
/// Filesystem / IO failure.
#[error("audit io error: {0}")]
Io(#[source] std::io::Error),
/// Anything else (bad line, lock poisoned, released file, …).
#[error("audit error: {0}")]
Other(String),
/// The hash chain is broken at a named line (`holger-audit-4`).
#[error("audit chain broken: {0}")]
ChainBroken(chain::ChainBreak),
/// The Arrow index could not be written or read (`holger-audit-2`).
#[error("audit index (Arrow IPC) failed: {0}")]
Arrow(String),
}
impl AuditError {
/// The FROZEN `holger-audit-<n>` code.
pub fn code(&self) -> &'static holger_errcode::ErrCode {
match self {
AuditError::Io(_) => &holger_errcode::AUDIT_IO,
AuditError::Other(_) => &holger_errcode::AUDIT_OTHER,
AuditError::ChainBroken(_) => &holger_errcode::AUDIT_CHAIN_BROKEN,
AuditError::Arrow(_) => &holger_errcode::AUDIT_ARROW,
}
}
/// BOTH channels in one string.
pub fn tagged(&self) -> String {
self.code().tag(self)
}
}
impl From<std::io::Error> for AuditError {
fn from(e: std::io::Error) -> Self {
AuditError::Io(e)
}
}
#[cfg(test)]
mod tests {
use super::*;
fn tmp(tag: &str) -> PathBuf {
std::env::temp_dir().join(format!("holger-audit-{tag}-{}", now_nanos()))
}
#[test]
fn action_wire_roundtrips() {
for a in [
AuditAction::Download,
AuditAction::Upload,
AuditAction::Delete,
AuditAction::List,
AuditAction::Promote,
AuditAction::Entitlement,
AuditAction::Purge,
AuditAction::Other,
] {
assert_eq!(AuditAction::from_wire(a.as_str()), a);
}
assert_eq!(AuditAction::from_wire("teleport"), AuditAction::Other);
}
#[test]
fn mode_wire_strings_roundtrip_and_unknown_is_refused() {
for m in [AuditMode::Off, AuditMode::Everything] {
assert_eq!(AuditMode::from_wire(m.as_str()), Some(m));
}
assert_eq!(AuditMode::from_wire("aggregated"), None);
assert_eq!(AuditMode::from_wire("Off"), None);
assert_eq!(AuditMode::default(), AuditMode::Everything);
}
#[test]
fn off_builds_an_inert_sink_and_never_touches_the_directory() {
let dir = tmp("off");
let log = audit_log_for_mode(AuditMode::Off, &dir).unwrap();
log.record(AuditEvent::new(
"a",
AuditAction::Download,
"r",
"x@1",
"ip:1",
200,
7,
))
.unwrap();
assert_eq!(log.mode(), AuditMode::Off);
assert!(log.query_source().is_none());
assert!(!dir.exists());
}
#[test]
fn the_file_sink_appends_and_reads_back_in_order_across_reopens() {
let dir = tmp("file");
{
let log = FileAuditLog::new(&dir).unwrap();
log.record(
AuditEvent::new(
"alice",
AuditAction::Upload,
"crates",
"serde@1",
"10.0.0.1:5",
200,
42,
)
.with_detail("a \"quoted\"\nline"),
)
.unwrap();
}
let log = FileAuditLog::new(&dir).unwrap();
log.record(AuditEvent::new(
"anonymous",
AuditAction::Download,
"crates",
"serde@1",
"10.0.0.2:6",
200,
42,
))
.unwrap();
let got = FileAuditLog::read_dir(&dir).unwrap();
assert_eq!(got.len(), 2);
assert_eq!(got[0].ident, "alice");
assert_eq!(got[0].detail, "a \"quoted\"\nline");
assert_eq!(got[1].action, AuditAction::Download);
assert!(matches!(
log.query_source(),
Some(QuerySource::Directory(_))
));
let _ = std::fs::remove_dir_all(&dir);
}
#[test]
fn a_released_file_sink_refuses_rows_until_reacquired() {
let dir = tmp("release");
let log = FileAuditLog::new(&dir).unwrap();
log.release_files().unwrap();
assert!(log
.record(AuditEvent::new("a", AuditAction::List, "r", "", "", 200, 0))
.is_err());
log.reacquire_files().unwrap();
log.record(AuditEvent::new("a", AuditAction::List, "r", "", "", 200, 0))
.unwrap();
assert_eq!(FileAuditLog::read_dir(&dir).unwrap().len(), 1);
let _ = std::fs::remove_dir_all(&dir);
}
#[test]
fn memory_log_keeps_every_event_from_concurrent_recorders() {
let log = Arc::new(MemoryAuditLog::new());
std::thread::scope(|scope| {
for t in 0..8 {
let log = Arc::clone(&log);
scope.spawn(move || {
for i in 0..50 {
log.record(AuditEvent::new(
format!("w-{t}"),
AuditAction::Download,
"c",
format!("p@{i}"),
"ip",
200,
1,
))
.unwrap();
}
});
}
});
assert_eq!(log.events().len(), 400);
}
}