1use async_trait::async_trait;
30use sqlx_core::row::Row;
31use sqlx_postgres::{PgConnectOptions, PgPool, PgPoolOptions};
32use std::str::FromStr;
33#[cfg(test)]
35use std::sync::Mutex;
36use std::time::Duration;
37
38use crate::audit_chain::Checkpoint;
39
40#[derive(Debug, thiserror::Error)]
41pub enum CheckpointSinkError {
42 #[error("checkpoint sink connection error: {0}")]
43 Connect(String),
44 #[error("checkpoint sink query error: {0}")]
45 Query(String),
46}
47
48impl From<sqlx_core::Error> for CheckpointSinkError {
49 fn from(e: sqlx_core::Error) -> Self {
50 CheckpointSinkError::Query(e.to_string())
51 }
52}
53
54#[async_trait]
58pub trait CheckpointSink: Send + Sync {
59 async fn append(&self, checkpoint: &Checkpoint) -> Result<(), CheckpointSinkError>;
61
62 async fn load_all(&self) -> Result<Vec<Checkpoint>, CheckpointSinkError>;
64}
65
66#[cfg(test)]
71#[derive(Debug, Default)]
72pub struct InMemoryCheckpointSink {
73 stored: Mutex<Vec<Checkpoint>>,
74}
75
76#[cfg(test)]
77impl InMemoryCheckpointSink {
78 pub fn new() -> Self {
79 Self::default()
80 }
81}
82
83#[cfg(test)]
84#[async_trait]
85impl CheckpointSink for InMemoryCheckpointSink {
86 async fn append(&self, checkpoint: &Checkpoint) -> Result<(), CheckpointSinkError> {
87 let mut stored = self
88 .stored
89 .lock()
90 .map_err(|e| CheckpointSinkError::Query(format!("lock poisoned: {e}")))?;
91 stored.push(checkpoint.clone());
92 Ok(())
93 }
94
95 async fn load_all(&self) -> Result<Vec<Checkpoint>, CheckpointSinkError> {
96 let stored = self
97 .stored
98 .lock()
99 .map_err(|e| CheckpointSinkError::Query(format!("lock poisoned: {e}")))?;
100 let mut out = stored.clone();
101 out.sort_by_key(|c| c.seq);
102 Ok(out)
103 }
104}
105
106#[derive(Debug)]
110pub struct PostgresCheckpointSink {
111 pool: PgPool,
112}
113
114impl PostgresCheckpointSink {
115 pub async fn connect(url: &str) -> Result<Self, CheckpointSinkError> {
118 let opts = PgConnectOptions::from_str(url).map_err(|_| {
120 CheckpointSinkError::Connect("invalid checkpoint database URL".to_string())
121 })?;
122 crate::pg_tls::require_tls_for_remote(&opts).map_err(CheckpointSinkError::Connect)?;
125 let pool = PgPoolOptions::new()
126 .max_connections(4)
127 .acquire_timeout(Duration::from_secs(10))
128 .connect_with(opts)
129 .await
130 .map_err(|e| CheckpointSinkError::Connect(e.to_string()))?;
131 let sink = Self { pool };
132 sink.initialize().await?;
133 Ok(sink)
134 }
135
136 async fn initialize(&self) -> Result<(), CheckpointSinkError> {
137 let row = sqlx_core::query::query(
142 "SELECT to_regclass('audit_checkpoints') IS NOT NULL AS present",
143 )
144 .fetch_one(&self.pool)
145 .await?;
146 let present: bool = row.try_get("present")?;
147 if present {
148 return Ok(());
149 }
150 sqlx_core::query::query(
151 "CREATE TABLE IF NOT EXISTS audit_checkpoints (\
152 seq BIGINT NOT NULL, \
153 chain_tip TEXT NOT NULL, \
154 created_at TEXT NOT NULL, \
155 signature TEXT NOT NULL\
156 )",
157 )
158 .execute(&self.pool)
159 .await?;
160 Ok(())
161 }
162}
163
164#[async_trait]
165impl CheckpointSink for PostgresCheckpointSink {
166 async fn append(&self, checkpoint: &Checkpoint) -> Result<(), CheckpointSinkError> {
167 sqlx_core::query::query(
168 "INSERT INTO audit_checkpoints (seq, chain_tip, created_at, signature) \
169 VALUES ($1, $2, $3, $4)",
170 )
171 .bind(i64::try_from(checkpoint.seq).map_err(|_| {
172 CheckpointSinkError::Query("checkpoint seq exceeds i64 range".to_string())
173 })?)
174 .bind(&checkpoint.chain_tip)
175 .bind(&checkpoint.created_at)
176 .bind(&checkpoint.signature)
177 .execute(&self.pool)
178 .await?;
179 Ok(())
180 }
181
182 async fn load_all(&self) -> Result<Vec<Checkpoint>, CheckpointSinkError> {
183 let rows = sqlx_core::query::query(
184 "SELECT seq, chain_tip, created_at, signature \
185 FROM audit_checkpoints ORDER BY seq ASC",
186 )
187 .fetch_all(&self.pool)
188 .await?;
189
190 let mut out = Vec::with_capacity(rows.len());
191 for row in rows {
192 let seq: i64 = row.try_get("seq")?;
193 out.push(Checkpoint {
194 seq: u64::try_from(seq).map_err(|e| {
195 CheckpointSinkError::Query(format!("negative checkpoint seq: {e}"))
196 })?,
197 chain_tip: row.try_get("chain_tip")?,
198 created_at: row.try_get("created_at")?,
199 signature: row.try_get("signature")?,
200 });
201 }
202 Ok(out)
203 }
204}
205
206#[derive(Debug, PartialEq, Eq)]
217pub enum AnchorOutcome {
218 Anchored { seq: u64, checkpoints_checked: u64 },
220 ChainEmpty,
222 ChainBroken(crate::audit_chain::VerifyOutcome),
224 ReadBackMissing,
226 Inconsistent(crate::audit_chain::CheckpointOutcome),
228}
229
230pub async fn anchor_once(
242 key: &crate::audit_chain::AuditKey,
243 rows: &[crate::audit_chain::ChainRow],
244 sink: &dyn CheckpointSink,
245 created_at: &str,
246) -> Result<AnchorOutcome, CheckpointSinkError> {
247 use crate::audit_chain::{verify_chain, verify_checkpoints, CheckpointOutcome, VerifyOutcome};
248
249 let Some(tip) = rows.last() else {
250 return Ok(AnchorOutcome::ChainEmpty);
251 };
252
253 match verify_chain(key, rows) {
254 VerifyOutcome::Intact { .. } => {}
255 broken => return Ok(AnchorOutcome::ChainBroken(broken)),
256 }
257
258 let checkpoint = key.sign_checkpoint(tip.seq, &tip.chain_hash, created_at);
259 sink.append(&checkpoint).await?;
260
261 let anchored = sink.load_all().await?;
262 if !anchored.contains(&checkpoint) {
263 return Ok(AnchorOutcome::ReadBackMissing);
264 }
265
266 match verify_checkpoints(&key.verifying_key_hex(), rows, &anchored) {
267 CheckpointOutcome::Consistent {
268 checkpoints_checked,
269 } => Ok(AnchorOutcome::Anchored {
270 seq: checkpoint.seq,
271 checkpoints_checked,
272 }),
273 other => Ok(AnchorOutcome::Inconsistent(other)),
274 }
275}
276
277pub const DEFAULT_ANCHOR_INTERVAL: Duration = Duration::from_secs(15 * 60);
284
285pub fn spawn_periodic_anchor(
293 audit: std::sync::Arc<dyn crate::store::AuditStore>,
294 key: std::sync::Arc<crate::audit_chain::AuditKey>,
295 sink: std::sync::Arc<dyn CheckpointSink>,
296 interval: Duration,
297) -> tokio::task::JoinHandle<()> {
298 tokio::spawn(async move {
299 loop {
300 tokio::time::sleep(interval).await;
301
302 let rows = match audit.fetch_chain_rows().await {
303 Ok(rows) => rows,
304 Err(e) => {
305 eprintln!("[sysknife-daemon] checkpoint anchor: reading chain failed: {e}");
306 continue;
307 }
308 };
309
310 let created_at = chrono::Utc::now().to_rfc3339();
311 match anchor_once(&key, &rows, sink.as_ref(), &created_at).await {
312 Ok(AnchorOutcome::Anchored {
313 seq,
314 checkpoints_checked,
315 }) => {
316 eprintln!(
317 "[sysknife-daemon] anchored checkpoint seq={seq} \
318 ({checkpoints_checked} verified)"
319 );
320 }
321 Ok(AnchorOutcome::ChainEmpty) => {}
322 Ok(AnchorOutcome::ChainBroken(outcome)) => {
326 eprintln!(
327 "[sysknife-daemon] checkpoint anchor REFUSED: the local audit chain \
328 does not verify ({outcome:?}); not anchoring a tampered chain"
329 );
330 }
331 Ok(AnchorOutcome::ReadBackMissing) => {
332 eprintln!(
333 "[sysknife-daemon] checkpoint anchor FAILED: the checkpoint was absent \
334 on read-back; the sink may be a lagging replica or a different database"
335 );
336 }
337 Ok(AnchorOutcome::Inconsistent(outcome)) => {
338 eprintln!(
339 "[sysknife-daemon] checkpoint verification FAILED: {outcome:?} — \
340 the local chain disagrees with a previously anchored checkpoint"
341 );
342 }
343 Err(e) => {
344 eprintln!("[sysknife-daemon] checkpoint anchor: sink error: {e}");
345 }
346 }
347 }
348 })
349}
350
351#[cfg(test)]
352mod tests {
353 use super::*;
354 use crate::audit_chain::{
355 verify_checkpoints, AuditKey, ChainContent, ChainIdentity, ChainRow, CheckpointOutcome,
356 CHAIN_VERSION_CURRENT,
357 };
358 use sysknife_types::RiskLevel;
359
360 fn key() -> AuditKey {
361 AuditKey::from_bytes(vec![0x42; 32])
362 }
363
364 #[tokio::test]
365 async fn refuses_a_downgradeable_remote_url_before_connecting() {
366 let refusal = tokio::time::timeout(
371 std::time::Duration::from_secs(2),
372 PostgresCheckpointSink::connect("postgres://u:p@db.example.com:5432/audit"),
373 )
374 .await
375 .expect("the guard must refuse without attempting a connection");
376 match refusal.expect_err("a remote checkpoint URL without TLS must be refused") {
377 CheckpointSinkError::Connect(m) => {
378 assert!(m.contains("sslmode"), "message should name sslmode: {m}")
379 }
380 other => panic!("expected a Connect refusal, got {other:?}"),
381 }
382 }
383
384 pub(super) fn build_chain(key: &AuditKey, count: usize) -> Vec<ChainRow> {
394 let mut rows = Vec::with_capacity(count);
395 let mut prev = String::new();
396 for i in 0..count {
397 let seq = (i + 1) as u64;
398 let txid = format!("tx{i}");
399 let content = ChainContent {
400 seq,
401 key_id: "v1",
402 transaction_id: &txid,
403 request_id: "req",
404 request_hash: "hash",
405 action_name: "UpdateSystem",
406 risk_level: RiskLevel::High,
407 summary: "s",
408 approval_id: None,
409 warnings_json: "[]",
410 created_at: "2026-04-24T12:00:00Z",
411 identity: ChainIdentity::V3 {
412 caller_role: "dev",
413 event_tip: "",
414 caller_principal: "uid:1000",
415 },
416 };
417 let hash = key.chain_hash(&content, &prev);
418 rows.push(ChainRow {
419 seq,
420 key_id: "v1".to_string(),
421 transaction_id: txid,
422 request_id: "req".to_string(),
423 request_hash: "hash".to_string(),
424 action_name: "UpdateSystem".to_string(),
425 risk_level: RiskLevel::High,
426 summary: "s".to_string(),
427 approval_id: None,
428 warnings_json: "[]".to_string(),
429 created_at: "2026-04-24T12:00:00Z".to_string(),
430 prev_chain_hash: prev.clone(),
431 chain_hash: hash.clone(),
432 chain_version: CHAIN_VERSION_CURRENT,
433 caller_role: Some("dev".to_string()),
434 event_tip: Some(String::new()),
435 caller_principal: Some("uid:1000".to_string()),
436 });
437 prev = hash;
438 }
439 rows
440 }
441
442 #[tokio::test]
443 async fn in_memory_append_and_load_round_trip() {
444 let key = key();
445 let rows = build_chain(&key, 3);
446 let sink = InMemoryCheckpointSink::new();
447 let cp = key.sign_checkpoint(3, &rows[2].chain_hash, "2026-04-24T12:00:00Z");
448 sink.append(&cp).await.unwrap();
449 let loaded = sink.load_all().await.unwrap();
450 assert_eq!(loaded, vec![cp]);
451 }
452
453 #[tokio::test]
454 async fn anchored_checkpoints_verify_and_catch_truncation() {
455 let key = key();
456 let full = build_chain(&key, 5);
457 let sink = InMemoryCheckpointSink::new();
458 let cp = key.sign_checkpoint(5, &full[4].chain_hash, "2026-04-24T12:00:00Z");
460 sink.append(&cp).await.unwrap();
461
462 let anchored = sink.load_all().await.unwrap();
464 assert_eq!(
465 verify_checkpoints(&key.verifying_key_hex(), &full, &anchored),
466 CheckpointOutcome::Consistent {
467 checkpoints_checked: 1
468 }
469 );
470
471 let truncated = &full[..3];
474 assert!(matches!(
475 verify_checkpoints(&key.verifying_key_hex(), truncated, &anchored),
476 CheckpointOutcome::Truncated {
477 checkpoint_seq: 5,
478 current_max_seq: 3
479 }
480 ));
481 }
482}
483
484#[cfg(test)]
485mod anchor_tests {
486 use super::*;
487 use crate::audit_chain::{AuditKey, ChainContent, ChainRow, CheckpointOutcome, VerifyOutcome};
488
489 fn key() -> AuditKey {
490 AuditKey::from_bytes(vec![0x42; 32])
491 }
492
493 fn reseal(key: &AuditKey, rows: &[ChainRow]) -> Vec<ChainRow> {
497 let mut out = Vec::with_capacity(rows.len());
498 let mut prev = String::new();
499 for row in rows {
500 let content = ChainContent {
501 seq: row.seq,
502 key_id: &row.key_id,
503 transaction_id: &row.transaction_id,
504 request_id: &row.request_id,
505 request_hash: &row.request_hash,
506 action_name: &row.action_name,
507 risk_level: row.risk_level,
508 summary: &row.summary,
509 approval_id: row.approval_id.as_deref(),
510 warnings_json: &row.warnings_json,
511 created_at: &row.created_at,
512 identity: row.identity().expect("fixture rows are well-formed"),
513 };
514 let hash = key.chain_hash(&content, &prev);
515 let mut cloned = row.clone();
516 cloned.prev_chain_hash = prev.clone();
517 cloned.chain_hash = hash.clone();
518 out.push(cloned);
519 prev = hash;
520 }
521 out
522 }
523
524 fn chain(key: &AuditKey, count: usize) -> Vec<ChainRow> {
525 super::tests::build_chain(key, count)
526 }
527
528 #[tokio::test]
529 async fn anchor_once_writes_and_verifies_the_tip() {
530 let key = key();
531 let rows = chain(&key, 3);
532 let sink = InMemoryCheckpointSink::new();
533
534 let outcome = anchor_once(&key, &rows, &sink, "2026-04-24T12:00:00Z")
535 .await
536 .unwrap();
537
538 assert_eq!(
539 outcome,
540 AnchorOutcome::Anchored {
541 seq: 3,
542 checkpoints_checked: 1
543 }
544 );
545 assert_eq!(sink.load_all().await.unwrap().len(), 1);
546 }
547
548 #[tokio::test]
549 async fn anchor_once_refuses_to_launder_a_tampered_chain() {
550 let key = key();
553 let mut rows = chain(&key, 3);
554 rows[1].summary = "tampered".to_string();
555 let sink = InMemoryCheckpointSink::new();
556
557 let outcome = anchor_once(&key, &rows, &sink, "2026-04-24T12:00:00Z")
558 .await
559 .unwrap();
560
561 assert!(
562 matches!(
563 outcome,
564 AnchorOutcome::ChainBroken(VerifyOutcome::Broken { .. })
565 ),
566 "expected a refusal, got {outcome:?}"
567 );
568 assert!(
569 sink.load_all().await.unwrap().is_empty(),
570 "nothing may be anchored when the chain does not verify"
571 );
572 }
573
574 #[tokio::test]
575 async fn anchor_once_reports_an_empty_chain_rather_than_anchoring_nothing() {
576 let key = key();
577 let sink = InMemoryCheckpointSink::new();
578 let outcome = anchor_once(&key, &[], &sink, "2026-04-24T12:00:00Z")
579 .await
580 .unwrap();
581 assert_eq!(outcome, AnchorOutcome::ChainEmpty);
582 assert!(sink.load_all().await.unwrap().is_empty());
583 }
584
585 #[derive(Default)]
588 struct BlackHoleSink;
589
590 #[async_trait]
591 impl CheckpointSink for BlackHoleSink {
592 async fn append(
593 &self,
594 _checkpoint: &crate::audit_chain::Checkpoint,
595 ) -> Result<(), CheckpointSinkError> {
596 Ok(())
597 }
598 async fn load_all(
599 &self,
600 ) -> Result<Vec<crate::audit_chain::Checkpoint>, CheckpointSinkError> {
601 Ok(Vec::new())
602 }
603 }
604
605 #[tokio::test]
606 async fn anchor_once_detects_a_write_that_did_not_land() {
607 let key = key();
612 let rows = chain(&key, 2);
613
614 let outcome = anchor_once(&key, &rows, &BlackHoleSink, "2026-04-24T12:00:00Z")
615 .await
616 .unwrap();
617
618 assert_eq!(outcome, AnchorOutcome::ReadBackMissing);
619 }
620
621 #[tokio::test]
622 async fn a_previously_anchored_checkpoint_catches_a_key_holder_rewrite() {
623 let key = key();
631 let original = chain(&key, 5);
632 let sink = InMemoryCheckpointSink::new();
633
634 anchor_once(&key, &original, &sink, "2026-04-24T12:00:00Z")
635 .await
636 .unwrap();
637 let anchored = sink.load_all().await.unwrap();
638
639 let mut tampered = original.clone();
640 tampered[3].summary = "quietly rewritten".to_string();
641 let tampered = reseal(&key, &tampered);
642
643 assert!(
644 matches!(
645 crate::audit_chain::verify_chain(&key, &tampered),
646 VerifyOutcome::Intact { .. }
647 ),
648 "the forged chain must verify — otherwise this test proves nothing \
649 beyond what a plain signature check already catches"
650 );
651
652 let outcome =
653 crate::audit_chain::verify_checkpoints(&key.verifying_key_hex(), &tampered, &anchored);
654 assert!(
655 matches!(outcome, CheckpointOutcome::TipMismatch { .. }),
656 "the anchored tip must expose the rewrite, got {outcome:?}"
657 );
658 }
659
660 #[tokio::test]
661 async fn verify_chain_alone_cannot_detect_tail_truncation() {
662 let key = key();
665 let full = chain(&key, 5);
666 let truncated = &full[..3];
667
668 assert!(
669 matches!(
670 crate::audit_chain::verify_chain(&key, truncated),
671 VerifyOutcome::Intact {
672 rows_checked: 3,
673 ..
674 }
675 ),
676 "a truncated chain still verifies on its own — this is the gap"
677 );
678
679 let sink = InMemoryCheckpointSink::new();
680 anchor_once(&key, &full, &sink, "2026-04-24T12:00:00Z")
681 .await
682 .unwrap();
683 let anchored = sink.load_all().await.unwrap();
684
685 assert!(
686 matches!(
687 crate::audit_chain::verify_checkpoints(
688 &key.verifying_key_hex(),
689 truncated,
690 &anchored
691 ),
692 CheckpointOutcome::Truncated { .. }
693 ),
694 "the anchor is what turns an undetectable truncation into a detected one"
695 );
696 }
697}