feat(channels): add Linq channel for iMessage/RCS/SMS support
The existing iMessage channel relies on AppleScript and only works on macOS. Linq provides a REST API for iMessage, RCS, and SMS — this gives ZeroClaw native iMessage support on any platform via webhooks. Implements LinqChannel following the same patterns as WhatsAppChannel: - Channel trait impl (send, listen, health_check, typing indicators) - Webhook handler with HMAC-SHA256 signature verification - Sender allowlist filtering - Onboarding wizard step with connection testing - 18 unit tests covering parsing, auth, and signature verification Resolves #656 — the prior issue was closed without a merged PR, so this is the actual implementation.
This commit is contained in:
George McCain
Chummy
parent
e23edde44b
commit
361e750576
5 changed files with 1003 additions and 5 deletions
|
|
@ -7,7 +7,7 @@
|
|||
//! - Request timeouts (30s) to prevent slow-loris attacks
|
||||
//! - Header sanitization (handled by axum/hyper)
|
||||
|
||||
use crate::channels::{Channel, SendMessage, WhatsAppChannel};
|
||||
use crate::channels::{Channel, LinqChannel, SendMessage, WhatsAppChannel};
|
||||
use crate::config::Config;
|
||||
use crate::memory::{self, Memory, MemoryCategory};
|
||||
use crate::providers::{self, Provider};
|
||||
|
|
@ -53,6 +53,10 @@ fn whatsapp_memory_key(msg: &crate::channels::traits::ChannelMessage) -> String
|
|||
format!("whatsapp_{}_{}", msg.sender, msg.id)
|
||||
}
|
||||
|
||||
fn linq_memory_key(msg: &crate::channels::traits::ChannelMessage) -> String {
|
||||
format!("linq_{}_{}", msg.sender, msg.id)
|
||||
}
|
||||
|
||||
fn hash_webhook_secret(value: &str) -> String {
|
||||
use sha2::{Digest, Sha256};
|
||||
|
||||
|
|
@ -274,6 +278,9 @@ pub struct AppState {
|
|||
pub whatsapp: Option<Arc<WhatsAppChannel>>,
|
||||
/// `WhatsApp` app secret for webhook signature verification (`X-Hub-Signature-256`)
|
||||
pub whatsapp_app_secret: Option<Arc<str>>,
|
||||
pub linq: Option<Arc<LinqChannel>>,
|
||||
/// Linq webhook signing secret for signature verification
|
||||
pub linq_signing_secret: Option<Arc<str>>,
|
||||
/// Observability backend for metrics scraping
|
||||
pub observer: Arc<dyn crate::observability::Observer>,
|
||||
}
|
||||
|
|
@ -389,6 +396,34 @@ pub async fn run_gateway(host: &str, port: u16, config: Config) -> Result<()> {
|
|||
})
|
||||
.map(Arc::from);
|
||||
|
||||
// Linq channel (if configured)
|
||||
let linq_channel: Option<Arc<LinqChannel>> = config.channels_config.linq.as_ref().map(|lq| {
|
||||
Arc::new(LinqChannel::new(
|
||||
lq.api_token.clone(),
|
||||
lq.from_phone.clone(),
|
||||
lq.allowed_senders.clone(),
|
||||
))
|
||||
});
|
||||
|
||||
// Linq signing secret for webhook signature verification
|
||||
// Priority: environment variable > config file
|
||||
let linq_signing_secret: Option<Arc<str>> = std::env::var("ZEROCLAW_LINQ_SIGNING_SECRET")
|
||||
.ok()
|
||||
.and_then(|secret| {
|
||||
let secret = secret.trim();
|
||||
(!secret.is_empty()).then(|| secret.to_owned())
|
||||
})
|
||||
.or_else(|| {
|
||||
config.channels_config.linq.as_ref().and_then(|lq| {
|
||||
lq.signing_secret
|
||||
.as_deref()
|
||||
.map(str::trim)
|
||||
.filter(|secret| !secret.is_empty())
|
||||
.map(ToOwned::to_owned)
|
||||
})
|
||||
})
|
||||
.map(Arc::from);
|
||||
|
||||
// ── Pairing guard ──────────────────────────────────────
|
||||
let pairing = Arc::new(PairingGuard::new(
|
||||
config.gateway.require_pairing,
|
||||
|
|
@ -440,6 +475,9 @@ pub async fn run_gateway(host: &str, port: u16, config: Config) -> Result<()> {
|
|||
println!(" GET /whatsapp — Meta webhook verification");
|
||||
println!(" POST /whatsapp — WhatsApp message webhook");
|
||||
}
|
||||
if linq_channel.is_some() {
|
||||
println!(" POST /linq — Linq message webhook (iMessage/RCS/SMS)");
|
||||
}
|
||||
println!(" GET /health — health check");
|
||||
println!(" GET /metrics — Prometheus metrics");
|
||||
if let Some(code) = pairing.pairing_code() {
|
||||
|
|
@ -476,6 +514,8 @@ pub async fn run_gateway(host: &str, port: u16, config: Config) -> Result<()> {
|
|||
idempotency_store,
|
||||
whatsapp: whatsapp_channel,
|
||||
whatsapp_app_secret,
|
||||
linq: linq_channel,
|
||||
linq_signing_secret,
|
||||
observer,
|
||||
};
|
||||
|
||||
|
|
@ -487,6 +527,7 @@ pub async fn run_gateway(host: &str, port: u16, config: Config) -> Result<()> {
|
|||
.route("/webhook", post(handle_webhook))
|
||||
.route("/whatsapp", get(handle_whatsapp_verify))
|
||||
.route("/whatsapp", post(handle_whatsapp_message))
|
||||
.route("/linq", post(handle_linq_webhook))
|
||||
.with_state(state)
|
||||
.layer(RequestBodyLimitLayer::new(MAX_BODY_SIZE))
|
||||
.layer(TimeoutLayer::with_status_code(
|
||||
|
|
@ -967,6 +1008,118 @@ async fn handle_whatsapp_message(
|
|||
(StatusCode::OK, Json(serde_json::json!({"status": "ok"})))
|
||||
}
|
||||
|
||||
/// POST /linq — incoming message webhook (iMessage/RCS/SMS via Linq)
|
||||
async fn handle_linq_webhook(
|
||||
State(state): State<AppState>,
|
||||
headers: HeaderMap,
|
||||
body: Bytes,
|
||||
) -> impl IntoResponse {
|
||||
let Some(ref linq) = state.linq else {
|
||||
return (
|
||||
StatusCode::NOT_FOUND,
|
||||
Json(serde_json::json!({"error": "Linq not configured"})),
|
||||
);
|
||||
};
|
||||
|
||||
let body_str = String::from_utf8_lossy(&body);
|
||||
|
||||
// ── Security: Verify X-Webhook-Signature if signing_secret is configured ──
|
||||
if let Some(ref signing_secret) = state.linq_signing_secret {
|
||||
let timestamp = headers
|
||||
.get("X-Webhook-Timestamp")
|
||||
.and_then(|v| v.to_str().ok())
|
||||
.unwrap_or("");
|
||||
|
||||
let signature = headers
|
||||
.get("X-Webhook-Signature")
|
||||
.and_then(|v| v.to_str().ok())
|
||||
.unwrap_or("");
|
||||
|
||||
if !crate::channels::linq::verify_linq_signature(
|
||||
signing_secret,
|
||||
&body_str,
|
||||
timestamp,
|
||||
signature,
|
||||
) {
|
||||
tracing::warn!(
|
||||
"Linq webhook signature verification failed (signature: {})",
|
||||
if signature.is_empty() {
|
||||
"missing"
|
||||
} else {
|
||||
"invalid"
|
||||
}
|
||||
);
|
||||
return (
|
||||
StatusCode::UNAUTHORIZED,
|
||||
Json(serde_json::json!({"error": "Invalid signature"})),
|
||||
);
|
||||
}
|
||||
}
|
||||
|
||||
// Parse JSON body
|
||||
let Ok(payload) = serde_json::from_slice::<serde_json::Value>(&body) else {
|
||||
return (
|
||||
StatusCode::BAD_REQUEST,
|
||||
Json(serde_json::json!({"error": "Invalid JSON payload"})),
|
||||
);
|
||||
};
|
||||
|
||||
// Parse messages from the webhook payload
|
||||
let messages = linq.parse_webhook_payload(&payload);
|
||||
|
||||
if messages.is_empty() {
|
||||
// Acknowledge the webhook even if no messages (could be status/delivery events)
|
||||
return (StatusCode::OK, Json(serde_json::json!({"status": "ok"})));
|
||||
}
|
||||
|
||||
// Process each message
|
||||
for msg in &messages {
|
||||
tracing::info!(
|
||||
"Linq message from {}: {}",
|
||||
msg.sender,
|
||||
truncate_with_ellipsis(&msg.content, 50)
|
||||
);
|
||||
|
||||
// Auto-save to memory
|
||||
if state.auto_save {
|
||||
let key = linq_memory_key(msg);
|
||||
let _ = state
|
||||
.mem
|
||||
.store(&key, &msg.content, MemoryCategory::Conversation, None)
|
||||
.await;
|
||||
}
|
||||
|
||||
// Call the LLM
|
||||
match state
|
||||
.provider
|
||||
.simple_chat(&msg.content, &state.model, state.temperature)
|
||||
.await
|
||||
{
|
||||
Ok(response) => {
|
||||
// Send reply via Linq
|
||||
if let Err(e) = linq
|
||||
.send(&SendMessage::new(response, &msg.reply_target))
|
||||
.await
|
||||
{
|
||||
tracing::error!("Failed to send Linq reply: {e}");
|
||||
}
|
||||
}
|
||||
Err(e) => {
|
||||
tracing::error!("LLM error for Linq message: {e:#}");
|
||||
let _ = linq
|
||||
.send(&SendMessage::new(
|
||||
"Sorry, I couldn't process your message right now.",
|
||||
&msg.reply_target,
|
||||
))
|
||||
.await;
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// Acknowledge the webhook
|
||||
(StatusCode::OK, Json(serde_json::json!({"status": "ok"})))
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use super::*;
|
||||
|
|
@ -1433,6 +1586,8 @@ mod tests {
|
|||
idempotency_store: Arc::new(IdempotencyStore::new(Duration::from_secs(300), 1000)),
|
||||
whatsapp: None,
|
||||
whatsapp_app_secret: None,
|
||||
linq: None,
|
||||
linq_signing_secret: None,
|
||||
observer: Arc::new(crate::observability::NoopObserver),
|
||||
};
|
||||
|
||||
|
|
@ -1489,6 +1644,8 @@ mod tests {
|
|||
idempotency_store: Arc::new(IdempotencyStore::new(Duration::from_secs(300), 1000)),
|
||||
whatsapp: None,
|
||||
whatsapp_app_secret: None,
|
||||
linq: None,
|
||||
linq_signing_secret: None,
|
||||
observer: Arc::new(crate::observability::NoopObserver),
|
||||
};
|
||||
|
||||
|
|
@ -1557,6 +1714,8 @@ mod tests {
|
|||
idempotency_store: Arc::new(IdempotencyStore::new(Duration::from_secs(300), 1000)),
|
||||
whatsapp: None,
|
||||
whatsapp_app_secret: None,
|
||||
linq: None,
|
||||
linq_signing_secret: None,
|
||||
observer: Arc::new(crate::observability::NoopObserver),
|
||||
};
|
||||
|
||||
|
|
@ -1597,6 +1756,8 @@ mod tests {
|
|||
idempotency_store: Arc::new(IdempotencyStore::new(Duration::from_secs(300), 1000)),
|
||||
whatsapp: None,
|
||||
whatsapp_app_secret: None,
|
||||
linq: None,
|
||||
linq_signing_secret: None,
|
||||
observer: Arc::new(crate::observability::NoopObserver),
|
||||
};
|
||||
|
||||
|
|
@ -1642,6 +1803,8 @@ mod tests {
|
|||
idempotency_store: Arc::new(IdempotencyStore::new(Duration::from_secs(300), 1000)),
|
||||
whatsapp: None,
|
||||
whatsapp_app_secret: None,
|
||||
linq: None,
|
||||
linq_signing_secret: None,
|
||||
observer: Arc::new(crate::observability::NoopObserver),
|
||||
};
|
||||
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue