use async_trait::async_trait;
use sqlx_core::row::Row;
use sqlx_postgres::{PgConnectOptions, PgPool, PgPoolOptions};
use std::str::FromStr;
#[cfg(test)]
use std::sync::Mutex;
use std::time::Duration;
use crate::audit_chain::{verify_checkpoints, ChainRow, Checkpoint, CheckpointOutcome};
#[derive(Debug, thiserror::Error)]
pub enum CheckpointSinkError {
#[error("checkpoint sink connection error: {0}")]
Connect(String),
#[error("checkpoint sink query error: {0}")]
Query(String),
}
impl From<sqlx_core::Error> for CheckpointSinkError {
fn from(e: sqlx_core::Error) -> Self {
CheckpointSinkError::Query(e.to_string())
}
}
#[async_trait]
pub trait CheckpointSink: Send + Sync {
async fn append(&self, checkpoint: &Checkpoint) -> Result<(), CheckpointSinkError>;
async fn load_all(&self) -> Result<Vec<Checkpoint>, CheckpointSinkError>;
}
#[cfg(test)]
#[derive(Debug, Default)]
pub struct InMemoryCheckpointSink {
stored: Mutex<Vec<Checkpoint>>,
}
#[cfg(test)]
impl InMemoryCheckpointSink {
pub fn new() -> Self {
Self::default()
}
}
#[cfg(test)]
#[async_trait]
impl CheckpointSink for InMemoryCheckpointSink {
async fn append(&self, checkpoint: &Checkpoint) -> Result<(), CheckpointSinkError> {
let mut stored = self
.stored
.lock()
.map_err(|e| CheckpointSinkError::Query(format!("lock poisoned: {e}")))?;
stored.push(checkpoint.clone());
Ok(())
}
async fn load_all(&self) -> Result<Vec<Checkpoint>, CheckpointSinkError> {
let stored = self
.stored
.lock()
.map_err(|e| CheckpointSinkError::Query(format!("lock poisoned: {e}")))?;
let mut out = stored.clone();
out.sort_by_key(|c| c.seq);
Ok(out)
}
}
#[derive(Debug)]
pub struct PostgresCheckpointSink {
pool: PgPool,
}
impl PostgresCheckpointSink {
pub async fn connect(url: &str) -> Result<Self, CheckpointSinkError> {
let opts = PgConnectOptions::from_str(url).map_err(|_| {
CheckpointSinkError::Connect("invalid checkpoint database URL".to_string())
})?;
crate::pg_tls::require_tls_for_remote(&opts).map_err(CheckpointSinkError::Connect)?;
let pool = PgPoolOptions::new()
.max_connections(4)
.acquire_timeout(Duration::from_secs(10))
.connect_with(opts)
.await
.map_err(|e| CheckpointSinkError::Connect(e.to_string()))?;
let sink = Self { pool };
sink.initialize().await?;
Ok(sink)
}
async fn initialize(&self) -> Result<(), CheckpointSinkError> {
let row = sqlx_core::query::query(
"SELECT to_regclass('audit_checkpoints') IS NOT NULL AS present",
)
.fetch_one(&self.pool)
.await?;
let present: bool = row.try_get("present")?;
if present {
return Ok(());
}
sqlx_core::query::query(
"CREATE TABLE IF NOT EXISTS audit_checkpoints (\
seq BIGINT NOT NULL, \
chain_tip TEXT NOT NULL, \
created_at TEXT NOT NULL, \
signature TEXT NOT NULL\
)",
)
.execute(&self.pool)
.await?;
Ok(())
}
}
#[async_trait]
impl CheckpointSink for PostgresCheckpointSink {
async fn append(&self, checkpoint: &Checkpoint) -> Result<(), CheckpointSinkError> {
sqlx_core::query::query(
"INSERT INTO audit_checkpoints (seq, chain_tip, created_at, signature) \
VALUES ($1, $2, $3, $4)",
)
.bind(i64::try_from(checkpoint.seq).map_err(|_| {
CheckpointSinkError::Query("checkpoint seq exceeds i64 range".to_string())
})?)
.bind(&checkpoint.chain_tip)
.bind(&checkpoint.created_at)
.bind(&checkpoint.signature)
.execute(&self.pool)
.await?;
Ok(())
}
async fn load_all(&self) -> Result<Vec<Checkpoint>, CheckpointSinkError> {
let rows = sqlx_core::query::query(
"SELECT seq, chain_tip, created_at, signature \
FROM audit_checkpoints ORDER BY seq ASC",
)
.fetch_all(&self.pool)
.await?;
let mut out = Vec::with_capacity(rows.len());
for row in rows {
let seq: i64 = row.try_get("seq")?;
out.push(Checkpoint {
seq: u64::try_from(seq).map_err(|e| {
CheckpointSinkError::Query(format!("negative checkpoint seq: {e}"))
})?,
chain_tip: row.try_get("chain_tip")?,
created_at: row.try_get("created_at")?,
signature: row.try_get("signature")?,
});
}
Ok(out)
}
}
#[derive(Debug, PartialEq, Eq)]
pub enum AnchorOutcome {
Anchored { seq: u64, checkpoints_checked: u64 },
ChainEmpty,
ChainBroken(crate::audit_chain::VerifyOutcome),
ReadBackMissing,
Inconsistent(crate::audit_chain::CheckpointOutcome),
}
pub async fn anchor_once(
key: &crate::audit_chain::AuditKey,
rows: &[crate::audit_chain::ChainRow],
sink: &dyn CheckpointSink,
created_at: &str,
) -> Result<AnchorOutcome, CheckpointSinkError> {
use crate::audit_chain::{verify_chain, verify_checkpoints, CheckpointOutcome, VerifyOutcome};
let Some(tip) = rows.last() else {
return Ok(AnchorOutcome::ChainEmpty);
};
match verify_chain(key, rows) {
VerifyOutcome::Intact { .. } => {}
broken => return Ok(AnchorOutcome::ChainBroken(broken)),
}
let checkpoint = key.sign_checkpoint(tip.seq, &tip.chain_hash, created_at);
sink.append(&checkpoint).await?;
let anchored = sink.load_all().await?;
if !anchored.contains(&checkpoint) {
return Ok(AnchorOutcome::ReadBackMissing);
}
match verify_checkpoints(&key.verifying_key_hex(), rows, &anchored) {
CheckpointOutcome::Consistent {
checkpoints_checked,
} => Ok(AnchorOutcome::Anchored {
seq: checkpoint.seq,
checkpoints_checked,
}),
other => Ok(AnchorOutcome::Inconsistent(other)),
}
}
pub async fn verify_against_anchor(
verifying_key_hex: &str,
rows: &[ChainRow],
sink: &dyn CheckpointSink,
) -> Result<CheckpointOutcome, CheckpointSinkError> {
let anchored = sink.load_all().await?;
if anchored.is_empty() {
return Ok(CheckpointOutcome::CannotVerify {
reason: "a checkpoint anchor is configured but holds no checkpoints, so a \
deleted tail would still go unnoticed; run `sysknife audit checkpoint` \
(periodically — the interval is the window an attacker can rewrite)"
.to_string(),
});
}
Ok(verify_checkpoints(verifying_key_hex, rows, &anchored))
}
pub const DEFAULT_ANCHOR_INTERVAL: Duration = Duration::from_secs(15 * 60);
pub fn spawn_periodic_anchor(
audit: std::sync::Arc<dyn crate::store::AuditStore>,
key: std::sync::Arc<crate::audit_chain::AuditKey>,
sink: std::sync::Arc<dyn CheckpointSink>,
interval: Duration,
) -> tokio::task::JoinHandle<()> {
tokio::spawn(async move {
loop {
tokio::time::sleep(interval).await;
let rows = match audit.fetch_chain_rows().await {
Ok(rows) => rows,
Err(e) => {
eprintln!("[sysknife-daemon] checkpoint anchor: reading chain failed: {e}");
continue;
}
};
let created_at = chrono::Utc::now().to_rfc3339();
match anchor_once(&key, &rows, sink.as_ref(), &created_at).await {
Ok(AnchorOutcome::Anchored {
seq,
checkpoints_checked,
}) => {
eprintln!(
"[sysknife-daemon] anchored checkpoint seq={seq} \
({checkpoints_checked} verified)"
);
}
Ok(AnchorOutcome::ChainEmpty) => {}
Ok(AnchorOutcome::ChainBroken(outcome)) => {
eprintln!(
"[sysknife-daemon] checkpoint anchor REFUSED: the local audit chain \
does not verify ({outcome:?}); not anchoring a tampered chain"
);
}
Ok(AnchorOutcome::ReadBackMissing) => {
eprintln!(
"[sysknife-daemon] checkpoint anchor FAILED: the checkpoint was absent \
on read-back; the sink may be a lagging replica or a different database"
);
}
Ok(AnchorOutcome::Inconsistent(outcome)) => {
eprintln!(
"[sysknife-daemon] checkpoint verification FAILED: {outcome:?} — \
the local chain disagrees with a previously anchored checkpoint"
);
}
Err(e) => {
eprintln!("[sysknife-daemon] checkpoint anchor: sink error: {e}");
}
}
}
})
}
#[cfg(test)]
mod tests {
use super::*;
use crate::audit_chain::{
verify_checkpoints, AuditKey, ChainContent, ChainIdentity, ChainRow, CheckpointOutcome,
CHAIN_VERSION_CURRENT,
};
use sysknife_types::RiskLevel;
fn key() -> AuditKey {
AuditKey::from_bytes(vec![0x42; 32])
}
#[tokio::test]
async fn refuses_a_downgradeable_remote_url_before_connecting() {
let refusal = tokio::time::timeout(
std::time::Duration::from_secs(2),
PostgresCheckpointSink::connect("postgres://u:p@db.example.com:5432/audit"),
)
.await
.expect("the guard must refuse without attempting a connection");
match refusal.expect_err("a remote checkpoint URL without TLS must be refused") {
CheckpointSinkError::Connect(m) => {
assert!(m.contains("sslmode"), "message should name sslmode: {m}")
}
other => panic!("expected a Connect refusal, got {other:?}"),
}
}
pub(super) fn build_chain(key: &AuditKey, count: usize) -> Vec<ChainRow> {
let mut rows = Vec::with_capacity(count);
let mut prev = String::new();
for i in 0..count {
let seq = (i + 1) as u64;
let txid = format!("tx{i}");
let content = ChainContent {
seq,
key_id: "v1",
transaction_id: &txid,
request_id: "req",
request_hash: "hash",
action_name: "UpdateSystem",
risk_level: RiskLevel::High,
summary: "s",
approval_id: None,
warnings_json: "[]",
created_at: "2026-04-24T12:00:00Z",
identity: ChainIdentity::V3 {
caller_role: "dev",
event_tip: "",
caller_principal: "uid:1000",
},
};
let hash = key.chain_hash(&content, &prev);
rows.push(ChainRow {
seq,
key_id: "v1".to_string(),
transaction_id: txid,
request_id: "req".to_string(),
request_hash: "hash".to_string(),
action_name: "UpdateSystem".to_string(),
risk_level: RiskLevel::High,
summary: "s".to_string(),
approval_id: None,
warnings_json: "[]".to_string(),
created_at: "2026-04-24T12:00:00Z".to_string(),
prev_chain_hash: prev.clone(),
chain_hash: hash.clone(),
chain_version: CHAIN_VERSION_CURRENT,
caller_role: Some("dev".to_string()),
event_tip: Some(String::new()),
caller_principal: Some("uid:1000".to_string()),
});
prev = hash;
}
rows
}
#[tokio::test]
async fn in_memory_append_and_load_round_trip() {
let key = key();
let rows = build_chain(&key, 3);
let sink = InMemoryCheckpointSink::new();
let cp = key.sign_checkpoint(3, &rows[2].chain_hash, "2026-04-24T12:00:00Z");
sink.append(&cp).await.unwrap();
let loaded = sink.load_all().await.unwrap();
assert_eq!(loaded, vec![cp]);
}
#[tokio::test]
async fn anchored_checkpoints_verify_and_catch_truncation() {
let key = key();
let full = build_chain(&key, 5);
let sink = InMemoryCheckpointSink::new();
let cp = key.sign_checkpoint(5, &full[4].chain_hash, "2026-04-24T12:00:00Z");
sink.append(&cp).await.unwrap();
let anchored = sink.load_all().await.unwrap();
assert_eq!(
verify_checkpoints(&key.verifying_key_hex(), &full, &anchored),
CheckpointOutcome::Consistent {
checkpoints_checked: 1
}
);
let truncated = &full[..3];
assert!(matches!(
verify_checkpoints(&key.verifying_key_hex(), truncated, &anchored),
CheckpointOutcome::Truncated {
checkpoint_seq: 5,
current_max_seq: 3
}
));
}
}
#[cfg(test)]
mod anchor_tests {
use super::*;
use crate::audit_chain::{AuditKey, ChainContent, ChainRow, CheckpointOutcome, VerifyOutcome};
fn key() -> AuditKey {
AuditKey::from_bytes(vec![0x42; 32])
}
fn reseal(key: &AuditKey, rows: &[ChainRow]) -> Vec<ChainRow> {
let mut out = Vec::with_capacity(rows.len());
let mut prev = String::new();
for row in rows {
let content = ChainContent {
seq: row.seq,
key_id: &row.key_id,
transaction_id: &row.transaction_id,
request_id: &row.request_id,
request_hash: &row.request_hash,
action_name: &row.action_name,
risk_level: row.risk_level,
summary: &row.summary,
approval_id: row.approval_id.as_deref(),
warnings_json: &row.warnings_json,
created_at: &row.created_at,
identity: row.identity().expect("fixture rows are well-formed"),
};
let hash = key.chain_hash(&content, &prev);
let mut cloned = row.clone();
cloned.prev_chain_hash = prev.clone();
cloned.chain_hash = hash.clone();
out.push(cloned);
prev = hash;
}
out
}
fn chain(key: &AuditKey, count: usize) -> Vec<ChainRow> {
super::tests::build_chain(key, count)
}
#[tokio::test]
async fn anchor_once_writes_and_verifies_the_tip() {
let key = key();
let rows = chain(&key, 3);
let sink = InMemoryCheckpointSink::new();
let outcome = anchor_once(&key, &rows, &sink, "2026-04-24T12:00:00Z")
.await
.unwrap();
assert_eq!(
outcome,
AnchorOutcome::Anchored {
seq: 3,
checkpoints_checked: 1
}
);
assert_eq!(sink.load_all().await.unwrap().len(), 1);
}
#[tokio::test]
async fn anchor_once_refuses_to_launder_a_tampered_chain() {
let key = key();
let mut rows = chain(&key, 3);
rows[1].summary = "tampered".to_string();
let sink = InMemoryCheckpointSink::new();
let outcome = anchor_once(&key, &rows, &sink, "2026-04-24T12:00:00Z")
.await
.unwrap();
assert!(
matches!(
outcome,
AnchorOutcome::ChainBroken(VerifyOutcome::Broken { .. })
),
"expected a refusal, got {outcome:?}"
);
assert!(
sink.load_all().await.unwrap().is_empty(),
"nothing may be anchored when the chain does not verify"
);
}
#[tokio::test]
async fn anchor_once_reports_an_empty_chain_rather_than_anchoring_nothing() {
let key = key();
let sink = InMemoryCheckpointSink::new();
let outcome = anchor_once(&key, &[], &sink, "2026-04-24T12:00:00Z")
.await
.unwrap();
assert_eq!(outcome, AnchorOutcome::ChainEmpty);
assert!(sink.load_all().await.unwrap().is_empty());
}
#[derive(Default)]
struct BlackHoleSink;
#[async_trait]
impl CheckpointSink for BlackHoleSink {
async fn append(
&self,
_checkpoint: &crate::audit_chain::Checkpoint,
) -> Result<(), CheckpointSinkError> {
Ok(())
}
async fn load_all(
&self,
) -> Result<Vec<crate::audit_chain::Checkpoint>, CheckpointSinkError> {
Ok(Vec::new())
}
}
#[tokio::test]
async fn anchor_once_detects_a_write_that_did_not_land() {
let key = key();
let rows = chain(&key, 2);
let outcome = anchor_once(&key, &rows, &BlackHoleSink, "2026-04-24T12:00:00Z")
.await
.unwrap();
assert_eq!(outcome, AnchorOutcome::ReadBackMissing);
}
#[tokio::test]
async fn a_previously_anchored_checkpoint_catches_a_key_holder_rewrite() {
let key = key();
let original = chain(&key, 5);
let sink = InMemoryCheckpointSink::new();
anchor_once(&key, &original, &sink, "2026-04-24T12:00:00Z")
.await
.unwrap();
let anchored = sink.load_all().await.unwrap();
let mut tampered = original.clone();
tampered[3].summary = "quietly rewritten".to_string();
let tampered = reseal(&key, &tampered);
assert!(
matches!(
crate::audit_chain::verify_chain(&key, &tampered),
VerifyOutcome::Intact { .. }
),
"the forged chain must verify — otherwise this test proves nothing \
beyond what a plain signature check already catches"
);
let outcome =
crate::audit_chain::verify_checkpoints(&key.verifying_key_hex(), &tampered, &anchored);
assert!(
matches!(outcome, CheckpointOutcome::TipMismatch { .. }),
"the anchored tip must expose the rewrite, got {outcome:?}"
);
}
#[tokio::test]
async fn verifying_against_the_anchor_catches_what_the_chain_alone_cannot() {
let key = key();
let full = chain(&key, 5);
let sink = InMemoryCheckpointSink::new();
anchor_once(&key, &full, &sink, "2026-04-24T12:00:00Z")
.await
.unwrap();
let outcome = verify_against_anchor(&key.verifying_key_hex(), &full, &sink)
.await
.unwrap();
assert!(
matches!(outcome, CheckpointOutcome::Consistent { .. }),
"an untouched chain must verify against its anchor, got {outcome:?}"
);
let truncated = &full[..3];
let outcome = verify_against_anchor(&key.verifying_key_hex(), truncated, &sink)
.await
.unwrap();
assert!(
matches!(outcome, CheckpointOutcome::Truncated { .. }),
"a deleted tail must be reported, got {outcome:?}"
);
}
#[tokio::test]
async fn an_anchor_with_no_checkpoints_is_not_reported_as_consistent() {
let key = key();
let full = chain(&key, 5);
let sink = InMemoryCheckpointSink::new();
let outcome = verify_against_anchor(&key.verifying_key_hex(), &full, &sink)
.await
.unwrap();
assert!(
matches!(outcome, CheckpointOutcome::CannotVerify { .. }),
"an empty anchor proves nothing and must not read as success, got {outcome:?}"
);
}
#[tokio::test]
async fn verify_chain_alone_cannot_detect_tail_truncation() {
let key = key();
let full = chain(&key, 5);
let truncated = &full[..3];
assert!(
matches!(
crate::audit_chain::verify_chain(&key, truncated),
VerifyOutcome::Intact {
rows_checked: 3,
..
}
),
"a truncated chain still verifies on its own — this is the gap"
);
let sink = InMemoryCheckpointSink::new();
anchor_once(&key, &full, &sink, "2026-04-24T12:00:00Z")
.await
.unwrap();
let anchored = sink.load_all().await.unwrap();
assert!(
matches!(
crate::audit_chain::verify_checkpoints(
&key.verifying_key_hex(),
truncated,
&anchored
),
CheckpointOutcome::Truncated { .. }
),
"the anchor is what turns an undetectable truncation into a detected one"
);
}
}