sequel_mcp/audit/
mod.rs

1//! Audit logging with redaction, hash chaining and epoch bookkeeping.
2
3pub mod db;
4pub mod redactor;
5pub mod retention;
6
7use crate::approval::outcomes::ApprovalOutcome;
8use crate::policy::model::{PolicyAction, SqlCategory};
9use rusqlite::params;
10use sha2::{Digest, Sha256};
11use std::sync::Arc;
12
13pub use db::AuditDb;
14
15#[derive(Debug, Clone)]
16pub struct AuditEntry {
17    pub request_id: String,
18    pub connection: String,
19    pub databases: Vec<String>,
20    pub category: SqlCategory,
21    pub ast_type: Option<String>,
22    pub sql: String,
23    pub decision: PolicyAction,
24    pub confirmed: bool,
25    pub outcome: ApprovalOutcome,
26    pub affected_rows: Option<i64>,
27    pub duration_ms: Option<i64>,
28    pub error: Option<String>,
29    pub backup_id: Option<i64>,
30    /// v2 linkage: approval scope + digest + policy revision.
31    pub approval_scope: Option<String>,
32    pub approval_digest: Option<[u8; 32]>,
33    pub policy_revision: Option<u64>,
34}
35
36#[derive(Default)]
37pub struct WriteOptions {
38    pub redact_sql_in_log: bool,
39    pub tamper_evident_chain: bool,
40}
41
42fn iso_now() -> String {
43    time::OffsetDateTime::now_utc()
44        .format(&time::format_description::well_known::Rfc3339)
45        .unwrap_or_else(|_| "1970-01-01T00:00:00Z".into())
46}
47
48fn hash_row(prev: Option<&[u8]>, payload: &str) -> [u8; 32] {
49    let mut h = Sha256::new();
50    if let Some(p) = prev {
51        h.update(p);
52    }
53    h.update(payload.as_bytes());
54    h.finalize().into()
55}
56
57/// Write one audit entry. When `tamper_evident_chain` is set, the row hash
58/// chains from the previous row's hash inside the same IMMEDIATE
59/// transaction, so concurrent writers cannot interleave.
60pub fn write_audit_entry(
61    db: &Arc<AuditDb>,
62    entry: &AuditEntry,
63    opts: &WriteOptions,
64) -> rusqlite::Result<i64> {
65    let ts = iso_now();
66    // Redaction is dialect-agnostic (tokenizer-level); MySQL dialect
67    // tokenization covers the shared syntax.
68    let dialect = crate::policy::classifier::Dialect::MySql;
69    let sql_redacted = redactor::redact_sql(&entry.sql, dialect);
70    let sql_raw = if opts.redact_sql_in_log {
71        sql_redacted.clone()
72    } else {
73        entry.sql.clone()
74    };
75    let databases = serde_json::to_string(&entry.databases).unwrap_or_else(|_| "[]".into());
76
77    db.with_tx(|c| {
78        let (prev_hash, row_hash): (Option<Vec<u8>>, Option<[u8; 32]>) = if opts.tamper_evident_chain {
79            let prev: Option<Vec<u8>> = c
80                .query_row(
81                    "SELECT row_hash FROM audit_log ORDER BY id DESC LIMIT 1",
82                    [],
83                    |r| r.get(0),
84                )
85                .unwrap_or(None);
86            let canonical = serde_json::json!({
87                "ts": ts,
88                "requestId": entry.request_id,
89                "connection": entry.connection,
90                "databases": databases,
91                "category": entry.category.as_str(),
92                "astType": entry.ast_type,
93                "sqlRedacted": sql_redacted,
94                "decision": entry.decision.as_str(),
95                "confirmed": entry.confirmed,
96                "outcome": entry.outcome.as_str(),
97                "affectedRows": entry.affected_rows,
98                "durationMs": entry.duration_ms,
99                "error": entry.error,
100                "backupId": entry.backup_id,
101            })
102            .to_string();
103            let rh = hash_row(prev.as_deref(), &canonical);
104            (prev, Some(rh))
105        } else {
106            (None, None)
107        };
108
109        c.execute(
110            "INSERT INTO audit_log
111               (ts, request_id, connection, databases, category, ast_type,
112                sql_raw, sql_redacted, decision, confirmed, outcome,
113                affected_rows, duration_ms, error_msg, backup_id,
114                prev_hash, row_hash, approval_scope, approval_digest, policy_revision)
115             VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9, ?10, ?11, ?12, ?13, ?14, ?15, ?16, ?17, ?18, ?19, ?20)",
116            params![
117                ts,
118                entry.request_id,
119                entry.connection,
120                databases,
121                entry.category.as_str(),
122                entry.ast_type,
123                sql_raw,
124                sql_redacted,
125                entry.decision.as_str(),
126                entry.confirmed,
127                entry.outcome.as_str(),
128                entry.affected_rows,
129                entry.duration_ms,
130                entry.error,
131                entry.backup_id,
132                prev_hash,
133                row_hash,
134                entry.approval_scope,
135                entry.approval_digest.map(|d| d.to_vec()),
136                entry.policy_revision.map(|r| r as i64),
137            ],
138        )?;
139        Ok(c.last_insert_rowid())
140    })
141}
142
143#[derive(Debug, Clone, serde::Serialize)]
144pub struct AuditRow {
145    pub id: i64,
146    pub ts: String,
147    pub request_id: String,
148    pub connection: String,
149    pub databases: Vec<String>,
150    pub category: String,
151    pub ast_type: Option<String>,
152    pub sql_redacted: String,
153    pub decision: String,
154    pub confirmed: bool,
155    pub outcome: String,
156    pub affected_rows: Option<i64>,
157    pub duration_ms: Option<i64>,
158    pub error_msg: Option<String>,
159    pub backup_id: Option<i64>,
160}
161
162#[derive(Debug, Clone, Default)]
163pub struct AuditSearchFilters {
164    pub since: Option<String>,
165    pub until: Option<String>,
166    pub connection: Option<String>,
167    pub category: Option<SqlCategory>,
168    pub outcome: Option<String>,
169    pub limit: u32,
170}
171
172pub fn search_audit_log(
173    db: &Arc<AuditDb>,
174    f: &AuditSearchFilters,
175) -> rusqlite::Result<Vec<AuditRow>> {
176    let mut conds: Vec<String> = Vec::new();
177    let mut params_vec: Vec<Box<dyn rusqlite::types::ToSql>> = Vec::new();
178    if let Some(s) = &f.since {
179        params_vec.push(Box::new(s.clone()));
180        conds.push(format!("ts >= ?{}", params_vec.len()));
181    }
182    if let Some(u) = &f.until {
183        params_vec.push(Box::new(u.clone()));
184        conds.push(format!("ts < ?{}", params_vec.len()));
185    }
186    if let Some(conn) = &f.connection {
187        params_vec.push(Box::new(conn.clone()));
188        conds.push(format!("connection = ?{}", params_vec.len()));
189    }
190    if let Some(cat) = &f.category {
191        params_vec.push(Box::new(cat.as_str().to_string()));
192        conds.push(format!("category = ?{}", params_vec.len()));
193    }
194    if let Some(out) = &f.outcome {
195        params_vec.push(Box::new(out.clone()));
196        conds.push(format!("outcome = ?{}", params_vec.len()));
197    }
198    let where_clause = if conds.is_empty() {
199        String::new()
200    } else {
201        format!("WHERE {}", conds.join(" AND "))
202    };
203    let limit = f.limit.min(5000);
204    params_vec.push(Box::new(limit));
205    let limit_idx = params_vec.len();
206
207    let sql = format!(
208        "SELECT id, ts, request_id, connection, databases, category, ast_type,
209                sql_redacted, decision, confirmed, outcome, affected_rows,
210                duration_ms, error_msg, backup_id
211           FROM audit_log {where_clause} ORDER BY id DESC LIMIT ?{limit_idx}"
212    );
213    let params_ref: Vec<&dyn rusqlite::types::ToSql> =
214        params_vec.iter().map(|p| p.as_ref()).collect();
215    db.with(|c| {
216        let mut stmt = c.prepare(&sql)?;
217        let rows = stmt.query_map(params_ref.as_slice(), |r| {
218            let databases_raw: String = r.get(4)?;
219            Ok(AuditRow {
220                id: r.get(0)?,
221                ts: r.get(1)?,
222                request_id: r.get(2)?,
223                connection: r.get(3)?,
224                databases: serde_json::from_str(&databases_raw).unwrap_or_default(),
225                category: r.get(5)?,
226                ast_type: r.get(6)?,
227                sql_redacted: r.get(7)?,
228                decision: r.get(8)?,
229                confirmed: r.get::<_, i64>(9)? != 0,
230                outcome: r.get(10)?,
231                affected_rows: r.get(11)?,
232                duration_ms: r.get(12)?,
233                error_msg: r.get(13)?,
234                backup_id: r.get(14)?,
235            })
236        })?;
237        rows.collect()
238    })
239}
240
241/// Verify the tamper-evident chain. Retention deletions create chain
242/// epochs; verification restarts at each epoch boundary row.
243#[derive(Debug)]
244pub struct ChainVerification {
245    pub ok: bool,
246    pub rows_checked: u64,
247    pub broken_at: Option<i64>,
248    pub epoch: u64,
249}
250
251pub fn verify_chain(db: &Arc<AuditDb>) -> rusqlite::Result<ChainVerification> {
252    #[derive(Debug)]
253    struct ChainRow {
254        id: i64,
255        canonical: String,
256        prev_hash: Option<Vec<u8>>,
257        row_hash: Option<Vec<u8>>,
258    }
259    #[allow(clippy::type_complexity)]
260    fn get_chain(r: &rusqlite::Row<'_>) -> rusqlite::Result<ChainRow> {
261        let confirmed: i64 = r.get(9)?;
262        let canonical = serde_json::json!({
263            "ts": r.get::<_, String>(1)?,
264            "requestId": r.get::<_, String>(2)?,
265            "connection": r.get::<_, String>(3)?,
266            "databases": r.get::<_, String>(4)?,
267            "category": r.get::<_, String>(5)?,
268            "astType": r.get::<_, Option<String>>(6)?,
269            "sqlRedacted": r.get::<_, String>(7)?,
270            "decision": r.get::<_, String>(8)?,
271            "confirmed": confirmed != 0,
272            "outcome": r.get::<_, String>(10)?,
273            "affectedRows": r.get::<_, Option<i64>>(11)?,
274            "durationMs": r.get::<_, Option<i64>>(12)?,
275            "error": r.get::<_, Option<String>>(13)?,
276            "backupId": r.get::<_, Option<i64>>(14)?,
277        })
278        .to_string();
279        Ok(ChainRow {
280            id: r.get(0)?,
281            canonical,
282            prev_hash: r.get(15)?,
283            row_hash: r.get(16)?,
284        })
285    }
286
287    let mut checked: u64 = 0;
288    let mut epoch: u64 = 0;
289    let mut prev: Option<Vec<u8>> = None;
290    let mut broken_at: Option<i64> = None;
291
292    db.with(|c| {
293        let mut stmt = c.prepare("SELECT id, ts, request_id, connection, databases, category, ast_type, sql_redacted, decision, confirmed, outcome, affected_rows, duration_ms, error_msg, backup_id, prev_hash, row_hash FROM audit_log ORDER BY id ASC")?;
294        let rows = stmt.query_map([], get_chain)?;
295        for row in rows {
296            let ChainRow { id, canonical, prev_hash, row_hash } = row?;
297            let Some(rh) = row_hash else {
298                // Chaining was off for this row: restart from here.
299                if prev_hash.is_none() {
300                    epoch += 1;
301                    prev = None;
302                }
303                continue;
304            };
305            // Content integrity: the row hash must cover its stored prev and
306            // the canonical payload. This detects any content tampering.
307            let expected = hash_row(prev_hash.as_deref(), &canonical);
308            if expected != rh.as_slice() {
309                broken_at = Some(id);
310                return Ok(());
311            }
312            // Chain linkage: when a predecessor is known in this scan, the
313            // stored prev must be that predecessor's hash. A dangling prev
314            // at a chain start is an epoch boundary (retention deleted the
315            // predecessor), recorded rather than reported as tampering.
316            match (&prev, &prev_hash) {
317                (Some(last), Some(stored)) if last != stored => {
318                    broken_at = Some(id);
319                    return Ok(());
320                }
321                (None, Some(_)) => {
322                    epoch += 1;
323                }
324                _ => {}
325            }
326            prev = Some(rh);
327            checked += 1;
328        }
329        Ok(())
330    })?;
331
332    Ok(ChainVerification {
333        ok: broken_at.is_none(),
334        rows_checked: checked,
335        broken_at,
336        epoch,
337    })
338}
339
340/// Record a chain epoch after retention deletes chained rows, so
341/// verification stays meaningful instead of silently seeing a broken link.
342pub fn write_chain_epoch(db: &Arc<AuditDb>, deleted: u64) -> rusqlite::Result<()> {
343    db.with_tx(|c| {
344        c.execute(
345            "INSERT INTO meta (key, value) VALUES ('chain_epoch', '0')
346             ON CONFLICT(key) DO UPDATE SET value = CAST(CAST(value AS INTEGER) + ?1 AS TEXT)",
347            params![deleted as i64],
348        )?;
349        Ok(())
350    })
351}
352
353#[cfg(test)]
354mod tests {
355    use super::*;
356
357    fn db() -> (tempfile::TempDir, Arc<AuditDb>) {
358        let dir = tempfile::tempdir().unwrap();
359        let d = Arc::new(AuditDb::at_path(&dir.path().join("a.sqlite")).unwrap());
360        (dir, d)
361    }
362
363    fn entry(sql: &str, outcome: ApprovalOutcome) -> AuditEntry {
364        AuditEntry {
365            request_id: uuid::Uuid::new_v4().to_string(),
366            connection: "c1".into(),
367            databases: vec!["app".into()],
368            category: SqlCategory::Write,
369            ast_type: Some("update".into()),
370            sql: sql.into(),
371            decision: PolicyAction::Confirm,
372            confirmed: false,
373            outcome,
374            affected_rows: None,
375            duration_ms: None,
376            error: None,
377            backup_id: None,
378            approval_scope: None,
379            approval_digest: None,
380            policy_revision: None,
381        }
382    }
383
384    #[test]
385    fn writes_and_reads_back_redacted() {
386        let (_dir, db) = db();
387        let id = write_audit_entry(
388            &db,
389            &entry(
390                "UPDATE users SET name = 'secret' WHERE id = 1",
391                ApprovalOutcome::Approved,
392            ),
393            &WriteOptions::default(),
394        )
395        .unwrap();
396        assert!(id > 0);
397        let rows = search_audit_log(
398            &db,
399            &AuditSearchFilters {
400                limit: 10,
401                ..Default::default()
402            },
403        )
404        .unwrap();
405        assert_eq!(rows.len(), 1);
406        assert_eq!(rows[0].id, id);
407        assert!(!rows[0].sql_redacted.contains("secret"));
408        assert_eq!(rows[0].outcome, "approved");
409    }
410
411    #[test]
412    fn raw_sql_hidden_when_redact_sql_in_log() {
413        let (_dir, db) = db();
414        write_audit_entry(
415            &db,
416            &entry(
417                "UPDATE users SET name = 'secret'",
418                ApprovalOutcome::Approved,
419            ),
420            &WriteOptions {
421                redact_sql_in_log: true,
422                ..Default::default()
423            },
424        )
425        .unwrap();
426        let raw: String = db
427            .with(|c| c.query_row("SELECT sql_raw FROM audit_log LIMIT 1", [], |r| r.get(0)))
428            .unwrap();
429        assert!(
430            !raw.contains("secret"),
431            "raw must be redacted when configured"
432        );
433    }
434
435    #[test]
436    fn chain_verifies_and_detects_tampering() {
437        let (_dir, db) = db();
438        for i in 0..5 {
439            write_audit_entry(
440                &db,
441                &entry(&format!("UPDATE t SET n = {i}"), ApprovalOutcome::Approved),
442                &WriteOptions {
443                    tamper_evident_chain: true,
444                    ..Default::default()
445                },
446            )
447            .unwrap();
448        }
449        let v = verify_chain(&db).unwrap();
450        assert!(v.ok);
451        assert_eq!(v.rows_checked, 5);
452
453        // Tamper directly.
454        db.with(|c| {
455            c.execute(
456                "UPDATE audit_log SET sql_redacted = 'tampered' WHERE id = 3",
457                [],
458            )
459        })
460        .unwrap();
461        let v2 = verify_chain(&db).unwrap();
462        assert!(!v2.ok);
463        assert_eq!(v2.broken_at, Some(3));
464    }
465
466    #[test]
467    fn epoch_rows_restart_chain() {
468        let (_dir, db) = db();
469        write_audit_entry(
470            &db,
471            &entry("UPDATE t SET n = 1", ApprovalOutcome::Approved),
472            &WriteOptions {
473                tamper_evident_chain: true,
474                ..Default::default()
475            },
476        )
477        .unwrap();
478        // Simulate retention deletion: rows 2.. with chain on but no
479        // predecessor.
480        write_audit_entry(
481            &db,
482            &entry("UPDATE t SET n = 2", ApprovalOutcome::Approved),
483            &WriteOptions {
484                tamper_evident_chain: true,
485                ..Default::default()
486            },
487        )
488        .unwrap();
489        db.with(|c| c.execute("DELETE FROM audit_log WHERE id = 1", []))
490            .unwrap();
491        write_chain_epoch(&db, 1).unwrap();
492        let v = verify_chain(&db).unwrap();
493        assert!(v.ok, "epoch boundary must not read as tampering: {v:?}");
494        assert!(v.epoch >= 1);
495    }
496}
sequel_mcp/audit/mod.rs

sequel_mcp/audit/
mod.rs
