From b0b3a50fa38d69b789fd5f0217e9b48fc6456193 Mon Sep 17 00:00:00 2001 From: ibhagwan <59988195+ibhagwan@users.noreply.github.com> Date: Tue, 24 Feb 2026 01:25:16 -0500 Subject: [PATCH] feat(channels): add native Signal channel via signal-cli HTTP daemon (#271) * feat(channels): add native Signal channel via signal-cli HTTP daemon Implement a native Rust Signal channel that connects to a running signal-cli daemon's HTTP endpoint, enabling Signal messaging without WASM overhead. Architecture: - SSE listener at /api/v1/events for receiving messages with automatic reconnection and exponential backoff - JSON-RPC client at /api/v1/rpc for sending messages and typing indicators - Reply target tracking via Arc> to route responses back to the correct DM or group conversation Features: - User allowlisting supporting E.164 phone numbers, bare UUIDs, and uuid:-prefixed identifiers (matching OpenClaw's format) - Group allowlisting with wildcard (*) support - Configurable story and attachment-only message filtering - Health check via signal-cli /api/v1/check - Broadcast support to all tracked reply targets Configuration via environment variables: - SIGNAL_HTTP_URL, SIGNAL_ACCOUNT (required) - SIGNAL_ALLOWED_USERS, SIGNAL_ALLOWED_GROUPS - SIGNAL_IGNORE_ATTACHMENTS (default: false) - SIGNAL_IGNORE_STORIES (default: true) Includes unit tests covering allowlist logic, envelope parsing, recipient targeting, SSE deserialization, and edge cases. * refactor(signal): remove expect|unwrap calls - Change SignalChannel::new to return Result - Replace .expect() on reqwest client build with proper error handling - Replace .expect() on NonZeroUsize with compile-time const using unsafe new_unchecked - Propagate errors through test helpers to avoid unwraps in tests * fix(signal): prevent OOM from chunked response without Content-Length Use bytes_stream() to check response size during download rather than buffering entire body first. This closes the OOM vector where a malicious signal-cli daemon could send unbounded chunked data. * fix(signal): align is_e164 minimum digits with setup wizard Both now require 7-15 digits after '+', preventing environment variable bypass of the stricter onboarding validation. * refactor(signal): extract from_parts constructor Extract SignalChannel::from_parts() used by both new() and sse_listener() to ensure consistent object construction. * chore: remove redundant unused var * refactor(signal): rename allowed_users to allow_from and add dm_policy/group_policy - Rename allowed_users -> allow_from for consistency with other channels - Rename allowed_groups -> allow_from_groups - Add dm_policy field: 'open', 'allowlist', or 'pairing' (default: 'pairing') - Add group_policy field: 'allowlist', 'open', or 'disabled' (default: 'allowlist') - Add group_allow_from field that inherits from allow_from if empty - Implement dm_policy and group_policy logic in message processing - Add environment variable resolution: SIGNAL_ALLOW_FROM, SIGNAL_ALLOW_FROM_GROUPS, SIGNAL_DM_POLICY, SIGNAL_GROUP_POLICY, SIGNAL_GROUP_ALLOW_FROM - Add setup wizard prompts for new policy options - Note: full pairing flow (PairingStore integration) marked as pending for future PR * feat(signal): implement DM pairing workflow for unapproved senders - Add PairingStore integration to check approved senders - Handle pairing requests for unknown senders with dm_policy=pairing - Send pairing reply message with approval instructions - Update FEATURE_PARITY.md to reflect DM pairing support * chore(ci): fix clippy warnings --- .env.example | 11 + Cargo.lock | 1 + Cargo.toml | 3 +- FEATURE_PARITY.md | 3 +- src/channels/mod.rs | 2 + src/channels/signal.rs | 2220 ++++++++++++++++++++++++++++++++++++ src/config/channels.rs | 97 ++ src/config/mod.rs | 2 +- src/extensions/registry.rs | 2 +- src/main.rs | 21 +- src/pairing/mod.rs | 2 +- src/safety/sanitizer.rs | 2 +- src/settings.rs | 35 + src/setup/channels.rs | 197 ++++ src/setup/wizard.rs | 69 +- src/skills/selector.rs | 2 +- 16 files changed, 2655 insertions(+), 14 deletions(-) create mode 100644 src/channels/signal.rs diff --git a/.env.example b/.env.example index 62583c1b..64a688a8 100644 --- a/.env.example +++ b/.env.example @@ -75,6 +75,17 @@ HTTP_HOST=0.0.0.0 HTTP_PORT=8080 HTTP_WEBHOOK_SECRET=your-webhook-secret +# Signal Channel (optional, requires signal-cli daemon --http) +# SIGNAL_HTTP_URL=http://127.0.0.1:8080 +# SIGNAL_ACCOUNT=+1234567890 +# SIGNAL_ALLOW_FROM=+1234567890,uuid:xxxxxxxx-xxxx-xxxx-xxxx-xxxxxxxxxxxx # comma-separated, * for all, empty = deny/require pairing +# SIGNAL_ALLOW_FROM_GROUPS= # comma-separated group IDs, * for all, empty = deny all groups +# SIGNAL_DM_POLICY=pairing # open | allowlist | pairing +# SIGNAL_GROUP_POLICY=allowlist # allowlist | open | disabled +# SIGNAL_GROUP_ALLOW_FROM= # comma-separated, empty = inherit from ALLOW_FROM +# SIGNAL_IGNORE_ATTACHMENTS=false +# SIGNAL_IGNORE_STORIES=true + # Agent Settings AGENT_NAME=ironclaw AGENT_MAX_PARALLEL_JOBS=5 diff --git a/Cargo.lock b/Cargo.lock index adfc1070..3695f892 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -2728,6 +2728,7 @@ dependencies = [ "hyper 1.8.1", "hyper-util", "libsql", + "lru", "mime_guess", "open", "pgvector", diff --git a/Cargo.toml b/Cargo.toml index 98f81f6a..78801a78 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -69,7 +69,7 @@ dotenvy = "0.15" toml = "0.8" # Core types -uuid = { version = "1", features = ["v4", "serde"] } +uuid = { version = "1", features = ["v4", "v5", "serde"] } chrono = { version = "0.4", features = ["serde"] } rust_decimal = { version = "1", features = ["serde", "serde-with-str", "maths"] } rust_decimal_macros = "1" @@ -149,6 +149,7 @@ bytes = "1" base64 = "0.22.1" mime_guess = "2.0.5" clap_complete = "4.5.0" +lru = "0.16.3" # HTML to Markdown conversion (feature gated) html-to-markdown-rs = { version = "2.3", optional = true } diff --git a/FEATURE_PARITY.md b/FEATURE_PARITY.md index ba7b5c24..82981a57 100644 --- a/FEATURE_PARITY.md +++ b/FEATURE_PARITY.md @@ -68,7 +68,7 @@ This document tracks feature parity between IronClaw (Rust implementation) and O | WhatsApp | ✅ | ❌ | P1 | Baileys (Web), same-phone mode with echo detection | | Telegram | ✅ | ✅ | - | WASM channel(MTProto), DM pairing, caption, /start, bot_username | | Discord | ✅ | ❌ | P2 | discord.js, thread parent binding inheritance | -| Signal | ✅ | ❌ | P2 | signal-cli | +| Signal | ✅ | ✅ | P2 | signal-cli daemonPC, SSE listener HTTP/JSON-R, user/group allowlists, DM pairing | | Slack | ✅ | ✅ | - | WASM tool | | iMessage | ✅ | ❌ | P3 | BlueBubbles or Linq recommended | | Linq | ✅ | ❌ | P3 | Real iMessage via API, no Mac required | @@ -540,7 +540,6 @@ This document tracks feature parity between IronClaw (Rust implementation) and O ### P3 - Lower Priority - ❌ Discord channel -- ❌ Signal channel - ❌ Matrix channel - ❌ Other messaging platforms - ❌ TTS/audio features diff --git a/src/channels/mod.rs b/src/channels/mod.rs index 08d742e8..ad7320d3 100644 --- a/src/channels/mod.rs +++ b/src/channels/mod.rs @@ -31,6 +31,7 @@ mod channel; mod http; mod manager; mod repl; +mod signal; pub mod wasm; pub mod web; mod webhook_server; @@ -39,5 +40,6 @@ pub use channel::{Channel, IncomingMessage, MessageStream, OutgoingResponse, Sta pub use http::HttpChannel; pub use manager::ChannelManager; pub use repl::ReplChannel; +pub use signal::SignalChannel; pub use web::GatewayChannel; pub use webhook_server::{WebhookServer, WebhookServerConfig}; diff --git a/src/channels/signal.rs b/src/channels/signal.rs new file mode 100644 index 00000000..a4f5867c --- /dev/null +++ b/src/channels/signal.rs @@ -0,0 +1,2220 @@ +//! Signal channel via signal-cli daemon HTTP/JSON-RPC. +//! +//! Connects to a running `signal-cli daemon --http `. +//! Listens for messages via SSE at `/api/v1/events` and sends via +//! JSON-RPC at `/api/v1/rpc`. + +use std::num::NonZeroUsize; +use std::sync::Arc; +use std::time::Duration; + +use async_trait::async_trait; +use futures::StreamExt; +use lru::LruCache; +use reqwest::Client; +use serde::Deserialize; +use tokio::sync::RwLock; +use uuid::Uuid; + +use crate::channels::{Channel, IncomingMessage, MessageStream, OutgoingResponse, StatusUpdate}; +use crate::config::SignalConfig; +use crate::error::ChannelError; +use crate::pairing::PairingStore; + +const GROUP_TARGET_PREFIX: &str = "group:"; +const SIGNAL_HEALTH_ENDPOINT: &str = "/api/v1/check"; + +const MAX_SSE_BUFFER_SIZE: usize = 1024 * 1024; +const MAX_SSE_EVENT_SIZE: usize = 256 * 1024; +const MAX_HTTP_RESPONSE_SIZE: usize = 10 * 1024 * 1024; +const MAX_REPLY_TARGETS: usize = 10000; +const MAX_ERROR_LOG_BODY: usize = 1024; + +const REPLY_TARGETS_CAP: NonZeroUsize = NonZeroUsize::new(MAX_REPLY_TARGETS).unwrap(); + +/// Recipient classification for outbound messages. +#[derive(Debug, Clone, PartialEq, Eq)] +enum RecipientTarget { + Direct(String), + Group(String), +} + +// ── signal-cli SSE event JSON shapes ──────────────────────────── + +#[derive(Debug, Deserialize)] +struct SseEnvelope { + #[serde(default)] + envelope: Option, +} + +#[derive(Debug, Deserialize)] +struct Envelope { + #[serde(default)] + source: Option, + #[serde(rename = "sourceNumber", default)] + source_number: Option, + #[serde(rename = "sourceName", default)] + source_name: Option, + #[serde(rename = "sourceUuid", default)] + source_uuid: Option, + #[serde(rename = "dataMessage", default)] + data_message: Option, + #[serde(rename = "storyMessage", default)] + story_message: Option, + #[serde(default)] + timestamp: Option, +} + +#[derive(Debug, Deserialize)] +struct DataMessage { + #[serde(default)] + message: Option, + #[serde(default)] + timestamp: Option, + #[serde(rename = "groupInfo", default)] + group_info: Option, + #[serde(default)] + attachments: Option>, +} + +#[derive(Debug, Deserialize)] +struct GroupInfo { + #[serde(rename = "groupId", default)] + group_id: Option, +} + +/// Signal channel using signal-cli daemon's native JSON-RPC + SSE API. +pub struct SignalChannel { + config: SignalConfig, + client: Client, + /// LRU cache of reply targets per incoming message, used by `respond()`. + /// Bounded to `MAX_REPLY_TARGETS` entries; least-recently-used entries + /// are evicted automatically when the cache is full. + reply_targets: Arc>>, +} + +impl SignalChannel { + /// Create a new Signal channel with normalized config and fresh client/cache. + pub fn new(config: SignalConfig) -> Result { + let mut config = config; + config.http_url = config.http_url.trim_end_matches('/').to_string(); + + let client = Client::builder() + .connect_timeout(Duration::from_secs(10)) + .build() + .map_err(|e| ChannelError::Http(e.to_string()))?; + + let cap = REPLY_TARGETS_CAP; + let reply_targets = Arc::new(RwLock::new(LruCache::new(cap))); + + Ok(Self::from_parts(config, client, reply_targets)) + } + + /// Construct a SignalChannel from pre-validated parts. + /// + /// Used by [`new()`][Self::new] after normalization and by [`sse_listener`] + /// to ensure both code paths use the same constructor. + fn from_parts( + config: SignalConfig, + client: Client, + reply_targets: Arc>>, + ) -> Self { + Self { + config, + client, + reply_targets, + } + } + + /// Effective sender: prefer `sourceNumber` (E.164), fall back to `source` + /// (UUID for privacy-enabled users). + fn sender(envelope: &Envelope) -> Option { + envelope + .source_number + .as_deref() + .or(envelope.source.as_deref()) + .map(String::from) + } + + /// Normalize an allowlist entry to the bare identifier. + /// + /// Strips the `uuid:` prefix if present, so `uuid:` and `` both + /// match against a bare UUID sender. + fn normalize_allow_entry(entry: &str) -> &str { + entry.strip_prefix("uuid:").unwrap_or(entry) + } + + /// Check whether a sender is in the allowed users list. + fn is_sender_allowed(&self, sender: &str) -> bool { + if self.config.allow_from.is_empty() { + return false; + } + self.config.allow_from.iter().any(|entry| { + entry == "*" + || Self::normalize_allow_entry(entry) == Self::normalize_allow_entry(sender) + }) + } + + /// Check if sender is allowed via config allow_from OR pairing store. + fn is_sender_allowed_with_pairing(&self, sender: &str) -> bool { + if self.is_sender_allowed(sender) { + return true; + } + let store = PairingStore::new(); + if let Ok(allowed) = store.read_allow_from("signal") { + return allowed.iter().any(|entry| entry == "*" || entry == sender); + } + false + } + + /// Handle pairing request for unapproved sender. + /// Returns Ok(true) if message should be allowed (was already paired), + /// Ok(false) if message was blocked but pairing request was processed. + fn handle_pairing_request(&self, sender: &str, source_name: Option<&str>) -> Result { + let store = PairingStore::new(); + let meta = serde_json::json!({ + "sender": sender, + "name": source_name, + }); + + match store.upsert_request("signal", sender, Some(meta)) { + Ok(result) => { + tracing::info!( + sender = %sender, + code = %result.code, + "Signal: pairing request upserted" + ); + if result.created { + let message = format!( + "To pair with this bot, run: `ironclaw pairing approve signal {}`", + result.code + ); + let http_url = self.config.http_url.clone(); + let account = self.config.account.clone(); + let sender_owned = sender.to_string(); + let message_owned = message.clone(); + tokio::spawn(async move { + if let Err(e) = Self::send_pairing_reply_async( + &http_url, + &account, + &sender_owned, + &message_owned, + ) + .await + { + tracing::error!(sender = %sender_owned, error = %e, "Signal: failed to send pairing reply"); + } + }); + } + Ok(false) + } + Err(e) => { + tracing::error!(sender = %sender, error = %e, "Signal: pairing upsert failed"); + Err(()) + } + } + } + + /// Send a pairing reply message to the sender (async helper for spawned task). + async fn send_pairing_reply_async( + http_url: &str, + account: &str, + recipient: &str, + message: &str, + ) -> Result<(), ChannelError> { + let client = Client::builder() + .connect_timeout(Duration::from_secs(10)) + .build() + .map_err(|e| ChannelError::Http(e.to_string()))?; + + let target = Self::parse_recipient_target(recipient); + let params = Self::build_rpc_params_static(http_url, account, &target, Some(message)); + + let url = format!("{}/api/v1/rpc", http_url); + let id = Uuid::new_v4().to_string(); + + let body = serde_json::json!({ + "jsonrpc": "2.0", + "method": "send", + "params": params, + "id": id, + }); + + let resp = client + .post(&url) + .timeout(Duration::from_secs(30)) + .header("Content-Type", "application/json") + .json(&body) + .send() + .await + .map_err(|e| ChannelError::SendFailed { + name: "signal".to_string(), + reason: format!("RPC request failed to {}: {e}", Self::redact_url(&url)), + })?; + + let status = resp.status(); + let is_success = status.is_success(); + + if status.as_u16() == 201 { + return Ok(()); + } + + if !is_success { + let bytes = resp.bytes().await.unwrap_or_default(); + let truncated_len = bytes.len().min(MAX_ERROR_LOG_BODY); + let truncated_body = String::from_utf8_lossy(&bytes[..truncated_len]); + return Err(ChannelError::SendFailed { + name: "signal".to_string(), + reason: format!("HTTP error {}: {}", status.as_u16(), truncated_body), + }); + } + + Ok(()) + } + + /// Get effective group allow_from list (inherits from allow_from if empty). + fn effective_group_allow_from(&self) -> &[String] { + if self.config.group_allow_from.is_empty() { + &self.config.allow_from + } else { + &self.config.group_allow_from + } + } + + /// Check whether a group is in the allowed groups list. + /// + /// - Empty list — deny all groups (DMs only, secure by default). + /// - `*` — allow all groups. + /// - Specific IDs — allow only those groups. + fn is_group_allowed(&self, group_id: &str) -> bool { + if self.config.allow_from_groups.is_empty() { + return false; + } + self.config + .allow_from_groups + .iter() + .any(|entry| entry == "*" || entry == group_id) + } + + /// Check whether a sender is allowed for group messages. + fn is_group_sender_allowed(&self, sender: &str) -> bool { + let effective_list = self.effective_group_allow_from(); + if effective_list.is_empty() { + return false; + } + effective_list.iter().any(|entry| { + entry == "*" + || Self::normalize_allow_entry(entry) == Self::normalize_allow_entry(sender) + }) + } + + /// Redact credentials from a URL for safe logging. + /// + /// Replaces any embedded username/password with `**REDACTED**` and returns + /// the sanitised string. Returns `""` when parsing fails. + pub fn redact_url(url: &str) -> String { + reqwest::Url::parse(url) + .map(|mut u| { + if u.password().is_some() || !u.username().is_empty() { + let _ = u.set_username("**REDACTED**"); + let _ = u.set_password(None); + } + u.to_string() + }) + .unwrap_or_else(|_| "".to_string()) + } + + fn is_e164(recipient: &str) -> bool { + let Some(number) = recipient.strip_prefix('+') else { + return false; + }; + (7..=15).contains(&number.len()) && number.chars().all(|c| c.is_ascii_digit()) + } + + /// Check whether a string is a valid UUID (signal-cli uses these for + /// privacy-enabled users who have opted out of sharing their phone number). + fn is_uuid(s: &str) -> bool { + Uuid::parse_str(s).is_ok() + } + + /// Generate a deterministic UUID from an identifier (phone number or group ID). + /// + /// This ensures that the same phone number or group always produces the same UUID, + /// allowing conversation history to persist across gateway restarts. + fn thread_id_from_identifier(identifier: &str) -> String { + // Use a stable, deterministic UUID v5 derived from the identifier. + // This avoids relying on `DefaultHasher` implementation details and + // provides a full 128 bits of entropy. + Uuid::new_v5(&Uuid::NAMESPACE_URL, identifier.as_bytes()).to_string() + } + + fn parse_recipient_target(recipient: &str) -> RecipientTarget { + if let Some(group_id) = recipient.strip_prefix(GROUP_TARGET_PREFIX) { + return RecipientTarget::Group(group_id.to_string()); + } + + if Self::is_e164(recipient) || Self::is_uuid(recipient) { + RecipientTarget::Direct(recipient.to_string()) + } else { + RecipientTarget::Group(recipient.to_string()) + } + } + + /// Determine the reply target: group id (prefixed) or the sender's identifier. + fn reply_target(data_msg: &DataMessage, sender: &str) -> String { + if let Some(group_id) = data_msg + .group_info + .as_ref() + .and_then(|g| g.group_id.as_deref()) + { + format!("{GROUP_TARGET_PREFIX}{group_id}") + } else { + sender.to_string() + } + } + + /// Send a JSON-RPC request to signal-cli daemon. + async fn rpc_request( + &self, + method: &str, + params: serde_json::Value, + ) -> Result, ChannelError> { + let url = format!("{}/api/v1/rpc", self.config.http_url); + let id = Uuid::new_v4().to_string(); + + let body = serde_json::json!({ + "jsonrpc": "2.0", + "method": method, + "params": params, + "id": id, + }); + + let resp = self + .client + .post(&url) + .timeout(Duration::from_secs(30)) + .header("Content-Type", "application/json") + .json(&body) + .send() + .await + .map_err(|e| ChannelError::SendFailed { + name: "signal".to_string(), + reason: format!("RPC request failed to {}: {e}", Self::redact_url(&url)), + })?; + + // 201 = success with no body (e.g. typing indicators). + if resp.status().as_u16() == 201 { + return Ok(None); + } + + // Reject obviously oversized responses before buffering. + if let Some(len) = resp.content_length() + && len as usize > MAX_HTTP_RESPONSE_SIZE + { + return Err(ChannelError::SendFailed { + name: "signal".to_string(), + reason: format!( + "RPC response Content-Length too large: {} bytes (max {})", + len, MAX_HTTP_RESPONSE_SIZE + ), + }); + } + + let status = resp.status(); + let mut stream = resp.bytes_stream(); + let mut total_bytes = 0usize; + let mut body = Vec::new(); + + while let Some(chunk) = stream.next().await { + let chunk = chunk.map_err(|e| ChannelError::SendFailed { + name: "signal".to_string(), + reason: format!("Failed to read RPC response: {e}"), + })?; + let chunk_len = chunk.len(); + total_bytes += chunk_len; + + if total_bytes > MAX_HTTP_RESPONSE_SIZE { + return Err(ChannelError::SendFailed { + name: "signal".to_string(), + reason: format!( + "RPC response too large: {} bytes (max {})", + total_bytes, MAX_HTTP_RESPONSE_SIZE + ), + }); + } + + body.extend_from_slice(&chunk); + } + + let bytes = body; + + if bytes.is_empty() { + return Ok(None); + } + + // Check for non-success HTTP status codes before parsing as JSON. + if !status.is_success() { + let truncated_len = std::cmp::min(bytes.len(), 512); + let truncated_body = String::from_utf8_lossy(&bytes[..truncated_len]); + return Err(ChannelError::SendFailed { + name: "signal".to_string(), + reason: format!("HTTP error {}: {}", status.as_u16(), truncated_body), + }); + } + + let parsed: serde_json::Value = + serde_json::from_slice(&bytes).map_err(|e| ChannelError::SendFailed { + name: "signal".to_string(), + reason: format!("Invalid RPC response JSON: {e}"), + })?; + + if let Some(err) = parsed.get("error") { + let code = err.get("code").and_then(|c| c.as_i64()).unwrap_or(-1); + let msg = err + .get("message") + .and_then(|m| m.as_str()) + .unwrap_or("unknown"); + return Err(ChannelError::SendFailed { + name: "signal".to_string(), + reason: format!("Signal RPC error {code}: {msg}"), + }); + } + + Ok(parsed.get("result").cloned()) + } + + /// Build JSON-RPC params for a send/typing call. + fn build_rpc_params( + &self, + target: &RecipientTarget, + message: Option<&str>, + ) -> serde_json::Value { + match target { + RecipientTarget::Direct(id) => { + let mut params = serde_json::json!({ + "recipient": [id], + "account": &self.config.account, + }); + if let Some(msg) = message { + params["message"] = serde_json::Value::String(msg.to_string()); + } + params + } + RecipientTarget::Group(group_id) => { + let mut params = serde_json::json!({ + "groupId": group_id, + "account": &self.config.account, + }); + if let Some(msg) = message { + params["message"] = serde_json::Value::String(msg.to_string()); + } + params + } + } + } + + /// Build JSON-RPC params for a send/typing call (static version). + fn build_rpc_params_static( + _http_url: &str, + account: &str, + target: &RecipientTarget, + message: Option<&str>, + ) -> serde_json::Value { + match target { + RecipientTarget::Direct(id) => { + let mut params = serde_json::json!({ + "recipient": [id], + "account": account, + }); + if let Some(msg) = message { + params["message"] = serde_json::Value::String(msg.to_string()); + } + params + } + RecipientTarget::Group(group_id) => { + let mut params = serde_json::json!({ + "groupId": group_id, + "account": account, + }); + if let Some(msg) = message { + params["message"] = serde_json::Value::String(msg.to_string()); + } + params + } + } + } + + /// Process a single SSE envelope, returning an `IncomingMessage` if valid. + fn process_envelope(&self, envelope: &Envelope) -> Option<(IncomingMessage, String)> { + // Skip story messages when configured. + if self.config.ignore_stories && envelope.story_message.is_some() { + return None; + } + + let data_msg = envelope.data_message.as_ref()?; + + // Skip attachment-only messages when configured. + let has_attachments = data_msg.attachments.as_ref().is_some_and(|a| !a.is_empty()); + let has_message_text = data_msg.message.as_ref().is_some_and(|m| !m.is_empty()); + if self.config.ignore_attachments && has_attachments && !has_message_text { + return None; + } + + // Use message text, or fall back to "[Attachment]" for attachment-only messages + // when ignore_attachments is false. This ensures attachment-only messages are + // still processed when the user wants them (rather than always being dropped). + let text = data_msg + .message + .as_deref() + .filter(|t| !t.is_empty()) + .map(String::from) + .or_else(|| { + if has_attachments { + Some("[Attachment]".to_string()) + } else { + None + } + })?; + let sender = Self::sender(envelope)?; + + // Log sender info including UUID if available + tracing::debug!( + sender = %sender, + uuid = ?envelope.source_uuid, + "Signal: received message" + ); + + // Check if this is a group message + let is_group = data_msg + .group_info + .as_ref() + .and_then(|g| g.group_id.as_deref()) + .is_some(); + + // Apply group policy first (before DM policy for group messages) + if is_group { + match self.config.group_policy.as_str() { + "disabled" => { + tracing::debug!("Signal: group messages disabled, dropping"); + return None; + } + "open" => { + // For "open" policy, check group allowlist but not sender allowlist + if let Some(group_id) = data_msg + .group_info + .as_ref() + .and_then(|g| g.group_id.as_deref()) + && !self.is_group_allowed(group_id) + { + tracing::debug!( + group_id = %group_id, + "Signal: group not in allow_from_groups, dropping" + ); + return None; + } + } + "allowlist" => { + // Default to allowlist - check group AND sender + if let Some(group_id) = data_msg + .group_info + .as_ref() + .and_then(|g| g.group_id.as_deref()) + { + if !self.is_group_allowed(group_id) { + tracing::debug!( + group_id = %group_id, + "Signal: group not in allow_from_groups, dropping" + ); + return None; + } + // Also check sender is allowed for group + if !self.is_group_sender_allowed(&sender) { + tracing::debug!( + sender = %sender, + group_id = %group_id, + "Signal: sender not in group_allow_from, dropping" + ); + return None; + } + } + } + _ => {} + } + } else { + // DM message - apply DM policy + match self.config.dm_policy.as_str() { + "open" => {} + "pairing" => { + // Pairing policy: check allow_from + pairing store + if !self.is_sender_allowed_with_pairing(&sender) { + // Handle pairing request - this will create a request and send reply if new + match self.handle_pairing_request(&sender, envelope.source_name.as_deref()) + { + Ok(_) => { + // Pairing request processed (new or existing), drop the message + return None; + } + Err(()) => { + // Error processing pairing, drop message + return None; + } + } + } + } + "allowlist" => { + // Default: check allow_from list + if !self.is_sender_allowed(&sender) { + tracing::debug!(sender = %sender, "Signal: sender not in allow_from, dropping"); + return None; + } + } + _ => {} + } + } + + let target = Self::reply_target(data_msg, &sender); + + let timestamp = data_msg + .timestamp + .or(envelope.timestamp) + .unwrap_or_else(|| { + u64::try_from( + std::time::SystemTime::now() + .duration_since(std::time::UNIX_EPOCH) + .unwrap_or_default() + .as_millis(), + ) + .unwrap_or(u64::MAX) + }); + + // Build metadata with signal-specific routing info. + let metadata = serde_json::json!({ + "signal_sender": &sender, + "signal_target": &target, + "signal_timestamp": timestamp, + }); + + let mut msg = IncomingMessage::new("signal", &sender, text).with_metadata(metadata); + + // Use sourceName as display name if available. + if let Some(ref name) = envelope.source_name + && !name.is_empty() + { + msg = msg.with_user_name(name); + } + + // Use a deterministic UUID as thread_id for all conversations. + // This ensures DMs and groups continue the same thread AND work with + // maybe_hydrate_thread, enabling conversation history persistence. + // Priority: source_uuid > generated UUID from phone/group + if data_msg.group_info.is_some() { + // For groups, use the group ID to generate a deterministic UUID + msg = msg.with_thread(Self::thread_id_from_identifier(&target)); + } else if let Some(ref uuid) = envelope.source_uuid { + // Privacy mode users already have a UUID + msg = msg.with_thread(uuid.clone()); + } else { + // For regular DMs, generate a deterministic UUID from the phone number + msg = msg.with_thread(Self::thread_id_from_identifier(&sender)); + } + + Some((msg, target)) + } +} + +#[async_trait] +impl Channel for SignalChannel { + fn name(&self) -> &str { + "signal" + } + + async fn start(&self) -> Result { + let (tx, rx) = tokio::sync::mpsc::channel(256); + + let config = self.config.clone(); + let client = self.client.clone(); + let reply_targets = Arc::clone(&self.reply_targets); + + tokio::spawn(async move { + if let Err(e) = sse_listener(config, client, tx, reply_targets).await { + tracing::error!("Signal SSE listener exited with error: {e}"); + } + }); + + // Log the URL with credentials redacted (if any). + let safe_url = Self::redact_url(&self.config.http_url); + tracing::info!( + url = %safe_url, + "Signal channel started" + ); + + Ok(Box::pin(tokio_stream::wrappers::ReceiverStream::new(rx))) + } + + async fn respond( + &self, + msg: &IncomingMessage, + response: OutgoingResponse, + ) -> Result<(), ChannelError> { + // Resolve reply target from stored metadata. + let target_str = { + let targets = self.reply_targets.read().await; + targets.peek(&msg.id).cloned() + } + .or_else(|| { + // Fall back to metadata if not in the map. + msg.metadata + .get("signal_target") + .and_then(|v| v.as_str()) + .map(String::from) + }) + .unwrap_or_else(|| msg.user_id.clone()); + + let target = Self::parse_recipient_target(&target_str); + let params = self.build_rpc_params(&target, Some(&response.content)); + self.rpc_request("send", params).await?; + + // Clean up stored target. + self.reply_targets.write().await.pop(&msg.id); + + Ok(()) + } + + async fn send_status( + &self, + status: StatusUpdate, + metadata: &serde_json::Value, + ) -> Result<(), ChannelError> { + // Send typing indicator for thinking status. + if matches!(status, StatusUpdate::Thinking(_)) + && let Some(target_str) = metadata.get("signal_target").and_then(|v| v.as_str()) + { + let target = Self::parse_recipient_target(target_str); + let params = self.build_rpc_params(&target, None); + let _ = self.rpc_request("sendTyping", params).await; + } + Ok(()) + } + + async fn broadcast( + &self, + user_id: &str, + response: OutgoingResponse, + ) -> Result<(), ChannelError> { + let target = Self::parse_recipient_target(user_id); + let params = self.build_rpc_params(&target, Some(&response.content)); + self.rpc_request("send", params).await?; + Ok(()) + } + + async fn health_check(&self) -> Result<(), ChannelError> { + let url = format!("{}{}", self.config.http_url, SIGNAL_HEALTH_ENDPOINT); + let resp = self + .client + .get(&url) + .timeout(Duration::from_secs(10)) + .send() + .await + .map_err(|e| ChannelError::HealthCheckFailed { + name: format!("signal ({}): {e}", Self::redact_url(&url)), + })?; + + if resp.status().is_success() { + Ok(()) + } else { + Err(ChannelError::HealthCheckFailed { + name: format!("signal: HTTP {}", resp.status()), + }) + } + } +} + +/// Long-running SSE listener that reconnects with exponential backoff. +async fn sse_listener( + config: SignalConfig, + client: Client, + tx: tokio::sync::mpsc::Sender, + reply_targets: Arc>>, +) -> Result<(), ChannelError> { + let channel = SignalChannel::from_parts(config, client, Arc::clone(&reply_targets)); + + let mut url = reqwest::Url::parse(&format!("{}/api/v1/events", channel.config.http_url)) + .map_err(|e| ChannelError::StartupFailed { + name: "signal".to_string(), + reason: format!("Invalid SSE URL: {e}"), + })?; + url.query_pairs_mut() + .append_pair("account", &channel.config.account); + + let mut retry_delay = Duration::from_secs(2); + let max_delay = Duration::from_secs(60); + + loop { + let resp = channel + .client + .get(url.clone()) + .header("Accept", "text/event-stream") + .send() + .await; + + let resp = match resp { + Ok(r) if r.status().is_success() => r, + Ok(r) => { + let status = r.status(); + let mut stream = r.bytes_stream(); + let mut bytes = Vec::new(); + let mut collected = 0usize; + while let Some(chunk) = stream.next().await { + let chunk = chunk.unwrap_or_default(); + let remaining = MAX_ERROR_LOG_BODY.saturating_sub(collected); + if remaining == 0 { + break; + } + bytes.extend_from_slice(&chunk[..chunk.len().min(remaining)]); + collected = bytes.len(); + if collected >= MAX_ERROR_LOG_BODY { + break; + } + } + let body = String::from_utf8_lossy(&bytes); + tracing::warn!("Signal SSE returned {status}: {body}"); + tokio::time::sleep(retry_delay).await; + retry_delay = (retry_delay * 2).min(max_delay); + continue; + } + Err(e) => { + let safe_url = SignalChannel::redact_url(url.as_str()); + tracing::warn!("Signal SSE connect error to {safe_url}: {e}, retrying..."); + tokio::time::sleep(retry_delay).await; + retry_delay = (retry_delay * 2).min(max_delay); + continue; + } + }; + + // Connection succeeded — reset backoff. + retry_delay = Duration::from_secs(2); + tracing::info!("Signal SSE connected"); + + let mut bytes_stream = resp.bytes_stream(); + let mut buffer = String::with_capacity(8192); + let mut current_data = String::with_capacity(4096); + // Holds trailing bytes from the previous chunk that form an incomplete + // multi-byte UTF-8 sequence. At most 3 bytes (the longest incomplete + // leading sequence for a 4-byte character). + let mut utf8_carry: Vec = Vec::with_capacity(4); + + while let Some(chunk) = bytes_stream.next().await { + let chunk = match chunk { + Ok(c) => c, + Err(e) => { + tracing::debug!("Signal SSE chunk error, reconnecting: {e}"); + break; + } + }; + + // Prepend any leftover bytes from the previous chunk. + let decode_buf = if utf8_carry.is_empty() { + chunk.to_vec() + } else { + let mut combined = std::mem::take(&mut utf8_carry); + combined.extend_from_slice(&chunk); + combined + }; + + // Decode as much valid UTF-8 as possible, carrying over any + // incomplete trailing sequence to the next iteration. + let (valid_len, carry_start) = match std::str::from_utf8(&decode_buf) { + Ok(_) => (decode_buf.len(), decode_buf.len()), + Err(e) => { + let valid_up_to = e.valid_up_to(); + match e.error_len() { + Some(bad_len) => { + // Genuinely invalid byte sequence (not just incomplete). + // Skip the bad byte(s) and keep going with what we have. + tracing::debug!( + "Signal SSE invalid UTF-8 byte at offset {valid_up_to}, \ + skipping" + ); + // Advance past the bad byte(s); remaining data (if any) + // will be carried over to the next chunk. + (valid_up_to, valid_up_to + bad_len) + } + None => { + // Incomplete multi-byte sequence at the end – carry it over. + (valid_up_to, valid_up_to) + } + } + } + }; + + use std::borrow::Cow; + + debug_assert!( + std::str::from_utf8(&decode_buf[..valid_len]).is_ok(), + "valid_len {} should be a valid UTF-8 boundary (buffer len: {})", + valid_len, + decode_buf.len() + ); + + let text: Cow = match std::str::from_utf8(&decode_buf[..valid_len]) { + Ok(s) => Cow::Borrowed(s), + Err(_) => { + tracing::warn!( + "Signal SSE: unexpected invalid UTF-8 boundary at valid_len {}, \ + falling back to lossy conversion", + valid_len + ); + Cow::Owned(String::from_utf8_lossy(&decode_buf[..valid_len]).into_owned()) + } + }; + + if buffer.len() + text.len() > MAX_SSE_BUFFER_SIZE { + tracing::warn!( + "Signal SSE buffer overflow, resetting: buffer_len={} text_len={} max={}", + buffer.len(), + text.len(), + MAX_SSE_BUFFER_SIZE + ); + buffer.clear(); + utf8_carry.clear(); + current_data.clear(); + continue; + } + buffer.push_str(&text); + + // Preserve any trailing incomplete bytes for the next chunk. + if carry_start < decode_buf.len() { + utf8_carry.extend_from_slice(&decode_buf[carry_start..]); + } + + while let Some(newline_pos) = buffer.find('\n') { + let line = buffer[..newline_pos].trim_end_matches('\r').to_string(); + buffer.drain(..=newline_pos); + + // Skip SSE comments (keepalive). + if line.starts_with(':') { + continue; + } + + if line.is_empty() { + // Empty line = event boundary, dispatch accumulated data. + if !current_data.is_empty() { + match serde_json::from_str::(¤t_data) { + Ok(sse) => { + if let Some(ref envelope) = sse.envelope + && let Some((msg, target)) = channel.process_envelope(envelope) + { + // Store reply target for respond(). + // LruCache automatically evicts the + // least-recently-used entry when full. + { + let mut targets = reply_targets.write().await; + targets.put(msg.id, target); + } + if tx.send(msg).await.is_err() { + tracing::debug!("Signal SSE: receiver dropped, exiting"); + return Ok(()); + } + } + } + Err(e) => { + tracing::debug!("Signal SSE parse skip: {e}"); + } + } + current_data.clear(); + } + } else if let Some(data) = line.strip_prefix("data:") { + if current_data.len() + data.len() > MAX_SSE_EVENT_SIZE { + tracing::warn!("Signal SSE event too large, dropping"); + current_data.clear(); + continue; + } + if !current_data.is_empty() { + current_data.push('\n'); + } + current_data.push_str(data.trim_start()); + } + // Ignore "event:", "id:", "retry:" lines. + } + } + + // Process any trailing data before reconnect. + if !current_data.is_empty() + && let Ok(sse) = serde_json::from_str::(¤t_data) + && let Some(ref envelope) = sse.envelope + && let Some((msg, target)) = channel.process_envelope(envelope) + { + reply_targets.write().await.put(msg.id, target); + let _ = tx.send(msg).await; + } + + tracing::debug!("Signal SSE stream ended, reconnecting with backoff..."); + tokio::time::sleep(retry_delay).await; + retry_delay = std::cmp::min(retry_delay * 2, max_delay); + } +} + +#[cfg(test)] +mod tests { + use super::*; + + fn make_config() -> SignalConfig { + SignalConfig { + http_url: "http://127.0.0.1:8686".to_string(), + account: "+1234567890".to_string(), + allow_from: vec!["+1111111111".to_string()], + allow_from_groups: vec![], + dm_policy: "allowlist".to_string(), + group_policy: "disabled".to_string(), + group_allow_from: vec![], + ignore_attachments: false, + ignore_stories: false, + } + } + + /// Create a config that allows a specific group (and all senders). + fn make_config_with_allowed_group(group_id: &str) -> SignalConfig { + SignalConfig { + http_url: "http://127.0.0.1:8686".to_string(), + account: "+1234567890".to_string(), + allow_from: vec!["*".to_string()], + allow_from_groups: vec![group_id.to_string()], + dm_policy: "allowlist".to_string(), + group_policy: "allowlist".to_string(), + group_allow_from: vec![], + ignore_attachments: true, + ignore_stories: true, + } + } + + fn make_channel() -> Result { + SignalChannel::new(make_config()) + } + + fn make_channel_with_allowed_group(group_id: &str) -> Result { + SignalChannel::new(make_config_with_allowed_group(group_id)) + } + + fn make_envelope(source_number: Option<&str>, message: Option<&str>) -> Envelope { + Envelope { + source: source_number.map(String::from), + source_number: source_number.map(String::from), + source_name: None, + source_uuid: None, + data_message: message.map(|m| DataMessage { + message: Some(m.to_string()), + timestamp: Some(1_700_000_000_000), + group_info: None, + attachments: None, + }), + story_message: None, + timestamp: Some(1_700_000_000_000), + } + } + + #[test] + fn creates_with_correct_fields() -> Result<(), ChannelError> { + let ch = make_channel()?; + assert_eq!(ch.config.http_url, "http://127.0.0.1:8686"); + assert_eq!(ch.config.account, "+1234567890"); + assert_eq!(ch.config.allow_from.len(), 1); + assert!(ch.config.allow_from_groups.is_empty()); + assert!(!ch.config.ignore_attachments); + assert!(!ch.config.ignore_stories); + Ok(()) + } + + #[test] + fn strips_trailing_slash() -> Result<(), ChannelError> { + let mut config = make_config(); + config.http_url = "http://127.0.0.1:8686/".to_string(); + let ch = SignalChannel::new(config)?; + assert_eq!(ch.config.http_url, "http://127.0.0.1:8686"); + Ok(()) + } + + #[test] + fn wildcard_allows_anyone() -> Result<(), ChannelError> { + let mut config = make_config(); + config.allow_from = vec!["*".to_string()]; + let ch = SignalChannel::new(config)?; + assert!(ch.is_sender_allowed("+9999999999")); + Ok(()) + } + + #[test] + fn specific_sender_allowed() -> Result<(), ChannelError> { + let ch = make_channel()?; + assert!(ch.is_sender_allowed("+1111111111")); + Ok(()) + } + + #[test] + fn unknown_sender_denied() -> Result<(), ChannelError> { + let ch = make_channel()?; + assert!(!ch.is_sender_allowed("+9999999999")); + Ok(()) + } + + #[test] + fn empty_allowlist_denies_all() -> Result<(), ChannelError> { + let mut config = make_config(); + config.allow_from = vec![]; + let ch = SignalChannel::new(config)?; + assert!(!ch.is_sender_allowed("+1111111111")); + Ok(()) + } + + #[test] + fn uuid_prefix_in_allowlist() -> Result<(), ChannelError> { + let uuid = "a1b2c3d4-e5f6-7890-abcd-ef1234567890"; + let mut config = make_config(); + config.allow_from = vec![format!("uuid:{uuid}")]; + let ch = SignalChannel::new(config)?; + assert!(ch.is_sender_allowed(uuid)); + // Should not match phone numbers. + assert!(!ch.is_sender_allowed("+1111111111")); + Ok(()) + } + + #[test] + fn bare_uuid_in_allowlist() -> Result<(), ChannelError> { + let uuid = "a1b2c3d4-e5f6-7890-abcd-ef1234567890"; + let mut config = make_config(); + config.allow_from = vec![uuid.to_string()]; + let ch = SignalChannel::new(config)?; + assert!(ch.is_sender_allowed(uuid)); + Ok(()) + } + + #[test] + fn group_allowlist_filtering() -> Result<(), ChannelError> { + let mut config = make_config(); + config.allow_from = vec!["*".to_string()]; + config.allow_from_groups = vec!["group123".to_string()]; + let ch = SignalChannel::new(config)?; + assert!(ch.is_group_allowed("group123")); + assert!(!ch.is_group_allowed("other_group")); + Ok(()) + } + + #[test] + fn group_allowlist_wildcard() -> Result<(), ChannelError> { + let mut config = make_config(); + config.allow_from_groups = vec!["*".to_string()]; + let ch = SignalChannel::new(config)?; + assert!(ch.is_group_allowed("any_group")); + Ok(()) + } + + #[test] + fn group_allowlist_empty_denies_all() -> Result<(), ChannelError> { + let mut config = make_config(); + config.allow_from_groups = vec![]; + let ch = SignalChannel::new(config)?; + assert!(!ch.is_group_allowed("any_group")); + Ok(()) + } + + #[test] + fn name_returns_signal() -> Result<(), ChannelError> { + let ch = make_channel()?; + assert_eq!(ch.name(), "signal"); + Ok(()) + } + + #[test] + fn process_envelope_dm_accepted_with_empty_allow_from_groups() -> Result<(), ChannelError> { + // Empty allow_from_groups = DMs only. DMs should be accepted. + let ch = make_channel()?; + let env = make_envelope(Some("+1111111111"), Some("Hello!")); + assert!(ch.process_envelope(&env).is_some()); + Ok(()) + } + + #[test] + fn process_envelope_group_denied_with_empty_allow_from_groups() -> Result<(), ChannelError> { + // Empty allow_from_groups = DMs only. Group messages should be denied. + let mut config = make_config(); + config.allow_from = vec!["*".to_string()]; + let ch = SignalChannel::new(config)?; + + let env = Envelope { + source: Some("+1111111111".to_string()), + source_number: Some("+1111111111".to_string()), + source_name: None, + source_uuid: None, + data_message: Some(DataMessage { + message: Some("hi".to_string()), + timestamp: Some(1000), + group_info: Some(GroupInfo { + group_id: Some("group123".to_string()), + }), + attachments: None, + }), + story_message: None, + timestamp: Some(1000), + }; + assert!(ch.process_envelope(&env).is_none()); + Ok(()) + } + + #[test] + fn process_envelope_group_accepted_when_in_allow_from_groups() -> Result<(), ChannelError> { + let ch = make_channel_with_allowed_group("group123")?; + + let env = Envelope { + source: Some("+1111111111".to_string()), + source_number: Some("+1111111111".to_string()), + source_name: None, + source_uuid: None, + data_message: Some(DataMessage { + message: Some("hi".to_string()), + timestamp: Some(1000), + group_info: Some(GroupInfo { + group_id: Some("group123".to_string()), + }), + attachments: None, + }), + story_message: None, + timestamp: Some(1000), + }; + assert!(ch.process_envelope(&env).is_some()); + + // Different group should be denied. + let env2 = Envelope { + source: Some("+1111111111".to_string()), + source_number: Some("+1111111111".to_string()), + source_name: None, + source_uuid: None, + data_message: Some(DataMessage { + message: Some("hi".to_string()), + timestamp: Some(1000), + group_info: Some(GroupInfo { + group_id: Some("other_group".to_string()), + }), + attachments: None, + }), + story_message: None, + timestamp: Some(1000), + }; + assert!(ch.process_envelope(&env2).is_none()); + Ok(()) + } + + #[test] + fn reply_target_dm() { + let dm = DataMessage { + message: Some("hi".to_string()), + timestamp: Some(1000), + group_info: None, + attachments: None, + }; + assert_eq!( + SignalChannel::reply_target(&dm, "+1111111111"), + "+1111111111" + ); + } + + #[test] + fn reply_target_group() { + let group = DataMessage { + message: Some("hi".to_string()), + timestamp: Some(1000), + group_info: Some(GroupInfo { + group_id: Some("group123".to_string()), + }), + attachments: None, + }; + assert_eq!( + SignalChannel::reply_target(&group, "+1111111111"), + "group:group123" + ); + } + + #[test] + fn parse_recipient_target_e164_is_direct() { + assert_eq!( + SignalChannel::parse_recipient_target("+1234567890"), + RecipientTarget::Direct("+1234567890".to_string()) + ); + } + + #[test] + fn parse_recipient_target_prefixed_group_is_group() { + assert_eq!( + SignalChannel::parse_recipient_target("group:abc123"), + RecipientTarget::Group("abc123".to_string()) + ); + } + + #[test] + fn parse_recipient_target_uuid_is_direct() { + let uuid = "a1b2c3d4-e5f6-7890-abcd-ef1234567890"; + assert_eq!( + SignalChannel::parse_recipient_target(uuid), + RecipientTarget::Direct(uuid.to_string()) + ); + } + + #[test] + fn parse_recipient_target_non_e164_plus_is_group() { + assert_eq!( + SignalChannel::parse_recipient_target("+abc123"), + RecipientTarget::Group("+abc123".to_string()) + ); + } + + #[test] + fn is_uuid_valid() { + assert!(SignalChannel::is_uuid( + "a1b2c3d4-e5f6-7890-abcd-ef1234567890" + )); + assert!(SignalChannel::is_uuid( + "00000000-0000-0000-0000-000000000000" + )); + } + + #[test] + fn is_uuid_invalid() { + assert!(!SignalChannel::is_uuid("+1234567890")); + assert!(!SignalChannel::is_uuid("not-a-uuid")); + assert!(!SignalChannel::is_uuid("group:abc123")); + assert!(!SignalChannel::is_uuid("")); + } + + #[test] + fn thread_id_from_identifier_is_deterministic() { + let id1 = SignalChannel::thread_id_from_identifier("+1234567890"); + let id2 = SignalChannel::thread_id_from_identifier("+1234567890"); + assert_eq!(id1, id2, "same input should produce same UUID"); + } + + #[test] + fn thread_id_from_identifier_is_valid_uuid() { + let id = SignalChannel::thread_id_from_identifier("+1234567890"); + assert!(Uuid::parse_str(&id).is_ok(), "should be a valid UUID"); + } + + #[test] + fn thread_id_from_identifier_different_inputs() { + let id1 = SignalChannel::thread_id_from_identifier("+1234567890"); + let id2 = SignalChannel::thread_id_from_identifier("+9876543210"); + assert_ne!(id1, id2, "different inputs should produce different UUIDs"); + } + + #[test] + fn sender_prefers_source_number() { + let env = Envelope { + source: Some("uuid-123".to_string()), + source_number: Some("+1111111111".to_string()), + source_name: None, + source_uuid: None, + data_message: None, + story_message: None, + timestamp: Some(1000), + }; + assert_eq!(SignalChannel::sender(&env), Some("+1111111111".to_string())); + } + + #[test] + fn sender_falls_back_to_source() { + let env = Envelope { + source: Some("a1b2c3d4-e5f6-7890-abcd-ef1234567890".to_string()), + source_number: None, + source_name: None, + source_uuid: None, + data_message: None, + story_message: None, + timestamp: Some(1000), + }; + assert_eq!( + SignalChannel::sender(&env), + Some("a1b2c3d4-e5f6-7890-abcd-ef1234567890".to_string()) + ); + } + + #[test] + fn sender_none_when_both_missing() { + let env = Envelope { + source: None, + source_number: None, + source_name: None, + source_uuid: None, + data_message: None, + story_message: None, + timestamp: None, + }; + assert_eq!(SignalChannel::sender(&env), None); + } + + #[test] + fn process_envelope_valid_dm() -> Result<(), ChannelError> { + let ch = make_channel()?; + let env = make_envelope(Some("+1111111111"), Some("Hello!")); + let (msg, target) = ch.process_envelope(&env).unwrap(); + assert_eq!(msg.content, "Hello!"); + assert_eq!(msg.user_id, "+1111111111"); + assert_eq!(msg.channel, "signal"); + assert_eq!(target, "+1111111111"); + Ok(()) + } + + #[test] + fn process_envelope_denied_sender() -> Result<(), ChannelError> { + let ch = make_channel()?; + let env = make_envelope(Some("+9999999999"), Some("Hello!")); + assert!(ch.process_envelope(&env).is_none()); + Ok(()) + } + + #[test] + fn process_envelope_empty_message() -> Result<(), ChannelError> { + let ch = make_channel()?; + let env = make_envelope(Some("+1111111111"), Some("")); + assert!(ch.process_envelope(&env).is_none()); + Ok(()) + } + + #[test] + fn process_envelope_no_data_message() -> Result<(), ChannelError> { + let ch = make_channel()?; + let env = make_envelope(Some("+1111111111"), None); + assert!(ch.process_envelope(&env).is_none()); + Ok(()) + } + + #[test] + fn process_envelope_skips_stories() -> Result<(), ChannelError> { + let mut config = make_config(); + config.allow_from = vec!["*".to_string()]; + config.ignore_stories = true; + let ch = SignalChannel::new(config)?; + let mut env = make_envelope(Some("+1111111111"), Some("story text")); + env.story_message = Some(serde_json::json!({})); + assert!(ch.process_envelope(&env).is_none()); + Ok(()) + } + + #[test] + fn process_envelope_skips_attachment_only() -> Result<(), ChannelError> { + let mut config = make_config(); + config.allow_from = vec!["*".to_string()]; + config.ignore_attachments = true; + let ch = SignalChannel::new(config)?; + let env = Envelope { + source: Some("+1111111111".to_string()), + source_number: Some("+1111111111".to_string()), + source_name: None, + source_uuid: None, + data_message: Some(DataMessage { + message: None, + timestamp: Some(1_700_000_000_000), + group_info: None, + attachments: Some(vec![serde_json::json!({"contentType": "image/png"})]), + }), + story_message: None, + timestamp: Some(1_700_000_000_000), + }; + assert!(ch.process_envelope(&env).is_none()); + Ok(()) + } + + #[test] + fn process_envelope_uuid_sender_dm() -> Result<(), ChannelError> { + let uuid = "a1b2c3d4-e5f6-7890-abcd-ef1234567890"; + let mut config = make_config(); + config.allow_from = vec!["*".to_string()]; + let ch = SignalChannel::new(config)?; + + let env = Envelope { + source: Some(uuid.to_string()), + source_number: None, + source_name: Some("Privacy User".to_string()), + source_uuid: None, + data_message: Some(DataMessage { + message: Some("Hello from privacy user".to_string()), + timestamp: Some(1_700_000_000_000), + group_info: None, + attachments: None, + }), + story_message: None, + timestamp: Some(1_700_000_000_000), + }; + let (msg, target) = ch.process_envelope(&env).unwrap(); + assert_eq!(msg.user_id, uuid); + assert_eq!(msg.user_name.as_deref(), Some("Privacy User")); + assert_eq!(msg.content, "Hello from privacy user"); + assert_eq!(target, uuid); + + // Verify reply routing: UUID sender in DM should route as Direct. + let parsed = SignalChannel::parse_recipient_target(&target); + assert_eq!(parsed, RecipientTarget::Direct(uuid.to_string())); + Ok(()) + } + + #[test] + fn process_envelope_uuid_sender_in_group() -> Result<(), ChannelError> { + let uuid = "a1b2c3d4-e5f6-7890-abcd-ef1234567890"; + let mut config = make_config_with_allowed_group("testgroup"); + config.ignore_attachments = false; + config.ignore_stories = false; + let ch = SignalChannel::new(config)?; + + let env = Envelope { + source: Some(uuid.to_string()), + source_number: None, + source_name: None, + source_uuid: None, + data_message: Some(DataMessage { + message: Some("Group msg from privacy user".to_string()), + timestamp: Some(1_700_000_000_000), + group_info: Some(GroupInfo { + group_id: Some("testgroup".to_string()), + }), + attachments: None, + }), + story_message: None, + timestamp: Some(1_700_000_000_000), + }; + let (msg, target) = ch.process_envelope(&env).unwrap(); + assert_eq!(msg.user_id, uuid); + assert_eq!(target, "group:testgroup"); + // Groups now use deterministic UUID derived from group ID + let expected_thread_id = SignalChannel::thread_id_from_identifier("group:testgroup"); + assert_eq!(msg.thread_id, Some(expected_thread_id)); + + // Verify reply routing: group message should still route as Group. + let parsed = SignalChannel::parse_recipient_target(&target); + assert_eq!(parsed, RecipientTarget::Group("testgroup".to_string())); + Ok(()) + } + + #[test] + fn process_envelope_group_not_in_allow_from_groups() -> Result<(), ChannelError> { + let mut config = make_config(); + config.allow_from = vec!["*".to_string()]; + config.allow_from_groups = vec!["allowed_group".to_string()]; + let ch = SignalChannel::new(config)?; + + let env = Envelope { + source: Some("+1111111111".to_string()), + source_number: Some("+1111111111".to_string()), + source_name: None, + source_uuid: None, + data_message: Some(DataMessage { + message: Some("Hi".to_string()), + timestamp: Some(1_700_000_000_000), + group_info: Some(GroupInfo { + group_id: Some("other_group".to_string()), + }), + attachments: None, + }), + story_message: None, + timestamp: Some(1_700_000_000_000), + }; + assert!(ch.process_envelope(&env).is_none()); + Ok(()) + } + + #[test] + fn sse_envelope_deserializes() { + let json = r#"{ + "envelope": { + "source": "+1111111111", + "sourceNumber": "+1111111111", + "sourceName": "Test User", + "timestamp": 1700000000000, + "dataMessage": { + "message": "Hello Signal!", + "timestamp": 1700000000000 + } + } + }"#; + let sse: SseEnvelope = serde_json::from_str(json).unwrap(); + let env = sse.envelope.unwrap(); + assert_eq!(env.source_number.as_deref(), Some("+1111111111")); + assert_eq!(env.source_name.as_deref(), Some("Test User")); + let dm = env.data_message.unwrap(); + assert_eq!(dm.message.as_deref(), Some("Hello Signal!")); + } + + #[test] + fn sse_envelope_deserializes_group() { + let json = r#"{ + "envelope": { + "sourceNumber": "+2222222222", + "dataMessage": { + "message": "Group msg", + "groupInfo": { + "groupId": "abc123" + } + } + } + }"#; + let sse: SseEnvelope = serde_json::from_str(json).unwrap(); + let env = sse.envelope.unwrap(); + let dm = env.data_message.unwrap(); + assert_eq!( + dm.group_info.as_ref().unwrap().group_id.as_deref(), + Some("abc123") + ); + } + + #[test] + fn envelope_defaults() { + let json = r#"{}"#; + let env: Envelope = serde_json::from_str(json).unwrap(); + assert!(env.source.is_none()); + assert!(env.source_number.is_none()); + assert!(env.source_name.is_none()); + assert!(env.data_message.is_none()); + assert!(env.story_message.is_none()); + assert!(env.timestamp.is_none()); + } + + #[test] + fn normalize_allow_entry_strips_uuid_prefix() { + assert_eq!( + SignalChannel::normalize_allow_entry("uuid:abc-123"), + "abc-123" + ); + assert_eq!( + SignalChannel::normalize_allow_entry("+1234567890"), + "+1234567890" + ); + assert_eq!(SignalChannel::normalize_allow_entry("*"), "*"); + } + + // ── build_rpc_params tests ────────────────────────────────────── + + #[test] + fn build_rpc_params_direct_with_message() -> Result<(), ChannelError> { + let ch = make_channel()?; + let target = RecipientTarget::Direct("+5555555555".to_string()); + let params = ch.build_rpc_params(&target, Some("Hello!")); + assert_eq!(params["recipient"], serde_json::json!(["+5555555555"])); + assert_eq!(params["account"], "+1234567890"); + assert_eq!(params["message"], "Hello!"); + // Direct targets must NOT include groupId. + assert!(params.get("groupId").is_none()); + Ok(()) + } + + #[test] + fn build_rpc_params_direct_without_message() -> Result<(), ChannelError> { + let ch = make_channel()?; + let target = RecipientTarget::Direct("+5555555555".to_string()); + let params = ch.build_rpc_params(&target, None); + assert_eq!(params["recipient"], serde_json::json!(["+5555555555"])); + assert_eq!(params["account"], "+1234567890"); + // No message key should be present for typing indicators. + assert!(params.get("message").is_none()); + Ok(()) + } + + #[test] + fn build_rpc_params_group_with_message() -> Result<(), ChannelError> { + let ch = make_channel()?; + let target = RecipientTarget::Group("abc123".to_string()); + let params = ch.build_rpc_params(&target, Some("Group msg")); + assert_eq!(params["groupId"], "abc123"); + assert_eq!(params["account"], "+1234567890"); + assert_eq!(params["message"], "Group msg"); + // Group targets must NOT include recipient. + assert!(params.get("recipient").is_none()); + Ok(()) + } + + #[test] + fn build_rpc_params_group_without_message() -> Result<(), ChannelError> { + let ch = make_channel()?; + let target = RecipientTarget::Group("abc123".to_string()); + let params = ch.build_rpc_params(&target, None); + assert_eq!(params["groupId"], "abc123"); + assert_eq!(params["account"], "+1234567890"); + assert!(params.get("message").is_none()); + Ok(()) + } + + #[test] + fn build_rpc_params_uuid_direct_target() -> Result<(), ChannelError> { + let ch = make_channel()?; + let uuid = "a1b2c3d4-e5f6-7890-abcd-ef1234567890"; + let target = RecipientTarget::Direct(uuid.to_string()); + let params = ch.build_rpc_params(&target, Some("hi")); + assert_eq!(params["recipient"], serde_json::json!([uuid])); + Ok(()) + } + + // ── metadata assertion tests ──────────────────────────────────── + + #[test] + fn process_envelope_metadata_has_signal_fields() -> Result<(), ChannelError> { + let ch = make_channel()?; + let env = make_envelope(Some("+1111111111"), Some("Hello!")); + let (msg, _) = ch.process_envelope(&env).unwrap(); + assert_eq!(msg.metadata["signal_sender"], "+1111111111"); + assert_eq!(msg.metadata["signal_target"], "+1111111111"); + assert_eq!(msg.metadata["signal_timestamp"], 1_700_000_000_000_u64); + Ok(()) + } + + #[test] + fn process_envelope_metadata_group_target() -> Result<(), ChannelError> { + let mut config = make_config(); + config.allow_from = vec!["*".to_string()]; + config.allow_from_groups = vec!["*".to_string()]; + config.group_policy = "allowlist".to_string(); + let ch = SignalChannel::new(config)?; + + let env = Envelope { + source: Some("+2222222222".to_string()), + source_number: Some("+2222222222".to_string()), + source_name: None, + source_uuid: None, + data_message: Some(DataMessage { + message: Some("In the group".to_string()), + timestamp: Some(1_700_000_000_000), + group_info: Some(GroupInfo { + group_id: Some("mygroup".to_string()), + }), + attachments: None, + }), + story_message: None, + timestamp: Some(1_700_000_000_000), + }; + let (msg, _) = ch.process_envelope(&env).unwrap(); + assert_eq!(msg.metadata["signal_target"], "group:mygroup"); + assert_eq!(msg.metadata["signal_sender"], "+2222222222"); + Ok(()) + } + + // ── attachment-with-text tests ────────────────────────────────── + + #[test] + fn process_envelope_attachment_with_text_not_skipped() -> Result<(), ChannelError> { + // Even with ignore_attachments=true, messages that have BOTH text + // and attachments should be processed (only attachment-only are skipped). + let mut config = make_config(); + config.allow_from = vec!["*".to_string()]; + config.ignore_attachments = true; + let ch = SignalChannel::new(config)?; + + let env = Envelope { + source: Some("+1111111111".to_string()), + source_number: Some("+1111111111".to_string()), + source_name: None, + source_uuid: None, + data_message: Some(DataMessage { + message: Some("Check this out".to_string()), + timestamp: Some(1_700_000_000_000), + group_info: None, + attachments: Some(vec![serde_json::json!({"contentType": "image/png"})]), + }), + story_message: None, + timestamp: Some(1_700_000_000_000), + }; + let result = ch.process_envelope(&env); + assert!( + result.is_some(), + "Message with text + attachment should not be skipped" + ); + let (msg, _) = result.unwrap(); + assert_eq!(msg.content, "Check this out"); + Ok(()) + } + + #[test] + fn process_envelope_attachment_only_not_skipped_when_ignore_disabled() + -> Result<(), ChannelError> { + // With ignore_attachments=false, attachment-only messages should be + // processed with the "[Attachment]" placeholder text. + let mut config = make_config(); + config.allow_from = vec!["*".to_string()]; + config.ignore_attachments = false; + let ch = SignalChannel::new(config)?; + + let env = Envelope { + source: Some("+1111111111".to_string()), + source_number: Some("+1111111111".to_string()), + source_name: None, + source_uuid: None, + data_message: Some(DataMessage { + message: None, + timestamp: Some(1_700_000_000_000), + group_info: None, + attachments: Some(vec![serde_json::json!({"contentType": "image/png"})]), + }), + story_message: None, + timestamp: Some(1_700_000_000_000), + }; + // With ignore_attachments=false, attachment-only messages are now + // processed with a placeholder "[Attachment]" text. + let result = ch.process_envelope(&env); + assert!( + result.is_some(), + "Attachment-only should be processed when ignore_attachments=false" + ); + let (msg, _) = result.unwrap(); + assert_eq!(msg.content, "[Attachment]"); + Ok(()) + } + + // ── source_name / display name tests ──────────────────────────── + + #[test] + fn process_envelope_source_name_sets_user_name() -> Result<(), ChannelError> { + let mut config = make_config(); + config.allow_from = vec!["*".to_string()]; + let ch = SignalChannel::new(config)?; + + let env = Envelope { + source: Some("+3333333333".to_string()), + source_number: Some("+3333333333".to_string()), + source_name: Some("Alice".to_string()), + source_uuid: None, + data_message: Some(DataMessage { + message: Some("Hey".to_string()), + timestamp: Some(1_700_000_000_000), + group_info: None, + attachments: None, + }), + story_message: None, + timestamp: Some(1_700_000_000_000), + }; + let (msg, _) = ch.process_envelope(&env).unwrap(); + assert_eq!(msg.user_name.as_deref(), Some("Alice")); + Ok(()) + } + + #[test] + fn process_envelope_empty_source_name_not_set() -> Result<(), ChannelError> { + let mut config = make_config(); + config.allow_from = vec!["*".to_string()]; + let ch = SignalChannel::new(config)?; + + let env = Envelope { + source: Some("+3333333333".to_string()), + source_number: Some("+3333333333".to_string()), + source_name: Some("".to_string()), + source_uuid: None, + data_message: Some(DataMessage { + message: Some("Hey".to_string()), + timestamp: Some(1_700_000_000_000), + group_info: None, + attachments: None, + }), + story_message: None, + timestamp: Some(1_700_000_000_000), + }; + let (msg, _) = ch.process_envelope(&env).unwrap(); + assert!( + msg.user_name.is_none(), + "Empty source_name should not set user_name" + ); + Ok(()) + } + + #[test] + fn process_envelope_no_source_name_not_set() -> Result<(), ChannelError> { + let ch = make_channel()?; + let env = make_envelope(Some("+1111111111"), Some("hi")); + let (msg, _) = ch.process_envelope(&env).unwrap(); + assert!(msg.user_name.is_none()); + Ok(()) + } + + // ── thread_id tests ───────────────────────────────────────────────────────────────── + + #[test] + fn process_envelope_dm_sets_thread_id_to_uuid() -> Result<(), ChannelError> { + let ch = make_channel()?; + let env = make_envelope(Some("+1111111111"), Some("DM")); + let (msg, _) = ch.process_envelope(&env).unwrap(); + // DMs now set thread_id to a deterministic UUID derived from phone number + let expected_thread_id = SignalChannel::thread_id_from_identifier("+1111111111"); + assert_eq!( + msg.thread_id, + Some(expected_thread_id), + "DMs should set thread_id to UUID" + ); + Ok(()) + } + + #[test] + fn process_envelope_group_sets_thread_id_to_uuid() -> Result<(), ChannelError> { + let mut config = make_config(); + config.allow_from = vec!["*".to_string()]; + config.allow_from_groups = vec!["*".to_string()]; + config.group_policy = "allowlist".to_string(); + let ch = SignalChannel::new(config)?; + + let env = Envelope { + source: Some("+1111111111".to_string()), + source_number: Some("+1111111111".to_string()), + source_name: None, + source_uuid: None, + data_message: Some(DataMessage { + message: Some("Group msg".to_string()), + timestamp: Some(1_700_000_000_000), + group_info: Some(GroupInfo { + group_id: Some("grp999".to_string()), + }), + attachments: None, + }), + story_message: None, + timestamp: Some(1_700_000_000_000), + }; + let (msg, _) = ch.process_envelope(&env).unwrap(); + // Groups now set thread_id to a deterministic UUID derived from group ID + let expected_thread_id = SignalChannel::thread_id_from_identifier("group:grp999"); + assert_eq!( + msg.thread_id, + Some(expected_thread_id), + "Groups should set thread_id to UUID" + ); + Ok(()) + } + + // ── timestamp edge cases ──────────────────────────────────────── + + #[test] + fn process_envelope_uses_data_message_timestamp() -> Result<(), ChannelError> { + let mut config = make_config(); + config.allow_from = vec!["*".to_string()]; + let ch = SignalChannel::new(config)?; + + let env = Envelope { + source: Some("+1111111111".to_string()), + source_number: Some("+1111111111".to_string()), + source_name: None, + source_uuid: None, + data_message: Some(DataMessage { + message: Some("hi".to_string()), + timestamp: Some(9999), + group_info: None, + attachments: None, + }), + story_message: None, + timestamp: Some(1111), + }; + let (msg, _) = ch.process_envelope(&env).unwrap(); + // data_message timestamp takes priority. + assert_eq!(msg.metadata["signal_timestamp"], 9999); + Ok(()) + } + + #[test] + fn process_envelope_falls_back_to_envelope_timestamp() -> Result<(), ChannelError> { + let mut config = make_config(); + config.allow_from = vec!["*".to_string()]; + let ch = SignalChannel::new(config)?; + + let env = Envelope { + source: Some("+1111111111".to_string()), + source_number: Some("+1111111111".to_string()), + source_name: None, + source_uuid: None, + data_message: Some(DataMessage { + message: Some("hi".to_string()), + timestamp: None, + group_info: None, + attachments: None, + }), + story_message: None, + timestamp: Some(7777), + }; + let (msg, _) = ch.process_envelope(&env).unwrap(); + assert_eq!(msg.metadata["signal_timestamp"], 7777); + Ok(()) + } + + #[test] + fn process_envelope_generates_timestamp_when_missing() -> Result<(), ChannelError> { + let mut config = make_config(); + config.allow_from = vec!["*".to_string()]; + let ch = SignalChannel::new(config)?; + + let env = Envelope { + source: Some("+1111111111".to_string()), + source_number: Some("+1111111111".to_string()), + source_name: None, + source_uuid: None, + data_message: Some(DataMessage { + message: Some("hi".to_string()), + timestamp: None, + group_info: None, + attachments: None, + }), + story_message: None, + timestamp: None, + }; + let (msg, _) = ch.process_envelope(&env).unwrap(); + // Should generate a timestamp (current time in millis), just verify it's positive. + let ts = msg.metadata["signal_timestamp"].as_u64().unwrap(); + assert!(ts > 0, "Generated timestamp should be positive"); + Ok(()) + } + + // ── SSE envelope deserialization edge cases ───────────────────── + + #[test] + fn sse_envelope_missing_envelope_field() { + let json = r#"{"account": "+1234567890"}"#; + let sse: SseEnvelope = serde_json::from_str(json).unwrap(); + assert!(sse.envelope.is_none()); + } + + #[test] + fn sse_envelope_with_story_message() { + let json = r#"{ + "envelope": { + "sourceNumber": "+1111111111", + "storyMessage": {"allowsReplies": true}, + "dataMessage": { + "message": "story text" + } + } + }"#; + let sse: SseEnvelope = serde_json::from_str(json).unwrap(); + let env = sse.envelope.unwrap(); + assert!(env.story_message.is_some()); + assert!(env.data_message.is_some()); + } + + #[test] + fn sse_envelope_with_attachments() { + let json = r#"{ + "envelope": { + "sourceNumber": "+1111111111", + "dataMessage": { + "message": "See attached", + "attachments": [ + {"contentType": "image/jpeg", "filename": "photo.jpg"}, + {"contentType": "application/pdf"} + ] + } + } + }"#; + let sse: SseEnvelope = serde_json::from_str(json).unwrap(); + let dm = sse.envelope.unwrap().data_message.unwrap(); + let attachments = dm.attachments.unwrap(); + assert_eq!(attachments.len(), 2); + } + + // ── is_e164 tests ─────────────────────────────────────────────── + + #[test] + fn is_e164_valid_numbers() { + assert!(SignalChannel::is_e164("+12345678901")); + assert!(SignalChannel::is_e164("+1234567")); // min 7 digits after + + assert!(SignalChannel::is_e164("+123456789012345")); // max 15 digits + } + + #[test] + fn is_e164_invalid_numbers() { + assert!(!SignalChannel::is_e164("12345678901")); // no + + assert!(!SignalChannel::is_e164("+1")); // too short (1 digit) + assert!(!SignalChannel::is_e164("+1234567890123456")); // too long (16 digits) + assert!(!SignalChannel::is_e164("+abc123")); // non-digit + assert!(!SignalChannel::is_e164("")); // empty + assert!(!SignalChannel::is_e164("+")); // plus only + } + + // ── config edge cases ─────────────────────────────────────────── + + #[test] + fn multiple_allow_from() -> Result<(), ChannelError> { + let mut config = make_config(); + config.allow_from = vec![ + "+1111111111".to_string(), + "+2222222222".to_string(), + "a1b2c3d4-e5f6-7890-abcd-ef1234567890".to_string(), + ]; + let ch = SignalChannel::new(config)?; + assert!(ch.is_sender_allowed("+1111111111")); + assert!(ch.is_sender_allowed("+2222222222")); + assert!(ch.is_sender_allowed("a1b2c3d4-e5f6-7890-abcd-ef1234567890")); + assert!(!ch.is_sender_allowed("+9999999999")); + Ok(()) + } + + #[test] + fn multiple_allow_from_groups() -> Result<(), ChannelError> { + let mut config = make_config(); + config.allow_from_groups = vec!["group_a".to_string(), "group_b".to_string()]; + let ch = SignalChannel::new(config)?; + assert!(ch.is_group_allowed("group_a")); + assert!(ch.is_group_allowed("group_b")); + assert!(!ch.is_group_allowed("group_c")); + Ok(()) + } + + #[test] + fn uuid_prefix_normalization_in_allowlist() -> Result<(), ChannelError> { + let uuid = "a1b2c3d4-e5f6-7890-abcd-ef1234567890"; + let mut config = make_config(); + config.allow_from = vec![format!("uuid:{uuid}"), "+1111111111".to_string()]; + let ch = SignalChannel::new(config)?; + // uuid:-prefixed entry should match bare UUID sender. + assert!(ch.is_sender_allowed(uuid)); + // Phone numbers still work alongside UUID entries. + assert!(ch.is_sender_allowed("+1111111111")); + // Non-matching should fail. + assert!(!ch.is_sender_allowed("+9999999999")); + Ok(()) + } + + // ── stories behavior tests ────────────────────────────────────── + + #[test] + fn process_envelope_stories_not_skipped_when_disabled() -> Result<(), ChannelError> { + // With ignore_stories=false, story messages with a data_message + // should still be processed. + let mut config = make_config(); + config.allow_from = vec!["*".to_string()]; + config.ignore_stories = false; + let ch = SignalChannel::new(config)?; + + let env = Envelope { + source: Some("+1111111111".to_string()), + source_number: Some("+1111111111".to_string()), + source_name: None, + source_uuid: None, + data_message: Some(DataMessage { + message: Some("story with text".to_string()), + timestamp: Some(1_700_000_000_000), + group_info: None, + attachments: None, + }), + story_message: Some(serde_json::json!({})), + timestamp: Some(1_700_000_000_000), + }; + let result = ch.process_envelope(&env); + assert!( + result.is_some(), + "Stories should not be skipped when ignore_stories=false" + ); + Ok(()) + } + + // ── trailing slash variations ─────────────────────────────────── + + #[test] + fn strips_multiple_trailing_slashes() -> Result<(), ChannelError> { + let mut config = make_config(); + config.http_url = "http://127.0.0.1:8686///".to_string(); + let ch = SignalChannel::new(config)?; + assert_eq!(ch.config.http_url, "http://127.0.0.1:8686"); + Ok(()) + } + + #[test] + fn preserves_url_without_trailing_slash() -> Result<(), ChannelError> { + let config = make_config(); + let ch = SignalChannel::new(config)?; + assert_eq!(ch.config.http_url, "http://127.0.0.1:8686"); + Ok(()) + } +} diff --git a/src/config/channels.rs b/src/config/channels.rs index ccfdecf3..31e4e42f 100644 --- a/src/config/channels.rs +++ b/src/config/channels.rs @@ -12,6 +12,7 @@ pub struct ChannelsConfig { pub cli: CliConfig, pub http: Option, pub gateway: Option, + pub signal: Option, /// Directory containing WASM channel modules (default: ~/.ironclaw/channels/). pub wasm_channels_dir: std::path::PathBuf, /// Whether WASM channels are enabled. @@ -43,6 +44,49 @@ pub struct GatewayConfig { pub user_id: String, } +/// Signal channel configuration (signal-cli daemon HTTP/JSON-RPC). +#[derive(Debug, Clone)] +pub struct SignalConfig { + /// Base URL of the signal-cli daemon HTTP endpoint (e.g. `http://127.0.0.1:8080`). + pub http_url: String, + /// Signal account identifier (E.164 phone number, e.g. `+1234567890`). + pub account: String, + /// Users allowed to interact with the bot in DMs. + /// + /// Each entry is one of: + /// - `*` — allow everyone + /// - E.164 phone number (e.g. `+1234567890`) + /// - bare UUID (e.g. `a1b2c3d4-e5f6-7890-abcd-ef1234567890`) + /// - `uuid:` prefix form (e.g. `uuid:a1b2c3d4-e5f6-7890-abcd-ef1234567890`) + /// + /// An empty list denies all senders (secure by default). + pub allow_from: Vec, + /// Groups allowed to interact with the bot. + /// + /// - Empty list — deny all group messages (DMs only, secure by default). + /// - `*` — allow all groups. + /// - Specific group IDs — allow only those groups. + pub allow_from_groups: Vec, + /// DM policy: "open", "allowlist", or "pairing". Default: "pairing". + /// + /// - "open" — allow all DM senders (ignores allow_from for DMs) + /// - "allowlist" — only allow senders in allow_from list + /// - "pairing" — allowlist + send pairing reply to unknown users + pub dm_policy: String, + /// Group policy: "allowlist", "open", or "disabled". Default: "allowlist". + /// + /// - "disabled" — deny all group messages + /// - "allowlist" — check allow_from_groups and group_allow_from + /// - "open" — accept all group messages (respects allow_from_groups for group ID) + pub group_policy: String, + /// Allow list for group message senders. If empty, inherits from allow_from. + pub group_allow_from: Vec, + /// Skip messages that contain only attachments (no text). + pub ignore_attachments: bool, + /// Skip story messages. + pub ignore_stories: bool, +} + impl ChannelsConfig { pub(crate) fn resolve(settings: &Settings) -> Result { let http = if optional_env("HTTP_PORT")?.is_some() || optional_env("HTTP_HOST")?.is_some() { @@ -68,6 +112,58 @@ impl ChannelsConfig { None }; + let signal = if let Some(http_url) = optional_env("SIGNAL_HTTP_URL")? { + let account = optional_env("SIGNAL_ACCOUNT")?.ok_or(ConfigError::InvalidValue { + key: "SIGNAL_ACCOUNT".to_string(), + message: "SIGNAL_ACCOUNT is required when SIGNAL_HTTP_URL is set".to_string(), + })?; + let allow_from = match std::env::var_os("SIGNAL_ALLOW_FROM") { + None => vec![account.clone()], + Some(val) => { + let s = val.to_string_lossy(); + s.split(',') + .map(|e| e.trim().to_string()) + .filter(|s| !s.is_empty()) + .collect() + } + }; + let dm_policy = + optional_env("SIGNAL_DM_POLICY")?.unwrap_or_else(|| "pairing".to_string()); + let group_policy = + optional_env("SIGNAL_GROUP_POLICY")?.unwrap_or_else(|| "allowlist".to_string()); + Some(SignalConfig { + http_url, + account, + allow_from, + allow_from_groups: optional_env("SIGNAL_ALLOW_FROM_GROUPS")? + .map(|s| { + s.split(',') + .map(|e| e.trim().to_string()) + .filter(|s| !s.is_empty()) + .collect() + }) + .unwrap_or_default(), + dm_policy, + group_policy, + group_allow_from: optional_env("SIGNAL_GROUP_ALLOW_FROM")? + .map(|s| { + s.split(',') + .map(|e| e.trim().to_string()) + .filter(|s| !s.is_empty()) + .collect() + }) + .unwrap_or_default(), + ignore_attachments: optional_env("SIGNAL_IGNORE_ATTACHMENTS")? + .map(|s| s.to_lowercase() == "true" || s == "1") + .unwrap_or(false), + ignore_stories: optional_env("SIGNAL_IGNORE_STORIES")? + .map(|s| s.to_lowercase() == "true" || s == "1") + .unwrap_or(true), + }) + } else { + None + }; + let cli_enabled = optional_env("CLI_ENABLED")? .map(|s| s.to_lowercase() != "false" && s != "0") .unwrap_or(true); @@ -78,6 +174,7 @@ impl ChannelsConfig { }, http, gateway, + signal, wasm_channels_dir: optional_env("WASM_CHANNELS_DIR")? .map(PathBuf::from) .unwrap_or_else(default_channels_dir), diff --git a/src/config/mod.rs b/src/config/mod.rs index 9326b682..a15dc505 100644 --- a/src/config/mod.rs +++ b/src/config/mod.rs @@ -31,7 +31,7 @@ use crate::settings::Settings; // Re-export all public types so `crate::config::FooConfig` continues to work. pub use self::agent::AgentConfig; pub use self::builder::BuilderModeConfig; -pub use self::channels::{ChannelsConfig, CliConfig, GatewayConfig, HttpConfig}; +pub use self::channels::{ChannelsConfig, CliConfig, GatewayConfig, HttpConfig, SignalConfig}; pub use self::database::{DatabaseBackend, DatabaseConfig, default_libsql_path}; pub use self::embeddings::EmbeddingsConfig; pub use self::heartbeat::HeartbeatConfig; diff --git a/src/extensions/registry.rs b/src/extensions/registry.rs index 95b96bd1..ef6e46e6 100644 --- a/src/extensions/registry.rs +++ b/src/extensions/registry.rs @@ -103,7 +103,7 @@ impl ExtensionRegistry { } } - scored.sort_by(|a, b| b.1.cmp(&a.1)); + scored.sort_by_key(|b| std::cmp::Reverse(b.1)); scored.into_iter().map(|(r, _)| r).collect() } diff --git a/src/main.rs b/src/main.rs index 0e9a3b48..3743fea1 100644 --- a/src/main.rs +++ b/src/main.rs @@ -9,7 +9,7 @@ use ironclaw::{ agent::{Agent, AgentDeps}, app::{AppBuilder, AppBuilderFlags}, channels::{ - ChannelManager, GatewayChannel, HttpChannel, ReplChannel, WebhookServer, + ChannelManager, GatewayChannel, HttpChannel, ReplChannel, SignalChannel, WebhookServer, WebhookServerConfig, wasm::{ RegisteredEndpoint, SharedWasmChannel, WasmChannelLoader, WasmChannelRouter, @@ -368,6 +368,25 @@ async fn main() -> anyhow::Result<()> { } } + // Add Signal channel if configured and not CLI-only mode. + if !cli.cli_only + && let Some(ref signal_config) = config.channels.signal + { + let signal_channel = SignalChannel::new(signal_config.clone())?; + channel_names.push("signal".to_string()); + channels.add(Box::new(signal_channel)).await; + let safe_url = SignalChannel::redact_url(&signal_config.http_url); + tracing::info!( + url = %safe_url, + "Signal channel enabled" + ); + if signal_config.allow_from.is_empty() { + tracing::warn!( + "Signal channel has empty allow_from list - ALL messages will be DENIED." + ); + } + } + // Add HTTP channel if configured and not CLI-only mode. let mut webhook_server_addr: Option = None; if !cli.cli_only diff --git a/src/pairing/mod.rs b/src/pairing/mod.rs index 6468524f..c35a5a70 100644 --- a/src/pairing/mod.rs +++ b/src/pairing/mod.rs @@ -7,4 +7,4 @@ mod store; -pub use store::{PairingRequest, PairingStore, PairingStoreError}; +pub use store::{PairingRequest, PairingStore, PairingStoreError, UpsertResult}; diff --git a/src/safety/sanitizer.rs b/src/safety/sanitizer.rs index 60ab9901..605db896 100644 --- a/src/safety/sanitizer.rs +++ b/src/safety/sanitizer.rs @@ -225,7 +225,7 @@ impl Sanitizer { } // Sort warnings by severity (critical first) - warnings.sort_by(|a, b| b.severity.cmp(&a.severity)); + warnings.sort_by_key(|b| std::cmp::Reverse(b.severity)); // Determine if we need to modify content let has_critical = warnings.iter().any(|w| w.severity == Severity::Critical); diff --git a/src/settings.rs b/src/settings.rs index 3ca28b3f..a6a0cac0 100644 --- a/src/settings.rs +++ b/src/settings.rs @@ -212,6 +212,41 @@ pub struct ChannelSettings { #[serde(default)] pub http_host: Option, + /// Whether Signal channel is enabled. + #[serde(default)] + pub signal_enabled: bool, + + /// Signal HTTP URL (signal-cli daemon endpoint). + #[serde(default)] + pub signal_http_url: Option, + + /// Signal account (E.164 phone number). + #[serde(default)] + pub signal_account: Option, + + /// Signal allow from list for DMs (comma-separated E.164 phone numbers). + /// Comma-separated identifiers: E.164 phone numbers, `*`, bare UUIDs, or `uuid:` entries. + /// Defaults to the configured account. + #[serde(default)] + pub signal_allow_from: Option, + + /// Signal allow from groups (comma-separated group IDs). + #[serde(default)] + pub signal_allow_from_groups: Option, + + /// Signal DM policy: "open", "allowlist", or "pairing". Default: "pairing". + #[serde(default)] + pub signal_dm_policy: Option, + + /// Signal group policy: "allowlist", "open", or "disabled". Default: "allowlist". + #[serde(default)] + pub signal_group_policy: Option, + + /// Signal group allow from (comma-separated group member IDs). + /// If empty, inherits from signal_allow_from. + #[serde(default)] + pub signal_group_allow_from: Option, + /// Telegram owner user ID. When set, the bot only responds to this user. /// Captured during setup by having the user message the bot. #[serde(default)] diff --git a/src/setup/channels.rs b/src/setup/channels.rs index 36cc7049..aafe5817 100644 --- a/src/setup/channels.rs +++ b/src/setup/channels.rs @@ -11,6 +11,8 @@ use std::sync::Arc; use reqwest::Client; use secrecy::{ExposeSecret, SecretString}; use serde::Deserialize; +use url::Url; +use uuid::Uuid; #[cfg(feature = "postgres")] use crate::secrets::SecretsCrypto; @@ -639,6 +641,19 @@ pub struct HttpSetupResult { pub host: String, } +/// Result of Signal channel setup. +#[derive(Debug, Clone)] +pub struct SignalSetupResult { + pub enabled: bool, + pub http_url: String, + pub account: String, + pub allow_from: String, + pub allow_from_groups: String, + pub dm_policy: String, + pub group_policy: String, + pub group_allow_from: String, +} + /// Set up HTTP webhook channel. pub async fn setup_http(secrets: &SecretsContext) -> Result { println!("HTTP Webhook Setup:"); @@ -684,6 +699,188 @@ pub fn generate_webhook_secret() -> String { generate_secret_with_length(32) } +fn validate_e164(account: &str) -> Result<(), String> { + if !account.starts_with('+') { + return Err("E.164 account must start with '+'".to_string()); + } + let digits = &account[1..]; + if digits.is_empty() { + return Err("E.164 account must have digits after '+'".to_string()); + } + if !digits.chars().all(|c| c.is_ascii_digit()) { + return Err("E.164 account must contain only digits after '+'".to_string()); + } + if digits.len() < 7 || digits.len() > 15 { + return Err("E.164 account must be 7-15 digits after '+'".to_string()); + } + Ok(()) +} + +fn validate_allow_from_list(list: &str) -> Result<(), String> { + if list.is_empty() { + return Ok(()); + } + for (i, item) in list.split(',').enumerate() { + let trimmed = item.trim(); + if trimmed.is_empty() { + continue; + } + if trimmed == "*" { + continue; + } + if let Some(uuid_part) = trimmed.strip_prefix("uuid:") { + if Uuid::parse_str(uuid_part).is_err() { + return Err(format!( + "allow_from[{}]: '{}' is not a valid UUID (after 'uuid:' prefix)", + i, trimmed + )); + } + continue; + } + if validate_e164(trimmed).is_ok() { + continue; + } + if Uuid::parse_str(trimmed).is_ok() { + continue; + } + return Err(format!( + "allow_from[{}]: '{}' must be '*', E.164 phone number, UUID, or 'uuid:'", + i, trimmed + )); + } + Ok(()) +} + +fn validate_allow_from_groups_list(list: &str) -> Result<(), String> { + if list.is_empty() { + return Ok(()); + } + for (i, item) in list.split(',').enumerate() { + let trimmed = item.trim(); + if trimmed.is_empty() { + continue; + } + if trimmed == "*" { + continue; + } + if trimmed.is_empty() { + return Err(format!( + "allow_from_groups[{}]: group ID cannot be empty", + i + )); + } + } + Ok(()) +} + +/// Set up Signal channel. +/// `Settings` is reserved for future use +pub async fn setup_signal(_settings: &Settings) -> Result { + println!("Signal Channel Setup:"); + println!(); + print_info("Signal channel connects to a signal-cli daemon running in HTTP mode."); + println!(); + + let http_url = input("Signal-cli HTTP URL")?; + match Url::parse(&http_url) { + Ok(url) if url.scheme() == "http" || url.scheme() == "https" => {} + Ok(_) => { + print_error("URL must use http or https scheme"); + return Err(ChannelSetupError::Validation( + "Invalid HTTP URL: must use http or https scheme".to_string(), + )); + } + Err(e) => { + print_error(&format!("Invalid URL: {}", e)); + return Err(ChannelSetupError::Validation(format!( + "Invalid HTTP URL: {}", + e + ))); + } + } + + let account = input("Signal account (E.164)")?; + if let Err(e) = validate_e164(&account) { + print_error(&e); + return Err(ChannelSetupError::Validation(e)); + } + + let allow_from = optional_input( + "Allow from (comma-separated: E.164 numbers, '*' for anyone, UUIDs or 'uuid:'; empty for self-only)", + Some(&format!("default: {} (self-only)", account)), + )? + .unwrap_or_else(|| account.clone()); + + let dm_policy = optional_input( + "DM policy (open, allowlist, pairing)", + Some("default: pairing"), + )? + .unwrap_or_else(|| "pairing".to_string()); + + let allow_from_groups = optional_input( + "Allow from groups (comma-separated group IDs, '*' for any group; empty for none)", + Some("default: (none)"), + )? + .unwrap_or_default(); + + let group_policy = optional_input( + "Group policy (allowlist, open, disabled)", + Some("default: allowlist"), + )? + .unwrap_or_else(|| "allowlist".to_string()); + + let group_allow_from = optional_input( + "Group allow from (comma-separated member IDs; empty to inherit from allow_from)", + Some("default: (inherit from allow_from)"), + )? + .unwrap_or_default(); + + if let Err(e) = validate_allow_from_list(&allow_from) { + print_error(&e); + return Err(ChannelSetupError::Validation(e)); + } + + if let Err(e) = validate_allow_from_groups_list(&allow_from_groups) { + print_error(&e); + return Err(ChannelSetupError::Validation(e)); + } + + println!(); + print_success(&format!( + "Signal channel configured for account: {}", + account + )); + print_info(&format!("HTTP URL: {}", http_url)); + if allow_from == account { + print_info("Allow from: self-only"); + } else { + print_info(&format!("Allow from: {}", allow_from)); + } + print_info(&format!("DM policy: {}", dm_policy)); + if allow_from_groups.is_empty() { + print_info("Allow from groups: (none)"); + } else { + print_info(&format!("Allow from groups: {}", allow_from_groups)); + } + print_info(&format!("Group policy: {}", group_policy)); + if group_allow_from.is_empty() { + print_info("Group allow from: (inherits from allow_from)"); + } else { + print_info(&format!("Group allow from: {}", group_allow_from)); + } + + Ok(SignalSetupResult { + enabled: true, + http_url, + account, + allow_from, + allow_from_groups, + dm_policy, + group_policy, + group_allow_from, + }) +} + /// Result of WASM channel setup. #[derive(Debug, Clone)] pub struct WasmChannelSetupResult { diff --git a/src/setup/wizard.rs b/src/setup/wizard.rs index ed548299..fd745161 100644 --- a/src/setup/wizard.rs +++ b/src/setup/wizard.rs @@ -27,13 +27,18 @@ use crate::llm::{SessionConfig, SessionManager}; use crate::secrets::{SecretsCrypto, SecretsStore}; use crate::settings::{KeySource, Settings}; use crate::setup::channels::{ - SecretsContext, setup_http, setup_telegram, setup_tunnel, setup_wasm_channel, + SecretsContext, setup_http, setup_signal, setup_telegram, setup_tunnel, setup_wasm_channel, }; use crate::setup::prompts::{ confirm, input, optional_input, print_error, print_header, print_info, print_step, print_success, secret_input, select_many, select_one, }; +// unused const, keep commented for clarity / future use +// const CHANNEL_INDEX_CLI: usize = 0; +const CHANNEL_INDEX_HTTP: usize = 1; +const CHANNEL_INDEX_SIGNAL: usize = 2; + /// Setup wizard error. #[derive(Debug, thiserror::Error)] pub enum SetupError { @@ -1443,8 +1448,11 @@ impl SetupWizard { "HTTP webhook".to_string(), self.settings.channels.http_enabled, ), + ("Signal".to_string(), self.settings.channels.signal_enabled), ]; + let non_wasm_count = options.len(); + // Add available WASM channels (installed + bundled + registry) for name in &wasm_channel_names { let is_enabled = self.settings.channels.wasm_channels.contains(name); @@ -1466,7 +1474,7 @@ impl SetupWizard { .iter() .enumerate() .filter_map(|(idx, name)| { - if selected.contains(&(idx + 2)) { + if selected.contains(&(non_wasm_count + idx)) { Some(name.clone()) } else { None @@ -1514,7 +1522,8 @@ impl SetupWizard { } // Determine if we need secrets context - let needs_secrets = selected.contains(&1) || !selected_wasm_channels.is_empty(); + let needs_secrets = + selected.contains(&CHANNEL_INDEX_HTTP) || !selected_wasm_channels.is_empty(); let secrets = if needs_secrets { match self.init_secrets_context().await { Ok(ctx) => Some(ctx), @@ -1528,8 +1537,8 @@ impl SetupWizard { None }; - // HTTP is index 1 - if selected.contains(&1) { + // HTTP channel + if selected.contains(&CHANNEL_INDEX_HTTP) { println!(); if let Some(ref ctx) = secrets { let result = setup_http(ctx).await?; @@ -1544,6 +1553,29 @@ impl SetupWizard { self.settings.channels.http_enabled = false; } + // Signal channel + if selected.contains(&CHANNEL_INDEX_SIGNAL) { + println!(); + let result = setup_signal(&self.settings).await?; + self.settings.channels.signal_enabled = result.enabled; + self.settings.channels.signal_http_url = Some(result.http_url); + self.settings.channels.signal_account = Some(result.account); + self.settings.channels.signal_allow_from = Some(result.allow_from); + self.settings.channels.signal_allow_from_groups = Some(result.allow_from_groups); + self.settings.channels.signal_dm_policy = Some(result.dm_policy); + self.settings.channels.signal_group_policy = Some(result.group_policy); + self.settings.channels.signal_group_allow_from = Some(result.group_allow_from); + } else { + self.settings.channels.signal_enabled = false; + self.settings.channels.signal_http_url = None; + self.settings.channels.signal_account = None; + self.settings.channels.signal_allow_from = None; + self.settings.channels.signal_allow_from_groups = None; + self.settings.channels.signal_dm_policy = None; + self.settings.channels.signal_group_policy = None; + self.settings.channels.signal_group_allow_from = None; + } + let discovered_by_name: HashMap = discovered_channels.into_iter().collect(); @@ -1939,6 +1971,33 @@ impl SetupWizard { env_vars.push(("ONBOARD_COMPLETED", "true".to_string())); } + // Signal channel env vars (chicken-and-egg: config resolves before DB). + if let Some(ref url) = self.settings.channels.signal_http_url { + env_vars.push(("SIGNAL_HTTP_URL", url.clone())); + } + if let Some(ref account) = self.settings.channels.signal_account { + env_vars.push(("SIGNAL_ACCOUNT", account.clone())); + } + if let Some(ref allow_from) = self.settings.channels.signal_allow_from { + env_vars.push(("SIGNAL_ALLOW_FROM", allow_from.clone())); + } + if let Some(ref allow_from_groups) = self.settings.channels.signal_allow_from_groups + && !allow_from_groups.is_empty() + { + env_vars.push(("SIGNAL_ALLOW_FROM_GROUPS", allow_from_groups.clone())); + } + if let Some(ref dm_policy) = self.settings.channels.signal_dm_policy { + env_vars.push(("SIGNAL_DM_POLICY", dm_policy.clone())); + } + if let Some(ref group_policy) = self.settings.channels.signal_group_policy { + env_vars.push(("SIGNAL_GROUP_POLICY", group_policy.clone())); + } + if let Some(ref group_allow_from) = self.settings.channels.signal_group_allow_from + && !group_allow_from.is_empty() + { + env_vars.push(("SIGNAL_GROUP_ALLOW_FROM", group_allow_from.clone())); + } + if !env_vars.is_empty() { let pairs: Vec<(&str, &str)> = env_vars.iter().map(|(k, v)| (*k, v.as_str())).collect(); crate::bootstrap::save_bootstrap_env(&pairs).map_err(|e| { diff --git a/src/skills/selector.rs b/src/skills/selector.rs index 060d0caf..f9a78aa9 100644 --- a/src/skills/selector.rs +++ b/src/skills/selector.rs @@ -62,7 +62,7 @@ pub fn prefilter_skills<'a>( .collect(); // Sort by score descending - scored.sort_by(|a, b| b.score.cmp(&a.score)); + scored.sort_by_key(|b| std::cmp::Reverse(b.score)); // Apply candidate limit and context budget let mut result = Vec::new();