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, Debug, PartialEq, Eq)]
50pub struct NewTransaction {
51 pub request_id: String,
52 pub request_hash: String,
53 pub action_name: String,
54 pub risk_level: RiskLevel,
55 pub summary: String,
75 pub warnings: Vec<String>,
76 pub caller_role: CallerRole,
85 pub caller_principal: CallerPrincipal,
93}
94
95struct SqliteMigration {
104 version: i64,
105 name: &'static str,
106 sql: &'static str,
107}
108
109const SQLITE_MIGRATIONS: &[SqliteMigration] = &[
127 SqliteMigration {
128 version: 1,
129 name: "initial_audit_schema",
130 sql: r#"
131 CREATE TABLE IF NOT EXISTS transactions (
132 transaction_id TEXT PRIMARY KEY,
133 request_id TEXT NOT NULL,
134 request_hash TEXT NOT NULL,
135 action_name TEXT NOT NULL,
136 risk_level TEXT NOT NULL,
137 status TEXT NOT NULL,
138 approval_id TEXT,
139 summary TEXT NOT NULL,
140 warnings_json TEXT NOT NULL,
141 created_at TEXT NOT NULL DEFAULT (datetime('now')),
142 seq INTEGER NOT NULL UNIQUE,
143 key_id TEXT NOT NULL,
144 chain_hash TEXT NOT NULL,
145 prev_chain_hash TEXT NOT NULL DEFAULT ''
146 );
147
148 CREATE TABLE IF NOT EXISTS transaction_previews (
149 transaction_id TEXT PRIMARY KEY,
150 preview_json TEXT NOT NULL
151 );
152
153 CREATE TABLE IF NOT EXISTS transaction_approvals (
154 transaction_id TEXT PRIMARY KEY,
155 receipt_digest TEXT NOT NULL,
156 approved_at TEXT NOT NULL DEFAULT (datetime('now')),
157 consumed_at TEXT
158 );
159
160 CREATE INDEX IF NOT EXISTS transactions_seq_idx ON transactions(seq);
161 "#,
162 },
163 SqliteMigration {
172 version: 2,
173 name: "caller_identity_and_approval_events",
174 sql: r#"
175 ALTER TABLE transactions ADD COLUMN chain_version INTEGER NOT NULL DEFAULT 1;
176 ALTER TABLE transactions ADD COLUMN caller_role TEXT;
177 ALTER TABLE transactions ADD COLUMN event_tip TEXT;
178
179 CREATE TABLE IF NOT EXISTS audit_events (
180 seq INTEGER PRIMARY KEY,
181 key_id TEXT NOT NULL,
182 kind TEXT NOT NULL,
183 transaction_id TEXT NOT NULL,
184 receipt_digest TEXT NOT NULL,
185 created_at TEXT NOT NULL,
186 chain_hash TEXT NOT NULL,
187 prev_chain_hash TEXT NOT NULL DEFAULT ''
188 );
189
190 CREATE INDEX IF NOT EXISTS audit_events_transaction_idx
191 ON audit_events(transaction_id);
192 "#,
193 },
194 SqliteMigration {
199 version: 3,
200 name: "caller_principal",
201 sql: r#"
202 ALTER TABLE transactions ADD COLUMN caller_principal TEXT;
203 "#,
204 },
205 SqliteMigration {
214 version: 4,
215 name: "event_approver_identity",
216 sql: r#"
217 ALTER TABLE audit_events ADD COLUMN chain_version INTEGER NOT NULL DEFAULT 1;
218 ALTER TABLE audit_events ADD COLUMN caller_principal TEXT;
219 "#,
220 },
221];
222
223use crate::store::CHAIN_ROW_COLUMNS;
227
228fn chain_row_from_sqlite(row: &rusqlite::Row<'_>) -> rusqlite::Result<ChainRow> {
229 Ok(ChainRow {
230 seq: row.get::<_, i64>(0)? as u64,
231 key_id: row.get(1)?,
232 transaction_id: row.get(2)?,
233 request_id: row.get(3)?,
234 request_hash: row.get(4)?,
235 action_name: row.get(5)?,
236 risk_level: deserialize_field(&row.get::<_, String>(6)?).map_err(|e| {
237 rusqlite::Error::FromSqlConversionFailure(6, rusqlite::types::Type::Text, Box::new(e))
238 })?,
239 summary: row.get(7)?,
240 approval_id: row.get(8)?,
241 warnings_json: row.get(9)?,
242 created_at: row.get(10)?,
243 prev_chain_hash: row.get(11)?,
244 chain_hash: row.get(12)?,
245 chain_version: row.get::<_, i64>(13)? as u32,
246 caller_role: row.get(14)?,
247 event_tip: row.get(15)?,
248 caller_principal: row.get(16)?,
249 })
250}
251
252fn event_row_from_sqlite(row: &rusqlite::Row<'_>) -> rusqlite::Result<EventRow> {
253 Ok(EventRow {
254 seq: row.get::<_, i64>(0)? as u64,
255 key_id: row.get(1)?,
256 kind: row.get(2)?,
257 transaction_id: row.get(3)?,
258 receipt_digest: row.get(4)?,
259 created_at: row.get(5)?,
260 prev_chain_hash: row.get(6)?,
261 chain_hash: row.get(7)?,
262 chain_version: row.get::<_, i64>(8)? as u32,
263 caller_principal: row.get(9)?,
264 })
265}
266
267fn query_optional<T>(
273 conn: &Connection,
274 sql: &str,
275 params: impl rusqlite::Params,
276 f: impl FnOnce(&rusqlite::Row<'_>) -> rusqlite::Result<T>,
277) -> rusqlite::Result<Option<T>> {
278 match conn.query_row(sql, params, f) {
279 Ok(v) => Ok(Some(v)),
280 Err(rusqlite::Error::QueryReturnedNoRows) => Ok(None),
281 Err(e) => Err(e),
282 }
283}
284
285#[derive(Clone, Debug)]
286pub struct TransactionStore {
287 path: PathBuf,
288 audit_key: Option<Arc<AuditKey>>,
291}
292
293#[derive(Clone, Debug, PartialEq)]
294pub struct RecordedPreviewedTransaction {
295 pub transaction: TransactionRecord,
296 pub preview: PreviewEnvelope,
297}
298
299struct InsertedTransaction {
300 record: TransactionRecord,
301 seq: u64,
302 chain_hash: String,
303}
304
305#[derive(Debug, thiserror::Error)]
306pub enum TransactionStoreError {
307 #[error("io error: {0}")]
308 Io(#[from] std::io::Error),
309
310 #[error("sqlite error: {0}")]
311 Sqlite(#[from] rusqlite::Error),
312
313 #[error("json error: {0}")]
314 Json(#[from] serde_json::Error),
315
316 #[error("database invariant violation: {0}")]
317 DatabaseInvariant(String),
318
319 #[error("transaction not found: {0}")]
320 NotFound(String),
321
322 #[error("invalid transition from {from:?} to {to:?}")]
323 InvalidTransition { from: JobState, to: JobState },
324
325 #[error("audit chain misconfiguration: {0}")]
326 AuditChainMissing(&'static str),
327
328 #[error("transaction {0} status changed concurrently; retry")]
336 ConcurrentStatusChange(String),
337}
338
339impl TransactionStore {
340 pub fn open_read_only(path: impl AsRef<Path>) -> Result<Self, TransactionStoreError> {
344 let path = path.as_ref().to_path_buf();
345 if let Some(parent) = path.parent() {
346 ensure_private_dir(parent)?;
347 }
348
349 let store = Self {
350 path,
351 audit_key: None,
352 };
353 store.initialize()?;
354 Ok(store)
355 }
356
357 pub fn open(path: impl AsRef<Path>) -> Result<Self, TransactionStoreError> {
364 let db_path = path.as_ref();
365 let key_path = crate::audit_chain::resolve_audit_key_path(db_path);
366 let key = AuditKey::load_or_generate(&key_path).map_err(|e| {
367 TransactionStoreError::Io(std::io::Error::new(
368 std::io::ErrorKind::InvalidData,
369 format!("audit key load failed: {e}"),
370 ))
371 })?;
372 Self::open_with_key(path, Arc::new(key))
373 }
374
375 pub fn open_with_key(
378 path: impl AsRef<Path>,
379 audit_key: Arc<AuditKey>,
380 ) -> Result<Self, TransactionStoreError> {
381 let path = path.as_ref().to_path_buf();
382 if let Some(parent) = path.parent() {
383 ensure_private_dir(parent)?;
384 }
385
386 let store = Self {
387 path,
388 audit_key: Some(audit_key),
389 };
390 store.initialize()?;
391 restrict_db_file_mode(&store.path);
392 Ok(store)
393 }
394
395 fn record_with(
404 &self,
405 transaction: NewTransaction,
406 in_tx: impl FnOnce(&Connection, &InsertedTransaction) -> Result<(), TransactionStoreError>,
407 ) -> Result<InsertedTransaction, TransactionStoreError> {
408 let key = self
409 .audit_key
410 .as_ref()
411 .ok_or(TransactionStoreError::AuditChainMissing(
412 "this TransactionStore was opened read-only; cannot record",
413 ))?;
414 let mut conn = self.connection()?;
415 let tx = conn.transaction_with_behavior(TransactionBehavior::Immediate)?;
420 let transaction_id = Uuid::new_v4().to_string();
421 let inserted = Self::insert_transaction(&tx, key, &transaction_id, transaction)?;
422 in_tx(&tx, &inserted)?;
423 tx.commit()?;
424 emit_chain_tip_watermark(inserted.seq, &inserted.chain_hash);
425 Ok(inserted)
426 }
427
428 pub fn record(
429 &self,
430 transaction: NewTransaction,
431 ) -> Result<TransactionRecord, TransactionStoreError> {
432 let inserted = self.record_with(transaction, |_, _| Ok(()))?;
433 Ok(inserted.record)
434 }
435
436 pub fn record_previewed(
437 &self,
438 transaction: NewTransaction,
439 preview: PreviewEnvelope,
440 ) -> Result<RecordedPreviewedTransaction, TransactionStoreError> {
441 let inserted = self.record_with(transaction, |conn, inserted| {
442 Self::insert_preview(conn, &inserted.record.transaction_id, &preview)
443 })?;
444 Ok(RecordedPreviewedTransaction {
445 transaction: inserted.record,
446 preview,
447 })
448 }
449
450 pub fn get(
451 &self,
452 transaction_id: &str,
453 ) -> Result<Option<TransactionRecord>, TransactionStoreError> {
454 let conn = self.connection()?;
455 let mut stmt = conn.prepare(
456 "SELECT
457 transaction_id,
458 request_id,
459 request_hash,
460 action_name,
461 risk_level,
462 status,
463 approval_id,
464 summary,
465 warnings_json
466 FROM transactions
467 WHERE transaction_id = ?1",
468 )?;
469 let mut rows = stmt.query(params![transaction_id])?;
470 if let Some(row) = rows.next()? {
471 Ok(Some(row_to_record(row)?))
472 } else {
473 Ok(None)
474 }
475 }
476
477 pub fn get_preview(
478 &self,
479 transaction_id: &str,
480 ) -> Result<Option<PreviewEnvelope>, TransactionStoreError> {
481 let conn = self.connection()?;
482 let mut stmt = conn.prepare(
483 "SELECT preview_json
484 FROM transaction_previews
485 WHERE transaction_id = ?1",
486 )?;
487 let mut rows = stmt.query(params![transaction_id])?;
488 if let Some(row) = rows.next()? {
489 let preview_json: String = row.get(0)?;
490 Ok(Some(serde_json::from_str(&preview_json)?))
491 } else {
492 Ok(None)
493 }
494 }
495
496 pub fn update_status(
497 &self,
498 transaction_id: &str,
499 new_status: JobState,
500 ) -> Result<(), TransactionStoreError> {
501 let conn = self.connection()?;
502
503 let current_status: String = conn
505 .query_row(
506 "SELECT status FROM transactions WHERE transaction_id = ?1",
507 params![transaction_id],
508 |row| row.get(0),
509 )
510 .map_err(|e| match e {
511 rusqlite::Error::QueryReturnedNoRows => {
512 TransactionStoreError::NotFound(transaction_id.to_string())
513 }
514 other => TransactionStoreError::Sqlite(other),
515 })?;
516
517 let current: JobState = deserialize_field(¤t_status)?;
518 if !crate::jobs::allowed_transition(¤t, &new_status) {
519 return Err(TransactionStoreError::InvalidTransition {
520 from: current,
521 to: new_status,
522 });
523 }
524
525 let mut conn = conn;
542 let tx = conn.transaction()?;
543 let rows_affected = tx.execute(
544 "UPDATE transactions SET status = ?1 WHERE transaction_id = ?2 AND status = ?3",
545 params![
546 serialize_field(&new_status)?,
547 transaction_id,
548 current_status
549 ],
550 )?;
551 if rows_affected == 0 {
552 return Err(TransactionStoreError::ConcurrentStatusChange(
553 transaction_id.to_string(),
554 ));
555 }
556 let Some(key) = self.audit_key.as_ref() else {
562 return Err(TransactionStoreError::AuditChainMissing(
563 "update_status has no signing key, so the outcome could not be chained",
564 ));
565 };
566 Self::append_event(
567 &tx,
568 key,
569 AuditEventKind::from(new_status),
570 transaction_id,
571 "",
572 EventIdentity::LegacyV1,
576 )?;
577 tx.commit()?;
578 Ok(())
579 }
580
581 pub fn status_matches_chain(
592 &self,
593 ) -> Result<audit_chain::StatusOutcome, TransactionStoreError> {
594 let conn = self.connection()?;
595 let mut stmt = conn.prepare(
596 "SELECT t.transaction_id, t.status, ( \
597 SELECT e.kind FROM audit_events e \
598 WHERE e.transaction_id = t.transaction_id AND e.kind LIKE 'status_%' \
599 ORDER BY e.seq DESC LIMIT 1 \
600 ) FROM transactions t",
601 )?;
602 let mut disagreements = Vec::new();
603 let mut checked: u64 = 0;
604 let rows = stmt.query_map([], |row| {
605 Ok((
606 row.get::<_, String>(0)?,
607 row.get::<_, String>(1)?,
608 row.get::<_, Option<String>>(2)?,
609 ))
610 })?;
611 for row in rows {
612 let (id, status_json, newest) = row?;
613 checked += 1;
614 let state: JobState = deserialize_field(&status_json)?;
615 let Some(newest) = newest else {
618 if state != JobState::Queued {
619 disagreements.push(format!(
620 "{id}: status is {state:?} and no status event was ever chained"
621 ));
622 }
623 continue;
624 };
625 let expected = AuditEventKind::from(state).as_str();
626 if newest != expected {
627 disagreements.push(format!(
628 "{id}: status column says {state:?} ({expected}) and the newest \
629 chained status event is {newest}"
630 ));
631 }
632 }
633 Ok(if disagreements.is_empty() {
634 audit_chain::StatusOutcome::Agrees {
635 rows_checked: checked,
636 }
637 } else {
638 audit_chain::StatusOutcome::Disagrees {
639 detail: disagreements.join("; "),
640 }
641 })
642 }
643
644 pub fn approve_transaction(
649 &self,
650 transaction_id: &str,
651 approver: CallerPrincipal,
652 ) -> Result<Option<String>, TransactionStoreError> {
653 let key = self
654 .audit_key
655 .as_ref()
656 .ok_or(TransactionStoreError::AuditChainMissing(
657 "this TransactionStore was opened read-only; cannot approve",
658 ))?;
659 let approver_signed = approver.as_signed_str();
662 let Some(record) = self.get(transaction_id)? else {
663 return Ok(None);
664 };
665 let receipt = key.approval_receipt(transaction_id, &record.request_hash);
666 let receipt_digest = audit_chain::approval_receipt_digest(&receipt);
667 let Some(committed_digest) = record.approval_id.as_deref() else {
668 return Err(TransactionStoreError::DatabaseInvariant(format!(
669 "transaction {transaction_id} has no signed approval commitment"
670 )));
671 };
672 if !bool::from(receipt_digest.as_bytes().ct_eq(committed_digest.as_bytes())) {
673 return Err(TransactionStoreError::DatabaseInvariant(format!(
674 "transaction {transaction_id} approval commitment does not match its signed preview"
675 )));
676 }
677
678 let mut conn = self.connection()?;
681 let tx = conn.transaction_with_behavior(TransactionBehavior::Immediate)?;
682 let queued_json = serialize_field(&JobState::Queued)?;
683 let rows_affected = tx.execute(
684 &format!(
685 "INSERT INTO transaction_approvals (transaction_id, receipt_digest) \
686 SELECT transaction_id, ?1 FROM transactions \
687 WHERE transaction_id = ?2 \
688 AND status = ?3 \
689 AND julianday(created_at) > julianday('now', '-{APPROVAL_RECEIPT_TTL_MINUTES} minutes') \
690 AND NOT EXISTS ( \
691 SELECT 1 FROM transaction_approvals WHERE transaction_id = ?2 \
692 )"
693 ),
694 params![receipt_digest, transaction_id, queued_json],
695 )?;
696 if rows_affected > 0 {
697 Self::append_event(
698 &tx,
699 key,
700 AuditEventKind::ApprovalGranted,
701 transaction_id,
702 &receipt_digest,
703 EventIdentity::V2 {
704 caller_principal: &approver_signed,
705 },
706 )?;
707 }
708 tx.commit()?;
709 Ok((rows_affected > 0).then_some(receipt))
710 }
711
712 pub fn revoke_unconsumed_approval(
718 &self,
719 transaction_id: &str,
720 revoker: CallerPrincipal,
721 ) -> Result<bool, TransactionStoreError> {
722 let key = self
723 .audit_key
724 .as_ref()
725 .ok_or(TransactionStoreError::AuditChainMissing(
726 "this TransactionStore was opened read-only; cannot revoke",
727 ))?;
728 let mut conn = self.connection()?;
729 let tx = conn.transaction_with_behavior(TransactionBehavior::Immediate)?;
730 let revoked = Self::revoke_unconsumed_approval_in_tx(
731 &tx,
732 key,
733 transaction_id,
734 &revoker.as_signed_str(),
735 )?;
736 tx.commit()?;
737 Ok(revoked)
738 }
739
740 fn revoke_unconsumed_approval_in_tx(
741 conn: &Connection,
742 key: &AuditKey,
743 transaction_id: &str,
744 revoker_signed: &str,
745 ) -> Result<bool, TransactionStoreError> {
746 let digest: Option<String> = query_optional(
749 conn,
750 "SELECT receipt_digest FROM transaction_approvals \
751 WHERE transaction_id = ?1 AND consumed_at IS NULL",
752 params![transaction_id],
753 |row| row.get(0),
754 )?;
755 let rows_affected = conn.execute(
756 "DELETE FROM transaction_approvals \
757 WHERE transaction_id = ?1 AND consumed_at IS NULL",
758 params![transaction_id],
759 )?;
760 if rows_affected > 0 {
761 let digest = digest.ok_or_else(|| {
762 TransactionStoreError::DatabaseInvariant(format!(
763 "revoked an approval for {transaction_id} that had no receipt digest"
764 ))
765 })?;
766 Self::append_event(
767 conn,
768 key,
769 AuditEventKind::ApprovalRevoked,
770 transaction_id,
771 &digest,
772 EventIdentity::V2 {
773 caller_principal: revoker_signed,
774 },
775 )?;
776 }
777 Ok(rows_affected > 0)
778 }
779
780 pub fn claim_approved_for_execution(
786 &self,
787 transaction_id: &str,
788 receipt_digest: &str,
789 executor: CallerPrincipal,
790 ) -> Result<bool, TransactionStoreError> {
791 let key = self
792 .audit_key
793 .as_ref()
794 .ok_or(TransactionStoreError::AuditChainMissing(
795 "this TransactionStore was opened read-only; cannot claim",
796 ))?;
797 let mut conn = self.connection()?;
798 let tx = conn.transaction_with_behavior(TransactionBehavior::Immediate)?;
799 let queued_json = serialize_field(&JobState::Queued)?;
800 let running_json = serialize_field(&JobState::Running)?;
801 let rows_affected = tx.execute(
802 &format!(
803 "UPDATE transactions SET status = ?1 \
804 WHERE transaction_id = ?2 \
805 AND status = ?3 \
806 AND julianday(created_at) > julianday('now', '-{APPROVAL_RECEIPT_TTL_MINUTES} minutes') \
807 AND EXISTS ( \
808 SELECT 1 FROM transaction_approvals \
809 WHERE transaction_id = ?2 \
810 AND receipt_digest = ?4 \
811 AND consumed_at IS NULL \
812 )"
813 ),
814 params![running_json, transaction_id, queued_json, receipt_digest],
815 )?;
816 if rows_affected > 0 {
817 tx.execute(
818 "UPDATE transaction_approvals \
819 SET consumed_at = datetime('now') \
820 WHERE transaction_id = ?1 AND consumed_at IS NULL",
821 params![transaction_id],
822 )?;
823 Self::append_event(
824 &tx,
825 key,
826 AuditEventKind::ApprovalConsumed,
827 transaction_id,
828 receipt_digest,
829 EventIdentity::V2 {
830 caller_principal: &executor.as_signed_str(),
831 },
832 )?;
833 }
834 tx.commit()?;
835 Ok(rows_affected > 0)
836 }
837
838 pub fn cleanup_stale_queued(&self) -> Result<u64, TransactionStoreError> {
853 let key = self
854 .audit_key
855 .as_ref()
856 .ok_or(TransactionStoreError::AuditChainMissing(
857 "this TransactionStore was opened read-only; cannot clean up",
858 ))?;
859 let mut conn = self.connection()?;
860 let tx = conn.transaction_with_behavior(TransactionBehavior::Immediate)?;
861 let canceled_json = serialize_field(&JobState::Canceled)?;
862 let queued_json = serialize_field(&JobState::Queued)?;
863 let stale_ids: Vec<String> = {
864 let mut statement = tx.prepare(&format!(
865 "SELECT transaction_id FROM transactions \
866 WHERE status = ?1 \
867 AND julianday(created_at) <= julianday('now', '-{APPROVAL_RECEIPT_TTL_MINUTES} minutes')"
868 ))?;
869 let stale_ids = statement
870 .query_map(params![queued_json], |row| row.get(0))?
871 .collect::<Result<_, _>>()?;
872 stale_ids
873 };
874 let mut canceled = 0;
875 for transaction_id in stale_ids {
876 let rows_affected = tx.execute(
877 "UPDATE transactions SET status = ?1 \
878 WHERE transaction_id = ?2 AND status = ?3",
879 params![canceled_json, transaction_id, queued_json],
880 )?;
881 if rows_affected > 0 {
882 Self::revoke_unconsumed_approval_in_tx(
886 &tx,
887 key,
888 &transaction_id,
889 &CallerPrincipal::Unattributed.as_signed_str(),
890 )?;
891 canceled += rows_affected;
892 }
893 }
894 tx.commit()?;
895 Ok(canceled as u64)
896 }
897
898 pub fn cancel_queued(
907 &self,
908 transaction_id: &str,
909 canceller: CallerPrincipal,
910 ) -> Result<bool, TransactionStoreError> {
911 let key = self
912 .audit_key
913 .as_ref()
914 .ok_or(TransactionStoreError::AuditChainMissing(
915 "this TransactionStore was opened read-only; cannot cancel",
916 ))?;
917 let mut conn = self.connection()?;
918 let tx = conn.transaction_with_behavior(TransactionBehavior::Immediate)?;
919 let canceled_json = serialize_field(&JobState::Canceled)?;
920 let queued_json = serialize_field(&JobState::Queued)?;
921 let rows_affected = tx.execute(
922 "UPDATE transactions SET status = ?1 \
923 WHERE transaction_id = ?2 AND status = ?3",
924 params![canceled_json, transaction_id, queued_json],
925 )?;
926 if rows_affected > 0 {
927 Self::revoke_unconsumed_approval_in_tx(
932 &tx,
933 key,
934 transaction_id,
935 &canceller.as_signed_str(),
936 )?;
937 }
938 tx.commit()?;
939 Ok(rows_affected > 0)
940 }
941
942 pub fn list_transactions(
951 &self,
952 limit: u32,
953 status_filter: Option<&str>,
954 action_filter: Option<&str>,
955 since_hours: Option<u32>,
956 ) -> Result<Vec<TransactionRecord>, TransactionStoreError> {
957 let conn = self.connection()?;
958 let (filter_sql, param_values) =
959 Self::build_history_filter(limit, status_filter, action_filter, since_hours)?;
960 let sql = format!(
961 "SELECT transaction_id, request_id, request_hash, action_name, \
962 risk_level, status, approval_id, summary, warnings_json \
963 FROM transactions WHERE 1=1{filter_sql}"
964 );
965 let params_ref: Vec<&dyn rusqlite::types::ToSql> =
966 param_values.iter().map(|b| b.as_ref()).collect();
967
968 let mut stmt = conn.prepare(&sql)?;
969 let rows = stmt.query_map(params_ref.as_slice(), |row| Ok(row_to_record(row)))?;
970
971 let mut results = Vec::new();
972 for row in rows {
973 results.push(row??);
974 }
975 Ok(results)
976 }
977
978 pub fn list_history(
983 &self,
984 limit: u32,
985 status_filter: Option<&str>,
986 action_filter: Option<&str>,
987 since_hours: Option<u32>,
988 ) -> Result<Vec<JobHistoryEntry>, TransactionStoreError> {
989 let conn = self.connection()?;
990 let (filter_sql, param_values) =
991 Self::build_history_filter(limit, status_filter, action_filter, since_hours)?;
992 let sql = format!(
993 "SELECT transaction_id, action_name, risk_level, status, summary, created_at \
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| {
1001 Ok((|| {
1002 Ok::<JobHistoryEntry, TransactionStoreError>(JobHistoryEntry {
1003 transaction_id: row.get(0)?,
1004 action_name: row.get(1)?,
1005 risk_level: deserialize_field(&row.get::<_, String>(2)?)?,
1006 status: deserialize_field(&row.get::<_, String>(3)?)?,
1007 summary: row.get(4)?,
1008 created_at: row.get(5)?,
1009 })
1010 })())
1011 })?;
1012
1013 let mut results = Vec::new();
1014 for row in rows {
1015 results.push(row??);
1016 }
1017 Ok(results)
1018 }
1019
1020 fn build_history_filter(
1025 limit: u32,
1026 status_filter: Option<&str>,
1027 action_filter: Option<&str>,
1028 since_hours: Option<u32>,
1029 ) -> Result<(String, Vec<Box<dyn rusqlite::types::ToSql>>), TransactionStoreError> {
1030 let mut sql = String::new();
1031 let mut param_values: Vec<Box<dyn rusqlite::types::ToSql>> = Vec::new();
1032
1033 if let Some(status) = status_filter {
1034 let validated: JobState = deserialize_field(&format!("\"{status}\""))?;
1038 let status_json = serialize_field(&validated)?;
1039 sql.push_str(" AND status = ?");
1040 param_values.push(Box::new(status_json));
1041 }
1042
1043 if let Some(action) = action_filter {
1044 sql.push_str(" AND action_name = ?");
1045 param_values.push(Box::new(action.to_string()));
1046 }
1047
1048 if let Some(hours) = since_hours {
1049 sql.push_str(" AND julianday(created_at) > julianday('now', '-' || ? || ' hours')");
1050 param_values.push(Box::new(hours));
1051 }
1052
1053 sql.push_str(" ORDER BY seq DESC LIMIT ?");
1054 param_values.push(Box::new(limit.min(100)));
1055
1056 Ok((sql, param_values))
1057 }
1058
1059 fn connection(&self) -> Result<Connection, TransactionStoreError> {
1060 let conn = Connection::open(&self.path)?;
1061 conn.pragma_update(None, "journal_mode", "WAL")?;
1072 conn.pragma_update(None, "synchronous", "NORMAL")?;
1073 conn.pragma_update(None, "foreign_keys", "ON")?;
1074 conn.busy_timeout(std::time::Duration::from_secs(5))?;
1075 Ok(conn)
1076 }
1077
1078 fn initialize(&self) -> Result<(), TransactionStoreError> {
1079 let mut conn = self.connection()?;
1080 let tx = conn.transaction_with_behavior(TransactionBehavior::Immediate)?;
1081 tx.execute_batch(
1082 "CREATE TABLE IF NOT EXISTS schema_migrations (\
1083 version INTEGER PRIMARY KEY,\
1084 name TEXT NOT NULL,\
1085 applied_at TEXT NOT NULL DEFAULT (datetime('now'))\
1086 );",
1087 )?;
1088 let current: i64 = tx.query_row(
1089 "SELECT COALESCE(MAX(version), 0) FROM schema_migrations",
1090 [],
1091 |row| row.get(0),
1092 )?;
1093 let latest = SQLITE_MIGRATIONS.last().map_or(0, |m| m.version);
1094 if current > latest {
1095 return Err(TransactionStoreError::DatabaseInvariant(format!(
1096 "sqlite schema version {current} is newer than this binary supports ({latest})"
1097 )));
1098 }
1099 for migration in SQLITE_MIGRATIONS.iter().filter(|m| m.version > current) {
1100 tx.execute_batch(migration.sql)?;
1101 tx.execute(
1102 "INSERT INTO schema_migrations (version, name) VALUES (?1, ?2)",
1103 params![migration.version, migration.name],
1104 )?;
1105 }
1106 tx.commit()?;
1107 Ok(())
1108 }
1109
1110 pub fn fetch_chain_rows(&self) -> Result<Vec<ChainRow>, TransactionStoreError> {
1112 let conn = self.connection()?;
1113 let mut stmt = conn.prepare(&format!(
1114 "SELECT {CHAIN_ROW_COLUMNS} FROM transactions ORDER BY seq ASC"
1115 ))?;
1116 let rows = stmt.query_map([], chain_row_from_sqlite)?;
1117 let mut out = Vec::new();
1118 for row in rows {
1119 out.push(row?);
1120 }
1121 Ok(out)
1122 }
1123
1124 pub fn fetch_event_rows(&self) -> Result<Vec<EventRow>, TransactionStoreError> {
1126 let conn = self.connection()?;
1127 let mut stmt = conn.prepare(
1128 "SELECT seq, key_id, kind, transaction_id, receipt_digest, \
1129 created_at, prev_chain_hash, chain_hash, chain_version, caller_principal \
1130 FROM audit_events ORDER BY seq ASC",
1131 )?;
1132 let rows = stmt.query_map([], event_row_from_sqlite)?;
1133 let mut out = Vec::new();
1134 for row in rows {
1135 out.push(row?);
1136 }
1137 Ok(out)
1138 }
1139
1140 pub fn verify_audit_chain(
1142 &self,
1143 key: &AuditKey,
1144 ) -> Result<VerifyOutcome, TransactionStoreError> {
1145 let rows = self.fetch_chain_rows()?;
1146 Ok(audit_chain::verify_chain(key, &rows))
1147 }
1148
1149 pub fn verify_event_chain(
1151 &self,
1152 key: &AuditKey,
1153 ) -> Result<VerifyOutcome, TransactionStoreError> {
1154 let rows = self.fetch_event_rows()?;
1155 Ok(audit_chain::verify_event_chain(key, &rows))
1156 }
1157
1158 pub fn verify_event_binding(
1161 &self,
1162 ) -> Result<audit_chain::BindingOutcome, TransactionStoreError> {
1163 let tx_rows = self.fetch_chain_rows()?;
1164 let event_rows = self.fetch_event_rows()?;
1165 Ok(audit_chain::verify_event_binding(&tx_rows, &event_rows))
1166 }
1167
1168 pub fn verify_audit_chain_with_pubkey(
1171 &self,
1172 verifying_key_hex: &str,
1173 ) -> Result<VerifyOutcome, TransactionStoreError> {
1174 let rows = self.fetch_chain_rows()?;
1175 Ok(audit_chain::verify_chain_with_pubkey(
1176 verifying_key_hex,
1177 &rows,
1178 ))
1179 }
1180
1181 pub fn fetch_chain_row(
1184 &self,
1185 transaction_id: &str,
1186 ) -> Result<Option<ChainRow>, TransactionStoreError> {
1187 let conn = self.connection()?;
1188 let mut stmt = conn.prepare(&format!(
1189 "SELECT {CHAIN_ROW_COLUMNS} FROM transactions WHERE transaction_id = ?1"
1190 ))?;
1191 let mut rows = stmt.query(params![transaction_id])?;
1192 if let Some(row) = rows.next()? {
1193 Ok(Some(chain_row_from_sqlite(row)?))
1194 } else {
1195 Ok(None)
1196 }
1197 }
1198
1199 fn next_seq_and_prev_hash(conn: &Connection) -> Result<(u64, String), TransactionStoreError> {
1203 let prev: Option<(i64, String)> = query_optional(
1204 conn,
1205 "SELECT seq, chain_hash FROM transactions ORDER BY seq DESC LIMIT 1",
1206 [],
1207 |row| Ok((row.get(0)?, row.get(1)?)),
1208 )?;
1209 Ok(match prev {
1210 Some((seq, hash)) => ((seq as u64) + 1, hash),
1211 None => (1, String::new()),
1212 })
1213 }
1214
1215 fn event_chain_tip(conn: &Connection) -> Result<Option<String>, TransactionStoreError> {
1218 Ok(query_optional(
1219 conn,
1220 "SELECT chain_hash FROM audit_events ORDER BY seq DESC LIMIT 1",
1221 [],
1222 |row| row.get(0),
1223 )?)
1224 }
1225
1226 fn append_event(
1231 conn: &Connection,
1232 key: &AuditKey,
1233 kind: AuditEventKind,
1234 transaction_id: &str,
1235 receipt_digest: &str,
1236 identity: EventIdentity<'_>,
1237 ) -> Result<(), TransactionStoreError> {
1238 let prev_chain_hash = Self::event_chain_tip(conn)?.unwrap_or_default();
1239 let seq: i64 = conn.query_row(
1240 "SELECT COALESCE(MAX(seq), 0) + 1 FROM audit_events",
1241 [],
1242 |row| row.get(0),
1243 )?;
1244 let created_at: String =
1245 conn.query_row("SELECT strftime('%Y-%m-%dT%H:%M:%fZ', 'now')", [], |row| {
1246 row.get(0)
1247 })?;
1248 let key_id = CURRENT_KEY_ID.to_string();
1249 let content = EventContent {
1254 seq: seq as u64,
1255 key_id: &key_id,
1256 kind,
1257 transaction_id,
1258 receipt_digest,
1259 created_at: &created_at,
1260 identity,
1261 };
1262 let chain_hash = key.event_hash(&content, &prev_chain_hash);
1263 let stored_principal = match identity {
1264 EventIdentity::V2 { caller_principal } => Some(caller_principal),
1265 EventIdentity::LegacyV1 => None,
1266 };
1267 conn.execute(
1268 "INSERT INTO audit_events (\
1269 seq, key_id, kind, transaction_id, receipt_digest, \
1270 created_at, chain_hash, prev_chain_hash, chain_version, caller_principal \
1271 ) VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9, ?10)",
1272 params![
1273 seq,
1274 key_id,
1275 kind.as_str(),
1276 transaction_id,
1277 receipt_digest,
1278 created_at,
1279 chain_hash,
1280 prev_chain_hash,
1281 identity.version() as i64,
1282 stored_principal,
1283 ],
1284 )?;
1285 Ok(())
1286 }
1287
1288 fn insert_transaction(
1289 conn: &Connection,
1290 key: &AuditKey,
1291 transaction_id: &str,
1292 transaction: NewTransaction,
1293 ) -> Result<InsertedTransaction, TransactionStoreError> {
1294 let request_id = transaction.request_id;
1295 let request_hash = transaction.request_hash;
1296 let action_name = transaction.action_name;
1297 let risk_level = transaction.risk_level;
1298 let status = JobState::Queued;
1300 let approval_id = Some(key.approval_commitment(transaction_id, &request_hash));
1307 let summary = transaction.summary;
1308 let warnings = transaction.warnings;
1309 let warnings_json = serde_json::to_string(&warnings)?;
1310
1311 let (seq, prev_chain_hash) = Self::next_seq_and_prev_hash(conn)?;
1314
1315 let created_at: String =
1319 conn.query_row("SELECT strftime('%Y-%m-%dT%H:%M:%fZ', 'now')", [], |row| {
1320 row.get(0)
1321 })?;
1322
1323 let key_id = CURRENT_KEY_ID.to_string();
1324 let event_tip = Self::event_chain_tip(conn)?.unwrap_or_default();
1328 let caller_role = transaction.caller_role.as_str();
1329 let caller_principal = transaction.caller_principal.as_signed_str();
1330 let identity = ChainIdentity::V3 {
1334 caller_role,
1335 event_tip: &event_tip,
1336 caller_principal: &caller_principal,
1337 };
1338 let chain_hash = key.chain_hash(
1339 &ChainContent {
1340 seq,
1341 key_id: &key_id,
1342 transaction_id,
1343 request_id: &request_id,
1344 request_hash: &request_hash,
1345 action_name: &action_name,
1346 risk_level,
1347 summary: &summary,
1348 approval_id: approval_id.as_deref(),
1349 warnings_json: &warnings_json,
1350 created_at: &created_at,
1351 identity,
1352 },
1353 &prev_chain_hash,
1354 );
1355
1356 let record = TransactionRecord {
1357 transaction_id: transaction_id.to_string(),
1358 request_id: request_id.clone(),
1359 request_hash: request_hash.clone(),
1360 action_name: action_name.clone(),
1361 risk_level,
1362 status,
1363 approval_id: approval_id.clone(),
1364 summary: summary.clone(),
1365 warnings: warnings.clone(),
1366 };
1367
1368 conn.execute(
1369 "INSERT INTO transactions (
1370 transaction_id,
1371 request_id,
1372 request_hash,
1373 action_name,
1374 risk_level,
1375 status,
1376 approval_id,
1377 summary,
1378 warnings_json,
1379 created_at,
1380 seq,
1381 key_id,
1382 chain_hash,
1383 prev_chain_hash,
1384 chain_version,
1385 caller_role,
1386 event_tip,
1387 caller_principal
1388 ) VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9, ?10, ?11, ?12, ?13, ?14, ?15, ?16, ?17, ?18)",
1389 params![
1390 transaction_id,
1391 request_id,
1392 request_hash,
1393 action_name,
1394 serialize_field(&risk_level)?,
1395 serialize_field(&status)?,
1396 approval_id,
1397 summary,
1398 warnings_json,
1399 created_at,
1400 seq as i64,
1401 key_id,
1402 chain_hash,
1403 prev_chain_hash,
1404 identity.version() as i64,
1405 caller_role,
1406 event_tip,
1407 caller_principal,
1408 ],
1409 )?;
1410
1411 Ok(InsertedTransaction {
1412 record,
1413 seq,
1414 chain_hash,
1415 })
1416 }
1417
1418 fn insert_preview(
1419 conn: &Connection,
1420 transaction_id: &str,
1421 preview: &PreviewEnvelope,
1422 ) -> Result<(), TransactionStoreError> {
1423 conn.execute(
1424 "INSERT INTO transaction_previews (transaction_id, preview_json)
1425 VALUES (?1, ?2)",
1426 params![transaction_id, serde_json::to_string(preview)?],
1427 )?;
1428 Ok(())
1429 }
1430}
1431
1432fn ensure_private_dir(dir: &Path) -> std::io::Result<()> {
1439 use std::os::unix::fs::DirBuilderExt;
1440 if dir.exists() {
1441 return Ok(());
1442 }
1443 std::fs::DirBuilder::new()
1444 .recursive(true)
1445 .mode(0o700)
1446 .create(dir)
1447}
1448
1449fn restrict_db_file_mode(db_path: &Path) {
1463 use std::os::unix::fs::PermissionsExt;
1464
1465 let sidecars = [
1469 db_path.to_path_buf(),
1470 with_suffix(db_path, "-wal"),
1471 with_suffix(db_path, "-shm"),
1472 ];
1473 for path in sidecars {
1474 let Ok(metadata) = std::fs::metadata(&path) else {
1475 continue; };
1477 let mode = metadata.permissions().mode() & 0o777;
1478 if mode & 0o077 == 0 {
1479 continue; }
1481 if let Err(err) = std::fs::set_permissions(&path, std::fs::Permissions::from_mode(0o600)) {
1482 eprintln!(
1483 "[sysknife-daemon] WARNING: {} is mode {mode:o} and could not be \
1484 narrowed to 0600 ({err}); it is readable by anyone who can reach \
1485 it. The parent directory should be 0700 — check it.",
1486 path.display()
1487 );
1488 }
1489 }
1490}
1491
1492fn with_suffix(path: &Path, suffix: &str) -> PathBuf {
1495 let mut name = path.as_os_str().to_os_string();
1496 name.push(suffix);
1497 PathBuf::from(name)
1498}
1499
1500fn row_to_record(row: &rusqlite::Row) -> Result<TransactionRecord, TransactionStoreError> {
1501 Ok(TransactionRecord {
1502 transaction_id: row.get(0)?,
1503 request_id: row.get(1)?,
1504 request_hash: row.get(2)?,
1505 action_name: row.get(3)?,
1506 risk_level: deserialize_field(&row.get::<_, String>(4)?)?,
1507 status: deserialize_field(&row.get::<_, String>(5)?)?,
1508 approval_id: row.get(6)?,
1509 summary: row.get(7)?,
1510 warnings: serde_json::from_str(&row.get::<_, String>(8)?)?,
1511 })
1512}
1513
1514fn serialize_field<T: Serialize>(value: &T) -> Result<String, serde_json::Error> {
1515 serde_json::to_string(value)
1516}
1517
1518fn deserialize_field<T: DeserializeOwned>(value: &str) -> Result<T, serde_json::Error> {
1519 serde_json::from_str(value)
1520}
1521
1522#[cfg(test)]
1523mod tests {
1524 use super::*;
1525 use crate::audit_chain::CHAIN_VERSION_CURRENT;
1526 use crate::audit_chain::CHAIN_VERSION_LEGACY;
1527 use crate::audit_chain::EVENT_VERSION_CURRENT;
1528 use crate::audit_chain::EVENT_VERSION_LEGACY;
1529 use std::os::unix::fs::PermissionsExt;
1530 use tempfile::tempdir;
1531
1532 fn test_store(path: impl AsRef<Path>) -> TransactionStore {
1536 let key = Arc::new(AuditKey::from_bytes(vec![0x42; 32]));
1537 TransactionStore::open_with_key(path, key).unwrap()
1538 }
1539
1540 fn legacy_v1_database(path: &Path, key: &AuditKey) -> String {
1547 let conn = Connection::open(path).unwrap();
1548 conn.execute_batch(
1549 "CREATE TABLE schema_migrations (
1550 version INTEGER PRIMARY KEY,
1551 name TEXT NOT NULL,
1552 applied_at TEXT NOT NULL DEFAULT (datetime('now'))
1553 );",
1554 )
1555 .unwrap();
1556 conn.execute_batch(SQLITE_MIGRATIONS[0].sql).unwrap();
1557 conn.execute(
1558 "INSERT INTO schema_migrations (version, name) VALUES (1, ?1)",
1559 params![SQLITE_MIGRATIONS[0].name],
1560 )
1561 .unwrap();
1562
1563 let transaction_id = "tx-legacy";
1564 let created_at = "2026-04-24T12:00:00.000Z";
1565 let chain_hash = key.chain_hash(
1566 &ChainContent {
1567 seq: 1,
1568 key_id: CURRENT_KEY_ID,
1569 transaction_id,
1570 request_id: "req-legacy",
1571 request_hash: "hash-legacy",
1572 action_name: "UpdateSystem",
1573 risk_level: RiskLevel::High,
1574 summary: "Upgrade the system",
1575 approval_id: None,
1576 warnings_json: "[]",
1577 created_at,
1578 identity: ChainIdentity::LegacyV1,
1579 },
1580 "",
1581 );
1582 conn.execute(
1583 "INSERT INTO transactions (
1584 transaction_id, request_id, request_hash, action_name, risk_level,
1585 status, approval_id, summary, warnings_json, created_at,
1586 seq, key_id, chain_hash, prev_chain_hash
1587 ) VALUES (?1, ?2, ?3, ?4, ?5, ?6, NULL, ?7, ?8, ?9, 1, ?10, ?11, '')",
1588 params![
1589 transaction_id,
1590 "req-legacy",
1591 "hash-legacy",
1592 "UpdateSystem",
1593 serialize_field(&RiskLevel::High).unwrap(),
1594 serialize_field(&JobState::Queued).unwrap(),
1595 "Upgrade the system",
1596 "[]",
1597 created_at,
1598 CURRENT_KEY_ID,
1599 chain_hash,
1600 ],
1601 )
1602 .unwrap();
1603 transaction_id.to_string()
1604 }
1605
1606 fn v2_database(path: &Path, key: &AuditKey) -> String {
1613 let conn = Connection::open(path).unwrap();
1614 conn.execute_batch(
1615 "CREATE TABLE schema_migrations (
1616 version INTEGER PRIMARY KEY,
1617 name TEXT NOT NULL,
1618 applied_at TEXT NOT NULL DEFAULT (datetime('now'))
1619 );",
1620 )
1621 .unwrap();
1622 for migration in &SQLITE_MIGRATIONS[0..=1] {
1623 conn.execute_batch(migration.sql).unwrap();
1624 conn.execute(
1625 "INSERT INTO schema_migrations (version, name) VALUES (?1, ?2)",
1626 params![migration.version, migration.name],
1627 )
1628 .unwrap();
1629 }
1630
1631 let transaction_id = "tx-v2";
1632 let created_at = "2026-07-01T12:00:00.000Z";
1633 let chain_hash = key.chain_hash(
1634 &ChainContent {
1635 seq: 1,
1636 key_id: CURRENT_KEY_ID,
1637 transaction_id,
1638 request_id: "req-v2",
1639 request_hash: "hash-v2",
1640 action_name: "UpdateSystem",
1641 risk_level: RiskLevel::High,
1642 summary: "Upgrade the system",
1643 approval_id: None,
1644 warnings_json: "[]",
1645 created_at,
1646 identity: ChainIdentity::V2 {
1647 caller_role: "admin",
1648 event_tip: "",
1649 },
1650 },
1651 "",
1652 );
1653 conn.execute(
1654 "INSERT INTO transactions (
1655 transaction_id, request_id, request_hash, action_name, risk_level,
1656 status, approval_id, summary, warnings_json, created_at,
1657 seq, key_id, chain_hash, prev_chain_hash,
1658 chain_version, caller_role, event_tip
1659 ) VALUES (?1, ?2, ?3, ?4, ?5, ?6, NULL, ?7, ?8, ?9, 1, ?10, ?11, '', 2, 'admin', '')",
1660 params![
1661 transaction_id,
1662 "req-v2",
1663 "hash-v2",
1664 "UpdateSystem",
1665 serialize_field(&RiskLevel::High).unwrap(),
1666 serialize_field(&JobState::Queued).unwrap(),
1667 "Upgrade the system",
1668 "[]",
1669 created_at,
1670 CURRENT_KEY_ID,
1671 chain_hash,
1672 ],
1673 )
1674 .unwrap();
1675 transaction_id.to_string()
1676 }
1677
1678 #[test]
1686 fn a_v2_database_keeps_verifying_and_accepts_v3_rows() {
1687 let dir = tempdir().unwrap();
1688 let db_path = dir.path().join("tx.db");
1689 let key = AuditKey::from_bytes(vec![0x42; 32]);
1690 let old_transaction = v2_database(&db_path, &key);
1691
1692 let store = test_store(&db_path);
1693
1694 let version: i64 = store
1695 .connection()
1696 .unwrap()
1697 .query_row("SELECT MAX(version) FROM schema_migrations", [], |row| {
1698 row.get(0)
1699 })
1700 .unwrap();
1701 assert_eq!(
1702 version, 4,
1703 "opening a v2 database must apply every migration up to 4"
1704 );
1705
1706 let rows = store.fetch_chain_rows().unwrap();
1707 assert_eq!(rows.len(), 1);
1708 assert_eq!(rows[0].transaction_id, old_transaction);
1709 assert_eq!(rows[0].chain_version, audit_chain::CHAIN_VERSION_V2);
1710 assert_eq!(
1711 rows[0].caller_principal, None,
1712 "migration 3 must not invent a principal for a row signed without one"
1713 );
1714 assert!(matches!(
1715 store.verify_audit_chain(&key).unwrap(),
1716 audit_chain::VerifyOutcome::Intact { rows_checked: 1 }
1717 ));
1718
1719 let mut fresh = queued_transaction();
1721 fresh.caller_principal = CallerPrincipal::Uid(4242);
1722 store.record(fresh).unwrap();
1723
1724 let rows = store.fetch_chain_rows().unwrap();
1725 assert_eq!(rows.len(), 2);
1726 assert_eq!(rows[1].chain_version, audit_chain::CHAIN_VERSION_CURRENT);
1727 assert_eq!(rows[1].caller_principal.as_deref(), Some("uid:4242"));
1728 assert_eq!(
1729 rows[1].prev_chain_hash, rows[0].chain_hash,
1730 "the v3 row must chain onto the v2 row, not start a new chain"
1731 );
1732 assert!(matches!(
1733 store.verify_audit_chain(&key).unwrap(),
1734 audit_chain::VerifyOutcome::Intact { rows_checked: 2 }
1735 ));
1736
1737 let census = audit_chain::AttributionCensus::of(&rows);
1745 assert_eq!(census.rows(), 2);
1746 assert_eq!(census.named(), 1, "only the v3 row names uid:4242");
1747 assert_eq!(
1748 census.not_recorded(),
1749 1,
1750 "the v2 row's encoding never signed a principal"
1751 );
1752 assert_eq!(census.attribution_failed(), 0);
1753 assert_eq!(census.unattested(), 0);
1754 assert_eq!(census.unnamed(), 1);
1755 }
1756
1757 #[test]
1758 fn a_database_written_before_the_migration_still_verifies_after_upgrading() {
1759 let dir = tempdir().unwrap();
1764 let db_path = dir.path().join("tx.db");
1765 let key = AuditKey::from_bytes(vec![0x42; 32]);
1766 legacy_v1_database(&db_path, &key);
1767
1768 let store = test_store(&db_path);
1769 assert_eq!(
1770 store.verify_audit_chain(&key).unwrap(),
1771 VerifyOutcome::Intact { rows_checked: 1 }
1772 );
1773 let rows = store.fetch_chain_rows().unwrap();
1774 assert_eq!(rows[0].chain_version, CHAIN_VERSION_LEGACY);
1775 assert_eq!(rows[0].caller_role, None);
1776 }
1777
1778 #[test]
1779 fn rows_appended_after_the_migration_chain_onto_legacy_rows() {
1780 let dir = tempdir().unwrap();
1784 let db_path = dir.path().join("tx.db");
1785 let key = AuditKey::from_bytes(vec![0x42; 32]);
1786 legacy_v1_database(&db_path, &key);
1787
1788 let store = test_store(&db_path);
1789 store.record(queued_transaction()).unwrap();
1790
1791 let rows = store.fetch_chain_rows().unwrap();
1792 assert_eq!(rows.len(), 2);
1793 assert_eq!(rows[1].prev_chain_hash, rows[0].chain_hash);
1794 assert_eq!(rows[1].chain_version, CHAIN_VERSION_CURRENT);
1795 assert_eq!(
1796 store.verify_audit_chain(&key).unwrap(),
1797 VerifyOutcome::Intact { rows_checked: 2 }
1798 );
1799 }
1800
1801 #[test]
1802 fn a_schema_newer_than_this_binary_is_refused() {
1803 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 let conn = Connection::open(&db_path).unwrap();
1808 conn.execute(
1809 "INSERT INTO schema_migrations (version, name) VALUES (99, 'from the future')",
1810 [],
1811 )
1812 .unwrap();
1813 drop(conn);
1814
1815 let err = TransactionStore::open_with_key(&db_path, Arc::new(key)).unwrap_err();
1816 assert!(
1817 matches!(err, TransactionStoreError::DatabaseInvariant(ref m) if m.contains("99")),
1818 "expected a refusal naming the unsupported version, got {err:?}"
1819 );
1820 }
1821
1822 #[test]
1825 fn the_signed_row_records_which_role_asked() {
1826 let dir = tempdir().unwrap();
1827 let store = test_store(dir.path().join("tx.db"));
1828 let mut tx = queued_transaction();
1829 tx.caller_role = CallerRole::Admin;
1830 store.record(tx).unwrap();
1831
1832 let rows = store.fetch_chain_rows().unwrap();
1833 assert_eq!(rows[0].caller_role.as_deref(), Some("admin"));
1834 }
1835
1836 #[test]
1837 fn approving_consuming_and_revoking_each_append_a_chained_event() {
1838 let dir = tempdir().unwrap();
1839 let key = AuditKey::from_bytes(vec![0x42; 32]);
1840 let store = test_store(dir.path().join("tx.db"));
1841
1842 let a = store.record(queued_transaction()).unwrap();
1844 let receipt = store
1845 .approve_transaction(&a.transaction_id, CallerPrincipal::Uid(1000))
1846 .unwrap()
1847 .expect("queued transaction approves");
1848 let digest = audit_chain::approval_receipt_digest(&receipt);
1849 assert!(store
1850 .claim_approved_for_execution(&a.transaction_id, &digest, CallerPrincipal::Uid(1000))
1851 .unwrap());
1852
1853 let b = store.record(queued_transaction()).unwrap();
1855 store
1856 .approve_transaction(&b.transaction_id, CallerPrincipal::Uid(1000))
1857 .unwrap();
1858 assert!(store
1859 .revoke_unconsumed_approval(&b.transaction_id, CallerPrincipal::Uid(1000))
1860 .unwrap());
1861
1862 let events = store.fetch_event_rows().unwrap();
1863 let kinds: Vec<&str> = events.iter().map(|e| e.kind.as_str()).collect();
1864 assert_eq!(
1865 kinds,
1866 vec![
1867 "approval_granted",
1868 "approval_consumed",
1869 "approval_granted",
1870 "approval_revoked",
1871 ]
1872 );
1873 assert_eq!(
1874 store.verify_event_chain(&key).unwrap(),
1875 VerifyOutcome::Intact { rows_checked: 4 }
1876 );
1877 }
1878
1879 #[test]
1880 fn a_stored_approval_event_names_the_account_that_acted() {
1881 let dir = tempdir().unwrap();
1888 let key = AuditKey::from_bytes(vec![0x42; 32]);
1889 let store = test_store(dir.path().join("tx.db"));
1890
1891 let tx = store.record(queued_transaction()).unwrap();
1892 let receipt = store
1894 .approve_transaction(&tx.transaction_id, CallerPrincipal::Uid(1000))
1895 .unwrap()
1896 .expect("queued transaction approves");
1897 let digest = audit_chain::approval_receipt_digest(&receipt);
1898 assert!(store
1900 .claim_approved_for_execution(&tx.transaction_id, &digest, CallerPrincipal::Uid(1001))
1901 .unwrap());
1902
1903 let events = store.fetch_event_rows().unwrap();
1904 assert_eq!(events.len(), 2);
1905 let granted = events
1906 .iter()
1907 .find(|e| e.kind == "approval_granted")
1908 .expect("a grant event");
1909 let consumed = events
1910 .iter()
1911 .find(|e| e.kind == "approval_consumed")
1912 .expect("a consume event");
1913
1914 assert_eq!(granted.chain_version, EVENT_VERSION_CURRENT);
1916 assert_eq!(consumed.chain_version, EVENT_VERSION_CURRENT);
1917 assert_eq!(
1918 granted.caller_principal.as_deref(),
1919 Some("uid:1000"),
1920 "the grant must name the approver"
1921 );
1922 assert_eq!(
1923 consumed.caller_principal.as_deref(),
1924 Some("uid:1001"),
1925 "the consume must name the executor"
1926 );
1927 assert_ne!(
1928 granted.caller_principal, consumed.caller_principal,
1929 "a grant and a consume by different accounts must be distinguishable"
1930 );
1931
1932 assert_eq!(
1936 store.verify_event_chain(&key).unwrap(),
1937 VerifyOutcome::Intact { rows_checked: 2 }
1938 );
1939 }
1940
1941 #[test]
1942 fn a_stored_status_event_carries_no_principal_and_stays_legacy() {
1943 let dir = tempdir().unwrap();
1948 let key = AuditKey::from_bytes(vec![0x42; 32]);
1949 let store = test_store(dir.path().join("tx.db"));
1950
1951 let tx = store.record(queued_transaction()).unwrap();
1952 let receipt = store
1953 .approve_transaction(&tx.transaction_id, CallerPrincipal::Uid(1000))
1954 .unwrap()
1955 .expect("queued transaction approves");
1956 let digest = audit_chain::approval_receipt_digest(&receipt);
1957 assert!(store
1958 .claim_approved_for_execution(&tx.transaction_id, &digest, CallerPrincipal::Uid(1000))
1959 .unwrap());
1960 store
1961 .update_status(&tx.transaction_id, JobState::Succeeded)
1962 .unwrap();
1963
1964 let events = store.fetch_event_rows().unwrap();
1965 let status = events
1966 .iter()
1967 .find(|e| e.kind == "status_succeeded")
1968 .expect("a status event");
1969 assert_eq!(status.chain_version, EVENT_VERSION_LEGACY);
1970 assert_eq!(
1971 status.caller_principal, None,
1972 "a status event names no account"
1973 );
1974 assert_eq!(
1976 store.verify_event_chain(&key).unwrap(),
1977 VerifyOutcome::Intact { rows_checked: 3 }
1978 );
1979 }
1980
1981 #[test]
1982 fn a_terminal_outcome_is_chained_so_rewriting_status_is_detectable() {
1983 let dir = tempdir().unwrap();
1991 let key = AuditKey::from_bytes(vec![0x42; 32]);
1992 let store = test_store(dir.path().join("tx.db"));
1993
1994 let tx = store.record(queued_transaction()).unwrap();
1995 let receipt = store
1996 .approve_transaction(&tx.transaction_id, CallerPrincipal::Uid(1000))
1997 .unwrap()
1998 .expect("queued transaction approves");
1999 let digest = audit_chain::approval_receipt_digest(&receipt);
2000 assert!(store
2001 .claim_approved_for_execution(&tx.transaction_id, &digest, CallerPrincipal::Uid(1000))
2002 .unwrap());
2003 store
2004 .update_status(&tx.transaction_id, JobState::Succeeded)
2005 .unwrap();
2006
2007 let kinds: Vec<String> = store
2009 .fetch_event_rows()
2010 .unwrap()
2011 .iter()
2012 .map(|e| e.kind.clone())
2013 .collect();
2014 assert!(
2015 kinds.iter().any(|k| k.contains("succeeded")),
2016 "no event records the terminal outcome; the chain covers only {kinds:?}"
2017 );
2018 assert!(
2019 matches!(
2020 store.status_matches_chain().unwrap(),
2021 audit_chain::StatusOutcome::Agrees { .. }
2022 ),
2023 "an untampered store must report agreement, got {:?}",
2024 store.status_matches_chain().unwrap()
2025 );
2026
2027 {
2030 let conn = rusqlite::Connection::open(dir.path().join("tx.db")).unwrap();
2031 let n = conn
2032 .execute(
2033 "UPDATE transactions SET status = ?1 WHERE transaction_id = ?2",
2034 rusqlite::params!["\"canceled\"", tx.transaction_id],
2035 )
2036 .unwrap();
2037 assert_eq!(n, 1, "the fixture did not rewrite the status column");
2038 }
2039 let audit_chain::StatusOutcome::Disagrees { detail } =
2040 store.status_matches_chain().unwrap()
2041 else {
2042 panic!("a rewritten status must be reported, not passed over");
2043 };
2044 assert!(
2045 detail.contains(&tx.transaction_id),
2046 "the report must name the transaction: {detail}"
2047 );
2048 assert!(
2051 matches!(
2052 store.verify_event_chain(&key).unwrap(),
2053 VerifyOutcome::Intact { .. }
2054 ),
2055 "event chain: {:?}, rows: {:?}",
2056 store.verify_event_chain(&key).unwrap(),
2057 store
2058 .fetch_event_rows()
2059 .unwrap()
2060 .iter()
2061 .map(|e| (e.seq, e.kind.clone()))
2062 .collect::<Vec<_>>()
2063 );
2064 }
2065
2066 #[test]
2067 fn a_failed_approval_appends_no_event() {
2068 let dir = tempdir().unwrap();
2072 let store = test_store(dir.path().join("tx.db"));
2073 let tx = store.record(queued_transaction()).unwrap();
2074 store
2075 .approve_transaction(&tx.transaction_id, CallerPrincipal::Uid(1000))
2076 .unwrap();
2077 store
2078 .approve_transaction(&tx.transaction_id, CallerPrincipal::Uid(1000))
2079 .unwrap();
2080 assert_eq!(store.fetch_event_rows().unwrap().len(), 1);
2081
2082 assert!(!store
2084 .revoke_unconsumed_approval("no-such-tx", CallerPrincipal::Uid(1000))
2085 .unwrap());
2086 assert_eq!(store.fetch_event_rows().unwrap().len(), 1);
2087 }
2088
2089 #[test]
2090 fn deleting_the_record_that_an_approval_happened_no_longer_goes_unnoticed() {
2091 let dir = tempdir().unwrap();
2097 let db_path = dir.path().join("tx.db");
2098 let key = AuditKey::from_bytes(vec![0x42; 32]);
2099 let store = test_store(&db_path);
2100
2101 let a = store.record(queued_transaction()).unwrap();
2102 store
2103 .approve_transaction(&a.transaction_id, CallerPrincipal::Uid(1000))
2104 .unwrap();
2105 store.record(queued_transaction()).unwrap();
2108
2109 assert_eq!(
2110 store.verify_event_chain(&key).unwrap(),
2111 VerifyOutcome::Intact { rows_checked: 1 }
2112 );
2113 assert_eq!(
2114 store.verify_event_binding().unwrap(),
2115 audit_chain::BindingOutcome::Consistent {
2116 bindings_checked: 1
2117 }
2118 );
2119
2120 let conn = Connection::open(&db_path).unwrap();
2122 conn.execute("DELETE FROM transaction_approvals", [])
2123 .unwrap();
2124 conn.execute("DELETE FROM audit_events", []).unwrap();
2125 drop(conn);
2126
2127 assert_eq!(
2131 store.verify_event_chain(&key).unwrap(),
2132 VerifyOutcome::Intact { rows_checked: 0 }
2133 );
2134 match store.verify_event_binding().unwrap() {
2135 audit_chain::BindingOutcome::MissingEvent {
2136 transaction_seq, ..
2137 } => assert_eq!(transaction_seq, 2),
2138 other => panic!("expected MissingEvent, got {other:?}"),
2139 }
2140 }
2141
2142 #[test]
2143 fn a_read_only_store_cannot_revoke_or_claim() {
2144 let dir = tempdir().unwrap();
2147 let db_path = dir.path().join("tx.db");
2148 let store = test_store(&db_path);
2149 store.record(queued_transaction()).unwrap();
2150
2151 let read_only = TransactionStore::open_read_only(&db_path).unwrap();
2152 assert!(matches!(
2153 read_only.revoke_unconsumed_approval("tx", CallerPrincipal::Uid(1000)),
2154 Err(TransactionStoreError::AuditChainMissing(_))
2155 ));
2156 assert!(matches!(
2157 read_only.claim_approved_for_execution("tx", "digest", CallerPrincipal::Uid(1000)),
2158 Err(TransactionStoreError::AuditChainMissing(_))
2159 ));
2160 }
2161
2162 #[test]
2169 fn record_persists_the_caller_principal_under_the_current_encoding() {
2170 let dir = tempdir().unwrap();
2171 let store = test_store(dir.path().join("tx.db"));
2172 let mut tx = queued_transaction();
2173 tx.caller_principal = CallerPrincipal::Uid(4242);
2174 store.record(tx).unwrap();
2175
2176 let conn = store.connection().unwrap();
2177 let (version, role, principal): (i64, Option<String>, Option<String>) = conn
2178 .query_row(
2179 "SELECT chain_version, caller_role, caller_principal FROM transactions",
2180 [],
2181 |row| Ok((row.get(0)?, row.get(1)?, row.get(2)?)),
2182 )
2183 .unwrap();
2184
2185 assert_eq!(version, i64::from(audit_chain::CHAIN_VERSION_CURRENT));
2186 assert_eq!(role.as_deref(), Some("dev"));
2187 assert_eq!(
2188 principal.as_deref(),
2189 Some("uid:4242"),
2190 "the row must name the account the dispatcher resolved"
2191 );
2192
2193 let key = AuditKey::from_bytes(vec![0x42; 32]);
2195 assert!(matches!(
2196 store.verify_audit_chain(&key).unwrap(),
2197 audit_chain::VerifyOutcome::Intact { .. }
2198 ));
2199 }
2200
2201 #[test]
2202 fn record_writes_audit_chain_columns() {
2203 let dir = tempdir().unwrap();
2204 let store = test_store(dir.path().join("tx.db"));
2205 let _record = store.record(queued_transaction()).unwrap();
2206
2207 let conn = store.connection().unwrap();
2208 let (seq, key_id, chain_hash, prev): (i64, String, String, String) = conn
2209 .query_row(
2210 "SELECT seq, key_id, chain_hash, prev_chain_hash FROM transactions",
2211 [],
2212 |row| Ok((row.get(0)?, row.get(1)?, row.get(2)?, row.get(3)?)),
2213 )
2214 .unwrap();
2215 assert_eq!(seq, 1, "first row gets seq=1");
2216 assert_eq!(key_id, audit_chain::CURRENT_KEY_ID);
2217 assert_eq!(prev, "", "first row has empty prev_chain_hash");
2218 assert_eq!(
2219 chain_hash.len(),
2220 audit_chain::HASH_HEX_LEN,
2221 "chain_hash is a hex-encoded Ed25519 signature"
2222 );
2223 }
2224
2225 #[test]
2226 fn sequential_records_produce_chained_hashes() {
2227 let dir = tempdir().unwrap();
2228 let store = test_store(dir.path().join("tx.db"));
2229 store.record(queued_transaction()).unwrap();
2230 store.record(queued_transaction()).unwrap();
2231 store.record(queued_transaction()).unwrap();
2232
2233 let rows = store.fetch_chain_rows().unwrap();
2234 assert_eq!(rows.len(), 3);
2235 assert_eq!(rows[0].seq, 1);
2236 assert_eq!(rows[1].seq, 2);
2237 assert_eq!(rows[2].seq, 3);
2238 assert_eq!(rows[1].prev_chain_hash, rows[0].chain_hash);
2240 assert_eq!(rows[2].prev_chain_hash, rows[1].chain_hash);
2241 }
2242
2243 #[test]
2255 fn concurrent_record_keeps_chain_intact_and_seqs_contiguous() {
2256 const WORKERS: usize = 8;
2257 const RECORDS_PER_WORKER: usize = 10;
2258 const TOTAL: usize = WORKERS * RECORDS_PER_WORKER;
2259
2260 let dir = tempdir().unwrap();
2261 let store = std::sync::Arc::new(test_store(dir.path().join("tx.db")));
2262
2263 let mut handles = Vec::with_capacity(WORKERS);
2264 for w in 0..WORKERS {
2265 let store = std::sync::Arc::clone(&store);
2266 handles.push(std::thread::spawn(move || {
2267 for r in 0..RECORDS_PER_WORKER {
2268 let tx = NewTransaction {
2269 request_id: format!("worker-{w}-record-{r}"),
2270 request_hash: format!("hash-{w}-{r}"),
2271 action_name: "GetSystemState".to_string(),
2272 risk_level: RiskLevel::Low,
2273 summary: format!("worker {w} record {r}"),
2274 warnings: vec![],
2275 caller_role: CallerRole::Dev,
2276 caller_principal: CallerPrincipal::Uid(3000 + w as u32),
2280 };
2281 store
2282 .record(tx)
2283 .expect("record must succeed under contention");
2284 }
2285 }));
2286 }
2287 for h in handles {
2288 h.join().expect("worker thread did not panic");
2289 }
2290
2291 let key = AuditKey::from_bytes(vec![0x42; 32]);
2293 let outcome = store.verify_audit_chain(&key).unwrap();
2294 match outcome {
2295 VerifyOutcome::Intact { rows_checked } => {
2296 assert_eq!(
2297 rows_checked, TOTAL as u64,
2298 "expected {TOTAL} rows checked, got {rows_checked}"
2299 );
2300 }
2301 other => panic!("chain must be Intact under concurrent writes; got {other:?}"),
2302 }
2303
2304 {
2308 let conn = store.connection().unwrap();
2309 let mut stmt = conn
2310 .prepare("SELECT summary, caller_principal FROM transactions")
2311 .unwrap();
2312 let pairs: Vec<(String, Option<String>)> = stmt
2313 .query_map([], |row| Ok((row.get(0)?, row.get(1)?)))
2314 .unwrap()
2315 .map(|r| r.unwrap())
2316 .collect();
2317 assert_eq!(pairs.len(), TOTAL);
2318 for (summary, principal) in pairs {
2319 let worker: u32 = summary
2320 .split_whitespace()
2321 .nth(1)
2322 .and_then(|w| w.parse().ok())
2323 .unwrap_or_else(|| panic!("unexpected summary {summary:?}"));
2324 assert_eq!(
2325 principal.as_deref(),
2326 Some(format!("uid:{}", 3000 + worker).as_str()),
2327 "row {summary:?} was attributed to the wrong account"
2328 );
2329 }
2330 }
2331
2332 let conn = store.connection().unwrap();
2334 let mut stmt = conn
2335 .prepare("SELECT seq FROM transactions ORDER BY seq ASC")
2336 .unwrap();
2337 let seqs: Vec<i64> = stmt
2338 .query_map([], |row| row.get::<_, i64>(0))
2339 .unwrap()
2340 .map(|r| r.unwrap())
2341 .collect();
2342 assert_eq!(seqs.len(), TOTAL, "row count mismatch");
2343 for (i, s) in seqs.iter().enumerate() {
2344 assert_eq!(
2345 *s,
2346 (i as i64) + 1,
2347 "seq column must be contiguous 1..={TOTAL}; saw {s} at position {i}"
2348 );
2349 }
2350 }
2351
2352 #[test]
2353 fn verify_audit_chain_intact_after_inserts() {
2354 let dir = tempdir().unwrap();
2355 let store = test_store(dir.path().join("tx.db"));
2356 for _ in 0..3 {
2357 store.record(queued_transaction()).unwrap();
2358 }
2359 let key = AuditKey::from_bytes(vec![0x42; 32]);
2360 let outcome = store.verify_audit_chain(&key).unwrap();
2361 assert!(matches!(outcome, VerifyOutcome::Intact { rows_checked: 3 }));
2362 }
2363
2364 #[test]
2365 fn verify_audit_chain_with_pubkey_intact_after_inserts() {
2366 let dir = tempdir().unwrap();
2367 let store = test_store(dir.path().join("tx.db"));
2368 for _ in 0..3 {
2369 store.record(queued_transaction()).unwrap();
2370 }
2371 let key = AuditKey::from_bytes(vec![0x42; 32]);
2373 let outcome = store
2374 .verify_audit_chain_with_pubkey(&key.verifying_key_hex())
2375 .unwrap();
2376 assert!(matches!(outcome, VerifyOutcome::Intact { rows_checked: 3 }));
2377 }
2378
2379 #[test]
2380 fn verify_audit_chain_with_wrong_pubkey_is_broken() {
2381 let dir = tempdir().unwrap();
2382 let store = test_store(dir.path().join("tx.db"));
2383 store.record(queued_transaction()).unwrap();
2384 let other = AuditKey::from_bytes(vec![0x99; 32]);
2386 let outcome = store
2387 .verify_audit_chain_with_pubkey(&other.verifying_key_hex())
2388 .unwrap();
2389 assert!(matches!(outcome, VerifyOutcome::Broken { .. }));
2390 }
2391
2392 #[test]
2393 fn verify_detects_tampered_summary() {
2394 let dir = tempdir().unwrap();
2395 let store = test_store(dir.path().join("tx.db"));
2396 let tx = store.record(queued_transaction()).unwrap();
2397
2398 let conn = store.connection().unwrap();
2401 conn.execute(
2402 "UPDATE transactions SET summary = ?1 WHERE transaction_id = ?2",
2403 params!["EVIL CHANGE", tx.transaction_id],
2404 )
2405 .unwrap();
2406
2407 let key = AuditKey::from_bytes(vec![0x42; 32]);
2408 let outcome = store.verify_audit_chain(&key).unwrap();
2409 assert!(matches!(outcome, VerifyOutcome::Broken { .. }));
2410 }
2411
2412 #[test]
2413 fn status_update_does_not_break_chain() {
2414 let dir = tempdir().unwrap();
2416 let store = test_store(dir.path().join("tx.db"));
2417 let tx = store.record(queued_transaction()).unwrap();
2418 store
2419 .update_status(&tx.transaction_id, JobState::Running)
2420 .unwrap();
2421 store
2422 .update_status(&tx.transaction_id, JobState::Succeeded)
2423 .unwrap();
2424
2425 let key = AuditKey::from_bytes(vec![0x42; 32]);
2426 let outcome = store.verify_audit_chain(&key).unwrap();
2427 assert!(
2428 matches!(outcome, VerifyOutcome::Intact { rows_checked: 1 }),
2429 "status mutation must not break the chain (status not in chain content): {outcome:?}"
2430 );
2431 }
2432
2433 #[test]
2434 fn open_read_only_rejects_record() {
2435 let dir = tempdir().unwrap();
2436 let key_path = dir.path().join("audit-key");
2437 std::fs::write(&key_path, vec![0x42; 32]).unwrap();
2438 std::fs::set_permissions(&key_path, std::fs::Permissions::from_mode(0o600)).unwrap();
2439
2440 let store = TransactionStore::open_read_only(dir.path().join("tx.db")).unwrap();
2441 let result = store.record(queued_transaction());
2442 assert!(matches!(
2443 result,
2444 Err(TransactionStoreError::AuditChainMissing(_))
2445 ));
2446 }
2447
2448 #[test]
2449 fn ensure_private_dir_creates_with_0700_mode() {
2450 let root = tempdir().unwrap();
2451 let target = root.path().join("a/b/c");
2452 ensure_private_dir(&target).unwrap();
2453 assert!(target.is_dir());
2454 let mode = std::fs::metadata(&target).unwrap().permissions().mode() & 0o777;
2455 assert_eq!(mode, 0o700, "leaf dir must be 0o700, got {mode:o}");
2456 }
2457
2458 #[test]
2466 fn the_audit_database_is_created_owner_only() {
2467 let dir = tempdir().unwrap();
2468 let db = dir.path().join("daemon.sqlite");
2469 let store = TransactionStore::open(&db).unwrap();
2470 store
2472 .record(NewTransaction {
2473 request_id: "mode-probe".to_string(),
2474 request_hash: "hash-mode-probe".to_string(),
2475 action_name: "GetSystemState".to_string(),
2476 risk_level: RiskLevel::Low,
2477 summary: "probe".to_string(),
2478 warnings: vec![],
2479 caller_role: CallerRole::Dev,
2480 caller_principal: CallerPrincipal::Uid(3000),
2481 })
2482 .unwrap();
2483
2484 for path in [db.clone(), db.with_extension("sqlite-wal")] {
2485 if !path.exists() {
2486 continue;
2487 }
2488 let mode = std::fs::metadata(&path).unwrap().permissions().mode() & 0o777;
2489 assert_eq!(
2490 mode,
2491 0o600,
2492 "{} must be owner-only, got {mode:o}",
2493 path.display()
2494 );
2495 }
2496 }
2497
2498 #[test]
2499 fn ensure_private_dir_is_idempotent_and_does_not_widen_existing_mode() {
2500 let root = tempdir().unwrap();
2501 let target = root.path().join("preexisting");
2502 std::fs::create_dir(&target).unwrap();
2503 std::fs::set_permissions(&target, std::fs::Permissions::from_mode(0o755)).unwrap();
2504 ensure_private_dir(&target).unwrap();
2505 let mode = std::fs::metadata(&target).unwrap().permissions().mode() & 0o777;
2507 assert_eq!(mode, 0o755);
2508 }
2509
2510 #[test]
2511 fn open_creates_parent_with_private_mode() {
2512 let root = tempdir().unwrap();
2513 let db_path = root.path().join("nested/dirs/daemon.sqlite");
2514 let _store = test_store(&db_path);
2515 let parent = db_path.parent().unwrap();
2516 let mode = std::fs::metadata(parent).unwrap().permissions().mode() & 0o777;
2517 assert_eq!(mode, 0o700);
2518 }
2519
2520 fn queued_transaction() -> NewTransaction {
2521 NewTransaction {
2522 request_id: "req-1".to_string(),
2523 request_hash: "hash-abc".to_string(),
2524 action_name: "UpdateSystem".to_string(),
2525 risk_level: RiskLevel::High,
2526 summary: "Upgrade the system".to_string(),
2527 warnings: vec![],
2528 caller_role: CallerRole::Dev,
2529 caller_principal: CallerPrincipal::Uid(1000),
2530 }
2531 }
2532
2533 #[test]
2534 fn update_status_transitions_queued_to_running() {
2535 let dir = tempdir().unwrap();
2536 let store = test_store(dir.path().join("tx.db"));
2537 let tx = store.record(queued_transaction()).unwrap();
2538
2539 store
2540 .update_status(&tx.transaction_id, JobState::Running)
2541 .unwrap();
2542
2543 let updated = store.get(&tx.transaction_id).unwrap().unwrap();
2544 assert_eq!(updated.status, JobState::Running);
2545 }
2546
2547 #[test]
2548 fn update_status_transitions_running_to_succeeded() {
2549 let dir = tempdir().unwrap();
2550 let store = test_store(dir.path().join("tx.db"));
2551 let tx = store.record(queued_transaction()).unwrap();
2552
2553 store
2554 .update_status(&tx.transaction_id, JobState::Running)
2555 .unwrap();
2556 store
2557 .update_status(&tx.transaction_id, JobState::Succeeded)
2558 .unwrap();
2559
2560 let updated = store.get(&tx.transaction_id).unwrap().unwrap();
2561 assert_eq!(updated.status, JobState::Succeeded);
2562 }
2563
2564 #[test]
2565 fn update_status_for_unknown_id_returns_not_found() {
2566 let dir = tempdir().unwrap();
2567 let store = test_store(dir.path().join("tx.db"));
2568
2569 let result = store.update_status("does-not-exist", JobState::Running);
2570 assert!(
2571 matches!(result, Err(TransactionStoreError::NotFound(ref id)) if id == "does-not-exist"),
2572 "expected NotFound, got: {result:?}"
2573 );
2574 }
2575
2576 #[test]
2577 fn update_status_leaves_other_fields_intact() {
2578 let dir = tempdir().unwrap();
2579 let store = test_store(dir.path().join("tx.db"));
2580 let tx = store.record(queued_transaction()).unwrap();
2581
2582 store
2583 .update_status(&tx.transaction_id, JobState::Running)
2584 .unwrap();
2585 store
2586 .update_status(&tx.transaction_id, JobState::Failed)
2587 .unwrap();
2588
2589 let updated = store.get(&tx.transaction_id).unwrap().unwrap();
2590 assert_eq!(updated.action_name, "UpdateSystem");
2591 assert_eq!(updated.risk_level, RiskLevel::High);
2592 assert_eq!(updated.status, JobState::Failed);
2593 }
2594
2595 #[test]
2604 fn update_status_is_atomic_under_concurrent_conflicting_transitions() {
2605 let dir = tempdir().unwrap();
2606 let store = std::sync::Arc::new(test_store(dir.path().join("tx.db")));
2607 let tx = store.record(queued_transaction()).unwrap();
2608 store
2609 .update_status(&tx.transaction_id, JobState::Running)
2610 .unwrap();
2611
2612 let store_a = std::sync::Arc::clone(&store);
2613 let id_a = tx.transaction_id.clone();
2614 let succeed = std::thread::spawn(move || store_a.update_status(&id_a, JobState::Succeeded));
2615
2616 let store_b = std::sync::Arc::clone(&store);
2617 let id_b = tx.transaction_id.clone();
2618 let fail = std::thread::spawn(move || store_b.update_status(&id_b, JobState::Failed));
2619
2620 let succeed_result = succeed.join().unwrap();
2621 let fail_result = fail.join().unwrap();
2622
2623 let oks = [succeed_result.is_ok(), fail_result.is_ok()]
2625 .into_iter()
2626 .filter(|ok| *ok)
2627 .count();
2628 assert_eq!(
2629 oks, 1,
2630 "exactly one concurrent transition should succeed: succeed={succeed_result:?} \
2631 fail={fail_result:?}"
2632 );
2633
2634 let loser = if succeed_result.is_ok() {
2650 &fail_result
2651 } else {
2652 &succeed_result
2653 };
2654 match loser {
2655 Err(TransactionStoreError::ConcurrentStatusChange(id)) => {
2656 assert_eq!(id, &tx.transaction_id);
2657 }
2658 Err(TransactionStoreError::InvalidTransition { from, to }) => {
2659 let winner = if succeed_result.is_ok() {
2661 JobState::Succeeded
2662 } else {
2663 JobState::Failed
2664 };
2665 assert_eq!(
2666 *from, winner,
2667 "the loser must have observed the winner's committed status"
2668 );
2669 assert_ne!(*to, winner, "the loser must be the other transition");
2670 }
2671 other => panic!(
2672 "the losing transition must be refused with either \
2673 ConcurrentStatusChange or InvalidTransition, got: {other:?}"
2674 ),
2675 }
2676
2677 let final_status = store.get(&tx.transaction_id).unwrap().unwrap().status;
2680 if succeed_result.is_ok() {
2681 assert_eq!(final_status, JobState::Succeeded);
2682 } else {
2683 assert_eq!(final_status, JobState::Failed);
2684 }
2685 }
2686
2687 #[test]
2688 fn revoke_unconsumed_approval_removes_an_unused_receipt() {
2689 let dir = tempdir().unwrap();
2692 let store = test_store(dir.path().join("tx.db"));
2693 let tx = store.record(queued_transaction()).unwrap();
2694 let receipt = store
2695 .approve_transaction(&tx.transaction_id, CallerPrincipal::Uid(1000))
2696 .unwrap()
2697 .expect("approved");
2698 let digest = audit_chain::approval_receipt_digest(&receipt);
2699
2700 assert!(
2701 store
2702 .revoke_unconsumed_approval(&tx.transaction_id, CallerPrincipal::Uid(1000))
2703 .unwrap(),
2704 "an unconsumed approval must be revocable"
2705 );
2706 assert!(
2707 !store
2708 .claim_approved_for_execution(
2709 &tx.transaction_id,
2710 &digest,
2711 CallerPrincipal::Uid(1000)
2712 )
2713 .unwrap(),
2714 "a revoked receipt must no longer be claimable"
2715 );
2716 }
2717
2718 #[test]
2719 fn revoke_cannot_retract_an_approval_that_was_already_used() {
2720 let dir = tempdir().unwrap();
2724 let store = test_store(dir.path().join("tx.db"));
2725 let tx = store.record(queued_transaction()).unwrap();
2726 let receipt = store
2727 .approve_transaction(&tx.transaction_id, CallerPrincipal::Uid(1000))
2728 .unwrap()
2729 .expect("approved");
2730 let digest = audit_chain::approval_receipt_digest(&receipt);
2731 assert!(store
2732 .claim_approved_for_execution(&tx.transaction_id, &digest, CallerPrincipal::Uid(1000))
2733 .unwrap());
2734
2735 assert!(
2736 !store
2737 .revoke_unconsumed_approval(&tx.transaction_id, CallerPrincipal::Uid(1000))
2738 .unwrap(),
2739 "a consumed approval must not be revocable"
2740 );
2741 assert_eq!(
2742 store.get(&tx.transaction_id).unwrap().unwrap().status,
2743 JobState::Running,
2744 "a failed revocation must not disturb the running transaction"
2745 );
2746 }
2747
2748 #[test]
2749 fn approve_refuses_a_transaction_that_is_no_longer_queued() {
2750 let dir = tempdir().unwrap();
2754 let store = test_store(dir.path().join("tx.db"));
2755 let tx = store.record(queued_transaction()).unwrap();
2756 let receipt = store
2757 .approve_transaction(&tx.transaction_id, CallerPrincipal::Uid(1000))
2758 .unwrap()
2759 .expect("approved");
2760 let digest = audit_chain::approval_receipt_digest(&receipt);
2761 assert!(store
2762 .claim_approved_for_execution(&tx.transaction_id, &digest, CallerPrincipal::Uid(1000))
2763 .unwrap());
2764
2765 assert!(
2767 store
2768 .approve_transaction(&tx.transaction_id, CallerPrincipal::Uid(1000))
2769 .unwrap()
2770 .is_none(),
2771 "a Running transaction must not be approvable"
2772 );
2773
2774 store
2776 .update_status(&tx.transaction_id, JobState::Succeeded)
2777 .unwrap();
2778 assert!(
2779 store
2780 .approve_transaction(&tx.transaction_id, CallerPrincipal::Uid(1000))
2781 .unwrap()
2782 .is_none(),
2783 "a completed transaction must not be approvable"
2784 );
2785 }
2786
2787 #[test]
2788 fn only_one_of_two_concurrent_claims_can_execute() {
2789 let dir = tempdir().unwrap();
2794 let store = std::sync::Arc::new(test_store(dir.path().join("tx.db")));
2795 let tx = store.record(queued_transaction()).unwrap();
2796 let receipt = store
2797 .approve_transaction(&tx.transaction_id, CallerPrincipal::Uid(1000))
2798 .unwrap()
2799 .expect("approved");
2800 let digest = audit_chain::approval_receipt_digest(&receipt);
2801
2802 let mut handles = Vec::new();
2803 for _ in 0..2 {
2804 let store = std::sync::Arc::clone(&store);
2805 let id = tx.transaction_id.clone();
2806 let digest = digest.clone();
2807 handles.push(std::thread::spawn(move || {
2808 store.claim_approved_for_execution(&id, &digest, CallerPrincipal::Uid(1000))
2809 }));
2810 }
2811 let claims: Vec<bool> = handles
2812 .into_iter()
2813 .map(|h| h.join().unwrap().unwrap_or(false))
2814 .collect();
2815
2816 assert_eq!(
2817 claims.iter().filter(|c| **c).count(),
2818 1,
2819 "exactly one claim may win: {claims:?}"
2820 );
2821 assert_eq!(
2822 store.get(&tx.transaction_id).unwrap().unwrap().status,
2823 JobState::Running
2824 );
2825 }
2826
2827 #[test]
2828 fn cancel_queued_cancels_a_queued_transaction_once() {
2829 let dir = tempdir().unwrap();
2830 let store = test_store(dir.path().join("tx.db"));
2831 let tx = store.record(queued_transaction()).unwrap();
2832
2833 assert!(
2834 store
2835 .cancel_queued(&tx.transaction_id, CallerPrincipal::Uid(1000))
2836 .unwrap(),
2837 "a queued transaction is cancelable"
2838 );
2839 assert_eq!(
2840 store.get(&tx.transaction_id).unwrap().unwrap().status,
2841 JobState::Canceled
2842 );
2843 assert!(
2844 !store
2845 .cancel_queued(&tx.transaction_id, CallerPrincipal::Uid(1000))
2846 .unwrap(),
2847 "an already-canceled transaction is not cancelable again"
2848 );
2849 assert!(
2850 !store
2851 .cancel_queued("no-such-transaction", CallerPrincipal::Uid(1000))
2852 .unwrap(),
2853 "a missing transaction is not cancelable"
2854 );
2855 assert!(
2856 store.fetch_event_rows().unwrap().is_empty(),
2857 "canceling an unapproved transaction must not append an approval event"
2858 );
2859 }
2860
2861 #[test]
2862 fn cancel_queued_revokes_an_unconsumed_approval_and_appends_event() {
2863 let dir = tempdir().unwrap();
2864 let store = test_store(dir.path().join("tx.db"));
2865 let tx = store.record(queued_transaction()).unwrap();
2866 let receipt = store
2867 .approve_transaction(&tx.transaction_id, CallerPrincipal::Uid(1000))
2868 .unwrap()
2869 .expect("queued transaction is approvable");
2870 let digest = audit_chain::approval_receipt_digest(&receipt);
2871
2872 assert!(store
2873 .cancel_queued(&tx.transaction_id, CallerPrincipal::Uid(1000))
2874 .unwrap());
2875 assert_eq!(
2876 store.get(&tx.transaction_id).unwrap().unwrap().status,
2877 JobState::Canceled
2878 );
2879 assert!(
2880 !store
2881 .claim_approved_for_execution(
2882 &tx.transaction_id,
2883 &digest,
2884 CallerPrincipal::Uid(1000)
2885 )
2886 .unwrap(),
2887 "canceling must revoke the unconsumed receipt"
2888 );
2889 assert_eq!(
2890 store
2891 .fetch_event_rows()
2892 .unwrap()
2893 .iter()
2894 .map(|event| event.kind.as_str())
2895 .collect::<Vec<_>>(),
2896 vec!["approval_granted", "approval_revoked"]
2897 );
2898 }
2899
2900 #[test]
2901 fn cancel_queued_refuses_a_running_transaction() {
2902 let dir = tempdir().unwrap();
2905 let store = test_store(dir.path().join("tx.db"));
2906 let tx = store.record(queued_transaction()).unwrap();
2907 let receipt = store
2908 .approve_transaction(&tx.transaction_id, CallerPrincipal::Uid(1000))
2909 .unwrap()
2910 .expect("approved");
2911 let digest = audit_chain::approval_receipt_digest(&receipt);
2912 assert!(store
2913 .claim_approved_for_execution(&tx.transaction_id, &digest, CallerPrincipal::Uid(1000))
2914 .unwrap());
2915
2916 assert!(
2917 !store
2918 .cancel_queued(&tx.transaction_id, CallerPrincipal::Uid(1000))
2919 .unwrap(),
2920 "a running transaction must not be cancelable"
2921 );
2922 assert_eq!(
2923 store.get(&tx.transaction_id).unwrap().unwrap().status,
2924 JobState::Running,
2925 "cancel must not disturb a running transaction"
2926 );
2927 }
2928
2929 #[test]
2930 fn approved_receipt_is_required_and_consumed_once() {
2931 let dir = tempdir().unwrap();
2932 let store = test_store(dir.path().join("tx.db"));
2933 let tx = store.record(queued_transaction()).unwrap();
2934
2935 assert!(
2936 !store
2937 .claim_approved_for_execution(
2938 &tx.transaction_id,
2939 "digest-a",
2940 CallerPrincipal::Uid(1000)
2941 )
2942 .unwrap(),
2943 "an unapproved preview must not execute"
2944 );
2945 let receipt = store
2946 .approve_transaction(&tx.transaction_id, CallerPrincipal::Uid(1000))
2947 .unwrap()
2948 .expect("first approval must succeed");
2949 let digest = audit_chain::approval_receipt_digest(&receipt);
2950 assert_eq!(tx.approval_id.as_deref(), Some(digest.as_str()));
2951 assert!(
2952 store
2953 .approve_transaction(&tx.transaction_id, CallerPrincipal::Uid(1000))
2954 .unwrap()
2955 .is_none(),
2956 "approval is immutable once issued"
2957 );
2958 assert!(
2959 !store
2960 .claim_approved_for_execution(
2961 &tx.transaction_id,
2962 "wrong-digest",
2963 CallerPrincipal::Uid(1000)
2964 )
2965 .unwrap(),
2966 "a forged receipt must not execute"
2967 );
2968 assert!(
2969 store
2970 .claim_approved_for_execution(
2971 &tx.transaction_id,
2972 &digest,
2973 CallerPrincipal::Uid(1000)
2974 )
2975 .unwrap(),
2976 "the exact approved receipt must execute"
2977 );
2978 assert!(
2979 !store
2980 .claim_approved_for_execution(
2981 &tx.transaction_id,
2982 &digest,
2983 CallerPrincipal::Uid(1000)
2984 )
2985 .unwrap(),
2986 "the receipt must be one-time"
2987 );
2988 }
2989
2990 #[test]
2991 fn approval_commitment_is_covered_by_the_signed_chain() {
2992 let dir = tempdir().unwrap();
2993 let store = test_store(dir.path().join("tx.db"));
2994 let tx = store.record(queued_transaction()).unwrap();
2995 assert!(tx.approval_id.is_some());
2996
2997 store
2998 .connection()
2999 .unwrap()
3000 .execute(
3001 "UPDATE transactions SET approval_id = 'forged' WHERE transaction_id = ?1",
3002 params![tx.transaction_id],
3003 )
3004 .unwrap();
3005
3006 let key = AuditKey::from_bytes(vec![0x42; 32]);
3007 assert!(matches!(
3008 store.verify_audit_chain(&key).unwrap(),
3009 VerifyOutcome::Broken { .. }
3010 ));
3011 }
3012
3013 #[test]
3014 fn approve_rejects_a_forged_commitment_at_runtime() {
3015 let dir = tempdir().unwrap();
3019 let store = test_store(dir.path().join("tx.db"));
3020 let tx = store.record(queued_transaction()).unwrap();
3021
3022 store
3023 .connection()
3024 .unwrap()
3025 .execute(
3026 "UPDATE transactions SET approval_id = 'forged' WHERE transaction_id = ?1",
3027 params![tx.transaction_id],
3028 )
3029 .unwrap();
3030
3031 let err = store
3032 .approve_transaction(&tx.transaction_id, CallerPrincipal::Uid(1000))
3033 .expect_err("a forged commitment must be rejected, not approved");
3034 assert!(
3035 matches!(err, TransactionStoreError::DatabaseInvariant(_)),
3036 "forged commitment must surface as a DatabaseInvariant, got {err:?}"
3037 );
3038 }
3039
3040 #[test]
3041 fn stale_iso_timestamp_cannot_be_approved() {
3042 let dir = tempdir().unwrap();
3043 let store = test_store(dir.path().join("tx.db"));
3044 let tx = store.record(queued_transaction()).unwrap();
3045 let conn = store.connection().unwrap();
3046 conn.execute(
3047 "UPDATE transactions \
3048 SET created_at = strftime('%Y-%m-%dT%H:%M:%fZ', 'now', '-20 minutes') \
3049 WHERE transaction_id = ?1",
3050 params![tx.transaction_id],
3051 )
3052 .unwrap();
3053
3054 assert!(
3055 store
3056 .approve_transaction(&tx.transaction_id, CallerPrincipal::Uid(1000))
3057 .unwrap()
3058 .is_none(),
3059 "a production-format timestamp outside the TTL must not be approved"
3060 );
3061 }
3062
3063 #[test]
3064 fn stale_iso_timestamp_cannot_be_claimed_for_execution() {
3065 let dir = tempdir().unwrap();
3070 let store = test_store(dir.path().join("tx.db"));
3071 let tx = store.record(queued_transaction()).unwrap();
3072 let receipt = store
3073 .approve_transaction(&tx.transaction_id, CallerPrincipal::Uid(1000))
3074 .unwrap()
3075 .expect("a fresh approval succeeds");
3076 let digest = audit_chain::approval_receipt_digest(&receipt);
3077
3078 let conn = store.connection().unwrap();
3080 conn.execute(
3081 "UPDATE transactions \
3082 SET created_at = strftime('%Y-%m-%dT%H:%M:%fZ', 'now', '-20 minutes') \
3083 WHERE transaction_id = ?1",
3084 params![tx.transaction_id],
3085 )
3086 .unwrap();
3087
3088 assert!(
3089 !store
3090 .claim_approved_for_execution(
3091 &tx.transaction_id,
3092 &digest,
3093 CallerPrincipal::Uid(1000)
3094 )
3095 .unwrap(),
3096 "an approval aged past the TTL must not be claimable at execute time"
3097 );
3098 assert_eq!(
3099 store.get(&tx.transaction_id).unwrap().unwrap().status,
3100 JobState::Queued,
3101 "a TTL-expired claim must leave the transaction Queued, never Running"
3102 );
3103 }
3104
3105 #[test]
3106 fn cleanup_stale_queued_cancels_old_records() {
3107 let dir = tempdir().unwrap();
3108 let store = test_store(dir.path().join("tx.db"));
3109
3110 let fresh = store.record(queued_transaction()).unwrap();
3112 let stale = store.record(queued_transaction()).unwrap();
3113
3114 let conn = store.connection().unwrap();
3116 conn.execute(
3117 "UPDATE transactions \
3118 SET created_at = strftime('%Y-%m-%dT%H:%M:%fZ', 'now', '-20 minutes') \
3119 WHERE transaction_id = ?1",
3120 params![stale.transaction_id],
3121 )
3122 .unwrap();
3123
3124 let canceled = store.cleanup_stale_queued().unwrap();
3125 assert_eq!(canceled, 1, "only the stale record should be canceled");
3126
3127 let stale_record = store.get(&stale.transaction_id).unwrap().unwrap();
3129 assert_eq!(stale_record.status, JobState::Canceled);
3130
3131 let fresh_record = store.get(&fresh.transaction_id).unwrap().unwrap();
3133 assert_eq!(fresh_record.status, JobState::Queued);
3134 }
3135
3136 #[test]
3137 fn cleanup_stale_queued_revokes_every_unconsumed_approval() {
3138 let dir = tempdir().unwrap();
3139 let store = test_store(dir.path().join("tx.db"));
3140 let transactions: Vec<_> = (0..3)
3141 .map(|_| {
3142 let transaction = store.record(queued_transaction()).unwrap();
3143 store
3144 .approve_transaction(&transaction.transaction_id, CallerPrincipal::Uid(1000))
3145 .unwrap()
3146 .expect("stale transaction is approvable");
3147 transaction
3148 })
3149 .collect();
3150
3151 let conn = store.connection().unwrap();
3152 for transaction in &transactions {
3153 conn.execute(
3154 "UPDATE transactions \
3155 SET created_at = strftime('%Y-%m-%dT%H:%M:%fZ', 'now', '-20 minutes') \
3156 WHERE transaction_id = ?1",
3157 params![transaction.transaction_id],
3158 )
3159 .unwrap();
3160 }
3161
3162 assert_eq!(store.cleanup_stale_queued().unwrap(), 3);
3163 for transaction in &transactions {
3164 assert_eq!(
3165 store
3166 .get(&transaction.transaction_id)
3167 .unwrap()
3168 .unwrap()
3169 .status,
3170 JobState::Canceled
3171 );
3172 assert!(!store
3173 .revoke_unconsumed_approval(&transaction.transaction_id, CallerPrincipal::Uid(1000))
3174 .unwrap());
3175 }
3176 assert_eq!(
3177 store
3178 .fetch_event_rows()
3179 .unwrap()
3180 .iter()
3181 .map(|event| event.kind.as_str())
3182 .collect::<Vec<_>>(),
3183 vec![
3184 "approval_granted",
3185 "approval_granted",
3186 "approval_granted",
3187 "approval_revoked",
3188 "approval_revoked",
3189 "approval_revoked"
3190 ]
3191 );
3192 }
3193
3194 #[test]
3197 fn update_status_rejects_queued_to_succeeded() {
3198 let dir = tempdir().unwrap();
3199 let store = test_store(dir.path().join("tx.db"));
3200 let tx = store.record(queued_transaction()).unwrap();
3201
3202 let result = store.update_status(&tx.transaction_id, JobState::Succeeded);
3203 assert!(
3204 matches!(
3205 result,
3206 Err(TransactionStoreError::InvalidTransition {
3207 from: JobState::Queued,
3208 to: JobState::Succeeded,
3209 })
3210 ),
3211 "Queued -> Succeeded must be rejected (must go through Running first): {result:?}"
3212 );
3213 }
3214
3215 #[test]
3216 fn update_status_rejects_succeeded_to_running() {
3217 let dir = tempdir().unwrap();
3218 let store = test_store(dir.path().join("tx.db"));
3219 let tx = store.record(queued_transaction()).unwrap();
3220
3221 store
3222 .update_status(&tx.transaction_id, JobState::Running)
3223 .unwrap();
3224 store
3225 .update_status(&tx.transaction_id, JobState::Succeeded)
3226 .unwrap();
3227
3228 let result = store.update_status(&tx.transaction_id, JobState::Running);
3229 assert!(
3230 matches!(
3231 result,
3232 Err(TransactionStoreError::InvalidTransition {
3233 from: JobState::Succeeded,
3234 to: JobState::Running,
3235 })
3236 ),
3237 "Succeeded -> Running must be rejected (terminal state): {result:?}"
3238 );
3239 }
3240
3241 #[test]
3242 fn update_status_accepts_running_to_failed() {
3243 let dir = tempdir().unwrap();
3244 let store = test_store(dir.path().join("tx.db"));
3245 let tx = store.record(queued_transaction()).unwrap();
3246
3247 store
3248 .update_status(&tx.transaction_id, JobState::Running)
3249 .unwrap();
3250 store
3251 .update_status(&tx.transaction_id, JobState::Failed)
3252 .unwrap();
3253
3254 let updated = store.get(&tx.transaction_id).unwrap().unwrap();
3255 assert_eq!(updated.status, JobState::Failed);
3256 }
3257
3258 #[test]
3259 fn update_status_accepts_running_to_rolled_back() {
3260 let dir = tempdir().unwrap();
3261 let store = test_store(dir.path().join("tx.db"));
3262 let tx = store.record(queued_transaction()).unwrap();
3263
3264 store
3265 .update_status(&tx.transaction_id, JobState::Running)
3266 .unwrap();
3267 store
3268 .update_status(&tx.transaction_id, JobState::RolledBack)
3269 .unwrap();
3270
3271 let updated = store.get(&tx.transaction_id).unwrap().unwrap();
3272 assert_eq!(updated.status, JobState::RolledBack);
3273 }
3274
3275 #[test]
3278 fn list_transactions_returns_empty_for_fresh_store() {
3279 let dir = tempdir().unwrap();
3280 let store = test_store(dir.path().join("tx.db"));
3281 let results = store.list_transactions(10, None, None, None).unwrap();
3282 assert!(results.is_empty());
3283 }
3284
3285 #[test]
3286 fn list_transactions_returns_all_records_ordered_by_newest_first() {
3287 let dir = tempdir().unwrap();
3288 let store = test_store(dir.path().join("tx.db"));
3289 store.record(queued_transaction()).unwrap();
3290
3291 let mut second = queued_transaction();
3292 second.action_name = "GetDiskUsage".to_string();
3293 second.risk_level = RiskLevel::Low;
3294 store.record(second).unwrap();
3295
3296 let results = store.list_transactions(10, None, None, None).unwrap();
3297 assert_eq!(results.len(), 2);
3298 assert_eq!(results[0].action_name, "GetDiskUsage");
3300 assert_eq!(results[1].action_name, "UpdateSystem");
3301 }
3302
3303 #[test]
3304 fn list_history_populates_created_at_and_risk_level() {
3305 let dir = tempdir().unwrap();
3306 let store = test_store(dir.path().join("tx.db"));
3307 store.record(queued_transaction()).unwrap();
3308
3309 let entries = store.list_history(10, None, None, None).unwrap();
3310 assert_eq!(entries.len(), 1);
3311 let entry = &entries[0];
3312 assert_eq!(entry.action_name, "UpdateSystem");
3313 assert_eq!(entry.risk_level, RiskLevel::High);
3314 assert_eq!(entry.status, JobState::Queued);
3315 assert!(
3316 !entry.created_at.is_empty(),
3317 "created_at must be populated from the stored row, not left blank"
3318 );
3319 }
3320
3321 #[test]
3322 fn list_history_applies_the_same_filters_as_list_transactions() {
3323 let dir = tempdir().unwrap();
3324 let store = test_store(dir.path().join("tx.db"));
3325 store.record(queued_transaction()).unwrap();
3326 let mut low = queued_transaction();
3327 low.action_name = "GetDiskUsage".to_string();
3328 low.risk_level = RiskLevel::Low;
3329 store.record(low).unwrap();
3330
3331 let only = store
3332 .list_history(10, None, Some("GetDiskUsage"), None)
3333 .unwrap();
3334 assert_eq!(only.len(), 1);
3335 assert_eq!(only[0].action_name, "GetDiskUsage");
3336 }
3337
3338 #[test]
3339 fn list_transactions_respects_limit() {
3340 let dir = tempdir().unwrap();
3341 let store = test_store(dir.path().join("tx.db"));
3342 for _ in 0..5 {
3343 store.record(queued_transaction()).unwrap();
3344 }
3345 let results = store.list_transactions(3, None, None, None).unwrap();
3346 assert_eq!(results.len(), 3);
3347 }
3348
3349 #[test]
3350 fn list_transactions_filters_by_status() {
3351 let dir = tempdir().unwrap();
3352 let store = test_store(dir.path().join("tx.db"));
3353 let tx = store.record(queued_transaction()).unwrap();
3354 store
3355 .update_status(&tx.transaction_id, JobState::Running)
3356 .unwrap();
3357 store
3358 .update_status(&tx.transaction_id, JobState::Succeeded)
3359 .unwrap();
3360
3361 store.record(queued_transaction()).unwrap();
3363
3364 let succeeded = store
3365 .list_transactions(10, Some("succeeded"), None, None)
3366 .unwrap();
3367 assert_eq!(succeeded.len(), 1);
3368 assert_eq!(succeeded[0].status, JobState::Succeeded);
3369
3370 let queued = store
3371 .list_transactions(10, Some("queued"), None, None)
3372 .unwrap();
3373 assert_eq!(queued.len(), 1);
3374 assert_eq!(queued[0].status, JobState::Queued);
3375 }
3376
3377 #[test]
3378 fn list_transactions_filters_by_action_name() {
3379 let dir = tempdir().unwrap();
3380 let store = test_store(dir.path().join("tx.db"));
3381 store.record(queued_transaction()).unwrap(); let mut disk = queued_transaction();
3384 disk.action_name = "GetDiskUsage".to_string();
3385 store.record(disk).unwrap();
3386
3387 let results = store
3388 .list_transactions(10, None, Some("GetDiskUsage"), None)
3389 .unwrap();
3390 assert_eq!(results.len(), 1);
3391 assert_eq!(results[0].action_name, "GetDiskUsage");
3392 }
3393
3394 #[test]
3395 fn list_transactions_filters_by_since_hours() {
3396 let dir = tempdir().unwrap();
3397 let store = test_store(dir.path().join("tx.db"));
3398
3399 let old = store.record(queued_transaction()).unwrap();
3401 let conn = store.connection().unwrap();
3402 conn.execute(
3403 "UPDATE transactions \
3404 SET created_at = strftime('%Y-%m-%dT%H:%M:%fZ', 'now', '-48 hours') \
3405 WHERE transaction_id = ?1",
3406 params![old.transaction_id],
3407 )
3408 .unwrap();
3409
3410 store.record(queued_transaction()).unwrap();
3412
3413 let results = store.list_transactions(10, None, None, Some(24)).unwrap();
3415 assert_eq!(results.len(), 1);
3416
3417 let results = store.list_transactions(10, None, None, Some(72)).unwrap();
3419 assert_eq!(results.len(), 2);
3420 }
3421
3422 #[test]
3423 fn list_transactions_rejects_invalid_status_filter() {
3424 let dir = tempdir().unwrap();
3425 let store = test_store(dir.path().join("tx.db"));
3426 store.record(queued_transaction()).unwrap();
3427 let result = store.list_transactions(10, Some("bogus"), None, None);
3428 assert!(result.is_err(), "invalid status filter should return error");
3429 }
3430
3431 #[test]
3439 fn record_emits_one_watermark_per_entry() {
3440 let sink = std::sync::Arc::new(std::sync::Mutex::new(Vec::new()));
3441 crate::audit_watermark::install_test_sink(std::sync::Arc::clone(&sink));
3442
3443 let dir = tempdir().unwrap();
3444 let store = test_store(dir.path().join("tx.db"));
3445 store.record(queued_transaction()).unwrap();
3446
3447 let calls = crate::audit_watermark::take_watermarks(&sink);
3448 assert_eq!(calls.len(), 1, "expected exactly 1 watermark per record()");
3449 }
3450
3451 #[test]
3453 fn record_previewed_emits_one_watermark() {
3454 let sink = std::sync::Arc::new(std::sync::Mutex::new(Vec::new()));
3455 crate::audit_watermark::install_test_sink(std::sync::Arc::clone(&sink));
3456
3457 let dir = tempdir().unwrap();
3458 let store = test_store(dir.path().join("tx.db"));
3459 let preview = PreviewEnvelope {
3460 summary: "Upgrade the system".to_string(),
3461 risk_level: RiskLevel::High,
3462 current_state: serde_json::Value::Null,
3463 proposed_change: serde_json::Value::Null,
3464 expected_side_effects: vec![],
3465 reboot_required: false,
3466 rollback_available: false,
3467 warnings: vec![],
3468 request_hash: sysknife_types::RequestHash::from("hash-abc".to_string()),
3469 };
3470 store
3471 .record_previewed(queued_transaction(), preview)
3472 .unwrap();
3473
3474 let calls = crate::audit_watermark::take_watermarks(&sink);
3475 assert_eq!(
3476 calls.len(),
3477 1,
3478 "expected exactly 1 watermark per record_previewed()"
3479 );
3480 }
3481
3482 #[test]
3484 fn watermark_seq_and_hash_match_chain_row() {
3485 let sink = std::sync::Arc::new(std::sync::Mutex::new(Vec::new()));
3486 crate::audit_watermark::install_test_sink(std::sync::Arc::clone(&sink));
3487
3488 let dir = tempdir().unwrap();
3489 let store = test_store(dir.path().join("tx.db"));
3490 store.record(queued_transaction()).unwrap();
3491
3492 let rows = store.fetch_chain_rows().unwrap();
3493 assert_eq!(rows.len(), 1);
3494 let row = &rows[0];
3495
3496 let calls = crate::audit_watermark::take_watermarks(&sink);
3497 assert_eq!(calls.len(), 1);
3498 assert_eq!(calls[0].seq, row.seq, "watermark seq must match chain row");
3499 assert_eq!(
3500 calls[0].chain_hash_hex, row.chain_hash,
3501 "watermark chain_hash_hex must match stored chain_hash"
3502 );
3503 }
3504
3505 #[test]
3507 fn multiple_records_produce_one_watermark_each() {
3508 let sink = std::sync::Arc::new(std::sync::Mutex::new(Vec::new()));
3509 crate::audit_watermark::install_test_sink(std::sync::Arc::clone(&sink));
3510
3511 let dir = tempdir().unwrap();
3512 let store = test_store(dir.path().join("tx.db"));
3513 for _ in 0..3 {
3514 store.record(queued_transaction()).unwrap();
3515 }
3516
3517 let calls = crate::audit_watermark::take_watermarks(&sink);
3518 assert_eq!(calls.len(), 3, "one watermark per record call");
3519 assert_eq!(calls[0].seq, 1);
3520 assert_eq!(calls[1].seq, 2);
3521 assert_eq!(calls[2].seq, 3);
3522 }
3523
3524 #[test]
3538 fn no_watermark_emitted_before_any_record() {
3539 let sink = std::sync::Arc::new(std::sync::Mutex::new(Vec::new()));
3540 crate::audit_watermark::install_test_sink(std::sync::Arc::clone(&sink));
3541
3542 let dir = tempdir().unwrap();
3543 let _store = test_store(dir.path().join("tx.db"));
3544
3545 let calls = crate::audit_watermark::take_watermarks(&sink);
3547 assert!(
3548 calls.is_empty(),
3549 "no watermark must be emitted without a record() call"
3550 );
3551 }
3552}