From 5901451603d164a0e5814855161e5d5b05cc0cf0 Mon Sep 17 00:00:00 2001 From: Pierre LE GUEN <26087574+PierreLeGuen@users.noreply.github.com> Date: Tue, 24 Mar 2026 17:49:13 +0000 Subject: [PATCH] fix: remove stale stream_token gate from channel-relay activation (#1623) * fix: remove stale stream_token gate from channel-relay activation The relay architecture now uses instance-scoped bearer auth + webhook callbacks, not streaming. The `relay::stream_token` secret was never written by the current OAuth flow, so activation always failed with AuthRequired. Replace stream_token with the team_id setting (already stored by the OAuth callback) as the persistent "auth completed" marker: - is_relay_channel(): check team_id setting instead of stream_token secret - activate_channel_relay(): gate on team_id emptiness, not stream_token - removal flow: delete team_id setting + oauth_state secret - configure(): return empty allowed-secrets set (relay is OAuth-only) - configure_token(): return AuthRequired (no manual token entry) - list(): surface activation_error for relay channels (was hardcoded None) - Clean up stale comments referencing stream_token / "stored token" - Update test to match OAuth-only model (no secrets to pass) Made-with: Cursor * fix: address CI and review feedback - Fix pre-existing tunnel/mod.rs test compilation (missing GatewayConfig fields: memory_layers, user_tokens, workspace_read_scopes) - Log warnings on failed team_id/oauth_state cleanup during removal instead of silently ignoring errors (gemini review) - Also delete legacy stream_token secret during removal for backward compatibility with pre-webhook installs (codex review) Made-with: Cursor --- src/extensions/manager.rs | 102 ++++++++++++++++++-------------------- src/tunnel/mod.rs | 6 +++ 2 files changed, 55 insertions(+), 53 deletions(-) diff --git a/src/extensions/manager.rs b/src/extensions/manager.rs index 7da9e980..39654305 100644 --- a/src/extensions/manager.rs +++ b/src/extensions/manager.rs @@ -891,24 +891,27 @@ impl ExtensionManager { *self.relay_channel_manager.write().await = Some(channel_manager); } - /// Check if a channel name corresponds to a relay extension (has stored stream token + /// Check if a channel name corresponds to a relay extension (has stored team_id /// or is tracked in the installed relay extensions set). pub async fn is_relay_channel(&self, name: &str, user_id: &str) -> bool { // Check in-memory installed set first (supports no-store mode) if self.installed_relay_extensions.read().await.contains(name) { return true; } - // Then check for stored stream token - self.secrets - .exists(user_id, &format!("relay:{}:stream_token", name)) - .await - .unwrap_or(false) + // Check for stored team_id (persisted across restarts by the OAuth callback) + if let Some(ref store) = self.store { + let key = format!("relay:{}:team_id", name); + if let Ok(Some(v)) = store.get_setting(user_id, &key).await { + return v.as_str().is_some_and(|s| !s.is_empty()); + } + } + false } /// Restore persisted relay channels after startup. /// /// Loads the persisted active channel list, filters to relay types (those with - /// a stored stream token), and activates each via `activate_stored_relay()`. + /// a stored team_id setting), and activates each via `activate_stored_relay()`. /// Skips channels that are already active. /// /// Call this only after `set_relay_channel_manager()` or `set_channel_runtime()`. @@ -1428,9 +1431,11 @@ impl ExtensionManager { if kind_filter.is_none() || kind_filter == Some(ExtensionKind::ChannelRelay) { let installed = self.installed_relay_extensions.read().await; let active_names = self.active_channel_names.read().await; + let errors = self.activation_errors.read().await; for name in installed.iter() { let active = active_names.contains(name); - let has_token = self.is_relay_channel(name, user_id).await; + let authenticated = self.is_relay_channel(name, user_id).await; + let activation_error = errors.get(name).cloned(); let registry_entry = self .registry .get_with_kind(name, Some(ExtensionKind::ChannelRelay)) @@ -1443,13 +1448,13 @@ impl ExtensionManager { display_name, description, url: None, - authenticated: has_token, + authenticated, active, tools: Vec::new(), needs_setup: false, has_auth: true, installed: true, - activation_error: None, + activation_error, version: None, }); } @@ -1626,7 +1631,22 @@ impl ExtensionManager { self.persist_active_channels(user_id).await; self.activation_errors.write().await.remove(name); - // Remove stored stream token + // Remove stored team_id setting and clean up secrets + if let Some(ref store) = self.store + && let Err(e) = store + .delete_setting(user_id, &format!("relay:{}:team_id", name)) + .await + { + tracing::warn!(error = %e, name, "Failed to delete relay team_id setting on removal"); + } + if let Err(e) = self + .secrets + .delete(user_id, &format!("relay:{}:oauth_state", name)) + .await + { + tracing::warn!(error = %e, name, "Failed to delete relay oauth_state secret on removal"); + } + // Clean up legacy stream_token secret from pre-webhook installs let _ = self .secrets .delete(user_id, &format!("relay:{}:stream_token", name)) @@ -4181,13 +4201,13 @@ impl ExtensionManager { /// /// For Slack: initiates OAuth flow (redirect-based). /// For Telegram: accepts a bot token, registers it with channel-relay, - /// and stores the returned stream token. + /// and stores the team_id setting. async fn auth_channel_relay( &self, name: &str, user_id: &str, ) -> Result { - // Check if already authenticated (stream token exists) + // Check if already authenticated (team_id setting exists) if self.is_relay_channel(name, user_id).await { return Ok(AuthResult::authenticated(name, ExtensionKind::ChannelRelay)); } @@ -4233,19 +4253,9 @@ impl ExtensionManager { name: &str, user_id: &str, ) -> Result { - let token_key = format!("relay:{}:stream_token", name); let team_id_key = format!("relay:{}:team_id", name); - // Check if we have a stream token - // Verify auth: stream token must exist (even though we don't use it in this constructor path) - let _stream_token = match self.secrets.get_decrypted(user_id, &token_key).await { - Ok(secret) => secret.expose().to_string(), - Err(_) => { - return Err(ExtensionError::AuthRequired); - } - }; - - // Get team_id from settings + // Get team_id from settings (stored by the OAuth callback) let team_id = if let Some(ref store) = self.store { store .get_setting(user_id, &team_id_key) @@ -4258,6 +4268,10 @@ impl ExtensionManager { String::new() }; + if team_id.is_empty() { + return Err(ExtensionError::AuthRequired); + } + // Use relay config captured at startup let relay_config = self.relay_config()?; @@ -4367,11 +4381,11 @@ impl ExtensionManager { return Ok(ExtensionKind::WasmChannel); } - // Check channel-relay extensions (installed in memory or has stored token) + // Check channel-relay extensions (installed in memory or has stored team_id) if self.installed_relay_extensions.read().await.contains(name) { return Ok(ExtensionKind::ChannelRelay); } - // Also check if there's a stored stream token (persisted across restarts) + // Also check if there's a stored team_id setting (persisted across restarts) if self.is_relay_channel(name, user_id).await { return Ok(ExtensionKind::ChannelRelay); } @@ -4999,11 +5013,7 @@ impl ExtensionManager { names.insert(server.token_secret_name()); (names, Vec::new()) } - ExtensionKind::ChannelRelay => { - let mut names = std::collections::HashSet::new(); - names.insert(format!("relay:{}:stream_token", name)); - (names, Vec::new()) - } + ExtensionKind::ChannelRelay => (std::collections::HashSet::new(), Vec::new()), }; let allowed_fields: std::collections::HashSet = @@ -5434,7 +5444,9 @@ impl ExtensionManager { .map_err(|e| ExtensionError::NotInstalled(e.to_string()))?; server.token_secret_name() } - ExtensionKind::ChannelRelay => format!("relay:{}:stream_token", name), + ExtensionKind::ChannelRelay => { + return Err(ExtensionError::AuthRequired); + } }; let mut secrets = std::collections::HashMap::new(); @@ -7043,7 +7055,7 @@ mod tests { let dir = tempfile::tempdir().expect("temp dir"); let mgr = make_test_manager(None, dir.path().to_path_buf()); - // No token stored → not a relay channel + // No store configured, no team_id → not a relay channel assert!(!mgr.is_relay_channel("slack-relay", "test").await); } @@ -7862,19 +7874,13 @@ mod tests { .await .insert("test-relay".to_string()); - // configure() should dispatch to activate_channel_relay(), not - // activate_wasm_channel(). Both will fail (no runtime configured), - // but the error should be about relay config, not WASM channels. - let mut secrets = std::collections::HashMap::new(); - secrets.insert( - "relay:test-relay:stream_token".to_string(), - "tok".to_string(), - ); - + // configure() with empty secrets should dispatch to + // activate_channel_relay(), not activate_wasm_channel(). Relay auth + // is OAuth-only so there are no manual secrets to pass. let result = mgr .configure( "test-relay", - &secrets, + &std::collections::HashMap::new(), &std::collections::HashMap::new(), "test", ) @@ -7886,7 +7892,6 @@ mod tests { ); let result = result.unwrap(); - // Activation will fail (no relay config), but secrets should still be stored assert!( !result.activated, "activation should fail without relay config" @@ -7896,15 +7901,6 @@ mod tests { "error should not mention WASM — got: {}", result.message ); - - // Verify the secret was stored - assert!( - mgr.secrets - .exists("test", "relay:test-relay:stream_token") - .await - .unwrap_or(false), - "configure should have stored the relay stream token" - ); } #[test] fn test_validation_failed_is_distinct_error_variant() { diff --git a/src/tunnel/mod.rs b/src/tunnel/mod.rs index a6869eda..8719b6e1 100644 --- a/src/tunnel/mod.rs +++ b/src/tunnel/mod.rs @@ -429,6 +429,9 @@ mod tests { port: 3000, auth_token: None, user_id: "test".to_string(), + workspace_read_scopes: Vec::new(), + memory_layers: Vec::new(), + user_tokens: None, }); c } @@ -440,6 +443,9 @@ mod tests { port, auth_token: None, user_id: "test".to_string(), + workspace_read_scopes: Vec::new(), + memory_layers: Vec::new(), + user_tokens: None, }); c }