Skip to content

fix(agent): block thread_id-based context pollution across users #760

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
zmanian merged 3 commits into from
Mar 11, 2026
Merged

fix(agent): block thread_id-based context pollution across users #760

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
a83dc30
fix(agent): prevent forged thread UUID context/write contamination
pikaxinge Mar 9, 2026
25c2228
fix(agent): close thread_id race and reject forged UUID hydration
pikaxinge Mar 10, 2026
16979dc
fix(ci): satisfy clippy and fmt checks after rebase
pikaxinge Mar 10, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
4 changes: 3 additions & 1 deletion src/agent/agent_loop.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -803,7 +803,9 @@ impl Agent {
thread_id = %external_thread_id,
"Hydrating thread from DB"
);
self.maybe_hydrate_thread(message, external_thread_id).await;
if let Some(rejection) = self.maybe_hydrate_thread(message, external_thread_id).await {
return Ok(Some(format!("Error: {}", rejection)));
}
}

// Resolve session and thread
Expand Down
190 changes: 164 additions & 26 deletions src/agent/thread_ops.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -23,6 +23,14 @@ use crate::error::Error;
use crate::llm::{ChatMessage, ToolCall};
use crate::tools::redact_params;

const FORGED_THREAD_ID_ERROR: &str = "Invalid or unauthorized thread ID.";

fn requires_preexisting_uuid_thread(channel: &str) -> bool {
// Gateway-style channels send server-issued conversation UUIDs.
// Unknown UUIDs should be rejected instead of silently creating a new thread.
matches!(channel, "gateway" | "test")
}

impl Agent {
/// Hydrate a historical thread from DB into memory if not already present.
///
Expand All @@ -37,11 +45,11 @@ impl Agent {
&self,
message: &IncomingMessage,
external_thread_id: &str,
) {
) -> Option<String> {
// Only hydrate UUID-shaped thread IDs (web gateway uses UUIDs)
let thread_uuid = match Uuid::parse_str(external_thread_id) {
Ok(id) => id,
Err(_) => return,
Err(_) => return None,

Copilot AI Mar 11, 2026

Copy link

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

For gateway/test channels you appear to require server-issued UUID thread IDs, but invalid/non-UUID external_thread_id currently returns None (no rejection). That allows clients to bypass the preexisting-UUID requirement by sending a non-UUID thread_id, causing resolve_thread to create a new thread keyed by that string. Consider returning the same safe rejection message when Uuid::parse_str fails and requires_preexisting_uuid_thread(&message.channel) is true.

Suggested change
Err(_) => return None,
Err(_) => {
// For gateway-style channels that require server-issued UUIDs,
// reject non-UUID / malformed thread IDs instead of silently
// letting a new thread be created for them.
if requires_preexisting_uuid_thread(&message.channel) {
return Some(FORGED_THREAD_ID_ERROR.to_string());
}
return None;
}

Copilot uses AI. Check for mistakes.
};

// Check if already in memory
Expand All @@ -52,7 +60,7 @@ impl Agent {
{
let sess = session.lock().await;
if sess.threads.contains_key(&thread_uuid) {
return;
return None;
}
}

Expand All @@ -61,6 +69,62 @@ impl Agent {
let msg_count;

if let Some(store) = self.store() {
// Never hydrate history from a conversation UUID that isn't owned
// by the current authenticated user.
let owned = match store
.conversation_belongs_to_user(thread_uuid, &message.user_id)
.await
Comment on lines 71 to +76

Copilot AI Mar 11, 2026

Copy link

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Hydration authorization checks only conversation_belongs_to_user(id, user_id) (no channel), but writes are later guarded by ensure_conversation(id, channel, user_id, ...) which enforces both user and channel. This mismatch can lead to a thread being hydrated from a different channel for the same user and then all persistence being rejected due to channel conflict. Consider validating (id, channel, user_id) consistently during hydration (or explicitly documenting/handling the cross-channel case).

Copilot uses AI. Check for mistakes.
{
Ok(v) => v,
Err(e) => {
tracing::warn!(
"Failed to verify conversation ownership for hydration {}: {}",
thread_uuid,
e
);
if requires_preexisting_uuid_thread(&message.channel) {
return Some(FORGED_THREAD_ID_ERROR.to_string());
}
return None;
}
};
if !owned {
let exists = match store.get_conversation_metadata(thread_uuid).await {
Ok(Some(_)) => true,
Ok(None) => false,
Err(e) => {
tracing::warn!(
"Failed to inspect conversation metadata for hydration {}: {}",
thread_uuid,
e
);
if requires_preexisting_uuid_thread(&message.channel) {
return Some(FORGED_THREAD_ID_ERROR.to_string());
}
return None;
}
};

if requires_preexisting_uuid_thread(&message.channel) {
tracing::warn!(
user = %message.user_id,
channel = %message.channel,
thread_id = %thread_uuid,
exists,
"Rejected message for unavailable thread id"
);
return Some(FORGED_THREAD_ID_ERROR.to_string());
}

tracing::warn!(
user = %message.user_id,
thread_id = %thread_uuid,
exists,
"Skipped hydration for thread id not owned by sender"
);
return None;
}

let db_messages = store
.list_conversation_messages(thread_uuid)
.await
Expand Down Expand Up @@ -104,6 +168,8 @@ impl Agent {
thread_uuid,
msg_count
);

None
}

pub(super) async fn process_user_input(
Expand Down Expand Up @@ -303,8 +369,13 @@ impl Agent {
thread_id = %thread_id,
"Persisting user message to DB"
);
self.persist_user_message(thread_id, &message.user_id, effective_content)
.await;
self.persist_user_message(
thread_id,
&message.channel,
&message.user_id,
effective_content,
)
.await;

tracing::debug!(
message_id = %message.id,
Expand Down Expand Up @@ -386,10 +457,21 @@ impl Agent {
.await;

// Persist tool calls then assistant response (user message already persisted at turn start)
self.persist_tool_calls(thread_id, &message.user_id, turn_number, &tool_calls)
.await;
self.persist_assistant_response(thread_id, &message.user_id, &response)
.await;
self.persist_tool_calls(
thread_id,
&message.channel,
&message.user_id,
turn_number,
&tool_calls,
)
.await;
self.persist_assistant_response(
thread_id,
&message.channel,
&message.user_id,
&response,
)
.await;

Ok(SubmissionResult::response(response))
}
Expand Down Expand Up @@ -423,13 +505,49 @@ impl Agent {
}
}

/// Ensure a thread UUID is writable for `(channel, user_id)`.
///
/// Returns `false` for foreign/unowned conversation IDs or DB errors.
async fn ensure_writable_conversation(
&self,
store: &Arc<dyn crate::db::Database>,
thread_id: Uuid,
channel: &str,
user_id: &str,
) -> bool {
match store
.ensure_conversation(thread_id, channel, user_id, None)
.await
{
Ok(true) => true,
Ok(false) => {
tracing::warn!(
user = %user_id,
channel = %channel,
thread_id = %thread_id,
"Rejected write for unavailable thread id"
);
false
}
Err(e) => {
tracing::warn!(
"Failed to ensure writable conversation {}: {}",
thread_id,
e
);
false
}
}
}

/// Persist the user message to the DB at turn start (before the agentic loop).
///
/// This ensures the user message is durable even if the process crashes
/// mid-response. Call this right after `thread.start_turn()`.
pub(super) async fn persist_user_message(
&self,
thread_id: Uuid,
channel: &str,
user_id: &str,
user_input: &str,
) {
Expand All @@ -438,11 +556,10 @@ impl Agent {
None => return,
};

if let Err(e) = store
.ensure_conversation(thread_id, "gateway", user_id, None)
if !self
.ensure_writable_conversation(&store, thread_id, channel, user_id)
.await
{
tracing::warn!("Failed to ensure conversation {}: {}", thread_id, e);
return;
}

Expand All @@ -462,6 +579,7 @@ impl Agent {
pub(super) async fn persist_assistant_response(
&self,
thread_id: Uuid,
channel: &str,
user_id: &str,
response: &str,
) {
Expand All @@ -470,11 +588,10 @@ impl Agent {
None => return,
};

if let Err(e) = store
.ensure_conversation(thread_id, "gateway", user_id, None)
if !self
.ensure_writable_conversation(&store, thread_id, channel, user_id)
.await
{
tracing::warn!("Failed to ensure conversation {}: {}", thread_id, e);
return;
}

Expand All @@ -494,6 +611,7 @@ impl Agent {
pub(super) async fn persist_tool_calls(
&self,
thread_id: Uuid,
channel: &str,
user_id: &str,
turn_number: usize,
tool_calls: &[crate::agent::session::TurnToolCall],
Expand Down Expand Up @@ -543,11 +661,10 @@ impl Agent {
}
};

if let Err(e) = store
.ensure_conversation(thread_id, "gateway", user_id, None)
if !self
.ensure_writable_conversation(&store, thread_id, channel, user_id)
.await
{
tracing::warn!("Failed to ensure conversation {}: {}", thread_id, e);
return;
}

Expand Down Expand Up @@ -1208,10 +1325,21 @@ impl Agent {
.map(|t| (t.turn_number, t.tool_calls.clone()))
.unwrap_or_default();
// User message already persisted at turn start; save tool calls then assistant response
self.persist_tool_calls(thread_id, &message.user_id, turn_number, &tool_calls)
.await;
self.persist_assistant_response(thread_id, &message.user_id, &response)
.await;
self.persist_tool_calls(
thread_id,
&message.channel,
&message.user_id,
turn_number,
&tool_calls,
)
.await;
self.persist_assistant_response(
thread_id,
&message.channel,
&message.user_id,
&response,
)
.await;
let _ = self
.channels
.send_status(
Expand Down Expand Up @@ -1264,8 +1392,13 @@ impl Agent {
thread.clear_pending_approval();
thread.complete_turn(&rejection);
// User message already persisted at turn start; save rejection response
self.persist_assistant_response(thread_id, &message.user_id, &rejection)
.await;
self.persist_assistant_response(
thread_id,
&message.channel,
&message.user_id,
&rejection,
)
.await;
}
}

Expand Down Expand Up @@ -1303,8 +1436,13 @@ impl Agent {
thread.enter_auth_mode(ext_name.clone());
thread.complete_turn(&instructions);
// User message already persisted at turn start; save auth instructions
self.persist_assistant_response(thread_id, &message.user_id, &instructions)
.await;
self.persist_assistant_response(
thread_id,
&message.channel,
&message.user_id,
&instructions,
)
.await;
}
}
let _ = self
Expand Down
10 changes: 8 additions & 2 deletions src/channels/web/handlers/chat.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -550,11 +550,17 @@ pub async fn chat_new_thread_handler(
// Persist the empty conversation row with thread_type metadata synchronously
// so that the subsequent loadThreads() call from the frontend sees it.
if let Some(ref store) = state.store {
if let Err(e) = store
match store
.ensure_conversation(thread_id, "gateway", &state.user_id, None)
.await
{
tracing::warn!("Failed to persist new thread: {}", e);
Ok(true) => {}
Ok(false) => tracing::warn!(
user = %state.user_id,
thread_id = %thread_id,
"Skipped persisting new thread due to ownership/channel conflict"
),
Err(e) => tracing::warn!("Failed to persist new thread: {}", e),
}
let metadata_val = serde_json::json!("thread");
if let Err(e) = store
Expand Down
10 changes: 8 additions & 2 deletions src/channels/web/server.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -1418,11 +1418,17 @@ async fn chat_new_thread_handler(
// Persist the empty conversation row with thread_type metadata synchronously
// so that the subsequent loadThreads() call from the frontend sees it.
if let Some(ref store) = state.store {
if let Err(e) = store
match store
.ensure_conversation(thread_id, "gateway", &state.user_id, None)
.await
{
tracing::warn!("Failed to persist new thread: {}", e);
Ok(true) => {}
Ok(false) => tracing::warn!(
user = %state.user_id,
thread_id = %thread_id,
"Skipped persisting new thread due to ownership/channel conflict"
),
Err(e) => tracing::warn!("Failed to persist new thread: {}", e),
}
let metadata_val = serde_json::json!("thread");
if let Err(e) = store
Expand Down
15 changes: 9 additions & 6 deletions src/db/libsql/conversations.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -67,20 +67,23 @@ impl ConversationStore for LibSqlBackend {
channel: &str,
user_id: &str,
thread_id: Option<&str>,
) -> Result<(), DatabaseError> {
) -> Result<bool, DatabaseError> {
let conn = self.connect().await?;
let now = fmt_ts(&Utc::now());
conn.execute(
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)
ON CONFLICT (id) DO UPDATE SET last_activity = ?5
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],
)
.await
.map_err(|e| DatabaseError::Query(e.to_string()))?;
Ok(())
.await
.map_err(|e| DatabaseError::Query(e.to_string()))?;
Ok(affected > 0)
}

async fn list_conversations_with_preview(
Expand Down
Loading
Loading