sysknife_daemon/
checkpoint_sink.rs

1//! External anchor sinks for signed audit checkpoints.
2//!
3//! A [`Checkpoint`] commits `(seq, chain_tip,
4//! timestamp)` with an Ed25519 signature. Anchoring those checkpoints to an
5//! **independent, append-only** store is what makes tail-truncation and
6//! rewrite of the local chain *detectable* by a host attacker who controls the
7//! primary database: they cannot reproduce a previously anchored signed tip
8//! (see [`verify_checkpoints`](crate::audit_chain::verify_checkpoints)).
9//!
10//! This module defines a small [`CheckpointSink`] interface with two backends:
11//!
12//! - [`PostgresCheckpointSink`] — writes checkpoints to an append-only
13//!   `audit_checkpoints` table on a separate Postgres database. INSERT-only by
14//!   construction (this code never issues UPDATE/DELETE). Operators should
15//!   additionally grant the SysKnife role only `INSERT`/`SELECT` on the table
16//!   and `REVOKE UPDATE, DELETE` so a stolen daemon credential cannot rewrite
17//!   the anchor either. Append-only permissions alone do not stop a DB
18//!   superuser; the *signature* is what makes tampering detectable.
19//! - [`InMemoryCheckpointSink`] — for tests and dry runs.
20//!
21//! Additional verifiable backends (immudb, WORM object storage, an RFC 3161
22//! timestamp authority) can implement the same trait.
23
24use async_trait::async_trait;
25use sqlx_core::row::Row;
26use sqlx_postgres::{PgConnectOptions, PgPool, PgPoolOptions};
27use std::str::FromStr;
28use std::sync::Mutex;
29use std::time::Duration;
30
31use crate::audit_chain::Checkpoint;
32
33#[derive(Debug, thiserror::Error)]
34pub enum CheckpointSinkError {
35    #[error("checkpoint sink connection error: {0}")]
36    Connect(String),
37    #[error("checkpoint sink query error: {0}")]
38    Query(String),
39}
40
41impl From<sqlx_core::Error> for CheckpointSinkError {
42    fn from(e: sqlx_core::Error) -> Self {
43        CheckpointSinkError::Query(e.to_string())
44    }
45}
46
47/// An append-only sink that stores signed checkpoints and hands them back for
48/// verification. Implementations must never mutate or delete a stored
49/// checkpoint.
50#[async_trait]
51pub trait CheckpointSink: Send + Sync {
52    /// Append one signed checkpoint. Must not overwrite prior checkpoints.
53    async fn append(&self, checkpoint: &Checkpoint) -> Result<(), CheckpointSinkError>;
54
55    /// Load every stored checkpoint, ordered by `seq` ascending.
56    async fn load_all(&self) -> Result<Vec<Checkpoint>, CheckpointSinkError>;
57}
58
59/// In-memory checkpoint sink for tests and dry runs. Append-only.
60#[derive(Debug, Default)]
61pub struct InMemoryCheckpointSink {
62    stored: Mutex<Vec<Checkpoint>>,
63}
64
65impl InMemoryCheckpointSink {
66    pub fn new() -> Self {
67        Self::default()
68    }
69}
70
71#[async_trait]
72impl CheckpointSink for InMemoryCheckpointSink {
73    async fn append(&self, checkpoint: &Checkpoint) -> Result<(), CheckpointSinkError> {
74        let mut stored = self
75            .stored
76            .lock()
77            .map_err(|e| CheckpointSinkError::Query(format!("lock poisoned: {e}")))?;
78        stored.push(checkpoint.clone());
79        Ok(())
80    }
81
82    async fn load_all(&self) -> Result<Vec<Checkpoint>, CheckpointSinkError> {
83        let stored = self
84            .stored
85            .lock()
86            .map_err(|e| CheckpointSinkError::Query(format!("lock poisoned: {e}")))?;
87        let mut out = stored.clone();
88        out.sort_by_key(|c| c.seq);
89        Ok(out)
90    }
91}
92
93/// Postgres append-only checkpoint sink. Anchors signed checkpoints to a
94/// separate database so a host attacker cannot silently rewrite or truncate
95/// the local chain without being detected on the next `verify`.
96#[derive(Debug)]
97pub struct PostgresCheckpointSink {
98    pool: PgPool,
99}
100
101impl PostgresCheckpointSink {
102    /// Connect to `url`, create the append-only `audit_checkpoints` table if it
103    /// does not exist, and return the sink.
104    pub async fn connect(url: &str) -> Result<Self, CheckpointSinkError> {
105        // Do not echo the URL/DSN into the error (it can carry credentials).
106        let opts = PgConnectOptions::from_str(url).map_err(|_| {
107            CheckpointSinkError::Connect("invalid checkpoint database URL".to_string())
108        })?;
109        let pool = PgPoolOptions::new()
110            .max_connections(4)
111            .acquire_timeout(Duration::from_secs(10))
112            .connect_with(opts)
113            .await
114            .map_err(|e| CheckpointSinkError::Connect(e.to_string()))?;
115        let sink = Self { pool };
116        sink.initialize().await?;
117        Ok(sink)
118    }
119
120    async fn initialize(&self) -> Result<(), CheckpointSinkError> {
121        // Only CREATE when the table is absent. This lets a hardened,
122        // least-privilege role (INSERT/SELECT only, no DDL, per the module
123        // docs) connect to an already-provisioned table; the one-time CREATE
124        // is an admin/bootstrap step that a DDL-capable role performs once.
125        let row = sqlx_core::query::query(
126            "SELECT to_regclass('audit_checkpoints') IS NOT NULL AS present",
127        )
128        .fetch_one(&self.pool)
129        .await?;
130        let present: bool = row.try_get("present")?;
131        if present {
132            return Ok(());
133        }
134        sqlx_core::query::query(
135            "CREATE TABLE IF NOT EXISTS audit_checkpoints (\
136                 seq BIGINT NOT NULL, \
137                 chain_tip TEXT NOT NULL, \
138                 created_at TEXT NOT NULL, \
139                 signature TEXT NOT NULL\
140             )",
141        )
142        .execute(&self.pool)
143        .await?;
144        Ok(())
145    }
146}
147
148#[async_trait]
149impl CheckpointSink for PostgresCheckpointSink {
150    async fn append(&self, checkpoint: &Checkpoint) -> Result<(), CheckpointSinkError> {
151        sqlx_core::query::query(
152            "INSERT INTO audit_checkpoints (seq, chain_tip, created_at, signature) \
153             VALUES ($1, $2, $3, $4)",
154        )
155        .bind(i64::try_from(checkpoint.seq).map_err(|_| {
156            CheckpointSinkError::Query("checkpoint seq exceeds i64 range".to_string())
157        })?)
158        .bind(&checkpoint.chain_tip)
159        .bind(&checkpoint.created_at)
160        .bind(&checkpoint.signature)
161        .execute(&self.pool)
162        .await?;
163        Ok(())
164    }
165
166    async fn load_all(&self) -> Result<Vec<Checkpoint>, CheckpointSinkError> {
167        let rows = sqlx_core::query::query(
168            "SELECT seq, chain_tip, created_at, signature \
169             FROM audit_checkpoints ORDER BY seq ASC",
170        )
171        .fetch_all(&self.pool)
172        .await?;
173
174        let mut out = Vec::with_capacity(rows.len());
175        for row in rows {
176            let seq: i64 = row.try_get("seq")?;
177            out.push(Checkpoint {
178                seq: u64::try_from(seq).map_err(|e| {
179                    CheckpointSinkError::Query(format!("negative checkpoint seq: {e}"))
180                })?,
181                chain_tip: row.try_get("chain_tip")?,
182                created_at: row.try_get("created_at")?,
183                signature: row.try_get("signature")?,
184            });
185        }
186        Ok(out)
187    }
188}
189
190#[cfg(test)]
191mod tests {
192    use super::*;
193    use crate::audit_chain::{
194        verify_checkpoints, AuditKey, ChainContent, ChainRow, CheckpointOutcome,
195    };
196    use sysknife_types::RiskLevel;
197
198    fn key() -> AuditKey {
199        AuditKey::from_bytes(vec![0x42; 32])
200    }
201
202    /// Build a small intact chain the same way the daemon would.
203    fn build_chain(key: &AuditKey, count: usize) -> Vec<ChainRow> {
204        let mut rows = Vec::with_capacity(count);
205        let mut prev = String::new();
206        for i in 0..count {
207            let seq = (i + 1) as u64;
208            let txid = format!("tx{i}");
209            let content = ChainContent {
210                seq,
211                key_id: "v1",
212                transaction_id: &txid,
213                request_id: "req",
214                request_hash: "hash",
215                action_name: "UpdateSystem",
216                risk_level: RiskLevel::High,
217                summary: "s",
218                approval_id: None,
219                warnings_json: "[]",
220                created_at: "2026-04-24T12:00:00Z",
221            };
222            let hash = key.chain_hash(&content, &prev);
223            rows.push(ChainRow {
224                seq,
225                key_id: "v1".to_string(),
226                transaction_id: txid,
227                request_id: "req".to_string(),
228                request_hash: "hash".to_string(),
229                action_name: "UpdateSystem".to_string(),
230                risk_level: RiskLevel::High,
231                summary: "s".to_string(),
232                approval_id: None,
233                warnings_json: "[]".to_string(),
234                created_at: "2026-04-24T12:00:00Z".to_string(),
235                prev_chain_hash: prev.clone(),
236                chain_hash: hash.clone(),
237            });
238            prev = hash;
239        }
240        rows
241    }
242
243    #[tokio::test]
244    async fn in_memory_append_and_load_round_trip() {
245        let key = key();
246        let rows = build_chain(&key, 3);
247        let sink = InMemoryCheckpointSink::new();
248        let cp = key.sign_checkpoint(3, &rows[2].chain_hash, "2026-04-24T12:00:00Z");
249        sink.append(&cp).await.unwrap();
250        let loaded = sink.load_all().await.unwrap();
251        assert_eq!(loaded, vec![cp]);
252    }
253
254    #[tokio::test]
255    async fn anchored_checkpoints_verify_and_catch_truncation() {
256        let key = key();
257        let full = build_chain(&key, 5);
258        let sink = InMemoryCheckpointSink::new();
259        // Anchor a checkpoint at the tip.
260        let cp = key.sign_checkpoint(5, &full[4].chain_hash, "2026-04-24T12:00:00Z");
261        sink.append(&cp).await.unwrap();
262
263        // Intact: loading the anchored checkpoints verifies against the full chain.
264        let anchored = sink.load_all().await.unwrap();
265        assert_eq!(
266            verify_checkpoints(&key.verifying_key_hex(), &full, &anchored),
267            CheckpointOutcome::Consistent {
268                checkpoints_checked: 1
269            }
270        );
271
272        // Truncated: the local chain is cut to 3, but the anchored tip (seq=5)
273        // can no longer be reproduced -> detected.
274        let truncated = &full[..3];
275        assert!(matches!(
276            verify_checkpoints(&key.verifying_key_hex(), truncated, &anchored),
277            CheckpointOutcome::Truncated {
278                checkpoint_seq: 5,
279                current_max_seq: 3
280            }
281        ));
282    }
283}
sysknife_daemon/checkpoint_sink.rs

sysknife_daemon/
checkpoint_sink.rs
