gitops/optimclaw
1
0
Fork 0
mirror of https://github.com/outbackdingo/optimclaw.git synced 2026-08-31 08:39:24 +00:00
Code Issues Packages Projects Releases Wiki Activity

DM pairing + Telegram channel improvements (#17)

Browse Source 
* feat: Implement DM pairing for channels

- Introduced a new pairing system to manage direct messages from unknown senders.
- Added `PairingStore` to handle pending requests and allowlist management.
- Implemented CLI commands for listing and approving pairing requests.
- Updated Telegram channel to utilize the new pairing logic, including workspace paths for storing pairing data.
- Enhanced WASM channel integration to support pairing functionality.

This feature enhances security by requiring approval for unknown senders before they can interact with the agent.

* Enhance Telegram channel support with media captioning and DM pairing features

- Added support for media captions in Telegram messages, allowing for richer content handling.
- Updated message processing to utilize either text or caption, improving message flexibility.
- Enhanced DM pairing functionality to include approval and listing capabilities for direct messages.
- Updated feature parity documentation to reflect new capabilities and improvements in Telegram integration.

* Update README and BUILDING_CHANNELS documentation for Telegram channel integration

- Enhanced README with instructions for building and running the Telegram channel, including a note on running `./scripts/build-all.sh` for full releases.
- Added detailed steps in BUILDING_CHANNELS.md for building and deploying the Telegram channel, emphasizing the need to run `./channels-src/telegram/build.sh` before building the main crate to ensure updated WASM is included.
- Updated CLI module to expose a new command for pairing with store functionality.

* Implement build script for Telegram channel WASM and enhance pairing error handling

- Added a new `build.rs` script to automate the compilation of the Telegram channel's WASM binary from source, ensuring reproducible builds and emphasizing supply chain security by preventing committed binaries.
- Updated `BUILDING_CHANNELS.md` to reflect the new build process and the importance of not committing compiled binaries.
- Enhanced error handling in the pairing approval process to include rate limiting for failed attempts, improving security and user feedback.

* Remove Telegram channel WASM binary file as part of the build process cleanup, ensuring no committed binaries are present in the repository.
This commit is contained in:
Ilgın Kanat
2026-02-12 00:46:47 +00:00
committed by GitHub
parent bb228f6315
commit 115b7f38fe
22 changed files with 1774 additions and 129 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/channels/wasm/loader.rs
+17 -5
View File
@@ -16,16 +16,21 @@ use crate::channels::wasm::error::WasmChannelError;
use crate::channels::wasm::runtime::WasmChannelRuntime;
use crate::channels::wasm::schema::ChannelCapabilitiesFile;
use crate::channels::wasm::wrapper::WasmChannel;
use crate::pairing::PairingStore;
/// Loads WASM channels from the filesystem.
pub struct WasmChannelLoader {
runtime: Arc<WasmChannelRuntime>,
pairing_store: Arc<PairingStore>,
}
impl WasmChannelLoader {
/// Create a new loader with the given runtime.
pub fn new(runtime: Arc<WasmChannelRuntime>) -> Self {
Self { runtime }
/// Create a new loader with the given runtime and pairing store.
pub fn new(runtime: Arc<WasmChannelRuntime>, pairing_store: Arc<PairingStore>) -> Self {
Self {
runtime,
pairing_store,
}
}
/// Load a single WASM channel from a file pair.
@@ -114,7 +119,13 @@ impl WasmChannelLoader {
.await?;
// Create the channel
let channel = WasmChannel::new(self.runtime.clone(), prepared, capabilities, config_json);
let channel = WasmChannel::new(
self.runtime.clone(),
prepared,
capabilities,
config_json,
self.pairing_store.clone(),
);
tracing::info!(
name = name,
@@ -352,6 +363,7 @@ mod tests {
use crate::channels::wasm::loader::{WasmChannelLoader, discover_channels};
use crate::channels::wasm::runtime::{WasmChannelRuntime, WasmChannelRuntimeConfig};
use crate::pairing::PairingStore;
use std::sync::Arc;
#[tokio::test]
@@ -408,7 +420,7 @@ mod tests {
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 loader = WasmChannelLoader::new(runtime, Arc::new(PairingStore::new()));
let dir = TempDir::new().unwrap();
let wasm_path = dir.path().join("test.wasm");
src/channels/wasm/router.rs
+3 -1
View File
@@ -469,7 +469,7 @@ pub fn create_wasm_channel_router(
}
#[cfg(test)]
mod tests {
mod tests {
use std::sync::Arc;
use crate::channels::wasm::capabilities::ChannelCapabilities;
@@ -478,6 +478,7 @@ mod tests {
PreparedChannelModule, WasmChannelRuntime, WasmChannelRuntimeConfig,
};
use crate::channels::wasm::wrapper::WasmChannel;
use crate::pairing::PairingStore;
use crate::tools::wasm::ResourceLimits;
fn create_test_channel(name: &str) -> Arc<WasmChannel> {
@@ -499,6 +500,7 @@ mod tests {
prepared,
capabilities,
"{}".to_string(),
Arc::new(PairingStore::new()),
))
}
src/channels/wasm/wrapper.rs
+125 -16
View File
@@ -43,6 +43,7 @@ use wasmtime_wasi::{ResourceTable, WasiCtx, WasiCtxBuilder, WasiView};
use crate::channels::wasm::capabilities::ChannelCapabilities;
use crate::channels::wasm::error::WasmChannelError;
use crate::channels::wasm::host::{ChannelEmitRateLimiter, ChannelHostState, EmittedMessage};
use crate::pairing::PairingStore;
use crate::channels::wasm::router::RegisteredEndpoint;
use crate::channels::wasm::runtime::{PreparedChannelModule, WasmChannelRuntime};
use crate::channels::wasm::schema::ChannelConfig;
@@ -73,6 +74,8 @@ struct ChannelStoreData {
/// Injected credentials for URL substitution (e.g., bot tokens).
/// Keys are placeholder names like "TELEGRAM_BOT_TOKEN".
credentials: HashMap<String, String>,
/// Pairing store for DM pairing (guest access control).
pairing_store: Arc<PairingStore>,
}
impl ChannelStoreData {
@@ -81,6 +84,7 @@ impl ChannelStoreData {
channel_name: &str,
capabilities: ChannelCapabilities,
credentials: HashMap<String, String>,
pairing_store: Arc<PairingStore>,
) -> Self {
// Create a minimal WASI context (no filesystem, no env vars for security)
let wasi = WasiCtxBuilder::new().build();
@@ -91,6 +95,7 @@ impl ChannelStoreData {
wasi,
table: ResourceTable::new(),
credentials,
pairing_store,
}
}
@@ -403,6 +408,43 @@ impl near::agent::channel_host::Host for ChannelStoreData {
}
}
}
fn pairing_upsert_request(
&mut self,
channel: String,
id: String,
meta_json: String,
) -> Result<near::agent::channel_host::PairingUpsertResult, String> {
let meta = if meta_json.is_empty() {
None
} else {
serde_json::from_str(&meta_json).ok()
};
match self.pairing_store.upsert_request(&channel, &id, meta) {
Ok(r) => Ok(near::agent::channel_host::PairingUpsertResult {
code: r.code,
created: r.created,
}),
Err(e) => Err(e.to_string()),
}
}
fn pairing_is_allowed(
&mut self,
channel: String,
id: String,
username: Option<String>,
) -> Result<bool, String> {
self.pairing_store
.is_sender_allowed(&channel, &id, username.as_deref())
.map_err(|e| e.to_string())
}
fn pairing_read_allow_from(&mut self, channel: String) -> Result<Vec<String>, String> {
self.pairing_store
.read_allow_from(&channel)
.map_err(|e| e.to_string())
}
}
/// A WASM-based channel implementing the Channel trait.
@@ -455,6 +497,9 @@ pub struct WasmChannel {
/// Background task that repeats typing indicators every 4 seconds.
/// Telegram's "typing..." indicator expires after ~5s, so we refresh it.
typing_task: RwLock<Option<tokio::task::JoinHandle<()>>>,
/// Pairing store for DM pairing (guest access control).
pairing_store: Arc<PairingStore>,
}
impl WasmChannel {
@@ -464,6 +509,7 @@ impl WasmChannel {
prepared: Arc<PreparedChannelModule>,
capabilities: ChannelCapabilities,
config_json: String,
pairing_store: Arc<PairingStore>,
) -> Self {
let name = prepared.name.clone();
let rate_limiter = ChannelEmitRateLimiter::new(capabilities.emit_rate_limit.clone());
@@ -483,6 +529,7 @@ impl WasmChannel {
endpoints: RwLock::new(Vec::new()),
credentials: Arc::new(RwLock::new(HashMap::new())),
typing_task: RwLock::new(None),
pairing_store,
}
}
@@ -564,6 +611,7 @@ impl WasmChannel {
prepared: &PreparedChannelModule,
capabilities: &ChannelCapabilities,
credentials: HashMap<String, String>,
pairing_store: Arc<PairingStore>,
) -> Result<Store<ChannelStoreData>, WasmChannelError> {
let engine = runtime.engine();
let limits = &prepared.limits;
@@ -574,6 +622,7 @@ impl WasmChannel {
&prepared.name,
capabilities.clone(),
credentials,
pairing_store,
);
let mut store = Store::new(engine, store_data);
@@ -674,12 +723,18 @@ impl WasmChannel {
let timeout = self.runtime.config().callback_timeout;
let channel_name = self.name.clone();
let credentials = self.get_credentials().await;
let pairing_store = self.pairing_store.clone();
// Execute in blocking task with timeout
let result = tokio::time::timeout(timeout, async move {
tokio::task::spawn_blocking(move || {
let mut store =
Self::create_store(&runtime, &prepared, &capabilities, credentials)?;
let mut store = Self::create_store(
&runtime,
&prepared,
&capabilities,
credentials,
pairing_store,
)?;
let instance = Self::instantiate_component(&runtime, &prepared, &mut store)?;
// Call on_start using the generated typed interface
@@ -784,6 +839,7 @@ impl WasmChannel {
let capabilities = self.capabilities.clone();
let timeout = self.runtime.config().callback_timeout;
let credentials = self.get_credentials().await;
let pairing_store = self.pairing_store.clone();
// Prepare request data
let method = method.to_string();
@@ -797,8 +853,13 @@ impl WasmChannel {
// Execute in blocking task with timeout
let result = tokio::time::timeout(timeout, async move {
tokio::task::spawn_blocking(move || {
let mut store =
Self::create_store(&runtime, &prepared, &capabilities, credentials)?;
let mut store = Self::create_store(
&runtime,
&prepared,
&capabilities,
credentials,
pairing_store,
)?;
let instance = Self::instantiate_component(&runtime, &prepared, &mut store)?;
// Build the WIT request type
@@ -871,12 +932,18 @@ impl WasmChannel {
let timeout = self.runtime.config().callback_timeout;
let channel_name = self.name.clone();
let credentials = self.get_credentials().await;
let pairing_store = self.pairing_store.clone();
// Execute in blocking task with timeout
let result = tokio::time::timeout(timeout, async move {
tokio::task::spawn_blocking(move || {
let mut store =
Self::create_store(&runtime, &prepared, &capabilities, credentials)?;
let mut store = Self::create_store(
&runtime,
&prepared,
&capabilities,
credentials,
pairing_store,
)?;
let instance = Self::instantiate_component(&runtime, &prepared, &mut store)?;
// Call on_poll using the generated typed interface
@@ -960,6 +1027,7 @@ impl WasmChannel {
let timeout = self.runtime.config().callback_timeout;
let channel_name = self.name.clone();
let credentials = self.get_credentials().await;
let pairing_store = self.pairing_store.clone();
// Prepare response data
let message_id_str = message_id.to_string();
@@ -973,8 +1041,13 @@ impl WasmChannel {
let result = tokio::time::timeout(timeout, async move {
tokio::task::spawn_blocking(move || {
tracing::info!("Creating WASM store for on_respond");
let mut store =
Self::create_store(&runtime, &prepared, &capabilities, credentials)?;
let mut store = Self::create_store(
&runtime,
&prepared,
&capabilities,
credentials,
pairing_store,
)?;
tracing::info!("Instantiating WASM component for on_respond");
let instance = Self::instantiate_component(&runtime, &prepared, &mut store)?;
@@ -1067,13 +1140,19 @@ impl WasmChannel {
let timeout = self.runtime.config().callback_timeout;
let channel_name = self.name.clone();
let credentials = self.get_credentials().await;
let pairing_store = self.pairing_store.clone();
let wit_update = status_to_wit(status, metadata);
let result = tokio::time::timeout(timeout, async move {
tokio::task::spawn_blocking(move || {
let mut store =
Self::create_store(&runtime, &prepared, &capabilities, credentials)?;
let mut store = Self::create_store(
&runtime,
&prepared,
&capabilities,
credentials,
pairing_store,
)?;
let instance = Self::instantiate_component(&runtime, &prepared, &mut store)?;
let channel_iface = instance.near_agent_channel();
@@ -1117,6 +1196,7 @@ impl WasmChannel {
prepared: &Arc<PreparedChannelModule>,
capabilities: &ChannelCapabilities,
credentials: &RwLock<HashMap<String, String>>,
pairing_store: Arc<PairingStore>,
timeout: Duration,
wit_update: wit_channel::StatusUpdate,
) -> Result<(), WasmChannelError> {
@@ -1132,8 +1212,13 @@ impl WasmChannel {
let result = tokio::time::timeout(timeout, async move {
tokio::task::spawn_blocking(move || {
let mut store =
Self::create_store(&runtime, &prepared, &capabilities, credentials_snapshot)?;
let mut store = Self::create_store(
&runtime,
&prepared,
&capabilities,
credentials_snapshot,
pairing_store,
)?;
let instance = Self::instantiate_component(&runtime, &prepared, &mut store)?;
let channel_iface = instance.near_agent_channel();
@@ -1201,6 +1286,7 @@ impl WasmChannel {
let prepared = Arc::clone(&self.prepared);
let capabilities = self.capabilities.clone();
let credentials = self.credentials.clone();
let pairing_store = self.pairing_store.clone();
let callback_timeout = self.runtime.config().callback_timeout;
let wit_update = status_to_wit(&status, metadata);
@@ -1220,6 +1306,7 @@ impl WasmChannel {
&prepared,
&capabilities,
&credentials,
pairing_store.clone(),
callback_timeout,
wit_update_clone,
)
@@ -1350,6 +1437,7 @@ impl WasmChannel {
let message_tx = self.message_tx.clone();
let rate_limiter = self.rate_limiter.clone();
let credentials = self.credentials.clone();
let pairing_store = self.pairing_store.clone();
let callback_timeout = self.runtime.config().callback_timeout;
tokio::spawn(async move {
@@ -1371,6 +1459,7 @@ impl WasmChannel {
&prepared,
&capabilities,
&credentials,
pairing_store.clone(),
callback_timeout,
).await;
@@ -1422,6 +1511,7 @@ impl WasmChannel {
prepared: &Arc<PreparedChannelModule>,
capabilities: &ChannelCapabilities,
credentials: &RwLock<HashMap<String, String>>,
pairing_store: Arc<PairingStore>,
timeout: Duration,
) -> Result<Vec<EmittedMessage>, WasmChannelError> {
// Skip if no WASM bytes (testing mode)
@@ -1442,8 +1532,13 @@ impl WasmChannel {
// Execute in blocking task with timeout
let result = tokio::time::timeout(timeout, async move {
tokio::task::spawn_blocking(move || {
let mut store =
Self::create_store(&runtime, &prepared, &capabilities, credentials_snapshot)?;
let mut store = Self::create_store(
&runtime,
&prepared,
&capabilities,
credentials_snapshot,
pairing_store,
)?;
let instance = Self::instantiate_component(&runtime, &prepared, &mut store)?;
// Call on_poll using the generated typed interface
@@ -1978,6 +2073,7 @@ mod tests {
use std::sync::Arc;
use crate::channels::Channel;
use crate::pairing::PairingStore;
use crate::channels::wasm::capabilities::ChannelCapabilities;
use crate::channels::wasm::runtime::{
PreparedChannelModule, WasmChannelRuntime, WasmChannelRuntimeConfig,
@@ -1998,7 +2094,13 @@ mod tests {
let capabilities = ChannelCapabilities::for_channel("test").with_path("/webhook/test");
WasmChannel::new(runtime, prepared, capabilities, "{}".to_string())
WasmChannel::new(
runtime,
prepared,
capabilities,
"{}".to_string(),
Arc::new(PairingStore::new()),
)
}
#[test]
@@ -2073,6 +2175,7 @@ mod tests {
&prepared,
&capabilities,
&credentials,
Arc::new(PairingStore::new()),
timeout,
)
.await;
@@ -2166,7 +2269,13 @@ mod tests {
.with_path("/webhook/poll")
.with_polling(1000);
let channel = WasmChannel::new(runtime, prepared, capabilities, "{}".to_string());
let channel = WasmChannel::new(
runtime,
prepared,
capabilities,
"{}".to_string(),
Arc::new(PairingStore::new()),
);
// Start the channel
let _stream = channel.start().await.expect("Channel should start");