Compare commits

..
Author SHA1 Message Date
ZakiandClaude Opus 4.6 3f970252d5 fix(security): address review feedback on cross-channel approval checks
1. thread_ops.rs: Remove .or(Some(&*message.channel)) fallback in
   maybe_hydrate_thread() so that when source_channel is NULL in the DB,
   it stays None rather than being stamped with the requesting channel.
   This preserves the fail-closed behavior of is_approval_authorized().

2. libsql_migrations.rs: Remove source_channel from base SCHEMA to
   eliminate duplicate column definition. The column is now added solely
   by V14 migration, preventing fresh databases from failing on startup.

3. wasm/setup.rs: Expand RESERVED_CHANNEL_NAMES to cover all built-in
   channels (http, signal, slack-relay, secret_save) and add a dynamic
   collision check against already-registered channel names passed from
   the startup sequence.

Co-Authored-By: Claude Opus 4.6 (1M context) <[email protected]>
2026-03-28 11:13:04 -07:00
Claude 3930184e71 style: fix cargo fmt formatting
https://claude.ai/code/session_01Ci7CAdGaHhssYdio7wxVvd
2026-03-28 16:01:35 +00:00
Claude 4fb0d53206 fix: resolve merge conflicts from staging rebase
- Fix Thread::with_id calls to include source_channel parameter
- Fix ensure_conversation calls to include source_channel parameter
- Bump libsql source_channel migration to V15 (V14 taken by users)
- Remove stale conflict markers
- Fix clippy warning in users.rs

https://claude.ai/code/session_01Esh8QQzHACYyfsVwCb479F
2026-03-28 15:20:53 +00:00
ZakiandClaude 5f50df7583 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]>
2026-03-28 15:11:03 +00:00
Claude 30feaabb7c fix(clippy): use is_none_or instead of map_or for Option check
is_none_or is stable since Rust 1.82 and preferred by clippy over
map_or(true, ...) pattern.

https://claude.ai/code/session_012nbbEyFXjDwdZZrHg7gFNK
2026-03-28 15:10:20 +00:00
ZakiandClaude 427f908e71 fix(security): address review feedback on source_channel
- hydrate_thread_from_db now passes message.channel as source_channel
  instead of None, ensuring DB-hydrated threads get proper channel auth
- Replace is_none_or (unstable) with map_or(true, ...) for MSRV compat
- Add "gateway" to trusted approval channels alongside "web"
- Document why bootstrap thread uses None for source_channel

Co-Authored-By: Claude Opus 4.6 (1M context) <[email protected]>
2026-03-28 15:10:20 +00:00
Claude 8bd30a970e style: run cargo fmt
https://claude.ai/code/session_01Mdiz3XwyZcjqMkqicaynGs
2026-03-28 15:10:20 +00:00
ZakiandClaude 5d1d504e11 fix(security): block cross-channel approval thread hijacking (#1485)
Add source_channel to Thread and verify channel authorization before
allowing approval messages to target threads by UUID. The web gateway
channel is allowed as a trusted approval UI. Threads without
source_channel (deserialized from older DB records) are permitted
for backward compatibility.

Closes #1485

Co-Authored-By: Claude Opus 4.6 (1M context) <[email protected]>
2026-03-28 15:10:20 +00:00
20 changed files with 401 additions and 102 deletions
@@ -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;
+35 -2
View File
@@ -843,7 +843,10 @@ impl Agent {
{
use crate::agent::session::Thread;
let mut sess = session.lock().await;
let thread = Thread::with_id(id, sess.id);
// 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);
}
@@ -1148,7 +1151,37 @@ impl Agent {
.get_or_create_session(&message.user_id)
.await;
let mut sess = session.lock().await;
if sess.threads.contains_key(&target_thread_id) {
if let Some(thread) = sess.threads.get(&target_thread_id) {
// 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,
source_channel = ?thread.source_channel,
approval_channel = %message.channel,
"Blocked cross-channel approval attempt"
);
drop(sess);
return Ok(Some(
"Error: approval not authorized for this channel".into(),
));
}
sess.active_thread = Some(target_thread_id);
sess.last_active_at = chrono::Utc::now();
drop(sess);
+5 -5
View File
@@ -319,7 +319,7 @@ mod tests {
#[test]
fn test_format_turns() {
let mut thread = Thread::new(Uuid::new_v4());
let mut thread = Thread::new(Uuid::new_v4(), None);
thread.start_turn("Hello");
thread.complete_turn("Hi there");
thread.start_turn("How are you?");
@@ -351,7 +351,7 @@ mod tests {
/// Helper: build a thread with `n` completed turns.
/// Turn `i` has user_input "msg-{i}" and response "resp-{i}".
fn make_thread(n: usize) -> Thread {
let mut thread = Thread::new(Uuid::new_v4());
let mut thread = Thread::new(Uuid::new_v4(), None);
for i in 0..n {
thread.start_turn(format!("msg-{}", i));
thread.complete_turn(format!("resp-{}", i));
@@ -457,7 +457,7 @@ mod tests {
async fn test_compact_truncate_empty_turns() {
let llm = Arc::new(StubLlm::new("unused"));
let compactor = make_compactor(llm);
let mut thread = Thread::new(Uuid::new_v4());
let mut thread = Thread::new(Uuid::new_v4(), None);
assert!(thread.turns.is_empty());
let result = compactor
@@ -698,7 +698,7 @@ mod tests {
#[test]
fn test_format_turns_for_storage_with_tool_calls() {
let mut thread = Thread::new(Uuid::new_v4());
let mut thread = Thread::new(Uuid::new_v4(), None);
thread.start_turn("Search for X");
// Record a tool call on the current turn
if let Some(turn) = thread.turns.last_mut() {
@@ -719,7 +719,7 @@ mod tests {
#[test]
fn test_format_turns_for_storage_incomplete_turn() {
let mut thread = Thread::new(Uuid::new_v4());
let mut thread = Thread::new(Uuid::new_v4(), None);
thread.start_turn("In progress message");
// Don't complete the turn
+2 -2
View File
@@ -2299,7 +2299,7 @@ mod tests {
// Initialize a thread in the session so the loop can record tool calls.
let thread_id = {
let mut sess = session.lock().await;
sess.create_thread().id
sess.create_thread(Some("test")).id
};
let message = IncomingMessage::new("test", "test-user", "do something");
@@ -2412,7 +2412,7 @@ mod tests {
let session = Arc::new(Mutex::new(Session::new("test-user")));
let thread_id = {
let mut sess = session.lock().await;
sess.create_thread().id
sess.create_thread(Some("test")).id
};
let message = IncomingMessage::new("test", "test-user", "keep calling tools");
+169 -54
View File
@@ -68,8 +68,8 @@ impl Session {
}
/// Create a new thread in this session.
pub fn create_thread(&mut self) -> &mut Thread {
let thread = Thread::new(self.id);
pub fn create_thread(&mut self, channel: Option<&str>) -> &mut Thread {
let thread = Thread::new(self.id, channel);
let thread_id = thread.id;
self.active_thread = Some(thread_id);
self.last_active_at = Utc::now();
@@ -87,9 +87,9 @@ impl Session {
}
/// Get or create the active thread.
pub fn get_or_create_thread(&mut self) -> &mut Thread {
pub fn get_or_create_thread(&mut self, channel: Option<&str>) -> &mut Thread {
match self.active_thread {
None => self.create_thread(),
None => self.create_thread(channel),
Some(id) => {
if self.threads.contains_key(&id) {
// Entry existence confirmed by contains_key above.
@@ -100,7 +100,7 @@ impl Session {
} else {
// Stale active_thread ID: create a new thread, which
// updates self.active_thread to the new thread's ID.
self.create_thread()
self.create_thread(channel)
}
}
}
@@ -225,6 +225,9 @@ pub struct Thread {
/// Messages queued while the thread was processing a turn.
#[serde(default, skip_serializing_if = "VecDeque::is_empty")]
pub pending_messages: VecDeque<String>,
/// Channel that created this thread (for approval authorization).
#[serde(default)]
pub source_channel: Option<String>,
}
/// Maximum number of messages that can be queued while a thread is processing.
@@ -233,9 +236,29 @@ 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) -> Self {
pub fn new(session_id: Uuid, source_channel: Option<&str>) -> Self {
let now = Utc::now();
Self {
id: Uuid::new_v4(),
@@ -248,11 +271,12 @@ impl Thread {
pending_approval: None,
pending_auth: None,
pending_messages: VecDeque::new(),
source_channel: source_channel.map(String::from),
}
}
/// Create a thread with a specific ID (for DB hydration).
pub fn with_id(id: Uuid, session_id: Uuid) -> Self {
pub fn with_id(id: Uuid, session_id: Uuid, source_channel: Option<&str>) -> Self {
let now = Utc::now();
Self {
id,
@@ -265,6 +289,7 @@ impl Thread {
pending_approval: None,
pending_auth: None,
pending_messages: VecDeque::new(),
source_channel: source_channel.map(String::from),
}
}
@@ -787,13 +812,13 @@ mod tests {
let mut session = Session::new("user-123");
assert!(session.active_thread.is_none());
session.create_thread();
session.create_thread(None);
assert!(session.active_thread.is_some());
}
#[test]
fn test_thread_turns() {
let mut thread = Thread::new(Uuid::new_v4());
let mut thread = Thread::new(Uuid::new_v4(), None);
thread.start_turn("Hello");
assert_eq!(thread.state, ThreadState::Processing);
@@ -806,7 +831,7 @@ mod tests {
#[test]
fn test_thread_messages() {
let mut thread = Thread::new(Uuid::new_v4());
let mut thread = Thread::new(Uuid::new_v4(), None);
thread.start_turn("First message");
thread.complete_turn("First response");
@@ -829,7 +854,7 @@ mod tests {
#[test]
fn test_restore_from_messages() {
let mut thread = Thread::new(Uuid::new_v4());
let mut thread = Thread::new(Uuid::new_v4(), None);
// First add some turns
thread.start_turn("Original message");
@@ -855,7 +880,7 @@ mod tests {
#[test]
fn test_restore_from_messages_incomplete_turn() {
let mut thread = Thread::new(Uuid::new_v4());
let mut thread = Thread::new(Uuid::new_v4(), None);
// Messages with incomplete last turn (no assistant response)
let messages = vec![
@@ -874,7 +899,7 @@ mod tests {
#[test]
fn test_enter_auth_mode() {
let before = Utc::now();
let mut thread = Thread::new(Uuid::new_v4());
let mut thread = Thread::new(Uuid::new_v4(), None);
assert!(thread.pending_auth.is_none());
thread.enter_auth_mode("telegram".to_string());
@@ -887,7 +912,7 @@ mod tests {
#[test]
fn test_take_pending_auth() {
let mut thread = Thread::new(Uuid::new_v4());
let mut thread = Thread::new(Uuid::new_v4(), None);
thread.enter_auth_mode("notion".to_string());
let pending = thread.take_pending_auth();
@@ -902,7 +927,7 @@ mod tests {
#[test]
fn test_pending_auth_serialization() {
let mut thread = Thread::new(Uuid::new_v4());
let mut thread = Thread::new(Uuid::new_v4(), None);
thread.enter_auth_mode("openai".to_string());
let json = serde_json::to_string(&thread).expect("should serialize");
@@ -932,7 +957,7 @@ mod tests {
#[test]
fn test_pending_auth_default_none() {
// Deserialization of old data without pending_auth should default to None
let mut thread = Thread::new(Uuid::new_v4());
let mut thread = Thread::new(Uuid::new_v4(), None);
thread.pending_auth = None;
let json = serde_json::to_string(&thread).expect("serialize");
@@ -946,7 +971,7 @@ mod tests {
fn test_thread_with_id() {
let specific_id = Uuid::new_v4();
let session_id = Uuid::new_v4();
let thread = Thread::with_id(specific_id, session_id);
let thread = Thread::with_id(specific_id, session_id, None);
assert_eq!(thread.id, specific_id);
assert_eq!(thread.session_id, session_id);
@@ -958,7 +983,7 @@ mod tests {
fn test_thread_with_id_restore_messages() {
let thread_id = Uuid::new_v4();
let session_id = Uuid::new_v4();
let mut thread = Thread::with_id(thread_id, session_id);
let mut thread = Thread::with_id(thread_id, session_id, None);
let messages = vec![
ChatMessage::user("Hello from DB"),
@@ -977,7 +1002,7 @@ mod tests {
#[test]
fn test_restore_from_messages_empty() {
let mut thread = Thread::new(Uuid::new_v4());
let mut thread = Thread::new(Uuid::new_v4(), None);
// Add a turn first, then restore with empty vec
thread.start_turn("hello");
@@ -993,7 +1018,7 @@ mod tests {
#[test]
fn test_restore_from_messages_only_assistant_messages() {
let mut thread = Thread::new(Uuid::new_v4());
let mut thread = Thread::new(Uuid::new_v4(), None);
// Only assistant messages (no user messages to anchor turns)
let messages = vec![
@@ -1010,7 +1035,7 @@ mod tests {
#[test]
fn test_restore_from_messages_multiple_user_messages_in_a_row() {
let mut thread = Thread::new(Uuid::new_v4());
let mut thread = Thread::new(Uuid::new_v4(), None);
// Two user messages with no assistant response between them
let messages = vec![
@@ -1037,8 +1062,8 @@ mod tests {
fn test_thread_switch() {
let mut session = Session::new("user-1");
let t1_id = session.create_thread().id;
let t2_id = session.create_thread().id;
let t1_id = session.create_thread(None).id;
let t2_id = session.create_thread(None).id;
// After creating two threads, active should be the last one
assert_eq!(session.active_thread, Some(t2_id));
@@ -1058,8 +1083,8 @@ mod tests {
fn test_get_or_create_thread_idempotent() {
let mut session = Session::new("user-1");
let tid1 = session.get_or_create_thread().id;
let tid2 = session.get_or_create_thread().id;
let tid1 = session.get_or_create_thread(None).id;
let tid2 = session.get_or_create_thread(None).id;
// Should return the same thread (not create a new one each time)
assert_eq!(tid1, tid2);
@@ -1068,7 +1093,7 @@ mod tests {
#[test]
fn test_truncate_turns() {
let mut thread = Thread::new(Uuid::new_v4());
let mut thread = Thread::new(Uuid::new_v4(), None);
for i in 0..5 {
thread.start_turn(format!("msg-{}", i));
@@ -1092,7 +1117,7 @@ mod tests {
#[test]
fn test_truncate_turns_noop_when_fewer() {
let mut thread = Thread::new(Uuid::new_v4());
let mut thread = Thread::new(Uuid::new_v4(), None);
thread.start_turn("only one");
thread.complete_turn("response");
@@ -1104,7 +1129,7 @@ mod tests {
#[test]
fn test_thread_interrupt_and_resume() {
let mut thread = Thread::new(Uuid::new_v4());
let mut thread = Thread::new(Uuid::new_v4(), None);
thread.start_turn("do something");
assert_eq!(thread.state, ThreadState::Processing);
@@ -1122,7 +1147,7 @@ mod tests {
#[test]
fn test_resume_only_from_interrupted() {
let mut thread = Thread::new(Uuid::new_v4());
let mut thread = Thread::new(Uuid::new_v4(), None);
// Idle thread: resume should be a no-op
assert_eq!(thread.state, ThreadState::Idle);
@@ -1138,7 +1163,7 @@ mod tests {
#[test]
fn test_turn_fail() {
let mut thread = Thread::new(Uuid::new_v4());
let mut thread = Thread::new(Uuid::new_v4(), None);
thread.start_turn("risky operation");
thread.fail_turn("connection timed out");
@@ -1154,7 +1179,7 @@ mod tests {
#[test]
fn test_messages_with_incomplete_last_turn() {
let mut thread = Thread::new(Uuid::new_v4());
let mut thread = Thread::new(Uuid::new_v4(), None);
thread.start_turn("first");
thread.complete_turn("first reply");
@@ -1170,7 +1195,7 @@ mod tests {
#[test]
fn test_thread_serialization_round_trip() {
let mut thread = Thread::new(Uuid::new_v4());
let mut thread = Thread::new(Uuid::new_v4(), None);
thread.start_turn("hello");
thread.complete_turn("world");
@@ -1188,7 +1213,7 @@ mod tests {
#[test]
fn test_session_serialization_round_trip() {
let mut session = Session::new("user-ser");
session.create_thread();
session.create_thread(None);
session.auto_approve_tool("echo");
let json = serde_json::to_string(&session).unwrap();
@@ -1226,7 +1251,7 @@ mod tests {
#[test]
fn test_turn_number_increments() {
let mut thread = Thread::new(Uuid::new_v4());
let mut thread = Thread::new(Uuid::new_v4(), None);
// Before any turns, turn_number() is 1 (1-indexed for display)
assert_eq!(thread.turn_number(), 1);
@@ -1241,7 +1266,7 @@ mod tests {
#[test]
fn test_complete_turn_on_empty_thread() {
let mut thread = Thread::new(Uuid::new_v4());
let mut thread = Thread::new(Uuid::new_v4(), None);
// Completing a turn when there are no turns should be a safe no-op
thread.complete_turn("phantom response");
@@ -1251,7 +1276,7 @@ mod tests {
#[test]
fn test_fail_turn_on_empty_thread() {
let mut thread = Thread::new(Uuid::new_v4());
let mut thread = Thread::new(Uuid::new_v4(), None);
// Failing a turn when there are no turns should be a safe no-op
thread.fail_turn("phantom error");
@@ -1261,7 +1286,7 @@ mod tests {
#[test]
fn test_pending_approval_flow() {
let mut thread = Thread::new(Uuid::new_v4());
let mut thread = Thread::new(Uuid::new_v4(), None);
let approval = PendingApproval {
request_id: Uuid::new_v4(),
@@ -1288,7 +1313,7 @@ mod tests {
#[test]
fn test_clear_pending_approval() {
let mut thread = Thread::new(Uuid::new_v4());
let mut thread = Thread::new(Uuid::new_v4(), None);
let approval = PendingApproval {
request_id: Uuid::new_v4(),
@@ -1317,7 +1342,7 @@ mod tests {
assert!(session.active_thread().is_none());
assert!(session.active_thread_mut().is_none());
let tid = session.create_thread().id;
let tid = session.create_thread(None).id;
assert!(session.active_thread().is_some());
assert_eq!(session.active_thread().unwrap().id, tid);
@@ -1334,7 +1359,7 @@ mod tests {
#[test]
fn test_messages_includes_tool_calls() {
let mut thread = Thread::new(Uuid::new_v4());
let mut thread = Thread::new(Uuid::new_v4(), None);
thread.start_turn("Search for X");
{
@@ -1366,7 +1391,7 @@ mod tests {
#[test]
fn test_messages_multiple_tool_calls_per_turn() {
let mut thread = Thread::new(Uuid::new_v4());
let mut thread = Thread::new(Uuid::new_v4(), None);
thread.start_turn("Do two things");
{
@@ -1393,7 +1418,7 @@ mod tests {
#[test]
fn test_restore_from_messages_with_tool_calls() {
let mut thread = Thread::new(Uuid::new_v4());
let mut thread = Thread::new(Uuid::new_v4(), None);
// Build a message sequence with tool calls
let tc = ToolCall {
@@ -1425,7 +1450,7 @@ mod tests {
#[test]
fn test_restore_from_messages_with_tool_error() {
let mut thread = Thread::new(Uuid::new_v4());
let mut thread = Thread::new(Uuid::new_v4(), None);
let tc = ToolCall {
id: "call_0".to_string(),
@@ -1456,7 +1481,7 @@ mod tests {
fn test_messages_round_trip_with_tools() {
// Build a thread with tool calls, get messages(), restore, get messages() again
// The two message sequences should be equivalent.
let mut thread = Thread::new(Uuid::new_v4());
let mut thread = Thread::new(Uuid::new_v4(), None);
thread.start_turn("Do search");
{
@@ -1469,7 +1494,7 @@ mod tests {
let messages_original = thread.messages();
// Restore into a new thread
let mut thread2 = Thread::new(Uuid::new_v4());
let mut thread2 = Thread::new(Uuid::new_v4(), None);
thread2.restore_from_messages(messages_original.clone());
let messages_restored = thread2.messages();
@@ -1491,7 +1516,7 @@ mod tests {
#[test]
fn test_restore_multi_stage_tool_calls() {
let mut thread = Thread::new(Uuid::new_v4());
let mut thread = Thread::new(Uuid::new_v4(), None);
let tc1 = ToolCall {
id: "call_a".to_string(),
@@ -1534,7 +1559,7 @@ mod tests {
#[test]
fn test_messages_truncates_large_tool_results() {
let mut thread = Thread::new(Uuid::new_v4());
let mut thread = Thread::new(Uuid::new_v4(), None);
thread.start_turn("Read big file");
{
@@ -1557,7 +1582,7 @@ mod tests {
#[test]
fn test_thread_message_queue() {
let mut thread = Thread::new(Uuid::new_v4());
let mut thread = Thread::new(Uuid::new_v4(), None);
// Queue is initially empty
assert!(thread.pending_messages.is_empty());
@@ -1593,7 +1618,7 @@ mod tests {
#[test]
fn test_thread_message_queue_serialization() {
let mut thread = Thread::new(Uuid::new_v4());
let mut thread = Thread::new(Uuid::new_v4(), None);
// Empty queue should not appear in serialization (skip_serializing_if)
let json = serde_json::to_string(&thread).unwrap();
@@ -1613,7 +1638,7 @@ mod tests {
#[test]
fn test_thread_message_queue_default_on_old_data() {
// Deserialization of old data without pending_messages should default to empty
let thread = Thread::new(Uuid::new_v4());
let thread = Thread::new(Uuid::new_v4(), None);
let json = serde_json::to_string(&thread).unwrap();
// The field is absent (skip_serializing_if), simulating old data
@@ -1624,7 +1649,7 @@ mod tests {
#[test]
fn test_interrupt_clears_pending_messages() {
let mut thread = Thread::new(Uuid::new_v4());
let mut thread = Thread::new(Uuid::new_v4(), None);
// Start a turn so there's something to interrupt
thread.start_turn("initial input");
@@ -1643,7 +1668,7 @@ mod tests {
#[test]
fn test_thread_state_idle_after_full_drain() {
let mut thread = Thread::new(Uuid::new_v4());
let mut thread = Thread::new(Uuid::new_v4(), None);
// Simulate a full drain cycle: start turn, queue messages, complete turn,
// then drain all queued messages as a single merged turn (#259).
@@ -1671,7 +1696,7 @@ mod tests {
#[test]
fn test_drain_pending_messages_merges_with_newlines() {
let mut thread = Thread::new(Uuid::new_v4());
let mut thread = Thread::new(Uuid::new_v4(), None);
// Empty queue returns None
assert!(thread.drain_pending_messages().is_none());
@@ -1700,7 +1725,7 @@ mod tests {
#[test]
fn test_requeue_drained_preserves_content_at_front() {
let mut thread = Thread::new(Uuid::new_v4());
let mut thread = Thread::new(Uuid::new_v4(), None);
// Re-queue into empty queue
thread.requeue_drained("failed batch".to_string());
@@ -1811,4 +1836,94 @@ mod tests {
&serde_json::json!("done")
);
}
#[test]
fn test_thread_new_stores_source_channel() {
let thread = Thread::new(Uuid::new_v4(), Some("telegram"));
assert_eq!(thread.source_channel.as_deref(), Some("telegram"));
}
#[test]
fn test_thread_new_none_channel() {
let thread = Thread::new(Uuid::new_v4(), None);
assert!(thread.source_channel.is_none());
}
#[test]
fn test_source_channel_serde_backcompat() {
// Simulate deserializing a Thread from older DB records that lack source_channel.
let thread = Thread::new(Uuid::new_v4(), Some("cli"));
let json = serde_json::to_string(&thread).unwrap();
// Remove the source_channel field to simulate an old record.
let mut value: serde_json::Value = serde_json::from_str(&json).unwrap();
value.as_object_mut().unwrap().remove("source_channel");
let old_json = serde_json::to_string(&value).unwrap();
let deserialized: Thread = serde_json::from_str(&old_json).unwrap();
assert!(
deserialized.source_channel.is_none(),
"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"
);
}
}
+28 -13
View File
@@ -200,7 +200,7 @@ impl SessionManager {
// Create new thread (always create a new one for a new key)
let thread_id = {
let mut sess = session.lock().await;
let thread = sess.create_thread();
let thread = sess.create_thread(Some(channel));
thread.id
};
@@ -476,7 +476,7 @@ mod tests {
let session = Arc::new(Mutex::new(Session::new("user-hydrate")));
{
let mut sess = session.lock().await;
let thread = Thread::with_id(thread_id, sess.id);
let thread = Thread::with_id(thread_id, sess.id, None);
sess.threads.insert(thread_id, thread);
sess.active_thread = Some(thread_id);
}
@@ -600,7 +600,7 @@ mod tests {
// Simulate hydration: create thread with a known UUID
{
let mut sess = session.lock().await;
let thread = Thread::with_id(known_uuid, session_id);
let thread = Thread::with_id(known_uuid, session_id, None);
sess.threads.insert(known_uuid, thread);
}
@@ -627,7 +627,7 @@ mod tests {
let session = Arc::new(Mutex::new(Session::new("user-idem")));
{
let mut sess = session.lock().await;
let thread = Thread::with_id(tid, sess.id);
let thread = Thread::with_id(tid, sess.id, None);
sess.threads.insert(tid, thread);
}
@@ -656,7 +656,7 @@ mod tests {
let session = Arc::new(Mutex::new(Session::new("user-undo")));
{
let mut sess = session.lock().await;
let thread = Thread::with_id(tid, sess.id);
let thread = Thread::with_id(tid, sess.id, None);
sess.threads.insert(tid, thread);
}
@@ -680,7 +680,7 @@ mod tests {
let session = Arc::new(Mutex::new(Session::new("user-new")));
{
let mut sess = session.lock().await;
let thread = Thread::with_id(tid, sess.id);
let thread = Thread::with_id(tid, sess.id, None);
sess.threads.insert(tid, thread);
}
@@ -788,7 +788,7 @@ mod tests {
let session = Arc::new(Mutex::new(Session::new("user-cross")));
{
let mut sess = session.lock().await;
let thread = Thread::with_id(tid, sess.id);
let thread = Thread::with_id(tid, sess.id, None);
sess.threads.insert(tid, thread);
}
@@ -815,7 +815,7 @@ mod tests {
let session = Arc::new(Mutex::new(Session::new("user-cross")));
{
let mut sess = session.lock().await;
let thread = Thread::with_id(tid, sess.id);
let thread = Thread::with_id(tid, sess.id, None);
sess.threads.insert(tid, thread);
}
@@ -966,7 +966,7 @@ mod tests {
let adopted_id = Uuid::new_v4();
{
let mut sess = session1.lock().await;
let thread = Thread::with_id(adopted_id, sess.id);
let thread = Thread::with_id(adopted_id, sess.id, None);
sess.threads.insert(adopted_id, thread);
}
// Resolve with the UUID as external_thread_id -- should adopt it
@@ -992,7 +992,7 @@ mod tests {
let session = Arc::new(Mutex::new(Session::new("user-direct")));
{
let mut sess = session.lock().await;
let thread = Thread::with_id(tid, sess.id);
let thread = Thread::with_id(tid, sess.id, None);
sess.threads.insert(tid, thread);
}
{
@@ -1030,7 +1030,7 @@ mod tests {
let known_id = Uuid::new_v4();
{
let mut sess = session.lock().await;
let thread = Thread::with_id(known_id, sess.id);
let thread = Thread::with_id(known_id, sess.id, None);
sess.threads.insert(known_id, thread);
}
@@ -1057,7 +1057,7 @@ mod tests {
let known_id = Uuid::new_v4();
{
let mut sess = session.lock().await;
let thread = Thread::with_id(known_id, sess.id);
let thread = Thread::with_id(known_id, sess.id, None);
sess.threads.insert(known_id, thread);
}
@@ -1081,7 +1081,7 @@ mod tests {
let known_id = Uuid::new_v4();
{
let mut sess = session.lock().await;
let thread = Thread::with_id(known_id, sess.id);
let thread = Thread::with_id(known_id, sess.id, None);
sess.threads.insert(known_id, thread);
}
@@ -1102,4 +1102,19 @@ mod tests {
"should NOT adopt UUID when external_thread_id is None"
);
}
#[tokio::test]
async fn test_thread_stores_source_channel() {
let manager = SessionManager::new();
let (session, thread_id) = manager.resolve_thread("user-1", "telegram", None).await;
let sess = session.lock().await;
let thread = sess.threads.get(&thread_id).unwrap();
assert_eq!(
thread.source_channel.as_deref(),
Some("telegram"),
"resolve_thread should store source_channel from the channel parameter"
);
}
}
+25 -9
View File
@@ -135,13 +135,29 @@ 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();
let session_id = {
let sess = session.lock().await;
sess.id
};
let mut thread = crate::agent::session::Thread::with_id(thread_uuid, session_id);
let mut thread = crate::agent::session::Thread::with_id(
thread_uuid,
session_id,
effective_source_channel,
);
if !chat_messages.is_empty() {
thread.restore_from_messages(chat_messages);
}
@@ -636,7 +652,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,
@@ -1781,7 +1797,7 @@ impl Agent {
.get_or_create_session(&message.user_id)
.await;
let mut sess = session.lock().await;
let thread = sess.create_thread();
let thread = sess.create_thread(Some(&message.channel));
let thread_id = thread.id;
Ok(SubmissionResult::ok_with_message(format!(
"New thread: {}",
@@ -2117,7 +2133,7 @@ mod tests {
let session_id = Uuid::new_v4();
let thread_id = Uuid::new_v4();
let mut thread = Thread::with_id(thread_id, session_id);
let mut thread = Thread::with_id(thread_id, session_id, None);
// Set thread to AwaitingApproval with a pending tool approval
let pending = PendingApproval {
@@ -2185,7 +2201,7 @@ mod tests {
use crate::agent::session::{MAX_PENDING_MESSAGES, Thread, ThreadState};
use uuid::Uuid;
let mut thread = Thread::new(Uuid::new_v4());
let mut thread = Thread::new(Uuid::new_v4(), None);
thread.start_turn("processing something");
assert_eq!(thread.state, ThreadState::Processing);
@@ -2211,7 +2227,7 @@ mod tests {
use crate::agent::session::{Thread, ThreadState};
use uuid::Uuid;
let mut thread = Thread::new(Uuid::new_v4());
let mut thread = Thread::new(Uuid::new_v4(), None);
thread.start_turn("processing");
thread.queue_message("pending-1".to_string());
@@ -2241,7 +2257,7 @@ mod tests {
let thread_id = Uuid::new_v4();
let session_id = Uuid::new_v4();
let mut thread = Thread::with_id(thread_id, session_id);
let mut thread = Thread::with_id(thread_id, session_id, None);
thread.start_turn("working");
assert_eq!(thread.state, ThreadState::Processing);
@@ -2268,7 +2284,7 @@ mod tests {
let thread_id = Uuid::new_v4();
let session_id = Uuid::new_v4();
let mut thread = Thread::with_id(thread_id, session_id);
let mut thread = Thread::with_id(thread_id, session_id, None);
thread.start_turn("working");
assert_eq!(thread.state, ThreadState::Processing);
+40
View File
@@ -34,6 +34,7 @@ pub async fn setup_wasm_channels(
secrets_store: &Option<Arc<dyn SecretsStore + Send + Sync>>,
extension_manager: Option<&Arc<ExtensionManager>>,
database: Option<&Arc<dyn Database>>,
registered_channel_names: &[String],
) -> Option<WasmChannelSetup> {
let runtime = match WasmChannelRuntime::new(WasmChannelRuntimeConfig::default()) {
Ok(r) => Arc::new(r),
@@ -71,7 +72,46 @@ 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.
// This list must cover every built-in channel name to prevent a WASM
// module from impersonating a built-in and satisfying same-channel
// approval checks.
const RESERVED_CHANNEL_NAMES: &[&str] = &[
"web",
"gateway",
"cli",
"repl",
"http",
"signal",
"slack-relay",
"secret_save",
];
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;
}
// Also reject any name that collides with an already-registered
// channel to prevent a WASM module from shadowing a channel that
// was registered earlier in the startup sequence.
if registered_channel_names
.iter()
.any(|n| n.to_ascii_lowercase() == name_lower)
{
tracing::warn!(
channel = %loaded.name(),
"Rejected WASM channel that collides with already-registered channel"
);
continue;
}
let (name, channel) = register_channel(
loaded,
config,
+8 -2
View File
@@ -574,7 +574,7 @@ pub async fn chat_new_thread_handler(
.await;
let (thread_id, info) = {
let mut sess = session.lock().await;
let thread = sess.create_thread();
let thread = sess.create_thread(Some("web"));
let id = thread.id;
let info = ThreadInfo {
id: thread.id,
@@ -593,7 +593,13 @@ 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) => {}
+2 -2
View File
@@ -2014,7 +2014,7 @@ async fn chat_new_thread_handler(
let session = session_manager.get_or_create_session(&user.user_id).await;
let (thread_id, info) = {
let mut sess = session.lock().await;
let thread = sess.create_thread();
let thread = sess.create_thread(Some("gateway"));
let id = thread.id;
let info = ThreadInfo {
id: thread.id,
@@ -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) => {}
+26 -3
View File
@@ -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)]
+1 -1
View File
@@ -981,7 +981,7 @@ mod tests {
assert!(alice_stats.last_active_at.is_some());
// Bob has no LLM calls so doesn't appear in summary stats
assert!(stats.iter().find(|s| s.user_id == "bob").is_none());
assert!(!stats.iter().any(|s| s.user_id == "bob"));
// Filter to single user
let alice_only = db.user_summary_stats(Some("alice")).await.unwrap();
+8
View File
@@ -785,6 +785,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;
"#,
),
];
+6
View File
@@ -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]
+11 -1
View File
@@ -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 ====================
+19 -3
View File
@@ -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)`.
+1
View File
@@ -449,6 +449,7 @@ async fn async_main() -> anyhow::Result<()> {
&components.secrets_store,
components.extension_manager.as_ref(),
components.db.as_ref(),
&channel_names,
)
.await;
+1 -1
View File
@@ -278,7 +278,7 @@ impl TenantScope {
thread_id: Option<&str>,
) -> Result<bool, DatabaseError> {
self.inner
.ensure_conversation(id, channel, &self.user_id, thread_id)
.ensure_conversation(id, channel, &self.user_id, thread_id, None)
.await
}
+3 -3
View File
@@ -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"
+7 -1
View File
@@ -48,7 +48,13 @@ 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"