fix(security): persist source_channel to DB, harden cross-channel authorization

Address PR #1590 review feedback:

1. Persist source_channel to DB: Add source_channel column to conversations
   table in both PostgreSQL (V14 migration) and libSQL (incremental migration
   + base schema). Add get_conversation_source_channel trait method to
   ConversationStore with both backend implementations.

2. Fix hydrate_thread_from_db: Read source_channel from DB instead of
   stamping the requesting message's channel, preventing channel confusion
   after server restart.

3. Reject reserved WASM channel names: Validate that WASM channels cannot
   register as "web", "gateway", "cli", or "repl" to prevent authorization
   bypass via name spoofing.

4. Require pending_approval exists: Authorization check now verifies
   thread.pending_approval.is_some() before allowing approval-shaped messages
   to target a thread.

5. Fail-closed for None source_channel: Use "__bootstrap__" sentinel for
   bootstrap threads (authorized from any channel). None now means "deny by
   default" instead of "allow by default".

6. Extract and test authorization predicate: is_approval_authorized() helper
   with 6 unit tests covering same-channel, cross-channel blocked, web/gateway
   always allowed, None denied, and bootstrap sentinel.

Co-Authored-By: Claude Opus 4.6 (1M context) <[email protected]>

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;

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,

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"
+        );
+    }
 }

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,

src/channels/wasm/setup.rs

@@ -71,7 +71,21 @@ pub async fn setup_wasm_channels(
     let mut channels: Vec<(String, Box<dyn crate::channels::Channel>)> = Vec::new();
     let mut channel_names: Vec<String> = 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,

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) => {}

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) => {}

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<bool, DatabaseError> {
         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<Option<String>, 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)]

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;
 "#,
     ),
 ];

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<bool, DatabaseError>;
     async fn list_conversations_with_preview(
         &self,
@@ -431,6 +432,11 @@ pub trait ConversationStore: Send + Sync {
         conversation_id: Uuid,
         user_id: &str,
     ) -> Result<bool, DatabaseError>;
+    /// Get the source_channel for a conversation (the channel that created it).
+    async fn get_conversation_source_channel(
+        &self,
+        conversation_id: Uuid,
+    ) -> Result<Option<String>, DatabaseError>;
 }
 
 #[async_trait]

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<bool, DatabaseError> {
         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<Option<String>, DatabaseError> {
+        self.store
+            .get_conversation_source_channel(conversation_id)
+            .await
+    }
 }
 
 // ==================== JobStore ====================

src/history/store.rs

@@ -1581,19 +1581,20 @@ impl Store {
         channel: &str,
         user_id: &str,
         thread_id: Option<&str>,
+        source_channel: Option<&str>,
     ) -> Result<bool, DatabaseError> {
         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<Option<String>, 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<String>>(0)))
+    }
+
     /// Load messages for a conversation with cursor-based pagination.
     ///
     /// Returns `(messages_oldest_first, has_more)`.

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"

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"