Implementing channels to be handled in wasm

This commit is contained in:
Illia Polosukhin
2026-02-03 14:57:27 -08:00
parent 2c26ba8431
commit 74d94d33b0
19 changed files with 5107 additions and 2 deletions
+29
View File
@@ -0,0 +1,29 @@
[package]
name = "slack-channel"
version = "0.1.0"
edition = "2021"
description = "Slack Events API channel for NEAR Agent"
license = "MIT OR Apache-2.0"
[lib]
crate-type = ["cdylib"]
[dependencies]
# WIT bindgen for WASM component model
wit-bindgen = "0.36"
# Serialization
serde = { version = "1.0", features = ["derive"] }
serde_json = "1.0"
# HMAC for signature validation
hmac = "0.12"
sha2 = "0.10"
hex = "0.4"
[profile.release]
# Optimize for size
opt-level = "s"
lto = true
strip = true
codegen-units = 1
+37
View File
@@ -0,0 +1,37 @@
#!/usr/bin/env bash
# Build the Slack channel WASM component
#
# Prerequisites:
# - Rust with wasm32-wasip2 target: rustup target add wasm32-wasip2
# - wasm-tools for component creation: cargo install wasm-tools
#
# Output:
# - slack.wasm - WASM component ready for deployment
# - slack.capabilities.json - Capabilities file (copy alongside .wasm)
set -euo pipefail
cd "$(dirname "$0")"
echo "Building Slack channel WASM component..."
# Build the WASM module
cargo build --release --target wasm32-wasip2
# Convert to component model (if not already a component)
# wasm-tools component new is idempotent on components
WASM_PATH="target/wasm32-wasip2/release/slack_channel.wasm"
if [ -f "$WASM_PATH" ]; then
# Create component if needed
wasm-tools component new "$WASM_PATH" -o slack.wasm 2>/dev/null || cp "$WASM_PATH" slack.wasm
# Optimize the component
wasm-tools strip slack.wasm -o slack.wasm
echo "Built: slack.wasm ($(du -h slack.wasm | cut -f1))"
echo "Copy slack.wasm and slack.capabilities.json to ~/.near-agent/channels/"
else
echo "Error: WASM output not found at $WASM_PATH"
exit 1
fi
@@ -0,0 +1,38 @@
{
"type": "channel",
"name": "slack",
"description": "Slack Events API channel for receiving and responding to Slack messages",
"capabilities": {
"http": {
"allowlist": [
{ "host": "slack.com", "path_prefix": "/api/" }
],
"credentials": {
"slack_bot": {
"secret_name": "slack_bot_token",
"location": { "type": "bearer" },
"host_patterns": ["slack.com"]
}
},
"rate_limit": {
"requests_per_minute": 50,
"requests_per_hour": 1000
}
},
"secrets": {
"allowed_names": ["slack_*"]
},
"channel": {
"allowed_paths": ["/webhook/slack"],
"allow_polling": false,
"workspace_prefix": "channels/slack/",
"emit_rate_limit": {
"messages_per_minute": 100,
"messages_per_hour": 5000
}
}
},
"config": {
"signing_secret_name": "slack_signing_secret"
}
}
+376
View File
@@ -0,0 +1,376 @@
//! Slack Events API channel for NEAR Agent.
//!
//! This WASM component implements the channel interface for handling Slack
//! webhooks and sending messages back to Slack.
//!
//! # Features
//!
//! - URL verification for Slack Events API
//! - Message event parsing (@mentions, DMs)
//! - Thread support for conversations
//! - Response posting via Slack Web API
//!
//! # Security
//!
//! - Signature validation is handled by the host (webhook secrets)
//! - Bot token is injected by host during HTTP requests
//! - WASM never sees raw credentials
// Generate bindings from the WIT file
wit_bindgen::generate!({
world: "sandboxed-channel",
path: "../../wit/channel.wit",
});
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};
/// Slack event wrapper.
#[derive(Debug, Deserialize)]
struct SlackEventWrapper {
/// Event type (url_verification, event_callback, etc.)
#[serde(rename = "type")]
event_type: String,
/// Challenge token for URL verification.
challenge: Option<String>,
/// The actual event payload (for event_callback).
event: Option<SlackEvent>,
/// Team ID that sent this event.
team_id: Option<String>,
/// Event ID for deduplication.
event_id: Option<String>,
}
/// Slack event payload.
#[derive(Debug, Deserialize)]
struct SlackEvent {
/// Event type (message, app_mention, etc.)
#[serde(rename = "type")]
event_type: String,
/// User who triggered the event.
user: Option<String>,
/// Channel where the event occurred.
channel: Option<String>,
/// Message text.
text: Option<String>,
/// Thread timestamp (for threaded messages).
thread_ts: Option<String>,
/// Message timestamp.
ts: Option<String>,
/// Bot ID (if message is from a bot).
bot_id: Option<String>,
/// Subtype (bot_message, etc.)
subtype: Option<String>,
}
/// Metadata stored with emitted messages for response routing.
#[derive(Debug, Serialize, Deserialize)]
struct SlackMessageMetadata {
/// Slack channel ID.
channel: String,
/// Thread timestamp for threaded replies.
thread_ts: Option<String>,
/// Original message timestamp.
message_ts: String,
/// Team ID.
team_id: Option<String>,
}
/// Slack API response for chat.postMessage.
#[derive(Debug, Deserialize)]
struct SlackPostMessageResponse {
ok: bool,
error: Option<String>,
ts: Option<String>,
}
/// Channel configuration from capabilities file.
#[derive(Debug, Deserialize)]
struct SlackConfig {
/// Name of secret containing signing secret (for verification by host).
#[serde(default = "default_signing_secret_name")]
signing_secret_name: String,
}
fn default_signing_secret_name() -> String {
"slack_signing_secret".to_string()
}
struct SlackChannel;
impl Guest for SlackChannel {
fn on_start(config_json: String) -> Result<ChannelConfig, String> {
// Parse configuration
let _config: SlackConfig = serde_json::from_str(&config_json)
.map_err(|e| format!("Failed to parse config: {}", e))?;
channel_host::log(channel_host::LogLevel::Info, "Slack channel starting");
Ok(ChannelConfig {
display_name: "Slack".to_string(),
http_endpoints: vec![HttpEndpointConfig {
path: "/webhook/slack".to_string(),
methods: vec!["POST".to_string()],
require_secret: true,
}],
poll: None, // Slack uses push via webhooks, no polling needed
})
}
fn on_http_request(req: IncomingHttpRequest) -> OutgoingHttpResponse {
// Parse the request body
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 as Slack event
let event_wrapper: SlackEventWrapper = match serde_json::from_str(body_str) {
Ok(e) => e,
Err(e) => {
channel_host::log(
channel_host::LogLevel::Error,
&format!("Failed to parse Slack event: {}", e),
);
return json_response(400, serde_json::json!({"error": "Invalid event payload"}));
}
};
match event_wrapper.event_type.as_str() {
// URL verification challenge (Slack setup)
"url_verification" => {
if let Some(challenge) = event_wrapper.challenge {
channel_host::log(
channel_host::LogLevel::Info,
"Responding to Slack URL verification",
);
json_response(200, serde_json::json!({"challenge": challenge}))
} else {
json_response(400, serde_json::json!({"error": "Missing challenge"}))
}
}
// Actual event callback
"event_callback" => {
if let Some(event) = event_wrapper.event {
handle_slack_event(
event,
event_wrapper.team_id,
event_wrapper.event_id,
);
}
// Always respond 200 quickly to Slack (they have a 3s timeout)
json_response(200, serde_json::json!({"ok": true}))
}
// Unknown event type
_ => {
channel_host::log(
channel_host::LogLevel::Warn,
&format!("Unknown Slack event type: {}", event_wrapper.event_type),
);
json_response(200, serde_json::json!({"ok": true}))
}
}
}
fn on_poll() {
// Slack uses webhooks, no polling needed
}
fn on_respond(response: AgentResponse) -> Result<(), String> {
// Parse metadata to get channel info
let metadata: SlackMessageMetadata = serde_json::from_str(&response.metadata_json)
.map_err(|e| format!("Failed to parse metadata: {}", e))?;
// Build Slack API request
let mut payload = serde_json::json!({
"channel": metadata.channel,
"text": response.content,
});
// Add thread_ts for threaded replies
if let Some(thread_ts) = response.thread_id.or(metadata.thread_ts) {
payload["thread_ts"] = serde_json::Value::String(thread_ts);
}
let payload_bytes = serde_json::to_vec(&payload)
.map_err(|e| format!("Failed to serialize payload: {}", e))?;
// Make HTTP request to Slack API
// The bot token is injected by the host based on credential configuration
let headers = serde_json::json!({
"Content-Type": "application/json"
});
let result = channel_host::http_request(
"POST",
"https://slack.com/api/chat.postMessage",
&headers.to_string(),
Some(&payload_bytes),
);
match result {
Ok(http_response) => {
if http_response.status != 200 {
return Err(format!(
"Slack API returned status {}",
http_response.status
));
}
// Parse Slack response
let slack_response: SlackPostMessageResponse =
serde_json::from_slice(&http_response.body).map_err(|e| {
format!("Failed to parse Slack response: {}", e)
})?;
if !slack_response.ok {
return Err(format!(
"Slack API error: {}",
slack_response.error.unwrap_or_else(|| "unknown".to_string())
));
}
channel_host::log(
channel_host::LogLevel::Debug,
&format!(
"Posted message to Slack channel {}: ts={}",
metadata.channel,
slack_response.ts.unwrap_or_default()
),
);
Ok(())
}
Err(e) => Err(format!("HTTP request failed: {}", e)),
}
}
fn on_shutdown() {
channel_host::log(channel_host::LogLevel::Info, "Slack channel shutting down");
}
}
/// Handle a Slack event and emit message if applicable.
fn handle_slack_event(
event: SlackEvent,
team_id: Option<String>,
_event_id: Option<String>,
) {
match event.event_type.as_str() {
// Direct mention of the bot
"app_mention" => {
if let (Some(user), Some(channel), Some(text), Some(ts)) =
(event.user, event.channel.clone(), event.text, event.ts.clone())
{
emit_message(user, text, channel, event.thread_ts.or(Some(ts)), team_id);
}
}
// Direct message to the bot
"message" => {
// Skip messages from bots (including ourselves)
if event.bot_id.is_some() || event.subtype.is_some() {
return;
}
if let (Some(user), Some(channel), Some(text), Some(ts)) =
(event.user, event.channel.clone(), event.text, event.ts.clone())
{
// Only process DMs (channel IDs starting with D)
if channel.starts_with('D') {
emit_message(user, text, channel, event.thread_ts.or(Some(ts)), team_id);
}
}
}
_ => {
channel_host::log(
channel_host::LogLevel::Debug,
&format!("Ignoring Slack event type: {}", event.event_type),
);
}
}
}
/// Emit a message to the agent.
fn emit_message(
user_id: String,
text: String,
channel: String,
thread_ts: Option<String>,
team_id: Option<String>,
) {
let message_ts = thread_ts.clone().unwrap_or_default();
let metadata = SlackMessageMetadata {
channel: channel.clone(),
thread_ts: thread_ts.clone(),
message_ts: message_ts.clone(),
team_id,
};
let metadata_json =
serde_json::to_string(&metadata).unwrap_or_else(|_| "{}".to_string());
// Strip @ mentions of the bot from the text for cleaner messages
let cleaned_text = strip_bot_mention(&text);
channel_host::emit_message(EmittedMessage {
user_id,
user_name: None, // Could fetch from Slack API if needed
content: cleaned_text,
thread_id: thread_ts,
metadata_json,
});
}
/// Strip leading bot mention from text.
fn strip_bot_mention(text: &str) -> String {
// Slack mentions look like <@U12345678>
let trimmed = text.trim();
if trimmed.starts_with("<@") {
if let Some(end) = trimmed.find('>') {
return trimmed[end + 1..].trim_start().to_string();
}
}
trimmed.to_string()
}
/// 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,
headers_json: headers.to_string(),
body,
}
}
// Export the component
export!(SlackChannel);
+25
View File
@@ -2,12 +2,37 @@
//! //!
//! Channels receive messages from external sources (CLI, HTTP, etc.) //! Channels receive messages from external sources (CLI, HTTP, etc.)
//! and convert them to a unified message format for the agent to process. //! and convert them to a unified message format for the agent to process.
//!
//! # Architecture
//!
//! ```text
//! ┌─────────────────────────────────────────────────────────────────────┐
//! │ ChannelManager │
//! │ │
//! │ ┌─────────────┐ ┌─────────────┐ ┌─────────────┐ │
//! │ │ TuiChannel │ │ HttpChannel │ │ WasmChannel │ ... │
//! │ └──────┬──────┘ └──────┬──────┘ └──────┬──────┘ │
//! │ │ │ │ │
//! │ └─────────────────┴─────────────────┘ │
//! │ │ │
//! │ select_all (futures) │
//! │ │ │
//! │ ▼ │
//! │ MessageStream │
//! └─────────────────────────────────────────────────────────────────────┘
//! ```
//!
//! # WASM Channels
//!
//! WASM channels allow dynamic loading of channel implementations at runtime.
//! See the [`wasm`] module for details.
mod channel; mod channel;
pub mod cli; pub mod cli;
mod http; mod http;
mod manager; mod manager;
mod repl; mod repl;
pub mod wasm;
pub use channel::{Channel, IncomingMessage, MessageStream, OutgoingResponse, StatusUpdate}; pub use channel::{Channel, IncomingMessage, MessageStream, OutgoingResponse, StatusUpdate};
pub use cli::{AppEvent, TuiChannel}; pub use cli::{AppEvent, TuiChannel};
+316
View File
@@ -0,0 +1,316 @@
//! Channel-specific capabilities for WASM channels.
//!
//! Defines the capability system that controls what a WASM channel can do.
//! Channels have additional capabilities beyond tools: HTTP endpoint registration,
//! message emission, and workspace write access within their namespace.
use std::time::Duration;
use serde::{Deserialize, Serialize};
use crate::tools::wasm::{Capabilities as ToolCapabilities, RateLimitConfig};
/// Minimum allowed polling interval (30 seconds).
pub const MIN_POLL_INTERVAL_MS: u32 = 30_000;
/// Default emit rate limit.
pub const DEFAULT_EMIT_RATE_PER_MINUTE: u32 = 100;
pub const DEFAULT_EMIT_RATE_PER_HOUR: u32 = 5000;
/// Capabilities specific to WASM channels.
///
/// Extends tool capabilities with channel-specific permissions.
#[derive(Debug, Clone)]
pub struct ChannelCapabilities {
/// Base tool capabilities (HTTP, secrets, workspace_read, etc.).
pub tool_capabilities: ToolCapabilities,
/// HTTP paths this channel can register for webhooks.
/// Paths must start with "/webhook/" by convention.
pub allowed_paths: Vec<String>,
/// Whether polling is allowed for this channel.
pub allow_polling: bool,
/// Minimum poll interval in milliseconds.
/// Enforced to be at least MIN_POLL_INTERVAL_MS.
pub min_poll_interval_ms: u32,
/// Workspace prefix for this channel's storage.
/// All workspace writes are automatically prefixed.
/// Example: "channels/slack/" means writes to "state.json" become "channels/slack/state.json".
pub workspace_prefix: String,
/// Rate limiting for emit_message calls.
pub emit_rate_limit: EmitRateLimitConfig,
/// Maximum message content size in bytes.
pub max_message_size: usize,
/// Callback timeout duration.
pub callback_timeout: Duration,
}
impl Default for ChannelCapabilities {
fn default() -> Self {
Self {
tool_capabilities: ToolCapabilities::default(),
allowed_paths: Vec::new(),
allow_polling: false,
min_poll_interval_ms: MIN_POLL_INTERVAL_MS,
workspace_prefix: String::new(),
emit_rate_limit: EmitRateLimitConfig::default(),
max_message_size: 64 * 1024, // 64 KB
callback_timeout: Duration::from_secs(30),
}
}
}
impl ChannelCapabilities {
/// Create capabilities for a channel with the given name.
pub fn for_channel(name: &str) -> Self {
Self {
workspace_prefix: format!("channels/{}/", name),
..Default::default()
}
}
/// Add an allowed HTTP path.
pub fn with_path(mut self, path: impl Into<String>) -> Self {
self.allowed_paths.push(path.into());
self
}
/// Enable polling with the given minimum interval.
pub fn with_polling(mut self, min_interval_ms: u32) -> Self {
self.allow_polling = true;
self.min_poll_interval_ms = min_interval_ms.max(MIN_POLL_INTERVAL_MS);
self
}
/// Set the emit rate limit.
pub fn with_emit_rate_limit(mut self, rate_limit: EmitRateLimitConfig) -> Self {
self.emit_rate_limit = rate_limit;
self
}
/// Set the callback timeout.
pub fn with_callback_timeout(mut self, timeout: Duration) -> Self {
self.callback_timeout = timeout;
self
}
/// Set the base tool capabilities.
pub fn with_tool_capabilities(mut self, capabilities: ToolCapabilities) -> Self {
self.tool_capabilities = capabilities;
self
}
/// Check if a path is allowed for this channel.
pub fn is_path_allowed(&self, path: &str) -> bool {
self.allowed_paths.iter().any(|p| p == path)
}
/// Validate and normalize a poll interval.
///
/// Returns the interval clamped to minimum, or an error if polling is disabled.
pub fn validate_poll_interval(&self, interval_ms: u32) -> Result<u32, String> {
if !self.allow_polling {
return Err("Polling not allowed for this channel".to_string());
}
Ok(interval_ms.max(self.min_poll_interval_ms))
}
/// Prefix a workspace path for this channel.
///
/// Ensures all workspace writes are scoped to the channel's namespace.
pub fn prefix_workspace_path(&self, path: &str) -> String {
if self.workspace_prefix.is_empty() {
path.to_string()
} else {
format!("{}{}", self.workspace_prefix, path)
}
}
/// Check if a workspace path is valid for this channel.
///
/// Paths cannot escape the channel's namespace.
pub fn validate_workspace_path(&self, path: &str) -> Result<String, String> {
// Block absolute paths
if path.starts_with('/') {
return Err("Absolute paths not allowed".to_string());
}
// Block path traversal
if path.contains("..") {
return Err("Parent directory references not allowed".to_string());
}
// Block null bytes
if path.contains('\0') {
return Err("Null bytes not allowed".to_string());
}
// Prefix with channel namespace
Ok(self.prefix_workspace_path(path))
}
}
/// Configuration for an HTTP endpoint the channel wants to register.
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct HttpEndpointConfig {
/// Path to register (e.g., "/webhook/slack").
pub path: String,
/// HTTP methods to accept (e.g., ["POST"]).
pub methods: Vec<String>,
/// Whether secret validation is required.
pub require_secret: bool,
}
impl HttpEndpointConfig {
/// Create a POST webhook endpoint.
pub fn post_webhook(path: impl Into<String>) -> Self {
Self {
path: path.into(),
methods: vec!["POST".to_string()],
require_secret: true,
}
}
}
/// Polling configuration returned by the channel.
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct PollConfig {
/// Polling interval in milliseconds.
pub interval_ms: u32,
/// Whether polling is enabled.
pub enabled: bool,
}
impl Default for PollConfig {
fn default() -> Self {
Self {
interval_ms: MIN_POLL_INTERVAL_MS,
enabled: false,
}
}
}
/// Rate limiting configuration for message emission.
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct EmitRateLimitConfig {
/// Maximum messages per minute.
pub messages_per_minute: u32,
/// Maximum messages per hour.
pub messages_per_hour: u32,
}
impl Default for EmitRateLimitConfig {
fn default() -> Self {
Self {
messages_per_minute: DEFAULT_EMIT_RATE_PER_MINUTE,
messages_per_hour: DEFAULT_EMIT_RATE_PER_HOUR,
}
}
}
impl From<RateLimitConfig> for EmitRateLimitConfig {
fn from(config: RateLimitConfig) -> Self {
Self {
messages_per_minute: config.requests_per_minute,
messages_per_hour: config.requests_per_hour,
}
}
}
#[cfg(test)]
mod tests {
use crate::channels::wasm::capabilities::{
ChannelCapabilities, EmitRateLimitConfig, HttpEndpointConfig, MIN_POLL_INTERVAL_MS,
};
#[test]
fn test_default_capabilities() {
let caps = ChannelCapabilities::default();
assert!(caps.allowed_paths.is_empty());
assert!(!caps.allow_polling);
assert_eq!(caps.min_poll_interval_ms, MIN_POLL_INTERVAL_MS);
}
#[test]
fn test_for_channel() {
let caps = ChannelCapabilities::for_channel("slack");
assert_eq!(caps.workspace_prefix, "channels/slack/");
}
#[test]
fn test_path_allowed() {
let caps = ChannelCapabilities::default()
.with_path("/webhook/slack")
.with_path("/webhook/slack/events");
assert!(caps.is_path_allowed("/webhook/slack"));
assert!(caps.is_path_allowed("/webhook/slack/events"));
assert!(!caps.is_path_allowed("/webhook/telegram"));
}
#[test]
fn test_poll_interval_validation() {
let caps = ChannelCapabilities::default().with_polling(60_000);
// Valid interval
assert_eq!(caps.validate_poll_interval(90_000).unwrap(), 90_000);
// Too short, clamped to minimum
assert_eq!(caps.validate_poll_interval(1000).unwrap(), 60_000);
// Polling disabled
let no_poll_caps = ChannelCapabilities::default();
assert!(no_poll_caps.validate_poll_interval(60_000).is_err());
}
#[test]
fn test_workspace_path_validation() {
let caps = ChannelCapabilities::for_channel("slack");
// Valid path
let result = caps.validate_workspace_path("state.json");
assert_eq!(result.unwrap(), "channels/slack/state.json");
// Nested path
let result = caps.validate_workspace_path("data/users.json");
assert_eq!(result.unwrap(), "channels/slack/data/users.json");
// Block absolute paths
let result = caps.validate_workspace_path("/etc/passwd");
assert!(result.is_err());
// Block path traversal
let result = caps.validate_workspace_path("../secrets/key.txt");
assert!(result.is_err());
// Block null bytes
let result = caps.validate_workspace_path("file\0.txt");
assert!(result.is_err());
}
#[test]
fn test_http_endpoint_config() {
let endpoint = HttpEndpointConfig::post_webhook("/webhook/slack");
assert_eq!(endpoint.path, "/webhook/slack");
assert_eq!(endpoint.methods, vec!["POST"]);
assert!(endpoint.require_secret);
}
#[test]
fn test_emit_rate_limit_default() {
let limit = EmitRateLimitConfig::default();
assert_eq!(limit.messages_per_minute, 100);
assert_eq!(limit.messages_per_hour, 5000);
}
}
+83
View File
@@ -0,0 +1,83 @@
//! Error types for WASM channels.
use std::path::PathBuf;
/// Error during WASM channel operations.
#[derive(Debug, thiserror::Error)]
pub enum WasmChannelError {
#[error("Channel {name} failed to start: {reason}")]
StartupFailed { name: String, reason: String },
#[error("Channel {name} callback failed: {reason}")]
CallbackFailed { name: String, reason: String },
#[error("Channel {name} WASM execution trapped: {reason}")]
Trapped { name: String, reason: String },
#[error("Channel {name} callback '{callback}' timed out")]
Timeout { name: String, callback: String },
#[error("Channel {name} execution panicked: {reason}")]
ExecutionPanicked { name: String, reason: String },
#[error("Channel {name} emit rate limited")]
EmitRateLimited { name: String },
#[error("Channel {name} HTTP path not allowed: {path}")]
PathNotAllowed { name: String, path: String },
#[error("Channel {name} polling interval too short: {interval_ms}ms (minimum: {min_ms}ms)")]
PollIntervalTooShort {
name: String,
interval_ms: u32,
min_ms: u32,
},
#[error("Channel {name} workspace path escape attempt: {path}")]
WorkspaceEscape { name: String, path: String },
#[error("Channel {name} exhausted fuel limit ({limit})")]
FuelExhausted { name: String, limit: u64 },
#[error("IO error: {0}")]
Io(#[from] std::io::Error),
#[error("WASM file not found: {0}")]
WasmNotFound(PathBuf),
#[error("Capabilities file not found: {0}")]
CapabilitiesNotFound(PathBuf),
#[error("Invalid capabilities JSON: {0}")]
InvalidCapabilities(String),
#[error("WASM compilation error: {0}")]
Compilation(String),
#[error("WASM instantiation error: {0}")]
Instantiation(String),
#[error("Invalid channel name: {0}")]
InvalidName(String),
#[error("Channel {name} not found")]
NotFound { name: String },
#[error("Channel module missing export: {0}")]
MissingExport(String),
#[error("Invalid response from WASM: {0}")]
InvalidResponse(String),
#[error("Runtime not initialized")]
RuntimeNotInitialized,
#[error("Configuration error: {0}")]
Config(String),
}
impl From<crate::tools::wasm::WasmError> for WasmChannelError {
fn from(err: crate::tools::wasm::WasmError) -> Self {
WasmChannelError::Compilation(err.to_string())
}
}
+500
View File
@@ -0,0 +1,500 @@
//! Host state for WASM channel execution.
//!
//! Extends the base tool host state with channel-specific functionality:
//! - Message emission (queueing messages to send to the agent)
//! - Workspace write access (scoped to channel namespace)
//! - Rate limiting for message emission
use std::time::{SystemTime, UNIX_EPOCH};
use crate::channels::wasm::capabilities::{ChannelCapabilities, EmitRateLimitConfig};
use crate::channels::wasm::error::WasmChannelError;
use crate::tools::wasm::{HostState, LogLevel};
/// Maximum emitted messages per callback execution.
const MAX_EMITS_PER_EXECUTION: usize = 100;
/// Maximum message content size (64 KB).
const MAX_MESSAGE_CONTENT_SIZE: usize = 64 * 1024;
/// A message emitted by a WASM channel to be sent to the agent.
#[derive(Debug, Clone)]
pub struct EmittedMessage {
/// User identifier within the channel.
pub user_id: String,
/// Optional user display name.
pub user_name: Option<String>,
/// Message content.
pub content: String,
/// Optional thread ID for threaded conversations.
pub thread_id: Option<String>,
/// Channel-specific metadata as JSON string.
pub metadata_json: String,
/// Timestamp when the message was emitted.
pub emitted_at_millis: u64,
}
impl EmittedMessage {
/// Create a new emitted message.
pub fn new(user_id: impl Into<String>, content: impl Into<String>) -> Self {
Self {
user_id: user_id.into(),
user_name: None,
content: content.into(),
thread_id: None,
metadata_json: "{}".to_string(),
emitted_at_millis: SystemTime::now()
.duration_since(UNIX_EPOCH)
.map(|d| d.as_millis() as u64)
.unwrap_or(0),
}
}
/// Set the user name.
pub fn with_user_name(mut self, name: impl Into<String>) -> Self {
self.user_name = Some(name.into());
self
}
/// Set the thread ID.
pub fn with_thread_id(mut self, thread_id: impl Into<String>) -> Self {
self.thread_id = Some(thread_id.into());
self
}
/// Set metadata JSON.
pub fn with_metadata(mut self, metadata_json: impl Into<String>) -> Self {
self.metadata_json = metadata_json.into();
self
}
}
/// A pending workspace write operation.
#[derive(Debug, Clone)]
pub struct PendingWorkspaceWrite {
/// Full path (already prefixed with channel namespace).
pub path: String,
/// Content to write.
pub content: String,
}
/// Host state for WASM channel callbacks.
///
/// Maintains all side effects during callback execution and enforces limits.
/// This is the channel-specific equivalent of HostState for tools.
pub struct ChannelHostState {
/// Base tool host state (logging, time, HTTP, etc.).
base: HostState,
/// Channel name (for error messages).
channel_name: String,
/// Channel capabilities.
capabilities: ChannelCapabilities,
/// Emitted messages (queued for delivery).
emitted_messages: Vec<EmittedMessage>,
/// Pending workspace writes.
pending_writes: Vec<PendingWorkspaceWrite>,
/// Emit count for rate limiting within this execution.
emit_count: u32,
/// Whether emit is still allowed (false after rate limit hit).
emit_enabled: bool,
/// Count of emits dropped due to rate limiting.
emits_dropped: usize,
}
impl std::fmt::Debug for ChannelHostState {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
f.debug_struct("ChannelHostState")
.field("channel_name", &self.channel_name)
.field("emitted_messages_count", &self.emitted_messages.len())
.field("pending_writes_count", &self.pending_writes.len())
.field("emit_count", &self.emit_count)
.field("emit_enabled", &self.emit_enabled)
.field("emits_dropped", &self.emits_dropped)
.finish()
}
}
impl ChannelHostState {
/// Create a new channel host state.
pub fn new(channel_name: impl Into<String>, capabilities: ChannelCapabilities) -> Self {
let base = HostState::new(capabilities.tool_capabilities.clone());
Self {
base,
channel_name: channel_name.into(),
capabilities,
emitted_messages: Vec::new(),
pending_writes: Vec::new(),
emit_count: 0,
emit_enabled: true,
emits_dropped: 0,
}
}
/// Get the channel name.
pub fn channel_name(&self) -> &str {
&self.channel_name
}
/// Get the capabilities.
pub fn capabilities(&self) -> &ChannelCapabilities {
&self.capabilities
}
/// Get the base host state for tool capabilities.
pub fn base(&self) -> &HostState {
&self.base
}
/// Get mutable access to the base host state.
pub fn base_mut(&mut self) -> &mut HostState {
&mut self.base
}
/// Emit a message from the channel.
///
/// Messages are queued and delivered after callback execution completes.
/// Rate limiting is enforced per-execution and globally.
pub fn emit_message(&mut self, msg: EmittedMessage) -> Result<(), WasmChannelError> {
// Check per-execution limit
if !self.emit_enabled {
self.emits_dropped += 1;
return Ok(()); // Silently drop, don't fail execution
}
if self.emitted_messages.len() >= MAX_EMITS_PER_EXECUTION {
self.emit_enabled = false;
self.emits_dropped += 1;
tracing::warn!(
channel = %self.channel_name,
limit = MAX_EMITS_PER_EXECUTION,
"Channel emit limit reached, further messages dropped"
);
return Ok(());
}
// Validate message content size
if msg.content.len() > MAX_MESSAGE_CONTENT_SIZE {
tracing::warn!(
channel = %self.channel_name,
size = msg.content.len(),
max = MAX_MESSAGE_CONTENT_SIZE,
"Message content too large, truncating"
);
let mut truncated = msg.content[..MAX_MESSAGE_CONTENT_SIZE].to_string();
truncated.push_str("... (truncated)");
let msg = EmittedMessage {
content: truncated,
..msg
};
self.emitted_messages.push(msg);
} else {
self.emitted_messages.push(msg);
}
self.emit_count += 1;
Ok(())
}
/// Take all emitted messages (clears the queue).
pub fn take_emitted_messages(&mut self) -> Vec<EmittedMessage> {
std::mem::take(&mut self.emitted_messages)
}
/// Get the number of emitted messages.
pub fn emitted_count(&self) -> usize {
self.emitted_messages.len()
}
/// Get the number of emits dropped due to rate limiting.
pub fn emits_dropped(&self) -> usize {
self.emits_dropped
}
/// Write to workspace (scoped to channel namespace).
///
/// Writes are queued and committed after callback execution completes.
pub fn workspace_write(&mut self, path: &str, content: String) -> Result<(), WasmChannelError> {
// Validate and prefix path
let full_path = self
.capabilities
.validate_workspace_path(path)
.map_err(|reason| WasmChannelError::WorkspaceEscape {
name: self.channel_name.clone(),
path: reason,
})?;
self.pending_writes.push(PendingWorkspaceWrite {
path: full_path,
content,
});
Ok(())
}
/// Take all pending workspace writes (clears the queue).
pub fn take_pending_writes(&mut self) -> Vec<PendingWorkspaceWrite> {
std::mem::take(&mut self.pending_writes)
}
/// Get the number of pending workspace writes.
pub fn pending_writes_count(&self) -> usize {
self.pending_writes.len()
}
/// Log a message (delegates to base).
pub fn log(
&mut self,
level: LogLevel,
message: String,
) -> Result<(), crate::tools::wasm::WasmError> {
self.base.log(level, message)
}
/// Get current timestamp in milliseconds (delegates to base).
pub fn now_millis(&self) -> u64 {
self.base.now_millis()
}
/// Read from workspace (delegates to base).
pub fn workspace_read(
&self,
path: &str,
) -> Result<Option<String>, crate::tools::wasm::WasmError> {
// Prefix the path with channel namespace before reading
let full_path = self.capabilities.prefix_workspace_path(path);
self.base.workspace_read(&full_path)
}
/// Check if a secret exists (delegates to base).
pub fn secret_exists(&self, name: &str) -> bool {
self.base.secret_exists(name)
}
/// Check if HTTP is allowed (delegates to base).
pub fn check_http_allowed(&self, url: &str, method: &str) -> Result<(), String> {
self.base.check_http_allowed(url, method)
}
/// Record an HTTP request (delegates to base).
pub fn record_http_request(&mut self) -> Result<(), String> {
self.base.record_http_request()
}
/// Take logs (delegates to base).
pub fn take_logs(&mut self) -> Vec<crate::tools::wasm::LogEntry> {
self.base.take_logs()
}
}
/// Rate limiter for channel message emission.
///
/// Tracks emission rates across multiple executions.
pub struct ChannelEmitRateLimiter {
config: EmitRateLimitConfig,
minute_window: RateWindow,
hour_window: RateWindow,
}
struct RateWindow {
count: u32,
window_start: u64,
window_duration_ms: u64,
}
impl RateWindow {
fn new(duration_ms: u64) -> Self {
Self {
count: 0,
window_start: 0,
window_duration_ms: duration_ms,
}
}
fn check_and_record(&mut self, now_ms: u64, limit: u32) -> bool {
// Reset window if expired
if now_ms.saturating_sub(self.window_start) > self.window_duration_ms {
self.count = 0;
self.window_start = now_ms;
}
if self.count >= limit {
return false;
}
self.count += 1;
true
}
}
#[allow(dead_code)]
impl ChannelEmitRateLimiter {
/// Create a new rate limiter with the given config.
pub fn new(config: EmitRateLimitConfig) -> Self {
Self {
config,
minute_window: RateWindow::new(60_000), // 1 minute
hour_window: RateWindow::new(3_600_000), // 1 hour
}
}
/// Check if an emit is allowed and record it if so.
///
/// Returns true if the emit is allowed, false if rate limited.
pub fn check_and_record(&mut self) -> bool {
let now = SystemTime::now()
.duration_since(UNIX_EPOCH)
.map(|d| d.as_millis() as u64)
.unwrap_or(0);
// Check both windows
let minute_ok = self
.minute_window
.check_and_record(now, self.config.messages_per_minute);
let hour_ok = self
.hour_window
.check_and_record(now, self.config.messages_per_hour);
minute_ok && hour_ok
}
/// Get the current emission count for the minute window.
pub fn minute_count(&self) -> u32 {
self.minute_window.count
}
/// Get the current emission count for the hour window.
pub fn hour_count(&self) -> u32 {
self.hour_window.count
}
}
#[cfg(test)]
mod tests {
use crate::channels::wasm::capabilities::{ChannelCapabilities, EmitRateLimitConfig};
use crate::channels::wasm::host::{
ChannelEmitRateLimiter, ChannelHostState, EmittedMessage, MAX_EMITS_PER_EXECUTION,
};
#[test]
fn test_emit_message_basic() {
let caps = ChannelCapabilities::for_channel("test");
let mut state = ChannelHostState::new("test", caps);
let msg = EmittedMessage::new("user123", "Hello, world!");
state.emit_message(msg).unwrap();
assert_eq!(state.emitted_count(), 1);
let messages = state.take_emitted_messages();
assert_eq!(messages.len(), 1);
assert_eq!(messages[0].user_id, "user123");
assert_eq!(messages[0].content, "Hello, world!");
// Queue should be cleared
assert_eq!(state.emitted_count(), 0);
}
#[test]
fn test_emit_message_with_metadata() {
let caps = ChannelCapabilities::for_channel("test");
let mut state = ChannelHostState::new("test", caps);
let msg = EmittedMessage::new("user123", "Hello")
.with_user_name("John Doe")
.with_thread_id("thread-1")
.with_metadata(r#"{"key": "value"}"#);
state.emit_message(msg).unwrap();
let messages = state.take_emitted_messages();
assert_eq!(messages[0].user_name, Some("John Doe".to_string()));
assert_eq!(messages[0].thread_id, Some("thread-1".to_string()));
assert_eq!(messages[0].metadata_json, r#"{"key": "value"}"#);
}
#[test]
fn test_emit_per_execution_limit() {
let caps = ChannelCapabilities::for_channel("test");
let mut state = ChannelHostState::new("test", caps);
// Fill up to limit
for i in 0..MAX_EMITS_PER_EXECUTION {
let msg = EmittedMessage::new("user", format!("Message {}", i));
state.emit_message(msg).unwrap();
}
// This should be dropped silently
let msg = EmittedMessage::new("user", "Should be dropped");
state.emit_message(msg).unwrap();
assert_eq!(state.emitted_count(), MAX_EMITS_PER_EXECUTION);
assert_eq!(state.emits_dropped(), 1);
}
#[test]
fn test_workspace_write_prefixing() {
let caps = ChannelCapabilities::for_channel("slack");
let mut state = ChannelHostState::new("slack", caps);
state
.workspace_write("state.json", "{}".to_string())
.unwrap();
let writes = state.take_pending_writes();
assert_eq!(writes.len(), 1);
assert_eq!(writes[0].path, "channels/slack/state.json");
}
#[test]
fn test_workspace_write_path_traversal_blocked() {
let caps = ChannelCapabilities::for_channel("slack");
let mut state = ChannelHostState::new("slack", caps);
// Try to escape namespace
let result = state.workspace_write("../secrets.json", "{}".to_string());
assert!(result.is_err());
// Absolute path
let result = state.workspace_write("/etc/passwd", "{}".to_string());
assert!(result.is_err());
}
#[test]
fn test_rate_limiter_basic() {
let config = EmitRateLimitConfig {
messages_per_minute: 10,
messages_per_hour: 100,
};
let mut limiter = ChannelEmitRateLimiter::new(config);
// Should allow 10 messages
for _ in 0..10 {
assert!(limiter.check_and_record());
}
// 11th should be blocked
assert!(!limiter.check_and_record());
}
#[test]
fn test_channel_name() {
let caps = ChannelCapabilities::for_channel("telegram");
let state = ChannelHostState::new("telegram", caps);
assert_eq!(state.channel_name(), "telegram");
}
}
+360
View File
@@ -0,0 +1,360 @@
//! WASM channel loader for loading channels from files or directories.
//!
//! Loads WASM channel modules from the filesystem (default: ~/.near-agent/channels/).
//! Each channel consists of:
//! - `<name>.wasm` - The compiled WASM component
//! - `<name>.capabilities.json` - Channel capabilities and configuration
use std::collections::HashMap;
use std::path::{Path, PathBuf};
use std::sync::Arc;
use tokio::fs;
use crate::channels::wasm::capabilities::ChannelCapabilities;
use crate::channels::wasm::error::WasmChannelError;
use crate::channels::wasm::runtime::WasmChannelRuntime;
use crate::channels::wasm::schema::ChannelCapabilitiesFile;
use crate::channels::wasm::wrapper::WasmChannel;
/// Loads WASM channels from the filesystem.
pub struct WasmChannelLoader {
runtime: Arc<WasmChannelRuntime>,
}
impl WasmChannelLoader {
/// Create a new loader with the given runtime.
pub fn new(runtime: Arc<WasmChannelRuntime>) -> Self {
Self { runtime }
}
/// Load a single WASM channel from a file pair.
///
/// Expects:
/// - `wasm_path`: Path to the `.wasm` file
/// - `capabilities_path`: Path to the `.capabilities.json` file (optional)
///
/// If no capabilities file is provided, the channel gets minimal capabilities.
pub async fn load_from_files(
&self,
name: &str,
wasm_path: &Path,
capabilities_path: Option<&Path>,
) -> Result<WasmChannel, WasmChannelError> {
// Validate name
if name.is_empty() || name.contains('/') || name.contains('\\') || name.contains("..") {
return Err(WasmChannelError::InvalidName(name.to_string()));
}
// Read WASM bytes
if !wasm_path.exists() {
return Err(WasmChannelError::WasmNotFound(wasm_path.to_path_buf()));
}
let wasm_bytes = fs::read(wasm_path).await?;
// Read capabilities file
let (capabilities, config_json, description) = if let Some(cap_path) = capabilities_path {
if cap_path.exists() {
let cap_bytes = fs::read(cap_path).await?;
let cap_file = ChannelCapabilitiesFile::from_bytes(&cap_bytes)
.map_err(|e| WasmChannelError::InvalidCapabilities(e.to_string()))?;
let caps = cap_file.to_capabilities();
let config = cap_file.config_json();
let desc = cap_file.description.clone();
(caps, config, desc)
} else {
tracing::warn!(
path = %cap_path.display(),
"Capabilities file not found, using defaults"
);
(
ChannelCapabilities::for_channel(name),
"{}".to_string(),
None,
)
}
} else {
(
ChannelCapabilities::for_channel(name),
"{}".to_string(),
None,
)
};
// Prepare the module
let prepared = self
.runtime
.prepare(name, &wasm_bytes, None, description)
.await?;
// Create the channel
let channel = WasmChannel::new(self.runtime.clone(), prepared, capabilities, config_json);
tracing::info!(
name = name,
wasm_path = %wasm_path.display(),
"Loaded WASM channel from file"
);
Ok(channel)
}
/// Load all WASM channels from a directory.
///
/// Scans the directory for `*.wasm` files and loads each one, looking for
/// a matching `*.capabilities.json` sidecar file.
///
/// # Directory Layout
///
/// ```text
/// channels/
/// ├── slack.wasm <- Channel WASM component
/// ├── slack.capabilities.json <- Capabilities (optional)
/// ├── telegram.wasm
/// └── telegram.capabilities.json
/// ```
pub async fn load_from_dir(&self, dir: &Path) -> Result<LoadResults, WasmChannelError> {
if !dir.is_dir() {
return Err(WasmChannelError::Io(std::io::Error::new(
std::io::ErrorKind::NotADirectory,
format!("{} is not a directory", dir.display()),
)));
}
let mut results = LoadResults::default();
let mut entries = fs::read_dir(dir).await?;
while let Some(entry) = entries.next_entry().await? {
let path = entry.path();
// Only process .wasm files
if path.extension().and_then(|e| e.to_str()) != Some("wasm") {
continue;
}
// Extract channel name from filename
let name = match path.file_stem().and_then(|s| s.to_str()) {
Some(n) => n.to_string(),
None => {
results.errors.push((
path.clone(),
WasmChannelError::InvalidName("invalid filename".to_string()),
));
continue;
}
};
// Look for sidecar capabilities file
let cap_path = path.with_extension("capabilities.json");
let cap_path_option = if cap_path.exists() {
Some(cap_path.as_path())
} else {
None
};
match self.load_from_files(&name, &path, cap_path_option).await {
Ok(channel) => {
results.loaded.push(channel);
}
Err(e) => {
tracing::error!(
name = name,
path = %path.display(),
error = %e,
"Failed to load WASM channel"
);
results.errors.push((path, e));
}
}
}
if !results.loaded.is_empty() {
tracing::info!(
count = results.loaded.len(),
channels = ?results.loaded.iter().map(|c| c.channel_name()).collect::<Vec<_>>(),
"Loaded WASM channels from directory"
);
}
Ok(results)
}
}
/// Results from loading multiple channels.
#[derive(Default)]
pub struct LoadResults {
/// Successfully loaded channels.
pub loaded: Vec<WasmChannel>,
/// Errors encountered (path, error).
pub errors: Vec<(PathBuf, WasmChannelError)>,
}
impl LoadResults {
/// Check if all channels loaded successfully.
pub fn all_succeeded(&self) -> bool {
self.errors.is_empty()
}
/// Get the count of successfully loaded channels.
pub fn success_count(&self) -> usize {
self.loaded.len()
}
/// Get the count of failed channels.
pub fn error_count(&self) -> usize {
self.errors.len()
}
/// Take ownership of loaded channels.
pub fn take_channels(self) -> Vec<WasmChannel> {
self.loaded
}
}
/// Discover WASM channel files in a directory without loading them.
///
/// Returns a map of channel name -> (wasm_path, capabilities_path).
#[allow(dead_code)]
pub async fn discover_channels(
dir: &Path,
) -> Result<HashMap<String, DiscoveredChannel>, std::io::Error> {
let mut channels = HashMap::new();
if !dir.is_dir() {
return Ok(channels);
}
let mut entries = fs::read_dir(dir).await?;
while let Some(entry) = entries.next_entry().await? {
let path = entry.path();
if path.extension().and_then(|e| e.to_str()) != Some("wasm") {
continue;
}
let name = match path.file_stem().and_then(|s| s.to_str()) {
Some(n) => n.to_string(),
None => continue,
};
let cap_path = path.with_extension("capabilities.json");
channels.insert(
name,
DiscoveredChannel {
wasm_path: path,
capabilities_path: if cap_path.exists() {
Some(cap_path)
} else {
None
},
},
);
}
Ok(channels)
}
/// A discovered WASM channel (not yet loaded).
#[derive(Debug)]
pub struct DiscoveredChannel {
/// Path to the WASM file.
pub wasm_path: PathBuf,
/// Path to the capabilities file (if present).
pub capabilities_path: Option<PathBuf>,
}
/// Get the default channels directory path.
///
/// Returns ~/.near-agent/channels/
#[allow(dead_code)]
pub fn default_channels_dir() -> PathBuf {
dirs::home_dir()
.unwrap_or_else(|| PathBuf::from("."))
.join(".near-agent")
.join("channels")
}
#[cfg(test)]
mod tests {
use std::io::Write;
use tempfile::TempDir;
use crate::channels::wasm::loader::{WasmChannelLoader, discover_channels};
use crate::channels::wasm::runtime::{WasmChannelRuntime, WasmChannelRuntimeConfig};
use std::sync::Arc;
#[tokio::test]
async fn test_discover_channels_empty_dir() {
let dir = TempDir::new().unwrap();
let channels = discover_channels(dir.path()).await.unwrap();
assert!(channels.is_empty());
}
#[tokio::test]
async fn test_discover_channels_with_wasm() {
let dir = TempDir::new().unwrap();
// Create a fake .wasm file
let wasm_path = dir.path().join("slack.wasm");
std::fs::File::create(&wasm_path).unwrap();
let channels = discover_channels(dir.path()).await.unwrap();
assert_eq!(channels.len(), 1);
assert!(channels.contains_key("slack"));
assert!(channels["slack"].capabilities_path.is_none());
}
#[tokio::test]
async fn test_discover_channels_with_capabilities() {
let dir = TempDir::new().unwrap();
// Create wasm and capabilities files
std::fs::File::create(dir.path().join("telegram.wasm")).unwrap();
let mut cap_file =
std::fs::File::create(dir.path().join("telegram.capabilities.json")).unwrap();
cap_file.write_all(b"{}").unwrap();
let channels = discover_channels(dir.path()).await.unwrap();
assert_eq!(channels.len(), 1);
assert!(channels["telegram"].capabilities_path.is_some());
}
#[tokio::test]
async fn test_discover_channels_ignores_non_wasm() {
let dir = TempDir::new().unwrap();
// Create non-wasm files
std::fs::File::create(dir.path().join("readme.md")).unwrap();
std::fs::File::create(dir.path().join("config.json")).unwrap();
std::fs::File::create(dir.path().join("channel.wasm")).unwrap();
let channels = discover_channels(dir.path()).await.unwrap();
assert_eq!(channels.len(), 1);
assert!(channels.contains_key("channel"));
}
#[tokio::test]
async fn test_loader_invalid_name() {
let config = WasmChannelRuntimeConfig::for_testing();
let runtime = Arc::new(WasmChannelRuntime::new(config).unwrap());
let loader = WasmChannelLoader::new(runtime);
let dir = TempDir::new().unwrap();
let wasm_path = dir.path().join("test.wasm");
// Invalid name with path separator
let result = loader.load_from_files("../escape", &wasm_path, None).await;
assert!(result.is_err());
// Empty name
let result = loader.load_from_files("", &wasm_path, None).await;
assert!(result.is_err());
}
}
+102
View File
@@ -0,0 +1,102 @@
//! WASM-extensible channel system.
//!
//! This module provides a runtime for executing WASM-based channels using a
//! Host-Managed Event Loop pattern. The host (Rust) manages infrastructure
//! (HTTP server, polling), while WASM modules define channel behavior through
//! callbacks.
//!
//! # Architecture
//!
//! ```text
//! ┌─────────────────────────────────────────────────────────────────────────────────┐
//! │ Host-Managed Event Loop │
//! │ │
//! │ ┌─────────────┐ ┌──────────────┐ ┌──────────────┐ │
//! │ │ HTTP │ │ Polling │ │ Timer │ │
//! │ │ Router │ │ Scheduler │ │ Scheduler │ │
//! │ └──────┬──────┘ └──────┬───────┘ └──────┬───────┘ │
//! │ │ │ │ │
//! │ └───────────────────┴────────────────────┘ │
//! │ │ │
//! │ ▼ │
//! │ ┌─────────────────┐ │
//! │ │ Event Router │ │
//! │ └────────┬────────┘ │
//! │ │ │
//! │ ┌──────────────────┼──────────────────┐ │
//! │ ▼ ▼ ▼ │
//! │ ┌─────────────┐ ┌─────────────┐ ┌─────────────┐ │
//! │ │ on_http_req │ │ on_poll │ │ on_respond │ WASM Exports │
//! │ └─────────────┘ └─────────────┘ └─────────────┘ │
//! │ │ │ │ │
//! │ └──────────────────┴──────────────────┘ │
//! │ │ │
//! │ ▼ │
//! │ ┌─────────────────┐ │
//! │ │ Host Imports │ │
//! │ │ emit_message │──────────▶ MessageStream │
//! │ │ http_request │ │
//! │ │ log, etc. │ │
//! │ └─────────────────┘ │
//! └─────────────────────────────────────────────────────────────────────────────────┘
//! ```
//!
//! # Key Design Decisions
//!
//! 1. **Fresh Instance Per Callback** (NEAR Pattern) - Full isolation, no shared mutable state
//! 2. **Host Manages Infrastructure** - HTTP server, polling, timing in Rust
//! 3. **WASM Defines Behavior** - Callbacks for events, message parsing, response handling
//! 4. **Reuse Tool Runtime** - Share Wasmtime engine, extend capabilities
//!
//! # Security Model
//!
//! | Threat | Mitigation |
//! |--------|------------|
//! | Path hijacking | `allowed_paths` restricts registrable endpoints |
//! | Token exposure | Injected at host boundary, WASM never sees |
//! | State pollution | Fresh instance per callback |
//! | Workspace escape | Paths prefixed with `channels/<name>/` |
//! | Message spam | Rate limiting on `emit_message` |
//! | Resource exhaustion | Fuel metering, memory limits, callback timeout |
//! | Polling abuse | Minimum 30s interval enforced |
//!
//! # Example Usage
//!
//! ```ignore
//! use near_agent::channels::wasm::{WasmChannelLoader, WasmChannelRuntime};
//!
//! // Create runtime (can share engine with tool runtime)
//! let runtime = WasmChannelRuntime::new(config)?;
//!
//! // Load channels from directory
//! let loader = WasmChannelLoader::new(runtime);
//! let channels = loader.load_from_dir(Path::new("~/.near-agent/channels/")).await?;
//!
//! // Add to channel manager
//! for channel in channels {
//! manager.add(Box::new(channel));
//! }
//! ```
mod capabilities;
mod error;
mod host;
mod loader;
mod router;
mod runtime;
mod schema;
mod wrapper;
// Core types
pub use capabilities::{ChannelCapabilities, EmitRateLimitConfig, HttpEndpointConfig, PollConfig};
pub use error::WasmChannelError;
pub use host::{ChannelEmitRateLimiter, ChannelHostState, EmittedMessage};
pub use loader::{
DiscoveredChannel, LoadResults, WasmChannelLoader, default_channels_dir, discover_channels,
};
pub use router::{
RegisteredEndpoint, WasmChannelRouter, WasmChannelServer, create_wasm_channel_router,
};
pub use runtime::{PreparedChannelModule, WasmChannelRuntime, WasmChannelRuntimeConfig};
pub use schema::{ChannelCapabilitiesFile, ChannelConfig};
pub use wrapper::{HttpResponse, WasmChannel};
+480
View File
@@ -0,0 +1,480 @@
//! HTTP router for WASM channel webhooks.
//!
//! Routes incoming HTTP requests to the appropriate WASM channel based on
//! registered paths. Handles secret validation at the host level.
use std::collections::HashMap;
use std::net::SocketAddr;
use std::sync::Arc;
use axum::{
Json, Router,
body::Bytes,
extract::{Path, Query, State},
http::{HeaderMap, Method, StatusCode},
response::IntoResponse,
routing::{get, post},
};
use serde::{Deserialize, Serialize};
use tokio::sync::RwLock;
use crate::channels::wasm::wrapper::WasmChannel;
/// A registered HTTP endpoint for a WASM channel.
#[derive(Debug, Clone)]
pub struct RegisteredEndpoint {
/// Channel name that owns this endpoint.
pub channel_name: String,
/// HTTP path (e.g., "/webhook/slack").
pub path: String,
/// Allowed HTTP methods.
pub methods: Vec<String>,
/// Whether secret validation is required.
pub require_secret: bool,
}
/// Router for WASM channel HTTP endpoints.
pub struct WasmChannelRouter {
/// Registered channels by name.
channels: RwLock<HashMap<String, Arc<WasmChannel>>>,
/// Path to channel mapping for fast lookup.
path_to_channel: RwLock<HashMap<String, String>>,
/// Expected webhook secrets by channel name.
secrets: RwLock<HashMap<String, String>>,
}
impl WasmChannelRouter {
/// Create a new router.
pub fn new() -> Self {
Self {
channels: RwLock::new(HashMap::new()),
path_to_channel: RwLock::new(HashMap::new()),
secrets: RwLock::new(HashMap::new()),
}
}
/// Register a channel with its endpoints.
pub async fn register(
&self,
channel: Arc<WasmChannel>,
endpoints: Vec<RegisteredEndpoint>,
secret: Option<String>,
) {
let name = channel.channel_name().to_string();
// Store the channel
self.channels.write().await.insert(name.clone(), channel);
// Register path mappings
let mut path_map = self.path_to_channel.write().await;
for endpoint in endpoints {
path_map.insert(endpoint.path.clone(), name.clone());
tracing::info!(
channel = %name,
path = %endpoint.path,
methods = ?endpoint.methods,
"Registered WASM channel HTTP endpoint"
);
}
// Store secret if provided
if let Some(s) = secret {
self.secrets.write().await.insert(name, s);
}
}
/// Unregister a channel and its endpoints.
pub async fn unregister(&self, channel_name: &str) {
self.channels.write().await.remove(channel_name);
self.secrets.write().await.remove(channel_name);
// Remove all paths for this channel
self.path_to_channel
.write()
.await
.retain(|_, name| name != channel_name);
tracing::info!(
channel = %channel_name,
"Unregistered WASM channel"
);
}
/// Get the channel for a given path.
pub async fn get_channel_for_path(&self, path: &str) -> Option<Arc<WasmChannel>> {
let path_map = self.path_to_channel.read().await;
let channel_name = path_map.get(path)?;
self.channels.read().await.get(channel_name).cloned()
}
/// Validate a secret for a channel.
pub async fn validate_secret(&self, channel_name: &str, provided: &str) -> bool {
let secrets = self.secrets.read().await;
match secrets.get(channel_name) {
Some(expected) => expected == provided,
None => true, // No secret required
}
}
/// Check if a channel requires a secret.
pub async fn requires_secret(&self, channel_name: &str) -> bool {
self.secrets.read().await.contains_key(channel_name)
}
/// List all registered channels.
pub async fn list_channels(&self) -> Vec<String> {
self.channels.read().await.keys().cloned().collect()
}
/// List all registered paths.
pub async fn list_paths(&self) -> Vec<String> {
self.path_to_channel.read().await.keys().cloned().collect()
}
}
impl Default for WasmChannelRouter {
fn default() -> Self {
Self::new()
}
}
/// Shared state for the HTTP server.
#[allow(dead_code)]
#[derive(Clone)]
pub struct RouterState {
router: Arc<WasmChannelRouter>,
}
impl RouterState {
pub fn new(router: Arc<WasmChannelRouter>) -> Self {
Self { router }
}
}
/// Webhook request body for WASM channels.
#[allow(dead_code)]
#[derive(Debug, Deserialize)]
pub struct WasmWebhookRequest {
/// Optional secret for authentication.
#[serde(default)]
pub secret: Option<String>,
}
/// Health response.
#[allow(dead_code)]
#[derive(Debug, Serialize)]
struct HealthResponse {
status: String,
channels: Vec<String>,
}
/// Handler for health check endpoint.
#[allow(dead_code)]
async fn health_handler(State(state): State<RouterState>) -> impl IntoResponse {
let channels = state.router.list_channels().await;
Json(HealthResponse {
status: "healthy".to_string(),
channels,
})
}
/// Generic webhook handler that routes to the appropriate WASM channel.
#[allow(dead_code)]
async fn webhook_handler(
State(state): State<RouterState>,
method: Method,
Path(path): Path<String>,
Query(query): Query<HashMap<String, String>>,
headers: HeaderMap,
body: Bytes,
) -> impl IntoResponse {
let full_path = format!("/webhook/{}", path);
// Find the channel for this path
let channel = match state.router.get_channel_for_path(&full_path).await {
Some(c) => c,
None => {
return (
StatusCode::NOT_FOUND,
Json(serde_json::json!({
"error": "Channel not found for path",
"path": full_path
})),
);
}
};
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())
});
match provided_secret {
Some(secret) => {
if !state.router.validate_secret(channel_name, &secret).await {
return (
StatusCode::UNAUTHORIZED,
Json(serde_json::json!({
"error": "Invalid webhook secret"
})),
);
}
}
None => {
return (
StatusCode::UNAUTHORIZED,
Json(serde_json::json!({
"error": "Webhook secret required"
})),
);
}
}
}
// Convert headers to HashMap
let headers_map: HashMap<String, String> = headers
.iter()
.filter_map(|(k, v)| {
v.to_str()
.ok()
.map(|v| (k.as_str().to_string(), v.to_string()))
})
.collect();
// Call the WASM channel
let secret_validated = state.router.requires_secret(channel_name).await;
match channel
.call_on_http_request(
method.as_str(),
&full_path,
&headers_map,
&query,
&body,
secret_validated,
)
.await
{
Ok(response) => {
let status =
StatusCode::from_u16(response.status).unwrap_or(StatusCode::INTERNAL_SERVER_ERROR);
// Build response with headers
let body_json: serde_json::Value = serde_json::from_slice(&response.body)
.unwrap_or_else(|_| {
serde_json::json!({
"raw": String::from_utf8_lossy(&response.body).to_string()
})
});
(status, Json(body_json))
}
Err(e) => {
tracing::error!(
channel = %channel_name,
error = %e,
"WASM channel callback failed"
);
(
StatusCode::INTERNAL_SERVER_ERROR,
Json(serde_json::json!({
"error": "Channel callback failed",
"details": e.to_string()
})),
)
}
}
}
/// 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<WasmChannelRouter>) -> 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))
.with_state(state)
}
/// HTTP server for WASM channel webhooks.
#[allow(dead_code)]
pub struct WasmChannelServer {
router: Arc<WasmChannelRouter>,
}
#[allow(dead_code)]
impl WasmChannelServer {
/// Create a new server.
pub fn new(router: Arc<WasmChannelRouter>) -> Self {
Self { router }
}
/// Start the HTTP server.
///
/// Returns a handle that can be used to shut down the server.
pub async fn start(
&self,
addr: SocketAddr,
) -> Result<tokio::task::JoinHandle<()>, std::io::Error> {
let app = create_wasm_channel_router(self.router.clone());
let listener = tokio::net::TcpListener::bind(addr).await?;
tracing::info!(
addr = %addr,
"WASM channel HTTP server started"
);
let handle = tokio::spawn(async move {
if let Err(e) = axum::serve(listener, app).await {
tracing::error!("WASM channel HTTP server error: {}", e);
}
});
Ok(handle)
}
}
#[cfg(test)]
mod tests {
use std::sync::Arc;
use crate::channels::wasm::capabilities::ChannelCapabilities;
use crate::channels::wasm::router::{RegisteredEndpoint, WasmChannelRouter};
use crate::channels::wasm::runtime::{
PreparedChannelModule, WasmChannelRuntime, WasmChannelRuntimeConfig,
};
use crate::channels::wasm::wrapper::WasmChannel;
use crate::tools::wasm::ResourceLimits;
fn create_test_channel(name: &str) -> Arc<WasmChannel> {
let config = WasmChannelRuntimeConfig::for_testing();
let runtime = Arc::new(WasmChannelRuntime::new(config).unwrap());
let prepared = Arc::new(PreparedChannelModule {
name: name.to_string(),
description: format!("Test channel: {}", name),
component_bytes: Vec::new(),
limits: ResourceLimits::default(),
});
let capabilities =
ChannelCapabilities::for_channel(name).with_path(format!("/webhook/{}", name));
Arc::new(WasmChannel::new(
runtime,
prepared,
capabilities,
"{}".to_string(),
))
}
#[tokio::test]
async fn test_router_register_and_lookup() {
let router = WasmChannelRouter::new();
let channel = create_test_channel("slack");
let endpoints = vec![RegisteredEndpoint {
channel_name: "slack".to_string(),
path: "/webhook/slack".to_string(),
methods: vec!["POST".to_string()],
require_secret: true,
}];
router
.register(channel, endpoints, Some("secret123".to_string()))
.await;
// Should find channel by path
let found = router.get_channel_for_path("/webhook/slack").await;
assert!(found.is_some());
assert_eq!(found.unwrap().channel_name(), "slack");
// Should not find non-existent path
let not_found = router.get_channel_for_path("/webhook/telegram").await;
assert!(not_found.is_none());
}
#[tokio::test]
async fn test_router_secret_validation() {
let router = WasmChannelRouter::new();
let channel = create_test_channel("slack");
router
.register(channel, vec![], Some("secret123".to_string()))
.await;
// Correct secret
assert!(router.validate_secret("slack", "secret123").await);
// Wrong secret
assert!(!router.validate_secret("slack", "wrong").await);
// Channel without secret always validates
let channel2 = create_test_channel("telegram");
router.register(channel2, vec![], None).await;
assert!(router.validate_secret("telegram", "anything").await);
}
#[tokio::test]
async fn test_router_unregister() {
let router = WasmChannelRouter::new();
let channel = create_test_channel("slack");
let endpoints = vec![RegisteredEndpoint {
channel_name: "slack".to_string(),
path: "/webhook/slack".to_string(),
methods: vec!["POST".to_string()],
require_secret: false,
}];
router.register(channel, endpoints, None).await;
// Should exist
assert!(
router
.get_channel_for_path("/webhook/slack")
.await
.is_some()
);
// Unregister
router.unregister("slack").await;
// Should no longer exist
assert!(
router
.get_channel_for_path("/webhook/slack")
.await
.is_none()
);
}
#[tokio::test]
async fn test_router_list_channels() {
let router = WasmChannelRouter::new();
let channel1 = create_test_channel("slack");
let channel2 = create_test_channel("telegram");
router.register(channel1, vec![], None).await;
router.register(channel2, vec![], None).await;
let channels = router.list_channels().await;
assert_eq!(channels.len(), 2);
assert!(channels.contains(&"slack".to_string()));
assert!(channels.contains(&"telegram".to_string()));
}
}
+285
View File
@@ -0,0 +1,285 @@
//! WASM channel runtime for managing compiled channel components.
//!
//! Similar to tool runtime, follows the principle: compile once at registration,
//! instantiate fresh per callback execution.
use std::collections::HashMap;
use std::path::PathBuf;
use std::sync::Arc;
use std::time::Duration;
use tokio::sync::RwLock;
use wasmtime::{Config, Engine, OptLevel};
use crate::channels::wasm::error::WasmChannelError;
use crate::tools::wasm::{FuelConfig, ResourceLimits};
/// Configuration for the WASM channel runtime.
#[derive(Debug, Clone)]
pub struct WasmChannelRuntimeConfig {
/// Default resource limits for channels.
pub default_limits: ResourceLimits,
/// Fuel configuration.
pub fuel_config: FuelConfig,
/// Whether to cache compiled modules.
pub cache_compiled: bool,
/// Directory for compiled module cache.
pub cache_dir: Option<PathBuf>,
/// Cranelift optimization level.
pub optimization_level: OptLevel,
/// Default callback timeout.
pub callback_timeout: Duration,
}
impl Default for WasmChannelRuntimeConfig {
fn default() -> Self {
Self {
default_limits: ResourceLimits {
// Channels may need more memory for message buffering
memory_bytes: 50 * 1024 * 1024, // 50 MB
fuel: 10_000_000,
timeout: Duration::from_secs(60),
},
fuel_config: FuelConfig::default(),
cache_compiled: true,
cache_dir: None,
optimization_level: OptLevel::Speed,
callback_timeout: Duration::from_secs(30),
}
}
}
impl WasmChannelRuntimeConfig {
/// Create a minimal config for testing.
pub fn for_testing() -> Self {
Self {
default_limits: ResourceLimits {
memory_bytes: 5 * 1024 * 1024, // 5 MB
fuel: 1_000_000,
timeout: Duration::from_secs(5),
},
fuel_config: FuelConfig::with_limit(1_000_000),
cache_compiled: false,
cache_dir: None,
optimization_level: OptLevel::None, // Faster compilation for tests
callback_timeout: Duration::from_secs(5),
}
}
}
/// A compiled WASM channel component ready for instantiation.
#[derive(Debug)]
pub struct PreparedChannelModule {
/// Channel name.
pub name: String,
/// Channel description.
pub description: String,
/// Compiled component bytes (public for testing, otherwise use component_bytes()).
pub(crate) component_bytes: Vec<u8>,
/// Resource limits for this channel.
pub limits: ResourceLimits,
}
impl PreparedChannelModule {
/// Get the compiled component bytes.
pub fn component_bytes(&self) -> &[u8] {
&self.component_bytes
}
/// Create a PreparedChannelModule for testing purposes.
///
/// Creates a module with no actual WASM bytes, suitable for testing
/// channel infrastructure without requiring a real WASM component.
pub fn for_testing(name: impl Into<String>, description: impl Into<String>) -> Self {
Self {
name: name.into(),
description: description.into(),
component_bytes: Vec::new(),
limits: ResourceLimits::default(),
}
}
}
/// WASM channel runtime.
///
/// Manages the Wasmtime engine and a cache of prepared channel modules.
pub struct WasmChannelRuntime {
/// Wasmtime engine with configured settings.
engine: Engine,
/// Runtime configuration.
config: WasmChannelRuntimeConfig,
/// Cache of prepared modules by name.
modules: RwLock<HashMap<String, Arc<PreparedChannelModule>>>,
}
impl WasmChannelRuntime {
/// Create a new runtime with the given configuration.
pub fn new(config: WasmChannelRuntimeConfig) -> Result<Self, WasmChannelError> {
let mut wasmtime_config = Config::new();
// Enable fuel consumption for CPU limiting
if config.fuel_config.enabled {
wasmtime_config.consume_fuel(true);
}
// Enable epoch interruption as a backup timeout mechanism
wasmtime_config.epoch_interruption(true);
// Enable component model (WASI Preview 2)
wasmtime_config.wasm_component_model(true);
// Disable threads (simplifies security model)
wasmtime_config.wasm_threads(false);
// Set optimization level
wasmtime_config.cranelift_opt_level(config.optimization_level);
// Disable debug info in production
wasmtime_config.debug_info(false);
let engine = Engine::new(&wasmtime_config).map_err(|e| {
WasmChannelError::Config(format!("Failed to create Wasmtime engine: {}", e))
})?;
Ok(Self {
engine,
config,
modules: RwLock::new(HashMap::new()),
})
}
/// Get the Wasmtime engine.
pub fn engine(&self) -> &Engine {
&self.engine
}
/// Get the runtime configuration.
pub fn config(&self) -> &WasmChannelRuntimeConfig {
&self.config
}
/// Prepare a WASM channel component for execution.
///
/// This validates and compiles the component.
/// The compiled component is cached for fast instantiation.
pub async fn prepare(
&self,
name: &str,
wasm_bytes: &[u8],
limits: Option<ResourceLimits>,
description: Option<String>,
) -> Result<Arc<PreparedChannelModule>, WasmChannelError> {
// Check if already prepared
if let Some(module) = self.modules.read().await.get(name) {
return Ok(Arc::clone(module));
}
let name = name.to_string();
let wasm_bytes = wasm_bytes.to_vec();
let engine = self.engine.clone();
let default_limits = self.config.default_limits.clone();
let desc = description.unwrap_or_else(|| format!("WASM channel: {}", name));
// Compile in blocking task (Wasmtime compilation is synchronous)
let prepared = tokio::task::spawn_blocking(move || {
// Validate and compile the component
let _component = wasmtime::component::Component::new(&engine, &wasm_bytes)
.map_err(|e| WasmChannelError::Compilation(e.to_string()))?;
Ok::<_, WasmChannelError>(PreparedChannelModule {
name: name.clone(),
description: desc,
component_bytes: wasm_bytes,
limits: limits.unwrap_or(default_limits),
})
})
.await
.map_err(|e| {
WasmChannelError::Compilation(format!("Preparation task panicked: {}", e))
})??;
let prepared = Arc::new(prepared);
// Cache the prepared module
if self.config.cache_compiled {
self.modules
.write()
.await
.insert(prepared.name.clone(), Arc::clone(&prepared));
}
tracing::info!(
name = %prepared.name,
"Prepared WASM channel for execution"
);
Ok(prepared)
}
/// Get a prepared module by name.
pub async fn get(&self, name: &str) -> Option<Arc<PreparedChannelModule>> {
self.modules.read().await.get(name).cloned()
}
/// Remove a prepared module from the cache.
pub async fn remove(&self, name: &str) -> Option<Arc<PreparedChannelModule>> {
self.modules.write().await.remove(name)
}
/// List all prepared module names.
pub async fn list(&self) -> Vec<String> {
self.modules.read().await.keys().cloned().collect()
}
/// Clear all cached modules.
pub async fn clear(&self) {
self.modules.write().await.clear();
}
}
impl std::fmt::Debug for WasmChannelRuntime {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
f.debug_struct("WasmChannelRuntime")
.field("config", &self.config)
.field("modules", &"<RwLock<HashMap>>")
.finish()
}
}
#[cfg(test)]
mod tests {
use crate::channels::wasm::runtime::{WasmChannelRuntime, WasmChannelRuntimeConfig};
#[test]
fn test_runtime_config_default() {
let config = WasmChannelRuntimeConfig::default();
assert!(config.cache_compiled);
assert!(config.fuel_config.enabled);
// Channels get more memory than tools
assert_eq!(config.default_limits.memory_bytes, 50 * 1024 * 1024);
}
#[test]
fn test_runtime_config_for_testing() {
let config = WasmChannelRuntimeConfig::for_testing();
assert!(!config.cache_compiled);
assert_eq!(config.default_limits.memory_bytes, 5 * 1024 * 1024);
}
#[test]
fn test_runtime_creation() {
let config = WasmChannelRuntimeConfig::for_testing();
let runtime = WasmChannelRuntime::new(config).unwrap();
assert!(runtime.config().fuel_config.enabled);
}
#[tokio::test]
async fn test_module_cache_operations() {
let config = WasmChannelRuntimeConfig::for_testing();
let runtime = WasmChannelRuntime::new(config).unwrap();
// Initially empty
assert!(runtime.list().await.is_empty());
assert!(runtime.get("test").await.is_none());
}
}
+417
View File
@@ -0,0 +1,417 @@
//! JSON schema for WASM channel capabilities files.
//!
//! External WASM channels declare their required capabilities via a sidecar JSON file
//! (e.g., `slack.capabilities.json`). This module defines the schema for those files
//! and provides conversion to runtime [`ChannelCapabilities`].
//!
//! # Example Capabilities File
//!
//! ```json
//! {
//! "type": "channel",
//! "name": "slack",
//! "description": "Slack Events API channel",
//! "capabilities": {
//! "http": {
//! "allowlist": [
//! { "host": "slack.com", "path_prefix": "/api/" }
//! ],
//! "credentials": {
//! "slack_bot": {
//! "secret_name": "slack_bot_token",
//! "location": { "type": "bearer" },
//! "host_patterns": ["slack.com"]
//! }
//! }
//! },
//! "secrets": { "allowed_names": ["slack_*"] },
//! "channel": {
//! "allowed_paths": ["/webhook/slack"],
//! "allow_polling": false,
//! "workspace_prefix": "channels/slack/",
//! "emit_rate_limit": { "messages_per_minute": 100 }
//! }
//! },
//! "config": {
//! "signing_secret_name": "slack_signing_secret"
//! }
//! }
//! ```
use std::collections::HashMap;
use std::time::Duration;
use serde::{Deserialize, Serialize};
use crate::channels::wasm::capabilities::{
ChannelCapabilities, EmitRateLimitConfig, MIN_POLL_INTERVAL_MS,
};
use crate::tools::wasm::{CapabilitiesFile as ToolCapabilitiesFile, RateLimitSchema};
/// Root schema for a channel capabilities JSON file.
#[derive(Debug, Clone, Default, Serialize, Deserialize)]
pub struct ChannelCapabilitiesFile {
/// File type, must be "channel".
#[serde(default = "default_type")]
pub r#type: String,
/// Channel name.
pub name: String,
/// Channel description.
#[serde(default)]
pub description: Option<String>,
/// Capabilities (tool + channel specific).
#[serde(default)]
pub capabilities: ChannelCapabilitiesSchema,
/// Channel-specific configuration passed to on_start.
#[serde(default)]
pub config: HashMap<String, serde_json::Value>,
}
fn default_type() -> String {
"channel".to_string()
}
impl ChannelCapabilitiesFile {
/// Parse from JSON string.
pub fn from_json(json: &str) -> Result<Self, serde_json::Error> {
serde_json::from_str(json)
}
/// Parse from JSON bytes.
pub fn from_bytes(bytes: &[u8]) -> Result<Self, serde_json::Error> {
serde_json::from_slice(bytes)
}
/// Convert to runtime ChannelCapabilities.
pub fn to_capabilities(&self) -> ChannelCapabilities {
self.capabilities.to_channel_capabilities(&self.name)
}
/// Get the channel config as JSON string.
pub fn config_json(&self) -> String {
serde_json::to_string(&self.config).unwrap_or_else(|_| "{}".to_string())
}
}
/// Schema for channel capabilities.
#[derive(Debug, Clone, Default, Serialize, Deserialize)]
pub struct ChannelCapabilitiesSchema {
/// Tool capabilities (HTTP, secrets, workspace_read).
#[serde(flatten)]
pub tool: Option<ToolCapabilitiesFile>,
/// Channel-specific capabilities.
#[serde(default)]
pub channel: Option<ChannelSpecificCapabilitiesSchema>,
}
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 mut caps =
ChannelCapabilities::for_channel(channel_name).with_tool_capabilities(tool_caps);
if let Some(channel) = &self.channel {
caps.allowed_paths = channel.allowed_paths.clone();
caps.allow_polling = channel.allow_polling;
caps.min_poll_interval_ms = channel
.min_poll_interval_ms
.unwrap_or(MIN_POLL_INTERVAL_MS)
.max(MIN_POLL_INTERVAL_MS);
if let Some(prefix) = &channel.workspace_prefix {
caps.workspace_prefix = prefix.clone();
}
if let Some(rate) = &channel.emit_rate_limit {
caps.emit_rate_limit = rate.to_emit_rate_limit();
}
if let Some(max_size) = channel.max_message_size {
caps.max_message_size = max_size;
}
if let Some(timeout_secs) = channel.callback_timeout_secs {
caps.callback_timeout = Duration::from_secs(timeout_secs);
}
}
caps
}
}
/// Channel-specific capabilities schema.
#[derive(Debug, Clone, Default, Serialize, Deserialize)]
pub struct ChannelSpecificCapabilitiesSchema {
/// HTTP paths the channel can register for webhooks.
#[serde(default)]
pub allowed_paths: Vec<String>,
/// Whether polling is allowed.
#[serde(default)]
pub allow_polling: bool,
/// Minimum poll interval in milliseconds.
#[serde(default)]
pub min_poll_interval_ms: Option<u32>,
/// Workspace prefix for storage (overrides default).
#[serde(default)]
pub workspace_prefix: Option<String>,
/// Rate limiting for emit_message.
#[serde(default)]
pub emit_rate_limit: Option<EmitRateLimitSchema>,
/// Maximum message content size in bytes.
#[serde(default)]
pub max_message_size: Option<usize>,
/// Callback timeout in seconds.
#[serde(default)]
pub callback_timeout_secs: Option<u64>,
}
/// Schema for emit rate limiting.
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct EmitRateLimitSchema {
/// Maximum messages per minute.
#[serde(default = "default_messages_per_minute")]
pub messages_per_minute: u32,
/// Maximum messages per hour.
#[serde(default = "default_messages_per_hour")]
pub messages_per_hour: u32,
}
fn default_messages_per_minute() -> u32 {
100
}
fn default_messages_per_hour() -> u32 {
5000
}
impl EmitRateLimitSchema {
fn to_emit_rate_limit(&self) -> EmitRateLimitConfig {
EmitRateLimitConfig {
messages_per_minute: self.messages_per_minute,
messages_per_hour: self.messages_per_hour,
}
}
}
impl From<RateLimitSchema> for EmitRateLimitSchema {
fn from(schema: RateLimitSchema) -> Self {
Self {
messages_per_minute: schema.requests_per_minute,
messages_per_hour: schema.requests_per_hour,
}
}
}
/// Channel configuration returned by on_start.
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct ChannelConfig {
/// Display name for the channel.
pub display_name: String,
/// HTTP endpoints to register.
#[serde(default)]
pub http_endpoints: Vec<HttpEndpointConfigSchema>,
/// Polling configuration.
#[serde(default)]
pub poll: Option<PollConfigSchema>,
}
impl Default for ChannelConfig {
fn default() -> Self {
Self {
display_name: "WASM Channel".to_string(),
http_endpoints: Vec::new(),
poll: None,
}
}
}
/// HTTP endpoint configuration schema.
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct HttpEndpointConfigSchema {
/// Path to register.
pub path: String,
/// HTTP methods to accept.
#[serde(default)]
pub methods: Vec<String>,
/// Whether secret validation is required.
#[serde(default)]
pub require_secret: bool,
}
/// Polling configuration schema.
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct PollConfigSchema {
/// Polling interval in milliseconds.
pub interval_ms: u32,
/// Whether polling is enabled.
#[serde(default)]
pub enabled: bool,
}
#[cfg(test)]
mod tests {
use crate::channels::wasm::schema::ChannelCapabilitiesFile;
#[test]
fn test_parse_minimal() {
let json = r#"{
"name": "test"
}"#;
let file = ChannelCapabilitiesFile::from_json(json).unwrap();
assert_eq!(file.name, "test");
assert_eq!(file.r#type, "channel");
}
#[test]
fn test_parse_full_slack_example() {
let json = r#"{
"type": "channel",
"name": "slack",
"description": "Slack Events API channel",
"capabilities": {
"http": {
"allowlist": [
{ "host": "slack.com", "path_prefix": "/api/" }
],
"credentials": {
"slack_bot": {
"secret_name": "slack_bot_token",
"location": { "type": "bearer" },
"host_patterns": ["slack.com"]
}
},
"rate_limit": { "requests_per_minute": 50, "requests_per_hour": 1000 }
},
"secrets": { "allowed_names": ["slack_*"] },
"channel": {
"allowed_paths": ["/webhook/slack"],
"allow_polling": false,
"emit_rate_limit": { "messages_per_minute": 100, "messages_per_hour": 5000 }
}
},
"config": {
"signing_secret_name": "slack_signing_secret"
}
}"#;
let file = ChannelCapabilitiesFile::from_json(json).unwrap();
assert_eq!(file.name, "slack");
assert_eq!(
file.description,
Some("Slack Events API channel".to_string())
);
let caps = file.to_capabilities();
assert!(caps.is_path_allowed("/webhook/slack"));
assert!(!caps.allow_polling);
assert_eq!(caps.workspace_prefix, "channels/slack/");
// Check tool capabilities were parsed
assert!(caps.tool_capabilities.http.is_some());
assert!(caps.tool_capabilities.secrets.is_some());
// Check config
let config_json = file.config_json();
assert!(config_json.contains("signing_secret_name"));
}
#[test]
fn test_parse_with_polling() {
let json = r#"{
"name": "telegram",
"capabilities": {
"channel": {
"allowed_paths": [],
"allow_polling": true,
"min_poll_interval_ms": 60000
}
}
}"#;
let file = ChannelCapabilitiesFile::from_json(json).unwrap();
let caps = file.to_capabilities();
assert!(caps.allow_polling);
assert_eq!(caps.min_poll_interval_ms, 60000);
}
#[test]
fn test_min_poll_interval_enforced() {
let json = r#"{
"name": "test",
"capabilities": {
"channel": {
"allow_polling": true,
"min_poll_interval_ms": 1000
}
}
}"#;
let file = ChannelCapabilitiesFile::from_json(json).unwrap();
let caps = file.to_capabilities();
// Should be clamped to minimum
assert_eq!(caps.min_poll_interval_ms, 30000);
}
#[test]
fn test_workspace_prefix_override() {
let json = r#"{
"name": "custom",
"capabilities": {
"channel": {
"workspace_prefix": "integrations/custom/"
}
}
}"#;
let file = ChannelCapabilitiesFile::from_json(json).unwrap();
let caps = file.to_capabilities();
assert_eq!(caps.workspace_prefix, "integrations/custom/");
}
#[test]
fn test_emit_rate_limit() {
let json = r#"{
"name": "test",
"capabilities": {
"channel": {
"emit_rate_limit": {
"messages_per_minute": 50,
"messages_per_hour": 1000
}
}
}
}"#;
let file = ChannelCapabilitiesFile::from_json(json).unwrap();
let caps = file.to_capabilities();
assert_eq!(caps.emit_rate_limit.messages_per_minute, 50);
assert_eq!(caps.emit_rate_limit.messages_per_hour, 1000);
}
}
File diff suppressed because it is too large Load Diff
+23
View File
@@ -180,6 +180,10 @@ fn default_session_path() -> PathBuf {
pub struct ChannelsConfig { pub struct ChannelsConfig {
pub cli: CliConfig, pub cli: CliConfig,
pub http: Option<HttpConfig>, pub http: Option<HttpConfig>,
/// 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,
} }
#[derive(Debug, Clone)] #[derive(Debug, Clone)]
@@ -222,10 +226,29 @@ impl ChannelsConfig {
enabled: cli_enabled, enabled: cli_enabled,
}, },
http, http,
wasm_channels_dir: optional_env("WASM_CHANNELS_DIR")?
.map(PathBuf::from)
.unwrap_or_else(default_channels_dir),
wasm_channels_enabled: optional_env("WASM_CHANNELS_ENABLED")?
.map(|s| s.parse())
.transpose()
.map_err(|e| ConfigError::InvalidValue {
key: "WASM_CHANNELS_ENABLED".to_string(),
message: format!("must be 'true' or 'false': {e}"),
})?
.unwrap_or(true),
}) })
} }
} }
/// Get the default channels directory (~/.near-agent/channels/).
fn default_channels_dir() -> PathBuf {
dirs::home_dir()
.unwrap_or_else(|| PathBuf::from("."))
.join(".near-agent")
.join("channels")
}
/// Agent behavior configuration. /// Agent behavior configuration.
#[derive(Debug, Clone)] #[derive(Debug, Clone)]
pub struct AgentConfig { pub struct AgentConfig {
+39 -1
View File
@@ -7,7 +7,10 @@ use tracing_subscriber::{EnvFilter, layer::SubscriberExt, util::SubscriberInitEx
use near_agent::{ use near_agent::{
agent::{Agent, AgentDeps}, agent::{Agent, AgentDeps},
channels::{AppEvent, ChannelManager, HttpChannel, ReplChannel, TuiChannel}, channels::{
AppEvent, ChannelManager, HttpChannel, ReplChannel, TuiChannel,
wasm::{WasmChannelLoader, WasmChannelRuntime, WasmChannelRuntimeConfig},
},
cli::{Cli, Command, run_tool_command}, cli::{Cli, Command, run_tool_command},
config::Config, config::Config,
history::Store, history::Store,
@@ -286,6 +289,41 @@ async fn main() -> anyhow::Result<()> {
} }
} }
// Load WASM channels if enabled
if config.channels.wasm_channels_enabled && config.channels.wasm_channels_dir.exists() {
match WasmChannelRuntime::new(WasmChannelRuntimeConfig::default()) {
Ok(runtime) => {
let runtime = Arc::new(runtime);
let loader = WasmChannelLoader::new(Arc::clone(&runtime));
match loader
.load_from_dir(&config.channels.wasm_channels_dir)
.await
{
Ok(results) => {
for channel in results.loaded {
tracing::info!("Loaded WASM channel: {}", channel.channel_name());
channels.add(Box::new(channel));
}
for (path, err) in &results.errors {
tracing::warn!(
"Failed to load WASM channel {}: {}",
path.display(),
err
);
}
}
Err(e) => {
tracing::warn!("Failed to scan WASM channels directory: {}", e);
}
}
}
Err(e) => {
tracing::warn!("Failed to initialize WASM channel runtime: {}", e);
}
}
}
// Create workspace for agent (shared with memory tools) // Create workspace for agent (shared with memory tools)
let workspace = store.as_ref().map(|s| { let workspace = store.as_ref().map(|s| {
let mut ws = Workspace::new("default", s.pool()); let mut ws = Workspace::new("default", s.pool());
+1 -1
View File
@@ -118,4 +118,4 @@ pub use storage::{
pub use loader::{DiscoveredTool, LoadResults, WasmLoadError, WasmToolLoader, discover_tools}; pub use loader::{DiscoveredTool, LoadResults, WasmLoadError, WasmToolLoader, discover_tools};
// Capabilities schema (for parsing *.capabilities.json files) // Capabilities schema (for parsing *.capabilities.json files)
pub use capabilities_schema::CapabilitiesFile; pub use capabilities_schema::{CapabilitiesFile, RateLimitSchema};
+446
View File
@@ -0,0 +1,446 @@
//! Integration tests for the WASM channel system.
//!
//! These tests verify the full flow of WASM channel operations:
//! - Channel loading from filesystem
//! - HTTP webhook routing
//! - Message emission and delivery
//! - Response handling
use std::collections::HashMap;
use std::sync::Arc;
use near_agent::channels::Channel;
use near_agent::channels::wasm::{
ChannelCapabilities, EmitRateLimitConfig, PreparedChannelModule, RegisteredEndpoint,
WasmChannel, WasmChannelRouter, WasmChannelRuntime, WasmChannelRuntimeConfig,
};
use tempfile::TempDir;
/// Create a test runtime for WASM channel operations.
fn create_test_runtime() -> Arc<WasmChannelRuntime> {
let config = WasmChannelRuntimeConfig::for_testing();
Arc::new(WasmChannelRuntime::new(config).expect("Failed to create runtime"))
}
/// Create a test channel with minimal configuration.
fn create_test_channel(
runtime: Arc<WasmChannelRuntime>,
name: &str,
paths: Vec<&str>,
) -> WasmChannel {
let prepared = Arc::new(PreparedChannelModule::for_testing(
name,
format!("Test channel: {}", name),
));
let mut capabilities = ChannelCapabilities::for_channel(name);
for path in paths {
capabilities = capabilities.with_path(path.to_string());
}
WasmChannel::new(runtime, prepared, capabilities, "{}".to_string())
}
mod router_tests {
use super::*;
#[tokio::test]
async fn test_register_and_route_channel() {
let router = WasmChannelRouter::new();
let runtime = create_test_runtime();
let channel = Arc::new(create_test_channel(
runtime,
"test-channel",
vec!["/webhook/test"],
));
let endpoints = vec![RegisteredEndpoint {
channel_name: "test-channel".to_string(),
path: "/webhook/test".to_string(),
methods: vec!["POST".to_string()],
require_secret: false,
}];
router.register(channel.clone(), endpoints, None).await;
// Verify channel is found by path
let found = router.get_channel_for_path("/webhook/test").await;
assert!(found.is_some());
assert_eq!(found.unwrap().channel_name(), "test-channel");
// Verify non-existent path returns None
let not_found = router.get_channel_for_path("/webhook/nonexistent").await;
assert!(not_found.is_none());
}
#[tokio::test]
async fn test_secret_validation() {
let router = WasmChannelRouter::new();
let runtime = create_test_runtime();
let channel = Arc::new(create_test_channel(
runtime,
"secure-channel",
vec!["/webhook/secure"],
));
router
.register(channel, vec![], Some("my-secret-123".to_string()))
.await;
// Correct secret validates
assert!(
router
.validate_secret("secure-channel", "my-secret-123")
.await
);
// Wrong secret fails
assert!(
!router
.validate_secret("secure-channel", "wrong-secret")
.await
);
// Non-existent channel without secret always validates
assert!(router.validate_secret("nonexistent", "anything").await);
}
#[tokio::test]
async fn test_unregister_channel() {
let router = WasmChannelRouter::new();
let runtime = create_test_runtime();
let channel = Arc::new(create_test_channel(
runtime,
"temp-channel",
vec!["/webhook/temp"],
));
let endpoints = vec![RegisteredEndpoint {
channel_name: "temp-channel".to_string(),
path: "/webhook/temp".to_string(),
methods: vec!["POST".to_string()],
require_secret: false,
}];
router.register(channel, endpoints, None).await;
// Channel exists
assert!(router.get_channel_for_path("/webhook/temp").await.is_some());
// Unregister
router.unregister("temp-channel").await;
// Channel no longer exists
assert!(router.get_channel_for_path("/webhook/temp").await.is_none());
}
#[tokio::test]
async fn test_multiple_channels() {
let router = WasmChannelRouter::new();
let runtime = create_test_runtime();
// Register multiple channels
for name in &["slack", "telegram", "discord"] {
let channel = Arc::new(create_test_channel(
Arc::clone(&runtime),
name,
vec![&format!("/webhook/{}", name)],
));
let endpoints = vec![RegisteredEndpoint {
channel_name: name.to_string(),
path: format!("/webhook/{}", name),
methods: vec!["POST".to_string()],
require_secret: false,
}];
router.register(channel, endpoints, None).await;
}
// Verify all channels are registered
let channels = router.list_channels().await;
assert_eq!(channels.len(), 3);
assert!(channels.contains(&"slack".to_string()));
assert!(channels.contains(&"telegram".to_string()));
assert!(channels.contains(&"discord".to_string()));
// Verify all paths work
for name in &["slack", "telegram", "discord"] {
let found = router
.get_channel_for_path(&format!("/webhook/{}", name))
.await;
assert!(found.is_some());
assert_eq!(found.unwrap().channel_name(), *name);
}
}
}
mod channel_lifecycle_tests {
use super::*;
#[tokio::test]
async fn test_channel_start_and_shutdown() {
let runtime = create_test_runtime();
let channel = create_test_channel(runtime, "lifecycle-test", vec!["/webhook/lifecycle"]);
// Start channel
let stream = channel.start().await;
assert!(stream.is_ok());
// Health check should pass
assert!(channel.health_check().await.is_ok());
// Shutdown
assert!(channel.shutdown().await.is_ok());
// Health check should fail after shutdown
assert!(channel.health_check().await.is_err());
}
#[tokio::test]
async fn test_channel_http_callback() {
let runtime = create_test_runtime();
let channel = create_test_channel(runtime, "http-test", vec!["/webhook/http"]);
// Start channel
let _stream = channel.start().await.expect("Failed to start channel");
// Call HTTP callback (stub implementation returns 200 OK)
let response = channel
.call_on_http_request(
"POST",
"/webhook/http",
&HashMap::new(),
&HashMap::new(),
b"{}",
true,
)
.await
.expect("HTTP callback failed");
assert_eq!(response.status, 200);
// Cleanup
channel.shutdown().await.expect("Shutdown failed");
}
}
mod loader_tests {
use super::*;
use std::io::Write;
#[tokio::test]
async fn test_discover_channels_empty_dir() {
let dir = TempDir::new().expect("Failed to create temp dir");
let channels = near_agent::channels::wasm::discover_channels(dir.path())
.await
.expect("Discovery failed");
assert!(channels.is_empty());
}
#[tokio::test]
async fn test_discover_channels_with_wasm_files() {
let dir = TempDir::new().expect("Failed to create temp dir");
// Create fake WASM files
std::fs::File::create(dir.path().join("slack.wasm")).expect("Failed to create file");
std::fs::File::create(dir.path().join("telegram.wasm")).expect("Failed to create file");
let channels = near_agent::channels::wasm::discover_channels(dir.path())
.await
.expect("Discovery failed");
assert_eq!(channels.len(), 2);
assert!(channels.contains_key("slack"));
assert!(channels.contains_key("telegram"));
}
#[tokio::test]
async fn test_discover_channels_with_capabilities() {
let dir = TempDir::new().expect("Failed to create temp dir");
// Create WASM and capabilities file
std::fs::File::create(dir.path().join("custom.wasm")).expect("Failed to create wasm");
let mut cap_file = std::fs::File::create(dir.path().join("custom.capabilities.json"))
.expect("Failed to create capabilities");
cap_file
.write_all(
br#"{
"name": "custom",
"capabilities": {
"channel": {
"allowed_paths": ["/webhook/custom"]
}
}
}"#,
)
.expect("Failed to write capabilities");
let channels = near_agent::channels::wasm::discover_channels(dir.path())
.await
.expect("Discovery failed");
assert_eq!(channels.len(), 1);
assert!(channels["custom"].capabilities_path.is_some());
}
#[tokio::test]
async fn test_discover_channels_ignores_non_wasm() {
let dir = TempDir::new().expect("Failed to create temp dir");
// Create various non-WASM files
std::fs::File::create(dir.path().join("readme.md")).expect("Failed to create file");
std::fs::File::create(dir.path().join("config.json")).expect("Failed to create file");
std::fs::File::create(dir.path().join("channel.wasm")).expect("Failed to create file");
let channels = near_agent::channels::wasm::discover_channels(dir.path())
.await
.expect("Discovery failed");
// Only the .wasm file should be discovered
assert_eq!(channels.len(), 1);
assert!(channels.contains_key("channel"));
}
}
mod capabilities_tests {
use super::*;
#[test]
fn test_capabilities_path_validation() {
let caps = ChannelCapabilities::for_channel("test")
.with_path("/webhook/test")
.with_path("/api/events");
assert!(caps.is_path_allowed("/webhook/test"));
assert!(caps.is_path_allowed("/api/events"));
assert!(!caps.is_path_allowed("/other/path"));
}
#[test]
fn test_capabilities_workspace_prefix() {
let caps = ChannelCapabilities::for_channel("slack");
assert_eq!(caps.workspace_prefix, "channels/slack/");
// Validate path prefixing
let prefixed = caps.prefix_workspace_path("state.json");
assert_eq!(prefixed, "channels/slack/state.json");
}
#[test]
fn test_capabilities_workspace_path_validation() {
let caps = ChannelCapabilities::for_channel("test");
// Valid paths
assert!(caps.validate_workspace_path("state.json").is_ok());
assert!(caps.validate_workspace_path("data/file.txt").is_ok());
// Invalid paths (traversal attempts)
assert!(caps.validate_workspace_path("../escape.txt").is_err());
assert!(caps.validate_workspace_path("/absolute/path").is_err());
assert!(caps.validate_workspace_path("data/../escape").is_err());
}
#[test]
fn test_capabilities_poll_interval_validation() {
let caps = ChannelCapabilities::for_channel("test").with_polling(30_000);
// Valid interval (returns as-is)
let result = caps.validate_poll_interval(60_000);
assert!(result.is_ok());
assert_eq!(result.unwrap(), 60_000);
// Too short interval is clamped to minimum (not rejected)
let result = caps.validate_poll_interval(1_000);
assert!(result.is_ok());
assert_eq!(result.unwrap(), 30_000);
// Minimum interval passes as-is
let result = caps.validate_poll_interval(30_000);
assert!(result.is_ok());
assert_eq!(result.unwrap(), 30_000);
// Polling disabled returns error
let no_poll_caps = ChannelCapabilities::for_channel("no-poll");
assert!(no_poll_caps.validate_poll_interval(60_000).is_err());
}
#[test]
fn test_emit_rate_limit_config() {
let config = EmitRateLimitConfig {
messages_per_minute: 100,
messages_per_hour: 5000,
};
assert_eq!(config.messages_per_minute, 100);
assert_eq!(config.messages_per_hour, 5000);
}
}
mod message_emission_tests {
use super::*;
use near_agent::channels::wasm::{ChannelHostState, EmittedMessage};
#[test]
fn test_emit_message_basic() {
let caps = ChannelCapabilities::for_channel("test");
let mut state = ChannelHostState::new("test", caps);
let msg = EmittedMessage::new("user123", "Hello, world!");
state.emit_message(msg).expect("Emit should succeed");
assert_eq!(state.emitted_count(), 1);
let messages = state.take_emitted_messages();
assert_eq!(messages.len(), 1);
assert_eq!(messages[0].user_id, "user123");
assert_eq!(messages[0].content, "Hello, world!");
// Queue should be cleared
assert_eq!(state.emitted_count(), 0);
}
#[test]
fn test_emit_message_with_metadata() {
let caps = ChannelCapabilities::for_channel("test");
let mut state = ChannelHostState::new("test", caps);
let msg = EmittedMessage::new("user123", "Hello")
.with_user_name("John Doe")
.with_thread_id("thread-1")
.with_metadata(r#"{"channel": "C123"}"#);
state.emit_message(msg).expect("Emit should succeed");
let messages = state.take_emitted_messages();
assert_eq!(messages[0].user_name, Some("John Doe".to_string()));
assert_eq!(messages[0].thread_id, Some("thread-1".to_string()));
assert!(messages[0].metadata_json.contains("channel"));
}
#[test]
fn test_emit_rate_limiting() {
let caps = ChannelCapabilities::for_channel("test");
let mut state = ChannelHostState::new("test", caps);
// Emit up to the per-execution limit
for i in 0..100 {
let msg = EmittedMessage::new("user", format!("Message {}", i));
state.emit_message(msg).expect("Emit should succeed");
}
// Messages beyond the limit are silently dropped
let msg = EmittedMessage::new("user", "Should be dropped");
state.emit_message(msg).expect("Emit should not fail");
assert_eq!(state.emitted_count(), 100);
assert_eq!(state.emits_dropped(), 1);
}
}
+276
View File
@@ -0,0 +1,276 @@
// WASM Channel Sandbox Interface
//
// Defines the contract between sandboxed channels and the host runtime.
// Channels export the `channel` interface; the host provides the `channel-host` interface.
//
// Architecture: Host-Managed Event Loop
// ┌─────────────────────────────────────────────────────────────────────────────────┐
// │ Host-Managed Event Loop │
// │ │
// │ ┌─────────────┐ ┌──────────────┐ ┌──────────────┐ │
// │ │ HTTP │ │ Polling │ │ Timer │ │
// │ │ Router │ │ Scheduler │ │ Scheduler │ │
// │ └──────┬──────┘ └──────┬───────┘ └──────┬───────┘ │
// │ └───────────────────┴────────────────────┘ │
// │ │ │
// │ ▼ │
// │ ┌──────────────────┬──────────────────┐ │
// │ ▼ ▼ ▼ │
// │ ┌─────────────┐ ┌─────────────┐ ┌─────────────┐ │
// │ │ on-http-req │ │ on-poll │ │ on-respond │ WASM Exports │
// │ └─────────────┘ └─────────────┘ └─────────────┘ │
// │ │ │ │ │
// │ └──────────────────┴──────────────────┘ │
// │ │ │
// │ ▼ │
// │ ┌─────────────────┐ │
// │ │ Host Imports │ │
// │ │ emit-message │──────────▶ MessageStream │
// │ │ http-request │ │
// │ └─────────────────┘ │
// └─────────────────────────────────────────────────────────────────────────────────┘
//
// Security Model:
// - WASM channels are untrusted and run in a sandbox
// - Fresh instance per callback (no shared mutable state)
// - All capabilities are opt-in (default: no access)
// - Secrets are NEVER exposed to WASM; credentials are injected at host boundary
// - Workspace writes are prefixed with channels/<name>/ to prevent escape
// - Message emission is rate-limited
package near:agent;
/// Host-provided capabilities for sandboxed channels.
///
/// Extends base tool capabilities with channel-specific functions:
/// - emit-message: Queue messages for delivery to the agent
/// - workspace-write: Write to channel-namespaced workspace
interface channel-host {
// ==================== Base Capabilities (from tool host) ====================
/// Log levels for structured logging.
enum log-level {
trace,
debug,
info,
warn,
error,
}
/// Emit a log message.
///
/// Messages are collected and emitted after execution completes.
/// Rate-limited to 1000 entries per execution, 4KB per message.
log: func(level: log-level, message: string);
/// Get the current timestamp in milliseconds since Unix epoch.
now-millis: func() -> u64;
/// Read a file from the workspace.
///
/// Path is automatically prefixed with channels/<name>/.
/// Path must be relative (no leading /) and cannot contain "..".
/// Returns None if the file doesn't exist.
workspace-read: func(path: string) -> option<string>;
/// Response from an HTTP request.
record http-response {
/// HTTP status code.
status: u16,
/// Response headers as JSON object string.
headers-json: string,
/// Response body bytes.
body: list<u8>,
}
/// Make an HTTP request (if capability granted).
///
/// Security:
/// - Only allowed endpoints (host/path patterns) can be accessed
/// - Credentials are injected by the host; WASM never sees them
/// - Response is scanned for leaked secrets before returning
/// - Rate-limited per channel
http-request: func(
method: string,
url: string,
headers-json: string,
body: option<list<u8>>
) -> result<http-response, string>;
/// Check if a secret exists (if capability granted).
///
/// Security:
/// - WASM can only check existence, NEVER read values
/// - Only allowed secret names can be checked
/// - Actual credentials are injected by host during HTTP requests
secret-exists: func(name: string) -> bool;
// ==================== Channel-Specific Capabilities ====================
/// A message to emit to the agent.
record emitted-message {
/// User identifier within the channel (e.g., Slack user ID).
user-id: string,
/// Optional human-readable user name.
user-name: option<string>,
/// Message content.
content: string,
/// Optional thread ID for threaded conversations.
thread-id: option<string>,
/// Channel-specific metadata as JSON string.
metadata-json: string,
}
/// Emit a message to the agent.
///
/// Messages are queued during callback execution and delivered after
/// the callback completes successfully.
///
/// Security:
/// - Rate-limited per execution (max 100 messages)
/// - Rate-limited globally per channel (configurable)
/// - Content size limited to 64KB
emit-message: func(msg: emitted-message);
/// Write a file to the workspace.
///
/// Path is automatically prefixed with channels/<name>/.
/// Path must be relative (no leading /) and cannot contain "..".
///
/// Returns Err if:
/// - Path validation fails (traversal attempt, absolute path)
/// - Write operation fails
workspace-write: func(path: string, content: string) -> result<_, string>;
}
/// Channel interface that sandboxed channels must implement.
interface channel {
// ==================== Configuration Types ====================
/// Configuration for an HTTP endpoint.
record http-endpoint-config {
/// Path to register (e.g., "/webhook/slack").
path: string,
/// Allowed HTTP methods (e.g., ["POST"]).
methods: list<string>,
/// Whether the endpoint requires secret validation.
require-secret: bool,
}
/// Configuration for polling behavior.
record poll-config {
/// Polling interval in milliseconds (minimum 30000).
interval-ms: u32,
/// Whether polling is enabled.
enabled: bool,
}
/// Channel configuration returned by on-start.
record channel-config {
/// Human-readable display name.
display-name: string,
/// HTTP endpoints to register.
http-endpoints: list<http-endpoint-config>,
/// Optional polling configuration.
poll: option<poll-config>,
}
// ==================== Request/Response Types ====================
/// Incoming HTTP request from a webhook.
record incoming-http-request {
/// HTTP method (GET, POST, etc.).
method: string,
/// Request path.
path: string,
/// Request headers as JSON object string.
headers-json: string,
/// Query parameters as JSON object string.
query-json: string,
/// Request body bytes.
body: list<u8>,
/// Whether the webhook secret was validated by the host.
secret-validated: bool,
}
/// HTTP response to return to the webhook caller.
record outgoing-http-response {
/// HTTP status code.
status: u16,
/// Response headers as JSON object string.
headers-json: string,
/// Response body bytes.
body: list<u8>,
}
/// Agent response to be sent back to the channel.
record agent-response {
/// Unique message ID for correlation.
message-id: string,
/// Response content from the agent.
content: string,
/// Optional thread ID for threaded replies.
thread-id: option<string>,
/// Channel-specific metadata as JSON string.
metadata-json: string,
}
// ==================== Lifecycle Callbacks ====================
/// Initialize the channel.
///
/// Called once when the channel is loaded. Returns configuration
/// describing HTTP endpoints and polling behavior.
///
/// Arguments:
/// - config-json: Channel configuration from the capabilities file.
///
/// Returns:
/// - Ok(channel-config): Configuration for the host to set up routing
/// - Err(string): Initialization failure message
on-start: func(config-json: string) -> result<channel-config, string>;
/// Handle an incoming HTTP request.
///
/// Called for each HTTP request to a registered endpoint.
/// Use emit-message to queue messages for the agent.
///
/// Arguments:
/// - req: The incoming HTTP request
///
/// Returns:
/// - HTTP response to send back to the caller
on-http-request: func(req: incoming-http-request) -> outgoing-http-response;
/// Handle a polling tick.
///
/// Called periodically if polling is configured.
/// Use emit-message to queue messages discovered during polling.
on-poll: func();
/// Deliver an agent response to the channel.
///
/// Called when the agent has generated a response to a message
/// that was emitted by this channel.
///
/// Arguments:
/// - response: The agent's response
///
/// Returns:
/// - Ok: Response delivered successfully
/// - Err(string): Delivery failure message
on-respond: func(response: agent-response) -> result<_, string>;
/// Clean up channel resources.
///
/// Called when the channel is being unloaded.
on-shutdown: func();
}
/// World definition for sandboxed channels.
///
/// Channels import host capabilities and export the channel interface.
world sandboxed-channel {
import channel-host;
export channel;
}