krabka_audit/
verify.rs

1//! Offline verification of an audit partition's hash-chain and signed checkpoints.
2//!
3//! The verifier reads `<dir>/*.log` segment files directly. It does NO recovery
4//! and NO truncation, so tail corruption stays visible. It recomputes the chain
5//! with the same primitives the writer used, and it validates each checkpoint
6//! signature against a trusted key. This file holds the report types and the
7//! segment walk; the per-record chain and checkpoint checks live in `walk`.
8
9use {
10    self::walk::{
11        WalkState, check_chained, check_checkpoint, check_records_lost, header,
12        records_lost_body_has_field,
13    },
14    crate::{
15        checkpoint::EVENT_CLASS_CHECKPOINT,
16        event::AuditEventClass,
17        ids::{CheckpointCount, RecordCount, Seq},
18        sink::AuditError,
19    },
20    krabka_protocol::records::RecordBatch,
21    std::{
22        collections::HashMap,
23        fs,
24        path::{Path, PathBuf},
25    },
26    tracing::field::Empty,
27};
28
29mod walk;
30
31#[cfg(test)]
32mod tests;
33
34/// Trusted public keys, keyed by `key_id`.
35#[derive(Debug, Default)]
36pub struct TrustedKeys {
37    keys: HashMap<String, Vec<u8>>,
38}
39
40impl TrustedKeys {
41    #[must_use]
42    pub fn single(key_id: String, public_key: Vec<u8>) -> Self {
43        let mut keys = HashMap::new();
44        keys.insert(key_id, public_key);
45        Self { keys }
46    }
47
48    #[must_use]
49    pub fn get(&self, key_id: &str) -> Option<&[u8]> { self.keys.get(key_id).map(Vec::as_slice) }
50}
51
52/// First detected break in the chain or signatures.
53#[derive(Debug, Clone)]
54pub struct VerifyBreak {
55    pub offset: i64,
56    pub seq: Option<Seq>,
57    pub reason: String,
58}
59
60/// A valid chain marker that declares fail-open audit loss.
61#[derive(Debug, Clone, PartialEq, Eq)]
62pub struct VerifyLoss {
63    pub offset: i64,
64    pub seq: Seq,
65    pub records: RecordCount,
66}
67
68/// Result of a partition verification.
69///
70/// `unanchored_records` is only meaningful when `ok` is `true`. It counts the
71/// records with a seq greater than the highest seq that the last valid signed
72/// checkpoint covers. When `ok` is `false`, this field is 0: the walk stopped
73/// at the break, before it could establish a reliable count.
74#[derive(Debug, Clone)]
75pub struct VerifyReport {
76    pub records: RecordCount,
77    pub checkpoints: CheckpointCount,
78    pub ok: bool,
79    pub first_break: Option<VerifyBreak>,
80    /// Valid, hash-chained declarations of fail-open record loss.
81    pub losses: Vec<VerifyLoss>,
82    /// Number of records that a signed checkpoint does NOT cover, that is, the
83    /// unsigned tail. Zero means the chain is fully attested. This field is
84    /// only meaningful when `ok` is `true`.
85    pub unanchored_records: RecordCount,
86}
87
88/// Verify the audit partition under `dir`.
89///
90/// This function reads all `*.log` segment files in base-offset order, which is
91/// the filename order. It decodes each `RecordBatch` directly and not through
92/// `Log::open`: that path runs recovery and truncation, which would silently
93/// mask tail corruption. The function then recomputes the hash-chain, and it
94/// validates every checkpoint signature against `trusted`.
95#[tracing::instrument(
96    level = "info",
97    skip_all,
98    fields(dir = %dir.display(), records = Empty, checkpoints = Empty, ok = Empty),
99    err
100)]
101/// # Errors
102/// Returns an error if the verifier cannot read `dir`. Returns an error if the
103/// verifier cannot read one of the segment files. A detected break in the chain
104/// or in a signature is not an error: it is an `Ok` report with `ok` set to
105/// `false`.
106pub fn verify_partition_dir(dir: &Path, trusted: &TrustedKeys) -> Result<VerifyReport, AuditError> {
107    let mut segments: Vec<PathBuf> = fs::read_dir(dir)
108        .map_err(|e| AuditError::Sink(format!("read dir {}: {e}", dir.display())))?
109        .filter_map(Result::ok)
110        .map(|e| e.path())
111        .filter(|p| p.extension().is_some_and(|x| x == "log"))
112        .collect();
113    segments.sort();
114
115    let mut state = WalkState::new();
116
117    for seg in segments {
118        let bytes = fs::read(&seg)
119            .map_err(|e| AuditError::Sink(format!("read segment {}: {e}", seg.display())))?;
120        let mut cur: &[u8] = &bytes;
121        while !cur.is_empty() {
122            let Ok(batch) = RecordBatch::decode(&mut cur) else {
123                break; // undecodable tail: stop this segment (visible truncation)
124            };
125            for rec in &batch.records {
126                let offset = batch.base_offset + i64::from(rec.offset_delta);
127                let class = header(rec, "event_class").unwrap_or_default();
128                let records_lost_header =
129                    class == AuditEventClass::RecordsLost.as_header().as_bytes();
130                let records_lost_body = records_lost_body_has_field(rec);
131                let result = if class == EVENT_CLASS_CHECKPOINT.as_bytes() {
132                    check_checkpoint(rec, offset, &mut state, trusted)
133                } else if records_lost_header || records_lost_body {
134                    check_records_lost(rec, offset, &mut state)
135                } else {
136                    check_chained(rec, offset, &mut state)
137                };
138                if let Err(report) = result {
139                    let span = tracing::Span::current();
140                    span.record("records", report.records.0);
141                    span.record("checkpoints", report.checkpoints.0);
142                    span.record("ok", report.ok);
143                    return Ok(report);
144                }
145            }
146        }
147    }
148
149    let unanchored_records = match state.last_checkpoint_seq_high {
150        Some(seq_high) => RecordCount(state.records.0.saturating_sub(seq_high.0 + 1)),
151        None => state.records,
152    };
153
154    let span = tracing::Span::current();
155    span.record("records", state.records.0);
156    span.record("checkpoints", state.checkpoints.0);
157    span.record("ok", true);
158    Ok(VerifyReport {
159        records: state.records,
160        checkpoints: state.checkpoints,
161        ok: true,
162        first_break: None,
163        losses: state.losses,
164        unanchored_records,
165    })
166}
krabka_audit/verify.rs

krabka_audit/
verify.rs
