1use std::sync::Arc;
37
38use cratefield_core::{Database, Statement};
39use sea_query::Value as SeaValue;
40use sha2::{Digest, Sha256};
41
42use crate::{Access, Audit, AuditEvent, SecretsError, StoreId};
43
44const GENESIS: [u8; 32] = [0; 32];
46
47pub struct ChainAudit {
49 db: Arc<dyn Database>,
50 store: StoreId,
51}
52
53impl ChainAudit {
54 #[must_use]
55 pub fn new(store: StoreId, db: Arc<dyn Database>) -> Self {
56 Self { db, store }
57 }
58
59 async fn tail(&self) -> Result<(i64, Vec<u8>), SecretsError> {
66 let rows = self
67 .db
68 .query(&Statement::new(
69 "SELECT seq FROM harness_secret_audit ORDER BY seq DESC LIMIT 1",
70 ))
71 .await?;
72 let next_seq = rows
73 .first()
74 .map(|row| {
75 row.get("seq")
76 .ok_or_else(|| SecretsError::NotAudited("an audit row has no seq".to_owned()))
77 })
78 .transpose()?
79 .map_or(1, |seq: i64| seq.saturating_add(1));
80
81 let rows = self
82 .db
83 .query(&Statement::with_values(
84 "SELECT hash FROM harness_secret_audit WHERE store = ? OR store = '' \
85 ORDER BY seq DESC LIMIT 1",
86 vec![text(self.store.as_str())],
87 ))
88 .await?;
89 let prev_hash = rows
90 .first()
91 .map(|row| {
92 row.get("hash")
93 .ok_or_else(|| SecretsError::NotAudited("an audit row has no hash".to_owned()))
94 })
95 .transpose()?
96 .unwrap_or_else(|| GENESIS.to_vec());
97 Ok((next_seq, prev_hash))
98 }
99}
100
101#[async_trait::async_trait]
102impl Audit for ChainAudit {
103 async fn record(&self, event: &AuditEvent<'_>) -> Result<(), SecretsError> {
104 if event.store.as_str().is_empty() {
109 return Err(SecretsError::Invalid(
110 "an audit event must name its store".to_owned(),
111 ));
112 }
113 if &self.store != event.store {
117 return Err(SecretsError::Invalid(format!(
118 "this chain sink is bound to store `{}`; it will not record an event for store \
119 `{}`",
120 self.store, event.store
121 )));
122 }
123 let (seq, prev_hash) = self.tail().await?;
124 let ts = now();
125 let name = event.name.unwrap_or("");
126 let entry = Entry {
127 seq,
128 ts: &ts,
129 actor: event.actor.as_str(),
130 name,
131 version: event.version,
132 action: event.access,
133 allowed: event.allowed,
134 request_id: event.request_id,
135 store: event.store.as_str(),
136 };
137 let hash = entry.hash(&prev_hash);
138
139 self.db
140 .execute(&Statement::with_values(
141 "INSERT INTO harness_secret_audit \
142 (seq, ts, actor, name, version, action, allowed, request_id, store, prev_hash, \
143 hash) \
144 VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)",
145 vec![
146 SeaValue::BigInt(Some(seq)),
147 text(&ts),
148 text(event.actor.as_str()),
149 text(name),
150 match event.version {
151 Some(version) => SeaValue::BigInt(Some(i64::from(version))),
152 None => SeaValue::BigInt(None),
153 },
154 text(event.access.as_str()),
155 SeaValue::BigInt(Some(i64::from(event.allowed))),
156 match event.request_id {
157 Some(id) => text(id),
158 None => SeaValue::String(None),
159 },
160 text(event.store.as_str()),
161 bytes(prev_hash),
162 bytes(hash),
163 ],
164 ))
165 .await
166 .map_err(|err| SecretsError::NotAudited(err.to_string()))?;
167 Ok(())
168 }
169}
170
171struct Entry<'a> {
175 seq: i64,
176 ts: &'a str,
177 actor: &'a str,
178 name: &'a str,
179 version: Option<u32>,
180 action: Access,
181 allowed: bool,
182 request_id: Option<&'a str>,
183 store: &'a str,
187}
188
189impl Entry<'_> {
190 fn hash(&self, prev_hash: &[u8]) -> Vec<u8> {
197 fn push(hasher: &mut Sha256, bytes: &[u8]) {
198 hasher.update(u32::try_from(bytes.len()).unwrap_or(u32::MAX).to_le_bytes());
199 hasher.update(bytes);
200 }
201 let mut hasher = Sha256::new();
202 hasher.update(b"FZ-SECRET-AUDIT-v1");
203 hasher.update(prev_hash);
204 push(&mut hasher, &self.seq.to_le_bytes());
205 push(&mut hasher, self.ts.as_bytes());
206 push(&mut hasher, self.actor.as_bytes());
207 push(&mut hasher, self.name.as_bytes());
208 match self.version {
210 Some(version) => {
211 push(&mut hasher, b"v");
212 push(&mut hasher, &version.to_le_bytes());
213 }
214 None => push(&mut hasher, b"-"),
215 }
216 push(&mut hasher, self.action.as_str().as_bytes());
217 push(&mut hasher, &[u8::from(self.allowed)]);
218 push(&mut hasher, self.request_id.unwrap_or("").as_bytes());
219 if !self.store.is_empty() {
220 push(&mut hasher, self.store.as_bytes());
221 }
222 hasher.finalize().to_vec()
223 }
224}
225
226#[derive(Debug, Clone, PartialEq, Eq)]
231pub struct Anchor {
232 pub store: String,
233 pub seq: i64,
234 pub hash: String,
236}
237
238async fn first_foreign(
241 store: &StoreId,
242 db: &dyn Database,
243) -> Result<Option<(i64, String)>, SecretsError> {
244 let rows = db
245 .query(&Statement::with_values(
246 "SELECT seq, store FROM harness_secret_audit WHERE store <> '' AND store <> ? \
247 ORDER BY seq ASC LIMIT 1",
248 vec![text(store.as_str())],
249 ))
250 .await?;
251 let Some(row) = rows.first() else {
252 return Ok(None);
253 };
254 Ok(Some((
255 row.get("seq")
256 .ok_or_else(|| broken(store, 0, "an audit row has no seq"))?,
257 row.get::<String>("store")
258 .ok_or_else(|| broken(store, 0, "an audit row has no store"))?,
259 )))
260}
261
262fn seq_violation(
268 store: &StoreId,
269 seq: i64,
270 expected_seq: i64,
271 last_seq: i64,
272 shared_ledger: bool,
273) -> Option<SecretsError> {
274 if !shared_ledger && seq != expected_seq {
275 return Some(broken(
276 store,
277 seq,
278 &format!("expected seq {expected_seq}: a row is missing or out of order"),
279 ));
280 }
281 if shared_ledger && seq <= last_seq {
282 return Some(broken(
283 store,
284 seq,
285 "its seq is not greater than its predecessor's",
286 ));
287 }
288 None
289}
290
291fn copied_chain(store: &StoreId, seq: i64, other: &str) -> SecretsError {
294 broken(
295 store,
296 seq,
297 &format!(
298 "it belongs to store `{other}`: a chain copied from another database is not this \
299 store's history"
300 ),
301 )
302}
303
304pub async fn verify(store: &StoreId, db: &dyn Database) -> Result<Anchor, SecretsError> {
313 let first_foreign = first_foreign(store, db).await?;
322 let shared_ledger = first_foreign.is_some();
323
324 let rows = db
325 .query(&Statement::new(
326 "SELECT seq, ts, actor, name, version, action, allowed, request_id, store, \
327 prev_hash, hash FROM harness_secret_audit ORDER BY seq ASC",
328 ))
329 .await?;
330
331 let mut prev_hash = GENESIS.to_vec();
332 let mut expected_seq: i64 = 1;
333 let mut mine: i64 = 0;
334 let mut last = Anchor {
335 store: store.to_string(),
336 seq: 0,
337 hash: hex(&GENESIS),
338 };
339
340 for row in &rows.rows {
341 let seq: i64 = row
342 .get("seq")
343 .ok_or_else(|| broken(store, expected_seq, "a row has no seq"))?;
344 let row_store: String = row.get("store").unwrap_or_default();
345 if !row_store.is_empty() && row_store != store.as_str() {
351 continue;
352 }
353 if let Some(err) = seq_violation(store, seq, expected_seq, last.seq, shared_ledger) {
354 return Err(err);
355 }
356 let recorded_prev: Vec<u8> = row
357 .get("prev_hash")
358 .ok_or_else(|| broken(store, seq, "a row has no prev_hash"))?;
359 if recorded_prev != prev_hash {
360 return Err(broken(
361 store,
362 seq,
363 "its prev_hash is not the previous row's hash",
364 ));
365 }
366 let ts: String = row
367 .get("ts")
368 .ok_or_else(|| broken(store, seq, "a row has no timestamp"))?;
369 let actor: String = row
370 .get("actor")
371 .ok_or_else(|| broken(store, seq, "a row has no actor"))?;
372 let name: String = row.get("name").unwrap_or_default();
373 let action: String = row
374 .get("action")
375 .ok_or_else(|| broken(store, seq, "a row has no action"))?;
376 let action = Access::parse(&action)
377 .ok_or_else(|| broken(store, seq, &format!("unknown action `{action}`")))?;
378 let version = row
379 .get::<i64>("version")
380 .and_then(|v| u32::try_from(v).ok());
381 let allowed = row.get::<i64>("allowed").unwrap_or_default() != 0;
382 let request_id: Option<String> = row.get("request_id");
383 let recorded_hash: Vec<u8> = row
384 .get("hash")
385 .ok_or_else(|| broken(store, seq, "a row has no hash"))?;
386
387 let computed = Entry {
388 seq,
389 ts: &ts,
390 actor: &actor,
391 name: &name,
392 version,
393 action,
394 allowed,
395 request_id: request_id.as_deref(),
396 store: &row_store,
397 }
398 .hash(&prev_hash);
399 if computed != recorded_hash {
400 return Err(broken(
401 store,
402 seq,
403 "its hash does not match its contents: the row was altered",
404 ));
405 }
406
407 prev_hash = recorded_hash;
408 last = Anchor {
409 store: store.to_string(),
410 seq,
411 hash: hex(&prev_hash),
412 };
413 expected_seq = seq.saturating_add(1);
414 mine += 1;
415 }
416 if mine == 0
417 && let Some((seq, other)) = &first_foreign
418 {
419 return Err(copied_chain(store, *seq, other));
420 }
421 Ok(last)
422}
423
424pub fn chain_sink(db: Arc<dyn Database>) -> Arc<dyn Audit> {
434 Arc::new(ChainSink { db })
435}
436
437struct ChainSink {
438 db: Arc<dyn Database>,
439}
440
441#[async_trait::async_trait]
442impl Audit for ChainSink {
443 async fn record(&self, event: &AuditEvent<'_>) -> Result<(), SecretsError> {
444 ChainAudit::new(event.store.clone(), Arc::clone(&self.db))
445 .record(event)
446 .await
447 }
448}
449
450fn broken(store: &StoreId, seq: i64, detail: &str) -> SecretsError {
451 SecretsError::ChainBroken {
452 store: store.to_string(),
453 seq,
454 detail: detail.to_owned(),
455 }
456}
457
458fn hex(bytes: &[u8]) -> String {
459 let mut out = String::with_capacity(bytes.len() * 2);
460 for byte in bytes {
461 use std::fmt::Write as _;
462 let _ = write!(out, "{byte:02x}");
463 }
464 out
465}
466
467fn text(value: &str) -> SeaValue {
468 SeaValue::String(Some(Box::new(value.to_owned())))
469}
470
471fn bytes(value: Vec<u8>) -> SeaValue {
472 SeaValue::Bytes(Some(Box::new(value)))
473}
474
475fn now() -> String {
476 time::OffsetDateTime::now_utc()
477 .format(&time::format_description::well_known::Rfc3339)
478 .unwrap_or_default()
479}
480
481#[cfg(test)]
482mod tests {
483 use super::*;
484 use std::sync::Arc;
485
486 use cratefield_adapter_sqlite::SqliteDatabase;
487
488 use crate::{Actor, Audit, StoreId};
489
490 #[allow(clippy::too_many_arguments)]
499 fn v1_hash(
500 prev_hash: &[u8],
501 seq: i64,
502 ts: &str,
503 actor: &str,
504 name: &str,
505 version: Option<u32>,
506 action: &str,
507 allowed: bool,
508 request_id: Option<&str>,
509 ) -> Vec<u8> {
510 fn push(hasher: &mut Sha256, bytes: &[u8]) {
511 hasher.update(u32::try_from(bytes.len()).unwrap_or(u32::MAX).to_le_bytes());
512 hasher.update(bytes);
513 }
514 let mut hasher = Sha256::new();
515 hasher.update(b"FZ-SECRET-AUDIT-v1");
516 hasher.update(prev_hash);
517 push(&mut hasher, &seq.to_le_bytes());
518 push(&mut hasher, ts.as_bytes());
519 push(&mut hasher, actor.as_bytes());
520 push(&mut hasher, name.as_bytes());
521 match version {
522 Some(version) => {
523 push(&mut hasher, b"v");
524 push(&mut hasher, &version.to_le_bytes());
525 }
526 None => push(&mut hasher, b"-"),
527 }
528 push(&mut hasher, action.as_bytes());
529 push(&mut hasher, &[u8::from(allowed)]);
530 push(&mut hasher, request_id.unwrap_or("").as_bytes());
531 hasher.finalize().to_vec()
532 }
533
534 fn entry(store: &'static str) -> Entry<'static> {
535 Entry {
536 seq: 7,
537 ts: "2026-09-01T00:00:00Z",
538 actor: "auditor",
539 name: "stripe/api_key",
540 version: Some(3),
541 action: Access::Get,
542 allowed: true,
543 request_id: Some("req-9"),
544 store,
545 }
546 }
547
548 #[test]
549 fn a_legacy_row_still_hashes_exactly_as_v1() {
550 let prev = GENESIS.to_vec();
551 let legacy = entry("").hash(&prev);
552 assert_eq!(
553 legacy,
554 v1_hash(
555 &prev,
556 7,
557 "2026-09-01T00:00:00Z",
558 "auditor",
559 "stripe/api_key",
560 Some(3),
561 "get",
562 true,
563 Some("req-9"),
564 ),
565 "the pre-#142 encoding must not move, or migrated chains stop verifying"
566 );
567 assert_ne!(legacy, entry("tenant-a").hash(&prev));
570 }
571
572 #[pollster::test]
573 async fn a_migrated_chain_continues_under_the_new_encoding() {
574 let db = SqliteDatabase::in_memory().expect("in-memory db");
575 db.apply_migrations("secrets", crate::migrations().sqlite)
576 .expect("schema");
577 let db: Arc<dyn Database> = Arc::new(db);
578
579 let hash = v1_hash(
582 GENESIS.as_ref(),
583 1,
584 "2026-08-01T00:00:00Z",
585 "auditor",
586 "stripe/api_key",
587 Some(1),
588 "get",
589 true,
590 None,
591 );
592 db.execute(&Statement::with_values(
593 "INSERT INTO harness_secret_audit \
594 (seq, ts, actor, name, version, action, allowed, request_id, prev_hash, hash) \
595 VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)",
596 vec![
597 SeaValue::BigInt(Some(1)),
598 text("2026-08-01T00:00:00Z"),
599 text("auditor"),
600 text("stripe/api_key"),
601 SeaValue::BigInt(Some(1)),
602 text("get"),
603 SeaValue::BigInt(Some(1)),
604 SeaValue::String(None),
605 bytes(GENESIS.to_vec()),
606 bytes(hash),
607 ],
608 ))
609 .await
610 .expect("legacy row");
611
612 let id = StoreId::Tenant("tenant-a".to_owned());
614 let sink = ChainAudit::new(id.clone(), Arc::clone(&db));
615 let actor = Actor::new("auditor").expect("named");
616 sink.record(&crate::AuditEvent {
617 store: &id,
618 access: Access::Put,
619 actor: &actor,
620 name: Some("stripe/api_key"),
621 version: Some(2),
622 allowed: true,
623 request_id: None,
624 })
625 .await
626 .expect("record");
627
628 let anchor = verify(&id, &*db).await.expect("chain verifies");
629 assert_eq!(
630 anchor.seq, 2,
631 "the legacy row is part of the chain, not a wall"
632 );
633 }
634
635 #[pollster::test]
636 async fn a_router_sink_keeps_each_stores_chain_its_own() {
637 let db = SqliteDatabase::in_memory().expect("in-memory db");
638 db.apply_migrations("secrets", crate::migrations().sqlite)
639 .expect("schema");
640 let db: Arc<dyn Database> = Arc::new(db);
641
642 let sink = chain_sink(Arc::clone(&db));
643 let actor = Actor::new("auditor").expect("named");
644 let a = StoreId::Tenant("tenant-a".to_owned());
645 let b = StoreId::Tenant("tenant-b".to_owned());
646 sink.record(&crate::AuditEvent {
647 store: &a,
648 access: Access::Put,
649 actor: &actor,
650 name: Some("k"),
651 version: Some(1),
652 allowed: true,
653 request_id: None,
654 })
655 .await
656 .expect("record for a");
657 sink.record(&crate::AuditEvent {
658 store: &b,
659 access: Access::Get,
660 actor: &actor,
661 name: Some("k"),
662 version: Some(1),
663 allowed: false,
664 request_id: None,
665 })
666 .await
667 .expect("record for b");
668
669 let anchor_a = verify(&a, &*db).await.expect("chain a verifies");
674 let anchor_b = verify(&b, &*db).await.expect("chain b verifies");
675 assert_eq!(anchor_a.seq, 1, "tenant-a's chain ends at its own row");
676 assert_eq!(anchor_b.seq, 2, "tenant-b's chain ends at its own row");
677 }
678
679 #[pollster::test]
680 async fn a_sink_refuses_events_for_another_store() {
681 let db = SqliteDatabase::in_memory().expect("in-memory db");
682 db.apply_migrations("secrets", crate::migrations().sqlite)
683 .expect("schema");
684 let db: Arc<dyn Database> = Arc::new(db);
685
686 let actor = Actor::new("auditor").expect("named");
687 let global = ChainAudit::new(StoreId::Global, Arc::clone(&db));
688 let tenant = StoreId::Tenant("tenant-a".to_owned());
689 let err = global
690 .record(&crate::AuditEvent {
691 store: &tenant,
692 access: Access::Get,
693 actor: &actor,
694 name: None,
695 version: None,
696 allowed: true,
697 request_id: None,
698 })
699 .await
700 .expect_err("a Global sink will not write a tenant's row");
701 assert!(err.to_string().contains("bound to store"), "{err}");
702 assert_eq!(
703 verify(&StoreId::Global, &*db)
704 .await
705 .expect("nothing was written")
706 .seq,
707 0
708 );
709 }
710}