diff --git a/migrations/V14__conversation_source_channel.sql b/migrations/V14__conversation_source_channel.sql new file mode 100644 index 00000000..340b48a6 --- /dev/null +++ b/migrations/V14__conversation_source_channel.sql @@ -0,0 +1,4 @@ +-- Add source_channel to conversations for cross-channel approval authorization. +-- Tracks which channel originally created a conversation so that approval +-- messages from other channels can be validated. +ALTER TABLE conversations ADD COLUMN source_channel TEXT; diff --git a/src/agent/agent_loop.rs b/src/agent/agent_loop.rs index 1c605ce1..f9745e58 100644 --- a/src/agent/agent_loop.rs +++ b/src/agent/agent_loop.rs @@ -843,9 +843,10 @@ impl Agent { { use crate::agent::session::Thread; let mut sess = session.lock().await; - // Bootstrap thread has no incoming message -- use None for - // source_channel so approvals from any channel are permitted. - let thread = Thread::with_id(id, sess.id, None); + // Bootstrap thread has no incoming message -- use the + // "__bootstrap__" sentinel so approvals from any channel are + // permitted. None means "deny by default" (fail-closed). + let thread = Thread::with_id(id, sess.id, Some("__bootstrap__")); sess.active_thread = Some(id); sess.threads.entry(id).or_insert(thread); } @@ -1151,11 +1152,26 @@ impl Agent { .await; let mut sess = session.lock().await; if let Some(thread) = sess.threads.get(&target_thread_id) { - let authorized = thread.source_channel.as_ref().is_none_or(|src| { - src == &message.channel - || message.channel == "web" - || message.channel == "gateway" - }); + // Verify the thread actually has a pending approval before + // allowing approval-shaped messages to target it. Without this + // check, an attacker could use approval messages to hijack any + // thread by UUID. + if thread.pending_approval.is_none() { + tracing::warn!( + %target_thread_id, + approval_channel = %message.channel, + "Blocked approval for thread with no pending approval" + ); + drop(sess); + return Ok(Some( + "Error: no pending approval on this thread".into(), + )); + } + + let authorized = crate::agent::session::is_approval_authorized( + thread.source_channel.as_deref(), + &message.channel, + ); if !authorized { tracing::warn!( %target_thread_id, diff --git a/src/agent/session.rs b/src/agent/session.rs index dd0ec726..548105c2 100644 --- a/src/agent/session.rs +++ b/src/agent/session.rs @@ -236,6 +236,28 @@ pub struct Thread { /// rapid follow-ups. The drain loop processes them as one newline-delimited turn. pub const MAX_PENDING_MESSAGES: usize = 10; +/// Sentinel value for bootstrap threads that accept approvals from any channel. +pub const BOOTSTRAP_SOURCE_CHANNEL: &str = "__bootstrap__"; + +/// Check whether an approval from `requesting_channel` is authorized for a +/// thread whose `source_channel` is `source`. +/// +/// Rules: +/// - `None` (unknown origin) -> denied (fail-closed) +/// - `Some("__bootstrap__")` -> authorized from any channel +/// - `Some(src) == requesting` -> same channel, authorized +/// - requesting is "web" or "gateway" -> always authorized (trusted UI) +/// - Otherwise -> denied +pub fn is_approval_authorized(source: Option<&str>, requesting: &str) -> bool { + match source { + None => false, + Some(src) if src == BOOTSTRAP_SOURCE_CHANNEL => true, + Some(src) => { + src == requesting || requesting == "web" || requesting == "gateway" + } + } +} + impl Thread { /// Create a new thread. pub fn new(session_id: Uuid, source_channel: Option<&str>) -> Self { @@ -1847,4 +1869,64 @@ mod tests { "missing source_channel should deserialize as None" ); } + + #[test] + fn test_approval_authorized_same_channel() { + assert!( + is_approval_authorized(Some("telegram"), "telegram"), + "same channel should be authorized" + ); + } + + #[test] + fn test_approval_authorized_different_channel_blocked() { + assert!( + !is_approval_authorized(Some("telegram"), "http"), + "different channel should be blocked" + ); + } + + #[test] + fn test_approval_authorized_web_always_allowed() { + assert!( + is_approval_authorized(Some("telegram"), "web"), + "web channel should always be authorized" + ); + } + + #[test] + fn test_approval_authorized_gateway_always_allowed() { + assert!( + is_approval_authorized(Some("telegram"), "gateway"), + "gateway channel should always be authorized" + ); + } + + #[test] + fn test_approval_authorized_none_denied() { + assert!( + !is_approval_authorized(None, "telegram"), + "None source_channel should be denied (fail-closed)" + ); + assert!( + !is_approval_authorized(None, "web"), + "None source_channel should be denied even for web" + ); + } + + #[test] + fn test_approval_authorized_bootstrap_any_channel() { + assert!( + is_approval_authorized(Some(BOOTSTRAP_SOURCE_CHANNEL), "telegram"), + "__bootstrap__ should be authorized from any channel" + ); + assert!( + is_approval_authorized(Some(BOOTSTRAP_SOURCE_CHANNEL), "http"), + "__bootstrap__ should be authorized from any channel" + ); + assert!( + is_approval_authorized(Some(BOOTSTRAP_SOURCE_CHANNEL), "cli"), + "__bootstrap__ should be authorized from any channel" + ); + } } diff --git a/src/agent/thread_ops.rs b/src/agent/thread_ops.rs index d39f285d..32bb55f7 100644 --- a/src/agent/thread_ops.rs +++ b/src/agent/thread_ops.rs @@ -135,14 +135,28 @@ impl Agent { msg_count = 0; } - // Create thread with the historical ID and restore messages + // 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. + let db_source_channel = if let Some(store) = self.store() { + store + .get_conversation_source_channel(thread_uuid) + .await + .unwrap_or(None) + } else { + None + }; + let effective_source_channel = db_source_channel + .as_deref() + .or(Some(&*message.channel)); + let session_id = { let sess = session.lock().await; sess.id }; let mut thread = - crate::agent::session::Thread::with_id(thread_uuid, session_id, Some(&message.channel)); + crate::agent::session::Thread::with_id(thread_uuid, session_id, effective_source_channel); if !chat_messages.is_empty() { thread.restore_from_messages(chat_messages); } @@ -637,7 +651,7 @@ impl Agent { user_id: &str, ) -> bool { match store - .ensure_conversation(thread_id, channel, user_id, None) + .ensure_conversation(thread_id, channel, user_id, None, Some(channel)) .await { Ok(true) => true, diff --git a/src/channels/wasm/setup.rs b/src/channels/wasm/setup.rs index 84df615f..fdda89e2 100644 --- a/src/channels/wasm/setup.rs +++ b/src/channels/wasm/setup.rs @@ -71,7 +71,21 @@ pub async fn setup_wasm_channels( let mut channels: Vec<(String, Box)> = Vec::new(); let mut channel_names: Vec = Vec::new(); + // Reserved channel names that WASM modules must not claim. + // A malicious module could otherwise register as a trusted built-in + // channel and bypass cross-channel authorization checks. + const RESERVED_CHANNEL_NAMES: &[&str] = &["web", "gateway", "cli", "repl"]; + for loaded in results.loaded { + let name_lower = loaded.name().to_ascii_lowercase(); + if RESERVED_CHANNEL_NAMES.contains(&name_lower.as_str()) { + tracing::warn!( + channel = %loaded.name(), + "Rejected WASM channel with reserved name" + ); + continue; + } + let (name, channel) = register_channel( loaded, config, diff --git a/src/channels/web/handlers/chat.rs b/src/channels/web/handlers/chat.rs index 302f8051..88d9050d 100644 --- a/src/channels/web/handlers/chat.rs +++ b/src/channels/web/handlers/chat.rs @@ -593,7 +593,7 @@ pub async fn chat_new_thread_handler( // so that the subsequent loadThreads() call from the frontend sees it. if let Some(ref store) = state.store { match store - .ensure_conversation(thread_id, "gateway", &identity.user_id, None) + .ensure_conversation(thread_id, "gateway", &identity.user_id, None, Some("gateway")) .await { Ok(true) => {} diff --git a/src/channels/web/server.rs b/src/channels/web/server.rs index 030f9d5e..f61765dd 100644 --- a/src/channels/web/server.rs +++ b/src/channels/web/server.rs @@ -2033,7 +2033,7 @@ async fn chat_new_thread_handler( // so that the subsequent loadThreads() call from the frontend sees it. if let Some(ref store) = state.store { match store - .ensure_conversation(thread_id, "gateway", &user.user_id, None) + .ensure_conversation(thread_id, "gateway", &user.user_id, None, Some("gateway")) .await { Ok(true) => {} diff --git a/src/db/libsql/conversations.rs b/src/db/libsql/conversations.rs index 911ee863..077e9d8f 100644 --- a/src/db/libsql/conversations.rs +++ b/src/db/libsql/conversations.rs @@ -67,19 +67,20 @@ impl ConversationStore for LibSqlBackend { channel: &str, user_id: &str, thread_id: Option<&str>, + source_channel: Option<&str>, ) -> Result { let conn = self.connect().await?; let now = fmt_ts(&Utc::now()); let affected = conn .execute( r#" - INSERT INTO conversations (id, channel, user_id, thread_id, started_at, last_activity) - VALUES (?1, ?2, ?3, ?4, ?5, ?5) + INSERT INTO conversations (id, channel, user_id, thread_id, source_channel, started_at, last_activity) + VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?6) ON CONFLICT (id) DO UPDATE SET last_activity = excluded.last_activity WHERE conversations.user_id = excluded.user_id AND conversations.channel = excluded.channel "#, - params![id.to_string(), channel, user_id, opt_text(thread_id), now], + params![id.to_string(), channel, user_id, opt_text(thread_id), opt_text(source_channel), now], ) .await .map_err(|e| DatabaseError::Query(e.to_string()))?; @@ -565,6 +566,28 @@ impl ConversationStore for LibSqlBackend { .map_err(|e| DatabaseError::Query(e.to_string()))?; Ok(found.is_some()) } + + async fn get_conversation_source_channel( + &self, + conversation_id: Uuid, + ) -> Result, DatabaseError> { + let conn = self.connect().await?; + let mut rows = conn + .query( + "SELECT source_channel FROM conversations WHERE id = ?1", + params![conversation_id.to_string()], + ) + .await + .map_err(|e| DatabaseError::Query(e.to_string()))?; + match rows + .next() + .await + .map_err(|e| DatabaseError::Query(e.to_string()))? + { + Some(row) => Ok(get_opt_text(&row, 0)), + None => Ok(None), + } + } } #[cfg(test)] diff --git a/src/db/libsql_migrations.rs b/src/db/libsql_migrations.rs index 2a4fa5c5..4dab6222 100644 --- a/src/db/libsql_migrations.rs +++ b/src/db/libsql_migrations.rs @@ -38,7 +38,8 @@ CREATE TABLE IF NOT EXISTS conversations ( thread_id TEXT, started_at TEXT NOT NULL DEFAULT (strftime('%Y-%m-%dT%H:%M:%fZ', 'now')), last_activity TEXT NOT NULL DEFAULT (strftime('%Y-%m-%dT%H:%M:%fZ', 'now')), - metadata TEXT NOT NULL DEFAULT '{}' + metadata TEXT NOT NULL DEFAULT '{}', + source_channel TEXT ); CREATE INDEX IF NOT EXISTS idx_conversations_channel ON conversations(channel); @@ -785,6 +786,14 @@ CREATE TABLE IF NOT EXISTS api_tokens ( ); CREATE INDEX IF NOT EXISTS idx_api_tokens_user ON api_tokens(user_id); CREATE INDEX IF NOT EXISTS idx_api_tokens_hash ON api_tokens(token_hash); +"#, + ), + ( + 15, + "conversation_source_channel", + // Add source_channel to conversations for cross-channel approval authorization. + r#" +ALTER TABLE conversations ADD COLUMN source_channel TEXT; "#, ), ]; diff --git a/src/db/mod.rs b/src/db/mod.rs index 14cad543..1c27ca8d 100644 --- a/src/db/mod.rs +++ b/src/db/mod.rs @@ -373,6 +373,7 @@ pub trait ConversationStore: Send + Sync { channel: &str, user_id: &str, thread_id: Option<&str>, + source_channel: Option<&str>, ) -> Result; async fn list_conversations_with_preview( &self, @@ -431,6 +432,11 @@ pub trait ConversationStore: Send + Sync { conversation_id: Uuid, user_id: &str, ) -> Result; + /// Get the source_channel for a conversation (the channel that created it). + async fn get_conversation_source_channel( + &self, + conversation_id: Uuid, + ) -> Result, DatabaseError>; } #[async_trait] diff --git a/src/db/postgres.rs b/src/db/postgres.rs index 2fba0b53..d942e3db 100644 --- a/src/db/postgres.rs +++ b/src/db/postgres.rs @@ -99,9 +99,10 @@ impl ConversationStore for PgBackend { channel: &str, user_id: &str, thread_id: Option<&str>, + source_channel: Option<&str>, ) -> Result { self.store - .ensure_conversation(id, channel, user_id, thread_id) + .ensure_conversation(id, channel, user_id, thread_id, source_channel) .await } @@ -212,6 +213,15 @@ impl ConversationStore for PgBackend { .conversation_belongs_to_user(conversation_id, user_id) .await } + + async fn get_conversation_source_channel( + &self, + conversation_id: Uuid, + ) -> Result, DatabaseError> { + self.store + .get_conversation_source_channel(conversation_id) + .await + } } // ==================== JobStore ==================== diff --git a/src/history/store.rs b/src/history/store.rs index e6e869b6..6e6c6915 100644 --- a/src/history/store.rs +++ b/src/history/store.rs @@ -1581,19 +1581,20 @@ impl Store { channel: &str, user_id: &str, thread_id: Option<&str>, + source_channel: Option<&str>, ) -> Result { let conn = self.conn().await?; let affected = conn .execute( r#" - INSERT INTO conversations (id, channel, user_id, thread_id) - VALUES ($1, $2, $3, $4) + INSERT INTO conversations (id, channel, user_id, thread_id, source_channel) + VALUES ($1, $2, $3, $4, $5) ON CONFLICT (id) DO UPDATE SET last_activity = NOW() WHERE conversations.user_id = EXCLUDED.user_id AND conversations.channel = EXCLUDED.channel "#, - &[&id, &channel, &user_id, &thread_id], + &[&id, &channel, &user_id, &thread_id, &source_channel], ) .await?; Ok(affected > 0) @@ -1892,6 +1893,21 @@ impl Store { Ok(row.is_some()) } + /// Get the source_channel for a conversation. + pub async fn get_conversation_source_channel( + &self, + conversation_id: Uuid, + ) -> Result, DatabaseError> { + let conn = self.conn().await?; + let row = conn + .query_opt( + "SELECT source_channel FROM conversations WHERE id = $1", + &[&conversation_id], + ) + .await?; + Ok(row.and_then(|r| r.get::<_, Option>(0))) + } + /// Load messages for a conversation with cursor-based pagination. /// /// Returns `(messages_oldest_first, has_more)`. diff --git a/src/testing/mod.rs b/src/testing/mod.rs index dfff4b10..bd1cc52f 100644 --- a/src/testing/mod.rs +++ b/src/testing/mod.rs @@ -753,7 +753,7 @@ mod tests { // ensure_conversation should create the row. assert!( - db.ensure_conversation(conv_id, "web", "carol", None) + db.ensure_conversation(conv_id, "web", "carol", None, Some("web")) .await .expect("ensure first"), "first ensure_conversation should create the row" @@ -761,7 +761,7 @@ mod tests { // Calling again with the same ID should not error. assert!( - db.ensure_conversation(conv_id, "web", "carol", None) + db.ensure_conversation(conv_id, "web", "carol", None, Some("web")) .await .expect("ensure second (idempotent)"), "second ensure_conversation should touch owned row" @@ -806,7 +806,7 @@ mod tests { tokio::time::sleep(std::time::Duration::from_millis(25)).await; assert!( - !db.ensure_conversation(conv_id, "web", "mallory", None) + !db.ensure_conversation(conv_id, "web", "mallory", None, None) .await .expect("foreign ensure should not error"), "foreign ensure_conversation should report not ensured" diff --git a/tests/e2e_thread_id_isolation.rs b/tests/e2e_thread_id_isolation.rs index baec73c1..1d04d848 100644 --- a/tests/e2e_thread_id_isolation.rs +++ b/tests/e2e_thread_id_isolation.rs @@ -48,7 +48,7 @@ mod tests { let store = rig.database(); assert!( store - .ensure_conversation(foreign_thread_id, "gateway", "victim-user", None) + .ensure_conversation(foreign_thread_id, "gateway", "victim-user", None, Some("gateway")) .await .expect("failed to create victim conversation"), "test setup failed: victim conversation was not created"