diff --git a/Cargo.lock b/Cargo.lock index 8c96241c..864042d9 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -84,6 +84,12 @@ version = "0.2.21" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "683d7910e743518b0e34f1186f92494becacb047c7b6bf616c96772180fef923" +[[package]] +name = "ambient-authority" +version = "0.0.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "e9d4ee0d472d1cd2e28c97dfa124b3d8d992e10eb0a035f33f5d12e3a177ba3b" + [[package]] name = "android_system_properties" version = "0.1.5" @@ -426,6 +432,84 @@ version = "1.11.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "b35204fbdc0b3f4446b89fc1ac2cf84a8a68971995d0bf2e925ec7cd960f9cb3" +[[package]] +name = "cap-fs-ext" +version = "3.4.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "d5528f85b1e134ae811704e41ef80930f56e795923f866813255bc342cc20654" +dependencies = [ + "cap-primitives", + "cap-std", + "io-lifetimes", + "windows-sys 0.59.0", +] + +[[package]] +name = "cap-net-ext" +version = "3.4.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "20a158160765c6a7d0d8c072a53d772e4cb243f38b04bfcf6b4939cfbe7482e7" +dependencies = [ + "cap-primitives", + "cap-std", + "rustix 1.1.3", + "smallvec", +] + +[[package]] +name = "cap-primitives" +version = "3.4.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "b6cf3aea8a5081171859ef57bc1606b1df6999df4f1110f8eef68b30098d1d3a" +dependencies = [ + "ambient-authority", + "fs-set-times", + "io-extras", + "io-lifetimes", + "ipnet", + "maybe-owned", + "rustix 1.1.3", + "rustix-linux-procfs", + "windows-sys 0.59.0", + "winx", +] + +[[package]] +name = "cap-rand" +version = "3.4.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "d8144c22e24bbcf26ade86cb6501a0916c46b7e4787abdb0045a467eb1645a1d" +dependencies = [ + "ambient-authority", + "rand 0.8.5", +] + +[[package]] +name = "cap-std" +version = "3.4.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "b6dc3090992a735d23219de5c204927163d922f42f575a0189b005c62d37549a" +dependencies = [ + "cap-primitives", + "io-extras", + "io-lifetimes", + "rustix 1.1.3", +] + +[[package]] +name = "cap-time-ext" +version = "3.4.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "def102506ce40c11710a9b16e614af0cde8e76ae51b1f48c04b8d79f4b671a80" +dependencies = [ + "ambient-authority", + "cap-primitives", + "iana-time-zone", + "once_cell", + "rustix 1.1.3", + "winx", +] + [[package]] name = "cassowary" version = "0.3.0" @@ -936,13 +1020,33 @@ dependencies = [ "dirs-sys-next", ] +[[package]] +name = "dirs" +version = "4.0.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ca3aa72a6f96ea37bbc5aa912f6788242832f75369bdfdadcb0e38423f100059" +dependencies = [ + "dirs-sys 0.3.7", +] + [[package]] name = "dirs" version = "6.0.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "c3e8aa94d75141228480295a7d0e7feb620b1a5ad9f12bc40be62411e38cce4e" dependencies = [ - "dirs-sys", + "dirs-sys 0.5.0", +] + +[[package]] +name = "dirs-sys" +version = "0.3.7" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "1b1d1d91c932ef41c0f2663aa8b0ca0342d444d842c06914aa0a7e352d0bada6" +dependencies = [ + "libc", + "redox_users 0.4.6", + "winapi", ] [[package]] @@ -1074,6 +1178,17 @@ version = "2.3.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "37909eebbb50d72f9059c3b6d82c0463f2ff062c9e95845c43a6c9c0355411be" +[[package]] +name = "fd-lock" +version = "4.0.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "0ce92ff622d6dadf7349484f42c93271a0d49b7cc4d466a936405bacbe10aa78" +dependencies = [ + "cfg-if", + "rustix 1.1.3", + "windows-sys 0.59.0", +] + [[package]] name = "filetime" version = "0.2.27" @@ -1112,6 +1227,17 @@ dependencies = [ "percent-encoding", ] +[[package]] +name = "fs-set-times" +version = "0.20.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "94e7099f6313ecacbe1256e8ff9d617b75d1bcb16a6fddef94866d225a01a14a" +dependencies = [ + "io-lifetimes", + "rustix 1.1.3", + "windows-sys 0.59.0", +] + [[package]] name = "funty" version = "2.0.0" @@ -1698,6 +1824,22 @@ dependencies = [ "syn 2.0.114", ] +[[package]] +name = "io-extras" +version = "0.18.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "2285ddfe3054097ef4b2fe909ef8c3bcd1ea52a8f0d274416caebeef39f04a65" +dependencies = [ + "io-lifetimes", + "windows-sys 0.59.0", +] + +[[package]] +name = "io-lifetimes" +version = "2.0.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "06432fb54d3be7964ecd3649233cddf80db2832f47fec34c01f65b3d9d774983" + [[package]] name = "ipnet" version = "2.11.0" @@ -1916,6 +2058,12 @@ version = "0.8.4" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "47e1ffaa40ddd1f3ed91f717a33c8c0ee23fff369e3aa8772b9605cc1d22f4c3" +[[package]] +name = "maybe-owned" +version = "0.3.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "4facc753ae494aeb6e3c22f839b158aebd4f9270f55cd3c79906c45476c47ab4" + [[package]] name = "md-5" version = "0.10.6" @@ -1973,7 +2121,7 @@ dependencies = [ "clap", "crossterm", "deadpool-postgres", - "dirs", + "dirs 6.0.0", "dotenvy", "futures", "hkdf", @@ -2007,6 +2155,7 @@ dependencies = [ "uuid", "wasmparser 0.220.1", "wasmtime", + "wasmtime-wasi", ] [[package]] @@ -2856,6 +3005,16 @@ dependencies = [ "windows-sys 0.61.2", ] +[[package]] +name = "rustix-linux-procfs" +version = "0.1.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "2fc84bf7e9aa16c4f2c758f27412dc9841341e16aa682d9c7ac308fe3ee12056" +dependencies = [ + "once_cell", + "rustix 1.1.3", +] + [[package]] name = "rustls" version = "0.23.36" @@ -3158,6 +3317,15 @@ dependencies = [ "lazy_static", ] +[[package]] +name = "shellexpand" +version = "2.1.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "7ccc8076840c4da029af4f87e4e8daeb0fca6b87bbb02e10cb60b791450e11e4" +dependencies = [ + "dirs 4.0.0", +] + [[package]] name = "shlex" version = "1.3.0" @@ -3360,6 +3528,22 @@ dependencies = [ "syn 2.0.114", ] +[[package]] +name = "system-interface" +version = "0.27.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "cc4592f674ce18521c2a81483873a49596655b179f71c5e05d10c1fe66c78745" +dependencies = [ + "bitflags 2.10.0", + "cap-fs-ext", + "cap-std", + "fd-lock", + "io-lifetimes", + "rustix 0.38.44", + "windows-sys 0.59.0", + "winx", +] + [[package]] name = "tap" version = "1.0.1" @@ -4383,6 +4567,36 @@ dependencies = [ "syn 2.0.114", ] +[[package]] +name = "wasmtime-wasi" +version = "28.0.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "1a8e04b9a4c68ad018b330a4f4914b82b01dc3582d715ce21a93564c7f26b19f" +dependencies = [ + "anyhow", + "async-trait", + "bitflags 2.10.0", + "bytes", + "cap-fs-ext", + "cap-net-ext", + "cap-rand", + "cap-std", + "cap-time-ext", + "fs-set-times", + "futures", + "io-extras", + "io-lifetimes", + "rustix 0.38.44", + "system-interface", + "thiserror 1.0.69", + "tokio", + "tracing", + "url", + "wasmtime", + "wiggle", + "windows-sys 0.59.0", +] + [[package]] name = "wasmtime-winch" version = "28.0.1" @@ -4412,6 +4626,15 @@ dependencies = [ "wit-parser", ] +[[package]] +name = "wast" +version = "35.0.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "2ef140f1b49946586078353a453a1d28ba90adfc54dde75710bc1931de204d68" +dependencies = [ + "leb128", +] + [[package]] name = "wast" version = "244.0.0" @@ -4431,7 +4654,7 @@ version = "1.244.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "bbf35b87ed352f9ab6cd0732abde5a67dd6153dfd02c493e61459218b19456fa" dependencies = [ - "wast", + "wast 244.0.0", ] [[package]] @@ -4474,6 +4697,48 @@ dependencies = [ "web-sys", ] +[[package]] +name = "wiggle" +version = "28.0.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "3b23e3dc273d1e35cab9f38a5f76487aeeedcfa6a3fb594e209ee7b6f8b41dcc" +dependencies = [ + "anyhow", + "async-trait", + "bitflags 2.10.0", + "thiserror 1.0.69", + "tracing", + "wasmtime", + "wiggle-macro", +] + +[[package]] +name = "wiggle-generate" +version = "28.0.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "8738c5a7ef3a9de0fae10f8b84091a2aa4e059d8fef23de202ab689812b6bc6e" +dependencies = [ + "anyhow", + "heck", + "proc-macro2", + "quote", + "shellexpand", + "syn 2.0.114", + "witx", +] + +[[package]] +name = "wiggle-macro" +version = "28.0.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "e882267ac583e013a38a5aaeb83a49b219456ba3aa6e6772440f7213b176e8ff" +dependencies = [ + "proc-macro2", + "quote", + "syn 2.0.114", + "wiggle-generate", +] + [[package]] name = "winapi" version = "0.3.9" @@ -4821,6 +5086,16 @@ dependencies = [ "memchr", ] +[[package]] +name = "winx" +version = "0.36.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "3f3fd376f71958b862e7afb20cfe5a22830e1963462f3a17f49d82a6c1d1f42d" +dependencies = [ + "bitflags 2.10.0", + "windows-sys 0.59.0", +] + [[package]] name = "wit-bindgen" version = "0.46.0" @@ -4845,6 +5120,18 @@ dependencies = [ "wasmparser 0.221.3", ] +[[package]] +name = "witx" +version = "0.9.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "e366f27a5cabcddb2706a78296a40b8fcc451e1a6aba2fc1d94b4a01bdaaef4b" +dependencies = [ + "anyhow", + "log", + "thiserror 1.0.69", + "wast 35.0.2", +] + [[package]] name = "writeable" version = "0.6.2" diff --git a/Cargo.toml b/Cargo.toml index 200bac58..8b05502f 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -79,6 +79,7 @@ pgvector = { version = "0.4", features = ["postgres"] } # WASM sandbox for untrusted tool execution wasmtime = { version = "28", features = ["component-model"] } +wasmtime-wasi = "28" # WASI support for component model wasmparser = "0.220" # WASM binary parsing for validation # Cryptography for secrets management diff --git a/channels-src/telegram/src/lib.rs b/channels-src/telegram/src/lib.rs index e4d97f88..013f37e7 100644 --- a/channels-src/telegram/src/lib.rs +++ b/channels-src/telegram/src/lib.rs @@ -154,6 +154,9 @@ struct SentMessage { message_id: i64, } +/// Workspace path for storing polling state. +const POLLING_STATE_PATH: &str = "state/last_update_id"; + // ============================================================================ // Channel Metadata // ============================================================================ @@ -186,12 +189,23 @@ struct TelegramConfig { respond_to_all_group_messages: bool, /// Whether to use polling instead of webhooks. + /// Automatically disabled if tunnel_url is set. #[serde(default)] polling_enabled: bool, /// Polling interval in milliseconds (if polling enabled). #[serde(default = "default_poll_interval")] poll_interval_ms: u32, + + /// Public tunnel URL for webhook mode (e.g., "https://abc123.ngrok.io"). + /// When set, webhook mode is enabled and polling is disabled. + #[serde(default)] + tunnel_url: Option, + + /// Secret token for webhook validation. + /// Telegram will include this in the X-Telegram-Bot-Api-Secret-Token header. + #[serde(default)] + webhook_secret: Option, } fn default_poll_interval() -> u32 { @@ -218,8 +232,29 @@ impl Guest for TelegramChannel { ); } - // Configure polling if enabled - let poll = if config.polling_enabled { + // Determine mode: webhook or polling + // Webhook mode is enabled if tunnel_url is set, which disables polling + let webhook_mode = config.tunnel_url.is_some(); + + if webhook_mode { + channel_host::log( + channel_host::LogLevel::Info, + "Webhook mode enabled (polling disabled)", + ); + if let Some(ref url) = config.tunnel_url { + channel_host::log( + channel_host::LogLevel::Info, + &format!("Tunnel URL: {}", url), + ); + } + } + + // Configure polling only if not in webhook mode and polling is enabled + let poll = if !webhook_mode && config.polling_enabled { + channel_host::log( + channel_host::LogLevel::Info, + &format!("Polling enabled (interval: {}ms)", config.poll_interval_ms.max(30000)), + ); Some(PollConfig { interval_ms: config.poll_interval_ms.max(30000), // Enforce minimum enabled: true, @@ -228,18 +263,37 @@ impl Guest for TelegramChannel { None }; + // Webhook secret validation is handled by the host (X-Telegram-Bot-Api-Secret-Token header) + // The require_secret flag tells the host to validate the secret_validated field + let require_secret = config.webhook_secret.is_some(); + Ok(ChannelConfig { display_name: "Telegram".to_string(), http_endpoints: vec![HttpEndpointConfig { path: "/webhook/telegram".to_string(), methods: vec!["POST".to_string()], - require_secret: false, // Telegram doesn't use signing secrets by default + require_secret, }], poll, }) } fn on_http_request(req: IncomingHttpRequest) -> OutgoingHttpResponse { + // Check if webhook secret validation passed (if required) + // The host validates X-Telegram-Bot-Api-Secret-Token header and sets secret_validated + // If require_secret was true in config but validation failed, secret_validated will be false + if !req.secret_validated { + // This means require_secret was set but the secret didn't match + // We still check the field even though the host should have already rejected invalid requests + // This is defense in depth + channel_host::log( + channel_host::LogLevel::Warn, + "Webhook request with invalid or missing secret token", + ); + // Return 401 but Telegram will keep retrying, so this is just for logging + // In practice, the host should reject these before they reach us + } + // Parse the request body as UTF-8 let body_str = match std::str::from_utf8(&req.body) { Ok(s) => s, @@ -269,12 +323,98 @@ impl Guest for TelegramChannel { } fn on_poll() { - // Polling mode: call getUpdates API - // For now, we focus on webhook mode. Polling can be added later. + // Read last offset from workspace storage + let offset = match channel_host::workspace_read(POLLING_STATE_PATH) { + Some(s) => s.parse::().unwrap_or(0), + None => 0, + }; + channel_host::log( channel_host::LogLevel::Debug, - "Polling tick (not implemented yet)", + &format!("Polling getUpdates with offset {}", offset), ); + + // Build getUpdates URL with parameters + // - offset: Identifier of the first update to be returned + // - timeout: Long polling timeout in seconds (Telegram recommends 30+) + // - allowed_updates: Only get message updates + let url = format!( + "https://api.telegram.org/bot{{TELEGRAM_BOT_TOKEN}}/getUpdates?offset={}&timeout=30&allowed_updates=[\"message\",\"edited_message\"]", + offset + ); + + let headers = serde_json::json!({}); + + let result = channel_host::http_request("GET", &url, &headers.to_string(), None); + + match result { + Ok(response) => { + if response.status != 200 { + let body_str = String::from_utf8_lossy(&response.body); + channel_host::log( + channel_host::LogLevel::Error, + &format!("getUpdates returned {}: {}", response.status, body_str), + ); + return; + } + + // Parse response + let api_response: Result>, _> = + serde_json::from_slice(&response.body); + + match api_response { + Ok(resp) if resp.ok => { + if let Some(updates) = resp.result { + let mut new_offset = offset; + + for update in updates { + // Track highest update_id for next poll + if update.update_id >= new_offset { + new_offset = update.update_id + 1; + } + + // Process the update (emits messages) + handle_update(update); + } + + // Save new offset if it changed + if new_offset != offset { + if let Err(e) = channel_host::workspace_write( + POLLING_STATE_PATH, + &new_offset.to_string(), + ) { + channel_host::log( + channel_host::LogLevel::Error, + &format!("Failed to save polling offset: {}", e), + ); + } + } + } + } + Ok(resp) => { + channel_host::log( + channel_host::LogLevel::Error, + &format!( + "Telegram API error: {}", + resp.description.unwrap_or_else(|| "unknown".to_string()) + ), + ); + } + Err(e) => { + channel_host::log( + channel_host::LogLevel::Error, + &format!("Failed to parse getUpdates response: {}", e), + ); + } + } + } + Err(e) => { + channel_host::log( + channel_host::LogLevel::Error, + &format!("getUpdates request failed: {}", e), + ); + } + } } fn on_respond(response: AgentResponse) -> Result<(), String> { @@ -320,14 +460,15 @@ impl Guest for TelegramChannel { // Parse Telegram response let api_response: TelegramApiResponse = - serde_json::from_slice(&http_response.body).map_err(|e| { - format!("Failed to parse Telegram response: {}", e) - })?; + serde_json::from_slice(&http_response.body) + .map_err(|e| format!("Failed to parse Telegram response: {}", e))?; if !api_response.ok { return Err(format!( "Telegram API error: {}", - api_response.description.unwrap_or_else(|| "unknown".to_string()) + api_response + .description + .unwrap_or_else(|| "unknown".to_string()) )); } @@ -347,7 +488,10 @@ impl Guest for TelegramChannel { } fn on_shutdown() { - channel_host::log(channel_host::LogLevel::Info, "Telegram channel shutting down"); + channel_host::log( + channel_host::LogLevel::Info, + "Telegram channel shutting down", + ); } } @@ -422,8 +566,7 @@ fn handle_message(message: TelegramMessage) { is_private, }; - let metadata_json = - serde_json::to_string(&metadata).unwrap_or_else(|_| "{}".to_string()); + let metadata_json = serde_json::to_string(&metadata).unwrap_or_else(|_| "{}".to_string()); // Clean the message text (strip bot mentions and commands) let cleaned_text = clean_message_text(&text); diff --git a/channels-src/telegram/telegram.wasm b/channels-src/telegram/telegram.wasm index be4c20b9..14791860 100644 Binary files a/channels-src/telegram/telegram.wasm and b/channels-src/telegram/telegram.wasm differ diff --git a/src/channels/wasm/error.rs b/src/channels/wasm/error.rs index 76ed4d34..aa0f717a 100644 --- a/src/channels/wasm/error.rs +++ b/src/channels/wasm/error.rs @@ -74,6 +74,12 @@ pub enum WasmChannelError { #[error("Configuration error: {0}")] Config(String), + + #[error("Webhook registration failed for channel {name}: {reason}")] + WebhookRegistration { name: String, reason: String }, + + #[error("HTTP request error: {0}")] + HttpRequest(String), } impl From for WasmChannelError { diff --git a/src/channels/wasm/loader.rs b/src/channels/wasm/loader.rs index 2d91e288..fc2df29d 100644 --- a/src/channels/wasm/loader.rs +++ b/src/channels/wasm/loader.rs @@ -59,7 +59,28 @@ impl WasmChannelLoader { let cap_file = ChannelCapabilitiesFile::from_bytes(&cap_bytes) .map_err(|e| WasmChannelError::InvalidCapabilities(e.to_string()))?; + // Debug: log raw capabilities + tracing::debug!( + channel = name, + raw_capabilities = ?cap_file.capabilities, + "Parsed capabilities file" + ); + let caps = cap_file.to_capabilities(); + + // Debug: log resulting capabilities + tracing::info!( + channel = name, + http_allowed = caps.tool_capabilities.http.is_some(), + http_allowlist_count = caps + .tool_capabilities + .http + .as_ref() + .map(|h| h.allowlist.len()) + .unwrap_or(0), + "Channel capabilities loaded" + ); + let config = cap_file.config_json(); let desc = cap_file.description.clone(); diff --git a/src/channels/wasm/mod.rs b/src/channels/wasm/mod.rs index 757969f9..55fafafd 100644 --- a/src/channels/wasm/mod.rs +++ b/src/channels/wasm/mod.rs @@ -99,4 +99,4 @@ pub use router::{ }; pub use runtime::{PreparedChannelModule, WasmChannelRuntime, WasmChannelRuntimeConfig}; pub use schema::{ChannelCapabilitiesFile, ChannelConfig}; -pub use wrapper::{HttpResponse, WasmChannel}; +pub use wrapper::{HttpResponse, SharedWasmChannel, WasmChannel}; diff --git a/src/channels/wasm/router.rs b/src/channels/wasm/router.rs index e7d66f6a..609ea9e9 100644 --- a/src/channels/wasm/router.rs +++ b/src/channels/wasm/router.rs @@ -180,7 +180,6 @@ async fn health_handler(State(state): State) -> impl IntoResponse { } /// Generic webhook handler that routes to the appropriate WASM channel. -#[allow(dead_code)] async fn webhook_handler( State(state): State, method: Method, @@ -191,10 +190,21 @@ async fn webhook_handler( ) -> impl IntoResponse { let full_path = format!("/webhook/{}", path); + tracing::info!( + method = %method, + path = %full_path, + body_len = body.len(), + "Webhook request received" + ); + // Find the channel for this path let channel = match state.router.get_channel_for_path(&full_path).await { Some(c) => c, None => { + tracing::warn!( + path = %full_path, + "No channel registered for webhook path" + ); return ( StatusCode::NOT_FOUND, Json(serde_json::json!({ @@ -205,21 +215,48 @@ async fn webhook_handler( } }; + tracing::info!( + channel = %channel.channel_name(), + "Found channel for webhook" + ); + let channel_name = channel.channel_name(); // Check if secret is required if state.router.requires_secret(channel_name).await { // Try to get secret from query param or header - let provided_secret = query.get("secret").cloned().or_else(|| { - headers - .get("X-Webhook-Secret") - .and_then(|v| v.to_str().ok()) - .map(|s| s.to_string()) - }); + // Telegram uses X-Telegram-Bot-Api-Secret-Token header + let provided_secret = query + .get("secret") + .cloned() + .or_else(|| { + headers + .get("X-Telegram-Bot-Api-Secret-Token") + .and_then(|v| v.to_str().ok()) + .map(|s| s.to_string()) + }) + .or_else(|| { + // Fallback to generic header + headers + .get("X-Webhook-Secret") + .and_then(|v| v.to_str().ok()) + .map(|s| s.to_string()) + }); + + tracing::debug!( + channel = %channel_name, + has_provided_secret = provided_secret.is_some(), + provided_secret_len = provided_secret.as_ref().map(|s| s.len()), + "Checking webhook secret" + ); match provided_secret { Some(secret) => { if !state.router.validate_secret(channel_name, &secret).await { + tracing::warn!( + channel = %channel_name, + "Webhook secret validation failed" + ); return ( StatusCode::UNAUTHORIZED, Json(serde_json::json!({ @@ -227,8 +264,13 @@ async fn webhook_handler( })), ); } + tracing::debug!(channel = %channel_name, "Webhook secret validated"); } None => { + tracing::warn!( + channel = %channel_name, + "Webhook secret required but not provided" + ); return ( StatusCode::UNAUTHORIZED, Json(serde_json::json!({ @@ -251,6 +293,13 @@ async fn webhook_handler( // Call the WASM channel let secret_validated = state.router.requires_secret(channel_name).await; + + tracing::info!( + channel = %channel_name, + secret_validated = secret_validated, + "Calling WASM channel on_http_request" + ); + match channel .call_on_http_request( method.as_str(), @@ -266,6 +315,13 @@ async fn webhook_handler( let status = StatusCode::from_u16(response.status).unwrap_or(StatusCode::INTERNAL_SERVER_ERROR); + tracing::info!( + channel = %channel_name, + status = %status, + body_len = response.body.len(), + "WASM channel on_http_request completed successfully" + ); + // Build response with headers let body_json: serde_json::Value = serde_json::from_slice(&response.body) .unwrap_or_else(|_| { @@ -296,25 +352,22 @@ async fn webhook_handler( /// Create an Axum router for WASM channel webhooks. /// /// This router can be merged with the existing HTTP channel router. -#[allow(dead_code)] pub fn create_wasm_channel_router(router: Arc) -> Router { let state = RouterState::new(router); Router::new() .route("/wasm-channels/health", get(health_handler)) // Catch-all for webhook paths - .route("/webhook/*path", get(webhook_handler)) - .route("/webhook/*path", post(webhook_handler)) + .route("/webhook/{*path}", get(webhook_handler)) + .route("/webhook/{*path}", post(webhook_handler)) .with_state(state) } /// HTTP server for WASM channel webhooks. -#[allow(dead_code)] pub struct WasmChannelServer { router: Arc, } -#[allow(dead_code)] impl WasmChannelServer { /// Create a new server. pub fn new(router: Arc) -> Self { diff --git a/src/channels/wasm/schema.rs b/src/channels/wasm/schema.rs index 228c0214..06876822 100644 --- a/src/channels/wasm/schema.rs +++ b/src/channels/wasm/schema.rs @@ -101,8 +101,10 @@ impl ChannelCapabilitiesFile { #[derive(Debug, Clone, Default, Serialize, Deserialize)] pub struct ChannelCapabilitiesSchema { /// Tool capabilities (HTTP, secrets, workspace_read). + /// Note: Using the struct directly (not Option) because #[serde(flatten)] + /// with Option doesn't work correctly when T has all-optional fields. #[serde(flatten)] - pub tool: Option, + pub tool: ToolCapabilitiesFile, /// Channel-specific capabilities. #[serde(default)] @@ -112,11 +114,7 @@ pub struct ChannelCapabilitiesSchema { impl ChannelCapabilitiesSchema { /// Convert to runtime ChannelCapabilities. pub fn to_channel_capabilities(&self, channel_name: &str) -> ChannelCapabilities { - let tool_caps = self - .tool - .as_ref() - .map(|t| t.to_capabilities()) - .unwrap_or_default(); + let tool_caps = self.tool.to_capabilities(); let mut caps = ChannelCapabilities::for_channel(channel_name).with_tool_capabilities(tool_caps); diff --git a/src/channels/wasm/wrapper.rs b/src/channels/wasm/wrapper.rs index 1996b7af..2bb1db40 100644 --- a/src/channels/wasm/wrapper.rs +++ b/src/channels/wasm/wrapper.rs @@ -37,7 +37,8 @@ use tokio::sync::{RwLock, mpsc, oneshot}; use tokio_stream::wrappers::ReceiverStream; use uuid::Uuid; use wasmtime::Store; -use wasmtime::component::{Component, Linker, Val}; +use wasmtime::component::{Component, Linker}; +use wasmtime_wasi::{ResourceTable, WasiCtx, WasiCtxBuilder, WasiView}; use crate::channels::wasm::capabilities::ChannelCapabilities; use crate::channels::wasm::error::WasmChannelError; @@ -50,19 +51,285 @@ use crate::error::ChannelError; use crate::tools::wasm::LogLevel; use crate::tools::wasm::WasmResourceLimiter; +// Generate component model bindings from the WIT file +wasmtime::component::bindgen!({ + path: "wit/channel.wit", + world: "sandboxed-channel", + async: false, + with: { + // Use our own store data type + }, +}); + /// Store data for WASM channel execution. /// -/// Contains the resource limiter and channel-specific host state. +/// Contains the resource limiter, channel-specific host state, and WASI context. struct ChannelStoreData { limiter: WasmResourceLimiter, host_state: ChannelHostState, + wasi: WasiCtx, + table: ResourceTable, + /// Injected credentials for URL substitution (e.g., bot tokens). + /// Keys are placeholder names like "TELEGRAM_BOT_TOKEN". + credentials: HashMap, } impl ChannelStoreData { - fn new(memory_limit: u64, channel_name: &str, capabilities: ChannelCapabilities) -> Self { + fn new( + memory_limit: u64, + channel_name: &str, + capabilities: ChannelCapabilities, + credentials: HashMap, + ) -> Self { + // Create a minimal WASI context (no filesystem, no env vars for security) + let wasi = WasiCtxBuilder::new().build(); + Self { limiter: WasmResourceLimiter::new(memory_limit), host_state: ChannelHostState::new(channel_name, capabilities), + wasi, + table: ResourceTable::new(), + credentials, + } + } + + /// Inject credentials into a URL by replacing placeholders. + /// + /// Replaces patterns like `{TELEGRAM_BOT_TOKEN}` with actual values from + /// the injected credentials map. This allows WASM channels to reference + /// credentials without ever seeing the actual values. + fn inject_credentials_into_url(&self, url: &str) -> String { + let mut result = url.to_string(); + + tracing::debug!( + url = %url, + credential_count = self.credentials.len(), + credential_names = ?self.credentials.keys().collect::>(), + "Injecting credentials into URL" + ); + + // Replace all known placeholders from the credentials map + for (name, value) in &self.credentials { + let placeholder = format!("{{{}}}", name); + if result.contains(&placeholder) { + tracing::debug!( + placeholder = %placeholder, + "Found and replacing credential placeholder" + ); + result = result.replace(&placeholder, value); + } + } + + // Check if any placeholders remain (indicates missing credential) + if result.contains('{') && result.contains('}') { + tracing::warn!( + original_url = %url, + result_url = %result, + "URL may contain unresolved placeholders" + ); + } + + result + } +} + +// Implement WasiView to provide WASI context and resource table +impl WasiView for ChannelStoreData { + fn ctx(&mut self) -> &mut WasiCtx { + &mut self.wasi + } + + fn table(&mut self) -> &mut ResourceTable { + &mut self.table + } +} + +// Implement the generated Host trait for channel-host interface +impl near::agent::channel_host::Host for ChannelStoreData { + fn log(&mut self, level: near::agent::channel_host::LogLevel, message: String) { + let log_level = match level { + near::agent::channel_host::LogLevel::Trace => LogLevel::Trace, + near::agent::channel_host::LogLevel::Debug => LogLevel::Debug, + near::agent::channel_host::LogLevel::Info => LogLevel::Info, + near::agent::channel_host::LogLevel::Warn => LogLevel::Warn, + near::agent::channel_host::LogLevel::Error => LogLevel::Error, + }; + let _ = self.host_state.log(log_level, message); + } + + fn now_millis(&mut self) -> u64 { + self.host_state.now_millis() + } + + fn workspace_read(&mut self, path: String) -> Option { + self.host_state.workspace_read(&path).ok().flatten() + } + + fn workspace_write(&mut self, path: String, content: String) -> Result<(), String> { + self.host_state + .workspace_write(&path, content) + .map_err(|e| e.to_string()) + } + + fn http_request( + &mut self, + method: String, + url: String, + headers_json: String, + body: Option>, + ) -> Result { + tracing::info!( + method = %method, + original_url = %url, + body_len = body.as_ref().map(|b| b.len()).unwrap_or(0), + "WASM http_request called" + ); + + // Inject credentials into URL (e.g., replace {TELEGRAM_BOT_TOKEN} with actual token) + let injected_url = self.inject_credentials_into_url(&url); + + // Log whether injection happened (without revealing the token) + let url_changed = injected_url != url; + tracing::info!( + url_changed = url_changed, + has_bot_token = injected_url.contains("/bot") && !injected_url.contains("{"), + "URL after credential injection" + ); + + // Check if HTTP is allowed for this URL + self.host_state + .check_http_allowed(&injected_url, &method) + .map_err(|e| { + tracing::error!(error = %e, "HTTP not allowed"); + format!("HTTP not allowed: {}", e) + })?; + + // Record the request for rate limiting + self.host_state.record_http_request().map_err(|e| { + tracing::error!(error = %e, "Rate limit exceeded"); + format!("Rate limit exceeded: {}", e) + })?; + + // Parse headers + let headers: std::collections::HashMap = + serde_json::from_str(&headers_json).unwrap_or_default(); + + tracing::debug!(header_count = headers.len(), "Parsed request headers"); + + let url = injected_url; + + // Make the HTTP request using blocking I/O + // We're already in a spawn_blocking context, so we can use block_on + let result = tokio::runtime::Handle::current().block_on(async { + let client = reqwest::Client::new(); + + let mut request = match method.to_uppercase().as_str() { + "GET" => client.get(&url), + "POST" => client.post(&url), + "PUT" => client.put(&url), + "DELETE" => client.delete(&url), + "PATCH" => client.patch(&url), + "HEAD" => client.head(&url), + _ => return Err(format!("Unsupported HTTP method: {}", method)), + }; + + // Add headers + for (key, value) in headers { + request = request.header(&key, &value); + } + + // Add body if present + if let Some(body_bytes) = body { + request = request.body(body_bytes); + } + + // Send request with timeout + let response = request + .timeout(std::time::Duration::from_secs(30)) + .send() + .await + .map_err(|e| format!("HTTP request failed: {}", e))?; + + let status = response.status().as_u16(); + let response_headers: std::collections::HashMap = response + .headers() + .iter() + .filter_map(|(k, v)| { + v.to_str() + .ok() + .map(|v| (k.as_str().to_string(), v.to_string())) + }) + .collect(); + let headers_json = serde_json::to_string(&response_headers).unwrap_or_default(); + let body = response + .bytes() + .await + .map_err(|e| format!("Failed to read response body: {}", e))? + .to_vec(); + + tracing::info!( + status = status, + body_len = body.len(), + "HTTP response received" + ); + + // Log response body for debugging (truncated) + if let Ok(body_str) = std::str::from_utf8(&body) { + let truncated = if body_str.len() > 500 { + format!("{}...", &body_str[..500]) + } else { + body_str.to_string() + }; + tracing::debug!(body = %truncated, "Response body"); + } + + Ok(near::agent::channel_host::HttpResponse { + status, + headers_json, + body, + }) + }); + + match &result { + Ok(resp) => { + tracing::info!(status = resp.status, "http_request completed successfully"); + } + Err(e) => { + tracing::error!(error = %e, "http_request failed"); + } + } + + result + } + + fn secret_exists(&mut self, name: String) -> bool { + self.host_state.secret_exists(&name) + } + + fn emit_message(&mut self, msg: near::agent::channel_host::EmittedMessage) { + tracing::info!( + user_id = %msg.user_id, + user_name = ?msg.user_name, + content_len = msg.content.len(), + "WASM emit_message called" + ); + + let mut emitted = EmittedMessage::new(msg.user_id.clone(), msg.content.clone()); + if let Some(name) = msg.user_name { + emitted = emitted.with_user_name(name); + } + if let Some(tid) = msg.thread_id { + emitted = emitted.with_thread_id(tid); + } + emitted = emitted.with_metadata(msg.metadata_json); + + match self.host_state.emit_message(emitted) { + Ok(()) => { + tracing::info!("Message emitted to host state successfully"); + } + Err(e) => { + tracing::error!(error = %e, "Failed to emit message to host state"); + } } } } @@ -102,6 +369,10 @@ pub struct WasmChannel { /// Registered HTTP endpoints. endpoints: RwLock>, + + /// Injected credentials for HTTP requests (e.g., bot tokens). + /// Keys are placeholder names like "TELEGRAM_BOT_TOKEN". + credentials: RwLock>, } impl WasmChannel { @@ -127,9 +398,23 @@ impl WasmChannel { rate_limiter: RwLock::new(rate_limiter), shutdown_tx: RwLock::new(None), endpoints: RwLock::new(Vec::new()), + credentials: RwLock::new(HashMap::new()), } } + /// Set a credential for URL injection. + pub async fn set_credential(&self, name: &str, value: String) { + self.credentials + .write() + .await + .insert(name.to_string(), value); + } + + /// Get a snapshot of credentials for use in callbacks. + pub async fn get_credentials(&self) -> HashMap { + self.credentials.read().await.clone() + } + /// Get the channel name. pub fn channel_name(&self) -> &str { &self.name @@ -145,155 +430,177 @@ impl WasmChannel { self.endpoints.read().await.clone() } - /// Add channel host functions to the linker. + /// Register a webhook URL with Telegram. /// - /// These functions are imported by the WASM channel module. - fn add_host_functions(linker: &mut Linker) -> Result<(), WasmChannelError> { - // host.log(level: log-level, message: string) - linker - .root() - .func_wrap( - "log", - |mut ctx: wasmtime::StoreContextMut<'_, ChannelStoreData>, - (level, message): (i32, String)| { - let log_level = match level { - 0 => LogLevel::Trace, - 1 => LogLevel::Debug, - 2 => LogLevel::Info, - 3 => LogLevel::Warn, - 4 => LogLevel::Error, - _ => LogLevel::Info, - }; - // Ignore errors from logging (rate limiting) - let _ = ctx.data_mut().host_state.log(log_level, message); - Ok(()) - }, - ) - .map_err(|e| WasmChannelError::Config(format!("Failed to add log function: {}", e)))?; + /// Called during channel startup if tunnel_url is configured. + /// This enables instant message delivery instead of polling. + pub async fn register_telegram_webhook( + &self, + tunnel_url: &str, + bot_token: &str, + secret_token: Option<&str>, + ) -> Result<(), WasmChannelError> { + let webhook_url = format!("{}/webhook/telegram", tunnel_url); - // host.now-millis() -> u64 - linker - .root() - .func_wrap( - "now-millis", - |ctx: wasmtime::StoreContextMut<'_, ChannelStoreData>, - (): ()| - -> anyhow::Result<(u64,)> { - Ok((ctx.data().host_state.now_millis(),)) - }, - ) - .map_err(|e| { - WasmChannelError::Config(format!("Failed to add now-millis function: {}", e)) - })?; + tracing::info!( + channel = %self.name, + webhook_url = %webhook_url, + "Registering Telegram webhook" + ); - // host.workspace-read(path: string) -> option - linker - .root() - .func_wrap( - "workspace-read", - |ctx: wasmtime::StoreContextMut<'_, ChannelStoreData>, - (path,): (String,)| - -> anyhow::Result<(Option,)> { - let result = ctx.data().host_state.workspace_read(&path).ok().flatten(); - Ok((result,)) - }, - ) - .map_err(|e| { - WasmChannelError::Config(format!("Failed to add workspace-read function: {}", e)) - })?; + // Build form parameters + let mut form_params = vec![ + ("url", webhook_url.as_str()), + ("allowed_updates", r#"["message","edited_message"]"#), + ]; - // host.workspace-write(path: string, content: string) -> result<_, string> - linker - .root() - .func_wrap( - "workspace-write", - |mut ctx: wasmtime::StoreContextMut<'_, ChannelStoreData>, - (path, content): (String, String)| - -> anyhow::Result<(Result<(), String>,)> { - let result = ctx - .data_mut() - .host_state - .workspace_write(&path, content) - .map_err(|e| e.to_string()); - Ok((result,)) - }, - ) - .map_err(|e| { - WasmChannelError::Config(format!("Failed to add workspace-write function: {}", e)) - })?; + let secret_owned: String; + if let Some(secret) = secret_token { + secret_owned = secret.to_string(); + form_params.push(("secret_token", &secret_owned)); + } - // host.emit-message(msg: emitted-message) - // The message is passed as a record with fields: user-id, user-name, content, thread-id, metadata-json - linker - .root() - .func_wrap( - "emit-message", - |mut ctx: wasmtime::StoreContextMut<'_, ChannelStoreData>, - (user_id, user_name, content, thread_id, metadata_json): ( - String, - Option, - String, - Option, - String, - )| { - let mut msg = EmittedMessage::new(user_id, content); - if let Some(name) = user_name { - msg = msg.with_user_name(name); - } - if let Some(tid) = thread_id { - msg = msg.with_thread_id(tid); - } - msg = msg.with_metadata(metadata_json); + let client = reqwest::Client::builder() + .timeout(std::time::Duration::from_secs(30)) + .build() + .map_err(|e| WasmChannelError::HttpRequest(e.to_string()))?; - // Ignore errors (rate limiting just drops messages) - let _ = ctx.data_mut().host_state.emit_message(msg); - Ok(()) - }, - ) - .map_err(|e| { - WasmChannelError::Config(format!("Failed to add emit-message function: {}", e)) - })?; + let response = client + .post(format!( + "https://api.telegram.org/bot{}/setWebhook", + bot_token + )) + .form(&form_params) + .send() + .await + .map_err(|e| WasmChannelError::HttpRequest(e.to_string()))?; - // host.secret-exists(name: string) -> bool - linker - .root() - .func_wrap( - "secret-exists", - |ctx: wasmtime::StoreContextMut<'_, ChannelStoreData>, - (name,): (String,)| - -> anyhow::Result<(bool,)> { - Ok((ctx.data().host_state.secret_exists(&name),)) - }, - ) - .map_err(|e| { - WasmChannelError::Config(format!("Failed to add secret-exists function: {}", e)) - })?; + if !response.status().is_success() { + let status = response.status(); + let body = response.text().await.unwrap_or_default(); + return Err(WasmChannelError::WebhookRegistration { + name: self.name.clone(), + reason: format!("HTTP {}: {}", status, body), + }); + } + + // Parse Telegram API response + let result: serde_json::Value = response + .json() + .await + .map_err(|e| WasmChannelError::HttpRequest(e.to_string()))?; + + if result["ok"].as_bool() != Some(true) { + let description = result["description"] + .as_str() + .unwrap_or("unknown error") + .to_string(); + return Err(WasmChannelError::WebhookRegistration { + name: self.name.clone(), + reason: description, + }); + } + + tracing::info!( + channel = %self.name, + webhook_url = %webhook_url, + "Telegram webhook registered successfully" + ); Ok(()) } - /// Execute a WASM callback synchronously (called from spawn_blocking). + /// Delete the webhook and switch back to polling mode. /// - /// This is the core execution logic shared by all callbacks. - fn execute_callback_sync( + /// Called during shutdown if webhook was registered. + pub async fn delete_telegram_webhook(&self, bot_token: &str) -> Result<(), WasmChannelError> { + tracing::info!( + channel = %self.name, + "Deleting Telegram webhook" + ); + + let client = reqwest::Client::builder() + .timeout(std::time::Duration::from_secs(30)) + .build() + .map_err(|e| WasmChannelError::HttpRequest(e.to_string()))?; + + let response = client + .post(format!( + "https://api.telegram.org/bot{}/deleteWebhook", + bot_token + )) + .send() + .await + .map_err(|e| WasmChannelError::HttpRequest(e.to_string()))?; + + if !response.status().is_success() { + let status = response.status(); + let body = response.text().await.unwrap_or_default(); + return Err(WasmChannelError::WebhookRegistration { + name: self.name.clone(), + reason: format!("HTTP {} (delete): {}", status, body), + }); + } + + let result: serde_json::Value = response + .json() + .await + .map_err(|e| WasmChannelError::HttpRequest(e.to_string()))?; + + if result["ok"].as_bool() != Some(true) { + let description = result["description"] + .as_str() + .unwrap_or("unknown error") + .to_string(); + return Err(WasmChannelError::WebhookRegistration { + name: self.name.clone(), + reason: format!("delete failed: {}", description), + }); + } + + tracing::info!( + channel = %self.name, + "Telegram webhook deleted" + ); + + Ok(()) + } + + /// Add channel host functions to the linker using generated bindings. + /// + /// Uses the wasmtime::component::bindgen! generated `add_to_linker` function + /// to properly register all host functions with correct component model signatures. + fn add_host_functions(linker: &mut Linker) -> Result<(), WasmChannelError> { + // Add WASI support (required by the component adapter) + wasmtime_wasi::add_to_linker_sync(linker).map_err(|e| { + WasmChannelError::Config(format!("Failed to add WASI functions: {}", e)) + })?; + + // Use the generated add_to_linker function from bindgen for our custom interface + near::agent::channel_host::add_to_linker(linker, |state| state).map_err(|e| { + WasmChannelError::Config(format!("Failed to add host functions: {}", e)) + })?; + + Ok(()) + } + + /// Create a fresh store configured for WASM execution. + fn create_store( runtime: &WasmChannelRuntime, prepared: &PreparedChannelModule, capabilities: &ChannelCapabilities, - export_name: &str, - build_args: F, - ) -> Result<(R, ChannelHostState), WasmChannelError> - where - F: FnOnce() -> ( - Vec, - Box Result>, - ), - { + credentials: HashMap, + ) -> Result, WasmChannelError> { let engine = runtime.engine(); let limits = &prepared.limits; // Create fresh store with channel state (NEAR pattern: fresh instance per call) - let store_data = - ChannelStoreData::new(limits.memory_bytes, &prepared.name, capabilities.clone()); + let store_data = ChannelStoreData::new( + limits.memory_bytes, + &prepared.name, + capabilities.clone(), + credentials, + ); let mut store = Store::new(engine, store_data); // Configure fuel if enabled @@ -310,6 +617,17 @@ impl WasmChannel { // Set up resource limiter store.limiter(|data| &mut data.limiter); + Ok(store) + } + + /// Instantiate the WASM component using generated bindings. + fn instantiate_component( + runtime: &WasmChannelRuntime, + prepared: &PreparedChannelModule, + store: &mut Store, + ) -> Result { + let engine = runtime.engine(); + // Compile the component (uses cached bytes) let component = Component::new(engine, prepared.component_bytes()) .map_err(|e| WasmChannelError::Compilation(e.to_string()))?; @@ -318,58 +636,44 @@ impl WasmChannel { let mut linker = Linker::new(engine); Self::add_host_functions(&mut linker)?; - // Instantiate the component - let instance = linker - .instantiate(&mut store, &component) + // Instantiate using the generated bindings + let instance = SandboxedChannel::instantiate(store, &component, &linker) .map_err(|e| WasmChannelError::Instantiation(e.to_string()))?; - // Get the export function - let func = instance - .get_func(&mut store, export_name) - .ok_or_else(|| WasmChannelError::MissingExport(export_name.to_string()))?; + Ok(instance) + } - // Build arguments and result extractor - let (args, extract_result) = build_args(); - - // Call the function - let mut results = vec![Val::Bool(false)]; // Placeholder - func.call(&mut store, &args, &mut results).map_err(|e| { - let error_str = e.to_string(); - if error_str.contains("out of fuel") { - WasmChannelError::FuelExhausted { - name: prepared.name.clone(), - limit: limits.fuel, - } - } else if error_str.contains("unreachable") { - WasmChannelError::Trapped { - name: prepared.name.clone(), - reason: "unreachable code executed".to_string(), - } - } else { - WasmChannelError::Trapped { - name: prepared.name.clone(), - reason: error_str, - } + /// Map WASM execution errors to our error types. + fn map_wasm_error(e: anyhow::Error, name: &str, fuel_limit: u64) -> WasmChannelError { + let error_str = e.to_string(); + if error_str.contains("out of fuel") { + WasmChannelError::FuelExhausted { + name: name.to_string(), + limit: fuel_limit, } - })?; + } else if error_str.contains("unreachable") { + WasmChannelError::Trapped { + name: name.to_string(), + reason: "unreachable code executed".to_string(), + } + } else { + WasmChannelError::Trapped { + name: name.to_string(), + reason: error_str, + } + } + } - // Post-call completion (cleanup) - func.post_return(&mut store) - .map_err(|e| WasmChannelError::Trapped { - name: prepared.name.clone(), - reason: format!("post_return failed: {}", e), - })?; - - // Extract result - let result = extract_result(&results[0])?; - - // Get host state with emitted messages and pending writes - let host_state = std::mem::replace( + /// Extract host state after callback execution. + fn extract_host_state( + store: &mut Store, + channel_name: &str, + capabilities: &ChannelCapabilities, + ) -> ChannelHostState { + std::mem::replace( &mut store.data_mut().host_state, - ChannelHostState::new(&prepared.name, capabilities.clone()), - ); - - Ok((result, host_state)) + ChannelHostState::new(channel_name, capabilities.clone()), + ) } /// Execute the on_start callback. @@ -394,20 +698,40 @@ impl WasmChannel { let capabilities = self.capabilities.clone(); let config_json = self.config_json.clone(); let timeout = self.runtime.config().callback_timeout; - let channel_name_for_error = self.name.clone(); + let channel_name = self.name.clone(); + let credentials = self.get_credentials().await; // Execute in blocking task with timeout let result = tokio::time::timeout(timeout, async move { tokio::task::spawn_blocking(move || { - Self::execute_callback_sync(&runtime, &prepared, &capabilities, "on-start", || { - let args = vec![Val::String(config_json)]; - let extract = Box::new(|result: &Val| extract_channel_config(result)); - (args, extract) - }) + let mut store = + Self::create_store(&runtime, &prepared, &capabilities, credentials)?; + let instance = Self::instantiate_component(&runtime, &prepared, &mut store)?; + + // Call on_start using the generated typed interface + let channel_iface = instance.near_agent_channel(); + let wasm_result = channel_iface + .call_on_start(&mut store, &config_json) + .map_err(|e| Self::map_wasm_error(e, &prepared.name, prepared.limits.fuel))?; + + // Convert the result + let config = match wasm_result { + Ok(wit_config) => convert_channel_config(wit_config), + Err(err_msg) => { + return Err(WasmChannelError::CallbackFailed { + name: prepared.name.clone(), + reason: err_msg, + }); + } + }; + + let host_state = + Self::extract_host_state(&mut store, &prepared.name, &capabilities); + Ok((config, host_state)) }) .await .map_err(|e| WasmChannelError::ExecutionPanicked { - name: channel_name_for_error.clone(), + name: channel_name.clone(), reason: e.to_string(), })? }) @@ -443,6 +767,33 @@ impl WasmChannel { body: &[u8], secret_validated: bool, ) -> Result { + tracing::info!( + channel = %self.name, + method = method, + path = path, + body_len = body.len(), + secret_validated = secret_validated, + "call_on_http_request invoked (webhook received)" + ); + + // Log the body for debugging (if it looks like JSON) + if let Ok(body_str) = std::str::from_utf8(body) { + let truncated = if body_str.len() > 1000 { + format!("{}...", &body_str[..1000]) + } else { + body_str.to_string() + }; + tracing::debug!(body = %truncated, "Webhook request body"); + } + + // Log credentials state (without values) + let creds = self.get_credentials().await; + tracing::info!( + credential_count = creds.len(), + credential_names = ?creds.keys().collect::>(), + "Credentials available for on_http_request" + ); + // If no WASM bytes, return 200 OK (for testing) if self.prepared.component_bytes.is_empty() { tracing::debug!( @@ -458,6 +809,7 @@ impl WasmChannel { let prepared = Arc::clone(&self.prepared); let capabilities = self.capabilities.clone(); let timeout = self.runtime.config().callback_timeout; + let credentials = self.get_credentials().await; // Prepare request data let method = method.to_string(); @@ -466,39 +818,39 @@ impl WasmChannel { let query_json = serde_json::to_string(&query).unwrap_or_default(); let body = body.to_vec(); - // Clone name for error handling before moving into closure - let channel_name_for_error = self.name.clone(); + let channel_name = self.name.clone(); // Execute in blocking task with timeout let result = tokio::time::timeout(timeout, async move { tokio::task::spawn_blocking(move || { - Self::execute_callback_sync( - &runtime, - &prepared, - &capabilities, - "on-http-request", - || { - // Build incoming-http-request record - let request = Val::Record(vec![ - ("method".to_string(), Val::String(method)), - ("path".to_string(), Val::String(path)), - ("headers-json".to_string(), Val::String(headers_json)), - ("query-json".to_string(), Val::String(query_json)), - ( - "body".to_string(), - Val::List(body.into_iter().map(Val::U8).collect()), - ), - ("secret-validated".to_string(), Val::Bool(secret_validated)), - ]); - let args = vec![request]; - let extract = Box::new(|result: &Val| extract_http_response(result)); - (args, extract) - }, - ) + let mut store = + Self::create_store(&runtime, &prepared, &capabilities, credentials)?; + let instance = Self::instantiate_component(&runtime, &prepared, &mut store)?; + + // Build the WIT request type + let wit_request = wit_channel::IncomingHttpRequest { + method, + path, + headers_json, + query_json, + body, + secret_validated, + }; + + // Call on_http_request using the generated typed interface + let channel_iface = instance.near_agent_channel(); + let wit_response = channel_iface + .call_on_http_request(&mut store, &wit_request) + .map_err(|e| Self::map_wasm_error(e, &prepared.name, prepared.limits.fuel))?; + + let response = convert_http_response(wit_response); + let host_state = + Self::extract_host_state(&mut store, &prepared.name, &capabilities); + Ok((response, host_state)) }) .await .map_err(|e| WasmChannelError::ExecutionPanicked { - name: channel_name_for_error.clone(), + name: channel_name.clone(), reason: e.to_string(), })? }) @@ -544,15 +896,24 @@ impl WasmChannel { let capabilities = self.capabilities.clone(); let timeout = self.runtime.config().callback_timeout; let channel_name = self.name.clone(); + let credentials = self.get_credentials().await; // Execute in blocking task with timeout let result = tokio::time::timeout(timeout, async move { tokio::task::spawn_blocking(move || { - Self::execute_callback_sync(&runtime, &prepared, &capabilities, "on-poll", || { - let args = vec![]; - let extract = Box::new(|_result: &Val| Ok(())); - (args, extract) - }) + let mut store = + Self::create_store(&runtime, &prepared, &capabilities, credentials)?; + let instance = Self::instantiate_component(&runtime, &prepared, &mut store)?; + + // Call on_poll using the generated typed interface + let channel_iface = instance.near_agent_channel(); + channel_iface + .call_on_poll(&mut store) + .map_err(|e| Self::map_wasm_error(e, &prepared.name, prepared.limits.fuel))?; + + let host_state = + Self::extract_host_state(&mut store, &prepared.name, &capabilities); + Ok(((), host_state)) }) .await .map_err(|e| WasmChannelError::ExecutionPanicked { @@ -593,6 +954,22 @@ impl WasmChannel { thread_id: Option<&str>, metadata_json: &str, ) -> Result<(), WasmChannelError> { + tracing::info!( + channel = %self.name, + message_id = %message_id, + content_len = content.len(), + thread_id = ?thread_id, + "call_on_respond invoked" + ); + + // Log credentials state (without values) + let creds = self.get_credentials().await; + tracing::info!( + credential_count = creds.len(), + credential_names = ?creds.keys().collect::>(), + "Credentials available for on_respond" + ); + // If no WASM bytes, do nothing (for testing) if self.prepared.component_bytes.is_empty() { tracing::debug!( @@ -608,6 +985,7 @@ impl WasmChannel { let capabilities = self.capabilities.clone(); let timeout = self.runtime.config().callback_timeout; let channel_name = self.name.clone(); + let credentials = self.get_credentials().await; // Prepare response data let message_id_str = message_id.to_string(); @@ -616,37 +994,62 @@ impl WasmChannel { let metadata_json = metadata_json.to_string(); // Execute in blocking task with timeout + tracing::info!(channel = %channel_name, "Starting on_respond WASM execution"); + let result = tokio::time::timeout(timeout, async move { tokio::task::spawn_blocking(move || { - Self::execute_callback_sync( - &runtime, - &prepared, - &capabilities, - "on-respond", - || { - // Build agent-response record - let response = Val::Record(vec![ - ("message-id".to_string(), Val::String(message_id_str)), - ("content".to_string(), Val::String(content)), - ( - "thread-id".to_string(), - match thread_id { - Some(tid) => Val::Option(Some(Box::new(Val::String(tid)))), - None => Val::Option(None), - }, - ), - ("metadata-json".to_string(), Val::String(metadata_json)), - ]); - let args = vec![response]; - let extract = Box::new(|result: &Val| extract_result_unit(result)); - (args, extract) - }, - ) + tracing::info!("Creating WASM store for on_respond"); + let mut store = + Self::create_store(&runtime, &prepared, &capabilities, credentials)?; + + tracing::info!("Instantiating WASM component for on_respond"); + let instance = Self::instantiate_component(&runtime, &prepared, &mut store)?; + + // Build the WIT response type + let wit_response = wit_channel::AgentResponse { + message_id: message_id_str, + content: content.clone(), + thread_id, + metadata_json, + }; + + tracing::info!( + content_preview = %if content.len() > 50 { &content[..50] } else { &content }, + "Calling WASM on_respond" + ); + + // Call on_respond using the generated typed interface + let channel_iface = instance.near_agent_channel(); + let wasm_result = channel_iface + .call_on_respond(&mut store, &wit_response) + .map_err(|e| { + tracing::error!(error = %e, "WASM on_respond call failed"); + Self::map_wasm_error(e, &prepared.name, prepared.limits.fuel) + })?; + + tracing::info!(wasm_result = ?wasm_result, "WASM on_respond returned"); + + // Check for WASM-level errors + if let Err(ref err_msg) = wasm_result { + tracing::error!(error = %err_msg, "WASM on_respond returned error"); + return Err(WasmChannelError::CallbackFailed { + name: prepared.name.clone(), + reason: err_msg.clone(), + }); + } + + let host_state = + Self::extract_host_state(&mut store, &prepared.name, &capabilities); + tracing::info!("on_respond WASM execution completed successfully"); + Ok(((), host_state)) }) .await - .map_err(|e| WasmChannelError::ExecutionPanicked { - name: channel_name.clone(), - reason: e.to_string(), + .map_err(|e| { + tracing::error!(error = %e, "spawn_blocking panicked"); + WasmChannelError::ExecutionPanicked { + name: channel_name.clone(), + reason: e.to_string(), + } })? }) .await; @@ -674,16 +1077,23 @@ impl WasmChannel { &self, messages: Vec, ) -> Result<(), WasmChannelError> { + tracing::info!( + channel = %self.name, + message_count = messages.len(), + "Processing emitted messages from WASM callback" + ); + if messages.is_empty() { + tracing::debug!(channel = %self.name, "No messages emitted"); return Ok(()); } let tx_guard = self.message_tx.read().await; let Some(tx) = tx_guard.as_ref() else { - tracing::warn!( + tracing::error!( channel = %self.name, count = messages.len(), - "Messages emitted but no sender available" + "Messages emitted but no sender available - channel may not be started!" ); return Ok(()); }; @@ -719,13 +1129,25 @@ impl WasmChannel { } // Send to stream + tracing::info!( + channel = %self.name, + user_id = %emitted.user_id, + content_len = emitted.content.len(), + "Sending emitted message to agent" + ); + if tx.send(msg).await.is_err() { - tracing::warn!( + tracing::error!( channel = %self.name, "Failed to send emitted message, channel closed" ); break; } + + tracing::info!( + channel = %self.name, + "Message successfully sent to agent queue" + ); } Ok(()) @@ -851,7 +1273,10 @@ impl Channel for WasmChannel { } // Call WASM on_respond - let metadata_json = serde_json::to_string(&response.metadata).unwrap_or_default(); + // IMPORTANT: Use the ORIGINAL message's metadata, not the response's metadata. + // The original metadata contains channel-specific routing info (e.g., Telegram chat_id) + // that the WASM channel needs to send the reply to the correct destination. + let metadata_json = serde_json::to_string(&msg.metadata).unwrap_or_default(); self.call_on_respond( msg.id, &response.content, @@ -915,241 +1340,103 @@ impl std::fmt::Debug for WasmChannel { } // ============================================================================ -// Value Extraction Helpers +// Shared Channel Wrapper // ============================================================================ -/// Extract ChannelConfig from a WIT result. -fn extract_channel_config(val: &Val) -> Result { - // Result is (ok: option, err: option) - match val { - Val::Result(result) => match result.as_ref() { - Ok(Some(config_val)) => extract_channel_config_inner(config_val), - Ok(None) => Err(WasmChannelError::InvalidResponse( - "on-start returned empty Ok".to_string(), - )), - Err(Some(err_val)) => { - if let Val::String(err) = err_val.as_ref() { - Err(WasmChannelError::CallbackFailed { - name: "channel".to_string(), - reason: err.clone(), - }) - } else { - Err(WasmChannelError::InvalidResponse( - "on-start error is not a string".to_string(), - )) - } - } - Err(None) => Err(WasmChannelError::InvalidResponse( - "on-start returned empty Err".to_string(), - )), - }, - // Fallback: try to parse as record directly (for simpler implementations) - Val::Record(_) => extract_channel_config_inner(val), - _ => Err(WasmChannelError::InvalidResponse(format!( - "Expected result or record, got {:?}", - std::mem::discriminant(val) - ))), +/// A wrapper around `Arc` that implements `Channel`. +/// +/// This allows sharing the same WasmChannel instance between: +/// - The WasmChannelRouter (for webhook handling) +/// - The ChannelManager (for message streaming and responses) +pub struct SharedWasmChannel { + inner: Arc, +} + +impl SharedWasmChannel { + /// Create a new shared wrapper. + pub fn new(channel: Arc) -> Self { + Self { inner: channel } + } + + /// Get the inner Arc. + pub fn inner(&self) -> &Arc { + &self.inner } } -/// Extract ChannelConfig from a channel-config record. -fn extract_channel_config_inner(val: &Val) -> Result { - match val { - Val::Record(fields) => { - let mut display_name = String::new(); - let mut http_endpoints = Vec::new(); - let mut poll = None; - - for (name, field_val) in fields { - match name.as_str() { - "display-name" => { - if let Val::String(s) = field_val { - display_name = s.clone(); - } - } - "http-endpoints" => { - if let Val::List(endpoints) = field_val { - for ep in endpoints { - if let Ok(endpoint) = extract_http_endpoint_config(ep) { - http_endpoints.push(endpoint); - } - } - } - } - "poll" => { - if let Val::Option(Some(poll_val)) = field_val { - poll = extract_poll_config(poll_val).ok(); - } - } - _ => {} - } - } - - Ok(ChannelConfig { - display_name, - http_endpoints, - poll, - }) - } - _ => Err(WasmChannelError::InvalidResponse( - "Expected record for channel-config".to_string(), - )), +impl std::fmt::Debug for SharedWasmChannel { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + f.debug_struct("SharedWasmChannel") + .field("inner", &self.inner) + .finish() } } -/// Extract HttpEndpointConfigSchema from a record. -fn extract_http_endpoint_config( - val: &Val, -) -> Result { - match val { - Val::Record(fields) => { - let mut path = String::new(); - let mut methods = Vec::new(); - let mut require_secret = false; +#[async_trait] +impl Channel for SharedWasmChannel { + fn name(&self) -> &str { + self.inner.name() + } - for (name, field_val) in fields { - match name.as_str() { - "path" => { - if let Val::String(s) = field_val { - path = s.clone(); - } - } - "methods" => { - if let Val::List(list) = field_val { - for item in list { - if let Val::String(s) = item { - methods.push(s.clone()); - } - } - } - } - "require-secret" => { - if let Val::Bool(b) = field_val { - require_secret = *b; - } - } - _ => {} - } - } + async fn start(&self) -> Result { + self.inner.start().await + } - Ok(crate::channels::wasm::schema::HttpEndpointConfigSchema { - path, - methods, - require_secret, - }) - } - _ => Err(WasmChannelError::InvalidResponse( - "Expected record for http-endpoint-config".to_string(), - )), + async fn respond( + &self, + msg: &IncomingMessage, + response: OutgoingResponse, + ) -> Result<(), ChannelError> { + self.inner.respond(msg, response).await + } + + async fn health_check(&self) -> Result<(), ChannelError> { + self.inner.health_check().await + } + + async fn shutdown(&self) -> Result<(), ChannelError> { + self.inner.shutdown().await } } -/// Extract PollConfigSchema from a record. -fn extract_poll_config( - val: &Val, -) -> Result { - match val { - Val::Record(fields) => { - let mut interval_ms = 30_000; - let mut enabled = false; +// ============================================================================ +// WIT Type Conversion Helpers +// ============================================================================ - for (name, field_val) in fields { - match name.as_str() { - "interval-ms" => { - if let Val::U32(n) = field_val { - interval_ms = *n; - } - } - "enabled" => { - if let Val::Bool(b) = field_val { - enabled = *b; - } - } - _ => {} - } - } +// Type aliases for the generated WIT types (exported interface) +use exports::near::agent::channel as wit_channel; - Ok(crate::channels::wasm::schema::PollConfigSchema { - interval_ms, - enabled, - }) - } - _ => Err(WasmChannelError::InvalidResponse( - "Expected record for poll-config".to_string(), - )), +/// Convert WIT-generated ChannelConfig to our internal type. +fn convert_channel_config(wit: wit_channel::ChannelConfig) -> ChannelConfig { + ChannelConfig { + display_name: wit.display_name, + http_endpoints: wit + .http_endpoints + .into_iter() + .map( + |ep| crate::channels::wasm::schema::HttpEndpointConfigSchema { + path: ep.path, + methods: ep.methods, + require_secret: ep.require_secret, + }, + ) + .collect(), + poll: wit + .poll + .map(|p| crate::channels::wasm::schema::PollConfigSchema { + interval_ms: p.interval_ms, + enabled: p.enabled, + }), } } -/// Extract HttpResponse from a WIT outgoing-http-response record. -fn extract_http_response(val: &Val) -> Result { - match val { - Val::Record(fields) => { - let mut status = 200u16; - let mut headers = HashMap::new(); - let mut body = Vec::new(); - - for (name, field_val) in fields { - match name.as_str() { - "status" => { - if let Val::U16(s) = field_val { - status = *s; - } - } - "headers-json" => { - if let Val::String(s) = field_val { - if let Ok(h) = serde_json::from_str::>(s) { - headers = h; - } - } - } - "body" => { - if let Val::List(bytes) = field_val { - body = bytes - .iter() - .filter_map(|v| if let Val::U8(b) = v { Some(*b) } else { None }) - .collect(); - } - } - _ => {} - } - } - - Ok(HttpResponse { - status, - headers, - body, - }) - } - _ => Err(WasmChannelError::InvalidResponse( - "Expected record for http-response".to_string(), - )), - } -} - -/// Extract unit result from a WIT result<_, string>. -fn extract_result_unit(val: &Val) -> Result<(), WasmChannelError> { - match val { - Val::Result(result) => match result.as_ref() { - Ok(_) => Ok(()), - Err(Some(err_val)) => { - if let Val::String(err) = err_val.as_ref() { - Err(WasmChannelError::CallbackFailed { - name: "channel".to_string(), - reason: err.clone(), - }) - } else { - Err(WasmChannelError::InvalidResponse( - "Error is not a string".to_string(), - )) - } - } - Err(None) => Err(WasmChannelError::InvalidResponse( - "Returned empty Err".to_string(), - )), - }, - // Unit return (for on-poll which returns nothing) - Val::Tuple(items) if items.is_empty() => Ok(()), - _ => Ok(()), // Treat anything else as success for unit-returning callbacks +/// Convert WIT-generated OutgoingHttpResponse to our HttpResponse type. +fn convert_http_response(wit: wit_channel::OutgoingHttpResponse) -> HttpResponse { + let headers = serde_json::from_str(&wit.headers_json).unwrap_or_default(); + HttpResponse { + status: wit.status, + headers, + body: wit.body, } } diff --git a/src/config.rs b/src/config.rs index 1d9e8d07..bdbd0200 100644 --- a/src/config.rs +++ b/src/config.rs @@ -13,6 +13,7 @@ pub struct Config { pub database: DatabaseConfig, pub llm: LlmConfig, pub embeddings: EmbeddingsConfig, + pub tunnel: TunnelConfig, pub channels: ChannelsConfig, pub agent: AgentConfig, pub safety: SafetyConfig, @@ -32,6 +33,7 @@ impl Config { database: DatabaseConfig::from_env()?, llm: LlmConfig::from_env()?, embeddings: EmbeddingsConfig::from_env()?, + tunnel: TunnelConfig::from_env()?, channels: ChannelsConfig::from_env()?, agent: AgentConfig::from_env()?, safety: SafetyConfig::from_env()?, @@ -43,6 +45,81 @@ impl Config { } } +/// Tunnel configuration for exposing the agent to the internet. +/// +/// Used by channels and tools that need public webhook endpoints. +/// The tunnel URL is shared across all channels (Telegram, Slack, etc.). +/// +/// # Security Notes +/// +/// **Webhook endpoints** (e.g., `/webhook/telegram`) should NOT use tunnel-level +/// authentication because webhook providers (Telegram, Slack, GitHub) need +/// unauthenticated access to POST updates. Security for webhooks comes from: +/// - Webhook signature verification (provider-specific secrets) +/// - IP allowlisting (if supported by provider) +/// +/// **Non-webhook endpoints** (admin APIs, health checks) CAN be protected using +/// tunnel provider features: +/// - ngrok: Basic Auth, OAuth, IP restrictions +/// - Cloudflare: Access policies, mTLS +/// +/// These protections are configured in the tunnel provider, not here. +/// +/// # Supported Providers +/// +/// - **ngrok**: `ngrok http 8080` -> `https://abc123.ngrok.io` +/// - **Cloudflare Tunnel**: `cloudflared tunnel --url http://localhost:8080` +/// - **localtunnel**: `lt --port 8080` +/// - Any service that provides a public HTTPS URL to localhost +#[derive(Debug, Clone, Default)] +pub struct TunnelConfig { + /// Public URL from tunnel provider (e.g., "https://abc123.ngrok.io"). + /// + /// When set, channels that support webhooks will register their endpoints + /// with this base URL instead of using polling. + pub public_url: Option, +} + +impl TunnelConfig { + fn from_env() -> Result { + // Priority: env var > settings file + let public_url = optional_env("TUNNEL_URL")?.or_else(|| { + crate::settings::Settings::load() + .tunnel + .public_url + .filter(|s| !s.is_empty()) + }); + + // Validate URL format if provided + if let Some(ref url) = public_url { + if !url.starts_with("https://") { + return Err(ConfigError::InvalidValue { + key: "TUNNEL_URL".to_string(), + message: "must start with https:// (webhooks require HTTPS)".to_string(), + }); + } + } + + Ok(Self { public_url }) + } + + /// Check if a tunnel is configured. + pub fn is_enabled(&self) -> bool { + self.public_url.is_some() + } + + /// Get the webhook URL for a given path. + /// + /// Returns `None` if no tunnel is configured. + pub fn webhook_url(&self, path: &str) -> Option { + self.public_url.as_ref().map(|base| { + let base = base.trim_end_matches('/'); + let path = path.trim_start_matches('/'); + format!("{}/{}", base, path) + }) + } +} + /// Database configuration. #[derive(Debug, Clone)] pub struct DatabaseConfig { @@ -231,12 +308,37 @@ fn default_session_path() -> PathBuf { pub struct ChannelsConfig { pub cli: CliConfig, pub http: Option, + pub telegram: TelegramChannelConfig, /// Directory containing WASM channel modules (default: ~/.near-agent/channels/). pub wasm_channels_dir: std::path::PathBuf, /// Whether WASM channels are enabled. pub wasm_channels_enabled: bool, } +/// Telegram channel configuration. +/// +/// The tunnel URL for webhook mode comes from the global `TunnelConfig`. +/// This config only contains Telegram-specific settings. +#[derive(Debug, Clone, Default)] +pub struct TelegramChannelConfig { + /// Secret token for webhook validation (optional but recommended). + /// + /// When set, Telegram will include this value in the + /// `X-Telegram-Bot-Api-Secret-Token` header of webhook requests. + /// The agent validates this header to ensure requests come from Telegram. + /// + /// Generate a secure random token (32+ characters recommended). + pub webhook_secret: Option, +} + +impl TelegramChannelConfig { + fn from_env() -> Result { + Ok(Self { + webhook_secret: optional_env("TELEGRAM_WEBHOOK_SECRET")?, + }) + } +} + #[derive(Debug, Clone)] pub struct CliConfig { pub enabled: bool, @@ -277,6 +379,7 @@ impl ChannelsConfig { enabled: cli_enabled, }, http, + telegram: TelegramChannelConfig::from_env()?, wasm_channels_dir: optional_env("WASM_CHANNELS_DIR")? .map(PathBuf::from) .unwrap_or_else(default_channels_dir), diff --git a/src/main.rs b/src/main.rs index 1d5f8cd4..532d80e9 100644 --- a/src/main.rs +++ b/src/main.rs @@ -9,13 +9,17 @@ use near_agent::{ agent::{Agent, AgentDeps}, channels::{ AppEvent, ChannelManager, HttpChannel, ReplChannel, TuiChannel, - wasm::{WasmChannelLoader, WasmChannelRuntime, WasmChannelRuntimeConfig}, + wasm::{ + RegisteredEndpoint, SharedWasmChannel, WasmChannelLoader, WasmChannelRouter, + WasmChannelRuntime, WasmChannelRuntimeConfig, WasmChannelServer, + }, }, cli::{Cli, Command, run_tool_command}, config::Config, history::Store, llm::{SessionConfig, create_llm_provider, create_session_manager}, safety::SafetyLayer, + secrets::{PostgresSecretsStore, SecretsCrypto, SecretsStore}, settings::Settings, setup::{SetupConfig, SetupWizard}, tools::{ @@ -327,6 +331,23 @@ async fn main() -> anyhow::Result<()> { } } + // Create secrets store if master key is configured (needed for Telegram webhook registration) + let secrets_store: Option> = + if let (Some(store), Some(master_key)) = (&store, config.secrets.master_key()) { + match SecretsCrypto::new(master_key.clone()) { + Ok(crypto) => Some(Arc::new(PostgresSecretsStore::new( + store.pool(), + Arc::new(crypto), + ))), + Err(e) => { + tracing::warn!("Failed to initialize secrets crypto: {}", e); + None + } + } + } else { + None + }; + // Load WASM channels if enabled if config.channels.wasm_channels_enabled && config.channels.wasm_channels_dir.exists() { match WasmChannelRuntime::new(WasmChannelRuntimeConfig::default()) { @@ -339,10 +360,140 @@ async fn main() -> anyhow::Result<()> { .await { Ok(results) => { + // Create router for WASM channel webhooks + let wasm_router = Arc::new(WasmChannelRouter::new()); + let mut has_webhook_channels = false; + for channel in results.loaded { - tracing::info!("Loaded WASM channel: {}", channel.channel_name()); - channels.add(Box::new(channel)); + let channel_name = channel.channel_name().to_string(); + tracing::info!("Loaded WASM channel: {}", channel_name); + + // Get webhook secret for this channel from secrets store + let webhook_secret = if let Some(ref secrets) = secrets_store { + let secret_name = format!("{}_webhook_secret", channel_name); + secrets + .get_decrypted("default", &secret_name) + .await + .ok() + .map(|s| s.expose().to_string()) + } else { + None + }; + + // Register channel with router for webhook handling + // Use known webhook path based on channel name + let webhook_path = format!("/webhook/{}", channel_name); + let endpoints = vec![RegisteredEndpoint { + channel_name: channel_name.clone(), + path: webhook_path.clone(), + methods: vec!["POST".to_string()], + require_secret: webhook_secret.is_some(), + }]; + + let channel_arc = Arc::new(channel); + + // Clone webhook_secret before moving it to register() + // We need it later for Telegram API registration + let webhook_secret_for_telegram = webhook_secret.clone(); + + tracing::info!( + channel = %channel_name, + has_webhook_secret = webhook_secret.is_some(), + "Registering channel with router" + ); + + wasm_router + .register(Arc::clone(&channel_arc), endpoints, webhook_secret) + .await; + has_webhook_channels = true; + + // Set up Telegram channel credentials and optionally register webhook + if channel_name == "telegram" { + if let Some(ref secrets) = secrets_store { + // Inject bot token for HTTP request URL substitution + // This is needed for both webhook and polling modes + match inject_telegram_credentials( + &channel_arc, + secrets.as_ref(), + ) + .await + { + Ok(()) => { + tracing::debug!("Telegram bot token injected"); + } + Err(e) => { + tracing::error!( + "Failed to inject Telegram credentials: {}", + e + ); + tracing::warn!( + "Telegram channel may not be able to send responses" + ); + } + } + + // Register webhook if tunnel URL is configured + // Use the SAME webhook_secret that the router expects (from secrets store) + if let Some(ref tunnel_url) = config.tunnel.public_url { + match register_telegram_webhook( + &channel_arc, + tunnel_url, + webhook_secret_for_telegram.as_deref(), + ) + .await + { + Ok(()) => { + tracing::info!( + "Telegram webhook registered at {}/webhook/telegram", + tunnel_url + ); + } + Err(e) => { + tracing::error!( + "Failed to register Telegram webhook: {}", + e + ); + tracing::warn!( + "Telegram will fall back to polling mode" + ); + } + } + } + } else { + tracing::warn!( + "Telegram channel loaded but secrets store not available" + ); + tracing::warn!( + "Set SECRETS_MASTER_KEY to enable Telegram bot token injection" + ); + } + } + + // Wrap in SharedWasmChannel for ChannelManager + // Both the router and ChannelManager share the same underlying channel + channels.add(Box::new(SharedWasmChannel::new(channel_arc))); } + + // Start WASM channel webhook server if we have channels with webhooks + if has_webhook_channels && config.tunnel.public_url.is_some() { + let server = WasmChannelServer::new(wasm_router); + let addr = std::net::SocketAddr::from(([0, 0, 0, 0], 8080)); + match server.start(addr).await { + Ok(_handle) => { + tracing::info!( + "WASM channel webhook server started on {}", + addr + ); + } + Err(e) => { + tracing::error!( + "Failed to start WASM channel webhook server: {}", + e + ); + } + } + } + for (path, err) in &results.errors { tracing::warn!( "Failed to load WASM channel {}: {}", @@ -407,3 +558,67 @@ async fn main() -> anyhow::Result<()> { tracing::info!("Agent shutdown complete"); Ok(()) } + +/// Inject Telegram bot token into the channel's credentials. +/// +/// This allows the WASM channel to use `{TELEGRAM_BOT_TOKEN}` in HTTP URLs +/// without ever seeing the actual token value. Required for both webhook +/// and polling modes to send responses. +async fn inject_telegram_credentials( + channel: &Arc, + secrets: &dyn SecretsStore, +) -> anyhow::Result<()> { + tracing::info!("Injecting Telegram bot token into channel credentials"); + + // Get bot token from secrets + let decrypted = secrets + .get_decrypted("default", "telegram_bot_token") + .await + .map_err(|e| { + tracing::error!(error = %e, "Failed to get telegram_bot_token from secrets"); + anyhow::anyhow!("Failed to get Telegram bot token: {}", e) + })?; + + let bot_token = decrypted.expose(); + let token_len = bot_token.len(); + + // Inject the token into the channel's credentials for URL substitution + channel + .set_credential("TELEGRAM_BOT_TOKEN", bot_token.to_string()) + .await; + + // Verify injection + let creds = channel.get_credentials().await; + tracing::info!( + token_length = token_len, + has_token = creds.contains_key("TELEGRAM_BOT_TOKEN"), + credential_count = creds.len(), + "Telegram bot token injected successfully" + ); + + Ok(()) +} + +/// Register Telegram webhook for instant message delivery. +/// +/// Calls the Telegram setWebhook API. Assumes credentials have already been +/// injected via `inject_telegram_credentials` (gets token from channel). +async fn register_telegram_webhook( + channel: &Arc, + tunnel_url: &str, + webhook_secret: Option<&str>, +) -> anyhow::Result<()> { + // Get the bot token via the public getter + let credentials = channel.get_credentials().await; + let bot_token = credentials.get("TELEGRAM_BOT_TOKEN").ok_or_else(|| { + anyhow::anyhow!("Bot token not injected - call inject_telegram_credentials first") + })?; + + // Register the webhook with Telegram API + channel + .register_telegram_webhook(tunnel_url, bot_token, webhook_secret) + .await + .map_err(|e| anyhow::anyhow!("Webhook registration failed: {}", e))?; + + Ok(()) +} diff --git a/src/settings.rs b/src/settings.rs index 3846bf3e..c52f1095 100644 --- a/src/settings.rs +++ b/src/settings.rs @@ -17,11 +17,25 @@ pub struct Settings { #[serde(default)] pub setup_completed: bool, + /// Tunnel configuration for exposing the agent to the internet. + #[serde(default)] + pub tunnel: TunnelSettings, + /// Channel configuration. #[serde(default)] pub channels: ChannelSettings, } +/// Tunnel settings for public webhook endpoints. +/// +/// The tunnel URL is shared across all channels that need webhooks. +#[derive(Debug, Clone, Serialize, Deserialize, Default)] +pub struct TunnelSettings { + /// Public URL from tunnel provider (e.g., "https://abc123.ngrok.io"). + #[serde(default)] + pub public_url: Option, +} + /// Channel-specific settings. #[derive(Debug, Clone, Serialize, Deserialize, Default)] pub struct ChannelSettings { diff --git a/src/setup/channels.rs b/src/setup/channels.rs index 095e09c5..b91d582d 100644 --- a/src/setup/channels.rs +++ b/src/setup/channels.rs @@ -13,8 +13,9 @@ use secrecy::{ExposeSecret, SecretString}; use serde::Deserialize; use crate::secrets::{CreateSecretParams, PostgresSecretsStore, SecretsCrypto, SecretsStore}; +use crate::settings::Settings; use crate::setup::prompts::{ - confirm, optional_input, print_error, print_info, print_success, secret_input, + confirm, input, optional_input, print_error, print_info, print_success, secret_input, }; /// Context for saving secrets during setup. @@ -58,6 +59,7 @@ impl SecretsContext { pub struct TelegramSetupResult { pub enabled: bool, pub bot_username: Option, + pub webhook_secret: Option, } /// Telegram Bot API response for getMe. @@ -94,9 +96,12 @@ pub async fn setup_telegram(secrets: &SecretsContext) -> Result Result { @@ -131,12 +140,107 @@ pub async fn setup_telegram(secrets: &SecretsContext) -> Result Result, String> { + // Check if already configured + let settings = Settings::load(); + if let Some(ref url) = settings.tunnel.public_url { + print_info(&format!("Existing tunnel configured: {}", url)); + if !confirm("Change tunnel configuration?", false).map_err(|e| e.to_string())? { + return Ok(Some(url.clone())); + } + } + + println!(); + print_info("Tunnel Configuration (for webhook endpoints):"); + print_info("A tunnel exposes your local agent to the internet, enabling:"); + print_info(" - Instant Telegram message delivery (instead of polling)"); + print_info(" - Future: Slack, Discord, GitHub webhooks"); + print_info(""); + print_info("Supported tunnel providers:"); + print_info(" - ngrok: ngrok http 8080"); + print_info(" - Cloudflare: cloudflared tunnel --url http://localhost:8080"); + print_info(" - localtunnel: lt --port 8080"); + print_info(""); + print_info("Security note: Webhook endpoints don't use tunnel-level auth."); + print_info("Security comes from provider-specific secrets (e.g., Telegram webhook secret)."); + println!(); + + if !confirm("Configure a tunnel?", false).map_err(|e| e.to_string())? { + return Ok(None); + } + + let tunnel_url = + input("Tunnel URL (e.g., https://abc123.ngrok.io)").map_err(|e| e.to_string())?; + + // Validate URL format + if !tunnel_url.starts_with("https://") { + print_error("URL must start with https:// (webhooks require HTTPS)"); + return Err("Invalid tunnel URL: must use HTTPS".to_string()); + } + + // Remove trailing slash if present + let tunnel_url = tunnel_url.trim_end_matches('/').to_string(); + + // Save to settings + let mut settings = Settings::load(); + settings.tunnel.public_url = Some(tunnel_url.clone()); + settings + .save() + .map_err(|e| format!("Failed to save settings: {}", e))?; + + print_success(&format!("Tunnel URL saved: {}", tunnel_url)); + print_info(""); + print_info("Make sure your tunnel is running before starting the agent."); + print_info("You can also set TUNNEL_URL environment variable to override."); + + Ok(Some(tunnel_url)) +} + +/// Set up Telegram webhook secret for signature validation. +/// +/// Returns the webhook secret if configured. +async fn setup_telegram_webhook_secret(secrets: &SecretsContext) -> Result, String> { + // Check if tunnel is configured + let settings = Settings::load(); + if settings.tunnel.public_url.is_none() { + print_info(""); + print_info("No tunnel configured. Telegram will use polling mode (30s+ delay)."); + print_info("Run setup again to configure a tunnel for instant delivery."); + return Ok(None); + } + + println!(); + print_info("Telegram Webhook Security:"); + print_info("A webhook secret adds an extra layer of security by validating"); + print_info("that requests actually come from Telegram's servers."); + + if !confirm("Generate a webhook secret?", true).map_err(|e| e.to_string())? { + return Ok(None); + } + + let secret = generate_webhook_secret(); + secrets + .save_secret( + "telegram_webhook_secret", + &SecretString::from(secret.clone()), + ) + .await?; + print_success("Webhook secret generated and saved"); + + Ok(Some(secret)) +} + /// Validate a Telegram bot token by calling the getMe API. /// /// Returns the bot's username if valid. diff --git a/src/setup/mod.rs b/src/setup/mod.rs index 3beb13c0..c082c56c 100644 --- a/src/setup/mod.rs +++ b/src/setup/mod.rs @@ -18,6 +18,8 @@ mod channels; mod prompts; mod wizard; -pub use channels::{SecretsContext, setup_http, setup_telegram, validate_telegram_token}; +pub use channels::{ + SecretsContext, setup_http, setup_telegram, setup_tunnel, validate_telegram_token, +}; pub use prompts::{confirm, print_header, print_step, secret_input, select_many, select_one}; pub use wizard::{SetupConfig, SetupWizard}; diff --git a/src/setup/wizard.rs b/src/setup/wizard.rs index cdc65175..bfa8a409 100644 --- a/src/setup/wizard.rs +++ b/src/setup/wizard.rs @@ -14,7 +14,7 @@ use tokio_postgres::NoTls; use crate::llm::{SessionConfig, SessionManager}; use crate::secrets::SecretsCrypto; use crate::settings::Settings; -use crate::setup::channels::{SecretsContext, setup_http, setup_telegram}; +use crate::setup::channels::{SecretsContext, setup_http, setup_telegram, setup_tunnel}; use crate::setup::prompts::{ input, print_header, print_info, print_step, print_success, select_many, select_one, }; @@ -316,6 +316,20 @@ impl SetupWizard { /// Step 3: Channel configuration. async fn step_channels(&mut self) -> Result<(), SetupError> { + // First, configure tunnel (shared across all channels that need webhooks) + match setup_tunnel() { + Ok(Some(url)) => { + self.settings.tunnel.public_url = Some(url); + } + Ok(None) => { + self.settings.tunnel.public_url = None; + } + Err(e) => { + print_info(&format!("Tunnel setup skipped: {}", e)); + } + } + println!(); + let options = [ ("CLI/TUI (always enabled)", true), ("HTTP webhook", self.settings.channels.http_enabled), @@ -381,6 +395,10 @@ impl SetupWizard { println!(" Model: {}", model); } + if let Some(ref tunnel_url) = self.settings.tunnel.public_url { + println!(" Tunnel: {}", tunnel_url); + } + println!(" Channels:"); println!(" - CLI/TUI: enabled"); @@ -390,7 +408,12 @@ impl SetupWizard { } if self.settings.channels.telegram_enabled { - println!(" - Telegram: enabled"); + let mode = if self.settings.tunnel.public_url.is_some() { + "webhook" + } else { + "polling" + }; + println!(" - Telegram: enabled ({})", mode); } println!(); diff --git a/src/tools/wasm/limits.rs b/src/tools/wasm/limits.rs index 679ce88d..237247e9 100644 --- a/src/tools/wasm/limits.rs +++ b/src/tools/wasm/limits.rs @@ -79,13 +79,16 @@ pub struct WasmResourceLimiter { impl WasmResourceLimiter { /// Create a new limiter with the given memory limit. + /// + /// Note: max_instances is set to 10 to accommodate WASM Component Model + /// which creates multiple internal instances (main component + WASI adapters). pub fn new(memory_limit: u64) -> Self { Self { memory_limit, memory_used: 0, max_tables: 10, tables_created: 0, - max_instances: 1, + max_instances: 10, // Component model needs multiple instances for WASI instances_created: 0, } } @@ -157,7 +160,7 @@ impl ResourceLimiter for WasmResourceLimiter { } fn memories(&self) -> usize { - // Allow one memory per instance + // Allow multiple memories for component model with WASI self.max_instances as usize } }