sequel_mcp/app/
gate.rs

1//! The operation gate: the single execution authority shared by MCP tools,
2//! CLI commands and (future) GUI actions.
3//!
4//! Pipeline: resolve connection → classify → two-layer policy → Touch ID
5//! (fail closed) → approval (elicitation/IPC; unavailable ⇒ fail closed) →
6//! revalidate → execute → audit.
7
8use crate::approval::digest::{OperationDigestInput, canonical_sql, operation_digest};
9use crate::approval::{ApprovalEngine, ConfirmOutcome, GrantChoice, SessionGrantKey};
10use crate::audit::{AuditDb, AuditEntry, WriteOptions};
11use crate::config::{Config, ConfigStore, Connection};
12use crate::policy::classifier::{self, ClassifiedStatement, Dialect};
13use crate::policy::model::{PolicyAction, SqlCategory, TableId};
14use crate::policy::resolver::{self, Resolution};
15use crate::sql::sqlite::{self, SqliteExecuteParams};
16use crate::vault::touchid::SessionAuthenticator;
17use serde_json::json;
18use std::sync::Arc;
19use thiserror::Error;
20use uuid::Uuid;
21
22#[derive(Debug, Error)]
23pub enum GateError {
24    #[error("{0}")]
25    NoConnection(String),
26    #[error("cannot run statement: {0}")]
27    Classify(String),
28    #[error(
29        "query tool only accepts read statements (got {0}). Use the \"execute\" tool for non-read statements."
30    )]
31    NotReadOnly(SqlCategory),
32    #[error("denied by policy: {0}")]
33    Denied(String),
34    #[error("Touch ID authentication failed")]
35    TouchIdFailed,
36    #[error("confirmation for {0} statement was declined")]
37    Declined(SqlCategory),
38    #[error(
39        "confirmation required for {0} statement, but no prompt could be shown: {1}. Statement not executed - nothing was changed. This is not a refusal: the prompt could not be delivered."
40    )]
41    Unavailable(SqlCategory, String),
42    #[error("approval expired before execution")]
43    Expired,
44    #[error("SQL execution failed: {0}")]
45    Execution(String),
46    #[error(
47        "no password stored for connection {0:?}. Run add_connection or import_from_sequel_ace first."
48    )]
49    NoPassword(String),
50    #[error("MySQL execution is not wired in this build: {0}")]
51    MySql(String),
52    #[error("config error: {0}")]
53    Config(#[from] crate::config::ConfigError),
54}
55
56/// Where approvals are sought.
57pub trait ApprovalSink: Send + Sync {
58    fn confirm(&self, request: ApprovalRequest) -> ConfirmOutcome;
59}
60
61#[derive(Debug, Clone)]
62pub struct ApprovalRequest {
63    pub category: SqlCategory,
64    pub statement_snippet: String,
65    pub connection_name: String,
66    pub database: Option<String>,
67    pub tables: Vec<TableId>,
68}
69
70/// Sink used when no interactive surface exists. Fails closed and reports
71/// the real reason — never fabricates a refusal.
72#[derive(Debug, Default, Clone, Copy)]
73pub struct UnavailableSink;
74
75impl ApprovalSink for UnavailableSink {
76    fn confirm(&self, _request: ApprovalRequest) -> ConfirmOutcome {
77        ConfirmOutcome::Unavailable {
78            reason: "no approval channel is available in this context".into(),
79        }
80    }
81}
82
83pub struct GateDeps {
84    pub config: Arc<ConfigStore>,
85    pub audit: Arc<AuditDb>,
86    pub approvals: Arc<ApprovalEngine>,
87    pub auth: Arc<SessionAuthenticator>,
88    pub sink: Box<dyn ApprovalSink>,
89    #[allow(dead_code)]
90    pub secrets: Arc<dyn crate::vault::keychain::SecretStore>,
91}
92
93impl GateDeps {
94    pub fn with_sink(sink: Box<dyn ApprovalSink>) -> Self {
95        Self::with_sink_and_approvals(sink, Arc::new(ApprovalEngine::new()))
96    }
97
98    /// Server entry point: the approval engine is PROCESS-SHARED so
99    /// "Allow for session" grants survive across tool calls (a fresh
100    /// engine per call silently voided the documented session scope).
101    pub fn with_sink_and_approvals(
102        sink: Box<dyn ApprovalSink>,
103        approvals: Arc<ApprovalEngine>,
104    ) -> Self {
105        Self {
106            config: Arc::new(ConfigStore::new()),
107            audit: AuditDb::shared(),
108            approvals,
109            auth: Arc::new(SessionAuthenticator::new(
110                crate::vault::touchid::system_touch_id(),
111            )),
112            sink,
113            secrets: crate::vault::keychain::default_store(),
114        }
115    }
116}
117
118pub struct RunSqlArgs {
119    pub connection: Option<String>,
120    pub sql: String,
121    pub database: Option<String>,
122    /// Plan-time approved DDL target set carried from an MRTR approval
123    /// (plan→retry gap enforcement); `None` for single-shot execution.
124    pub expected_ddl_targets: Option<Vec<(String, String)>>,
125}
126
127#[derive(Debug)]
128pub struct RunOutcome {
129    pub connection: String,
130    pub category: SqlCategory,
131    pub ast_type: String,
132    pub resolution: Resolution,
133    pub rows: Vec<serde_json::Value>,
134    pub fields: Vec<String>,
135    pub affected_rows: u64,
136    pub truncated: bool,
137    pub duration_ms: u64,
138    pub backup_id: Option<i64>,
139    pub backup_row_count: u64,
140    pub request_id: String,
141    /// DDL absent-target no-op (IF EXISTS over missing tables): nothing
142    /// was sent to the server (D4).
143    pub ddl_no_op: bool,
144    /// Targets absent under IF EXISTS in a Mixed set (recorded, never
145    /// suppressed); empty unless a Mixed DROP occurred.
146    pub ddl_absent_targets: Vec<String>,
147    /// Targets actually named by the rewritten Mixed DROP statement
148    /// (the preflight-approved existing subset); empty otherwise.
149    pub ddl_executed_targets: Vec<String>,
150    /// Protection-model warnings (nontransactional DDL snapshot).
151    pub warnings: Vec<&'static str>,
152}
153
154fn dialect_for(conn: &Connection) -> Dialect {
155    if conn.is_mysql() {
156        Dialect::MySql
157    } else {
158        Dialect::SQLite
159    }
160}
161
162fn resolve_table_ids(classified: &ClassifiedStatement, fallback: Option<&str>) -> Vec<TableId> {
163    let mut out: Vec<TableId> = Vec::new();
164    for t in classified
165        .read_tables
166        .iter()
167        .chain(classified.mutated_tables.iter())
168    {
169        let db = t.database.clone().or_else(|| fallback.map(str::to_string));
170        if let Some(db) = db {
171            let id = TableId {
172                database: db,
173                table: t.table.clone(),
174            };
175            if !out.contains(&id) {
176                out.push(id);
177            }
178        }
179    }
180    out
181}
182
183/// The single statement pipeline. `expect_read_only` rejects non-read
184/// categories before any policy work (legacy `query` behaviour).
185pub fn run_sql(
186    deps: &GateDeps,
187    args: &RunSqlArgs,
188    expect_read_only: bool,
189) -> Result<RunOutcome, GateError> {
190    let cfg: Config = deps.config.load()?;
191    let conn = cfg
192        .resolve(args.connection.as_deref())
193        .cloned()
194        .ok_or_else(|| {
195            GateError::NoConnection(no_connection_message(args.connection.as_deref()))
196        })?;
197
198    let dialect = dialect_for(&conn);
199    let classified = classifier::classify_statement(&args.sql, dialect)
200        .map_err(|e| GateError::Classify(e.message()))?;
201
202    if expect_read_only && classified.category != SqlCategory::Read {
203        return Err(GateError::NotReadOnly(classified.category));
204    }
205
206    let fallback = args
207        .database
208        .clone()
209        .or_else(|| conn.database().map(str::to_string));
210    let resolution = resolver::resolve(&conn, &classified, fallback.as_deref());
211
212    let request_id = Uuid::new_v4().to_string();
213    let databases_for_log: Vec<String> = if !classified.target_databases.is_empty() {
214        classified.target_databases.clone()
215    } else {
216        fallback.clone().into_iter().collect()
217    };
218
219    let write_opts = WriteOptions {
220        redact_sql_in_log: cfg.retention.redact_sql_in_log,
221        tamper_evident_chain: cfg.retention.tamper_evident_chain,
222    };
223
224    // Policy decision + approval.
225    let mut approval_scope: Option<String> = None;
226    let mut approval_digest: Option<[u8; 32]> = None;
227    match resolution.action {
228        PolicyAction::Deny => {
229            if let Err(e) = audit(
230                deps,
231                &request_id,
232                &conn,
233                &databases_for_log,
234                &classified,
235                &args.sql,
236                &resolution,
237                false,
238                crate::approval::outcomes::ApprovalOutcome::Denied,
239                None,
240                None,
241                None,
242                None,
243                None,
244                None,
245                None,
246                &write_opts,
247            ) {
248                eprintln!("sequel-mcp: audit write failed for a denied statement: {e}");
249            }
250            let hint = resolution
251                .contributions
252                .iter()
253                .map(|c| format!("{}.{}", c.table.database, c.table.table))
254                .next()
255                .map(|t| format!(" ({t})"))
256                .unwrap_or_default();
257            return Err(GateError::Denied(format!(
258                "{} statements not allowed on {:?}{}.",
259                classified.category.as_str(),
260                conn.name(),
261                hint
262            )));
263        }
264        PolicyAction::Allow => {}
265        PolicyAction::Confirm => {
266            // Touch ID first (fail closed).
267            if resolution.effective.require_touch_id {
268                let ok = deps.auth.ensure_authenticated(&format!(
269                    "Authenticate to run {} on {}",
270                    classified.category.as_str(),
271                    conn.name()
272                ));
273                if !ok {
274                    if let Err(audit_err) = audit(
275                        deps,
276                        &request_id,
277                        &conn,
278                        &databases_for_log,
279                        &classified,
280                        &args.sql,
281                        &resolution,
282                        false,
283                        crate::approval::outcomes::ApprovalOutcome::Denied,
284                        None,
285                        None,
286                        None,
287                        None,
288                        None,
289                        None,
290                        None,
291                        &write_opts,
292                    ) {
293                        eprintln!(
294                            "sequel-mcp: audit write failed for a touch-id-denied statement: {audit_err}"
295                        );
296                    }
297                    return Err(GateError::TouchIdFailed);
298                }
299            }
300
301            let tables = resolve_table_ids(&classified, fallback.as_deref());
302            // Narrow session grant first (exact table set + category).
303            let session_key = SessionGrantKey {
304                connection: conn.name().to_string(),
305                category: classified.category,
306                tables: tables.clone(),
307            };
308            if deps.approvals.session_covers(&session_key) {
309                approval_scope = Some("session".to_string());
310            } else {
311                let snippet = if args.sql.len() > 800 {
312                    format!("{}…", &args.sql[..800])
313                } else {
314                    args.sql.clone()
315                };
316                let outcome = deps.sink.confirm(ApprovalRequest {
317                    category: classified.category,
318                    statement_snippet: snippet,
319                    connection_name: conn.name().to_string(),
320                    database: databases_for_log.first().cloned(),
321                    tables: tables.clone(),
322                });
323                match outcome {
324                    ConfirmOutcome::Unavailable { reason } => {
325                        if let Err(e) = audit(
326                            deps,
327                            &request_id,
328                            &conn,
329                            &databases_for_log,
330                            &classified,
331                            &args.sql,
332                            &resolution,
333                            false,
334                            crate::approval::outcomes::ApprovalOutcome::Unavailable,
335                            None,
336                            None,
337                            None,
338                            Some(&reason),
339                            None,
340                            None,
341                            None,
342                            &write_opts,
343                        ) {
344                            eprintln!("sequel-mcp: audit write failed: {e}");
345                        }
346                        return Err(GateError::Unavailable(classified.category, reason));
347                    }
348                    ConfirmOutcome::Chosen(GrantChoice::Decline) => {
349                        if let Err(e) = audit(
350                            deps,
351                            &request_id,
352                            &conn,
353                            &databases_for_log,
354                            &classified,
355                            &args.sql,
356                            &resolution,
357                            false,
358                            crate::approval::outcomes::ApprovalOutcome::Declined,
359                            None,
360                            None,
361                            None,
362                            None,
363                            None,
364                            None,
365                            None,
366                            &write_opts,
367                        ) {
368                            eprintln!("sequel-mcp: audit write failed: {e}");
369                        }
370                        return Err(GateError::Declined(classified.category));
371                    }
372                    ConfirmOutcome::Chosen(GrantChoice::Session) => {
373                        deps.approvals.grant_session(session_key);
374                        approval_scope = Some("session".to_string());
375                    }
376                    ConfirmOutcome::Chosen(GrantChoice::Once) => {
377                        let canon = canonical_sql(&args.sql, dialect);
378                        let nonce = crate::approval::random_nonce_32();
379                        let digest = operation_digest(&OperationDigestInput {
380                            connection: conn.name(),
381                            driver: if conn.is_mysql() { "mysql" } else { "sqlite" },
382                            database: fallback.as_deref(),
383                            read_tables: &classified.read_tables,
384                            mutated_tables: &classified.mutated_tables,
385                            category: classified.category,
386                            canonical_sql: &canon,
387                            parameter_hashes: &[],
388                            backup_plan_identity: None,
389                            policy_revision: cfg.revision,
390                            metadata_revision: 0,
391                            nonce: *nonce,
392                        });
393                        deps.approvals.grant_once(digest);
394                        approval_digest = Some(digest);
395                        approval_scope = Some("once".to_string());
396                    }
397                }
398            }
399        }
400    }
401
402    // Revalidate: the policy may have changed since resolution began.
403    let cfg2: Config = deps.config.load()?;
404    if cfg2.revision != cfg.revision {
405        let conn2 = cfg2
406            .resolve(args.connection.as_deref())
407            .cloned()
408            .ok_or_else(|| {
409                GateError::NoConnection(no_connection_message(args.connection.as_deref()))
410            })?;
411        let resolution2 = resolver::resolve(&conn2, &classified, fallback.as_deref());
412        if resolution2.action != resolution.action {
413            if let Err(e) = audit(
414                deps,
415                &request_id,
416                &conn,
417                &databases_for_log,
418                &classified,
419                &args.sql,
420                &resolution,
421                false,
422                crate::approval::outcomes::ApprovalOutcome::Denied,
423                approval_scope.as_deref(),
424                approval_digest,
425                Some(cfg2.revision),
426                None,
427                None,
428                None,
429                None,
430                &write_opts,
431            ) {
432                eprintln!("sequel-mcp: audit write failed: {e}");
433            }
434            return Err(GateError::Denied(
435                "policy changed during approval; re-run the statement".into(),
436            ));
437        }
438    }
439
440    // Consume the one-time approval atomically (expiry fails closed).
441    if let Some(digest) = approval_digest
442        && deps.approvals.consume_once(digest).is_err()
443    {
444        return Err(GateError::Expired);
445    }
446
447    // Execute.
448    let started = std::time::Instant::now();
449    let exec_result: Result<crate::sql::sqlite::ExecuteResult, String> = match &conn {
450        Connection::Sqlite(sc) => sqlite::execute_sqlite_statement(SqliteExecuteParams {
451            connection: sc,
452            sql: &args.sql,
453            classified: &classified,
454            policy: &resolution.effective,
455            database: args.database.as_deref(),
456            audit: Some(deps.audit.clone()),
457        })
458        .map_err(|e| e.to_string()),
459        Connection::Mysql(mc) => {
460            // MySQL is async; the gate runs on the blocking pool, so
461            // block_on the runtime handle from here.
462            let password = match deps.secrets.get_password(&mc.name, &mc.user) {
463                Ok(p) => p,
464                Err(_) => {
465                    if let Err(e) = audit(
466                        deps,
467                        &request_id,
468                        &conn,
469                        &databases_for_log,
470                        &classified,
471                        &args.sql,
472                        &resolution,
473                        false,
474                        crate::approval::outcomes::ApprovalOutcome::Denied,
475                        approval_scope.as_deref(),
476                        approval_digest,
477                        Some(cfg.revision),
478                        None,
479                        None,
480                        None,
481                        None,
482                        &write_opts,
483                    ) {
484                        eprintln!("sequel-mcp: audit write failed: {e}");
485                    }
486                    return Err(GateError::NoPassword(mc.name.clone()));
487                }
488            };
489            let db = args.database.clone();
490            let audit_db = deps.audit.clone();
491            let revision = cfg.revision;
492            let sql = args.sql.clone();
493            let expected_ddl = args.expected_ddl_targets.clone();
494            // SSH direct transport: establish (or reuse) the tunnel and
495            // hand the executor the lease (loopback endpoint + transport
496            // generation). The MySQL pool keys on the generation, so
497            // tunneled pools stay separate from direct ones and can never
498            // be spliced onto a later transport that reuses the port.
499            let tunnel_endpoint: Option<crate::sql::ssh::TunnelLease> = match &mc.ssh {
500                Some(ssh) => {
501                    // The secret is the SSH password under password auth
502                    // (required) and the private-key passphrase under key
503                    // auth (optional — unencrypted keys need none).
504                    let ssh_password = match deps
505                        .secrets
506                        .get_password(&format!("{}::ssh", mc.name), &ssh.user)
507                    {
508                        Ok(p) => Some(p),
509                        Err(_) if ssh.auth_method == crate::config::SshAuthMethod::Key => None,
510                        Err(_) => {
511                            if let Err(e) = audit(
512                                deps,
513                                &request_id,
514                                &conn,
515                                &databases_for_log,
516                                &classified,
517                                &args.sql,
518                                &resolution,
519                                false,
520                                crate::approval::outcomes::ApprovalOutcome::Denied,
521                                approval_scope.as_deref(),
522                                approval_digest,
523                                Some(cfg.revision),
524                                None,
525                                None,
526                                None,
527                                None,
528                                &write_opts,
529                            ) {
530                                eprintln!("sequel-mcp: audit write failed: {e}");
531                            }
532                            return Err(GateError::NoPassword(format!("{}::ssh", mc.name)));
533                        }
534                    };
535                    let mc_name = mc.name.clone();
536                    let ssh_cfg = ssh.clone();
537                    let mysql_host = mc.host.clone();
538                    let mysql_port = mc.port;
539                    let res = tokio::task::block_in_place(|| {
540                        tokio::runtime::Handle::current().block_on(async {
541                            crate::sql::ssh::tunnel_endpoint(
542                                &mc_name,
543                                &ssh_cfg,
544                                ssh_password.as_ref().map(|p| p.as_str()),
545                                &mysql_host,
546                                mysql_port,
547                                revision,
548                            )
549                            .await
550                        })
551                    });
552                    match res {
553                        Ok(lease) => Some(lease),
554                        Err(e) => {
555                            let msg = format!("ssh tunnel: {e}");
556                            if let Err(audit_err) = audit(
557                                deps,
558                                &request_id,
559                                &conn,
560                                &databases_for_log,
561                                &classified,
562                                &args.sql,
563                                &resolution,
564                                false,
565                                crate::approval::outcomes::ApprovalOutcome::ExecutionError,
566                                approval_scope.as_deref(),
567                                approval_digest,
568                                Some(cfg.revision),
569                                Some(&msg),
570                                None,
571                                None,
572                                None,
573                                &write_opts,
574                            ) {
575                                eprintln!("sequel-mcp: audit write failed: {audit_err}");
576                            }
577                            return Err(GateError::Execution(msg));
578                        }
579                    }
580                }
581                None => None,
582            };
583            let res = tokio::task::block_in_place(|| {
584                tokio::runtime::Handle::current().block_on(async {
585                    crate::sql::mysql::execute_mysql_statement(
586                        crate::sql::mysql::MySqlExecuteParams {
587                            connection: mc,
588                            request_id: request_id.clone(),
589                            databases_for_log: databases_for_log.clone(),
590                            password,
591                            sql: &sql,
592                            classified: &classified,
593                            policy: &resolution.effective,
594                            database: db.as_deref(),
595                            audit: Some(audit_db),
596                            revision,
597                            tunnel_endpoint,
598                            expected_ddl_targets: expected_ddl,
599                        },
600                    )
601                    .await
602                })
603            });
604            res.map(|r| crate::sql::sqlite::ExecuteResult {
605                journal_id: r.journal_id,
606                ddl_no_op: r.ddl_no_op,
607                ddl_absent_targets: r.ddl_absent_targets,
608                ddl_executed_targets: r.ddl_executed_targets,
609                warnings: r.warnings,
610                rows: r.rows,
611                fields: r.fields,
612                affected_rows: r.affected_rows,
613                truncated: r.truncated,
614                duration_ms: r.duration_ms,
615                backup_id: r.backup_id,
616                backup_row_count: r.backup_row_count,
617            })
618            .map_err(|e| e.to_string())
619        }
620    };
621
622    match exec_result {
623        Ok(r) => {
624            let audit_result = audit(
625                deps,
626                &request_id,
627                &conn,
628                &databases_for_log,
629                &classified,
630                &args.sql,
631                &resolution,
632                resolution.action == PolicyAction::Confirm,
633                crate::approval::outcomes::ApprovalOutcome::Approved,
634                approval_scope.as_deref(),
635                approval_digest,
636                Some(cfg.revision),
637                None,
638                Some(r.affected_rows),
639                Some(r.duration_ms.max(started.elapsed().as_millis() as u64)),
640                r.backup_id,
641                &write_opts,
642            );
643            // Review blocker: a committed mutation with a FAILED audit
644            // write must never be reported as a clean success — and the
645            // journal must NOT close out over a missing row. Report an
646            // Ok outcome (the mutation happened; an error invites a
647            // client retry and a double execution) with a loud warning.
648            let audit_ok = audit_result.is_ok();
649            let mut warnings = r.warnings;
650            if let Err(e) = &audit_result {
651                eprintln!(
652                    "sequel-mcp: AUDIT WRITE FAILED after a committed {cat} statement on {conn}: {e}. The audit chain is incomplete for request {rid}.",
653                    cat = classified.category,
654                    conn = conn.name(),
655                    rid = request_id,
656                );
657                warnings.push(Box::leak(
658                    format!(
659                        "audit-write-failed: the statement committed but its audit entry could not be persisted ({e}); the audit chain is incomplete"
660                    )
661                    .into_boxed_str(),
662                ));
663            }
664            // D3: only close the journal (mutation_committed ->
665            // audit_finalized) when the audit row is actually durable —
666            // and LINK the row (review P1: link_audit never fired
667            // before, so the journal could not point at its audit entry).
668            if let (true, Some(jid)) = (audit_ok, r.journal_id) {
669                let j = crate::backup::journal::Journal::from_id(&deps.audit, jid);
670                let _ = j.transition(crate::backup::journal::JournalState::AuditFinalized, None);
671                if let Ok(audit_row) = audit_result {
672                    let _ = j.link_audit(audit_row);
673                }
674            }
675            Ok(RunOutcome {
676                connection: conn.name().to_string(),
677                category: classified.category,
678                ast_type: classified.ast_type.to_string(),
679                resolution,
680                rows: r.rows,
681                fields: r.fields,
682                affected_rows: r.affected_rows,
683                truncated: r.truncated,
684                duration_ms: r.duration_ms.max(started.elapsed().as_millis() as u64),
685                backup_id: r.backup_id,
686                backup_row_count: r.backup_row_count,
687                ddl_no_op: r.ddl_no_op,
688                ddl_absent_targets: r.ddl_absent_targets,
689                ddl_executed_targets: r.ddl_executed_targets,
690                warnings,
691                request_id,
692            })
693        }
694        Err(e) => {
695            if let Err(audit_err) = audit(
696                deps,
697                &request_id,
698                &conn,
699                &databases_for_log,
700                &classified,
701                &args.sql,
702                &resolution,
703                resolution.action == PolicyAction::Confirm,
704                crate::approval::outcomes::ApprovalOutcome::ExecutionError,
705                approval_scope.as_deref(),
706                approval_digest,
707                Some(cfg.revision),
708                Some(&e),
709                None,
710                Some(started.elapsed().as_millis() as u64),
711                None,
712                &write_opts,
713            ) {
714                eprintln!(
715                    "sequel-mcp: audit write also failed for the errored statement (request {rid}): {audit_err}",
716                    rid = request_id,
717                );
718            }
719            Err(GateError::Execution(e))
720        }
721    }
722}
723
724#[allow(clippy::too_many_arguments)]
725fn audit(
726    deps: &GateDeps,
727    request_id: &str,
728    conn: &Connection,
729    databases: &[String],
730    classified: &ClassifiedStatement,
731    sql: &str,
732    resolution: &Resolution,
733    confirmed: bool,
734    outcome: crate::approval::outcomes::ApprovalOutcome,
735    approval_scope: Option<&str>,
736    approval_digest: Option<[u8; 32]>,
737    policy_revision: Option<u64>,
738    error: Option<&str>,
739    affected_rows: Option<u64>,
740    duration_ms: Option<u64>,
741    backup_id: Option<i64>,
742    opts: &WriteOptions,
743) -> Result<i64, String> {
744    let entry = AuditEntry {
745        request_id: request_id.to_string(),
746        connection: conn.name().to_string(),
747        databases: databases.to_vec(),
748        category: classified.category,
749        ast_type: Some(classified.ast_type.to_string()),
750        sql: sql.to_string(),
751        decision: resolution.action,
752        confirmed,
753        outcome,
754        affected_rows: affected_rows.map(|v| v as i64),
755        duration_ms: duration_ms.map(|v| v as i64),
756        error: error.map(str::to_string),
757        backup_id,
758        approval_scope: approval_scope.map(str::to_string),
759        approval_digest,
760        policy_revision,
761    };
762    crate::audit::write_audit_entry(&deps.audit, &entry, opts).map_err(|e| e.to_string())
763}
764
765impl ApprovalRequest {
766    /// Rebuild an IPC-able request from the elicitation message the
767    /// sink produced. The message shape (from the sink):
768    /// "About to run a {label} statement on {conn}[ · {db}].\n\n
769    /// --- SQL ---\n{statement}\n--- end ---\n\nAffected tables: a.b, c.d…"
770    pub fn from_message(message: &str) -> crate::approval::ipc::ApprovalRequest {
771        // "About to run a {label} statement on {target}."
772        let mut category = String::new();
773        let mut connection = String::new();
774        let mut database = None;
775        // Category label sits between "run a " and " statement on".
776        if let Some(run) = message.find("run a ")
777            && let Some(stmt) = message[run..].find(" statement on ")
778        {
779            category = message[run + "run a ".len()..run + stmt].trim().to_string();
780            let target = &message[run + stmt + " statement on ".len()..];
781            // Target ends at the first '.' that terminates it (the
782            // sentence) — take up to newline or ".\n".
783            let line = target.split('\n').next().unwrap_or("");
784            let line = line.trim_end_matches('.');
785            if let Some(dot) = line.find(" · ") {
786                connection = line[..dot].trim_matches('"').to_string();
787                database = Some(line[dot + 3..].trim().trim_matches('"').to_string());
788            } else {
789                connection = line.trim_matches('"').to_string();
790            }
791        }
792        // The SQL sits on the line AFTER the "--- SQL ---" marker.
793        let mut snippet = String::new();
794        if let Some(pos) = message.find("--- SQL ---") {
795            let after = &message[pos + "--- SQL ---".len()..];
796            let mut lines = after.lines();
797            let _ = lines.next(); // the marker's own remainder (empty)
798            for line in lines {
799                let trimmed = line.trim();
800                if trimmed == "--- end ---" {
801                    break;
802                }
803                if !snippet.is_empty() {
804                    snippet.push('\n');
805                }
806                snippet.push_str(line.trim_end());
807            }
808        }
809        let mut tables = Vec::new();
810        if let Some(idx) = message.find("Affected tables: ") {
811            let rest = &message[idx + "Affected tables: ".len()..];
812            let tail = rest.split('\n').next().unwrap_or("");
813            for part in tail.split(", ") {
814                let part = part.trim();
815                if part.contains('.') && !part.contains(' ') {
816                    tables.push(part.to_string());
817                }
818            }
819        }
820        crate::approval::ipc::ApprovalRequest {
821            id: uuid::Uuid::new_v4().to_string(),
822            category,
823            connection,
824            database,
825            tables,
826            snippet,
827        }
828    }
829}
830
831pub fn no_connection_message(explicit: Option<&str>) -> String {
832    match explicit {
833        Some(name) => {
834            format!("Unknown connection {name:?}. Use list_connections to see available names.")
835        }
836        None => "No connection specified and no default set. Pass \"connection\" or call set_default_connection first.".into(),
837    }
838}
839
840/// Map an outcome to the legacy tool JSON shape.
841pub fn outcome_to_json(o: &RunOutcome) -> serde_json::Value {
842    json!({
843        "connection": o.connection,
844        "category": o.category.as_str(),
845        "contributingDatabase": o.resolution.contributing_databases.first(),
846        "contributingDatabases": o.resolution.contributing_databases,
847        "rows": o.rows,
848        "fields": o.fields,
849        "affectedRows": o.affected_rows,
850        "truncated": o.truncated,
851        "rowCap": o.resolution.effective.row_cap,
852        "durationMs": o.duration_ms,
853        "backupId": o.backup_id,
854        "backupRowCount": o.backup_row_count,
855        "requestId": o.request_id,
856        "ddlNoOp": o.ddl_no_op,
857        "ddlAbsentTargets": o.ddl_absent_targets,
858        "ddlExecutedTargets": o.ddl_executed_targets,
859        "warnings": o.warnings,
860    })
861}
862
863#[cfg(test)]
864mod tests {
865    use super::*;
866    use crate::config::{ConfigStore, SqliteConnection};
867    use crate::policy::model::{PartialPolicy, PolicyPresetName, TableRuleKey, policy_from_preset};
868    use crate::vault::touchid::NoTouchId;
869    use std::sync::Mutex;
870
871    /// The IPC fallback rebuilds its request from the elicitation
872    /// message; category / connection / database / snippet / tables must
873    /// all come through (review finding: the snippet used to parse as
874    /// empty and the category as "a", so companion approvals showed no
875    /// SQL).
876    #[test]
877    fn approval_request_from_message_round_trip() {
878        let message = "About to run a WRITE statement on prod · staging.\n\n--- SQL ---\nUPDATE users SET email = 'x' WHERE id = 1\n--- end ---\n\nAffected tables: staging.users, other.things\n\nPick an authorization scope.";
879        let r = ApprovalRequest::from_message(message);
880        assert_eq!(r.category, "WRITE");
881        assert_eq!(r.connection, "prod");
882        assert_eq!(r.database.as_deref(), Some("staging"));
883        assert_eq!(r.snippet, "UPDATE users SET email = 'x' WHERE id = 1");
884        assert_eq!(
885            r.tables,
886            vec!["staging.users".to_string(), "other.things".to_string()]
887        );
888
889        // No database scope variant.
890        let message = "About to run a DDL (schema-changing) statement on \"local-dev\".\n\n--- SQL ---\nDROP TABLE x\n--- end ---\n\nAffected tables: (statement scope)\n";
891        let r = ApprovalRequest::from_message(message);
892        assert_eq!(r.category, "DDL (schema-changing)");
893        assert_eq!(r.connection, "local-dev");
894        assert!(r.database.is_none());
895        assert_eq!(r.snippet, "DROP TABLE x");
896        assert!(r.tables.is_empty());
897    }
898
899    struct ScriptedSink(Vec<Mutex<Option<ConfirmOutcome>>>);
900
901    impl ApprovalSink for ScriptedSink {
902        fn confirm(&self, _r: ApprovalRequest) -> ConfirmOutcome {
903            self.0
904                .first()
905                .and_then(|m| m.lock().unwrap().take())
906                .unwrap_or(ConfirmOutcome::Unavailable {
907                    reason: "script exhausted".into(),
908                })
909        }
910    }
911
912    fn sink_with(outcome: ConfirmOutcome) -> Box<dyn ApprovalSink> {
913        Box::new(ScriptedSink(vec![Mutex::new(Some(outcome))]))
914    }
915
916    fn deps(dir: &tempfile::TempDir, sink: Box<dyn ApprovalSink>) -> GateDeps {
917        let mut d = GateDeps::with_sink(sink);
918        d.config = Arc::new(ConfigStore::with_path(dir.path().join("cfg.json")));
919        d.audit = Arc::new(AuditDb::at_path(&dir.path().join("audit.sqlite")).unwrap());
920        // Touch ID must never trigger in tests: unavailable prompt fails
921        // closed, which is exactly what the read-only preset needs.
922        d.auth = Arc::new(SessionAuthenticator::new(Box::new(NoTouchId)));
923        d
924    }
925
926    fn sqlite_config(
927        dir: &tempfile::TempDir,
928        policy: crate::policy::model::Policy,
929        rules: Vec<(TableRuleKey, PartialPolicy)>,
930    ) {
931        let mut sc = SqliteConnection {
932            name: "local-sqlite".into(),
933            path: dir.path().join("app.sqlite").display().to_string(),
934            ..SqliteConnection::default()
935        };
936        sc.policy = policy;
937        for (k, v) in rules {
938            sc.table_policies.insert(k, v);
939        }
940        let store = ConfigStore::with_path(dir.path().join("cfg.json"));
941        let cfg = store.load().unwrap();
942        store
943            .update(cfg.revision, |c| {
944                c.connections.push(crate::config::Connection::Sqlite(sc));
945                c.default_connection = Some("local-sqlite".into());
946                Ok(())
947            })
948            .unwrap();
949    }
950
951    fn args(sql: &str) -> RunSqlArgs {
952        RunSqlArgs {
953            connection: None,
954            sql: sql.into(),
955            database: None,
956            expected_ddl_targets: None,
957        }
958    }
959
960    #[test]
961    fn read_runs_without_prompt() {
962        let dir = tempfile::tempdir().unwrap();
963        let d = deps(
964            &dir,
965            sink_with(ConfirmOutcome::Unavailable { reason: "x".into() }),
966        );
967        sqlite_config(&dir, policy_from_preset(PolicyPresetName::ReadOnly), vec![]);
968        bootstrap_table(&d, &dir);
969        let out = run_sql(&d, &args("SELECT id FROM users"), true).unwrap();
970        assert_eq!(out.rows.len(), 1);
971        assert_eq!(out.category, SqlCategory::Read);
972    }
973
974    /// Review finding: the approval engine must be shared across calls,
975    /// or an "Allow for session" grant dies with the request. The sink
976    /// holds ONE scripted Session outcome; the second call must succeed
977    /// without prompting (an exhausted sink reports unavailable, which
978    /// would fail the run).
979    #[test]
980    fn session_grant_persists_across_calls() {
981        let dir = tempfile::tempdir().unwrap();
982        let mut d = GateDeps::with_sink_and_approvals(
983            sink_with(ConfirmOutcome::Chosen(GrantChoice::Session)),
984            Arc::new(ApprovalEngine::new()),
985        );
986        d.config = Arc::new(ConfigStore::with_path(dir.path().join("cfg.json")));
987        d.audit = Arc::new(AuditDb::at_path(&dir.path().join("audit.sqlite")).unwrap());
988        d.auth = Arc::new(SessionAuthenticator::new(Box::new(NoTouchId)));
989        sqlite_config(
990            &dir,
991            policy_from_preset(PolicyPresetName::Development),
992            vec![],
993        );
994        bootstrap_table(&d, &dir);
995        let out = run_sql(&d, &args("UPDATE users SET id = 2 WHERE id = 1"), false).unwrap();
996        assert_eq!(out.affected_rows, 1);
997        // Same engine, exhausted sink: the session grant must cover it.
998        let out2 = run_sql(&d, &args("UPDATE users SET id = 1 WHERE id = 2"), false).unwrap();
999        assert_eq!(out2.affected_rows, 1);
1000        // A different table set is NOT covered by the grant.
1001        let err = run_sql(&d, &args("UPDATE users SET id = 3 WHERE id = 1"), false);
1002        // (still covered — same table set; see approval tests for the
1003        // exact-set semantics. Here the sink is exhausted, so a miss
1004        // would surface as Unavailable.)
1005        assert!(err.is_ok(), "{err:?}");
1006    }
1007
1008    #[test]
1009    fn write_denied_by_baseline() {
1010        let dir = tempfile::tempdir().unwrap();
1011        let d = deps(
1012            &dir,
1013            sink_with(ConfirmOutcome::Unavailable { reason: "x".into() }),
1014        );
1015        sqlite_config(&dir, policy_from_preset(PolicyPresetName::ReadOnly), vec![]);
1016        bootstrap_table(&d, &dir);
1017        let err = run_sql(&d, &args("UPDATE users SET id = 2"), false).unwrap_err();
1018        assert!(matches!(err, GateError::Denied(_)), "{err}");
1019    }
1020
1021    #[test]
1022    fn elevation_prompts_and_decline_blocks() {
1023        let dir = tempfile::tempdir().unwrap();
1024        let d = deps(
1025            &dir,
1026            sink_with(ConfirmOutcome::Chosen(GrantChoice::Decline)),
1027        );
1028        let rule = PartialPolicy {
1029            write: Some(PolicyAction::Allow),
1030            ..PartialPolicy::default()
1031        };
1032        sqlite_config(
1033            &dir,
1034            policy_from_preset(PolicyPresetName::ReadOnly),
1035            vec![(TableRuleKey::parse("main.users").unwrap(), rule)],
1036        );
1037        bootstrap_table(&d, &dir);
1038        let err = run_sql(&d, &args("UPDATE users SET id = 2 WHERE id = 1"), false).unwrap_err();
1039        assert!(matches!(err, GateError::Declined(_)), "{err}");
1040        // Nothing changed.
1041        let d2 = deps(
1042            &dir,
1043            sink_with(ConfirmOutcome::Unavailable { reason: "x".into() }),
1044        );
1045        let out = run_sql(&d2, &args("SELECT id FROM users"), true).unwrap();
1046        assert_eq!(out.rows[0]["id"], json!(1));
1047    }
1048
1049    #[test]
1050    fn unavailable_is_reported_as_unavailable_not_declined() {
1051        let dir = tempfile::tempdir().unwrap();
1052        let d = deps(&dir, Box::new(UnavailableSink));
1053        let rule = PartialPolicy {
1054            write: Some(PolicyAction::Allow),
1055            ..PartialPolicy::default()
1056        };
1057        sqlite_config(
1058            &dir,
1059            policy_from_preset(PolicyPresetName::ReadOnly),
1060            vec![(TableRuleKey::parse("main.users").unwrap(), rule)],
1061        );
1062        bootstrap_table(&d, &dir);
1063        let err = run_sql(&d, &args("UPDATE users SET id = 2 WHERE id = 1"), false).unwrap_err();
1064        assert!(matches!(err, GateError::Unavailable(_, _)), "{err}");
1065        // Audit recorded `unavailable`, not `declined`.
1066        let filters = crate::audit::AuditSearchFilters {
1067            limit: 10,
1068            ..Default::default()
1069        };
1070        let rows = crate::audit::search_audit_log(&d.audit, &filters).unwrap();
1071        assert!(rows.iter().any(|r| r.outcome == "unavailable"), "{rows:?}");
1072        assert!(!rows.iter().any(|r| r.outcome == "declined"));
1073    }
1074
1075    #[test]
1076    fn approved_once_executes_and_backs_up() {
1077        let dir = tempfile::tempdir().unwrap();
1078        let d = deps(&dir, sink_with(ConfirmOutcome::Chosen(GrantChoice::Once)));
1079        let rule = PartialPolicy {
1080            write: Some(PolicyAction::Allow),
1081            ..PartialPolicy::default()
1082        };
1083        sqlite_config(
1084            &dir,
1085            policy_from_preset(PolicyPresetName::ReadOnly),
1086            vec![(TableRuleKey::parse("main.users").unwrap(), rule)],
1087        );
1088        bootstrap_table(&d, &dir);
1089        let out = run_sql(&d, &args("UPDATE users SET id = 2 WHERE id = 1"), false).unwrap();
1090        assert_eq!(out.affected_rows, 1);
1091        assert!(out.backup_id.is_some());
1092        let rows = crate::audit::search_audit_log(
1093            &d.audit,
1094            &crate::audit::AuditSearchFilters {
1095                limit: 10,
1096                ..Default::default()
1097            },
1098        )
1099        .unwrap();
1100        assert!(rows.iter().any(|r| r.outcome == "approved"));
1101    }
1102
1103    #[test]
1104    fn query_tool_rejects_writes() {
1105        let dir = tempfile::tempdir().unwrap();
1106        let d = deps(
1107            &dir,
1108            sink_with(ConfirmOutcome::Unavailable { reason: "x".into() }),
1109        );
1110        sqlite_config(
1111            &dir,
1112            policy_from_preset(PolicyPresetName::Development),
1113            vec![],
1114        );
1115        bootstrap_table(&d, &dir);
1116        let err = run_sql(&d, &args("DELETE FROM users"), true).unwrap_err();
1117        assert!(matches!(err, GateError::NotReadOnly(_)));
1118    }
1119
1120    fn bootstrap_table(d: &GateDeps, dir: &tempfile::TempDir) {
1121        let mut sc = SqliteConnection {
1122            name: "bootstrap".into(),
1123            path: dir.path().join("app.sqlite").display().to_string(),
1124            ..SqliteConnection::default()
1125        };
1126        sc.policy = policy_from_preset(PolicyPresetName::Administration);
1127        sc.policy.require_touch_id = false;
1128        let classified = crate::policy::classifier::classify_statement(
1129            "CREATE TABLE IF NOT EXISTS users (id INTEGER PRIMARY KEY)",
1130            Dialect::SQLite,
1131        )
1132        .unwrap();
1133        sqlite::execute_sqlite_statement(SqliteExecuteParams {
1134            connection: &sc,
1135            sql: "CREATE TABLE IF NOT EXISTS users (id INTEGER PRIMARY KEY)",
1136            classified: &classified,
1137            policy: &sc.policy,
1138            database: None,
1139            audit: Some(d.audit.clone()),
1140        })
1141        .unwrap();
1142        let classified = crate::policy::classifier::classify_statement(
1143            "INSERT INTO users (id) VALUES (1)",
1144            Dialect::SQLite,
1145        )
1146        .unwrap();
1147        sqlite::execute_sqlite_statement(SqliteExecuteParams {
1148            connection: &sc,
1149            sql: "INSERT INTO users (id) VALUES (1)",
1150            classified: &classified,
1151            policy: &sc.policy,
1152            database: None,
1153            audit: Some(d.audit.clone()),
1154        })
1155        .unwrap();
1156    }
1157}
sequel_mcp/app/gate.rs

sequel_mcp/app/
gate.rs
