1use crate::audit_chain::{
2 self, AuditEventKind, AuditKey, ChainContent, ChainIdentity, ChainRow, EventContent,
3 EventIdentity, EventRow, VerifyOutcome, CURRENT_KEY_ID,
4};
5use crate::audit_watermark::emit_chain_tip_watermark;
6use crate::auth::CallerPrincipal;
7use rusqlite::{params, Connection, TransactionBehavior};
8use serde::{de::DeserializeOwned, Deserialize, Serialize};
9use std::path::{Path, PathBuf};
10use std::sync::Arc;
11use subtle::ConstantTimeEq;
12use sysknife_types::{CallerRole, JobState, PreviewEnvelope, RiskLevel, TransactionRecord};
13use uuid::Uuid;
14
15pub(crate) const APPROVAL_RECEIPT_TTL_MINUTES: i64 = 15;
26
27#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
35pub struct JobHistoryEntry {
36 pub transaction_id: String,
37 pub action_name: String,
38 pub risk_level: RiskLevel,
39 pub status: JobState,
40 pub summary: String,
41 pub created_at: String,
42}
43
44#[derive(Clone, Copy, Debug, PartialEq, Eq)]
52pub enum HistorySince {
53 Hours(u32),
55 Cutoff(chrono::DateTime<chrono::Utc>),
57}
58
59impl HistorySince {
60 pub fn cutoff_rfc3339(at: &chrono::DateTime<chrono::Utc>) -> String {
63 at.to_rfc3339_opts(chrono::SecondsFormat::Millis, true)
64 }
65}
66
67#[derive(Clone, Debug, PartialEq, Eq)]
73pub struct NewTransaction {
74 pub request_id: String,
75 pub request_hash: String,
76 pub action_name: String,
77 pub risk_level: RiskLevel,
78 pub summary: String,
98 pub warnings: Vec<String>,
99 pub caller_role: CallerRole,
108 pub caller_principal: CallerPrincipal,
116}
117
118struct SqliteMigration {
127 version: i64,
128 name: &'static str,
129 sql: &'static str,
130}
131
132const SQLITE_MIGRATIONS: &[SqliteMigration] = &[
150 SqliteMigration {
151 version: 1,
152 name: "initial_audit_schema",
153 sql: r#"
154 CREATE TABLE IF NOT EXISTS transactions (
155 transaction_id TEXT PRIMARY KEY,
156 request_id TEXT NOT NULL,
157 request_hash TEXT NOT NULL,
158 action_name TEXT NOT NULL,
159 risk_level TEXT NOT NULL,
160 status TEXT NOT NULL,
161 approval_id TEXT,
162 summary TEXT NOT NULL,
163 warnings_json TEXT NOT NULL,
164 created_at TEXT NOT NULL DEFAULT (datetime('now')),
165 seq INTEGER NOT NULL UNIQUE,
166 key_id TEXT NOT NULL,
167 chain_hash TEXT NOT NULL,
168 prev_chain_hash TEXT NOT NULL DEFAULT ''
169 );
170
171 CREATE TABLE IF NOT EXISTS transaction_previews (
172 transaction_id TEXT PRIMARY KEY,
173 preview_json TEXT NOT NULL
174 );
175
176 CREATE TABLE IF NOT EXISTS transaction_approvals (
177 transaction_id TEXT PRIMARY KEY,
178 receipt_digest TEXT NOT NULL,
179 approved_at TEXT NOT NULL DEFAULT (datetime('now')),
180 consumed_at TEXT
181 );
182
183 CREATE INDEX IF NOT EXISTS transactions_seq_idx ON transactions(seq);
184 "#,
185 },
186 SqliteMigration {
195 version: 2,
196 name: "caller_identity_and_approval_events",
197 sql: r#"
198 ALTER TABLE transactions ADD COLUMN chain_version INTEGER NOT NULL DEFAULT 1;
199 ALTER TABLE transactions ADD COLUMN caller_role TEXT;
200 ALTER TABLE transactions ADD COLUMN event_tip TEXT;
201
202 CREATE TABLE IF NOT EXISTS audit_events (
203 seq INTEGER PRIMARY KEY,
204 key_id TEXT NOT NULL,
205 kind TEXT NOT NULL,
206 transaction_id TEXT NOT NULL,
207 receipt_digest TEXT NOT NULL,
208 created_at TEXT NOT NULL,
209 chain_hash TEXT NOT NULL,
210 prev_chain_hash TEXT NOT NULL DEFAULT ''
211 );
212
213 CREATE INDEX IF NOT EXISTS audit_events_transaction_idx
214 ON audit_events(transaction_id);
215 "#,
216 },
217 SqliteMigration {
222 version: 3,
223 name: "caller_principal",
224 sql: r#"
225 ALTER TABLE transactions ADD COLUMN caller_principal TEXT;
226 "#,
227 },
228 SqliteMigration {
237 version: 4,
238 name: "event_approver_identity",
239 sql: r#"
240 ALTER TABLE audit_events ADD COLUMN chain_version INTEGER NOT NULL DEFAULT 1;
241 ALTER TABLE audit_events ADD COLUMN caller_principal TEXT;
242 "#,
243 },
244];
245
246use crate::store::CHAIN_ROW_COLUMNS;
250
251fn chain_row_from_sqlite(row: &rusqlite::Row<'_>) -> rusqlite::Result<ChainRow> {
252 Ok(ChainRow {
253 seq: row.get::<_, i64>(0)? as u64,
254 key_id: row.get(1)?,
255 transaction_id: row.get(2)?,
256 request_id: row.get(3)?,
257 request_hash: row.get(4)?,
258 action_name: row.get(5)?,
259 risk_level: deserialize_field(&row.get::<_, String>(6)?).map_err(|e| {
260 rusqlite::Error::FromSqlConversionFailure(6, rusqlite::types::Type::Text, Box::new(e))
261 })?,
262 summary: row.get(7)?,
263 approval_id: row.get(8)?,
264 warnings_json: row.get(9)?,
265 created_at: row.get(10)?,
266 prev_chain_hash: row.get(11)?,
267 chain_hash: row.get(12)?,
268 chain_version: row.get::<_, i64>(13)? as u32,
269 caller_role: row.get(14)?,
270 event_tip: row.get(15)?,
271 caller_principal: row.get(16)?,
272 })
273}
274
275fn event_row_from_sqlite(row: &rusqlite::Row<'_>) -> rusqlite::Result<EventRow> {
276 Ok(EventRow {
277 seq: row.get::<_, i64>(0)? as u64,
278 key_id: row.get(1)?,
279 kind: row.get(2)?,
280 transaction_id: row.get(3)?,
281 receipt_digest: row.get(4)?,
282 created_at: row.get(5)?,
283 prev_chain_hash: row.get(6)?,
284 chain_hash: row.get(7)?,
285 chain_version: row.get::<_, i64>(8)? as u32,
286 caller_principal: row.get(9)?,
287 })
288}
289
290fn query_optional<T>(
296 conn: &Connection,
297 sql: &str,
298 params: impl rusqlite::Params,
299 f: impl FnOnce(&rusqlite::Row<'_>) -> rusqlite::Result<T>,
300) -> rusqlite::Result<Option<T>> {
301 match conn.query_row(sql, params, f) {
302 Ok(v) => Ok(Some(v)),
303 Err(rusqlite::Error::QueryReturnedNoRows) => Ok(None),
304 Err(e) => Err(e),
305 }
306}
307
308#[derive(Clone, Debug)]
309pub struct TransactionStore {
310 path: PathBuf,
311 audit_key: Option<Arc<AuditKey>>,
314}
315
316#[derive(Clone, Debug, PartialEq)]
317pub struct RecordedPreviewedTransaction {
318 pub transaction: TransactionRecord,
319 pub preview: PreviewEnvelope,
320}
321
322struct InsertedTransaction {
323 record: TransactionRecord,
324 seq: u64,
325 chain_hash: String,
326}
327
328#[derive(Debug, thiserror::Error)]
329pub enum TransactionStoreError {
330 #[error("io error: {0}")]
331 Io(#[from] std::io::Error),
332
333 #[error("sqlite error: {0}")]
334 Sqlite(#[from] rusqlite::Error),
335
336 #[error("json error: {0}")]
337 Json(#[from] serde_json::Error),
338
339 #[error("database invariant violation: {0}")]
340 DatabaseInvariant(String),
341
342 #[error("transaction not found: {0}")]
343 NotFound(String),
344
345 #[error("invalid transition from {from:?} to {to:?}")]
346 InvalidTransition { from: JobState, to: JobState },
347
348 #[error("audit chain misconfiguration: {0}")]
349 AuditChainMissing(&'static str),
350
351 #[error("transaction {0} status changed concurrently; retry")]
359 ConcurrentStatusChange(String),
360}
361
362impl TransactionStore {
363 pub fn open_read_only(path: impl AsRef<Path>) -> Result<Self, TransactionStoreError> {
367 let path = path.as_ref().to_path_buf();
368 if let Some(parent) = path.parent() {
369 ensure_private_dir(parent)?;
370 }
371
372 let store = Self {
373 path,
374 audit_key: None,
375 };
376 store.initialize()?;
377 Ok(store)
378 }
379
380 pub fn open(path: impl AsRef<Path>) -> Result<Self, TransactionStoreError> {
387 let db_path = path.as_ref();
388 let key_path = crate::audit_chain::resolve_audit_key_path(db_path);
389 let key = AuditKey::load_or_generate(&key_path).map_err(|e| {
390 TransactionStoreError::Io(std::io::Error::new(
391 std::io::ErrorKind::InvalidData,
392 format!("audit key load failed: {e}"),
393 ))
394 })?;
395 Self::open_with_key(path, Arc::new(key))
396 }
397
398 pub fn open_with_key(
401 path: impl AsRef<Path>,
402 audit_key: Arc<AuditKey>,
403 ) -> Result<Self, TransactionStoreError> {
404 let path = path.as_ref().to_path_buf();
405 if let Some(parent) = path.parent() {
406 ensure_private_dir(parent)?;
407 }
408
409 let store = Self {
410 path,
411 audit_key: Some(audit_key),
412 };
413 store.initialize()?;
414 restrict_db_file_mode(&store.path);
415 Ok(store)
416 }
417
418 fn record_with(
427 &self,
428 transaction: NewTransaction,
429 in_tx: impl FnOnce(&Connection, &InsertedTransaction) -> Result<(), TransactionStoreError>,
430 ) -> Result<InsertedTransaction, TransactionStoreError> {
431 let key = self
432 .audit_key
433 .as_ref()
434 .ok_or(TransactionStoreError::AuditChainMissing(
435 "this TransactionStore was opened read-only; cannot record",
436 ))?;
437 let mut conn = self.connection()?;
438 let tx = conn.transaction_with_behavior(TransactionBehavior::Immediate)?;
443 let transaction_id = Uuid::new_v4().to_string();
444 let inserted = Self::insert_transaction(&tx, key, &transaction_id, transaction)?;
445 in_tx(&tx, &inserted)?;
446 tx.commit()?;
447 emit_chain_tip_watermark(inserted.seq, &inserted.chain_hash);
448 Ok(inserted)
449 }
450
451 pub fn record(
452 &self,
453 transaction: NewTransaction,
454 ) -> Result<TransactionRecord, TransactionStoreError> {
455 let inserted = self.record_with(transaction, |_, _| Ok(()))?;
456 Ok(inserted.record)
457 }
458
459 pub fn record_previewed(
460 &self,
461 transaction: NewTransaction,
462 preview: PreviewEnvelope,
463 ) -> Result<RecordedPreviewedTransaction, TransactionStoreError> {
464 let inserted = self.record_with(transaction, |conn, inserted| {
465 Self::insert_preview(conn, &inserted.record.transaction_id, &preview)
466 })?;
467 Ok(RecordedPreviewedTransaction {
468 transaction: inserted.record,
469 preview,
470 })
471 }
472
473 pub fn get(
474 &self,
475 transaction_id: &str,
476 ) -> Result<Option<TransactionRecord>, TransactionStoreError> {
477 let conn = self.connection()?;
478 let mut stmt = conn.prepare(
479 "SELECT
480 transaction_id,
481 request_id,
482 request_hash,
483 action_name,
484 risk_level,
485 status,
486 approval_id,
487 summary,
488 warnings_json
489 FROM transactions
490 WHERE transaction_id = ?1",
491 )?;
492 let mut rows = stmt.query(params![transaction_id])?;
493 if let Some(row) = rows.next()? {
494 Ok(Some(row_to_record(row)?))
495 } else {
496 Ok(None)
497 }
498 }
499
500 pub fn get_preview(
501 &self,
502 transaction_id: &str,
503 ) -> Result<Option<PreviewEnvelope>, TransactionStoreError> {
504 let conn = self.connection()?;
505 let mut stmt = conn.prepare(
506 "SELECT preview_json
507 FROM transaction_previews
508 WHERE transaction_id = ?1",
509 )?;
510 let mut rows = stmt.query(params![transaction_id])?;
511 if let Some(row) = rows.next()? {
512 let preview_json: String = row.get(0)?;
513 Ok(Some(serde_json::from_str(&preview_json)?))
514 } else {
515 Ok(None)
516 }
517 }
518
519 pub fn update_status(
520 &self,
521 transaction_id: &str,
522 new_status: JobState,
523 ) -> Result<(), TransactionStoreError> {
524 let conn = self.connection()?;
525
526 let current_status: String = conn
528 .query_row(
529 "SELECT status FROM transactions WHERE transaction_id = ?1",
530 params![transaction_id],
531 |row| row.get(0),
532 )
533 .map_err(|e| match e {
534 rusqlite::Error::QueryReturnedNoRows => {
535 TransactionStoreError::NotFound(transaction_id.to_string())
536 }
537 other => TransactionStoreError::Sqlite(other),
538 })?;
539
540 let current: JobState = deserialize_field(¤t_status)?;
541 if !crate::jobs::allowed_transition(¤t, &new_status) {
542 return Err(TransactionStoreError::InvalidTransition {
543 from: current,
544 to: new_status,
545 });
546 }
547
548 let mut conn = conn;
565 let tx = conn.transaction()?;
566 let rows_affected = tx.execute(
567 "UPDATE transactions SET status = ?1 WHERE transaction_id = ?2 AND status = ?3",
568 params![
569 serialize_field(&new_status)?,
570 transaction_id,
571 current_status
572 ],
573 )?;
574 if rows_affected == 0 {
575 return Err(TransactionStoreError::ConcurrentStatusChange(
576 transaction_id.to_string(),
577 ));
578 }
579 let Some(key) = self.audit_key.as_ref() else {
585 return Err(TransactionStoreError::AuditChainMissing(
586 "update_status has no signing key, so the outcome could not be chained",
587 ));
588 };
589 Self::append_event(
590 &tx,
591 key,
592 AuditEventKind::from(new_status),
593 transaction_id,
594 "",
595 EventIdentity::LegacyV1,
599 )?;
600 tx.commit()?;
601 Ok(())
602 }
603
604 pub fn status_matches_chain(
617 &self,
618 ) -> Result<audit_chain::StatusOutcome, TransactionStoreError> {
619 use crate::store::{APPROVAL_CONSUMED_SQL, NEWEST_STATUS_EVENT_SQL};
620 let conn = self.connection()?;
621 let mut stmt = conn.prepare(&format!(
622 "SELECT t.transaction_id, t.status, {NEWEST_STATUS_EVENT_SQL}, \
623 {APPROVAL_CONSUMED_SQL} FROM transactions t"
624 ))?;
625 let rows = stmt.query_map([], |row| {
626 Ok((
627 row.get::<_, String>(0)?,
628 row.get::<_, String>(1)?,
629 row.get::<_, Option<String>>(2)?,
630 row.get::<_, bool>(3)?,
631 ))
632 })?;
633 let mut evidence = Vec::new();
634 for row in rows {
635 let (transaction_id, status_json, newest_status_event, approval_consumed) = row?;
636 evidence.push(audit_chain::StatusEvidence {
637 transaction_id,
638 status: deserialize_field(&status_json)?,
639 newest_status_event,
640 approval_consumed,
641 predates_status_events: false,
644 });
645 }
646 Ok(audit_chain::status_outcome(evidence))
647 }
648
649 pub fn approve_transaction(
654 &self,
655 transaction_id: &str,
656 approver: CallerPrincipal,
657 ) -> Result<Option<String>, TransactionStoreError> {
658 let key = self
659 .audit_key
660 .as_ref()
661 .ok_or(TransactionStoreError::AuditChainMissing(
662 "this TransactionStore was opened read-only; cannot approve",
663 ))?;
664 let approver_signed = approver.as_signed_str();
667 let Some(record) = self.get(transaction_id)? else {
668 return Ok(None);
669 };
670 let receipt = key.approval_receipt(transaction_id, &record.request_hash);
671 let receipt_digest = audit_chain::approval_receipt_digest(&receipt);
672 let Some(committed_digest) = record.approval_id.as_deref() else {
673 return Err(TransactionStoreError::DatabaseInvariant(format!(
674 "transaction {transaction_id} has no signed approval commitment"
675 )));
676 };
677 if !bool::from(receipt_digest.as_bytes().ct_eq(committed_digest.as_bytes())) {
678 return Err(TransactionStoreError::DatabaseInvariant(format!(
679 "transaction {transaction_id} approval commitment does not match its signed preview"
680 )));
681 }
682
683 let mut conn = self.connection()?;
686 let tx = conn.transaction_with_behavior(TransactionBehavior::Immediate)?;
687 let queued_json = serialize_field(&JobState::Queued)?;
688 let rows_affected = tx.execute(
689 &format!(
690 "INSERT INTO transaction_approvals (transaction_id, receipt_digest) \
691 SELECT transaction_id, ?1 FROM transactions \
692 WHERE transaction_id = ?2 \
693 AND status = ?3 \
694 AND julianday(created_at) > julianday('now', '-{APPROVAL_RECEIPT_TTL_MINUTES} minutes') \
695 AND NOT EXISTS ( \
696 SELECT 1 FROM transaction_approvals WHERE transaction_id = ?2 \
697 )"
698 ),
699 params![receipt_digest, transaction_id, queued_json],
700 )?;
701 if rows_affected > 0 {
702 Self::append_event(
703 &tx,
704 key,
705 AuditEventKind::ApprovalGranted,
706 transaction_id,
707 &receipt_digest,
708 EventIdentity::V2 {
709 caller_principal: &approver_signed,
710 },
711 )?;
712 }
713 tx.commit()?;
714 Ok((rows_affected > 0).then_some(receipt))
715 }
716
717 pub fn revoke_unconsumed_approval(
723 &self,
724 transaction_id: &str,
725 revoker: CallerPrincipal,
726 ) -> Result<bool, TransactionStoreError> {
727 let key = self
728 .audit_key
729 .as_ref()
730 .ok_or(TransactionStoreError::AuditChainMissing(
731 "this TransactionStore was opened read-only; cannot revoke",
732 ))?;
733 let mut conn = self.connection()?;
734 let tx = conn.transaction_with_behavior(TransactionBehavior::Immediate)?;
735 let revoked = Self::revoke_unconsumed_approval_in_tx(
736 &tx,
737 key,
738 transaction_id,
739 &revoker.as_signed_str(),
740 )?;
741 tx.commit()?;
742 Ok(revoked)
743 }
744
745 fn revoke_unconsumed_approval_in_tx(
746 conn: &Connection,
747 key: &AuditKey,
748 transaction_id: &str,
749 revoker_signed: &str,
750 ) -> Result<bool, TransactionStoreError> {
751 let digest: Option<String> = query_optional(
754 conn,
755 "SELECT receipt_digest FROM transaction_approvals \
756 WHERE transaction_id = ?1 AND consumed_at IS NULL",
757 params![transaction_id],
758 |row| row.get(0),
759 )?;
760 let rows_affected = conn.execute(
761 "DELETE FROM transaction_approvals \
762 WHERE transaction_id = ?1 AND consumed_at IS NULL",
763 params![transaction_id],
764 )?;
765 if rows_affected > 0 {
766 let digest = digest.ok_or_else(|| {
767 TransactionStoreError::DatabaseInvariant(format!(
768 "revoked an approval for {transaction_id} that had no receipt digest"
769 ))
770 })?;
771 Self::append_event(
772 conn,
773 key,
774 AuditEventKind::ApprovalRevoked,
775 transaction_id,
776 &digest,
777 EventIdentity::V2 {
778 caller_principal: revoker_signed,
779 },
780 )?;
781 }
782 Ok(rows_affected > 0)
783 }
784
785 pub fn claim_approved_for_execution(
791 &self,
792 transaction_id: &str,
793 receipt_digest: &str,
794 executor: CallerPrincipal,
795 ) -> Result<bool, TransactionStoreError> {
796 let key = self
797 .audit_key
798 .as_ref()
799 .ok_or(TransactionStoreError::AuditChainMissing(
800 "this TransactionStore was opened read-only; cannot claim",
801 ))?;
802 let mut conn = self.connection()?;
803 let tx = conn.transaction_with_behavior(TransactionBehavior::Immediate)?;
804 let queued_json = serialize_field(&JobState::Queued)?;
805 let running_json = serialize_field(&JobState::Running)?;
806 let rows_affected = tx.execute(
807 &format!(
808 "UPDATE transactions SET status = ?1 \
809 WHERE transaction_id = ?2 \
810 AND status = ?3 \
811 AND julianday(created_at) > julianday('now', '-{APPROVAL_RECEIPT_TTL_MINUTES} minutes') \
812 AND EXISTS ( \
813 SELECT 1 FROM transaction_approvals \
814 WHERE transaction_id = ?2 \
815 AND receipt_digest = ?4 \
816 AND consumed_at IS NULL \
817 )"
818 ),
819 params![running_json, transaction_id, queued_json, receipt_digest],
820 )?;
821 if rows_affected > 0 {
822 tx.execute(
823 "UPDATE transaction_approvals \
824 SET consumed_at = datetime('now') \
825 WHERE transaction_id = ?1 AND consumed_at IS NULL",
826 params![transaction_id],
827 )?;
828 Self::append_event(
829 &tx,
830 key,
831 AuditEventKind::ApprovalConsumed,
832 transaction_id,
833 receipt_digest,
834 EventIdentity::V2 {
835 caller_principal: &executor.as_signed_str(),
836 },
837 )?;
838 Self::append_event(
839 &tx,
840 key,
841 AuditEventKind::StatusRunning,
842 transaction_id,
843 "",
844 EventIdentity::LegacyV1,
845 )?;
846 }
847 tx.commit()?;
848 Ok(rows_affected > 0)
849 }
850
851 pub fn cleanup_stale_queued(&self) -> Result<u64, TransactionStoreError> {
866 let key = self
867 .audit_key
868 .as_ref()
869 .ok_or(TransactionStoreError::AuditChainMissing(
870 "this TransactionStore was opened read-only; cannot clean up",
871 ))?;
872 let mut conn = self.connection()?;
873 let tx = conn.transaction_with_behavior(TransactionBehavior::Immediate)?;
874 let canceled_json = serialize_field(&JobState::Canceled)?;
875 let queued_json = serialize_field(&JobState::Queued)?;
876 let stale_ids: Vec<String> = {
877 let mut statement = tx.prepare(&format!(
878 "SELECT transaction_id FROM transactions \
879 WHERE status = ?1 \
880 AND julianday(created_at) <= julianday('now', '-{APPROVAL_RECEIPT_TTL_MINUTES} minutes')"
881 ))?;
882 let stale_ids = statement
883 .query_map(params![queued_json], |row| row.get(0))?
884 .collect::<Result<_, _>>()?;
885 stale_ids
886 };
887 let mut canceled = 0;
888 for transaction_id in stale_ids {
889 let rows_affected = tx.execute(
890 "UPDATE transactions SET status = ?1 \
891 WHERE transaction_id = ?2 AND status = ?3",
892 params![canceled_json, transaction_id, queued_json],
893 )?;
894 if rows_affected > 0 {
895 Self::revoke_unconsumed_approval_in_tx(
899 &tx,
900 key,
901 &transaction_id,
902 &CallerPrincipal::Unattributed.as_signed_str(),
903 )?;
904 Self::append_event(
905 &tx,
906 key,
907 AuditEventKind::StatusCanceled,
908 &transaction_id,
909 "",
910 EventIdentity::LegacyV1,
911 )?;
912 canceled += rows_affected;
913 }
914 }
915 tx.commit()?;
916 Ok(canceled as u64)
917 }
918
919 pub fn cancel_queued(
928 &self,
929 transaction_id: &str,
930 canceller: CallerPrincipal,
931 ) -> Result<bool, TransactionStoreError> {
932 let key = self
933 .audit_key
934 .as_ref()
935 .ok_or(TransactionStoreError::AuditChainMissing(
936 "this TransactionStore was opened read-only; cannot cancel",
937 ))?;
938 let mut conn = self.connection()?;
939 let tx = conn.transaction_with_behavior(TransactionBehavior::Immediate)?;
940 let canceled_json = serialize_field(&JobState::Canceled)?;
941 let queued_json = serialize_field(&JobState::Queued)?;
942 let rows_affected = tx.execute(
943 "UPDATE transactions SET status = ?1 \
944 WHERE transaction_id = ?2 AND status = ?3",
945 params![canceled_json, transaction_id, queued_json],
946 )?;
947 if rows_affected > 0 {
948 Self::revoke_unconsumed_approval_in_tx(
953 &tx,
954 key,
955 transaction_id,
956 &canceller.as_signed_str(),
957 )?;
958 Self::append_event(
959 &tx,
960 key,
961 AuditEventKind::StatusCanceled,
962 transaction_id,
963 "",
964 EventIdentity::LegacyV1,
965 )?;
966 }
967 tx.commit()?;
968 Ok(rows_affected > 0)
969 }
970
971 pub fn list_transactions(
982 &self,
983 limit: u32,
984 status_filter: Option<&str>,
985 action_filter: Option<&str>,
986 since: Option<HistorySince>,
987 ) -> Result<Vec<TransactionRecord>, TransactionStoreError> {
988 let conn = self.connection()?;
989 let (filter_sql, param_values) =
990 Self::build_history_filter(limit, status_filter, action_filter, since)?;
991 let sql = format!(
992 "SELECT transaction_id, request_id, request_hash, action_name, \
993 risk_level, status, approval_id, summary, warnings_json \
994 FROM transactions WHERE 1=1{filter_sql}"
995 );
996 let params_ref: Vec<&dyn rusqlite::types::ToSql> =
997 param_values.iter().map(|b| b.as_ref()).collect();
998
999 let mut stmt = conn.prepare(&sql)?;
1000 let rows = stmt.query_map(params_ref.as_slice(), |row| Ok(row_to_record(row)))?;
1001
1002 let mut results = Vec::new();
1003 for row in rows {
1004 results.push(row??);
1005 }
1006 Ok(results)
1007 }
1008
1009 pub fn list_history(
1014 &self,
1015 limit: u32,
1016 status_filter: Option<&str>,
1017 action_filter: Option<&str>,
1018 since: Option<HistorySince>,
1019 ) -> Result<Vec<JobHistoryEntry>, TransactionStoreError> {
1020 let conn = self.connection()?;
1021 let (filter_sql, param_values) =
1022 Self::build_history_filter(limit, status_filter, action_filter, since)?;
1023 let sql = format!(
1024 "SELECT transaction_id, action_name, risk_level, status, summary, created_at \
1025 FROM transactions WHERE 1=1{filter_sql}"
1026 );
1027 let params_ref: Vec<&dyn rusqlite::types::ToSql> =
1028 param_values.iter().map(|b| b.as_ref()).collect();
1029
1030 let mut stmt = conn.prepare(&sql)?;
1031 let rows = stmt.query_map(params_ref.as_slice(), |row| {
1032 Ok((|| {
1033 Ok::<JobHistoryEntry, TransactionStoreError>(JobHistoryEntry {
1034 transaction_id: row.get(0)?,
1035 action_name: row.get(1)?,
1036 risk_level: deserialize_field(&row.get::<_, String>(2)?)?,
1037 status: deserialize_field(&row.get::<_, String>(3)?)?,
1038 summary: row.get(4)?,
1039 created_at: row.get(5)?,
1040 })
1041 })())
1042 })?;
1043
1044 let mut results = Vec::new();
1045 for row in rows {
1046 results.push(row??);
1047 }
1048 Ok(results)
1049 }
1050
1051 fn build_history_filter(
1056 limit: u32,
1057 status_filter: Option<&str>,
1058 action_filter: Option<&str>,
1059 since: Option<HistorySince>,
1060 ) -> Result<(String, Vec<Box<dyn rusqlite::types::ToSql>>), TransactionStoreError> {
1061 let mut sql = String::new();
1062 let mut param_values: Vec<Box<dyn rusqlite::types::ToSql>> = Vec::new();
1063
1064 if let Some(status) = status_filter {
1065 let validated: JobState = deserialize_field(&format!("\"{status}\""))?;
1069 let status_json = serialize_field(&validated)?;
1070 sql.push_str(" AND status = ?");
1071 param_values.push(Box::new(status_json));
1072 }
1073
1074 if let Some(action) = action_filter {
1075 sql.push_str(" AND action_name = ?");
1076 param_values.push(Box::new(action.to_string()));
1077 }
1078
1079 match since {
1082 Some(HistorySince::Hours(hours)) => {
1083 sql.push_str(" AND julianday(created_at) > julianday('now', '-' || ? || ' hours')");
1084 param_values.push(Box::new(hours));
1085 }
1086 Some(HistorySince::Cutoff(at)) => {
1087 sql.push_str(" AND julianday(created_at) >= julianday(?)");
1088 param_values.push(Box::new(HistorySince::cutoff_rfc3339(&at)));
1089 }
1090 None => {}
1091 }
1092
1093 sql.push_str(" ORDER BY seq DESC LIMIT ?");
1094 param_values.push(Box::new(limit.min(100)));
1095
1096 Ok((sql, param_values))
1097 }
1098
1099 fn connection(&self) -> Result<Connection, TransactionStoreError> {
1100 let conn = Connection::open(&self.path)?;
1101 conn.pragma_update(None, "journal_mode", "WAL")?;
1112 conn.pragma_update(None, "synchronous", "NORMAL")?;
1113 conn.pragma_update(None, "foreign_keys", "ON")?;
1114 conn.busy_timeout(std::time::Duration::from_secs(5))?;
1115 Ok(conn)
1116 }
1117
1118 fn initialize(&self) -> Result<(), TransactionStoreError> {
1119 let mut conn = self.connection()?;
1120 let tx = conn.transaction_with_behavior(TransactionBehavior::Immediate)?;
1121 tx.execute_batch(
1122 "CREATE TABLE IF NOT EXISTS schema_migrations (\
1123 version INTEGER PRIMARY KEY,\
1124 name TEXT NOT NULL,\
1125 applied_at TEXT NOT NULL DEFAULT (datetime('now'))\
1126 );",
1127 )?;
1128 let current: i64 = tx.query_row(
1129 "SELECT COALESCE(MAX(version), 0) FROM schema_migrations",
1130 [],
1131 |row| row.get(0),
1132 )?;
1133 let latest = SQLITE_MIGRATIONS.last().map_or(0, |m| m.version);
1134 if current > latest {
1135 return Err(TransactionStoreError::DatabaseInvariant(format!(
1136 "sqlite schema version {current} is newer than this binary supports ({latest})"
1137 )));
1138 }
1139 for migration in SQLITE_MIGRATIONS.iter().filter(|m| m.version > current) {
1140 tx.execute_batch(migration.sql)?;
1141 tx.execute(
1142 "INSERT INTO schema_migrations (version, name) VALUES (?1, ?2)",
1143 params![migration.version, migration.name],
1144 )?;
1145 }
1146 tx.commit()?;
1147 Ok(())
1148 }
1149
1150 pub fn fetch_chain_rows(&self) -> Result<Vec<ChainRow>, TransactionStoreError> {
1152 let conn = self.connection()?;
1153 let mut stmt = conn.prepare(&format!(
1154 "SELECT {CHAIN_ROW_COLUMNS} FROM transactions ORDER BY seq ASC"
1155 ))?;
1156 let rows = stmt.query_map([], chain_row_from_sqlite)?;
1157 let mut out = Vec::new();
1158 for row in rows {
1159 out.push(row?);
1160 }
1161 Ok(out)
1162 }
1163
1164 pub fn fetch_event_rows(&self) -> Result<Vec<EventRow>, TransactionStoreError> {
1166 let conn = self.connection()?;
1167 let mut stmt = conn.prepare(
1168 "SELECT seq, key_id, kind, transaction_id, receipt_digest, \
1169 created_at, prev_chain_hash, chain_hash, chain_version, caller_principal \
1170 FROM audit_events ORDER BY seq ASC",
1171 )?;
1172 let rows = stmt.query_map([], event_row_from_sqlite)?;
1173 let mut out = Vec::new();
1174 for row in rows {
1175 out.push(row?);
1176 }
1177 Ok(out)
1178 }
1179
1180 pub fn verify_audit_chain(
1182 &self,
1183 key: &AuditKey,
1184 ) -> Result<VerifyOutcome, TransactionStoreError> {
1185 let rows = self.fetch_chain_rows()?;
1186 Ok(audit_chain::verify_chain(key, &rows))
1187 }
1188
1189 pub fn verify_event_chain(
1191 &self,
1192 key: &AuditKey,
1193 ) -> Result<VerifyOutcome, TransactionStoreError> {
1194 let rows = self.fetch_event_rows()?;
1195 Ok(audit_chain::verify_event_chain(key, &rows))
1196 }
1197
1198 pub fn verify_event_binding(
1201 &self,
1202 ) -> Result<audit_chain::BindingOutcome, TransactionStoreError> {
1203 let tx_rows = self.fetch_chain_rows()?;
1204 let event_rows = self.fetch_event_rows()?;
1205 Ok(audit_chain::verify_event_binding(&tx_rows, &event_rows))
1206 }
1207
1208 pub fn verify_audit_chain_with_pubkey(
1211 &self,
1212 verifying_key_hex: &str,
1213 ) -> Result<VerifyOutcome, TransactionStoreError> {
1214 let rows = self.fetch_chain_rows()?;
1215 Ok(audit_chain::verify_chain_with_pubkey(
1216 verifying_key_hex,
1217 &rows,
1218 ))
1219 }
1220
1221 pub fn fetch_chain_row(
1224 &self,
1225 transaction_id: &str,
1226 ) -> Result<Option<ChainRow>, TransactionStoreError> {
1227 let conn = self.connection()?;
1228 let mut stmt = conn.prepare(&format!(
1229 "SELECT {CHAIN_ROW_COLUMNS} FROM transactions WHERE transaction_id = ?1"
1230 ))?;
1231 let mut rows = stmt.query(params![transaction_id])?;
1232 if let Some(row) = rows.next()? {
1233 Ok(Some(chain_row_from_sqlite(row)?))
1234 } else {
1235 Ok(None)
1236 }
1237 }
1238
1239 fn next_seq_and_prev_hash(conn: &Connection) -> Result<(u64, String), TransactionStoreError> {
1243 let prev: Option<(i64, String)> = query_optional(
1244 conn,
1245 "SELECT seq, chain_hash FROM transactions ORDER BY seq DESC LIMIT 1",
1246 [],
1247 |row| Ok((row.get(0)?, row.get(1)?)),
1248 )?;
1249 Ok(match prev {
1250 Some((seq, hash)) => ((seq as u64) + 1, hash),
1251 None => (1, String::new()),
1252 })
1253 }
1254
1255 fn event_chain_tip(conn: &Connection) -> Result<Option<String>, TransactionStoreError> {
1258 Ok(query_optional(
1259 conn,
1260 "SELECT chain_hash FROM audit_events ORDER BY seq DESC LIMIT 1",
1261 [],
1262 |row| row.get(0),
1263 )?)
1264 }
1265
1266 fn append_event(
1271 conn: &Connection,
1272 key: &AuditKey,
1273 kind: AuditEventKind,
1274 transaction_id: &str,
1275 receipt_digest: &str,
1276 identity: EventIdentity<'_>,
1277 ) -> Result<(), TransactionStoreError> {
1278 let prev_chain_hash = Self::event_chain_tip(conn)?.unwrap_or_default();
1279 let seq: i64 = conn.query_row(
1280 "SELECT COALESCE(MAX(seq), 0) + 1 FROM audit_events",
1281 [],
1282 |row| row.get(0),
1283 )?;
1284 let created_at: String =
1285 conn.query_row("SELECT strftime('%Y-%m-%dT%H:%M:%fZ', 'now')", [], |row| {
1286 row.get(0)
1287 })?;
1288 let key_id = CURRENT_KEY_ID.to_string();
1289 let content = EventContent {
1294 seq: seq as u64,
1295 key_id: &key_id,
1296 kind,
1297 transaction_id,
1298 receipt_digest,
1299 created_at: &created_at,
1300 identity,
1301 };
1302 let chain_hash = key.event_hash(&content, &prev_chain_hash);
1303 let stored_principal = match identity {
1304 EventIdentity::V2 { caller_principal } => Some(caller_principal),
1305 EventIdentity::LegacyV1 => None,
1306 };
1307 conn.execute(
1308 "INSERT INTO audit_events (\
1309 seq, key_id, kind, transaction_id, receipt_digest, \
1310 created_at, chain_hash, prev_chain_hash, chain_version, caller_principal \
1311 ) VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9, ?10)",
1312 params![
1313 seq,
1314 key_id,
1315 kind.as_str(),
1316 transaction_id,
1317 receipt_digest,
1318 created_at,
1319 chain_hash,
1320 prev_chain_hash,
1321 identity.version() as i64,
1322 stored_principal,
1323 ],
1324 )?;
1325 Ok(())
1326 }
1327
1328 fn insert_transaction(
1329 conn: &Connection,
1330 key: &AuditKey,
1331 transaction_id: &str,
1332 transaction: NewTransaction,
1333 ) -> Result<InsertedTransaction, TransactionStoreError> {
1334 let request_id = transaction.request_id;
1335 let request_hash = transaction.request_hash;
1336 let action_name = transaction.action_name;
1337 let risk_level = transaction.risk_level;
1338 let status = JobState::Queued;
1340 let approval_id = Some(key.approval_commitment(transaction_id, &request_hash));
1347 let summary = transaction.summary;
1348 let warnings = transaction.warnings;
1349 let warnings_json = serde_json::to_string(&warnings)?;
1350
1351 let (seq, prev_chain_hash) = Self::next_seq_and_prev_hash(conn)?;
1354
1355 let created_at: String =
1359 conn.query_row("SELECT strftime('%Y-%m-%dT%H:%M:%fZ', 'now')", [], |row| {
1360 row.get(0)
1361 })?;
1362
1363 let key_id = CURRENT_KEY_ID.to_string();
1364 let event_tip = Self::event_chain_tip(conn)?.unwrap_or_default();
1368 let caller_role = transaction.caller_role.as_str();
1369 let caller_principal = transaction.caller_principal.as_signed_str();
1370 let identity = ChainIdentity::V4 {
1374 caller_role,
1375 event_tip: &event_tip,
1376 caller_principal: &caller_principal,
1377 };
1378 let chain_hash = key.chain_hash(
1379 &ChainContent {
1380 seq,
1381 key_id: &key_id,
1382 transaction_id,
1383 request_id: &request_id,
1384 request_hash: &request_hash,
1385 action_name: &action_name,
1386 risk_level,
1387 summary: &summary,
1388 approval_id: approval_id.as_deref(),
1389 warnings_json: &warnings_json,
1390 created_at: &created_at,
1391 identity,
1392 },
1393 &prev_chain_hash,
1394 );
1395
1396 let record = TransactionRecord {
1397 transaction_id: transaction_id.to_string(),
1398 request_id: request_id.clone(),
1399 request_hash: request_hash.clone(),
1400 action_name: action_name.clone(),
1401 risk_level,
1402 status,
1403 approval_id: approval_id.clone(),
1404 summary: summary.clone(),
1405 warnings: warnings.clone(),
1406 };
1407
1408 conn.execute(
1409 "INSERT INTO transactions (
1410 transaction_id,
1411 request_id,
1412 request_hash,
1413 action_name,
1414 risk_level,
1415 status,
1416 approval_id,
1417 summary,
1418 warnings_json,
1419 created_at,
1420 seq,
1421 key_id,
1422 chain_hash,
1423 prev_chain_hash,
1424 chain_version,
1425 caller_role,
1426 event_tip,
1427 caller_principal
1428 ) VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9, ?10, ?11, ?12, ?13, ?14, ?15, ?16, ?17, ?18)",
1429 params![
1430 transaction_id,
1431 request_id,
1432 request_hash,
1433 action_name,
1434 serialize_field(&risk_level)?,
1435 serialize_field(&status)?,
1436 approval_id,
1437 summary,
1438 warnings_json,
1439 created_at,
1440 seq as i64,
1441 key_id,
1442 chain_hash,
1443 prev_chain_hash,
1444 identity.version() as i64,
1445 caller_role,
1446 event_tip,
1447 caller_principal,
1448 ],
1449 )?;
1450
1451 Ok(InsertedTransaction {
1452 record,
1453 seq,
1454 chain_hash,
1455 })
1456 }
1457
1458 fn insert_preview(
1459 conn: &Connection,
1460 transaction_id: &str,
1461 preview: &PreviewEnvelope,
1462 ) -> Result<(), TransactionStoreError> {
1463 conn.execute(
1464 "INSERT INTO transaction_previews (transaction_id, preview_json)
1465 VALUES (?1, ?2)",
1466 params![transaction_id, serde_json::to_string(preview)?],
1467 )?;
1468 Ok(())
1469 }
1470}
1471
1472fn ensure_private_dir(dir: &Path) -> std::io::Result<()> {
1479 use std::os::unix::fs::DirBuilderExt;
1480 if dir.exists() {
1481 return Ok(());
1482 }
1483 std::fs::DirBuilder::new()
1484 .recursive(true)
1485 .mode(0o700)
1486 .create(dir)
1487}
1488
1489fn restrict_db_file_mode(db_path: &Path) {
1503 use std::os::unix::fs::PermissionsExt;
1504
1505 let sidecars = [
1509 db_path.to_path_buf(),
1510 with_suffix(db_path, "-wal"),
1511 with_suffix(db_path, "-shm"),
1512 ];
1513 for path in sidecars {
1514 let Ok(metadata) = std::fs::metadata(&path) else {
1515 continue; };
1517 let mode = metadata.permissions().mode() & 0o777;
1518 if mode & 0o077 == 0 {
1519 continue; }
1521 if let Err(err) = std::fs::set_permissions(&path, std::fs::Permissions::from_mode(0o600)) {
1522 eprintln!(
1523 "[sysknife-daemon] WARNING: {} is mode {mode:o} and could not be \
1524 narrowed to 0600 ({err}); it is readable by anyone who can reach \
1525 it. The parent directory should be 0700 — check it.",
1526 path.display()
1527 );
1528 }
1529 }
1530}
1531
1532fn with_suffix(path: &Path, suffix: &str) -> PathBuf {
1535 let mut name = path.as_os_str().to_os_string();
1536 name.push(suffix);
1537 PathBuf::from(name)
1538}
1539
1540fn row_to_record(row: &rusqlite::Row) -> Result<TransactionRecord, TransactionStoreError> {
1541 Ok(TransactionRecord {
1542 transaction_id: row.get(0)?,
1543 request_id: row.get(1)?,
1544 request_hash: row.get(2)?,
1545 action_name: row.get(3)?,
1546 risk_level: deserialize_field(&row.get::<_, String>(4)?)?,
1547 status: deserialize_field(&row.get::<_, String>(5)?)?,
1548 approval_id: row.get(6)?,
1549 summary: row.get(7)?,
1550 warnings: serde_json::from_str(&row.get::<_, String>(8)?)?,
1551 })
1552}
1553
1554fn serialize_field<T: Serialize>(value: &T) -> Result<String, serde_json::Error> {
1555 serde_json::to_string(value)
1556}
1557
1558fn deserialize_field<T: DeserializeOwned>(value: &str) -> Result<T, serde_json::Error> {
1559 serde_json::from_str(value)
1560}
1561
1562#[cfg(test)]
1563mod tests {
1564 use super::*;
1565 use crate::audit_chain::CHAIN_VERSION_CURRENT;
1566 use crate::audit_chain::CHAIN_VERSION_LEGACY;
1567 use crate::audit_chain::EVENT_VERSION_CURRENT;
1568 use crate::audit_chain::EVENT_VERSION_LEGACY;
1569 use std::os::unix::fs::PermissionsExt;
1570 use tempfile::tempdir;
1571
1572 fn test_store(path: impl AsRef<Path>) -> TransactionStore {
1576 let key = Arc::new(AuditKey::from_bytes(vec![0x42; 32]));
1577 TransactionStore::open_with_key(path, key).unwrap()
1578 }
1579
1580 fn legacy_v1_database(path: &Path, key: &AuditKey) -> String {
1587 let conn = Connection::open(path).unwrap();
1588 conn.execute_batch(
1589 "CREATE TABLE schema_migrations (
1590 version INTEGER PRIMARY KEY,
1591 name TEXT NOT NULL,
1592 applied_at TEXT NOT NULL DEFAULT (datetime('now'))
1593 );",
1594 )
1595 .unwrap();
1596 conn.execute_batch(SQLITE_MIGRATIONS[0].sql).unwrap();
1597 conn.execute(
1598 "INSERT INTO schema_migrations (version, name) VALUES (1, ?1)",
1599 params![SQLITE_MIGRATIONS[0].name],
1600 )
1601 .unwrap();
1602
1603 let transaction_id = "tx-legacy";
1604 let created_at = "2026-04-24T12:00:00.000Z";
1605 let chain_hash = key.chain_hash(
1606 &ChainContent {
1607 seq: 1,
1608 key_id: CURRENT_KEY_ID,
1609 transaction_id,
1610 request_id: "req-legacy",
1611 request_hash: "hash-legacy",
1612 action_name: "UpdateSystem",
1613 risk_level: RiskLevel::High,
1614 summary: "Upgrade the system",
1615 approval_id: None,
1616 warnings_json: "[]",
1617 created_at,
1618 identity: ChainIdentity::LegacyV1,
1619 },
1620 "",
1621 );
1622 conn.execute(
1623 "INSERT INTO transactions (
1624 transaction_id, request_id, request_hash, action_name, risk_level,
1625 status, approval_id, summary, warnings_json, created_at,
1626 seq, key_id, chain_hash, prev_chain_hash
1627 ) VALUES (?1, ?2, ?3, ?4, ?5, ?6, NULL, ?7, ?8, ?9, 1, ?10, ?11, '')",
1628 params![
1629 transaction_id,
1630 "req-legacy",
1631 "hash-legacy",
1632 "UpdateSystem",
1633 serialize_field(&RiskLevel::High).unwrap(),
1634 serialize_field(&JobState::Queued).unwrap(),
1635 "Upgrade the system",
1636 "[]",
1637 created_at,
1638 CURRENT_KEY_ID,
1639 chain_hash,
1640 ],
1641 )
1642 .unwrap();
1643 transaction_id.to_string()
1644 }
1645
1646 fn v2_database(path: &Path, key: &AuditKey) -> String {
1653 let conn = Connection::open(path).unwrap();
1654 conn.execute_batch(
1655 "CREATE TABLE schema_migrations (
1656 version INTEGER PRIMARY KEY,
1657 name TEXT NOT NULL,
1658 applied_at TEXT NOT NULL DEFAULT (datetime('now'))
1659 );",
1660 )
1661 .unwrap();
1662 for migration in &SQLITE_MIGRATIONS[0..=1] {
1663 conn.execute_batch(migration.sql).unwrap();
1664 conn.execute(
1665 "INSERT INTO schema_migrations (version, name) VALUES (?1, ?2)",
1666 params![migration.version, migration.name],
1667 )
1668 .unwrap();
1669 }
1670
1671 let transaction_id = "tx-v2";
1672 let created_at = "2026-07-01T12:00:00.000Z";
1673 let chain_hash = key.chain_hash(
1674 &ChainContent {
1675 seq: 1,
1676 key_id: CURRENT_KEY_ID,
1677 transaction_id,
1678 request_id: "req-v2",
1679 request_hash: "hash-v2",
1680 action_name: "UpdateSystem",
1681 risk_level: RiskLevel::High,
1682 summary: "Upgrade the system",
1683 approval_id: None,
1684 warnings_json: "[]",
1685 created_at,
1686 identity: ChainIdentity::V2 {
1687 caller_role: "admin",
1688 event_tip: "",
1689 },
1690 },
1691 "",
1692 );
1693 conn.execute(
1694 "INSERT INTO transactions (
1695 transaction_id, request_id, request_hash, action_name, risk_level,
1696 status, approval_id, summary, warnings_json, created_at,
1697 seq, key_id, chain_hash, prev_chain_hash,
1698 chain_version, caller_role, event_tip
1699 ) VALUES (?1, ?2, ?3, ?4, ?5, ?6, NULL, ?7, ?8, ?9, 1, ?10, ?11, '', 2, 'admin', '')",
1700 params![
1701 transaction_id,
1702 "req-v2",
1703 "hash-v2",
1704 "UpdateSystem",
1705 serialize_field(&RiskLevel::High).unwrap(),
1706 serialize_field(&JobState::Queued).unwrap(),
1707 "Upgrade the system",
1708 "[]",
1709 created_at,
1710 CURRENT_KEY_ID,
1711 chain_hash,
1712 ],
1713 )
1714 .unwrap();
1715 transaction_id.to_string()
1716 }
1717
1718 #[test]
1726 fn a_v2_database_keeps_verifying_and_accepts_v3_rows() {
1727 let dir = tempdir().unwrap();
1728 let db_path = dir.path().join("tx.db");
1729 let key = AuditKey::from_bytes(vec![0x42; 32]);
1730 let old_transaction = v2_database(&db_path, &key);
1731
1732 let store = test_store(&db_path);
1733
1734 let version: i64 = store
1735 .connection()
1736 .unwrap()
1737 .query_row("SELECT MAX(version) FROM schema_migrations", [], |row| {
1738 row.get(0)
1739 })
1740 .unwrap();
1741 assert_eq!(
1742 version, 4,
1743 "opening a v2 database must apply every migration up to 4"
1744 );
1745
1746 let rows = store.fetch_chain_rows().unwrap();
1747 assert_eq!(rows.len(), 1);
1748 assert_eq!(rows[0].transaction_id, old_transaction);
1749 assert_eq!(rows[0].chain_version, audit_chain::CHAIN_VERSION_V2);
1750 assert_eq!(
1751 rows[0].caller_principal, None,
1752 "migration 3 must not invent a principal for a row signed without one"
1753 );
1754 assert!(matches!(
1755 store.verify_audit_chain(&key).unwrap(),
1756 audit_chain::VerifyOutcome::Intact { rows_checked: 1 }
1757 ));
1758
1759 let mut fresh = queued_transaction();
1761 fresh.caller_principal = CallerPrincipal::Uid(4242);
1762 store.record(fresh).unwrap();
1763
1764 let rows = store.fetch_chain_rows().unwrap();
1765 assert_eq!(rows.len(), 2);
1766 assert_eq!(rows[1].chain_version, audit_chain::CHAIN_VERSION_CURRENT);
1767 assert_eq!(rows[1].caller_principal.as_deref(), Some("uid:4242"));
1768 assert_eq!(
1769 rows[1].prev_chain_hash, rows[0].chain_hash,
1770 "the v3 row must chain onto the v2 row, not start a new chain"
1771 );
1772 assert!(matches!(
1773 store.verify_audit_chain(&key).unwrap(),
1774 audit_chain::VerifyOutcome::Intact { rows_checked: 2 }
1775 ));
1776
1777 let census = audit_chain::AttributionCensus::of(&rows);
1785 assert_eq!(census.rows(), 2);
1786 assert_eq!(census.named(), 1, "only the v3 row names uid:4242");
1787 assert_eq!(
1788 census.not_recorded(),
1789 1,
1790 "the v2 row's encoding never signed a principal"
1791 );
1792 assert_eq!(census.attribution_failed(), 0);
1793 assert_eq!(census.unattested(), 0);
1794 assert_eq!(census.unnamed(), 1);
1795 }
1796
1797 #[test]
1798 fn a_database_written_before_the_migration_still_verifies_after_upgrading() {
1799 let dir = tempdir().unwrap();
1804 let db_path = dir.path().join("tx.db");
1805 let key = AuditKey::from_bytes(vec![0x42; 32]);
1806 legacy_v1_database(&db_path, &key);
1807
1808 let store = test_store(&db_path);
1809 assert_eq!(
1810 store.verify_audit_chain(&key).unwrap(),
1811 VerifyOutcome::Intact { rows_checked: 1 }
1812 );
1813 let rows = store.fetch_chain_rows().unwrap();
1814 assert_eq!(rows[0].chain_version, CHAIN_VERSION_LEGACY);
1815 assert_eq!(rows[0].caller_role, None);
1816 }
1817
1818 #[test]
1819 fn rows_appended_after_the_migration_chain_onto_legacy_rows() {
1820 let dir = tempdir().unwrap();
1824 let db_path = dir.path().join("tx.db");
1825 let key = AuditKey::from_bytes(vec![0x42; 32]);
1826 legacy_v1_database(&db_path, &key);
1827
1828 let store = test_store(&db_path);
1829 store.record(queued_transaction()).unwrap();
1830
1831 let rows = store.fetch_chain_rows().unwrap();
1832 assert_eq!(rows.len(), 2);
1833 assert_eq!(rows[1].prev_chain_hash, rows[0].chain_hash);
1834 assert_eq!(rows[1].chain_version, CHAIN_VERSION_CURRENT);
1835 assert_eq!(
1836 store.verify_audit_chain(&key).unwrap(),
1837 VerifyOutcome::Intact { rows_checked: 2 }
1838 );
1839 }
1840
1841 #[test]
1842 fn a_schema_newer_than_this_binary_is_refused() {
1843 let dir = tempdir().unwrap();
1844 let db_path = dir.path().join("tx.db");
1845 let key = AuditKey::from_bytes(vec![0x42; 32]);
1846 legacy_v1_database(&db_path, &key);
1847 let conn = Connection::open(&db_path).unwrap();
1848 conn.execute(
1849 "INSERT INTO schema_migrations (version, name) VALUES (99, 'from the future')",
1850 [],
1851 )
1852 .unwrap();
1853 drop(conn);
1854
1855 let err = TransactionStore::open_with_key(&db_path, Arc::new(key)).unwrap_err();
1856 assert!(
1857 matches!(err, TransactionStoreError::DatabaseInvariant(ref m) if m.contains("99")),
1858 "expected a refusal naming the unsupported version, got {err:?}"
1859 );
1860 }
1861
1862 #[test]
1865 fn the_signed_row_records_which_role_asked() {
1866 let dir = tempdir().unwrap();
1867 let store = test_store(dir.path().join("tx.db"));
1868 let mut tx = queued_transaction();
1869 tx.caller_role = CallerRole::Admin;
1870 store.record(tx).unwrap();
1871
1872 let rows = store.fetch_chain_rows().unwrap();
1873 assert_eq!(rows[0].caller_role.as_deref(), Some("admin"));
1874 }
1875
1876 #[test]
1877 fn approving_consuming_and_revoking_each_append_a_chained_event() {
1878 let dir = tempdir().unwrap();
1879 let key = AuditKey::from_bytes(vec![0x42; 32]);
1880 let store = test_store(dir.path().join("tx.db"));
1881
1882 let a = store.record(queued_transaction()).unwrap();
1884 let receipt = store
1885 .approve_transaction(&a.transaction_id, CallerPrincipal::Uid(1000))
1886 .unwrap()
1887 .expect("queued transaction approves");
1888 let digest = audit_chain::approval_receipt_digest(&receipt);
1889 assert!(store
1890 .claim_approved_for_execution(&a.transaction_id, &digest, CallerPrincipal::Uid(1000))
1891 .unwrap());
1892
1893 let b = store.record(queued_transaction()).unwrap();
1895 store
1896 .approve_transaction(&b.transaction_id, CallerPrincipal::Uid(1000))
1897 .unwrap();
1898 assert!(store
1899 .revoke_unconsumed_approval(&b.transaction_id, CallerPrincipal::Uid(1000))
1900 .unwrap());
1901
1902 let events = store.fetch_event_rows().unwrap();
1903 let kinds: Vec<&str> = events.iter().map(|e| e.kind.as_str()).collect();
1904 assert_eq!(
1905 kinds,
1906 vec![
1907 "approval_granted",
1908 "approval_consumed",
1909 "status_running",
1910 "approval_granted",
1911 "approval_revoked",
1912 ]
1913 );
1914 assert_eq!(
1915 store.verify_event_chain(&key).unwrap(),
1916 VerifyOutcome::Intact { rows_checked: 5 }
1917 );
1918 }
1919
1920 #[test]
1921 fn a_stored_approval_event_names_the_account_that_acted() {
1922 let dir = tempdir().unwrap();
1929 let key = AuditKey::from_bytes(vec![0x42; 32]);
1930 let store = test_store(dir.path().join("tx.db"));
1931
1932 let tx = store.record(queued_transaction()).unwrap();
1933 let receipt = store
1935 .approve_transaction(&tx.transaction_id, CallerPrincipal::Uid(1000))
1936 .unwrap()
1937 .expect("queued transaction approves");
1938 let digest = audit_chain::approval_receipt_digest(&receipt);
1939 assert!(store
1941 .claim_approved_for_execution(&tx.transaction_id, &digest, CallerPrincipal::Uid(1001))
1942 .unwrap());
1943
1944 let events = store.fetch_event_rows().unwrap();
1945 assert_eq!(events.len(), 3);
1946 let granted = events
1947 .iter()
1948 .find(|e| e.kind == "approval_granted")
1949 .expect("a grant event");
1950 let consumed = events
1951 .iter()
1952 .find(|e| e.kind == "approval_consumed")
1953 .expect("a consume event");
1954
1955 assert_eq!(granted.chain_version, EVENT_VERSION_CURRENT);
1957 assert_eq!(consumed.chain_version, EVENT_VERSION_CURRENT);
1958 assert_eq!(
1959 granted.caller_principal.as_deref(),
1960 Some("uid:1000"),
1961 "the grant must name the approver"
1962 );
1963 assert_eq!(
1964 consumed.caller_principal.as_deref(),
1965 Some("uid:1001"),
1966 "the consume must name the executor"
1967 );
1968 assert_ne!(
1969 granted.caller_principal, consumed.caller_principal,
1970 "a grant and a consume by different accounts must be distinguishable"
1971 );
1972
1973 assert_eq!(
1975 store.verify_event_chain(&key).unwrap(),
1976 VerifyOutcome::Intact { rows_checked: 3 }
1977 );
1978 }
1979
1980 #[test]
1981 fn a_stored_status_event_carries_no_principal_and_stays_legacy() {
1982 let dir = tempdir().unwrap();
1987 let key = AuditKey::from_bytes(vec![0x42; 32]);
1988 let store = test_store(dir.path().join("tx.db"));
1989
1990 let tx = store.record(queued_transaction()).unwrap();
1991 let receipt = store
1992 .approve_transaction(&tx.transaction_id, CallerPrincipal::Uid(1000))
1993 .unwrap()
1994 .expect("queued transaction approves");
1995 let digest = audit_chain::approval_receipt_digest(&receipt);
1996 assert!(store
1997 .claim_approved_for_execution(&tx.transaction_id, &digest, CallerPrincipal::Uid(1000))
1998 .unwrap());
1999 store
2000 .update_status(&tx.transaction_id, JobState::Succeeded)
2001 .unwrap();
2002
2003 let events = store.fetch_event_rows().unwrap();
2004 let status = events
2005 .iter()
2006 .find(|e| e.kind == "status_succeeded")
2007 .expect("a status event");
2008 assert_eq!(status.chain_version, EVENT_VERSION_LEGACY);
2009 assert_eq!(
2010 status.caller_principal, None,
2011 "a status event names no account"
2012 );
2013 assert_eq!(
2015 store.verify_event_chain(&key).unwrap(),
2016 VerifyOutcome::Intact { rows_checked: 4 }
2017 );
2018 }
2019
2020 #[test]
2021 fn audit_status_event_failure_rolls_back_the_entire_transition() {
2022 for operation in ["claim", "cancel", "sweep"] {
2023 let dir = tempdir().unwrap();
2024 let store = test_store(dir.path().join("tx.db"));
2025 let tx = store.record(queued_transaction()).unwrap();
2026 let receipt = store
2027 .approve_transaction(&tx.transaction_id, CallerPrincipal::Uid(1000))
2028 .unwrap()
2029 .unwrap();
2030 let digest = audit_chain::approval_receipt_digest(&receipt);
2031 let conn = store.connection().unwrap();
2032 if operation == "sweep" {
2033 conn.execute(
2034 "UPDATE transactions SET created_at = datetime('now', '-20 minutes')",
2035 [],
2036 )
2037 .unwrap();
2038 }
2039 conn.execute_batch("CREATE TRIGGER reject_status_event BEFORE INSERT ON audit_events WHEN NEW.kind LIKE 'status_%' BEGIN SELECT RAISE(ABORT, 'injected event failure'); END;").unwrap();
2040 let result = match operation {
2041 "claim" => store.claim_approved_for_execution(
2042 &tx.transaction_id,
2043 &digest,
2044 CallerPrincipal::Uid(1000),
2045 ),
2046 "cancel" => store.cancel_queued(&tx.transaction_id, CallerPrincipal::Uid(1000)),
2047 _ => store.cleanup_stale_queued().map(|count| count > 0),
2048 };
2049 assert!(
2050 result.is_err(),
2051 "{operation} must fail when its event cannot be written"
2052 );
2053 assert_eq!(
2054 store.get(&tx.transaction_id).unwrap().unwrap().status,
2055 JobState::Queued
2056 );
2057 let unconsumed: i64 = conn.query_row("SELECT COUNT(*) FROM transaction_approvals WHERE transaction_id = ?1 AND consumed_at IS NULL", params![tx.transaction_id], |row| row.get(0)).unwrap();
2058 assert_eq!(
2059 unconsumed, 1,
2060 "{operation} must roll back approval changes too"
2061 );
2062 assert_eq!(store.fetch_event_rows().unwrap().len(), 1);
2063 }
2064 }
2065
2066 #[test]
2067 fn audit_status_legacy_consumption_cannot_be_rewritten_as_queued() {
2068 let dir = tempdir().unwrap();
2069 let store = test_store(dir.path().join("tx.db"));
2070 let tx = store.record(queued_transaction()).unwrap();
2071 let receipt = store
2072 .approve_transaction(&tx.transaction_id, CallerPrincipal::Uid(1000))
2073 .unwrap()
2074 .unwrap();
2075 let digest = audit_chain::approval_receipt_digest(&receipt);
2076 store
2077 .claim_approved_for_execution(&tx.transaction_id, &digest, CallerPrincipal::Uid(1000))
2078 .unwrap();
2079 let conn = store.connection().unwrap();
2080 conn.execute("DELETE FROM audit_events WHERE kind LIKE 'status_%'", [])
2082 .unwrap();
2083 let outcome = store.status_matches_chain().unwrap();
2084 assert_eq!(
2085 serde_json::to_value(&outcome).unwrap()["state"],
2086 "cannot_verify"
2087 );
2088 conn.execute(
2089 "UPDATE transactions SET status = ?1 WHERE transaction_id = ?2",
2090 params![
2091 serialize_field(&JobState::Queued).unwrap(),
2092 tx.transaction_id
2093 ],
2094 )
2095 .unwrap();
2096 assert!(matches!(
2097 store.status_matches_chain().unwrap(),
2098 audit_chain::StatusOutcome::Disagrees { .. }
2099 ));
2100 }
2101
2102 #[test]
2103 fn audit_status_tracks_an_in_flight_claim() {
2104 let dir = tempdir().unwrap();
2105 let store = test_store(dir.path().join("tx.db"));
2106 let tx = store.record(queued_transaction()).unwrap();
2107 let receipt = store
2108 .approve_transaction(&tx.transaction_id, CallerPrincipal::Uid(1000))
2109 .unwrap()
2110 .unwrap();
2111 let digest = audit_chain::approval_receipt_digest(&receipt);
2112 assert!(store
2113 .claim_approved_for_execution(&tx.transaction_id, &digest, CallerPrincipal::Uid(1000))
2114 .unwrap());
2115 assert_eq!(
2116 store.status_matches_chain().unwrap(),
2117 audit_chain::StatusOutcome::Agrees { rows_checked: 1 }
2118 );
2119 assert_eq!(
2120 store.fetch_event_rows().unwrap().last().unwrap().kind,
2121 "status_running"
2122 );
2123 assert_eq!(
2124 store
2125 .verify_event_chain(&AuditKey::from_bytes(vec![0x42; 32]))
2126 .unwrap(),
2127 VerifyOutcome::Intact { rows_checked: 3 }
2128 );
2129 }
2130
2131 #[test]
2132 fn audit_status_tracks_cancellation_with_and_without_approval() {
2133 for approved in [false, true] {
2134 let dir = tempdir().unwrap();
2135 let store = test_store(dir.path().join("tx.db"));
2136 let tx = store.record(queued_transaction()).unwrap();
2137 if approved {
2138 store
2139 .approve_transaction(&tx.transaction_id, CallerPrincipal::Uid(1000))
2140 .unwrap();
2141 }
2142 assert!(store
2143 .cancel_queued(&tx.transaction_id, CallerPrincipal::Uid(1000))
2144 .unwrap());
2145 assert_eq!(
2146 store.status_matches_chain().unwrap(),
2147 audit_chain::StatusOutcome::Agrees { rows_checked: 1 }
2148 );
2149 assert_eq!(
2150 store.fetch_event_rows().unwrap().last().unwrap().kind,
2151 "status_canceled"
2152 );
2153 assert!(matches!(
2154 store
2155 .verify_event_chain(&AuditKey::from_bytes(vec![0x42; 32]))
2156 .unwrap(),
2157 VerifyOutcome::Intact { .. }
2158 ));
2159 let count = store.fetch_event_rows().unwrap().len();
2160 assert!(!store
2161 .cancel_queued(&tx.transaction_id, CallerPrincipal::Uid(1000))
2162 .unwrap());
2163 assert_eq!(store.fetch_event_rows().unwrap().len(), count);
2164 }
2165 }
2166
2167 #[test]
2168 fn audit_status_tracks_stale_sweep_with_and_without_approval() {
2169 for approved in [false, true] {
2170 let dir = tempdir().unwrap();
2171 let store = test_store(dir.path().join("tx.db"));
2172 let tx = store.record(queued_transaction()).unwrap();
2173 if approved {
2174 store
2175 .approve_transaction(&tx.transaction_id, CallerPrincipal::Uid(1000))
2176 .unwrap();
2177 }
2178 store.connection().unwrap().execute(
2179 "UPDATE transactions SET created_at = datetime('now', '-20 minutes') WHERE transaction_id = ?1",
2180 params![tx.transaction_id],
2181 ).unwrap();
2182 assert_eq!(store.cleanup_stale_queued().unwrap(), 1);
2183 assert_eq!(
2184 store.status_matches_chain().unwrap(),
2185 audit_chain::StatusOutcome::Agrees { rows_checked: 1 }
2186 );
2187 assert_eq!(
2188 store.fetch_event_rows().unwrap().last().unwrap().kind,
2189 "status_canceled"
2190 );
2191 let count = store.fetch_event_rows().unwrap().len();
2192 assert_eq!(store.cleanup_stale_queued().unwrap(), 0);
2193 assert_eq!(store.fetch_event_rows().unwrap().len(), count);
2194 }
2195 }
2196
2197 #[test]
2198 fn audit_status_only_historical_claims_and_cancellations_are_inconclusive() {
2199 let dir = tempdir().unwrap();
2200 let store = test_store(dir.path().join("tx.db"));
2201 let tx = store.record(queued_transaction()).unwrap();
2202 store
2205 .approve_transaction(&tx.transaction_id, CallerPrincipal::Uid(1000))
2206 .unwrap();
2207 store
2208 .revoke_unconsumed_approval(&tx.transaction_id, CallerPrincipal::Uid(1000))
2209 .unwrap();
2210 assert_eq!(
2211 store.status_matches_chain().unwrap(),
2212 audit_chain::StatusOutcome::Agrees { rows_checked: 1 }
2213 );
2214 for state in [
2215 JobState::Canceled,
2216 JobState::Running,
2217 JobState::Succeeded,
2218 JobState::Failed,
2219 JobState::RolledBack,
2220 JobState::NeedsReboot,
2221 ] {
2222 store
2223 .connection()
2224 .unwrap()
2225 .execute(
2226 "UPDATE transactions SET status = ?1 WHERE transaction_id = ?2",
2227 params![serialize_field(&state).unwrap(), tx.transaction_id],
2228 )
2229 .unwrap();
2230 let outcome = store.status_matches_chain().unwrap();
2231 let historical = matches!(state, JobState::Running | JobState::Canceled);
2232 assert_eq!(
2233 serde_json::to_value(&outcome).unwrap()["state"],
2234 if historical {
2235 "cannot_verify"
2236 } else {
2237 "disagrees"
2238 },
2239 "unsigned state {state:?}"
2240 );
2241 assert_eq!(
2242 audit_chain::status_outcome_to_exit_code(&outcome),
2243 if historical { 2 } else { 1 },
2244 "unsigned state {state:?}"
2245 );
2246 }
2247 }
2248
2249 #[test]
2250 fn audit_status_new_event_contradictions_remain_broken() {
2251 let dir = tempdir().unwrap();
2252 let store = test_store(dir.path().join("tx.db"));
2253 let tx = store.record(queued_transaction()).unwrap();
2254 store
2255 .cancel_queued(&tx.transaction_id, CallerPrincipal::Uid(1000))
2256 .unwrap();
2257 store
2258 .connection()
2259 .unwrap()
2260 .execute(
2261 "UPDATE transactions SET status = ?1 WHERE transaction_id = ?2",
2262 params![
2263 serialize_field(&JobState::Queued).unwrap(),
2264 tx.transaction_id
2265 ],
2266 )
2267 .unwrap();
2268 let legacy = store.record(queued_transaction()).unwrap();
2271 store
2272 .connection()
2273 .unwrap()
2274 .execute(
2275 "UPDATE transactions SET status = ?1 WHERE transaction_id = ?2",
2276 params![
2277 serialize_field(&JobState::Canceled).unwrap(),
2278 legacy.transaction_id
2279 ],
2280 )
2281 .unwrap();
2282 assert!(matches!(
2283 store.status_matches_chain().unwrap(),
2284 audit_chain::StatusOutcome::Disagrees { .. }
2285 ));
2286 }
2287
2288 #[test]
2289 fn a_terminal_outcome_is_chained_so_rewriting_status_is_detectable() {
2290 let dir = tempdir().unwrap();
2298 let key = AuditKey::from_bytes(vec![0x42; 32]);
2299 let store = test_store(dir.path().join("tx.db"));
2300
2301 let tx = store.record(queued_transaction()).unwrap();
2302 let receipt = store
2303 .approve_transaction(&tx.transaction_id, CallerPrincipal::Uid(1000))
2304 .unwrap()
2305 .expect("queued transaction approves");
2306 let digest = audit_chain::approval_receipt_digest(&receipt);
2307 assert!(store
2308 .claim_approved_for_execution(&tx.transaction_id, &digest, CallerPrincipal::Uid(1000))
2309 .unwrap());
2310 store
2311 .update_status(&tx.transaction_id, JobState::Succeeded)
2312 .unwrap();
2313
2314 let kinds: Vec<String> = store
2316 .fetch_event_rows()
2317 .unwrap()
2318 .iter()
2319 .map(|e| e.kind.clone())
2320 .collect();
2321 assert!(
2322 kinds.iter().any(|k| k.contains("succeeded")),
2323 "no event records the terminal outcome; the chain covers only {kinds:?}"
2324 );
2325 assert!(
2326 matches!(
2327 store.status_matches_chain().unwrap(),
2328 audit_chain::StatusOutcome::Agrees { .. }
2329 ),
2330 "an untampered store must report agreement, got {:?}",
2331 store.status_matches_chain().unwrap()
2332 );
2333
2334 {
2337 let conn = rusqlite::Connection::open(dir.path().join("tx.db")).unwrap();
2338 let n = conn
2339 .execute(
2340 "UPDATE transactions SET status = ?1 WHERE transaction_id = ?2",
2341 rusqlite::params!["\"canceled\"", tx.transaction_id],
2342 )
2343 .unwrap();
2344 assert_eq!(n, 1, "the fixture did not rewrite the status column");
2345 }
2346 let audit_chain::StatusOutcome::Disagrees { detail } =
2347 store.status_matches_chain().unwrap()
2348 else {
2349 panic!("a rewritten status must be reported, not passed over");
2350 };
2351 assert!(
2352 detail.contains(&tx.transaction_id),
2353 "the report must name the transaction: {detail}"
2354 );
2355 assert!(
2358 matches!(
2359 store.verify_event_chain(&key).unwrap(),
2360 VerifyOutcome::Intact { .. }
2361 ),
2362 "event chain: {:?}, rows: {:?}",
2363 store.verify_event_chain(&key).unwrap(),
2364 store
2365 .fetch_event_rows()
2366 .unwrap()
2367 .iter()
2368 .map(|e| (e.seq, e.kind.clone()))
2369 .collect::<Vec<_>>()
2370 );
2371 }
2372
2373 #[test]
2374 fn a_failed_approval_appends_no_event() {
2375 let dir = tempdir().unwrap();
2379 let store = test_store(dir.path().join("tx.db"));
2380 let tx = store.record(queued_transaction()).unwrap();
2381 store
2382 .approve_transaction(&tx.transaction_id, CallerPrincipal::Uid(1000))
2383 .unwrap();
2384 store
2385 .approve_transaction(&tx.transaction_id, CallerPrincipal::Uid(1000))
2386 .unwrap();
2387 assert_eq!(store.fetch_event_rows().unwrap().len(), 1);
2388
2389 assert!(!store
2391 .revoke_unconsumed_approval("no-such-tx", CallerPrincipal::Uid(1000))
2392 .unwrap());
2393 assert_eq!(store.fetch_event_rows().unwrap().len(), 1);
2394 }
2395
2396 #[test]
2397 fn deleting_the_record_that_an_approval_happened_no_longer_goes_unnoticed() {
2398 let dir = tempdir().unwrap();
2404 let db_path = dir.path().join("tx.db");
2405 let key = AuditKey::from_bytes(vec![0x42; 32]);
2406 let store = test_store(&db_path);
2407
2408 let a = store.record(queued_transaction()).unwrap();
2409 store
2410 .approve_transaction(&a.transaction_id, CallerPrincipal::Uid(1000))
2411 .unwrap();
2412 store.record(queued_transaction()).unwrap();
2415
2416 assert_eq!(
2417 store.verify_event_chain(&key).unwrap(),
2418 VerifyOutcome::Intact { rows_checked: 1 }
2419 );
2420 assert_eq!(
2421 store.verify_event_binding().unwrap(),
2422 audit_chain::BindingOutcome::Consistent {
2423 bindings_checked: 1
2424 }
2425 );
2426
2427 let conn = Connection::open(&db_path).unwrap();
2429 conn.execute("DELETE FROM transaction_approvals", [])
2430 .unwrap();
2431 conn.execute("DELETE FROM audit_events", []).unwrap();
2432 drop(conn);
2433
2434 assert_eq!(
2438 store.verify_event_chain(&key).unwrap(),
2439 VerifyOutcome::Intact { rows_checked: 0 }
2440 );
2441 match store.verify_event_binding().unwrap() {
2442 audit_chain::BindingOutcome::MissingEvent {
2443 transaction_seq, ..
2444 } => assert_eq!(transaction_seq, 2),
2445 other => panic!("expected MissingEvent, got {other:?}"),
2446 }
2447 }
2448
2449 #[test]
2450 fn a_read_only_store_cannot_revoke_or_claim() {
2451 let dir = tempdir().unwrap();
2454 let db_path = dir.path().join("tx.db");
2455 let store = test_store(&db_path);
2456 store.record(queued_transaction()).unwrap();
2457
2458 let read_only = TransactionStore::open_read_only(&db_path).unwrap();
2459 assert!(matches!(
2460 read_only.revoke_unconsumed_approval("tx", CallerPrincipal::Uid(1000)),
2461 Err(TransactionStoreError::AuditChainMissing(_))
2462 ));
2463 assert!(matches!(
2464 read_only.claim_approved_for_execution("tx", "digest", CallerPrincipal::Uid(1000)),
2465 Err(TransactionStoreError::AuditChainMissing(_))
2466 ));
2467 }
2468
2469 #[test]
2476 fn record_persists_the_caller_principal_under_the_current_encoding() {
2477 let dir = tempdir().unwrap();
2478 let store = test_store(dir.path().join("tx.db"));
2479 let mut tx = queued_transaction();
2480 tx.caller_principal = CallerPrincipal::Uid(4242);
2481 store.record(tx).unwrap();
2482
2483 let conn = store.connection().unwrap();
2484 let (version, role, principal): (i64, Option<String>, Option<String>) = conn
2485 .query_row(
2486 "SELECT chain_version, caller_role, caller_principal FROM transactions",
2487 [],
2488 |row| Ok((row.get(0)?, row.get(1)?, row.get(2)?)),
2489 )
2490 .unwrap();
2491
2492 assert_eq!(version, i64::from(audit_chain::CHAIN_VERSION_CURRENT));
2493 assert_eq!(role.as_deref(), Some("dev"));
2494 assert_eq!(
2495 principal.as_deref(),
2496 Some("uid:4242"),
2497 "the row must name the account the dispatcher resolved"
2498 );
2499
2500 let key = AuditKey::from_bytes(vec![0x42; 32]);
2502 assert!(matches!(
2503 store.verify_audit_chain(&key).unwrap(),
2504 audit_chain::VerifyOutcome::Intact { .. }
2505 ));
2506 }
2507
2508 #[test]
2509 fn record_writes_audit_chain_columns() {
2510 let dir = tempdir().unwrap();
2511 let store = test_store(dir.path().join("tx.db"));
2512 let _record = store.record(queued_transaction()).unwrap();
2513
2514 let conn = store.connection().unwrap();
2515 let (seq, key_id, chain_hash, prev): (i64, String, String, String) = conn
2516 .query_row(
2517 "SELECT seq, key_id, chain_hash, prev_chain_hash FROM transactions",
2518 [],
2519 |row| Ok((row.get(0)?, row.get(1)?, row.get(2)?, row.get(3)?)),
2520 )
2521 .unwrap();
2522 assert_eq!(seq, 1, "first row gets seq=1");
2523 assert_eq!(key_id, audit_chain::CURRENT_KEY_ID);
2524 assert_eq!(prev, "", "first row has empty prev_chain_hash");
2525 assert_eq!(
2526 chain_hash.len(),
2527 audit_chain::HASH_HEX_LEN,
2528 "chain_hash is a hex-encoded Ed25519 signature"
2529 );
2530 }
2531
2532 #[test]
2533 fn sequential_records_produce_chained_hashes() {
2534 let dir = tempdir().unwrap();
2535 let store = test_store(dir.path().join("tx.db"));
2536 store.record(queued_transaction()).unwrap();
2537 store.record(queued_transaction()).unwrap();
2538 store.record(queued_transaction()).unwrap();
2539
2540 let rows = store.fetch_chain_rows().unwrap();
2541 assert_eq!(rows.len(), 3);
2542 assert_eq!(rows[0].seq, 1);
2543 assert_eq!(rows[1].seq, 2);
2544 assert_eq!(rows[2].seq, 3);
2545 assert_eq!(rows[1].prev_chain_hash, rows[0].chain_hash);
2547 assert_eq!(rows[2].prev_chain_hash, rows[1].chain_hash);
2548 }
2549
2550 #[test]
2562 fn concurrent_record_keeps_chain_intact_and_seqs_contiguous() {
2563 const WORKERS: usize = 8;
2564 const RECORDS_PER_WORKER: usize = 10;
2565 const TOTAL: usize = WORKERS * RECORDS_PER_WORKER;
2566
2567 let dir = tempdir().unwrap();
2568 let store = std::sync::Arc::new(test_store(dir.path().join("tx.db")));
2569
2570 let mut handles = Vec::with_capacity(WORKERS);
2571 for w in 0..WORKERS {
2572 let store = std::sync::Arc::clone(&store);
2573 handles.push(std::thread::spawn(move || {
2574 for r in 0..RECORDS_PER_WORKER {
2575 let tx = NewTransaction {
2576 request_id: format!("worker-{w}-record-{r}"),
2577 request_hash: format!("hash-{w}-{r}"),
2578 action_name: "GetSystemState".to_string(),
2579 risk_level: RiskLevel::Low,
2580 summary: format!("worker {w} record {r}"),
2581 warnings: vec![],
2582 caller_role: CallerRole::Dev,
2583 caller_principal: CallerPrincipal::Uid(3000 + w as u32),
2587 };
2588 store
2589 .record(tx)
2590 .expect("record must succeed under contention");
2591 }
2592 }));
2593 }
2594 for h in handles {
2595 h.join().expect("worker thread did not panic");
2596 }
2597
2598 let key = AuditKey::from_bytes(vec![0x42; 32]);
2600 let outcome = store.verify_audit_chain(&key).unwrap();
2601 match outcome {
2602 VerifyOutcome::Intact { rows_checked } => {
2603 assert_eq!(
2604 rows_checked, TOTAL as u64,
2605 "expected {TOTAL} rows checked, got {rows_checked}"
2606 );
2607 }
2608 other => panic!("chain must be Intact under concurrent writes; got {other:?}"),
2609 }
2610
2611 {
2615 let conn = store.connection().unwrap();
2616 let mut stmt = conn
2617 .prepare("SELECT summary, caller_principal FROM transactions")
2618 .unwrap();
2619 let pairs: Vec<(String, Option<String>)> = stmt
2620 .query_map([], |row| Ok((row.get(0)?, row.get(1)?)))
2621 .unwrap()
2622 .map(|r| r.unwrap())
2623 .collect();
2624 assert_eq!(pairs.len(), TOTAL);
2625 for (summary, principal) in pairs {
2626 let worker: u32 = summary
2627 .split_whitespace()
2628 .nth(1)
2629 .and_then(|w| w.parse().ok())
2630 .unwrap_or_else(|| panic!("unexpected summary {summary:?}"));
2631 assert_eq!(
2632 principal.as_deref(),
2633 Some(format!("uid:{}", 3000 + worker).as_str()),
2634 "row {summary:?} was attributed to the wrong account"
2635 );
2636 }
2637 }
2638
2639 let conn = store.connection().unwrap();
2641 let mut stmt = conn
2642 .prepare("SELECT seq FROM transactions ORDER BY seq ASC")
2643 .unwrap();
2644 let seqs: Vec<i64> = stmt
2645 .query_map([], |row| row.get::<_, i64>(0))
2646 .unwrap()
2647 .map(|r| r.unwrap())
2648 .collect();
2649 assert_eq!(seqs.len(), TOTAL, "row count mismatch");
2650 for (i, s) in seqs.iter().enumerate() {
2651 assert_eq!(
2652 *s,
2653 (i as i64) + 1,
2654 "seq column must be contiguous 1..={TOTAL}; saw {s} at position {i}"
2655 );
2656 }
2657 }
2658
2659 #[test]
2660 fn verify_audit_chain_intact_after_inserts() {
2661 let dir = tempdir().unwrap();
2662 let store = test_store(dir.path().join("tx.db"));
2663 for _ in 0..3 {
2664 store.record(queued_transaction()).unwrap();
2665 }
2666 let key = AuditKey::from_bytes(vec![0x42; 32]);
2667 let outcome = store.verify_audit_chain(&key).unwrap();
2668 assert!(matches!(outcome, VerifyOutcome::Intact { rows_checked: 3 }));
2669 }
2670
2671 #[test]
2672 fn verify_audit_chain_with_pubkey_intact_after_inserts() {
2673 let dir = tempdir().unwrap();
2674 let store = test_store(dir.path().join("tx.db"));
2675 for _ in 0..3 {
2676 store.record(queued_transaction()).unwrap();
2677 }
2678 let key = AuditKey::from_bytes(vec![0x42; 32]);
2680 let outcome = store
2681 .verify_audit_chain_with_pubkey(&key.verifying_key_hex())
2682 .unwrap();
2683 assert!(matches!(outcome, VerifyOutcome::Intact { rows_checked: 3 }));
2684 }
2685
2686 #[test]
2687 fn verify_audit_chain_with_wrong_pubkey_is_broken() {
2688 let dir = tempdir().unwrap();
2689 let store = test_store(dir.path().join("tx.db"));
2690 store.record(queued_transaction()).unwrap();
2691 let other = AuditKey::from_bytes(vec![0x99; 32]);
2693 let outcome = store
2694 .verify_audit_chain_with_pubkey(&other.verifying_key_hex())
2695 .unwrap();
2696 assert!(matches!(outcome, VerifyOutcome::Broken { .. }));
2697 }
2698
2699 #[test]
2700 fn verify_detects_tampered_summary() {
2701 let dir = tempdir().unwrap();
2702 let store = test_store(dir.path().join("tx.db"));
2703 let tx = store.record(queued_transaction()).unwrap();
2704
2705 let conn = store.connection().unwrap();
2708 conn.execute(
2709 "UPDATE transactions SET summary = ?1 WHERE transaction_id = ?2",
2710 params!["EVIL CHANGE", tx.transaction_id],
2711 )
2712 .unwrap();
2713
2714 let key = AuditKey::from_bytes(vec![0x42; 32]);
2715 let outcome = store.verify_audit_chain(&key).unwrap();
2716 assert!(matches!(outcome, VerifyOutcome::Broken { .. }));
2717 }
2718
2719 #[test]
2720 fn status_update_does_not_break_chain() {
2721 let dir = tempdir().unwrap();
2723 let store = test_store(dir.path().join("tx.db"));
2724 let tx = store.record(queued_transaction()).unwrap();
2725 store
2726 .update_status(&tx.transaction_id, JobState::Running)
2727 .unwrap();
2728 store
2729 .update_status(&tx.transaction_id, JobState::Succeeded)
2730 .unwrap();
2731
2732 let key = AuditKey::from_bytes(vec![0x42; 32]);
2733 let outcome = store.verify_audit_chain(&key).unwrap();
2734 assert!(
2735 matches!(outcome, VerifyOutcome::Intact { rows_checked: 1 }),
2736 "status mutation must not break the chain (status not in chain content): {outcome:?}"
2737 );
2738 }
2739
2740 #[test]
2741 fn open_read_only_rejects_record() {
2742 let dir = tempdir().unwrap();
2743 let key_path = dir.path().join("audit-key");
2744 std::fs::write(&key_path, vec![0x42; 32]).unwrap();
2745 std::fs::set_permissions(&key_path, std::fs::Permissions::from_mode(0o600)).unwrap();
2746
2747 let store = TransactionStore::open_read_only(dir.path().join("tx.db")).unwrap();
2748 let result = store.record(queued_transaction());
2749 assert!(matches!(
2750 result,
2751 Err(TransactionStoreError::AuditChainMissing(_))
2752 ));
2753 }
2754
2755 #[test]
2756 fn ensure_private_dir_creates_with_0700_mode() {
2757 let root = tempdir().unwrap();
2758 let target = root.path().join("a/b/c");
2759 ensure_private_dir(&target).unwrap();
2760 assert!(target.is_dir());
2761 let mode = std::fs::metadata(&target).unwrap().permissions().mode() & 0o777;
2762 assert_eq!(mode, 0o700, "leaf dir must be 0o700, got {mode:o}");
2763 }
2764
2765 #[test]
2773 fn the_audit_database_is_created_owner_only() {
2774 let dir = tempdir().unwrap();
2775 let db = dir.path().join("daemon.sqlite");
2776 let store = TransactionStore::open(&db).unwrap();
2777 store
2779 .record(NewTransaction {
2780 request_id: "mode-probe".to_string(),
2781 request_hash: "hash-mode-probe".to_string(),
2782 action_name: "GetSystemState".to_string(),
2783 risk_level: RiskLevel::Low,
2784 summary: "probe".to_string(),
2785 warnings: vec![],
2786 caller_role: CallerRole::Dev,
2787 caller_principal: CallerPrincipal::Uid(3000),
2788 })
2789 .unwrap();
2790
2791 for path in [db.clone(), db.with_extension("sqlite-wal")] {
2792 if !path.exists() {
2793 continue;
2794 }
2795 let mode = std::fs::metadata(&path).unwrap().permissions().mode() & 0o777;
2796 assert_eq!(
2797 mode,
2798 0o600,
2799 "{} must be owner-only, got {mode:o}",
2800 path.display()
2801 );
2802 }
2803 }
2804
2805 #[test]
2806 fn ensure_private_dir_is_idempotent_and_does_not_widen_existing_mode() {
2807 let root = tempdir().unwrap();
2808 let target = root.path().join("preexisting");
2809 std::fs::create_dir(&target).unwrap();
2810 std::fs::set_permissions(&target, std::fs::Permissions::from_mode(0o755)).unwrap();
2811 ensure_private_dir(&target).unwrap();
2812 let mode = std::fs::metadata(&target).unwrap().permissions().mode() & 0o777;
2814 assert_eq!(mode, 0o755);
2815 }
2816
2817 #[test]
2818 fn open_creates_parent_with_private_mode() {
2819 let root = tempdir().unwrap();
2820 let db_path = root.path().join("nested/dirs/daemon.sqlite");
2821 let _store = test_store(&db_path);
2822 let parent = db_path.parent().unwrap();
2823 let mode = std::fs::metadata(parent).unwrap().permissions().mode() & 0o777;
2824 assert_eq!(mode, 0o700);
2825 }
2826
2827 fn queued_transaction() -> NewTransaction {
2828 NewTransaction {
2829 request_id: "req-1".to_string(),
2830 request_hash: "hash-abc".to_string(),
2831 action_name: "UpdateSystem".to_string(),
2832 risk_level: RiskLevel::High,
2833 summary: "Upgrade the system".to_string(),
2834 warnings: vec![],
2835 caller_role: CallerRole::Dev,
2836 caller_principal: CallerPrincipal::Uid(1000),
2837 }
2838 }
2839
2840 #[test]
2841 fn update_status_transitions_queued_to_running() {
2842 let dir = tempdir().unwrap();
2843 let store = test_store(dir.path().join("tx.db"));
2844 let tx = store.record(queued_transaction()).unwrap();
2845
2846 store
2847 .update_status(&tx.transaction_id, JobState::Running)
2848 .unwrap();
2849
2850 let updated = store.get(&tx.transaction_id).unwrap().unwrap();
2851 assert_eq!(updated.status, JobState::Running);
2852 }
2853
2854 #[test]
2855 fn update_status_transitions_running_to_succeeded() {
2856 let dir = tempdir().unwrap();
2857 let store = test_store(dir.path().join("tx.db"));
2858 let tx = store.record(queued_transaction()).unwrap();
2859
2860 store
2861 .update_status(&tx.transaction_id, JobState::Running)
2862 .unwrap();
2863 store
2864 .update_status(&tx.transaction_id, JobState::Succeeded)
2865 .unwrap();
2866
2867 let updated = store.get(&tx.transaction_id).unwrap().unwrap();
2868 assert_eq!(updated.status, JobState::Succeeded);
2869 }
2870
2871 #[test]
2872 fn update_status_for_unknown_id_returns_not_found() {
2873 let dir = tempdir().unwrap();
2874 let store = test_store(dir.path().join("tx.db"));
2875
2876 let result = store.update_status("does-not-exist", JobState::Running);
2877 assert!(
2878 matches!(result, Err(TransactionStoreError::NotFound(ref id)) if id == "does-not-exist"),
2879 "expected NotFound, got: {result:?}"
2880 );
2881 }
2882
2883 #[test]
2884 fn update_status_leaves_other_fields_intact() {
2885 let dir = tempdir().unwrap();
2886 let store = test_store(dir.path().join("tx.db"));
2887 let tx = store.record(queued_transaction()).unwrap();
2888
2889 store
2890 .update_status(&tx.transaction_id, JobState::Running)
2891 .unwrap();
2892 store
2893 .update_status(&tx.transaction_id, JobState::Failed)
2894 .unwrap();
2895
2896 let updated = store.get(&tx.transaction_id).unwrap().unwrap();
2897 assert_eq!(updated.action_name, "UpdateSystem");
2898 assert_eq!(updated.risk_level, RiskLevel::High);
2899 assert_eq!(updated.status, JobState::Failed);
2900 }
2901
2902 #[test]
2911 fn update_status_is_atomic_under_concurrent_conflicting_transitions() {
2912 let dir = tempdir().unwrap();
2913 let store = std::sync::Arc::new(test_store(dir.path().join("tx.db")));
2914 let tx = store.record(queued_transaction()).unwrap();
2915 store
2916 .update_status(&tx.transaction_id, JobState::Running)
2917 .unwrap();
2918
2919 let store_a = std::sync::Arc::clone(&store);
2920 let id_a = tx.transaction_id.clone();
2921 let succeed = std::thread::spawn(move || store_a.update_status(&id_a, JobState::Succeeded));
2922
2923 let store_b = std::sync::Arc::clone(&store);
2924 let id_b = tx.transaction_id.clone();
2925 let fail = std::thread::spawn(move || store_b.update_status(&id_b, JobState::Failed));
2926
2927 let succeed_result = succeed.join().unwrap();
2928 let fail_result = fail.join().unwrap();
2929
2930 let oks = [succeed_result.is_ok(), fail_result.is_ok()]
2932 .into_iter()
2933 .filter(|ok| *ok)
2934 .count();
2935 assert_eq!(
2936 oks, 1,
2937 "exactly one concurrent transition should succeed: succeed={succeed_result:?} \
2938 fail={fail_result:?}"
2939 );
2940
2941 let loser = if succeed_result.is_ok() {
2957 &fail_result
2958 } else {
2959 &succeed_result
2960 };
2961 match loser {
2962 Err(TransactionStoreError::ConcurrentStatusChange(id)) => {
2963 assert_eq!(id, &tx.transaction_id);
2964 }
2965 Err(TransactionStoreError::InvalidTransition { from, to }) => {
2966 let winner = if succeed_result.is_ok() {
2968 JobState::Succeeded
2969 } else {
2970 JobState::Failed
2971 };
2972 assert_eq!(
2973 *from, winner,
2974 "the loser must have observed the winner's committed status"
2975 );
2976 assert_ne!(*to, winner, "the loser must be the other transition");
2977 }
2978 other => panic!(
2979 "the losing transition must be refused with either \
2980 ConcurrentStatusChange or InvalidTransition, got: {other:?}"
2981 ),
2982 }
2983
2984 let final_status = store.get(&tx.transaction_id).unwrap().unwrap().status;
2987 if succeed_result.is_ok() {
2988 assert_eq!(final_status, JobState::Succeeded);
2989 } else {
2990 assert_eq!(final_status, JobState::Failed);
2991 }
2992 }
2993
2994 #[test]
2995 fn revoke_unconsumed_approval_removes_an_unused_receipt() {
2996 let dir = tempdir().unwrap();
2999 let store = test_store(dir.path().join("tx.db"));
3000 let tx = store.record(queued_transaction()).unwrap();
3001 let receipt = store
3002 .approve_transaction(&tx.transaction_id, CallerPrincipal::Uid(1000))
3003 .unwrap()
3004 .expect("approved");
3005 let digest = audit_chain::approval_receipt_digest(&receipt);
3006
3007 assert!(
3008 store
3009 .revoke_unconsumed_approval(&tx.transaction_id, CallerPrincipal::Uid(1000))
3010 .unwrap(),
3011 "an unconsumed approval must be revocable"
3012 );
3013 assert!(
3014 !store
3015 .claim_approved_for_execution(
3016 &tx.transaction_id,
3017 &digest,
3018 CallerPrincipal::Uid(1000)
3019 )
3020 .unwrap(),
3021 "a revoked receipt must no longer be claimable"
3022 );
3023 }
3024
3025 #[test]
3026 fn revoke_cannot_retract_an_approval_that_was_already_used() {
3027 let dir = tempdir().unwrap();
3031 let store = test_store(dir.path().join("tx.db"));
3032 let tx = store.record(queued_transaction()).unwrap();
3033 let receipt = store
3034 .approve_transaction(&tx.transaction_id, CallerPrincipal::Uid(1000))
3035 .unwrap()
3036 .expect("approved");
3037 let digest = audit_chain::approval_receipt_digest(&receipt);
3038 assert!(store
3039 .claim_approved_for_execution(&tx.transaction_id, &digest, CallerPrincipal::Uid(1000))
3040 .unwrap());
3041
3042 assert!(
3043 !store
3044 .revoke_unconsumed_approval(&tx.transaction_id, CallerPrincipal::Uid(1000))
3045 .unwrap(),
3046 "a consumed approval must not be revocable"
3047 );
3048 assert_eq!(
3049 store.get(&tx.transaction_id).unwrap().unwrap().status,
3050 JobState::Running,
3051 "a failed revocation must not disturb the running transaction"
3052 );
3053 }
3054
3055 #[test]
3056 fn approve_refuses_a_transaction_that_is_no_longer_queued() {
3057 let dir = tempdir().unwrap();
3061 let store = test_store(dir.path().join("tx.db"));
3062 let tx = store.record(queued_transaction()).unwrap();
3063 let receipt = store
3064 .approve_transaction(&tx.transaction_id, CallerPrincipal::Uid(1000))
3065 .unwrap()
3066 .expect("approved");
3067 let digest = audit_chain::approval_receipt_digest(&receipt);
3068 assert!(store
3069 .claim_approved_for_execution(&tx.transaction_id, &digest, CallerPrincipal::Uid(1000))
3070 .unwrap());
3071
3072 assert!(
3074 store
3075 .approve_transaction(&tx.transaction_id, CallerPrincipal::Uid(1000))
3076 .unwrap()
3077 .is_none(),
3078 "a Running transaction must not be approvable"
3079 );
3080
3081 store
3083 .update_status(&tx.transaction_id, JobState::Succeeded)
3084 .unwrap();
3085 assert!(
3086 store
3087 .approve_transaction(&tx.transaction_id, CallerPrincipal::Uid(1000))
3088 .unwrap()
3089 .is_none(),
3090 "a completed transaction must not be approvable"
3091 );
3092 }
3093
3094 #[test]
3095 fn only_one_of_two_concurrent_claims_can_execute() {
3096 let dir = tempdir().unwrap();
3101 let store = std::sync::Arc::new(test_store(dir.path().join("tx.db")));
3102 let tx = store.record(queued_transaction()).unwrap();
3103 let receipt = store
3104 .approve_transaction(&tx.transaction_id, CallerPrincipal::Uid(1000))
3105 .unwrap()
3106 .expect("approved");
3107 let digest = audit_chain::approval_receipt_digest(&receipt);
3108
3109 let mut handles = Vec::new();
3110 for _ in 0..2 {
3111 let store = std::sync::Arc::clone(&store);
3112 let id = tx.transaction_id.clone();
3113 let digest = digest.clone();
3114 handles.push(std::thread::spawn(move || {
3115 store.claim_approved_for_execution(&id, &digest, CallerPrincipal::Uid(1000))
3116 }));
3117 }
3118 let claims: Vec<bool> = handles
3119 .into_iter()
3120 .map(|h| h.join().unwrap().unwrap_or(false))
3121 .collect();
3122
3123 assert_eq!(
3124 claims.iter().filter(|c| **c).count(),
3125 1,
3126 "exactly one claim may win: {claims:?}"
3127 );
3128 assert_eq!(
3129 store.get(&tx.transaction_id).unwrap().unwrap().status,
3130 JobState::Running
3131 );
3132 }
3133
3134 #[test]
3135 fn cancel_queued_cancels_a_queued_transaction_once() {
3136 let dir = tempdir().unwrap();
3137 let store = test_store(dir.path().join("tx.db"));
3138 let tx = store.record(queued_transaction()).unwrap();
3139
3140 assert!(
3141 store
3142 .cancel_queued(&tx.transaction_id, CallerPrincipal::Uid(1000))
3143 .unwrap(),
3144 "a queued transaction is cancelable"
3145 );
3146 assert_eq!(
3147 store.get(&tx.transaction_id).unwrap().unwrap().status,
3148 JobState::Canceled
3149 );
3150 assert!(
3151 !store
3152 .cancel_queued(&tx.transaction_id, CallerPrincipal::Uid(1000))
3153 .unwrap(),
3154 "an already-canceled transaction is not cancelable again"
3155 );
3156 assert!(
3157 !store
3158 .cancel_queued("no-such-transaction", CallerPrincipal::Uid(1000))
3159 .unwrap(),
3160 "a missing transaction is not cancelable"
3161 );
3162 assert!(
3163 store
3164 .fetch_event_rows()
3165 .unwrap()
3166 .iter()
3167 .all(|event| event.kind == "status_canceled"),
3168 "canceling an unapproved transaction must not append an approval event"
3169 );
3170 }
3171
3172 #[test]
3173 fn cancel_queued_revokes_an_unconsumed_approval_and_appends_event() {
3174 let dir = tempdir().unwrap();
3175 let store = test_store(dir.path().join("tx.db"));
3176 let tx = store.record(queued_transaction()).unwrap();
3177 let receipt = store
3178 .approve_transaction(&tx.transaction_id, CallerPrincipal::Uid(1000))
3179 .unwrap()
3180 .expect("queued transaction is approvable");
3181 let digest = audit_chain::approval_receipt_digest(&receipt);
3182
3183 assert!(store
3184 .cancel_queued(&tx.transaction_id, CallerPrincipal::Uid(1000))
3185 .unwrap());
3186 assert_eq!(
3187 store.get(&tx.transaction_id).unwrap().unwrap().status,
3188 JobState::Canceled
3189 );
3190 assert!(
3191 !store
3192 .claim_approved_for_execution(
3193 &tx.transaction_id,
3194 &digest,
3195 CallerPrincipal::Uid(1000)
3196 )
3197 .unwrap(),
3198 "canceling must revoke the unconsumed receipt"
3199 );
3200 assert_eq!(
3201 store
3202 .fetch_event_rows()
3203 .unwrap()
3204 .iter()
3205 .map(|event| event.kind.as_str())
3206 .collect::<Vec<_>>(),
3207 vec!["approval_granted", "approval_revoked", "status_canceled"]
3208 );
3209 }
3210
3211 #[test]
3212 fn cancel_queued_refuses_a_running_transaction() {
3213 let dir = tempdir().unwrap();
3216 let store = test_store(dir.path().join("tx.db"));
3217 let tx = store.record(queued_transaction()).unwrap();
3218 let receipt = store
3219 .approve_transaction(&tx.transaction_id, CallerPrincipal::Uid(1000))
3220 .unwrap()
3221 .expect("approved");
3222 let digest = audit_chain::approval_receipt_digest(&receipt);
3223 assert!(store
3224 .claim_approved_for_execution(&tx.transaction_id, &digest, CallerPrincipal::Uid(1000))
3225 .unwrap());
3226
3227 assert!(
3228 !store
3229 .cancel_queued(&tx.transaction_id, CallerPrincipal::Uid(1000))
3230 .unwrap(),
3231 "a running transaction must not be cancelable"
3232 );
3233 assert_eq!(
3234 store.get(&tx.transaction_id).unwrap().unwrap().status,
3235 JobState::Running,
3236 "cancel must not disturb a running transaction"
3237 );
3238 }
3239
3240 #[test]
3241 fn approved_receipt_is_required_and_consumed_once() {
3242 let dir = tempdir().unwrap();
3243 let store = test_store(dir.path().join("tx.db"));
3244 let tx = store.record(queued_transaction()).unwrap();
3245
3246 assert!(
3247 !store
3248 .claim_approved_for_execution(
3249 &tx.transaction_id,
3250 "digest-a",
3251 CallerPrincipal::Uid(1000)
3252 )
3253 .unwrap(),
3254 "an unapproved preview must not execute"
3255 );
3256 let receipt = store
3257 .approve_transaction(&tx.transaction_id, CallerPrincipal::Uid(1000))
3258 .unwrap()
3259 .expect("first approval must succeed");
3260 let digest = audit_chain::approval_receipt_digest(&receipt);
3261 assert_eq!(tx.approval_id.as_deref(), Some(digest.as_str()));
3262 assert!(
3263 store
3264 .approve_transaction(&tx.transaction_id, CallerPrincipal::Uid(1000))
3265 .unwrap()
3266 .is_none(),
3267 "approval is immutable once issued"
3268 );
3269 assert!(
3270 !store
3271 .claim_approved_for_execution(
3272 &tx.transaction_id,
3273 "wrong-digest",
3274 CallerPrincipal::Uid(1000)
3275 )
3276 .unwrap(),
3277 "a forged receipt must not execute"
3278 );
3279 assert!(
3280 store
3281 .claim_approved_for_execution(
3282 &tx.transaction_id,
3283 &digest,
3284 CallerPrincipal::Uid(1000)
3285 )
3286 .unwrap(),
3287 "the exact approved receipt must execute"
3288 );
3289 assert!(
3290 !store
3291 .claim_approved_for_execution(
3292 &tx.transaction_id,
3293 &digest,
3294 CallerPrincipal::Uid(1000)
3295 )
3296 .unwrap(),
3297 "the receipt must be one-time"
3298 );
3299 }
3300
3301 #[test]
3302 fn approval_commitment_is_covered_by_the_signed_chain() {
3303 let dir = tempdir().unwrap();
3304 let store = test_store(dir.path().join("tx.db"));
3305 let tx = store.record(queued_transaction()).unwrap();
3306 assert!(tx.approval_id.is_some());
3307
3308 store
3309 .connection()
3310 .unwrap()
3311 .execute(
3312 "UPDATE transactions SET approval_id = 'forged' WHERE transaction_id = ?1",
3313 params![tx.transaction_id],
3314 )
3315 .unwrap();
3316
3317 let key = AuditKey::from_bytes(vec![0x42; 32]);
3318 assert!(matches!(
3319 store.verify_audit_chain(&key).unwrap(),
3320 VerifyOutcome::Broken { .. }
3321 ));
3322 }
3323
3324 #[test]
3325 fn approve_rejects_a_forged_commitment_at_runtime() {
3326 let dir = tempdir().unwrap();
3330 let store = test_store(dir.path().join("tx.db"));
3331 let tx = store.record(queued_transaction()).unwrap();
3332
3333 store
3334 .connection()
3335 .unwrap()
3336 .execute(
3337 "UPDATE transactions SET approval_id = 'forged' WHERE transaction_id = ?1",
3338 params![tx.transaction_id],
3339 )
3340 .unwrap();
3341
3342 let err = store
3343 .approve_transaction(&tx.transaction_id, CallerPrincipal::Uid(1000))
3344 .expect_err("a forged commitment must be rejected, not approved");
3345 assert!(
3346 matches!(err, TransactionStoreError::DatabaseInvariant(_)),
3347 "forged commitment must surface as a DatabaseInvariant, got {err:?}"
3348 );
3349 }
3350
3351 #[test]
3352 fn stale_iso_timestamp_cannot_be_approved() {
3353 let dir = tempdir().unwrap();
3354 let store = test_store(dir.path().join("tx.db"));
3355 let tx = store.record(queued_transaction()).unwrap();
3356 let conn = store.connection().unwrap();
3357 conn.execute(
3358 "UPDATE transactions \
3359 SET created_at = strftime('%Y-%m-%dT%H:%M:%fZ', 'now', '-20 minutes') \
3360 WHERE transaction_id = ?1",
3361 params![tx.transaction_id],
3362 )
3363 .unwrap();
3364
3365 assert!(
3366 store
3367 .approve_transaction(&tx.transaction_id, CallerPrincipal::Uid(1000))
3368 .unwrap()
3369 .is_none(),
3370 "a production-format timestamp outside the TTL must not be approved"
3371 );
3372 }
3373
3374 #[test]
3375 fn stale_iso_timestamp_cannot_be_claimed_for_execution() {
3376 let dir = tempdir().unwrap();
3381 let store = test_store(dir.path().join("tx.db"));
3382 let tx = store.record(queued_transaction()).unwrap();
3383 let receipt = store
3384 .approve_transaction(&tx.transaction_id, CallerPrincipal::Uid(1000))
3385 .unwrap()
3386 .expect("a fresh approval succeeds");
3387 let digest = audit_chain::approval_receipt_digest(&receipt);
3388
3389 let conn = store.connection().unwrap();
3391 conn.execute(
3392 "UPDATE transactions \
3393 SET created_at = strftime('%Y-%m-%dT%H:%M:%fZ', 'now', '-20 minutes') \
3394 WHERE transaction_id = ?1",
3395 params![tx.transaction_id],
3396 )
3397 .unwrap();
3398
3399 assert!(
3400 !store
3401 .claim_approved_for_execution(
3402 &tx.transaction_id,
3403 &digest,
3404 CallerPrincipal::Uid(1000)
3405 )
3406 .unwrap(),
3407 "an approval aged past the TTL must not be claimable at execute time"
3408 );
3409 assert_eq!(
3410 store.get(&tx.transaction_id).unwrap().unwrap().status,
3411 JobState::Queued,
3412 "a TTL-expired claim must leave the transaction Queued, never Running"
3413 );
3414 }
3415
3416 #[test]
3417 fn cleanup_stale_queued_cancels_old_records() {
3418 let dir = tempdir().unwrap();
3419 let store = test_store(dir.path().join("tx.db"));
3420
3421 let fresh = store.record(queued_transaction()).unwrap();
3423 let stale = store.record(queued_transaction()).unwrap();
3424
3425 let conn = store.connection().unwrap();
3427 conn.execute(
3428 "UPDATE transactions \
3429 SET created_at = strftime('%Y-%m-%dT%H:%M:%fZ', 'now', '-20 minutes') \
3430 WHERE transaction_id = ?1",
3431 params![stale.transaction_id],
3432 )
3433 .unwrap();
3434
3435 let canceled = store.cleanup_stale_queued().unwrap();
3436 assert_eq!(canceled, 1, "only the stale record should be canceled");
3437
3438 let stale_record = store.get(&stale.transaction_id).unwrap().unwrap();
3440 assert_eq!(stale_record.status, JobState::Canceled);
3441
3442 let fresh_record = store.get(&fresh.transaction_id).unwrap().unwrap();
3444 assert_eq!(fresh_record.status, JobState::Queued);
3445 }
3446
3447 #[test]
3448 fn cleanup_stale_queued_revokes_every_unconsumed_approval() {
3449 let dir = tempdir().unwrap();
3450 let store = test_store(dir.path().join("tx.db"));
3451 let transactions: Vec<_> = (0..3)
3452 .map(|_| {
3453 let transaction = store.record(queued_transaction()).unwrap();
3454 store
3455 .approve_transaction(&transaction.transaction_id, CallerPrincipal::Uid(1000))
3456 .unwrap()
3457 .expect("stale transaction is approvable");
3458 transaction
3459 })
3460 .collect();
3461
3462 let conn = store.connection().unwrap();
3463 for transaction in &transactions {
3464 conn.execute(
3465 "UPDATE transactions \
3466 SET created_at = strftime('%Y-%m-%dT%H:%M:%fZ', 'now', '-20 minutes') \
3467 WHERE transaction_id = ?1",
3468 params![transaction.transaction_id],
3469 )
3470 .unwrap();
3471 }
3472
3473 assert_eq!(store.cleanup_stale_queued().unwrap(), 3);
3474 for transaction in &transactions {
3475 assert_eq!(
3476 store
3477 .get(&transaction.transaction_id)
3478 .unwrap()
3479 .unwrap()
3480 .status,
3481 JobState::Canceled
3482 );
3483 assert!(!store
3484 .revoke_unconsumed_approval(&transaction.transaction_id, CallerPrincipal::Uid(1000))
3485 .unwrap());
3486 }
3487 assert_eq!(
3488 store
3489 .fetch_event_rows()
3490 .unwrap()
3491 .iter()
3492 .map(|event| event.kind.as_str())
3493 .collect::<Vec<_>>(),
3494 vec![
3495 "approval_granted",
3496 "approval_granted",
3497 "approval_granted",
3498 "approval_revoked",
3499 "status_canceled",
3500 "approval_revoked",
3501 "status_canceled",
3502 "approval_revoked",
3503 "status_canceled"
3504 ]
3505 );
3506 }
3507
3508 #[test]
3511 fn update_status_rejects_queued_to_succeeded() {
3512 let dir = tempdir().unwrap();
3513 let store = test_store(dir.path().join("tx.db"));
3514 let tx = store.record(queued_transaction()).unwrap();
3515
3516 let result = store.update_status(&tx.transaction_id, JobState::Succeeded);
3517 assert!(
3518 matches!(
3519 result,
3520 Err(TransactionStoreError::InvalidTransition {
3521 from: JobState::Queued,
3522 to: JobState::Succeeded,
3523 })
3524 ),
3525 "Queued -> Succeeded must be rejected (must go through Running first): {result:?}"
3526 );
3527 }
3528
3529 #[test]
3530 fn update_status_rejects_succeeded_to_running() {
3531 let dir = tempdir().unwrap();
3532 let store = test_store(dir.path().join("tx.db"));
3533 let tx = store.record(queued_transaction()).unwrap();
3534
3535 store
3536 .update_status(&tx.transaction_id, JobState::Running)
3537 .unwrap();
3538 store
3539 .update_status(&tx.transaction_id, JobState::Succeeded)
3540 .unwrap();
3541
3542 let result = store.update_status(&tx.transaction_id, JobState::Running);
3543 assert!(
3544 matches!(
3545 result,
3546 Err(TransactionStoreError::InvalidTransition {
3547 from: JobState::Succeeded,
3548 to: JobState::Running,
3549 })
3550 ),
3551 "Succeeded -> Running must be rejected (terminal state): {result:?}"
3552 );
3553 }
3554
3555 #[test]
3556 fn update_status_accepts_running_to_failed() {
3557 let dir = tempdir().unwrap();
3558 let store = test_store(dir.path().join("tx.db"));
3559 let tx = store.record(queued_transaction()).unwrap();
3560
3561 store
3562 .update_status(&tx.transaction_id, JobState::Running)
3563 .unwrap();
3564 store
3565 .update_status(&tx.transaction_id, JobState::Failed)
3566 .unwrap();
3567
3568 let updated = store.get(&tx.transaction_id).unwrap().unwrap();
3569 assert_eq!(updated.status, JobState::Failed);
3570 }
3571
3572 #[test]
3573 fn update_status_accepts_running_to_rolled_back() {
3574 let dir = tempdir().unwrap();
3575 let store = test_store(dir.path().join("tx.db"));
3576 let tx = store.record(queued_transaction()).unwrap();
3577
3578 store
3579 .update_status(&tx.transaction_id, JobState::Running)
3580 .unwrap();
3581 store
3582 .update_status(&tx.transaction_id, JobState::RolledBack)
3583 .unwrap();
3584
3585 let updated = store.get(&tx.transaction_id).unwrap().unwrap();
3586 assert_eq!(updated.status, JobState::RolledBack);
3587 }
3588
3589 #[test]
3592 fn list_transactions_returns_empty_for_fresh_store() {
3593 let dir = tempdir().unwrap();
3594 let store = test_store(dir.path().join("tx.db"));
3595 let results = store.list_transactions(10, None, None, None).unwrap();
3596 assert!(results.is_empty());
3597 }
3598
3599 #[test]
3600 fn list_transactions_returns_all_records_ordered_by_newest_first() {
3601 let dir = tempdir().unwrap();
3602 let store = test_store(dir.path().join("tx.db"));
3603 store.record(queued_transaction()).unwrap();
3604
3605 let mut second = queued_transaction();
3606 second.action_name = "GetDiskUsage".to_string();
3607 second.risk_level = RiskLevel::Low;
3608 store.record(second).unwrap();
3609
3610 let results = store.list_transactions(10, None, None, None).unwrap();
3611 assert_eq!(results.len(), 2);
3612 assert_eq!(results[0].action_name, "GetDiskUsage");
3614 assert_eq!(results[1].action_name, "UpdateSystem");
3615 }
3616
3617 #[test]
3618 fn list_history_populates_created_at_and_risk_level() {
3619 let dir = tempdir().unwrap();
3620 let store = test_store(dir.path().join("tx.db"));
3621 store.record(queued_transaction()).unwrap();
3622
3623 let entries = store.list_history(10, None, None, None).unwrap();
3624 assert_eq!(entries.len(), 1);
3625 let entry = &entries[0];
3626 assert_eq!(entry.action_name, "UpdateSystem");
3627 assert_eq!(entry.risk_level, RiskLevel::High);
3628 assert_eq!(entry.status, JobState::Queued);
3629 assert!(
3630 !entry.created_at.is_empty(),
3631 "created_at must be populated from the stored row, not left blank"
3632 );
3633 }
3634
3635 #[test]
3636 fn list_history_applies_the_same_filters_as_list_transactions() {
3637 let dir = tempdir().unwrap();
3638 let store = test_store(dir.path().join("tx.db"));
3639 store.record(queued_transaction()).unwrap();
3640 let mut low = queued_transaction();
3641 low.action_name = "GetDiskUsage".to_string();
3642 low.risk_level = RiskLevel::Low;
3643 store.record(low).unwrap();
3644
3645 let only = store
3646 .list_history(10, None, Some("GetDiskUsage"), None)
3647 .unwrap();
3648 assert_eq!(only.len(), 1);
3649 assert_eq!(only[0].action_name, "GetDiskUsage");
3650 }
3651
3652 #[test]
3653 fn list_transactions_respects_limit() {
3654 let dir = tempdir().unwrap();
3655 let store = test_store(dir.path().join("tx.db"));
3656 for _ in 0..5 {
3657 store.record(queued_transaction()).unwrap();
3658 }
3659 let results = store.list_transactions(3, None, None, None).unwrap();
3660 assert_eq!(results.len(), 3);
3661 }
3662
3663 #[test]
3664 fn list_transactions_filters_by_status() {
3665 let dir = tempdir().unwrap();
3666 let store = test_store(dir.path().join("tx.db"));
3667 let tx = store.record(queued_transaction()).unwrap();
3668 store
3669 .update_status(&tx.transaction_id, JobState::Running)
3670 .unwrap();
3671 store
3672 .update_status(&tx.transaction_id, JobState::Succeeded)
3673 .unwrap();
3674
3675 store.record(queued_transaction()).unwrap();
3677
3678 let succeeded = store
3679 .list_transactions(10, Some("succeeded"), None, None)
3680 .unwrap();
3681 assert_eq!(succeeded.len(), 1);
3682 assert_eq!(succeeded[0].status, JobState::Succeeded);
3683
3684 let queued = store
3685 .list_transactions(10, Some("queued"), None, None)
3686 .unwrap();
3687 assert_eq!(queued.len(), 1);
3688 assert_eq!(queued[0].status, JobState::Queued);
3689 }
3690
3691 #[test]
3692 fn list_transactions_filters_by_action_name() {
3693 let dir = tempdir().unwrap();
3694 let store = test_store(dir.path().join("tx.db"));
3695 store.record(queued_transaction()).unwrap(); let mut disk = queued_transaction();
3698 disk.action_name = "GetDiskUsage".to_string();
3699 store.record(disk).unwrap();
3700
3701 let results = store
3702 .list_transactions(10, None, Some("GetDiskUsage"), None)
3703 .unwrap();
3704 assert_eq!(results.len(), 1);
3705 assert_eq!(results[0].action_name, "GetDiskUsage");
3706 }
3707
3708 #[test]
3709 fn list_transactions_filters_by_since_hours() {
3710 let dir = tempdir().unwrap();
3711 let store = test_store(dir.path().join("tx.db"));
3712
3713 let old = store.record(queued_transaction()).unwrap();
3715 let conn = store.connection().unwrap();
3716 conn.execute(
3717 "UPDATE transactions \
3718 SET created_at = strftime('%Y-%m-%dT%H:%M:%fZ', 'now', '-48 hours') \
3719 WHERE transaction_id = ?1",
3720 params![old.transaction_id],
3721 )
3722 .unwrap();
3723
3724 store.record(queued_transaction()).unwrap();
3726
3727 let results = store
3729 .list_transactions(10, None, None, Some(HistorySince::Hours(24)))
3730 .unwrap();
3731 assert_eq!(results.len(), 1);
3732
3733 let results = store
3735 .list_transactions(10, None, None, Some(HistorySince::Hours(72)))
3736 .unwrap();
3737 assert_eq!(results.len(), 2);
3738 }
3739
3740 #[test]
3741 fn list_history_filters_by_exact_cutoff_below_one_hour() {
3742 let dir = tempdir().unwrap();
3746 let store = test_store(dir.path().join("tx.db"));
3747 let conn = store.connection().unwrap();
3748 for minutes_ago in [75, 20] {
3749 let tx = store.record(queued_transaction()).unwrap();
3750 conn.execute(
3751 "UPDATE transactions \
3752 SET created_at = strftime('%Y-%m-%dT%H:%M:%fZ', 'now', ?1) \
3753 WHERE transaction_id = ?2",
3754 params![format!("-{minutes_ago} minutes"), tx.transaction_id],
3755 )
3756 .unwrap();
3757 }
3758
3759 let cutoff = |minutes: i64| {
3760 Some(HistorySince::Cutoff(
3761 chrono::Utc::now() - chrono::Duration::minutes(minutes),
3762 ))
3763 };
3764 assert_eq!(
3765 store
3766 .list_history(10, None, None, cutoff(30))
3767 .unwrap()
3768 .len(),
3769 1
3770 );
3771 assert_eq!(
3772 store
3773 .list_history(10, None, None, cutoff(90))
3774 .unwrap()
3775 .len(),
3776 2
3777 );
3778 assert_eq!(
3779 store
3780 .list_history(10, None, None, cutoff(10))
3781 .unwrap()
3782 .len(),
3783 0
3784 );
3785 assert_eq!(
3786 store
3787 .list_transactions(10, None, None, cutoff(30))
3788 .unwrap()
3789 .len(),
3790 1
3791 );
3792 }
3793
3794 #[test]
3795 fn list_history_cutoff_normalises_offsets_and_legacy_rows() {
3796 let dir = tempdir().unwrap();
3799 let store = test_store(dir.path().join("tx.db"));
3800 let conn = store.connection().unwrap();
3801 let ten_min_ago = chrono::Utc::now() - chrono::Duration::minutes(10);
3802 let shapes = [
3803 ten_min_ago
3804 .with_timezone(&chrono::FixedOffset::east_opt(2 * 3600).unwrap())
3805 .to_rfc3339(),
3806 ten_min_ago.format("%Y-%m-%d %H:%M:%S").to_string(),
3807 ];
3808 for created_at in &shapes {
3809 let tx = store.record(queued_transaction()).unwrap();
3810 conn.execute(
3811 "UPDATE transactions SET created_at = ?1 WHERE transaction_id = ?2",
3812 params![created_at, tx.transaction_id],
3813 )
3814 .unwrap();
3815 }
3816
3817 let at = |minutes: i64| {
3818 Some(HistorySince::Cutoff(
3819 chrono::Utc::now() - chrono::Duration::minutes(minutes),
3820 ))
3821 };
3822 assert_eq!(store.list_history(10, None, None, at(15)).unwrap().len(), 2);
3823 assert_eq!(store.list_history(10, None, None, at(5)).unwrap().len(), 0);
3824 }
3825
3826 #[test]
3827 fn list_transactions_rejects_invalid_status_filter() {
3828 let dir = tempdir().unwrap();
3829 let store = test_store(dir.path().join("tx.db"));
3830 store.record(queued_transaction()).unwrap();
3831 let result = store.list_transactions(10, Some("bogus"), None, None);
3832 assert!(result.is_err(), "invalid status filter should return error");
3833 }
3834
3835 #[test]
3843 fn record_emits_one_watermark_per_entry() {
3844 let sink = std::sync::Arc::new(std::sync::Mutex::new(Vec::new()));
3845 crate::audit_watermark::install_test_sink(std::sync::Arc::clone(&sink));
3846
3847 let dir = tempdir().unwrap();
3848 let store = test_store(dir.path().join("tx.db"));
3849 store.record(queued_transaction()).unwrap();
3850
3851 let calls = crate::audit_watermark::take_watermarks(&sink);
3852 assert_eq!(calls.len(), 1, "expected exactly 1 watermark per record()");
3853 }
3854
3855 #[test]
3857 fn record_previewed_emits_one_watermark() {
3858 let sink = std::sync::Arc::new(std::sync::Mutex::new(Vec::new()));
3859 crate::audit_watermark::install_test_sink(std::sync::Arc::clone(&sink));
3860
3861 let dir = tempdir().unwrap();
3862 let store = test_store(dir.path().join("tx.db"));
3863 let preview = PreviewEnvelope {
3864 summary: "Upgrade the system".to_string(),
3865 risk_level: RiskLevel::High,
3866 current_state: serde_json::Value::Null,
3867 proposed_change: serde_json::Value::Null,
3868 expected_side_effects: vec![],
3869 reboot_required: false,
3870 rollback_available: false,
3871 warnings: vec![],
3872 request_hash: sysknife_types::RequestHash::from("hash-abc".to_string()),
3873 };
3874 store
3875 .record_previewed(queued_transaction(), preview)
3876 .unwrap();
3877
3878 let calls = crate::audit_watermark::take_watermarks(&sink);
3879 assert_eq!(
3880 calls.len(),
3881 1,
3882 "expected exactly 1 watermark per record_previewed()"
3883 );
3884 }
3885
3886 #[test]
3888 fn watermark_seq_and_hash_match_chain_row() {
3889 let sink = std::sync::Arc::new(std::sync::Mutex::new(Vec::new()));
3890 crate::audit_watermark::install_test_sink(std::sync::Arc::clone(&sink));
3891
3892 let dir = tempdir().unwrap();
3893 let store = test_store(dir.path().join("tx.db"));
3894 store.record(queued_transaction()).unwrap();
3895
3896 let rows = store.fetch_chain_rows().unwrap();
3897 assert_eq!(rows.len(), 1);
3898 let row = &rows[0];
3899
3900 let calls = crate::audit_watermark::take_watermarks(&sink);
3901 assert_eq!(calls.len(), 1);
3902 assert_eq!(calls[0].seq, row.seq, "watermark seq must match chain row");
3903 assert_eq!(
3904 calls[0].chain_hash_hex, row.chain_hash,
3905 "watermark chain_hash_hex must match stored chain_hash"
3906 );
3907 }
3908
3909 #[test]
3911 fn multiple_records_produce_one_watermark_each() {
3912 let sink = std::sync::Arc::new(std::sync::Mutex::new(Vec::new()));
3913 crate::audit_watermark::install_test_sink(std::sync::Arc::clone(&sink));
3914
3915 let dir = tempdir().unwrap();
3916 let store = test_store(dir.path().join("tx.db"));
3917 for _ in 0..3 {
3918 store.record(queued_transaction()).unwrap();
3919 }
3920
3921 let calls = crate::audit_watermark::take_watermarks(&sink);
3922 assert_eq!(calls.len(), 3, "one watermark per record call");
3923 assert_eq!(calls[0].seq, 1);
3924 assert_eq!(calls[1].seq, 2);
3925 assert_eq!(calls[2].seq, 3);
3926 }
3927
3928 #[test]
3942 fn no_watermark_emitted_before_any_record() {
3943 let sink = std::sync::Arc::new(std::sync::Mutex::new(Vec::new()));
3944 crate::audit_watermark::install_test_sink(std::sync::Arc::clone(&sink));
3945
3946 let dir = tempdir().unwrap();
3947 let _store = test_store(dir.path().join("tx.db"));
3948
3949 let calls = crate::audit_watermark::take_watermarks(&sink);
3951 assert!(
3952 calls.is_empty(),
3953 "no watermark must be emitted without a record() call"
3954 );
3955 }
3956}