From 92138b8ccf6b39b7705c5a4cd019e07f8e64b3cd Mon Sep 17 00:00:00 2001 From: Zaki Date: Mon, 30 Mar 2026 19:51:02 -0700 Subject: [PATCH] fix: address code review findings for cross-channel approval security 1. Add "telegram" to WASM channel name blocklist -- bundled channels like telegram were claimable by malicious WASM modules that load before the bundled one, bypassing cross-channel approval auth. 2. Make V16 libSQL migration (ADD COLUMN source_channel) idempotent -- the runner now checks pragma_table_info before executing ALTER TABLE, preventing startup failures if the base schema already includes the column. 3. Replace silent .unwrap_or(None) in thread hydration with explicit match on DB result -- legacy threads without stored source_channel now log a warning, and DB errors log an error. Both cases remain fail-closed (approvals denied) but are no longer silent. Co-Authored-By: Claude Opus 4.6 (1M context) --- src/agent/thread_ops.rs | 31 +++++++++++++++++--- src/channels/wasm/setup.rs | 4 ++- src/db/libsql_migrations.rs | 58 +++++++++++++++++++++++++++++++++++-- 3 files changed, 85 insertions(+), 8 deletions(-) diff --git a/src/agent/thread_ops.rs b/src/agent/thread_ops.rs index c53d37eb40..1797fbcde9 100644 --- a/src/agent/thread_ops.rs +++ b/src/agent/thread_ops.rs @@ -138,11 +138,34 @@ impl Agent { // Create thread with the historical ID and restore messages. // Read source_channel from DB so the authorization check uses the // original creator's channel, not the requesting message's channel. + // + // Fail-closed policy: if the DB lookup fails or the conversation has + // no stored source_channel (legacy row), the thread is hydrated with + // source_channel = None. `is_approval_authorized(None, _)` returns + // false, so approvals are denied until the conversation is backfilled + // with a source_channel via an explicit migration or re-creation. let db_source_channel = if let Some(store) = self.store() { - store - .get_conversation_source_channel(thread_uuid) - .await - .unwrap_or(None) + match store.get_conversation_source_channel(thread_uuid).await { + Ok(sc) => { + if sc.is_none() { + tracing::warn!( + thread_id = %thread_uuid, + "Legacy thread has no stored source_channel; \ + cross-channel approvals will be denied (fail-closed)" + ); + } + sc + } + Err(e) => { + tracing::error!( + thread_id = %thread_uuid, + error = %e, + "Failed to read source_channel from DB; \ + cross-channel approvals will be denied (fail-closed)" + ); + None + } + } } else { None }; diff --git a/src/channels/wasm/setup.rs b/src/channels/wasm/setup.rs index edf0123cde..638fae56c6 100644 --- a/src/channels/wasm/setup.rs +++ b/src/channels/wasm/setup.rs @@ -87,6 +87,7 @@ pub async fn setup_wasm_channels( "repl", "http", "signal", + "telegram", "slack-relay", "secret_save", ]; @@ -505,6 +506,7 @@ mod tests { "repl", "http", "signal", + "telegram", "slack-relay", "secret_save", ]; @@ -554,7 +556,7 @@ mod tests { #[test] fn non_reserved_names_allowed() { let reserved = reserved_names(); - let allowed = ["telegram", "discord", "my-custom-channel", "slack-bot"]; + let allowed = ["discord", "my-custom-channel", "slack-bot"]; for name in allowed { assert!( !reserved.contains(&name), diff --git a/src/db/libsql_migrations.rs b/src/db/libsql_migrations.rs index 0d9adf36e1..6bdaac1703 100644 --- a/src/db/libsql_migrations.rs +++ b/src/db/libsql_migrations.rs @@ -791,12 +791,42 @@ CREATE INDEX IF NOT EXISTS idx_api_tokens_hash ON api_tokens(token_hash); 16, "conversation_source_channel", // Add source_channel to conversations for cross-channel approval authorization. + // Marked as idempotent (see IDEMPOTENT_ADD_COLUMN_MIGRATIONS below) + // because SQLite does not support IF NOT EXISTS for ADD COLUMN. + // The runner checks pragma_table_info before executing the ALTER. r#" ALTER TABLE conversations ADD COLUMN source_channel TEXT; "#, ), ]; +/// Migrations whose ADD COLUMN should be skipped when the column already +/// exists (e.g. because the base SCHEMA was updated to include it). +/// Each entry is `(version, table_name, column_name)`. +const IDEMPOTENT_ADD_COLUMN_MIGRATIONS: &[(i64, &str, &str)] = + &[(16, "conversations", "source_channel")]; + +/// Check whether `table` already contains `column` via `pragma_table_info`. +async fn column_exists( + conn: &libsql::Connection, + table: &str, + column: &str, +) -> Result { + use crate::error::DatabaseError; + + let sql = format!( + "SELECT 1 FROM pragma_table_info('{}') WHERE name = ?1", + table + ); + let mut rows = conn + .query(&sql, libsql::params![column]) + .await + .map_err(|e| { + DatabaseError::Migration(format!("Failed to check column {table}.{column}: {e}")) + })?; + Ok(rows.next().await.ok().flatten().is_some()) +} + /// Run incremental migrations that haven't been applied yet. /// /// Each migration is wrapped in a transaction. On success the version is @@ -821,6 +851,18 @@ pub async fn run_incremental(conn: &libsql::Connection) -> Result<(), crate::err continue; // Already applied } + // For ADD COLUMN migrations, skip the ALTER if the column already + // exists (e.g. because the base SCHEMA was updated to include it) + // and just record the migration as applied. + let skip_sql = if let Some(&(_, table, column)) = IDEMPOTENT_ADD_COLUMN_MIGRATIONS + .iter() + .find(|(v, _, _)| *v == version) + { + column_exists(conn, table, column).await? + } else { + false + }; + // Wrap migration + recording in a transaction for atomicity. // If the process crashes mid-migration, the transaction rolls back // and the migration will be retried on next startup. @@ -830,9 +872,19 @@ pub async fn run_incremental(conn: &libsql::Connection) -> Result<(), crate::err )) })?; - tx.execute_batch(sql).await.map_err(|e| { - DatabaseError::Migration(format!("libSQL migration V{version} ({name}) failed: {e}")) - })?; + if skip_sql { + tracing::debug!( + version, + name, + "libSQL: column already exists, recording migration as applied" + ); + } else { + tx.execute_batch(sql).await.map_err(|e| { + DatabaseError::Migration(format!( + "libSQL migration V{version} ({name}) failed: {e}" + )) + })?; + } // Record as applied (inside the same transaction) tx.execute(