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::{verify_checkpoints, ChainRow, Checkpoint, CheckpointOutcome};
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 async fn verify_against_anchor(
296 verifying_key_hex: &str,
297 rows: &[ChainRow],
298 sink: &dyn CheckpointSink,
299) -> Result<CheckpointOutcome, CheckpointSinkError> {
300 let anchored = sink.load_all().await?;
301 if anchored.is_empty() {
302 return Ok(CheckpointOutcome::CannotVerify {
303 reason: "a checkpoint anchor is configured but holds no checkpoints, so a \
304 deleted tail would still go unnoticed; run `sysknife audit checkpoint` \
305 (periodically — the interval is the window an attacker can rewrite)"
306 .to_string(),
307 });
308 }
309 Ok(verify_checkpoints(verifying_key_hex, rows, &anchored))
310}
311
312pub const DEFAULT_ANCHOR_INTERVAL: Duration = Duration::from_secs(15 * 60);
319
320pub fn spawn_periodic_anchor(
328 audit: std::sync::Arc<dyn crate::store::AuditStore>,
329 key: std::sync::Arc<crate::audit_chain::AuditKey>,
330 sink: std::sync::Arc<dyn CheckpointSink>,
331 interval: Duration,
332) -> tokio::task::JoinHandle<()> {
333 tokio::spawn(async move {
334 loop {
335 tokio::time::sleep(interval).await;
336
337 let rows = match audit.fetch_chain_rows().await {
338 Ok(rows) => rows,
339 Err(e) => {
340 eprintln!("[sysknife-daemon] checkpoint anchor: reading chain failed: {e}");
341 continue;
342 }
343 };
344
345 let created_at = chrono::Utc::now().to_rfc3339();
346 match anchor_once(&key, &rows, sink.as_ref(), &created_at).await {
347 Ok(AnchorOutcome::Anchored {
348 seq,
349 checkpoints_checked,
350 }) => {
351 eprintln!(
352 "[sysknife-daemon] anchored checkpoint seq={seq} \
353 ({checkpoints_checked} verified)"
354 );
355 }
356 Ok(AnchorOutcome::ChainEmpty) => {}
357 Ok(AnchorOutcome::ChainBroken(outcome)) => {
361 eprintln!(
362 "[sysknife-daemon] checkpoint anchor REFUSED: the local audit chain \
363 does not verify ({outcome:?}); not anchoring a tampered chain"
364 );
365 }
366 Ok(AnchorOutcome::ReadBackMissing) => {
367 eprintln!(
368 "[sysknife-daemon] checkpoint anchor FAILED: the checkpoint was absent \
369 on read-back; the sink may be a lagging replica or a different database"
370 );
371 }
372 Ok(AnchorOutcome::Inconsistent(outcome)) => {
373 eprintln!(
374 "[sysknife-daemon] checkpoint verification FAILED: {outcome:?} — \
375 the local chain disagrees with a previously anchored checkpoint"
376 );
377 }
378 Err(e) => {
379 eprintln!("[sysknife-daemon] checkpoint anchor: sink error: {e}");
380 }
381 }
382 }
383 })
384}
385
386#[cfg(test)]
387mod tests {
388 use super::*;
389 use crate::audit_chain::{
390 verify_checkpoints, AuditKey, ChainContent, ChainIdentity, ChainRow, CheckpointOutcome,
391 CHAIN_VERSION_CURRENT,
392 };
393 use sysknife_types::RiskLevel;
394
395 fn key() -> AuditKey {
396 AuditKey::from_bytes(vec![0x42; 32])
397 }
398
399 #[tokio::test]
400 async fn refuses_a_downgradeable_remote_url_before_connecting() {
401 let refusal = tokio::time::timeout(
406 std::time::Duration::from_secs(2),
407 PostgresCheckpointSink::connect("postgres://u:p@db.example.com:5432/audit"),
408 )
409 .await
410 .expect("the guard must refuse without attempting a connection");
411 match refusal.expect_err("a remote checkpoint URL without TLS must be refused") {
412 CheckpointSinkError::Connect(m) => {
413 assert!(m.contains("sslmode"), "message should name sslmode: {m}")
414 }
415 other => panic!("expected a Connect refusal, got {other:?}"),
416 }
417 }
418
419 pub(super) fn build_chain(key: &AuditKey, count: usize) -> Vec<ChainRow> {
429 let mut rows = Vec::with_capacity(count);
430 let mut prev = String::new();
431 for i in 0..count {
432 let seq = (i + 1) as u64;
433 let txid = format!("tx{i}");
434 let content = ChainContent {
435 seq,
436 key_id: "v1",
437 transaction_id: &txid,
438 request_id: "req",
439 request_hash: "hash",
440 action_name: "UpdateSystem",
441 risk_level: RiskLevel::High,
442 summary: "s",
443 approval_id: None,
444 warnings_json: "[]",
445 created_at: "2026-04-24T12:00:00Z",
446 identity: ChainIdentity::V3 {
447 caller_role: "dev",
448 event_tip: "",
449 caller_principal: "uid:1000",
450 },
451 };
452 let hash = key.chain_hash(&content, &prev);
453 rows.push(ChainRow {
454 seq,
455 key_id: "v1".to_string(),
456 transaction_id: txid,
457 request_id: "req".to_string(),
458 request_hash: "hash".to_string(),
459 action_name: "UpdateSystem".to_string(),
460 risk_level: RiskLevel::High,
461 summary: "s".to_string(),
462 approval_id: None,
463 warnings_json: "[]".to_string(),
464 created_at: "2026-04-24T12:00:00Z".to_string(),
465 prev_chain_hash: prev.clone(),
466 chain_hash: hash.clone(),
467 chain_version: CHAIN_VERSION_CURRENT,
468 caller_role: Some("dev".to_string()),
469 event_tip: Some(String::new()),
470 caller_principal: Some("uid:1000".to_string()),
471 });
472 prev = hash;
473 }
474 rows
475 }
476
477 #[tokio::test]
478 async fn in_memory_append_and_load_round_trip() {
479 let key = key();
480 let rows = build_chain(&key, 3);
481 let sink = InMemoryCheckpointSink::new();
482 let cp = key.sign_checkpoint(3, &rows[2].chain_hash, "2026-04-24T12:00:00Z");
483 sink.append(&cp).await.unwrap();
484 let loaded = sink.load_all().await.unwrap();
485 assert_eq!(loaded, vec![cp]);
486 }
487
488 #[tokio::test]
489 async fn anchored_checkpoints_verify_and_catch_truncation() {
490 let key = key();
491 let full = build_chain(&key, 5);
492 let sink = InMemoryCheckpointSink::new();
493 let cp = key.sign_checkpoint(5, &full[4].chain_hash, "2026-04-24T12:00:00Z");
495 sink.append(&cp).await.unwrap();
496
497 let anchored = sink.load_all().await.unwrap();
499 assert_eq!(
500 verify_checkpoints(&key.verifying_key_hex(), &full, &anchored),
501 CheckpointOutcome::Consistent {
502 checkpoints_checked: 1
503 }
504 );
505
506 let truncated = &full[..3];
509 assert!(matches!(
510 verify_checkpoints(&key.verifying_key_hex(), truncated, &anchored),
511 CheckpointOutcome::Truncated {
512 checkpoint_seq: 5,
513 current_max_seq: 3
514 }
515 ));
516 }
517}
518
519#[cfg(test)]
520mod anchor_tests {
521 use super::*;
522 use crate::audit_chain::{AuditKey, ChainContent, ChainRow, CheckpointOutcome, VerifyOutcome};
523
524 fn key() -> AuditKey {
525 AuditKey::from_bytes(vec![0x42; 32])
526 }
527
528 fn reseal(key: &AuditKey, rows: &[ChainRow]) -> Vec<ChainRow> {
532 let mut out = Vec::with_capacity(rows.len());
533 let mut prev = String::new();
534 for row in rows {
535 let content = ChainContent {
536 seq: row.seq,
537 key_id: &row.key_id,
538 transaction_id: &row.transaction_id,
539 request_id: &row.request_id,
540 request_hash: &row.request_hash,
541 action_name: &row.action_name,
542 risk_level: row.risk_level,
543 summary: &row.summary,
544 approval_id: row.approval_id.as_deref(),
545 warnings_json: &row.warnings_json,
546 created_at: &row.created_at,
547 identity: row.identity().expect("fixture rows are well-formed"),
548 };
549 let hash = key.chain_hash(&content, &prev);
550 let mut cloned = row.clone();
551 cloned.prev_chain_hash = prev.clone();
552 cloned.chain_hash = hash.clone();
553 out.push(cloned);
554 prev = hash;
555 }
556 out
557 }
558
559 fn chain(key: &AuditKey, count: usize) -> Vec<ChainRow> {
560 super::tests::build_chain(key, count)
561 }
562
563 #[tokio::test]
564 async fn anchor_once_writes_and_verifies_the_tip() {
565 let key = key();
566 let rows = chain(&key, 3);
567 let sink = InMemoryCheckpointSink::new();
568
569 let outcome = anchor_once(&key, &rows, &sink, "2026-04-24T12:00:00Z")
570 .await
571 .unwrap();
572
573 assert_eq!(
574 outcome,
575 AnchorOutcome::Anchored {
576 seq: 3,
577 checkpoints_checked: 1
578 }
579 );
580 assert_eq!(sink.load_all().await.unwrap().len(), 1);
581 }
582
583 #[tokio::test]
584 async fn anchor_once_refuses_to_launder_a_tampered_chain() {
585 let key = key();
588 let mut rows = chain(&key, 3);
589 rows[1].summary = "tampered".to_string();
590 let sink = InMemoryCheckpointSink::new();
591
592 let outcome = anchor_once(&key, &rows, &sink, "2026-04-24T12:00:00Z")
593 .await
594 .unwrap();
595
596 assert!(
597 matches!(
598 outcome,
599 AnchorOutcome::ChainBroken(VerifyOutcome::Broken { .. })
600 ),
601 "expected a refusal, got {outcome:?}"
602 );
603 assert!(
604 sink.load_all().await.unwrap().is_empty(),
605 "nothing may be anchored when the chain does not verify"
606 );
607 }
608
609 #[tokio::test]
610 async fn anchor_once_reports_an_empty_chain_rather_than_anchoring_nothing() {
611 let key = key();
612 let sink = InMemoryCheckpointSink::new();
613 let outcome = anchor_once(&key, &[], &sink, "2026-04-24T12:00:00Z")
614 .await
615 .unwrap();
616 assert_eq!(outcome, AnchorOutcome::ChainEmpty);
617 assert!(sink.load_all().await.unwrap().is_empty());
618 }
619
620 #[derive(Default)]
623 struct BlackHoleSink;
624
625 #[async_trait]
626 impl CheckpointSink for BlackHoleSink {
627 async fn append(
628 &self,
629 _checkpoint: &crate::audit_chain::Checkpoint,
630 ) -> Result<(), CheckpointSinkError> {
631 Ok(())
632 }
633 async fn load_all(
634 &self,
635 ) -> Result<Vec<crate::audit_chain::Checkpoint>, CheckpointSinkError> {
636 Ok(Vec::new())
637 }
638 }
639
640 #[tokio::test]
641 async fn anchor_once_detects_a_write_that_did_not_land() {
642 let key = key();
647 let rows = chain(&key, 2);
648
649 let outcome = anchor_once(&key, &rows, &BlackHoleSink, "2026-04-24T12:00:00Z")
650 .await
651 .unwrap();
652
653 assert_eq!(outcome, AnchorOutcome::ReadBackMissing);
654 }
655
656 #[tokio::test]
657 async fn a_previously_anchored_checkpoint_catches_a_key_holder_rewrite() {
658 let key = key();
666 let original = chain(&key, 5);
667 let sink = InMemoryCheckpointSink::new();
668
669 anchor_once(&key, &original, &sink, "2026-04-24T12:00:00Z")
670 .await
671 .unwrap();
672 let anchored = sink.load_all().await.unwrap();
673
674 let mut tampered = original.clone();
675 tampered[3].summary = "quietly rewritten".to_string();
676 let tampered = reseal(&key, &tampered);
677
678 assert!(
679 matches!(
680 crate::audit_chain::verify_chain(&key, &tampered),
681 VerifyOutcome::Intact { .. }
682 ),
683 "the forged chain must verify — otherwise this test proves nothing \
684 beyond what a plain signature check already catches"
685 );
686
687 let outcome =
688 crate::audit_chain::verify_checkpoints(&key.verifying_key_hex(), &tampered, &anchored);
689 assert!(
690 matches!(outcome, CheckpointOutcome::TipMismatch { .. }),
691 "the anchored tip must expose the rewrite, got {outcome:?}"
692 );
693 }
694
695 #[tokio::test]
702 async fn verifying_against_the_anchor_catches_what_the_chain_alone_cannot() {
703 let key = key();
704 let full = chain(&key, 5);
705 let sink = InMemoryCheckpointSink::new();
706 anchor_once(&key, &full, &sink, "2026-04-24T12:00:00Z")
707 .await
708 .unwrap();
709
710 let outcome = verify_against_anchor(&key.verifying_key_hex(), &full, &sink)
712 .await
713 .unwrap();
714 assert!(
715 matches!(outcome, CheckpointOutcome::Consistent { .. }),
716 "an untouched chain must verify against its anchor, got {outcome:?}"
717 );
718
719 for remaining in [3, 0] {
722 let truncated = &full[..remaining];
723 let outcome = verify_against_anchor(&key.verifying_key_hex(), truncated, &sink)
724 .await
725 .unwrap();
726 assert!(
727 matches!(outcome, CheckpointOutcome::Truncated { .. }),
728 "a deleted tail, including complete erasure, must be reported; remaining={remaining}, got {outcome:?}"
729 );
730 }
731 }
732
733 #[tokio::test]
737 async fn an_anchor_with_no_checkpoints_is_not_reported_as_consistent() {
738 let key = key();
739 let full = chain(&key, 5);
740 let sink = InMemoryCheckpointSink::new();
741
742 let outcome = verify_against_anchor(&key.verifying_key_hex(), &full, &sink)
743 .await
744 .unwrap();
745 assert!(
746 matches!(outcome, CheckpointOutcome::CannotVerify { .. }),
747 "an empty anchor proves nothing and must not read as success, got {outcome:?}"
748 );
749 }
750
751 #[tokio::test]
752 async fn verify_chain_alone_cannot_detect_tail_truncation() {
753 let key = key();
756 let full = chain(&key, 5);
757 let truncated = &full[..3];
758
759 assert!(
760 matches!(
761 crate::audit_chain::verify_chain(&key, truncated),
762 VerifyOutcome::Intact {
763 rows_checked: 3,
764 ..
765 }
766 ),
767 "a truncated chain still verifies on its own — this is the gap"
768 );
769
770 let sink = InMemoryCheckpointSink::new();
771 anchor_once(&key, &full, &sink, "2026-04-24T12:00:00Z")
772 .await
773 .unwrap();
774 let anchored = sink.load_all().await.unwrap();
775
776 assert!(
777 matches!(
778 crate::audit_chain::verify_checkpoints(
779 &key.verifying_key_hex(),
780 truncated,
781 &anchored
782 ),
783 CheckpointOutcome::Truncated { .. }
784 ),
785 "the anchor is what turns an undetectable truncation into a detected one"
786 );
787 }
788}