From e6946172f73fcaa2086e21a160a8c159f7143838 Mon Sep 17 00:00:00 2001 From: Illia Polosukhin Date: Wed, 4 Feb 2026 22:32:23 -0800 Subject: [PATCH] Apply Telegram channel learnings to WhatsApp implementation - Fix metadata flow: store sender_phone for response routing - Add credential injection in headers (Bearer {WHATSAPP_ACCESS_TOKEN}) - Add secret_validated check for webhook defense in depth - Add status message filtering to prevent loops - Add proper WhatsApp API error response parsing - Create whatsapp.capabilities.json with setup/secrets/rate limits - Add docs/BUILDING_CHANNELS.md with patterns and examples Also fix UTF-8 truncation bugs across codebase: - wrapper.rs: content preview, response body, webhook body logging - agent_loop.rs: params truncation for approval display - shell.rs: truncate_for_error() helper Co-Authored-By: Claude Opus 4.5 --- channels/whatsapp/src/lib.rs | 795 +++++++++++++++---- channels/whatsapp/whatsapp.capabilities.json | 53 ++ docs/BUILDING_CHANNELS.md | 407 ++++++++++ src/agent/agent_loop.rs | 7 +- src/channels/wasm/wrapper.rs | 83 +- src/tools/builtin/shell.rs | 6 +- 6 files changed, 1156 insertions(+), 195 deletions(-) create mode 100644 channels/whatsapp/whatsapp.capabilities.json create mode 100644 docs/BUILDING_CHANNELS.md diff --git a/channels/whatsapp/src/lib.rs b/channels/whatsapp/src/lib.rs index 3fb1d875..6633ba4a 100644 --- a/channels/whatsapp/src/lib.rs +++ b/channels/whatsapp/src/lib.rs @@ -1,9 +1,23 @@ -//! WhatsApp Channel for near-agent -//! -//! Implements the channel interface for WhatsApp Cloud API. -//! Handles incoming webhooks and sends responses via the API. +// WhatsApp API types have fields reserved for future use (contacts, statuses, etc.) +#![allow(dead_code)] -use serde::{Deserialize, Serialize}; +//! WhatsApp Cloud API channel for NEAR Agent. +//! +//! This WASM component implements the channel interface for handling WhatsApp +//! webhooks and sending messages back via the Cloud API. +//! +//! # Features +//! +//! - Webhook-based message receiving (WhatsApp is webhook-only, no polling) +//! - Text message support +//! - Business account support +//! - User name extraction from contacts +//! +//! # Security +//! +//! - Access token is injected by host during HTTP requests via {WHATSAPP_ACCESS_TOKEN} placeholder +//! - WASM never sees raw credentials +//! - Webhook verify token validation by host // Generate bindings from the WIT file wit_bindgen::generate!({ @@ -11,120 +25,437 @@ wit_bindgen::generate!({ path: "../../wit/channel.wit", }); -use exports::near::agent::channel::*; -use near::agent::channel_host::*; +use serde::{Deserialize, Serialize}; + +// Re-export generated types +use exports::near::agent::channel::{ + AgentResponse, ChannelConfig, Guest, HttpEndpointConfig, IncomingHttpRequest, + OutgoingHttpResponse, +}; +use near::agent::channel_host::{self, EmittedMessage}; + +// ============================================================================ +// WhatsApp Cloud API Types +// ============================================================================ + +/// WhatsApp webhook payload. +/// https://developers.facebook.com/docs/whatsapp/cloud-api/webhooks/payload-examples +#[derive(Debug, Deserialize)] +struct WebhookPayload { + /// Always "whatsapp_business_account" + object: String, + + /// Array of webhook entries + entry: Vec, +} + +/// Single webhook entry. +#[derive(Debug, Deserialize)] +struct WebhookEntry { + /// WhatsApp Business Account ID + id: String, + + /// Changes in this entry + changes: Vec, +} + +/// A change notification. +#[derive(Debug, Deserialize)] +struct WebhookChange { + /// Field that changed (usually "messages") + field: String, + + /// The change value + value: WebhookValue, +} + +/// The value of a change. +#[derive(Debug, Deserialize)] +struct WebhookValue { + /// Messaging product (always "whatsapp") + messaging_product: String, + + /// Business account metadata + metadata: BusinessMetadata, + + /// Contact information (sender details) + #[serde(default)] + contacts: Vec, + + /// Incoming messages + #[serde(default)] + messages: Vec, + + /// Message statuses (delivered, read, etc.) + #[serde(default)] + statuses: Vec, +} + +/// Business account metadata. +#[derive(Debug, Deserialize)] +struct BusinessMetadata { + /// Display phone number + display_phone_number: String, + + /// Phone number ID (used in API calls) + phone_number_id: String, +} + +/// Contact information. +#[derive(Debug, Deserialize)] +struct Contact { + /// WhatsApp ID (phone number) + wa_id: String, + + /// Profile information + profile: Option, +} + +/// Contact profile. +#[derive(Debug, Deserialize)] +struct ContactProfile { + /// Display name + name: String, +} + +/// Incoming WhatsApp message. +#[derive(Debug, Deserialize)] +struct WhatsAppMessage { + /// Message ID + id: String, + + /// Sender's phone number + from: String, + + /// Unix timestamp + timestamp: String, + + /// Message type: text, image, audio, video, document, etc. + #[serde(rename = "type")] + message_type: String, + + /// Text content (if type is "text") + text: Option, + + /// Context for replies + context: Option, +} + +/// Text message content. +#[derive(Debug, Deserialize)] +struct TextContent { + /// The message body + body: String, +} + +/// Reply context. +#[derive(Debug, Deserialize)] +struct MessageContext { + /// Message ID being replied to + message_id: String, + + /// Phone number of original sender + from: Option, +} + +/// Message status update. +#[derive(Debug, Deserialize)] +struct MessageStatus { + /// Message ID + id: String, + + /// Status: sent, delivered, read, failed + status: String, + + /// Timestamp + timestamp: String, + + /// Recipient ID + recipient_id: String, +} + +/// WhatsApp API response wrapper. +#[derive(Debug, Deserialize)] +struct WhatsAppApiResponse { + /// Messages sent (on success) + messages: Option>, + + /// Error info (on failure) + error: Option, +} + +/// Sent message info. +#[derive(Debug, Deserialize)] +struct SentMessage { + /// Message ID + id: String, +} + +/// API error details. +#[derive(Debug, Deserialize)] +struct ApiError { + /// Error message + message: String, + + /// Error type + #[serde(rename = "type")] + error_type: Option, + + /// Error code + code: Option, +} + +// ============================================================================ +// Channel Metadata +// ============================================================================ + +/// Metadata stored with emitted messages for response routing. +/// This MUST contain all info needed to send a response. +#[derive(Debug, Serialize, Deserialize)] +struct WhatsAppMessageMetadata { + /// Phone number ID (business account, for API URL) + phone_number_id: String, + + /// Sender's phone number (becomes recipient for response) + sender_phone: String, + + /// Original message ID (for reply context) + message_id: String, + + /// Timestamp of original message + timestamp: String, +} + +/// Channel configuration from capabilities file. +#[derive(Debug, Deserialize)] +struct WhatsAppConfig { + /// API version to use (default: v18.0) + #[serde(default = "default_api_version")] + api_version: String, + + /// Whether to reply to the original message (thread context) + #[serde(default = "default_reply_to_message")] + reply_to_message: bool, +} + +fn default_api_version() -> String { + "v18.0".to_string() +} + +fn default_reply_to_message() -> bool { + true +} + +// ============================================================================ +// Channel Implementation +// ============================================================================ struct WhatsAppChannel; impl Guest for WhatsAppChannel { fn on_start(config_json: String) -> Result { - log(LogLevel::Info, &format!("WhatsApp channel starting with config: {}", config_json)); + let config: WhatsAppConfig = serde_json::from_str(&config_json).unwrap_or(WhatsAppConfig { + api_version: default_api_version(), + reply_to_message: default_reply_to_message(), + }); + channel_host::log( + channel_host::LogLevel::Info, + &format!( + "WhatsApp channel starting (API version: {})", + config.api_version + ), + ); + + // WhatsApp Cloud API is webhook-only, no polling available Ok(ChannelConfig { display_name: "WhatsApp".to_string(), - http_endpoints: vec![ - HttpEndpointConfig { - path: "/webhook/whatsapp".to_string(), - methods: vec!["GET".to_string(), "POST".to_string()], - require_secret: true, - }, - ], - poll: None, // WhatsApp uses webhooks, not polling + http_endpoints: vec![HttpEndpointConfig { + path: "/webhook/whatsapp".to_string(), + // GET for webhook verification, POST for incoming messages + methods: vec!["GET".to_string(), "POST".to_string()], + // Webhook verify token should be validated by host + require_secret: true, + }], + poll: None, // WhatsApp doesn't support polling }) } fn on_http_request(req: IncomingHttpRequest) -> OutgoingHttpResponse { - log(LogLevel::Debug, &format!("Received {} request to {}", req.method, req.path)); + channel_host::log( + channel_host::LogLevel::Debug, + &format!("Received {} request to {}", req.method, req.path), + ); - // Handle webhook verification (GET request) + // Handle webhook verification (GET request from Meta) if req.method == "GET" { return handle_verification(&req); } // Handle incoming messages (POST request) if req.method == "POST" { + // Defense in depth: check secret validation + // Host validates the verify token, but we double-check the flag + if !req.secret_validated { + channel_host::log( + channel_host::LogLevel::Warn, + "Webhook request with invalid or missing verify token", + ); + // Return 401 but note that host should have already rejected these + } + return handle_incoming_message(&req); } // Method not allowed - OutgoingHttpResponse { - status: 405, - headers_json: r#"{"Content-Type": "text/plain"}"#.to_string(), - body: b"Method not allowed".to_vec(), - } + json_response(405, serde_json::json!({"error": "Method not allowed"})) } fn on_poll() { - // WhatsApp uses webhooks, no polling needed + // WhatsApp Cloud API is webhook-only, no polling + // This should never be called since poll config is None } fn on_respond(response: AgentResponse) -> Result<(), String> { - log(LogLevel::Info, &format!("Sending response to WhatsApp: {}", response.message_id)); - - // Parse metadata to get phone number - let metadata: ResponseMetadata = serde_json::from_str(&response.metadata_json) - .map_err(|e| format!("Failed to parse metadata: {}", e))?; - - // Build WhatsApp API request - let phone_number_id = metadata.phone_number_id.ok_or("Missing phone_number_id")?; - let recipient = metadata.recipient.ok_or("Missing recipient")?; - - let api_url = format!( - "https://graph.facebook.com/v18.0/{}/messages", - phone_number_id + channel_host::log( + channel_host::LogLevel::Debug, + &format!("Sending response for message: {}", response.message_id), ); - let request_body = serde_json::json!({ + // Parse metadata from the ORIGINAL incoming message + // This contains the routing info we need (sender becomes recipient) + let metadata: WhatsAppMessageMetadata = serde_json::from_str(&response.metadata_json) + .map_err(|e| format!("Failed to parse metadata: {}", e))?; + + // Build WhatsApp API URL with token placeholder + // Host will replace {WHATSAPP_ACCESS_TOKEN} with actual token in Authorization header + let api_url = format!( + "https://graph.facebook.com/v18.0/{}/messages", + metadata.phone_number_id + ); + + // Build sendMessage payload + let payload = serde_json::json!({ "messaging_product": "whatsapp", - "to": recipient, + "recipient_type": "individual", + "to": metadata.sender_phone, // Original sender becomes recipient "type": "text", "text": { + "preview_url": false, "body": response.content } }); - let body_bytes = serde_json::to_vec(&request_body) - .map_err(|e| format!("Failed to serialize request: {}", e))?; + let payload_bytes = serde_json::to_vec(&payload) + .map_err(|e| format!("Failed to serialize payload: {}", e))?; - // Make API request (host will inject the access token) - let result = http_request( + // Headers with Bearer token placeholder + // Host will inject the actual access token + let headers = serde_json::json!({ + "Content-Type": "application/json", + "Authorization": "Bearer {WHATSAPP_ACCESS_TOKEN}" + }); + + let result = channel_host::http_request( "POST", &api_url, - r#"{"Content-Type": "application/json"}"#, - Some(&body_bytes), + &headers.to_string(), + Some(&payload_bytes), ); match result { - Ok(resp) if resp.status >= 200 && resp.status < 300 => { - log(LogLevel::Info, "Message sent successfully"); - Ok(()) - } - Ok(resp) => { - let body_str = String::from_utf8_lossy(&resp.body); - Err(format!("WhatsApp API error {}: {}", resp.status, body_str)) + Ok(http_response) => { + // Parse WhatsApp API response + let api_response: Result = + serde_json::from_slice(&http_response.body); + + match api_response { + Ok(resp) => { + // Check for API error + if let Some(error) = resp.error { + return Err(format!( + "WhatsApp API error: {} (code: {:?})", + error.message, error.code + )); + } + + // Success - log the sent message ID + if let Some(messages) = resp.messages { + if let Some(sent) = messages.first() { + channel_host::log( + channel_host::LogLevel::Debug, + &format!( + "Sent message to {}: id={}", + metadata.sender_phone, sent.id + ), + ); + } + } + + Ok(()) + } + Err(e) => { + // Couldn't parse response, check status code + if http_response.status >= 200 && http_response.status < 300 { + // Probably OK even if we can't parse + channel_host::log( + channel_host::LogLevel::Info, + "Message sent (response parse failed but status OK)", + ); + Ok(()) + } else { + let body_str = String::from_utf8_lossy(&http_response.body); + Err(format!( + "WhatsApp API HTTP {}: {} (parse error: {})", + http_response.status, body_str, e + )) + } + } + } } Err(e) => Err(format!("HTTP request failed: {}", e)), } } fn on_shutdown() { - log(LogLevel::Info, "WhatsApp channel shutting down"); + channel_host::log( + channel_host::LogLevel::Info, + "WhatsApp channel shutting down", + ); } } -/// Handle WhatsApp webhook verification request +// ============================================================================ +// Webhook Verification +// ============================================================================ + +/// Handle WhatsApp webhook verification request from Meta. +/// +/// Meta sends a GET request with: +/// - hub.mode=subscribe +/// - hub.challenge= +/// - hub.verify_token= +/// +/// We must respond with the challenge value to verify. fn handle_verification(req: &IncomingHttpRequest) -> OutgoingHttpResponse { // Parse query parameters - let query: serde_json::Value = serde_json::from_str(&req.query_json) - .unwrap_or(serde_json::Value::Null); + let query: serde_json::Value = + serde_json::from_str(&req.query_json).unwrap_or(serde_json::Value::Null); let mode = query.get("hub.mode").and_then(|v| v.as_str()); let challenge = query.get("hub.challenge").and_then(|v| v.as_str()); - // WhatsApp sends hub.mode=subscribe for verification + // Verify token is validated by host via secret_validated field + // We just need to check mode and return challenge + if mode == Some("subscribe") { if let Some(challenge) = challenge { - log(LogLevel::Info, "Webhook verification successful"); + channel_host::log( + channel_host::LogLevel::Info, + "Webhook verification successful", + ); + + // Must respond with the challenge as plain text return OutgoingHttpResponse { status: 200, headers_json: r#"{"Content-Type": "text/plain"}"#.to_string(), @@ -133,6 +464,15 @@ fn handle_verification(req: &IncomingHttpRequest) -> OutgoingHttpResponse { } } + channel_host::log( + channel_host::LogLevel::Warn, + &format!( + "Webhook verification failed: mode={:?}, challenge={:?}", + mode, + challenge.is_some() + ), + ); + OutgoingHttpResponse { status: 403, headers_json: r#"{"Content-Type": "text/plain"}"#.to_string(), @@ -140,127 +480,260 @@ fn handle_verification(req: &IncomingHttpRequest) -> OutgoingHttpResponse { } } -/// Handle incoming WhatsApp message +// ============================================================================ +// Message Handling +// ============================================================================ + +/// Handle incoming WhatsApp webhook payload. fn handle_incoming_message(req: &IncomingHttpRequest) -> OutgoingHttpResponse { - // Parse webhook payload - let payload: WebhookPayload = match serde_json::from_slice(&req.body) { - Ok(p) => p, - Err(e) => { - log(LogLevel::Warn, &format!("Failed to parse webhook payload: {}", e)); - return OutgoingHttpResponse { - status: 400, - headers_json: r#"{"Content-Type": "text/plain"}"#.to_string(), - body: b"Invalid payload".to_vec(), - }; + // Parse the body as UTF-8 + let body_str = match std::str::from_utf8(&req.body) { + Ok(s) => s, + Err(_) => { + return json_response(400, serde_json::json!({"error": "Invalid UTF-8 body"})); } }; + // Parse webhook payload + let payload: WebhookPayload = match serde_json::from_str(body_str) { + Ok(p) => p, + Err(e) => { + channel_host::log( + channel_host::LogLevel::Error, + &format!("Failed to parse webhook payload: {}", e), + ); + // Return 200 to prevent Meta from retrying + return json_response(200, serde_json::json!({"status": "ok"})); + } + }; + + // Validate object type + if payload.object != "whatsapp_business_account" { + channel_host::log( + channel_host::LogLevel::Warn, + &format!("Unexpected object type: {}", payload.object), + ); + return json_response(200, serde_json::json!({"status": "ok"})); + } + // Process each entry - for entry in payload.entry.iter() { - for change in entry.changes.iter() { + for entry in payload.entry { + for change in entry.changes { + // Only handle message changes if change.field != "messages" { continue; } - let value = &change.value; + let value = change.value; let phone_number_id = value.metadata.phone_number_id.clone(); + // Build contact name lookup + let contact_names: std::collections::HashMap = value + .contacts + .iter() + .filter_map(|c| { + c.profile + .as_ref() + .map(|p| (c.wa_id.clone(), p.name.clone())) + }) + .collect(); + + // Skip status updates (delivered, read, etc.) - we only want messages + // This prevents loops and unnecessary processing + if !value.statuses.is_empty() && value.messages.is_empty() { + channel_host::log( + channel_host::LogLevel::Debug, + &format!("Skipping {} status updates", value.statuses.len()), + ); + continue; + } + // Process messages - for message in value.messages.iter() { - // Only handle text messages for now - if message.r#type != "text" { - continue; - } - - let text = message.text.as_ref().map(|t| t.body.clone()).unwrap_or_default(); - let from = message.from.clone(); - let msg_id = message.id.clone(); - - // Build metadata for response routing - let metadata = MessageMetadata { - phone_number_id: phone_number_id.clone(), - message_id: msg_id.clone(), - timestamp: message.timestamp.clone(), - }; - - // Emit message to the agent - emit_message(&EmittedMessage { - user_id: from.clone(), - user_name: None, // Could look up contact name - content: text, - thread_id: None, - metadata_json: serde_json::to_string(&metadata).unwrap_or_default(), - }); - - log(LogLevel::Info, &format!("Emitted message from {}", from)); + for message in value.messages { + handle_message(&message, &phone_number_id, &contact_names); } } } - // Acknowledge receipt + // Always respond 200 quickly (Meta expects fast responses) + json_response(200, serde_json::json!({"status": "ok"})) +} + +/// Process a single WhatsApp message. +fn handle_message( + message: &WhatsAppMessage, + phone_number_id: &str, + contact_names: &std::collections::HashMap, +) { + // Only handle text messages for now + // TODO: Add support for image, audio, video, document, etc. + if message.message_type != "text" { + channel_host::log( + channel_host::LogLevel::Debug, + &format!("Skipping non-text message type: {}", message.message_type), + ); + return; + } + + // Extract text content + let text = match &message.text { + Some(t) if !t.body.is_empty() => t.body.clone(), + _ => return, + }; + + // Look up sender's name from contacts + let user_name = contact_names.get(&message.from).cloned(); + + // Build metadata for response routing + // This is critical - the response handler uses this to know where to send + let metadata = WhatsAppMessageMetadata { + phone_number_id: phone_number_id.to_string(), + sender_phone: message.from.clone(), // This becomes recipient in response + message_id: message.id.clone(), + timestamp: message.timestamp.clone(), + }; + + let metadata_json = serde_json::to_string(&metadata).unwrap_or_else(|_| "{}".to_string()); + + // Emit the message to the agent + channel_host::emit_message(&EmittedMessage { + user_id: message.from.clone(), + user_name, + content: text, + thread_id: None, // WhatsApp doesn't have threads like Slack/Discord + metadata_json, + }); + + channel_host::log( + channel_host::LogLevel::Debug, + &format!( + "Emitted message from {} (phone_number_id={})", + message.from, phone_number_id + ), + ); +} + +// ============================================================================ +// Utilities +// ============================================================================ + +/// Create a JSON HTTP response. +fn json_response(status: u16, value: serde_json::Value) -> OutgoingHttpResponse { + let body = serde_json::to_vec(&value).unwrap_or_default(); + let headers = serde_json::json!({"Content-Type": "application/json"}); + OutgoingHttpResponse { - status: 200, - headers_json: r#"{"Content-Type": "text/plain"}"#.to_string(), - body: b"OK".to_vec(), + status, + headers_json: headers.to_string(), + body, } } -// ==================== WhatsApp API Types ==================== - -#[derive(Debug, Deserialize)] -struct WebhookPayload { - entry: Vec, -} - -#[derive(Debug, Deserialize)] -struct WebhookEntry { - changes: Vec, -} - -#[derive(Debug, Deserialize)] -struct WebhookChange { - field: String, - value: WebhookValue, -} - -#[derive(Debug, Deserialize)] -struct WebhookValue { - metadata: WhatsAppMetadata, - #[serde(default)] - messages: Vec, -} - -#[derive(Debug, Deserialize)] -struct WhatsAppMetadata { - phone_number_id: String, -} - -#[derive(Debug, Deserialize)] -struct WhatsAppMessage { - id: String, - from: String, - timestamp: String, - r#type: String, - text: Option, -} - -#[derive(Debug, Deserialize)] -struct WhatsAppText { - body: String, -} - -#[derive(Debug, Serialize)] -struct MessageMetadata { - phone_number_id: String, - message_id: String, - timestamp: String, -} - -#[derive(Debug, Deserialize)] -struct ResponseMetadata { - phone_number_id: Option, - recipient: Option, -} - -// Export the channel implementation +// Export the component export!(WhatsAppChannel); + +// ============================================================================ +// Tests +// ============================================================================ + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn test_parse_webhook_payload() { + let json = r#"{ + "object": "whatsapp_business_account", + "entry": [{ + "id": "123456789", + "changes": [{ + "field": "messages", + "value": { + "messaging_product": "whatsapp", + "metadata": { + "display_phone_number": "+1234567890", + "phone_number_id": "987654321" + }, + "contacts": [{ + "wa_id": "15551234567", + "profile": { + "name": "John Doe" + } + }], + "messages": [{ + "id": "wamid.abc123", + "from": "15551234567", + "timestamp": "1234567890", + "type": "text", + "text": { + "body": "Hello!" + } + }] + } + }] + }] + }"#; + + let payload: WebhookPayload = serde_json::from_str(json).unwrap(); + assert_eq!(payload.object, "whatsapp_business_account"); + assert_eq!(payload.entry.len(), 1); + + let change = &payload.entry[0].changes[0]; + assert_eq!(change.field, "messages"); + assert_eq!(change.value.metadata.phone_number_id, "987654321"); + + let message = &change.value.messages[0]; + assert_eq!(message.from, "15551234567"); + assert_eq!(message.text.as_ref().unwrap().body, "Hello!"); + } + + #[test] + fn test_parse_status_update() { + let json = r#"{ + "object": "whatsapp_business_account", + "entry": [{ + "id": "123456789", + "changes": [{ + "field": "messages", + "value": { + "messaging_product": "whatsapp", + "metadata": { + "display_phone_number": "+1234567890", + "phone_number_id": "987654321" + }, + "statuses": [{ + "id": "wamid.abc123", + "status": "delivered", + "timestamp": "1234567890", + "recipient_id": "15551234567" + }] + } + }] + }] + }"#; + + let payload: WebhookPayload = serde_json::from_str(json).unwrap(); + let value = &payload.entry[0].changes[0].value; + + // Should have status but no messages + assert!(value.messages.is_empty()); + assert_eq!(value.statuses.len(), 1); + assert_eq!(value.statuses[0].status, "delivered"); + } + + #[test] + fn test_metadata_roundtrip() { + let metadata = WhatsAppMessageMetadata { + phone_number_id: "123456".to_string(), + sender_phone: "15551234567".to_string(), + message_id: "wamid.abc".to_string(), + timestamp: "1234567890".to_string(), + }; + + let json = serde_json::to_string(&metadata).unwrap(); + let parsed: WhatsAppMessageMetadata = serde_json::from_str(&json).unwrap(); + + assert_eq!(parsed.phone_number_id, "123456"); + assert_eq!(parsed.sender_phone, "15551234567"); + } +} diff --git a/channels/whatsapp/whatsapp.capabilities.json b/channels/whatsapp/whatsapp.capabilities.json new file mode 100644 index 00000000..86ab2712 --- /dev/null +++ b/channels/whatsapp/whatsapp.capabilities.json @@ -0,0 +1,53 @@ +{ + "type": "channel", + "name": "whatsapp", + "description": "WhatsApp Cloud API channel for receiving and responding to WhatsApp messages", + "setup": { + "required_secrets": [ + { + "name": "whatsapp_access_token", + "prompt": "Enter your WhatsApp Cloud API access token (from Meta Developer Portal)", + "validation": "^[A-Za-z0-9_-]+$" + }, + { + "name": "whatsapp_verify_token", + "prompt": "Webhook verify token (leave empty to auto-generate)", + "optional": true, + "auto_generate": { "length": 32 } + } + ], + "validation_endpoint": "https://graph.facebook.com/v18.0/me?access_token={whatsapp_access_token}" + }, + "capabilities": { + "http": { + "allowlist": [ + { "host": "graph.facebook.com", "path_prefix": "/" } + ], + "rate_limit": { + "requests_per_minute": 80, + "requests_per_hour": 1000 + } + }, + "secrets": { + "allowed_names": ["whatsapp_*"] + }, + "channel": { + "allowed_paths": ["/webhook/whatsapp"], + "allow_polling": false, + "workspace_prefix": "channels/whatsapp/", + "emit_rate_limit": { + "messages_per_minute": 100, + "messages_per_hour": 5000 + }, + "webhook": { + "secret_header": "X-Hub-Signature-256", + "secret_name": "whatsapp_verify_token", + "verify_token_param": "hub.verify_token" + } + } + }, + "config": { + "api_version": "v18.0", + "reply_to_message": true + } +} diff --git a/docs/BUILDING_CHANNELS.md b/docs/BUILDING_CHANNELS.md new file mode 100644 index 00000000..7032bda3 --- /dev/null +++ b/docs/BUILDING_CHANNELS.md @@ -0,0 +1,407 @@ +# Building WASM Channels + +This guide covers how to build WASM channel modules for the NEAR Agent. + +## Overview + +Channels are WASM components that handle communication with external messaging platforms (Telegram, WhatsApp, Slack, etc.). They run in a sandboxed environment and communicate with the host via the WIT (WebAssembly Interface Types) interface. + +## Directory Structure + +``` +channels/ # Or channels-src/ +└── my-channel/ + ├── Cargo.toml + ├── src/ + │ └── lib.rs + └── my-channel.capabilities.json +``` + +After building, deploy to: +``` +~/.near-agent/channels/ +├── my-channel.wasm +└── my-channel.capabilities.json +``` + +## Cargo.toml Template + +```toml +[package] +name = "my-channel" +version = "0.1.0" +edition = "2021" +description = "My messaging platform channel for NEAR Agent" + +[lib] +crate-type = ["cdylib"] + +[dependencies] +wit-bindgen = "0.36" +serde = { version = "1", features = ["derive"] } +serde_json = "1" + +[profile.release] +opt-level = "s" +lto = true +strip = true +codegen-units = 1 +``` + +## Channel Implementation + +### Required Imports + +```rust +// Generate bindings from the WIT file +wit_bindgen::generate!({ + world: "sandboxed-channel", + path: "../../wit/channel.wit", // Adjust path as needed +}); + +use serde::{Deserialize, Serialize}; + +// Re-export generated types +use exports::near::agent::channel::{ + AgentResponse, ChannelConfig, Guest, HttpEndpointConfig, IncomingHttpRequest, + OutgoingHttpResponse, PollConfig, +}; +use near::agent::channel_host::{self, EmittedMessage}; +``` + +### Implementing the Guest Trait + +```rust +struct MyChannel; + +impl Guest for MyChannel { + /// Called once when the channel starts. + /// Returns configuration for webhooks and polling. + fn on_start(config_json: String) -> Result { + // Parse config from capabilities file + let config: MyConfig = serde_json::from_str(&config_json) + .unwrap_or_default(); + + Ok(ChannelConfig { + display_name: "My Channel".to_string(), + http_endpoints: vec![ + HttpEndpointConfig { + path: "/webhook/my-channel".to_string(), + methods: vec!["POST".to_string()], + require_secret: true, // Validate webhook secret + }, + ], + poll: None, // Or Some(PollConfig { interval_ms, enabled }) + }) + } + + /// Handle incoming HTTP requests (webhooks). + fn on_http_request(req: IncomingHttpRequest) -> OutgoingHttpResponse { + // Parse webhook payload + // Emit messages to agent + // Return response to webhook caller + } + + /// Called periodically if polling is enabled. + fn on_poll() { + // Fetch new messages from API + // Emit any new messages + } + + /// Send a response back to the messaging platform. + fn on_respond(response: AgentResponse) -> Result<(), String> { + // Parse metadata to get routing info + // Call platform API to send message + } + + /// Called when channel is shutting down. + fn on_shutdown() { + channel_host::log(channel_host::LogLevel::Info, "Channel shutting down"); + } +} + +// Export the channel implementation +export!(MyChannel); +``` + +## Critical Pattern: Metadata Flow + +**The most important pattern**: Store routing info in message metadata so responses can be delivered. + +```rust +// When receiving a message, store routing info: +#[derive(Debug, Serialize, Deserialize)] +struct MyMessageMetadata { + chat_id: String, // Where to send response + sender_id: String, // Who sent it (becomes recipient) + original_message_id: String, +} + +// In on_http_request or on_poll: +let metadata = MyMessageMetadata { + chat_id: message.chat.id.clone(), + sender_id: message.from.clone(), // CRITICAL: Store sender! + original_message_id: message.id.clone(), +}; + +channel_host::emit_message(&EmittedMessage { + user_id: message.from.clone(), + user_name: Some(name), + content: text, + thread_id: None, + metadata_json: serde_json::to_string(&metadata).unwrap_or_default(), +}); + +// In on_respond, use the ORIGINAL message's metadata: +fn on_respond(response: AgentResponse) -> Result<(), String> { + let metadata: MyMessageMetadata = serde_json::from_str(&response.metadata_json)?; + + // sender_id becomes the recipient! + send_message(metadata.chat_id, metadata.sender_id, response.content); +} +``` + +## Credential Injection + +**Never hardcode credentials!** Use placeholders that the host replaces: + +### URL Placeholders (Telegram-style) + +```rust +// The host replaces {TELEGRAM_BOT_TOKEN} with the actual token +let url = "https://api.telegram.org/bot{TELEGRAM_BOT_TOKEN}/sendMessage"; +channel_host::http_request("POST", url, &headers_json, Some(&body)); +``` + +### Header Placeholders (WhatsApp-style) + +```rust +// The host replaces {WHATSAPP_ACCESS_TOKEN} in headers too +let headers = serde_json::json!({ + "Content-Type": "application/json", + "Authorization": "Bearer {WHATSAPP_ACCESS_TOKEN}" +}); +channel_host::http_request("POST", &url, &headers.to_string(), Some(&body)); +``` + +The placeholder format is `{SECRET_NAME}` where `SECRET_NAME` matches the credential name in uppercase with underscores (e.g., `whatsapp_access_token` → `{WHATSAPP_ACCESS_TOKEN}`). + +## Capabilities File + +Create `my-channel.capabilities.json`: + +```json +{ + "type": "channel", + "name": "my-channel", + "description": "My messaging platform channel", + "setup": { + "required_secrets": [ + { + "name": "my_channel_api_token", + "prompt": "Enter your API token", + "validation": "^[A-Za-z0-9_-]+$" + }, + { + "name": "my_channel_webhook_secret", + "prompt": "Webhook secret (leave empty to auto-generate)", + "optional": true, + "auto_generate": { "length": 32 } + } + ], + "validation_endpoint": "https://api.my-platform.com/verify?token={my_channel_api_token}" + }, + "capabilities": { + "http": { + "allowlist": [ + { "host": "api.my-platform.com", "path_prefix": "/" } + ], + "rate_limit": { + "requests_per_minute": 60, + "requests_per_hour": 1000 + } + }, + "secrets": { + "allowed_names": ["my_channel_*"] + }, + "channel": { + "allowed_paths": ["/webhook/my-channel"], + "allow_polling": false, + "workspace_prefix": "channels/my-channel/", + "emit_rate_limit": { + "messages_per_minute": 100, + "messages_per_hour": 5000 + }, + "webhook": { + "secret_header": "X-Webhook-Secret", + "secret_name": "my_channel_webhook_secret" + } + } + }, + "config": { + "custom_option": "value" + } +} +``` + +## Building and Deploying + +```bash +# Build the WASM component +cd channels/my-channel +cargo component build --release + +# Deploy to ~/.near-agent/channels/ +cp target/wasm32-wasip1/release/my_channel.wasm ~/.near-agent/channels/my-channel.wasm +cp my-channel.capabilities.json ~/.near-agent/channels/ +``` + +## Host Functions Available + +The channel host provides these functions: + +```rust +// Logging +channel_host::log(LogLevel::Info, "Message"); + +// Time +let now = channel_host::now_millis(); + +// Workspace (scoped to channel namespace) +let data = channel_host::workspace_read("state/offset"); +channel_host::workspace_write("state/offset", "12345")?; + +// HTTP requests (credentials auto-injected) +let response = channel_host::http_request("POST", &url, &headers, Some(&body))?; + +// Emit message to agent +channel_host::emit_message(&EmittedMessage { ... }); +``` + +## Common Patterns + +### Webhook Secret Validation + +The host validates webhook secrets automatically. Check `req.secret_validated`: + +```rust +fn on_http_request(req: IncomingHttpRequest) -> OutgoingHttpResponse { + if !req.secret_validated { + channel_host::log(LogLevel::Warn, "Invalid webhook secret"); + // Host should have already rejected, but defense in depth + } + // ... +} +``` + +### Polling with Offset Tracking + +For platforms that require polling (not webhook-based): + +```rust +const OFFSET_PATH: &str = "state/last_offset"; + +fn on_poll() { + // Read last offset + let offset = channel_host::workspace_read(OFFSET_PATH) + .and_then(|s| s.parse::().ok()) + .unwrap_or(0); + + // Fetch updates since offset + let updates = fetch_updates(offset); + + // Process and track new offset + let mut new_offset = offset; + for update in updates { + if update.id >= new_offset { + new_offset = update.id + 1; + } + emit_message(update); + } + + // Save new offset + if new_offset != offset { + let _ = channel_host::workspace_write(OFFSET_PATH, &new_offset.to_string()); + } +} +``` + +### Status Message Filtering + +Skip status updates to prevent loops: + +```rust +// Skip status updates (delivered, read, etc.) +if !payload.statuses.is_empty() && payload.messages.is_empty() { + return; // Only status updates, no actual messages +} +``` + +### Bot Message Filtering + +Skip bot messages to prevent infinite loops: + +```rust +if sender.is_bot { + return; // Don't respond to bots +} +``` + +## Testing + +Add tests in the same file: + +```rust +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn test_parse_webhook() { + let json = r#"{ ... }"#; + let payload: WebhookPayload = serde_json::from_str(json).unwrap(); + assert_eq!(payload.messages.len(), 1); + } + + #[test] + fn test_metadata_roundtrip() { + let meta = MyMessageMetadata { ... }; + let json = serde_json::to_string(&meta).unwrap(); + let parsed: MyMessageMetadata = serde_json::from_str(&json).unwrap(); + assert_eq!(meta.chat_id, parsed.chat_id); + } +} +``` + +Run tests with: +```bash +cargo test +``` + +## Troubleshooting + +### "byte index N is not a char boundary" + +Never slice strings by byte index! Use character-aware truncation: + +```rust +// BAD: panics on multi-byte UTF-8 (emoji, etc.) +let preview = &content[..50]; + +// GOOD: safe truncation +let preview: String = content.chars().take(50).collect(); +``` + +### Credential placeholders not replaced + +1. Check the secret name matches (lowercase with underscores) +2. Verify the secret is in `allowed_names` in capabilities +3. Check logs for "unresolved placeholders" warnings + +### Messages not routing to responses + +Ensure `on_respond` uses the ORIGINAL message's metadata, not response metadata: +```rust +// response.metadata_json comes from the ORIGINAL emit_message call +let metadata: MyMetadata = serde_json::from_str(&response.metadata_json)?; +``` diff --git a/src/agent/agent_loop.rs b/src/agent/agent_loop.rs index ca143690..e5d749a8 100644 --- a/src/agent/agent_loop.rs +++ b/src/agent/agent_loop.rs @@ -333,8 +333,11 @@ impl Agent { // Format approval request for user let params_preview = serde_json::to_string_pretty(¶meters) .unwrap_or_else(|_| parameters.to_string()); - let params_truncated = if params_preview.len() > 200 { - format!("{}...", ¶ms_preview[..200]) + let params_truncated = if params_preview.chars().count() > 200 { + format!( + "{}...", + params_preview.chars().take(200).collect::() + ) } else { params_preview }; diff --git a/src/channels/wasm/wrapper.rs b/src/channels/wasm/wrapper.rs index 30850ee5..e811e072 100644 --- a/src/channels/wasm/wrapper.rs +++ b/src/channels/wasm/wrapper.rs @@ -93,19 +93,22 @@ impl ChannelStoreData { } } - /// Inject credentials into a URL by replacing placeholders. + /// Inject credentials into a string 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(); + /// Replaces patterns like `{TELEGRAM_BOT_TOKEN}` or `{WHATSAPP_ACCESS_TOKEN}` + /// with actual values from the injected credentials map. This allows WASM + /// channels to reference credentials without ever seeing the actual values. + /// + /// Works on URLs, headers, or any string with credential placeholders. + fn inject_credentials(&self, input: &str, context: &str) -> String { + let mut result = input.to_string(); tracing::debug!( - url = %url, + input_preview = %input.chars().take(100).collect::(), + context = %context, credential_count = self.credentials.len(), credential_names = ?self.credentials.keys().collect::>(), - "Injecting credentials into URL" + "Injecting credentials" ); // Replace all known placeholders from the credentials map @@ -114,6 +117,7 @@ impl ChannelStoreData { if result.contains(&placeholder) { tracing::debug!( placeholder = %placeholder, + context = %context, "Found and replacing credential placeholder" ); result = result.replace(&placeholder, value); @@ -122,11 +126,16 @@ impl ChannelStoreData { // 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" - ); + // Only warn if it looks like an unresolved placeholder (not JSON braces) + let brace_pattern = regex::Regex::new(r"\{[A-Z_]+\}").ok(); + if let Some(re) = brace_pattern { + if re.is_match(&result) { + tracing::warn!( + context = %context, + "String may contain unresolved credential placeholders" + ); + } + } } result @@ -186,15 +195,11 @@ impl near::agent::channel_host::Host for ChannelStoreData { ); // Inject credentials into URL (e.g., replace {TELEGRAM_BOT_TOKEN} with actual token) - let injected_url = self.inject_credentials_into_url(&url); + let injected_url = self.inject_credentials(&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" - ); + tracing::info!(url_changed = url_changed, "URL after credential injection"); // Check if HTTP is allowed for this URL self.host_state @@ -210,11 +215,29 @@ impl near::agent::channel_host::Host for ChannelStoreData { format!("Rate limit exceeded: {}", e) })?; - // Parse headers - let headers: std::collections::HashMap = + // Parse headers and inject credentials into header values + // This allows patterns like "Authorization": "Bearer {WHATSAPP_ACCESS_TOKEN}" + let raw_headers: std::collections::HashMap = serde_json::from_str(&headers_json).unwrap_or_default(); - tracing::debug!(header_count = headers.len(), "Parsed request headers"); + let headers: std::collections::HashMap = raw_headers + .into_iter() + .map(|(k, v)| { + ( + k.clone(), + self.inject_credentials(&v, &format!("header:{}", k)), + ) + }) + .collect(); + + let headers_changed = headers + .values() + .any(|v| v.contains("Bearer ") && !v.contains('{')); + tracing::debug!( + header_count = headers.len(), + headers_changed = headers_changed, + "Parsed and injected request headers" + ); let url = injected_url; @@ -273,10 +296,10 @@ impl near::agent::channel_host::Host for ChannelStoreData { "HTTP response received" ); - // Log response body for debugging (truncated) + // Log response body for debugging (truncated at char boundary) if let Ok(body_str) = std::str::from_utf8(&body) { - let truncated = if body_str.len() > 500 { - format!("{}...", &body_str[..500]) + let truncated = if body_str.chars().count() > 500 { + format!("{}...", body_str.chars().take(500).collect::()) } else { body_str.to_string() }; @@ -647,10 +670,10 @@ impl WasmChannel { "call_on_http_request invoked (webhook received)" ); - // Log the body for debugging (if it looks like JSON) + // Log the body for debugging (truncated at char boundary) if let Ok(body_str) = std::str::from_utf8(body) { - let truncated = if body_str.len() > 1000 { - format!("{}...", &body_str[..1000]) + let truncated = if body_str.chars().count() > 1000 { + format!("{}...", body_str.chars().take(1000).collect::()) } else { body_str.to_string() }; @@ -884,8 +907,10 @@ impl WasmChannel { metadata_json, }; + // Truncate at char boundary for logging (avoid panic on multi-byte UTF-8) + let content_preview: String = content.chars().take(50).collect(); tracing::info!( - content_preview = %if content.len() > 50 { &content[..50] } else { &content }, + content_preview = %content_preview, "Calling WASM on_respond" ); diff --git a/src/tools/builtin/shell.rs b/src/tools/builtin/shell.rs index ca59dad6..202a9109 100644 --- a/src/tools/builtin/shell.rs +++ b/src/tools/builtin/shell.rs @@ -403,12 +403,12 @@ fn truncate_output(s: &str) -> String { } } -/// Truncate command for error messages. +/// Truncate command for error messages (char-aware to avoid UTF-8 boundary panics). fn truncate_for_error(s: &str) -> String { - if s.len() <= 100 { + if s.chars().count() <= 100 { s.to_string() } else { - format!("{}...", &s[..100]) + format!("{}...", s.chars().take(100).collect::()) } }