krabka_audit/
sink.rs

1//! The audit sink abstraction and an in-memory test sink.
2
3use {
4    crate::{
5        chain::to_hex,
6        event::{AuditEvent, AuditEventClass},
7        ocsf::{ProductInfo, to_ocsf},
8    },
9    async_trait::async_trait,
10    std::{
11        fmt,
12        sync::{Mutex, PoisonError},
13    },
14    tracing::field::Empty,
15};
16
17/// Record-header key for the per-broker chain sequence number in decimal ASCII.
18pub const HEADER_SEQ: &str = "seq";
19/// Record-header key for the previous chain head, as lowercase hex of 32 bytes.
20pub const HEADER_PREV_HASH: &str = "prev_hash";
21
22/// A serialized, sink-ready audit record.
23#[derive(Debug, Clone, PartialEq, Eq)]
24pub struct AuditRecord {
25    pub class: AuditEventClass,
26    pub value: Vec<u8>,
27    pub headers: Vec<(String, Vec<u8>)>,
28}
29
30impl AuditRecord {
31    /// Build the chained meta-record that declares fail-open audit loss.
32    ///
33    /// `generation` is the persisted loss-state generation the marker settles;
34    /// the verifier requires it to advance strictly across a partition.
35    #[must_use]
36    pub fn records_lost(count: u64, generation: u64) -> Self {
37        Self {
38            class: AuditEventClass::RecordsLost,
39            value: format!(r#"{{"records_lost":{count},"loss_generation":{generation}}}"#)
40                .into_bytes(),
41            headers: vec![(
42                "event_class".to_owned(),
43                AuditEventClass::RecordsLost.as_header().as_bytes().to_vec(),
44            )],
45        }
46    }
47
48    /// Stamp the hash-chain headers onto this record.
49    ///
50    /// The `seq` header is decimal and the `prev_hash` header is lowercase
51    /// hex. The writer calls this method, and the writer owns the chain.
52    pub fn push_chain_headers(&mut self, seq: u64, prev_head: &[u8; 32]) {
53        self.headers
54            .push((HEADER_SEQ.to_owned(), seq.to_string().into_bytes()));
55        self.headers
56            .push((HEADER_PREV_HASH.to_owned(), to_hex(prev_head).into_bytes()));
57    }
58
59    /// Serialize an event to OCSF JSON and attach the SIEM-filterable headers.
60    #[must_use]
61    #[tracing::instrument(
62        level = "debug",
63        skip_all,
64        fields(class = ?event.class(), bytes = Empty)
65    )]
66    pub fn from_event(event: &AuditEvent, product: &ProductInfo) -> Self {
67        let class = event.class();
68        let value = serde_json::to_vec(&to_ocsf(event, product)).unwrap_or_else(|_| b"{}".to_vec());
69        tracing::Span::current().record("bytes", value.len());
70        let mut headers = vec![(
71            "event_class".to_owned(),
72            class.as_header().as_bytes().to_vec(),
73        )];
74        if let Some(PrincipalStatus { principal, status }) = principal_and_status(event) {
75            headers.push(("principal".to_owned(), principal.into_bytes()));
76            headers.push(("status".to_owned(), status.as_bytes().to_vec()));
77        }
78        Self {
79            class,
80            value,
81            headers,
82        }
83    }
84}
85
86/// The principal name and outcome status an audit record carries as headers.
87struct PrincipalStatus {
88    principal: String,
89    status: &'static str,
90}
91
92fn principal_and_status(event: &AuditEvent) -> Option<PrincipalStatus> {
93    match event {
94        AuditEvent::Authentication {
95            principal, outcome, ..
96        }
97        | AuditEvent::AdminOperation {
98            principal, outcome, ..
99        }
100        | AuditEvent::PrivilegedAction {
101            principal, outcome, ..
102        } => Some(PrincipalStatus {
103            principal: principal.name.clone(),
104            status: outcome.as_str(),
105        }),
106        AuditEvent::AuthorizationDenied { principal, .. } => Some(PrincipalStatus {
107            principal: principal.name.clone(),
108            status: "denied",
109        }),
110        AuditEvent::Lifecycle { .. } => None,
111    }
112}
113
114/// Errors that a sink can report.
115#[derive(Debug, thiserror::Error)]
116pub enum AuditError {
117    #[error("audit sink: {0}")]
118    Sink(String),
119    #[error("audit key: {0}")]
120    Key(String),
121    #[error("audit spool I/O: {0}")]
122    Io(String),
123    #[error("audit unavailable: {0}")]
124    Unavailable(String),
125    /// The sink may have appended the record but could not prove durability.
126    #[error("audit durability indeterminate: {0}")]
127    Indeterminate(String),
128    #[error("audit spool requires explicit recovery: {0}")]
129    Poisoned(String),
130    /// A file in the audit spool directory has no version header, or one this
131    /// build does not read.
132    #[error("audit spool file {file}: {}", describe_spool_version(*found))]
133    UnsupportedSpoolFormat {
134        /// The file name, such as `audit.spool`.
135        file: String,
136        /// The version the header holds, or `None` when the file has no
137        /// header at all.
138        found: Option<i16>,
139    },
140}
141
142fn describe_spool_version(found: Option<i16>) -> String {
143    found.map_or_else(
144        || {
145            "has no version header; it predates krabka 1.0, so the audit spool directory must \
146             be emptied (the node reformatted) before this broker starts"
147                .to_owned()
148        },
149        |version| format!("has version {version}, which this build does not read"),
150    )
151}
152
153/// Destination for serialized audit records.
154#[async_trait]
155pub trait AuditSink: Send + Sync + fmt::Debug {
156    /// Write one record. When `durable` is true, return only after the record
157    /// has reached stable storage.
158    async fn write(&self, record: AuditRecord, durable: bool) -> Result<(), AuditError>;
159}
160
161/// In-memory sink for tests.
162#[derive(Debug, Default)]
163pub struct MemorySink {
164    records: Mutex<Vec<AuditRecord>>,
165}
166
167impl MemorySink {
168    #[must_use]
169    pub fn records(&self) -> Vec<AuditRecord> {
170        self.records
171            .lock()
172            .unwrap_or_else(PoisonError::into_inner)
173            .clone()
174    }
175}
176
177#[async_trait]
178impl AuditSink for MemorySink {
179    async fn write(&self, record: AuditRecord, _durable: bool) -> Result<(), AuditError> {
180        self.records
181            .lock()
182            .unwrap_or_else(PoisonError::into_inner)
183            .push(record);
184        Ok(())
185    }
186}
187
188#[cfg(test)]
189mod tests {
190    use {super::*, crate::test_support::authentication, assert2::check};
191
192    #[test]
193    fn io_error_displays() {
194        let e = AuditError::Io("disk full".to_owned());
195        check!(format!("{e}").contains("disk full"));
196    }
197
198    #[test]
199    fn records_lost_is_a_distinct_meta_record() {
200        check!(
201            AuditRecord::records_lost(3, 7)
202                == AuditRecord {
203                    class: AuditEventClass::RecordsLost,
204                    value: br#"{"records_lost":3,"loss_generation":7}"#.to_vec(),
205                    headers: vec![("event_class".to_owned(), b"records_lost".to_vec())],
206                }
207        );
208    }
209
210    #[test]
211    fn unavailable_error_displays() {
212        check!(
213            AuditError::Unavailable("spool full".into()).to_string()
214                == "audit unavailable: spool full"
215        );
216    }
217
218    fn hdr(r: &AuditRecord, k: &str) -> Option<String> {
219        r.headers
220            .iter()
221            .find(|(hk, _)| hk == k)
222            .map(|(_, v)| String::from_utf8_lossy(v).into_owned())
223    }
224
225    #[test]
226    fn from_event_sets_principal_and_status_headers() {
227        use crate::{
228            event::{AuditEndpoint, AuditEvent, AuditOutcome, AuditPrincipal, LifecycleKind},
229            ocsf::ProductInfo,
230        };
231        let product = ProductInfo {
232            vendor_name: "v".into(),
233            name: "n".into(),
234            version: "0".into(),
235        };
236        let authn = authentication(1, None, 0);
237        let r = AuditRecord::from_event(&authn, &product);
238        check!(
239            (hdr(&r, "principal"), hdr(&r, "status"))
240                == (Some("alice".into()), Some("failure".into()))
241        );
242        let admin = AuditEvent::AdminOperation {
243            outcome: AuditOutcome::Success,
244            principal: AuditPrincipal {
245                name: "bob".into(),
246                auth_method: "MTls".into(),
247            },
248            source: AuditEndpoint {
249                ip: "10.0.0.2".into(),
250                port: 2,
251            },
252            operation: "CreateTopics".into(),
253            resources: vec![],
254            time_ms: 0,
255        };
256        let r = AuditRecord::from_event(&admin, &product);
257        check!(
258            (hdr(&r, "principal"), hdr(&r, "status"))
259                == (Some("bob".into()), Some("success".into()))
260        );
261        let deny = AuditEvent::AuthorizationDenied {
262            principal: AuditPrincipal {
263                name: "carol".into(),
264                auth_method: "MTls".into(),
265            },
266            source: AuditEndpoint {
267                ip: "10.0.0.3".into(),
268                port: 3,
269            },
270            resource_type: "Topic".into(),
271            resource_name: "t".into(),
272            operation: "Write".into(),
273            time_ms: 0,
274        };
275        let r = AuditRecord::from_event(&deny, &product);
276        check!(
277            (hdr(&r, "principal"), hdr(&r, "status"))
278                == (Some("carol".into()), Some("denied".into()))
279        );
280        let life = AuditEvent::Lifecycle {
281            kind: LifecycleKind::BrokerStarted,
282            node_id: 1,
283            time_ms: 0,
284        };
285        let r = AuditRecord::from_event(&life, &product);
286        check!((hdr(&r, "principal"), hdr(&r, "status")) == (None, None));
287    }
288}
krabka_audit/sink.rs

krabka_audit/
sink.rs
