From a4d50174bd6530ce724c447df4e04e017d2247a7 Mon Sep 17 00:00:00 2001 From: Steven Enamakel Date: Sun, 5 Jul 2026 03:31:42 -0700 Subject: [PATCH 1/9] channels: consume ported providers from tinychannels Replace the in-tree implementations of the portable channel providers with thin re-exports from the tinychannels crate (submodule bumped): dingtalk, discord, email_channel, imessage, irc, lark, linq, mattermost, qq, yuanbao Each provider file now re-exports its module from `tinychannels::providers::

` (glob, to preserve the internal submodule/test-seam surface that raw-coverage e2e tests reach into). The per-provider `*_tests.rs` files and the discord/ and yuanbao/ subdirectories move into tinychannels. - startup.rs / commands.rs: providers whose reqwest client was previously proxied internally (discord, dingtalk, mattermost, qq) now build the proxy client at the call site via `build_runtime_proxy_client("channel.")` and pass it to `with_http_client(...)`, preserving proxy behavior. - One lark WS raw-coverage test now builds its tungstenite Message via `tinychannels::tokio_tungstenite` so the type matches across the 0.24/0.29 version boundary. The 4 host-coupled providers (telegram, web, presentation, whatsapp_web) remain in-tree: they depend on the agent harness, event bus, voice, and approval surfaces, which are not part of the tinychannels boundary. Core crate, Tauri shell, and the affected channels integration tests all build and pass. Claude-Session: https://claude.ai/code/session_013NzEbXoeB3YzsbmpjVUTPW --- Cargo.lock | 19 + app/src-tauri/Cargo.lock | 19 + src/openhuman/channels/commands.rs | 9 +- src/openhuman/channels/providers/dingtalk.rs | 385 +------ src/openhuman/channels/providers/discord.rs | 1 + .../channels/providers/discord/api.rs | 511 --------- .../channels/providers/discord/api_tests.rs | 451 -------- .../channels/providers/discord/channel.rs | 525 --------- .../providers/discord/channel_tests.rs | 581 ---------- .../channels/providers/discord/mod.rs | 4 - .../channels/providers/email_channel.rs | 554 +--------- .../channels/providers/email_channel_tests.rs | 535 ---------- src/openhuman/channels/providers/imessage.rs | 319 +----- .../channels/providers/imessage_tests.rs | 672 ------------ src/openhuman/channels/providers/irc.rs | 657 +----------- src/openhuman/channels/providers/irc_tests.rs | 406 ------- src/openhuman/channels/providers/lark.rs | 993 +----------------- .../channels/providers/lark_tests.rs | 613 ----------- src/openhuman/channels/providers/linq.rs | 398 +------ .../channels/providers/linq_tests.rs | 390 ------- .../channels/providers/mattermost.rs | 491 +-------- .../channels/providers/mattermost_tests.rs | 462 -------- src/openhuman/channels/providers/qq.rs | 452 +------- src/openhuman/channels/providers/qq_tests.rs | 84 -- src/openhuman/channels/providers/yuanbao.rs | 1 + .../channels/providers/yuanbao/channel.rs | 710 ------------- .../channels/providers/yuanbao/config.rs | 79 -- .../channels/providers/yuanbao/connection.rs | 903 ---------------- .../channels/providers/yuanbao/cos.rs | 567 ---------- .../channels/providers/yuanbao/errors.rs | 61 -- .../channels/providers/yuanbao/ids.rs | 115 -- .../channels/providers/yuanbao/inbound.rs | 623 ----------- .../channels/providers/yuanbao/media.rs | 619 ----------- .../channels/providers/yuanbao/mod.rs | 29 - .../channels/providers/yuanbao/outbound.rs | 449 -------- .../channels/providers/yuanbao/proto.rs | 916 ---------------- .../channels/providers/yuanbao/proto_biz.rs | 638 ----------- .../providers/yuanbao/proto_constants.rs | 90 -- .../channels/providers/yuanbao/sign.rs | 629 ----------- .../channels/providers/yuanbao/splitter.rs | 209 ---- .../channels/providers/yuanbao/types.rs | 441 -------- .../channels/providers/yuanbao/wire.rs | 397 ------- src/openhuman/channels/runtime/startup.rs | 12 +- ...email_dispatch_round21_raw_coverage_e2e.rs | 4 +- vendor/tinychannels | 2 +- 45 files changed, 66 insertions(+), 16959 deletions(-) create mode 100644 src/openhuman/channels/providers/discord.rs delete mode 100644 src/openhuman/channels/providers/discord/api.rs delete mode 100644 src/openhuman/channels/providers/discord/api_tests.rs delete mode 100644 src/openhuman/channels/providers/discord/channel.rs delete mode 100644 src/openhuman/channels/providers/discord/channel_tests.rs delete mode 100644 src/openhuman/channels/providers/discord/mod.rs delete mode 100644 src/openhuman/channels/providers/email_channel_tests.rs delete mode 100644 src/openhuman/channels/providers/imessage_tests.rs delete mode 100644 src/openhuman/channels/providers/irc_tests.rs delete mode 100644 src/openhuman/channels/providers/lark_tests.rs delete mode 100644 src/openhuman/channels/providers/linq_tests.rs delete mode 100644 src/openhuman/channels/providers/mattermost_tests.rs delete mode 100644 src/openhuman/channels/providers/qq_tests.rs create mode 100644 src/openhuman/channels/providers/yuanbao.rs delete mode 100644 src/openhuman/channels/providers/yuanbao/channel.rs delete mode 100644 src/openhuman/channels/providers/yuanbao/config.rs delete mode 100644 src/openhuman/channels/providers/yuanbao/connection.rs delete mode 100644 src/openhuman/channels/providers/yuanbao/cos.rs delete mode 100644 src/openhuman/channels/providers/yuanbao/errors.rs delete mode 100644 src/openhuman/channels/providers/yuanbao/ids.rs delete mode 100644 src/openhuman/channels/providers/yuanbao/inbound.rs delete mode 100644 src/openhuman/channels/providers/yuanbao/media.rs delete mode 100644 src/openhuman/channels/providers/yuanbao/mod.rs delete mode 100644 src/openhuman/channels/providers/yuanbao/outbound.rs delete mode 100644 src/openhuman/channels/providers/yuanbao/proto.rs delete mode 100644 src/openhuman/channels/providers/yuanbao/proto_biz.rs delete mode 100644 src/openhuman/channels/providers/yuanbao/proto_constants.rs delete mode 100644 src/openhuman/channels/providers/yuanbao/sign.rs delete mode 100644 src/openhuman/channels/providers/yuanbao/splitter.rs delete mode 100644 src/openhuman/channels/providers/yuanbao/types.rs delete mode 100644 src/openhuman/channels/providers/yuanbao/wire.rs diff --git a/Cargo.lock b/Cargo.lock index b44a50c403..0e375ffd86 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -6830,20 +6830,39 @@ name = "tinychannels" version = "0.1.0" dependencies = [ "anyhow", + "async-imap", "async-trait", + "axum", "base64 0.22.1", + "chrono", + "directories", + "futures", "futures-util", + "hex", "hmac 0.12.1", + "lettre", + "mail-parser", + "parking_lot", + "prost", + "rand 0.10.1", "reqwest 0.12.28", + "rusqlite", + "rustls", + "rustls-pki-types", "schemars", "serde", "serde_json", + "sha1", "sha2 0.10.9", "thiserror 2.0.18", "tokio", + "tokio-rustls", "tokio-tungstenite 0.29.0", "tracing", + "url", + "urlencoding", "uuid 1.23.1", + "webpki-roots 1.0.7", ] [[package]] diff --git a/app/src-tauri/Cargo.lock b/app/src-tauri/Cargo.lock index 7422ebe595..7666dacd88 100644 --- a/app/src-tauri/Cargo.lock +++ b/app/src-tauri/Cargo.lock @@ -9179,20 +9179,39 @@ name = "tinychannels" version = "0.1.0" dependencies = [ "anyhow", + "async-imap", "async-trait", + "axum", "base64 0.22.1", + "chrono", + "directories 6.0.0", + "futures", "futures-util", + "hex", "hmac 0.12.1", + "lettre", + "mail-parser", + "parking_lot", + "prost", + "rand 0.10.1", "reqwest 0.12.28", + "rusqlite", + "rustls", + "rustls-pki-types", "schemars 1.2.1", "serde", "serde_json", + "sha1", "sha2 0.10.9", "thiserror 2.0.18", "tokio", + "tokio-rustls", "tokio-tungstenite 0.29.0", "tracing", + "url", + "urlencoding", "uuid 1.23.1", + "webpki-roots 1.0.7", ] [[package]] diff --git a/src/openhuman/channels/commands.rs b/src/openhuman/channels/commands.rs index 41edc4ece9..a28dee4250 100644 --- a/src/openhuman/channels/commands.rs +++ b/src/openhuman/channels/commands.rs @@ -64,13 +64,14 @@ pub async fn doctor_channels(config: Config) -> Result<()> { if let Some(ref dc) = config.channels_config.discord { channels.push(( "Discord", - Arc::new(DiscordChannel::new( + Arc::new(DiscordChannel::with_http_client( dc.bot_token.clone(), dc.guild_id.clone(), dc.channel_id.clone(), dc.allowed_users.clone(), dc.listen_to_bots, dc.mention_only, + crate::openhuman::config::build_runtime_proxy_client("channel.discord"), )), )); } @@ -209,10 +210,11 @@ pub async fn doctor_channels(config: Config) -> Result<()> { if let Some(ref dt) = config.channels_config.dingtalk { channels.push(( "DingTalk", - Arc::new(DingTalkChannel::new( + Arc::new(DingTalkChannel::with_http_client( dt.client_id.clone(), dt.client_secret.clone(), dt.allowed_users.clone(), + crate::openhuman::config::build_runtime_proxy_client("channel.dingtalk"), )), )); } @@ -220,10 +222,11 @@ pub async fn doctor_channels(config: Config) -> Result<()> { if let Some(ref qq) = config.channels_config.qq { channels.push(( "QQ", - Arc::new(QQChannel::new( + Arc::new(QQChannel::with_http_client( qq.app_id.clone(), qq.app_secret.clone(), qq.allowed_users.clone(), + crate::openhuman::config::build_runtime_proxy_client("channel.qq"), )), )); } diff --git a/src/openhuman/channels/providers/dingtalk.rs b/src/openhuman/channels/providers/dingtalk.rs index 8b935414c2..432fafddb5 100644 --- a/src/openhuman/channels/providers/dingtalk.rs +++ b/src/openhuman/channels/providers/dingtalk.rs @@ -1,384 +1 @@ -use crate::openhuman::channels::traits::{Channel, ChannelMessage, SendMessage}; -use async_trait::async_trait; -use futures_util::{SinkExt, StreamExt}; -use std::collections::HashMap; -use std::sync::Arc; -use tokio::sync::RwLock; -use tokio_tungstenite::tungstenite::Message; -use uuid::Uuid; - -const DINGTALK_BOT_CALLBACK_TOPIC: &str = "/v1.0/im/bot/messages/get"; - -/// DingTalk channel — connects via Stream Mode WebSocket for real-time messages. -/// Replies are sent through per-message session webhook URLs. -pub struct DingTalkChannel { - client_id: String, - client_secret: String, - allowed_users: Vec, - /// Per-chat session webhooks for sending replies (chatID -> webhook URL). - /// DingTalk provides a unique webhook URL with each incoming message. - session_webhooks: Arc>>, -} - -/// Response from DingTalk gateway connection registration. -#[derive(serde::Deserialize)] -struct GatewayResponse { - endpoint: String, - ticket: String, -} - -impl DingTalkChannel { - pub fn new(client_id: String, client_secret: String, allowed_users: Vec) -> Self { - Self { - client_id, - client_secret, - allowed_users, - session_webhooks: Arc::new(RwLock::new(HashMap::new())), - } - } - - fn http_client(&self) -> reqwest::Client { - crate::openhuman::config::build_runtime_proxy_client("channel.dingtalk") - } - - fn is_user_allowed(&self, user_id: &str) -> bool { - self.allowed_users.iter().any(|u| u == "*" || u == user_id) - } - - fn parse_stream_data(frame: &serde_json::Value) -> Option { - match frame.get("data") { - Some(serde_json::Value::String(raw)) => serde_json::from_str(raw).ok(), - Some(serde_json::Value::Object(_)) => frame.get("data").cloned(), - _ => None, - } - } - - fn resolve_chat_id(data: &serde_json::Value, sender_id: &str) -> String { - let is_private_chat = data - .get("conversationType") - .and_then(|value| { - value - .as_str() - .map(|v| v == "1") - .or_else(|| value.as_i64().map(|v| v == 1)) - }) - .unwrap_or(true); - - if is_private_chat { - sender_id.to_string() - } else { - data.get("conversationId") - .and_then(|c| c.as_str()) - .unwrap_or(sender_id) - .to_string() - } - } - - /// Register a connection with DingTalk's gateway to get a WebSocket endpoint. - async fn register_connection(&self) -> anyhow::Result { - let body = serde_json::json!({ - "clientId": self.client_id, - "clientSecret": self.client_secret, - "subscriptions": [ - { - "type": "CALLBACK", - "topic": DINGTALK_BOT_CALLBACK_TOPIC, - } - ], - }); - - let resp = self - .http_client() - .post("https://api.dingtalk.com/v1.0/gateway/connections/open") - .json(&body) - .send() - .await?; - - if !resp.status().is_success() { - let status = resp.status(); - let err = resp.text().await.unwrap_or_default(); - anyhow::bail!("DingTalk gateway registration failed ({status}): {err}"); - } - - let gw: GatewayResponse = resp.json().await?; - Ok(gw) - } -} - -#[async_trait] -impl Channel for DingTalkChannel { - fn name(&self) -> &str { - "dingtalk" - } - - async fn send(&self, message: &SendMessage) -> anyhow::Result<()> { - let webhooks = self.session_webhooks.read().await; - let webhook_url = webhooks.get(&message.recipient).ok_or_else(|| { - anyhow::anyhow!( - "No session webhook found for chat {}. \ - The user must send a message first to establish a session.", - message.recipient - ) - })?; - - let title = message.subject.as_deref().unwrap_or("OpenHuman"); - let body = serde_json::json!({ - "msgtype": "markdown", - "markdown": { - "title": title, - "text": message.content, - } - }); - - let resp = self - .http_client() - .post(webhook_url) - .json(&body) - .send() - .await?; - - if !resp.status().is_success() { - let status = resp.status(); - let err = resp.text().await.unwrap_or_default(); - anyhow::bail!("DingTalk webhook reply failed ({status}): {err}"); - } - - Ok(()) - } - - async fn listen(&self, tx: tokio::sync::mpsc::Sender) -> anyhow::Result<()> { - tracing::info!("DingTalk: registering gateway connection..."); - - let gw = self.register_connection().await?; - let ws_url = format!("{}?ticket={}", gw.endpoint, gw.ticket); - - tracing::info!("DingTalk: connecting to stream WebSocket..."); - let (ws_stream, _) = tokio_tungstenite::connect_async(&ws_url).await?; - let (mut write, mut read) = ws_stream.split(); - - tracing::info!("DingTalk: connected and listening for messages..."); - - while let Some(msg) = read.next().await { - let msg = match msg { - Ok(Message::Text(t)) => t, - Ok(Message::Close(_)) => break, - Err(e) => { - tracing::warn!("DingTalk WebSocket error: {e}"); - break; - } - _ => continue, - }; - - let frame: serde_json::Value = match serde_json::from_str(msg.as_ref()) { - Ok(v) => v, - Err(_) => continue, - }; - - let frame_type = frame.get("type").and_then(|t| t.as_str()).unwrap_or(""); - - match frame_type { - "SYSTEM" => { - // Respond to system pings to keep the connection alive - let message_id = frame - .get("headers") - .and_then(|h| h.get("messageId")) - .and_then(|m| m.as_str()) - .unwrap_or(""); - - let pong = serde_json::json!({ - "code": 200, - "headers": { - "contentType": "application/json", - "messageId": message_id, - }, - "message": "OK", - "data": "", - }); - - if let Err(e) = write.send(Message::Text(pong.to_string())).await { - tracing::warn!("DingTalk: failed to send pong: {e}"); - break; - } - } - "EVENT" | "CALLBACK" => { - // Parse the chatbot callback data from the frame. - let data = match Self::parse_stream_data(&frame) { - Some(v) => v, - None => { - tracing::debug!("DingTalk: frame has no parseable data payload"); - continue; - } - }; - - // Extract message content - let content = data - .get("text") - .and_then(|t| t.get("content")) - .and_then(|c| c.as_str()) - .unwrap_or("") - .trim(); - - if content.is_empty() { - continue; - } - - let sender_id = data - .get("senderStaffId") - .and_then(|s| s.as_str()) - .unwrap_or("unknown"); - - if !self.is_user_allowed(sender_id) { - tracing::warn!( - "DingTalk: ignoring message from unauthorized user: {sender_id}" - ); - continue; - } - - // Private chat uses sender ID, group chat uses conversation ID. - let chat_id = Self::resolve_chat_id(&data, sender_id); - - // Store session webhook for later replies - if let Some(webhook) = data.get("sessionWebhook").and_then(|w| w.as_str()) { - let webhook = webhook.to_string(); - let mut webhooks = self.session_webhooks.write().await; - // Use both keys so reply routing works for both group and private flows. - webhooks.insert(chat_id.clone(), webhook.clone()); - webhooks.insert(sender_id.to_string(), webhook); - } - - // Acknowledge the event - let message_id = frame - .get("headers") - .and_then(|h| h.get("messageId")) - .and_then(|m| m.as_str()) - .unwrap_or(""); - - let ack = serde_json::json!({ - "code": 200, - "headers": { - "contentType": "application/json", - "messageId": message_id, - }, - "message": "OK", - "data": "", - }); - let _ = write.send(Message::Text(ack.to_string())).await; - - let channel_msg = ChannelMessage { - id: Uuid::new_v4().to_string(), - sender: sender_id.to_string(), - reply_target: chat_id, - content: content.to_string(), - channel: "dingtalk".to_string(), - timestamp: std::time::SystemTime::now() - .duration_since(std::time::UNIX_EPOCH) - .unwrap_or_default() - .as_secs(), - thread_ts: None, - }; - - if tx.send(channel_msg).await.is_err() { - tracing::warn!("DingTalk: message channel closed"); - break; - } - } - _ => {} - } - } - - anyhow::bail!("DingTalk WebSocket stream ended") - } - - async fn health_check(&self) -> bool { - self.register_connection().await.is_ok() - } -} - -#[cfg(test)] -mod tests { - use super::*; - - #[test] - fn test_name() { - let ch = DingTalkChannel::new("id".into(), "secret".into(), vec![]); - assert_eq!(ch.name(), "dingtalk"); - } - - #[test] - fn test_user_allowed_wildcard() { - let ch = DingTalkChannel::new("id".into(), "secret".into(), vec!["*".into()]); - assert!(ch.is_user_allowed("anyone")); - } - - #[test] - fn test_user_allowed_specific() { - let ch = DingTalkChannel::new("id".into(), "secret".into(), vec!["user123".into()]); - assert!(ch.is_user_allowed("user123")); - assert!(!ch.is_user_allowed("other")); - } - - #[test] - fn test_user_denied_empty() { - let ch = DingTalkChannel::new("id".into(), "secret".into(), vec![]); - assert!(!ch.is_user_allowed("anyone")); - } - - #[test] - fn test_config_serde() { - let toml_str = r#" -client_id = "app_id_123" -client_secret = "secret_456" -allowed_users = ["user1", "*"] -"#; - let config: crate::openhuman::config::schema::DingTalkConfig = - toml::from_str(toml_str).unwrap(); - assert_eq!(config.client_id, "app_id_123"); - assert_eq!(config.client_secret, "secret_456"); - assert_eq!(config.allowed_users, vec!["user1", "*"]); - } - - #[test] - fn test_config_serde_defaults() { - let toml_str = r#" -client_id = "id" -client_secret = "secret" -"#; - let config: crate::openhuman::config::schema::DingTalkConfig = - toml::from_str(toml_str).unwrap(); - assert!(config.allowed_users.is_empty()); - } - - #[test] - fn parse_stream_data_supports_string_payload() { - let frame = serde_json::json!({ - "data": "{\"text\":{\"content\":\"hello\"}}" - }); - let parsed = DingTalkChannel::parse_stream_data(&frame).unwrap(); - assert_eq!( - parsed.get("text").and_then(|v| v.get("content")), - Some(&serde_json::json!("hello")) - ); - } - - #[test] - fn parse_stream_data_supports_object_payload() { - let frame = serde_json::json!({ - "data": {"text": {"content": "hello"}} - }); - let parsed = DingTalkChannel::parse_stream_data(&frame).unwrap(); - assert_eq!( - parsed.get("text").and_then(|v| v.get("content")), - Some(&serde_json::json!("hello")) - ); - } - - #[test] - fn resolve_chat_id_handles_numeric_group_conversation_type() { - let data = serde_json::json!({ - "conversationType": 2, - "conversationId": "cid-group", - }); - let chat_id = DingTalkChannel::resolve_chat_id(&data, "staff-1"); - assert_eq!(chat_id, "cid-group"); - } -} +pub use tinychannels::providers::dingtalk::*; diff --git a/src/openhuman/channels/providers/discord.rs b/src/openhuman/channels/providers/discord.rs new file mode 100644 index 0000000000..83ce3b2d39 --- /dev/null +++ b/src/openhuman/channels/providers/discord.rs @@ -0,0 +1 @@ +pub use tinychannels::providers::discord::*; diff --git a/src/openhuman/channels/providers/discord/api.rs b/src/openhuman/channels/providers/discord/api.rs deleted file mode 100644 index 7b9ca8556b..0000000000 --- a/src/openhuman/channels/providers/discord/api.rs +++ /dev/null @@ -1,511 +0,0 @@ -//! Discord REST API helpers for guild/channel discovery and permission checks. - -use serde::{Deserialize, Serialize}; - -const DISCORD_API_BASE: &str = "https://discord.com/api/v10"; - -/// Minimal guild (server) info returned by `GET /users/@me/guilds`. -#[derive(Debug, Clone, Serialize, Deserialize)] -pub struct DiscordGuild { - pub id: String, - pub name: String, - pub icon: Option, -} - -/// Minimal channel info returned by `GET /guilds/{guild_id}/channels`. -#[derive(Debug, Clone, Serialize, Deserialize)] -pub struct DiscordTextChannel { - pub id: String, - pub name: String, - /// Discord channel type — 0 = text, 2 = voice, 4 = category, etc. - #[serde(rename = "type")] - pub channel_type: u64, - #[serde(default)] - pub position: u64, - /// Parent category ID (if nested under a category). - pub parent_id: Option, -} - -/// Result of a bot permission check for a given channel. -#[derive(Debug, Clone, Serialize, Deserialize)] -pub struct BotPermissionCheck { - pub can_view_channel: bool, - pub can_send_messages: bool, - pub can_read_message_history: bool, - pub missing_permissions: Vec, -} - -// Discord permission flag bits -const VIEW_CHANNEL: u64 = 1 << 10; // 0x400 -const SEND_MESSAGES: u64 = 1 << 11; // 0x800 -const READ_MESSAGE_HISTORY: u64 = 1 << 16; // 0x10000 - -fn build_client() -> reqwest::Client { - crate::openhuman::config::build_runtime_proxy_client("channel.discord") -} - -fn auth_header(token: &str) -> String { - format!("Bot {token}") -} - -/// Format a non-2xx response from the Discord REST API as a string -/// suitable for a JSON-RPC error result. -/// -/// **Load-bearing for issue #2285**: the global JSON-RPC dispatcher -/// at `src/core/jsonrpc.rs::is_session_expired_error` matches any -/// error string that contains `"401"` AND `"unauthorized"` as a -/// signal that the OpenHuman backend session has expired, and -/// publishes a `DomainEvent::SessionExpired` event that signs the -/// user out. A raw upstream Discord 401 (revoked bot token) would -/// previously trip that classifier — opening the connected-Discord -/// card on the Channels page logged the user out of OpenHuman over -/// a *Discord* credentials problem. -/// -/// The fix is to convert auth failures here into a Discord-domain -/// message that: -/// -/// 1. Does NOT contain both `"401"` and `"unauthorized"` as a pair -/// (so the global classifier ignores it). -/// 2. Tells the user the actual remediation: rotate the bot token -/// at `Settings → Channels → Discord`. -/// 3. Preserves the originating endpoint identifier in the message -/// so triage can still see WHICH Discord call failed without -/// plumbing a separate error code. -/// -/// Other non-2xx statuses (400 / 404 / 5xx) pass through with a -/// `Discord API error` prefix — they don't match the -/// `is_session_expired_error` predicate even when verbatim. -fn format_discord_http_error(endpoint: &str, status: reqwest::StatusCode, body: &str) -> String { - if status == reqwest::StatusCode::UNAUTHORIZED || status == reqwest::StatusCode::FORBIDDEN { - let kind = if status == reqwest::StatusCode::UNAUTHORIZED { - "bot token was rejected" - } else { - "bot token lacks required Discord permissions" - }; - // Spell out the HTTP code so `lower.contains("401")` does NOT - // match — see #2285 rationale on this helper. Also avoid the - // word `unauthorized` for the same reason; "rejected"/"forbidden" - // are the user-visible equivalents. - let code_word = if status == reqwest::StatusCode::UNAUTHORIZED { - "four-oh-one" - } else { - "four-oh-three" - }; - // Deliberately do NOT splice the upstream body into this - // user-facing message — Discord's auth-error bodies often - // include the literal words "401" and "Unauthorized", which - // would smuggle the cascade trigger back in. The body is - // still in `tracing::debug!` logs above the call site for - // triage; the user-facing message only needs the remediation. - let _ = body; - format!( - "Discord {endpoint}: {kind} (upstream HTTP {code_word}). \ - Open Settings → Channels → Discord and rotate / reconnect the bot \ - token." - ) - } else { - format!("Discord {endpoint} failed ({status}): {body}") - } -} - -/// List all guilds (servers) the bot is a member of. -pub async fn list_bot_guilds(token: &str) -> anyhow::Result> { - list_bot_guilds_at_base(DISCORD_API_BASE, token).await -} - -/// Test seam: list guilds against an arbitrary API base. Used by -/// `list_bot_guilds` in production and by unit tests that drive a -/// local mock Discord API. -async fn list_bot_guilds_at_base(base: &str, token: &str) -> anyhow::Result> { - let url = format!("{base}/users/@me/guilds"); - tracing::debug!("[discord-api] listing guilds for bot"); - - let resp = build_client() - .get(&url) - .header("Authorization", auth_header(token)) - .send() - .await?; - - if !resp.status().is_success() { - let status = resp.status(); - let body = resp.text().await.unwrap_or_default(); - tracing::debug!( - target: "discord-api", - endpoint = "list_guilds", - %url, - %status, - body = %body, - "[discord-api] non-success response" - ); - anyhow::bail!( - "{}", - format_discord_http_error("list_guilds", status, &body) - ); - } - - let guilds: Vec = resp.json().await?; - tracing::debug!("[discord-api] found {} guilds", guilds.len()); - Ok(guilds) -} - -/// List text channels in a guild. Filters to type=0 (text channels) only. -pub async fn list_guild_channels( - token: &str, - guild_id: &str, -) -> anyhow::Result> { - list_guild_channels_at_base(DISCORD_API_BASE, token, guild_id).await -} - -/// Test seam: list guild channels against an arbitrary API base. -async fn list_guild_channels_at_base( - base: &str, - token: &str, - guild_id: &str, -) -> anyhow::Result> { - let url = format!("{base}/guilds/{guild_id}/channels"); - tracing::debug!("[discord-api] listing channels for guild {guild_id}"); - - let resp = build_client() - .get(&url) - .header("Authorization", auth_header(token)) - .send() - .await?; - - if !resp.status().is_success() { - let status = resp.status(); - let body = resp.text().await.unwrap_or_default(); - tracing::debug!( - target: "discord-api", - endpoint = "list_guild_channels", - %guild_id, - %url, - %status, - body = %body, - "[discord-api] non-success response" - ); - anyhow::bail!( - "{}", - format_discord_http_error("list_channels", status, &body) - ); - } - - let all_channels: Vec = resp.json().await?; - - // Filter to text channels (type 0) and sort by position - let mut text_channels: Vec = all_channels - .into_iter() - .filter(|c| c.channel_type == 0) - .collect(); - text_channels.sort_by_key(|c| c.position); - - tracing::debug!( - "[discord-api] found {} text channels in guild {guild_id}", - text_channels.len() - ); - Ok(text_channels) -} - -/// Check bot permissions in a specific channel. -/// -/// Uses `GET /channels/{channel_id}` combined with the bot's guild member -/// permissions to determine if the bot can view, send, and read history. -pub async fn check_channel_permissions( - token: &str, - guild_id: &str, - channel_id: &str, -) -> anyhow::Result { - check_channel_permissions_at_base(DISCORD_API_BASE, token, guild_id, channel_id).await -} - -/// Test seam: see [`check_channel_permissions`]. -async fn check_channel_permissions_at_base( - base: &str, - token: &str, - guild_id: &str, - channel_id: &str, -) -> anyhow::Result { - tracing::debug!( - "[discord-api] checking permissions in channel {channel_id} (guild {guild_id})" - ); - - // Resolve bot user id first (`members/@me` is not a valid Discord route). - let me_url = format!("{base}/users/@me"); - let me_resp = build_client() - .get(&me_url) - .header("Authorization", auth_header(token)) - .send() - .await?; - - if !me_resp.status().is_success() { - let status = me_resp.status(); - let body = me_resp.text().await.unwrap_or_default(); - tracing::debug!( - target: "discord-api", - endpoint = "check_bot_permissions.me", - %guild_id, - %channel_id, - url = %me_url, - %status, - body = %body, - "[discord-api] non-success response" - ); - anyhow::bail!( - "{}", - format_discord_http_error("get_bot_user", status, &body) - ); - } - let me: serde_json::Value = me_resp.json().await?; - let bot_user_id = me.get("id").and_then(|i| i.as_str()).unwrap_or("").trim(); - if bot_user_id.is_empty() { - anyhow::bail!("Discord get bot user returned empty id"); - } - - // Fetch the bot's guild member info which includes role ids. - let member_url = format!("{base}/guilds/{guild_id}/members/{bot_user_id}"); - let member_resp = build_client() - .get(&member_url) - .header("Authorization", auth_header(token)) - .send() - .await?; - - if !member_resp.status().is_success() { - let status = member_resp.status(); - let body = member_resp.text().await.unwrap_or_default(); - tracing::debug!( - target: "discord-api", - endpoint = "check_bot_permissions.member", - %guild_id, - %channel_id, - url = %member_url, - %status, - body = %body, - "[discord-api] non-success response" - ); - anyhow::bail!( - "{}", - format_discord_http_error("get_member_info", status, &body) - ); - } - - let member: serde_json::Value = member_resp.json().await?; - - // Fetch guild roles to compute permissions - let roles_url = format!("{base}/guilds/{guild_id}/roles"); - let roles_resp = build_client() - .get(&roles_url) - .header("Authorization", auth_header(token)) - .send() - .await?; - if !roles_resp.status().is_success() { - let status = roles_resp.status(); - let body = roles_resp.text().await.unwrap_or_default(); - tracing::debug!( - target: "discord-api", - endpoint = "check_bot_permissions.roles", - %guild_id, - %channel_id, - url = %roles_url, - %status, - body = %body, - "[discord-api] non-success response" - ); - anyhow::bail!( - "{}", - format_discord_http_error("get_guild_roles", status, &body) - ); - } - let guild_roles: Vec = roles_resp.json().await?; - - // Get the member's role IDs - let member_role_ids: Vec<&str> = member - .get("roles") - .and_then(|r| r.as_array()) - .map(|arr| arr.iter().filter_map(|v| v.as_str()).collect::>()) - .unwrap_or_default(); - - // Compute base permissions from @everyone role + member roles - let mut permissions: u64 = 0; - for role in &guild_roles { - let role_id = role.get("id").and_then(|i| i.as_str()).unwrap_or(""); - let is_everyone = role_id == guild_id; // @everyone role ID == guild ID - let is_member_role = member_role_ids.contains(&role_id); - - if is_everyone || is_member_role { - if let Some(perms_str) = role.get("permissions").and_then(|p| p.as_str()) { - if let Ok(perms) = perms_str.parse::() { - permissions |= perms; - } - } - } - } - - // Administrator bypasses all permission checks - const ADMINISTRATOR: u64 = 1 << 3; - if permissions & ADMINISTRATOR != 0 { - return Ok(BotPermissionCheck { - can_view_channel: true, - can_send_messages: true, - can_read_message_history: true, - missing_permissions: vec![], - }); - } - - // Now check channel-level permission overwrites - let channel_url = format!("{base}/channels/{channel_id}"); - let ch_resp = build_client() - .get(&channel_url) - .header("Authorization", auth_header(token)) - .send() - .await?; - if !ch_resp.status().is_success() { - let status = ch_resp.status(); - let body = ch_resp.text().await.unwrap_or_default(); - tracing::debug!( - target: "discord-api", - endpoint = "check_bot_permissions.channel", - %guild_id, - %channel_id, - url = %channel_url, - %status, - body = %body, - "[discord-api] non-success response" - ); - anyhow::bail!( - "{}", - format_discord_http_error("get_channel", status, &body) - ); - } - let channel_data: serde_json::Value = ch_resp.json().await?; - if let Some(overwrites) = channel_data - .get("permission_overwrites") - .and_then(|o| o.as_array()) - { - // Intentional shadowing: prefer the ID returned inside the member - // object over the one fetched from /users/@me, because the guild - // member record is more authoritative for permission overwrite lookups. - let bot_user_id = member - .get("user") - .and_then(|u| u.get("id")) - .and_then(|i| i.as_str()) - .unwrap_or(bot_user_id); - - let mut everyone_allow = 0_u64; - let mut everyone_deny = 0_u64; - let mut role_allow = 0_u64; - let mut role_deny = 0_u64; - let mut member_allow = 0_u64; - let mut member_deny = 0_u64; - - for overwrite in overwrites { - let ow_id = overwrite.get("id").and_then(|i| i.as_str()).unwrap_or(""); - let ow_type = overwrite.get("type").and_then(|t| t.as_u64()).unwrap_or(0); - let allow = overwrite - .get("allow") - .and_then(|a| a.as_str()) - .and_then(|s| s.parse::().ok()) - .unwrap_or(0); - let deny = overwrite - .get("deny") - .and_then(|d| d.as_str()) - .and_then(|s| s.parse::().ok()) - .unwrap_or(0); - - match ow_type { - // @everyone overwrite (role id == guild id) - 0 if ow_id == guild_id => { - everyone_allow = allow; - everyone_deny = deny; - } - // Aggregate all role overwrites - 0 if member_role_ids.contains(&ow_id) => { - role_allow |= allow; - role_deny |= deny; - } - // Member-specific overwrite - 1 if ow_id == bot_user_id => { - member_allow = allow; - member_deny = deny; - } - _ => {} - } - } - - // Apply Discord overwrite precedence: everyone -> roles -> member. - permissions &= !everyone_deny; - permissions |= everyone_allow; - permissions &= !role_deny; - permissions |= role_allow; - permissions &= !member_deny; - permissions |= member_allow; - } - - let can_view = permissions & VIEW_CHANNEL != 0; - let can_send = permissions & SEND_MESSAGES != 0; - let can_read_history = permissions & READ_MESSAGE_HISTORY != 0; - - let mut missing = Vec::new(); - if !can_view { - missing.push("VIEW_CHANNEL".to_string()); - } - if !can_send { - missing.push("SEND_MESSAGES".to_string()); - } - if !can_read_history { - missing.push("READ_MESSAGE_HISTORY".to_string()); - } - - tracing::debug!( - "[discord-api] permissions for channel {channel_id}: view={can_view}, send={can_send}, history={can_read_history}" - ); - - Ok(BotPermissionCheck { - can_view_channel: can_view, - can_send_messages: can_send, - can_read_message_history: can_read_history, - missing_permissions: missing, - }) -} - -#[cfg(test)] -#[path = "api_tests.rs"] -mod tests; - -#[cfg(any(test, debug_assertions))] -pub mod test_support { - //! Debug-build wrappers for raw integration tests that drive the Discord - //! REST helpers against loopback servers. - - use super::*; - - pub fn format_discord_http_error_for_test( - endpoint: &str, - status: reqwest::StatusCode, - body: &str, - ) -> String { - format_discord_http_error(endpoint, status, body) - } - - pub async fn list_bot_guilds_at_base_for_test( - base: &str, - token: &str, - ) -> anyhow::Result> { - list_bot_guilds_at_base(base, token).await - } - - pub async fn list_guild_channels_at_base_for_test( - base: &str, - token: &str, - guild_id: &str, - ) -> anyhow::Result> { - list_guild_channels_at_base(base, token, guild_id).await - } - - pub async fn check_channel_permissions_at_base_for_test( - base: &str, - token: &str, - guild_id: &str, - channel_id: &str, - ) -> anyhow::Result { - check_channel_permissions_at_base(base, token, guild_id, channel_id).await - } -} diff --git a/src/openhuman/channels/providers/discord/api_tests.rs b/src/openhuman/channels/providers/discord/api_tests.rs deleted file mode 100644 index b222e15f08..0000000000 --- a/src/openhuman/channels/providers/discord/api_tests.rs +++ /dev/null @@ -1,451 +0,0 @@ -use super::*; - -#[test] -fn guild_deserializes() { - let json = r#"{"id":"123","name":"Test Server","icon":"abc123"}"#; - let guild: DiscordGuild = serde_json::from_str(json).unwrap(); - assert_eq!(guild.id, "123"); - assert_eq!(guild.name, "Test Server"); - assert_eq!(guild.icon, Some("abc123".to_string())); -} - -#[test] -fn guild_deserializes_without_icon() { - let json = r#"{"id":"456","name":"No Icon","icon":null}"#; - let guild: DiscordGuild = serde_json::from_str(json).unwrap(); - assert_eq!(guild.id, "456"); - assert!(guild.icon.is_none()); -} - -#[test] -fn text_channel_deserializes() { - let json = r#"{"id":"789","name":"general","type":0,"position":1,"parent_id":"100"}"#; - let ch: DiscordTextChannel = serde_json::from_str(json).unwrap(); - assert_eq!(ch.id, "789"); - assert_eq!(ch.name, "general"); - assert_eq!(ch.channel_type, 0); - assert_eq!(ch.position, 1); - assert_eq!(ch.parent_id, Some("100".to_string())); -} - -#[test] -fn text_channel_without_parent() { - let json = r#"{"id":"789","name":"general","type":0,"position":0,"parent_id":null}"#; - let ch: DiscordTextChannel = serde_json::from_str(json).unwrap(); - assert!(ch.parent_id.is_none()); -} - -#[test] -fn permission_check_serializes() { - let check = BotPermissionCheck { - can_view_channel: true, - can_send_messages: true, - can_read_message_history: false, - missing_permissions: vec!["READ_MESSAGE_HISTORY".to_string()], - }; - let json = serde_json::to_string(&check).unwrap(); - assert!(json.contains("READ_MESSAGE_HISTORY")); -} - -#[test] -fn permission_bits_are_correct() { - assert_eq!(VIEW_CHANNEL, 1024); - assert_eq!(SEND_MESSAGES, 2048); - assert_eq!(READ_MESSAGE_HISTORY, 65536); -} - -#[test] -fn auth_header_has_bot_prefix() { - assert_eq!(auth_header("abc"), "Bot abc"); - assert_eq!(auth_header(""), "Bot "); -} - -#[test] -fn permission_check_lists_all_missing_permissions_when_bot_lacks_any() { - let check = BotPermissionCheck { - can_view_channel: false, - can_send_messages: false, - can_read_message_history: false, - missing_permissions: vec![ - "VIEW_CHANNEL".into(), - "SEND_MESSAGES".into(), - "READ_MESSAGE_HISTORY".into(), - ], - }; - let json = serde_json::to_string(&check).unwrap(); - assert!(json.contains("VIEW_CHANNEL")); - assert!(json.contains("SEND_MESSAGES")); - assert!(json.contains("READ_MESSAGE_HISTORY")); -} - -#[test] -fn permission_check_with_all_granted_has_empty_missing_list() { - let check = BotPermissionCheck { - can_view_channel: true, - can_send_messages: true, - can_read_message_history: true, - missing_permissions: vec![], - }; - let json = serde_json::to_string(&check).unwrap(); - assert!(json.contains("\"missing_permissions\":[]")); -} - -#[test] -fn text_channel_type_zero_is_standard_text() { - let json = r#"{"id":"1","name":"general","type":0,"position":0,"parent_id":null}"#; - let ch: DiscordTextChannel = serde_json::from_str(json).unwrap(); - assert_eq!(ch.channel_type, 0); -} - -#[test] -fn guild_deserializes_with_full_payload() { - let json = r#"{ - "id": "999", - "name": "Full Guild", - "icon": "hash" - }"#; - let g: DiscordGuild = serde_json::from_str(json).unwrap(); - assert_eq!(g.id, "999"); - assert_eq!(g.name, "Full Guild"); -} - -#[test] -fn permission_bit_flags_are_disjoint() { - // Sanity: each permission is a single bit and distinct. - assert_eq!(VIEW_CHANNEL.count_ones(), 1); - assert_eq!(SEND_MESSAGES.count_ones(), 1); - assert_eq!(READ_MESSAGE_HISTORY.count_ones(), 1); - assert_ne!(VIEW_CHANNEL, SEND_MESSAGES); - assert_ne!(SEND_MESSAGES, READ_MESSAGE_HISTORY); -} - -// ── Mock Discord server integration tests ────────────────────── - -use axum::{extract::Path, http::StatusCode, routing::get, Json, Router}; -use serde_json::json; - -async fn spawn_mock(app: Router) -> String { - let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); - let addr = listener.local_addr().unwrap(); - tokio::spawn(async move { axum::serve(listener, app).await.unwrap() }); - format!("http://127.0.0.1:{}", addr.port()) -} - -#[tokio::test] -async fn list_bot_guilds_parses_discord_response() { - let app = Router::new().route( - "/users/@me/guilds", - get(|| async { - Json(json!([ - {"id": "g1", "name": "Guild One", "icon": "hash1"}, - {"id": "g2", "name": "Guild Two", "icon": null} - ])) - }), - ); - let base = spawn_mock(app).await; - let guilds = list_bot_guilds_at_base(&base, "test-token").await.unwrap(); - assert_eq!(guilds.len(), 2); - assert_eq!(guilds[0].id, "g1"); - assert_eq!(guilds[0].name, "Guild One"); - assert_eq!(guilds[1].icon, None); -} - -#[tokio::test] -async fn list_bot_guilds_rewraps_401_so_global_session_cascade_does_not_fire() { - // Upstream returns 401 with the canonical Discord auth-error body. - // BEFORE #2285 the error string flowed up to JSON-RPC as - // "Discord list guilds failed (401 Unauthorized): {\"message\": - // \"401: Unauthorized\",\"code\":0}" — that pair tripped - // `jsonrpc::is_session_expired_error` ("401" + "unauthorized") - // and logged the user out of OpenHuman over a *Discord* - // credentials problem. - // - // After the fix the user-facing message: - // - does NOT contain "401" or "unauthorized" as substrings - // (so `is_session_expired_error` returns false), AND - // - names the endpoint + the actionable Settings → Channels → - // Discord remediation path. - let app = Router::new().route( - "/users/@me/guilds", - get(|| async { - ( - StatusCode::UNAUTHORIZED, - r#"{"message":"401: Unauthorized","code":0}"#, - ) - }), - ); - let base = spawn_mock(app).await; - let err = list_bot_guilds_at_base(&base, "t") - .await - .unwrap_err() - .to_string(); - let lower = err.to_ascii_lowercase(); - assert!( - !lower.contains("401"), - "must NOT contain '401' substring: {err}" - ); - assert!( - !lower.contains("unauthorized"), - "must NOT contain 'unauthorized' substring: {err}" - ); - assert!( - err.contains("list_guilds"), - "endpoint identifier preserved for triage: {err}" - ); - assert!( - err.contains("Settings → Channels → Discord"), - "remediation path present: {err}" - ); -} - -#[tokio::test] -async fn list_bot_guilds_5xx_still_carries_raw_status() { - // Non-auth errors fall through to the legacy verbose format — - // those don't match `is_session_expired_error` even verbatim, so - // surfacing the raw status code helps the user / triage. - let app = Router::new().route( - "/users/@me/guilds", - get(|| async { (StatusCode::INTERNAL_SERVER_ERROR, "discord melting") }), - ); - let base = spawn_mock(app).await; - let err = list_bot_guilds_at_base(&base, "t") - .await - .unwrap_err() - .to_string(); - assert!( - err.contains("500"), - "5xx must surface verbatim status: {err}" - ); - assert!(err.contains("list_guilds")); -} - -#[tokio::test] -async fn list_guild_channels_filters_text_channels_and_sorts_by_position() { - let app = Router::new().route( - "/guilds/{guild_id}/channels", - get(|Path(guild_id): Path| async move { - assert_eq!(guild_id, "g1"); - Json(json!([ - {"id": "c3", "name": "category", "type": 4, "position": 0, "parent_id": null}, - {"id": "c1", "name": "general", "type": 0, "position": 2, "parent_id": null}, - {"id": "c2", "name": "random", "type": 0, "position": 1, "parent_id": null}, - {"id": "c4", "name": "voice", "type": 2, "position": 3, "parent_id": null} - ])) - }), - ); - let base = spawn_mock(app).await; - let channels = list_guild_channels_at_base(&base, "t", "g1").await.unwrap(); - // Only text channels (type=0) remain, sorted by position ascending. - assert_eq!(channels.len(), 2); - assert_eq!(channels[0].id, "c2"); - assert_eq!(channels[1].id, "c1"); -} - -#[tokio::test] -async fn list_guild_channels_rewraps_403_with_remediation_and_no_session_keywords() { - // 403 follows the same rewrap path as 401 (#2285) — both can - // happen on a stale/disabled bot token AND both share enough - // substrings with `is_session_expired_error` to be a problem if - // raw upstream text reaches the JSON-RPC layer. The user-facing - // message must use the safer wording. - let app = Router::new().route( - "/guilds/{guild_id}/channels", - get(|| async { - ( - StatusCode::FORBIDDEN, - r#"{"message":"Missing Access","code":50001}"#, - ) - }), - ); - let base = spawn_mock(app).await; - let err = list_guild_channels_at_base(&base, "t", "g1") - .await - .unwrap_err() - .to_string(); - let lower = err.to_ascii_lowercase(); - assert!(!lower.contains("403"), "raw 403 must not leak: {err}"); - assert!( - err.contains("list_channels"), - "endpoint identifier preserved: {err}" - ); - assert!( - err.contains("Settings → Channels → Discord"), - "remediation path present: {err}" - ); -} - -#[tokio::test] -async fn list_guild_channels_empty_returns_empty_vec() { - let app = Router::new().route( - "/guilds/{guild_id}/channels", - get(|| async { Json(json!([])) }), - ); - let base = spawn_mock(app).await; - let channels = list_guild_channels_at_base(&base, "t", "g").await.unwrap(); - assert!(channels.is_empty()); -} - -// ── check_channel_permissions ───────────────────────────────── - -/// Build a mock Discord that answers all endpoints the permissions check -/// touches: `/users/@me`, `/guilds//members/`, -/// `/guilds//roles`, and `/channels/`. -fn permissions_mock( - member: serde_json::Value, - roles: serde_json::Value, - channel: serde_json::Value, -) -> Router { - use axum::extract::Path; - Router::new() - .route( - "/users/@me", - get(|| async { Json(json!({ "id": "bot-1" })) }), - ) - .route( - "/guilds/{guild_id}/members/{member_id}", - get(move |Path((_g, member_id)): Path<(String, String)>| { - assert_eq!(member_id, "bot-1"); - let m = member.clone(); - async move { Json(m) } - }), - ) - .route( - "/guilds/{guild_id}/roles", - get(move |Path(_g): Path| { - let r = roles.clone(); - async move { Json(r) } - }), - ) - .route( - "/channels/{channel_id}", - get(move |Path(_c): Path| { - let c = channel.clone(); - async move { Json(c) } - }), - ) -} - -#[tokio::test] -async fn check_channel_permissions_administrator_bypasses_everything() { - let member = json!({ "roles": ["role-admin"], "user": { "id": "bot-1" } }); - // Role with Administrator bit (1<<3 = 8) — overrides all other checks. - let roles = json!([ - { "id": "role-admin", "permissions": "8" } - ]); - let channel = json!({ "permission_overwrites": [] }); - let base = spawn_mock(permissions_mock(member, roles, channel)).await; - let out = check_channel_permissions_at_base(&base, "token", "guild-1", "channel-1") - .await - .unwrap(); - assert!(out.can_view_channel); - assert!(out.can_send_messages); - assert!(out.can_read_message_history); - assert!(out.missing_permissions.is_empty()); -} - -#[tokio::test] -async fn check_channel_permissions_flags_missing_bits_when_role_lacks_them() { - // No roles grant any of the 3 permissions → all missing. - let member = json!({ "roles": ["role-nobody"], "user": { "id": "bot-1" } }); - let roles = json!([ - { "id": "role-nobody", "permissions": "0" } - ]); - let channel = json!({ "permission_overwrites": [] }); - let base = spawn_mock(permissions_mock(member, roles, channel)).await; - let out = check_channel_permissions_at_base(&base, "t", "guild-1", "channel-1") - .await - .unwrap(); - assert!(!out.can_view_channel); - assert!(!out.can_send_messages); - assert!(!out.can_read_message_history); - assert!(out - .missing_permissions - .contains(&"VIEW_CHANNEL".to_string())); - assert!(out - .missing_permissions - .contains(&"SEND_MESSAGES".to_string())); - assert!(out - .missing_permissions - .contains(&"READ_MESSAGE_HISTORY".to_string())); -} - -#[tokio::test] -async fn check_channel_permissions_grants_everything_when_everyone_role_allows() { - // @everyone role (id == guild_id) grants VIEW|SEND|HISTORY - // = 1024 | 2048 | 65536 = 68608 - let member = json!({ "roles": [], "user": { "id": "bot-1" } }); - let roles = json!([ - { "id": "guild-1", "permissions": "68608" } - ]); - let channel = json!({ "permission_overwrites": [] }); - let base = spawn_mock(permissions_mock(member, roles, channel)).await; - let out = check_channel_permissions_at_base(&base, "t", "guild-1", "channel-1") - .await - .unwrap(); - assert!(out.can_view_channel); - assert!(out.can_send_messages); - assert!(out.can_read_message_history); - assert!(out.missing_permissions.is_empty()); -} - -#[tokio::test] -async fn check_channel_permissions_channel_overwrite_can_deny_permission() { - // @everyone role grants everything, but the channel's @everyone - // overwrite denies VIEW_CHANNEL — expect VIEW missing. - let member = json!({ "roles": [], "user": { "id": "bot-1" } }); - let roles = json!([ - { "id": "guild-1", "permissions": "68608" } - ]); - let channel = json!({ - "permission_overwrites": [ - { - "id": "guild-1", - "type": 0, - "allow": "0", - "deny": "1024" // VIEW_CHANNEL - } - ] - }); - let base = spawn_mock(permissions_mock(member, roles, channel)).await; - let out = check_channel_permissions_at_base(&base, "t", "guild-1", "channel-1") - .await - .unwrap(); - assert!(!out.can_view_channel); - assert!(out - .missing_permissions - .contains(&"VIEW_CHANNEL".to_string())); -} - -#[tokio::test] -async fn check_channel_permissions_errors_on_member_lookup_failure() { - use axum::http::StatusCode; - let app = Router::new() - .route( - "/users/@me", - get(|| async { Json(json!({ "id": "bot-1" })) }), - ) - .route( - "/guilds/{guild_id}/members/{member_id}", - get(|Path((_g, _member_id)): Path<(String, String)>| async { - (StatusCode::UNAUTHORIZED, "bad token") - }), - ); - let base = spawn_mock(app).await; - let err = check_channel_permissions_at_base(&base, "t", "g", "c") - .await - .unwrap_err() - .to_string(); - // Endpoint identifier preserved in the rewrap (#2285), and the - // 401 path keeps the substrings "401"/"unauthorized" out of the - // user-facing message so the JSON-RPC session-expired classifier - // ignores it. - assert!(err.contains("get_member_info")); - assert!( - !err.to_ascii_lowercase().contains("401"), - "rewrapped message must not contain '401': {err}" - ); - assert!( - !err.to_ascii_lowercase().contains("unauthorized"), - "rewrapped message must not contain 'unauthorized': {err}" - ); -} diff --git a/src/openhuman/channels/providers/discord/channel.rs b/src/openhuman/channels/providers/discord/channel.rs deleted file mode 100644 index 73227fb55f..0000000000 --- a/src/openhuman/channels/providers/discord/channel.rs +++ /dev/null @@ -1,525 +0,0 @@ -use crate::openhuman::channels::traits::{Channel, ChannelMessage, SendMessage}; -use async_trait::async_trait; -use futures_util::{SinkExt, StreamExt}; -use parking_lot::Mutex; -use serde_json::json; -use tinychannels::channel::LengthUnit; -use tinychannels::text::{chunk_text_with_options, ChunkMode, TextChunkOptions}; -use tokio_tungstenite::tungstenite::Message; -use uuid::Uuid; - -/// Discord channel — connects via Gateway WebSocket for real-time messages -pub struct DiscordChannel { - bot_token: String, - guild_id: Option, - channel_id: Option, - allowed_users: Vec, - listen_to_bots: bool, - mention_only: bool, - typing_handle: Mutex>>, -} - -impl DiscordChannel { - pub fn new( - bot_token: String, - guild_id: Option, - channel_id: Option, - allowed_users: Vec, - listen_to_bots: bool, - mention_only: bool, - ) -> Self { - Self { - bot_token, - guild_id, - channel_id, - allowed_users, - listen_to_bots, - mention_only, - typing_handle: Mutex::new(None), - } - } - - fn http_client(&self) -> reqwest::Client { - crate::openhuman::config::build_runtime_proxy_client("channel.discord") - } - - /// Check if a Discord user ID is in the allowlist. - /// - /// Empty list ⇒ allow-all: an unconfigured allowlist applies no per-user - /// restriction (the bot is still scoped to its configured guild/channel). - /// Previously an empty list denied *everyone*, so a bot connected via the UI - /// with the default-empty allowlist silently ignored every message and never - /// replied (issue #3712). This now matches the WhatsApp provider's - /// empty-⇒-allow-all convention. `"*"` also allows everyone; populate the - /// list with specific user IDs to restrict. - fn is_user_allowed(&self, user_id: &str) -> bool { - self.allowed_users.is_empty() || self.allowed_users.iter().any(|u| u == "*" || u == user_id) - } - - /// Decide whether a message passes the bot's guild scoping. - /// - /// - No configured guild → all messages pass (guild filter inactive). - /// - Configured guild, message in a *different* guild → blocked. - /// - Configured guild, message in that guild → allowed. - /// - Configured guild, DM (no `guild_id`) → only when the allowlist is - /// non-empty (explicit). `is_user_allowed` treats an empty list as - /// allow-all (the intended *within-guild* default), but DMs bypass the - /// guild filter — so a blank allowlist must not open a guild-scoped bot to - /// arbitrary DMs (#3794 review — Codex P1). `"*"` (non-empty) still allows. - fn passes_guild_scope( - configured_guild: Option<&str>, - msg_guild: Option<&str>, - allowlist_empty: bool, - ) -> bool { - let Some(gid) = configured_guild else { - return true; - }; - match msg_guild { - Some(g) => g == gid, - None => !allowlist_empty, - } - } - - /// Resolve the outbound recipient channel id. Prefer the message's explicit - /// recipient (e.g. the channel a reply targets); fall back to the bot's - /// configured `channel_id` for recipient-less sends such as proactive - /// cron/heartbeat delivery (#3794 review — Codex P2). `None` when neither is - /// available, so the caller surfaces an error instead of POSTing to an empty - /// channel id. - fn resolve_recipient<'a>( - msg_recipient: &'a str, - configured: Option<&'a str>, - ) -> Option<&'a str> { - let recipient = if msg_recipient.is_empty() { - configured.unwrap_or("") - } else { - msg_recipient - }; - (!recipient.is_empty()).then_some(recipient) - } - - fn bot_user_id_from_token(token: &str) -> Option { - // Discord bot tokens are base64(bot_user_id).timestamp.hmac - let part = token.split('.').next()?; - base64_decode(part) - } -} - -const BASE64_ALPHABET: &[u8] = b"ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz0123456789+/"; - -/// Discord's maximum message length for regular messages. -/// -/// Discord rejects longer payloads with `50035 Invalid Form Body`. -const DISCORD_MAX_MESSAGE_LENGTH: usize = 2000; - -/// Split a message into chunks that respect Discord's 2000-character limit. -/// Tries to split at word boundaries when possible. -fn split_message_for_discord(message: &str) -> Vec { - if message.is_empty() { - return vec![String::new()]; - } - chunk_text_with_options( - message, - TextChunkOptions { - limit: DISCORD_MAX_MESSAGE_LENGTH, - length_unit: LengthUnit::Utf16Units, - mode: ChunkMode::Length, - markdown: true, - indicators: false, - }, - ) -} - -fn mention_tags(bot_user_id: &str) -> [String; 2] { - [format!("<@{bot_user_id}>"), format!("<@!{bot_user_id}>")] -} - -fn contains_bot_mention(content: &str, bot_user_id: &str) -> bool { - let tags = mention_tags(bot_user_id); - content.contains(&tags[0]) || content.contains(&tags[1]) -} - -fn normalize_incoming_content( - content: &str, - mention_only: bool, - bot_user_id: &str, -) -> Option { - if content.is_empty() { - return None; - } - - if mention_only && !contains_bot_mention(content, bot_user_id) { - return None; - } - - let mut normalized = content.to_string(); - if mention_only { - for tag in mention_tags(bot_user_id) { - normalized = normalized.replace(&tag, " "); - } - } - - let normalized = normalized.trim().to_string(); - if normalized.is_empty() { - return None; - } - - Some(normalized) -} - -/// Minimal base64 decode (no extra dep) — only needs to decode the user ID portion -#[allow(clippy::cast_possible_truncation)] -fn base64_decode(input: &str) -> Option { - let padded = match input.len() % 4 { - 2 => format!("{input}=="), - 3 => format!("{input}="), - _ => input.to_string(), - }; - - let mut bytes = Vec::new(); - let chars: Vec = padded.bytes().collect(); - - for chunk in chars.chunks(4) { - if chunk.len() < 4 { - break; - } - - let mut v = [0usize; 4]; - for (i, &b) in chunk.iter().enumerate() { - if b == b'=' { - v[i] = 0; - } else { - v[i] = BASE64_ALPHABET.iter().position(|&a| a == b)?; - } - } - - bytes.push(((v[0] << 2) | (v[1] >> 4)) as u8); - if chunk[2] != b'=' { - bytes.push((((v[1] & 0xF) << 4) | (v[2] >> 2)) as u8); - } - if chunk[3] != b'=' { - bytes.push((((v[2] & 0x3) << 6) | v[3]) as u8); - } - } - - String::from_utf8(bytes).ok() -} - -#[async_trait] -impl Channel for DiscordChannel { - fn name(&self) -> &str { - "discord" - } - - /// Recipient-less proactive sends (cron/heartbeat) deliver to the bot's - /// configured default `channel_id`. `None` when unconfigured, so proactive - /// routing skips Discord rather than letting `send` bail on an empty target - /// (#3794 review — Codex P2). - fn proactive_target(&self) -> Option { - self.channel_id - .as_deref() - .map(str::trim) - .filter(|s| !s.is_empty()) - .map(str::to_string) - } - - async fn send(&self, message: &SendMessage) -> anyhow::Result<()> { - // Resolve the target channel: explicit recipient (replies) or the - // configured default channel (recipient-less proactive sends). Bail with - // a clear error rather than POSTing to an empty channel id (#3794 review). - let Some(recipient) = - Self::resolve_recipient(&message.recipient, self.channel_id.as_deref()) - else { - anyhow::bail!( - "Discord send: no target channel — message had no recipient and no channel_id is configured" - ); - }; - - let chunks = split_message_for_discord(&message.content); - - for (i, chunk) in chunks.iter().enumerate() { - let url = format!("https://discord.com/api/v10/channels/{recipient}/messages"); - - let body = json!({ "content": chunk }); - - let resp = self - .http_client() - .post(&url) - .header("Authorization", format!("Bot {}", self.bot_token)) - .json(&body) - .send() - .await?; - - if !resp.status().is_success() { - let status = resp.status(); - let err = resp - .text() - .await - .unwrap_or_else(|e| format!("")); - anyhow::bail!("Discord send message failed ({status}): {err}"); - } - - // Add a small delay between chunks to avoid rate limiting - if i < chunks.len() - 1 { - tokio::time::sleep(std::time::Duration::from_millis(500)).await; - } - } - - Ok(()) - } - - #[allow(clippy::too_many_lines)] - async fn listen(&self, tx: tokio::sync::mpsc::Sender) -> anyhow::Result<()> { - let bot_user_id = Self::bot_user_id_from_token(&self.bot_token).unwrap_or_default(); - - // Get Gateway URL - let gw_resp: serde_json::Value = self - .http_client() - .get("https://discord.com/api/v10/gateway/bot") - .header("Authorization", format!("Bot {}", self.bot_token)) - .send() - .await? - .json() - .await?; - - let gw_url = gw_resp - .get("url") - .and_then(|u| u.as_str()) - .unwrap_or("wss://gateway.discord.gg"); - - let ws_url = format!("{gw_url}/?v=10&encoding=json"); - tracing::info!("Discord: connecting to gateway..."); - - let (ws_stream, _) = tokio_tungstenite::connect_async(&ws_url).await?; - let (mut write, mut read) = ws_stream.split(); - - // Read Hello (opcode 10) - let hello = read.next().await.ok_or(anyhow::anyhow!("No hello"))??; - let hello_data: serde_json::Value = serde_json::from_str(&hello.to_string())?; - let heartbeat_interval = hello_data - .get("d") - .and_then(|d| d.get("heartbeat_interval")) - .and_then(serde_json::Value::as_u64) - .unwrap_or(41250); - - // Send Identify (opcode 2) - let identify = json!({ - "op": 2, - "d": { - "token": self.bot_token, - "intents": 37377, // GUILDS | GUILD_MESSAGES | MESSAGE_CONTENT | DIRECT_MESSAGES - "properties": { - "os": "linux", - "browser": "openhuman", - "device": "openhuman" - } - } - }); - write.send(Message::Text(identify.to_string())).await?; - - tracing::info!("Discord: connected and identified"); - - // Track the last sequence number for heartbeats and resume. - // Only accessed in the select! loop below, so a plain i64 suffices. - let mut sequence: i64 = -1; - - // Spawn heartbeat timer — sends a tick signal, actual heartbeat - // is assembled in the select! loop where `sequence` lives. - let (hb_tx, mut hb_rx) = tokio::sync::mpsc::channel::<()>(1); - let hb_interval = heartbeat_interval; - tokio::spawn(async move { - let mut interval = tokio::time::interval(std::time::Duration::from_millis(hb_interval)); - loop { - interval.tick().await; - if hb_tx.send(()).await.is_err() { - break; - } - } - }); - - let guild_filter = self.guild_id.clone(); - let channel_filter = self.channel_id.clone(); - - loop { - tokio::select! { - _ = hb_rx.recv() => { - let d = if sequence >= 0 { json!(sequence) } else { json!(null) }; - let hb = json!({"op": 1, "d": d}); - if write.send(Message::Text(hb.to_string())).await.is_err() { - break; - } - } - msg = read.next() => { - let msg = match msg { - Some(Ok(Message::Text(t))) => t, - Some(Ok(Message::Close(_))) | None => break, - _ => continue, - }; - - let event: serde_json::Value = match serde_json::from_str(msg.as_ref()) { - Ok(e) => e, - Err(_) => continue, - }; - - // Track sequence number from all dispatch events - if let Some(s) = event.get("s").and_then(serde_json::Value::as_i64) { - sequence = s; - } - - let op = event.get("op").and_then(serde_json::Value::as_u64).unwrap_or(0); - - match op { - // Op 1: Server requests an immediate heartbeat - 1 => { - let d = if sequence >= 0 { json!(sequence) } else { json!(null) }; - let hb = json!({"op": 1, "d": d}); - if write.send(Message::Text(hb.to_string())).await.is_err() { - break; - } - continue; - } - // Op 7: Reconnect - 7 => { - tracing::warn!("Discord: received Reconnect (op 7), closing for restart"); - break; - } - // Op 9: Invalid Session - 9 => { - tracing::warn!("Discord: received Invalid Session (op 9), closing for restart"); - break; - } - _ => {} - } - - // Only handle MESSAGE_CREATE (opcode 0, type "MESSAGE_CREATE") - let event_type = event.get("t").and_then(|t| t.as_str()).unwrap_or(""); - if event_type != "MESSAGE_CREATE" { - continue; - } - - let Some(d) = event.get("d") else { - continue; - }; - - // Skip messages from the bot itself - let author_id = d.get("author").and_then(|a| a.get("id")).and_then(|i| i.as_str()).unwrap_or(""); - if author_id == bot_user_id { - continue; - } - - // Skip bot messages (unless listen_to_bots is enabled) - if !self.listen_to_bots && d.get("author").and_then(|a| a.get("bot")).and_then(serde_json::Value::as_bool).unwrap_or(false) { - continue; - } - - // Sender validation - if !self.is_user_allowed(author_id) { - tracing::warn!("Discord: ignoring message from unauthorized user: {author_id}"); - continue; - } - - // Guild filter + DM scoping (#3794 review — Codex P1) - if !Self::passes_guild_scope( - guild_filter.as_deref(), - d.get("guild_id").and_then(serde_json::Value::as_str), - self.allowed_users.is_empty(), - ) { - continue; - } - - // Channel filter — only process messages from the configured channel - if let Some(ref cid) = channel_filter { - let msg_channel = d.get("channel_id").and_then(serde_json::Value::as_str).unwrap_or(""); - if msg_channel != cid { - continue; - } - } - - let content = d.get("content").and_then(|c| c.as_str()).unwrap_or(""); - let Some(clean_content) = - normalize_incoming_content(content, self.mention_only, &bot_user_id) - else { - continue; - }; - - let message_id = d.get("id").and_then(|i| i.as_str()).unwrap_or(""); - let channel_id = d.get("channel_id").and_then(|c| c.as_str()).unwrap_or("").to_string(); - - let channel_msg = ChannelMessage { - id: if message_id.is_empty() { - format!("discord_{}", Uuid::new_v4()) - } else { - format!("discord_{message_id}") - }, - sender: author_id.to_string(), - reply_target: if channel_id.is_empty() { - author_id.to_string() - } else { - channel_id.clone() - }, - content: clean_content, - channel: "discord".to_string(), - timestamp: std::time::SystemTime::now() - .duration_since(std::time::UNIX_EPOCH) - .unwrap_or_default() - .as_secs(), - thread_ts: None, - }; - - if tx.send(channel_msg).await.is_err() { - break; - } - } - } - } - - Ok(()) - } - - async fn health_check(&self) -> bool { - self.http_client() - .get("https://discord.com/api/v10/users/@me") - .header("Authorization", format!("Bot {}", self.bot_token)) - .send() - .await - .map(|r| r.status().is_success()) - .unwrap_or(false) - } - - async fn start_typing(&self, recipient: &str) -> anyhow::Result<()> { - self.stop_typing(recipient).await?; - - let client = self.http_client(); - let token = self.bot_token.clone(); - let channel_id = recipient.to_string(); - - let handle = tokio::spawn(async move { - let url = format!("https://discord.com/api/v10/channels/{channel_id}/typing"); - loop { - let _ = client - .post(&url) - .header("Authorization", format!("Bot {token}")) - .send() - .await; - tokio::time::sleep(std::time::Duration::from_secs(8)).await; - } - }); - - let mut guard = self.typing_handle.lock(); - *guard = Some(handle); - - Ok(()) - } - - async fn stop_typing(&self, _recipient: &str) -> anyhow::Result<()> { - let mut guard = self.typing_handle.lock(); - if let Some(handle) = guard.take() { - handle.abort(); - } - Ok(()) - } -} - -#[cfg(test)] -#[path = "channel_tests.rs"] -mod tests; diff --git a/src/openhuman/channels/providers/discord/channel_tests.rs b/src/openhuman/channels/providers/discord/channel_tests.rs deleted file mode 100644 index 229e13977f..0000000000 --- a/src/openhuman/channels/providers/discord/channel_tests.rs +++ /dev/null @@ -1,581 +0,0 @@ -use super::*; - -fn discord_units(text: &str) -> usize { - text.encode_utf16().count() -} - -#[test] -fn discord_channel_name() { - let ch = DiscordChannel::new("fake".into(), None, None, vec![], false, false); - assert_eq!(ch.name(), "discord"); -} - -#[test] -fn base64_decode_bot_id() { - // "MTIzNDU2" decodes to "123456" - let decoded = base64_decode("MTIzNDU2"); - assert_eq!(decoded, Some("123456".to_string())); -} - -#[test] -fn bot_user_id_extraction() { - // Token format: base64(user_id).timestamp.hmac - let token = "MTIzNDU2.fake.hmac"; - let id = DiscordChannel::bot_user_id_from_token(token); - assert_eq!(id, Some("123456".to_string())); -} - -#[test] -fn empty_allowlist_allows_everyone() { - // Issue #3712: an unconfigured (empty) allowlist must apply no per-user - // restriction — otherwise a UI-connected bot silently ignores every message - // and never replies. Scope is still enforced by guild/channel filters. - let ch = DiscordChannel::new("fake".into(), None, None, vec![], false, false); - assert!(ch.is_user_allowed("12345")); - assert!(ch.is_user_allowed("anyone")); -} - -#[test] -fn wildcard_allows_everyone() { - let ch = DiscordChannel::new("fake".into(), None, None, vec!["*".into()], false, false); - assert!(ch.is_user_allowed("12345")); - assert!(ch.is_user_allowed("anyone")); -} - -#[test] -fn specific_allowlist_filters() { - let ch = DiscordChannel::new( - "fake".into(), - None, - None, - vec!["111".into(), "222".into()], - false, - false, - ); - assert!(ch.is_user_allowed("111")); - assert!(ch.is_user_allowed("222")); - assert!(!ch.is_user_allowed("333")); - assert!(!ch.is_user_allowed("unknown")); -} - -#[test] -fn allowlist_is_exact_match_not_substring() { - let ch = DiscordChannel::new("fake".into(), None, None, vec!["111".into()], false, false); - assert!(!ch.is_user_allowed("1111")); - assert!(!ch.is_user_allowed("11")); - assert!(!ch.is_user_allowed("0111")); -} - -#[test] -fn allowlist_empty_string_user_id() { - let ch = DiscordChannel::new("fake".into(), None, None, vec!["111".into()], false, false); - assert!(!ch.is_user_allowed("")); -} - -#[test] -fn allowlist_with_wildcard_and_specific() { - let ch = DiscordChannel::new( - "fake".into(), - None, - None, - vec!["111".into(), "*".into()], - false, - false, - ); - assert!(ch.is_user_allowed("111")); - assert!(ch.is_user_allowed("anyone_else")); -} - -#[test] -fn allowlist_case_sensitive() { - let ch = DiscordChannel::new("fake".into(), None, None, vec!["ABC".into()], false, false); - assert!(ch.is_user_allowed("ABC")); - assert!(!ch.is_user_allowed("abc")); - assert!(!ch.is_user_allowed("Abc")); -} - -#[test] -fn base64_decode_empty_string() { - let decoded = base64_decode(""); - assert_eq!(decoded, Some(String::new())); -} - -#[test] -fn base64_decode_invalid_chars() { - let decoded = base64_decode("!!!!"); - assert!(decoded.is_none()); -} - -#[test] -fn bot_user_id_from_empty_token() { - let id = DiscordChannel::bot_user_id_from_token(""); - assert_eq!(id, Some(String::new())); -} - -#[test] -fn contains_bot_mention_supports_plain_and_nick_forms() { - assert!(contains_bot_mention("hi <@12345>", "12345")); - assert!(contains_bot_mention("hi <@!12345>", "12345")); - assert!(!contains_bot_mention("hi <@99999>", "12345")); -} - -#[test] -fn normalize_incoming_content_requires_mention_when_enabled() { - let cleaned = normalize_incoming_content("hello there", true, "12345"); - assert!(cleaned.is_none()); -} - -#[test] -fn normalize_incoming_content_strips_mentions_and_trims() { - let cleaned = normalize_incoming_content(" <@!12345> run status ", true, "12345"); - assert_eq!(cleaned.as_deref(), Some("run status")); -} - -#[test] -fn normalize_incoming_content_rejects_empty_after_strip() { - let cleaned = normalize_incoming_content("<@12345>", true, "12345"); - assert!(cleaned.is_none()); -} - -// Message splitting tests - -#[test] -fn split_empty_message() { - let chunks = split_message_for_discord(""); - assert_eq!(chunks, vec![""]); -} - -#[test] -fn split_short_message_under_limit() { - let msg = "Hello, world!"; - let chunks = split_message_for_discord(msg); - assert_eq!(chunks, vec![msg]); -} - -#[test] -fn split_message_exactly_2000_chars() { - let msg = "a".repeat(DISCORD_MAX_MESSAGE_LENGTH); - let chunks = split_message_for_discord(&msg); - assert_eq!(chunks.len(), 1); - assert_eq!(chunks[0].chars().count(), DISCORD_MAX_MESSAGE_LENGTH); -} - -#[test] -fn split_message_just_over_limit() { - let msg = "a".repeat(DISCORD_MAX_MESSAGE_LENGTH + 1); - let chunks = split_message_for_discord(&msg); - assert_eq!(chunks.len(), 2); - assert_eq!(chunks[0].chars().count(), DISCORD_MAX_MESSAGE_LENGTH); - assert_eq!(chunks[1].chars().count(), 1); -} - -#[test] -fn split_very_long_message() { - let msg = "word ".repeat(2000); // 10000 characters (5 chars per "word ") - let chunks = split_message_for_discord(&msg); - // The shared chunker prefers whitespace boundaries, so it may produce more - // than the mathematical minimum while preserving Discord's UTF-16 limit. - assert!(chunks.len() > 1); - assert!(chunks - .iter() - .all(|chunk| discord_units(chunk) <= DISCORD_MAX_MESSAGE_LENGTH)); - // Verify total content is preserved - let reconstructed = chunks.concat(); - assert_eq!(reconstructed, msg); -} - -#[test] -fn split_prefer_newline_break() { - let msg = format!("{}\n{}", "a".repeat(1500), "b".repeat(500)); - let chunks = split_message_for_discord(&msg); - // Should split at the newline - assert_eq!(chunks.len(), 2); - assert!(chunks[1].starts_with('\n')); - assert!(chunks[1].trim_start_matches('\n').starts_with('b')); -} - -#[test] -fn split_prefer_space_break() { - let msg = format!("{} {}", "a".repeat(1500), "b".repeat(600)); - let chunks = split_message_for_discord(&msg); - assert_eq!(chunks.len(), 2); -} - -#[test] -fn split_without_good_break_points_hard_split() { - // No spaces or newlines - should hard split at 2000 - let msg = "a".repeat(5000); - let chunks = split_message_for_discord(&msg); - assert_eq!(chunks.len(), 3); - assert_eq!(chunks[0].chars().count(), DISCORD_MAX_MESSAGE_LENGTH); - assert_eq!(chunks[1].chars().count(), DISCORD_MAX_MESSAGE_LENGTH); - assert_eq!(chunks[2].chars().count(), 1000); -} - -#[test] -fn split_multiple_breaks() { - // Create a message with multiple newlines - let part1 = "a".repeat(900); - let part2 = "b".repeat(900); - let part3 = "c".repeat(900); - let msg = format!("{part1}\n{part2}\n{part3}"); - let chunks = split_message_for_discord(&msg); - // Should split into 2 chunks (first two parts + third part) - assert_eq!(chunks.len(), 2); - assert!(chunks[0].chars().count() <= DISCORD_MAX_MESSAGE_LENGTH); - assert!(chunks[1].chars().count() <= DISCORD_MAX_MESSAGE_LENGTH); -} - -#[test] -fn split_preserves_content() { - let original = "Hello world! This is a test message with some content. ".repeat(200); - let chunks = split_message_for_discord(&original); - let reconstructed = chunks.concat(); - assert_eq!(reconstructed, original); -} - -#[test] -fn split_unicode_content() { - // Test with emoji and multi-byte characters - let msg = "🦀 Rust is awesome! ".repeat(500); - let chunks = split_message_for_discord(&msg); - // All chunks should be valid UTF-8 - for chunk in &chunks { - assert!(std::str::from_utf8(chunk.as_bytes()).is_ok()); - assert!(chunk.chars().count() <= DISCORD_MAX_MESSAGE_LENGTH); - } - // Reconstruct and verify - let reconstructed = chunks.concat(); - assert_eq!(reconstructed, msg); -} - -#[test] -fn split_newline_too_close_to_end() { - // If newline is in the first half, don't use it - use space instead or hard split - let msg = format!("{}\n{}", "a".repeat(1900), "b".repeat(500)); - let chunks = split_message_for_discord(&msg); - // Should split at newline since it's in the second half of the window - assert_eq!(chunks.len(), 2); -} - -#[test] -fn split_multibyte_only_content_without_panics() { - let msg = "🦀".repeat(2500); - let chunks = split_message_for_discord(&msg); - assert_eq!(chunks.len(), 3); - assert!(chunks - .iter() - .all(|chunk| discord_units(chunk) <= DISCORD_MAX_MESSAGE_LENGTH)); - let reconstructed = chunks.concat(); - assert_eq!(reconstructed, msg); -} - -#[test] -fn split_chunks_always_within_discord_limit() { - let msg = "x".repeat(12_345); - let chunks = split_message_for_discord(&msg); - assert!(chunks - .iter() - .all(|chunk| chunk.chars().count() <= DISCORD_MAX_MESSAGE_LENGTH)); -} - -#[test] -fn split_message_with_multiple_newlines() { - let msg = "Line 1\nLine 2\nLine 3\n".repeat(1000); - let chunks = split_message_for_discord(&msg); - assert!(chunks.len() > 1); - let reconstructed = chunks.concat(); - assert_eq!(reconstructed, msg); -} - -#[test] -fn typing_handle_starts_as_none() { - let ch = DiscordChannel::new("fake".into(), None, None, vec![], false, false); - let guard = ch.typing_handle.lock(); - assert!(guard.is_none()); -} - -#[tokio::test] -async fn start_typing_sets_handle() { - let ch = DiscordChannel::new("fake".into(), None, None, vec![], false, false); - let _ = ch.start_typing("123456").await; - let guard = ch.typing_handle.lock(); - assert!(guard.is_some()); -} - -#[tokio::test] -async fn stop_typing_clears_handle() { - let ch = DiscordChannel::new("fake".into(), None, None, vec![], false, false); - let _ = ch.start_typing("123456").await; - let _ = ch.stop_typing("123456").await; - let guard = ch.typing_handle.lock(); - assert!(guard.is_none()); -} - -#[tokio::test] -async fn stop_typing_is_idempotent() { - let ch = DiscordChannel::new("fake".into(), None, None, vec![], false, false); - assert!(ch.stop_typing("123456").await.is_ok()); - assert!(ch.stop_typing("123456").await.is_ok()); -} - -#[tokio::test] -async fn start_typing_replaces_existing_task() { - let ch = DiscordChannel::new("fake".into(), None, None, vec![], false, false); - let _ = ch.start_typing("111").await; - let _ = ch.start_typing("222").await; - let guard = ch.typing_handle.lock(); - assert!(guard.is_some()); -} - -// ── Message ID edge cases ───────────────────────────────────── - -#[test] -fn discord_message_id_format_includes_discord_prefix() { - // Verify that message IDs follow the format: discord_{message_id} - let message_id = "123456789012345678"; - let expected_id = format!("discord_{message_id}"); - assert_eq!(expected_id, "discord_123456789012345678"); -} - -#[test] -fn discord_message_id_is_deterministic() { - // Same message_id = same ID (prevents duplicates after restart) - let message_id = "123456789012345678"; - let id1 = format!("discord_{message_id}"); - let id2 = format!("discord_{message_id}"); - assert_eq!(id1, id2); -} - -#[test] -fn discord_message_id_different_message_different_id() { - // Different message IDs produce different IDs - let id1 = "discord_123456789012345678".to_string(); - let id2 = "discord_987654321098765432".to_string(); - assert_ne!(id1, id2); -} - -#[test] -fn discord_message_id_uses_snowflake_id() { - // Discord snowflake IDs are numeric strings - let message_id = "123456789012345678"; // Typical snowflake format - let id = format!("discord_{message_id}"); - assert!(id.starts_with("discord_")); - // Snowflake IDs are numeric - assert!(message_id.chars().all(|c| c.is_ascii_digit())); -} - -#[test] -fn discord_message_id_fallback_to_uuid_on_empty() { - // Edge case: empty message_id falls back to UUID - let message_id = ""; - let id = if message_id.is_empty() { - format!("discord_{}", uuid::Uuid::new_v4()) - } else { - format!("discord_{message_id}") - }; - assert!(id.starts_with("discord_")); - // Should have UUID dashes - assert!(id.contains('-')); -} - -// ───────────────────────────────────────────────────────────────────── -// TG6: Channel platform limit edge cases for Discord (2000 char limit) -// Prevents: Pattern 6 — issues #574, #499 -// ───────────────────────────────────────────────────────────────────── - -#[test] -fn split_message_code_block_at_boundary() { - // Code block that spans the split boundary - let mut msg = String::new(); - msg.push_str("```rust\n"); - msg.push_str(&"x".repeat(1990)); - msg.push_str("\n```\nMore text after code block"); - let parts = split_message_for_discord(&msg); - assert!( - parts.len() >= 2, - "code block spanning boundary should split" - ); - for part in &parts { - assert!( - part.len() <= DISCORD_MAX_MESSAGE_LENGTH, - "each part must be <= {DISCORD_MAX_MESSAGE_LENGTH}, got {}", - part.len() - ); - } -} - -#[test] -fn split_message_single_long_word_exceeds_limit() { - // A single word longer than 2000 chars must be hard-split - let long_word = "a".repeat(2500); - let parts = split_message_for_discord(&long_word); - assert!(parts.len() >= 2, "word exceeding limit must be split"); - for part in &parts { - assert!( - part.len() <= DISCORD_MAX_MESSAGE_LENGTH, - "hard-split part must be <= {DISCORD_MAX_MESSAGE_LENGTH}, got {}", - part.len() - ); - } - // Reassembled content should match original - let reassembled: String = parts.join(""); - assert_eq!(reassembled, long_word); -} - -#[test] -fn split_message_exactly_at_limit_no_split() { - let msg = "a".repeat(DISCORD_MAX_MESSAGE_LENGTH); - let parts = split_message_for_discord(&msg); - assert_eq!(parts.len(), 1, "message exactly at limit should not split"); - assert_eq!(parts[0].len(), DISCORD_MAX_MESSAGE_LENGTH); -} - -#[test] -fn split_message_one_over_limit_splits() { - let msg = "a".repeat(DISCORD_MAX_MESSAGE_LENGTH + 1); - let parts = split_message_for_discord(&msg); - assert!(parts.len() >= 2, "message 1 char over limit must split"); -} - -#[test] -fn split_message_many_short_lines() { - // Many short lines should be batched into chunks under the limit - let msg: String = (0..500).map(|i| format!("line {i}\n")).collect(); - let parts = split_message_for_discord(&msg); - for part in &parts { - assert!( - part.len() <= DISCORD_MAX_MESSAGE_LENGTH, - "short-line batch must be <= limit" - ); - } - // All content should be preserved - let reassembled: String = parts.join(""); - assert_eq!(reassembled.trim(), msg.trim()); -} - -#[test] -fn split_message_only_whitespace() { - let msg = " \n\n\t "; - let parts = split_message_for_discord(msg); - // Should handle gracefully without panic - assert!(parts.len() <= 1); -} - -#[test] -fn split_message_emoji_at_boundary() { - // Emoji are multi-byte; ensure we don't split mid-emoji - let mut msg = "a".repeat(1998); - msg.push_str("🎉🎊"); // 2 emoji at the boundary (2000 chars total) - let parts = split_message_for_discord(&msg); - for part in &parts { - // The function splits on character count, not byte count - assert!( - part.chars().count() <= DISCORD_MAX_MESSAGE_LENGTH, - "emoji boundary split must respect limit" - ); - } -} - -#[test] -fn split_message_consecutive_newlines_at_boundary() { - let mut msg = "a".repeat(1995); - msg.push_str("\n\n\n\n\n"); - msg.push_str(&"b".repeat(100)); - let parts = split_message_for_discord(&msg); - for part in &parts { - assert!(part.len() <= DISCORD_MAX_MESSAGE_LENGTH); - } -} - -// ── channel_id field tests ─────────────────────────────────── - -#[test] -fn channel_id_stored_in_struct() { - let ch = DiscordChannel::new( - "token".into(), - Some("guild1".into()), - Some("channel1".into()), - vec![], - false, - false, - ); - assert_eq!(ch.channel_id.as_deref(), Some("channel1")); - assert_eq!(ch.guild_id.as_deref(), Some("guild1")); -} - -#[test] -fn channel_id_defaults_to_none() { - let ch = DiscordChannel::new("token".into(), None, None, vec![], false, false); - assert!(ch.channel_id.is_none()); -} - -#[test] -fn passes_guild_scope_covers_guild_dm_and_unscoped_cases() { - // No configured guild → everything passes (filter inactive). - assert!(DiscordChannel::passes_guild_scope(None, Some("g1"), true)); - assert!(DiscordChannel::passes_guild_scope(None, None, true)); - // Configured guild: same guild passes, other guild blocked. - assert!(DiscordChannel::passes_guild_scope( - Some("g1"), - Some("g1"), - true - )); - assert!(!DiscordChannel::passes_guild_scope( - Some("g1"), - Some("g2"), - true - )); - // #3794 review (Codex P1): DM (no guild_id) under guild scope is blocked - // with a blank allowlist, allowed with an explicit one. - assert!(!DiscordChannel::passes_guild_scope(Some("g1"), None, true)); - assert!(DiscordChannel::passes_guild_scope(Some("g1"), None, false)); -} - -#[test] -fn resolve_recipient_prefers_explicit_then_configured_channel() { - // #3794 review (Codex P2): recipient-less proactive sends fall back to the - // configured channel_id; an explicit recipient always wins. - assert_eq!( - DiscordChannel::resolve_recipient("123", Some("999")), - Some("123") - ); - assert_eq!( - DiscordChannel::resolve_recipient("", Some("999")), - Some("999") - ); - // Neither available → None, so the caller errors instead of POSTing to "". - assert_eq!(DiscordChannel::resolve_recipient("", None), None); - assert_eq!(DiscordChannel::resolve_recipient("", Some("")), None); -} - -#[test] -fn proactive_target_uses_configured_channel_id() { - use crate::openhuman::channels::traits::Channel; - - // Configured channel_id ⇒ recipient-less proactive sends have a target. - let with_channel = DiscordChannel::new( - "fake".into(), - None, - Some("12345".into()), - vec![], - false, - false, - ); - assert_eq!(with_channel.proactive_target(), Some("12345".to_string())); - - // No channel_id ⇒ None, so proactive routing skips Discord (#3794 Codex P2). - let no_channel = DiscordChannel::new("fake".into(), None, None, vec![], false, false); - assert_eq!(no_channel.proactive_target(), None); - - // Whitespace-only channel_id is treated as unset. - let blank_channel = DiscordChannel::new( - "fake".into(), - None, - Some(" ".into()), - vec![], - false, - false, - ); - assert_eq!(blank_channel.proactive_target(), None); -} diff --git a/src/openhuman/channels/providers/discord/mod.rs b/src/openhuman/channels/providers/discord/mod.rs deleted file mode 100644 index 4f87d61dc8..0000000000 --- a/src/openhuman/channels/providers/discord/mod.rs +++ /dev/null @@ -1,4 +0,0 @@ -pub mod api; -pub mod channel; - -pub use channel::DiscordChannel; diff --git a/src/openhuman/channels/providers/email_channel.rs b/src/openhuman/channels/providers/email_channel.rs index d11265c069..e602b0d5ea 100644 --- a/src/openhuman/channels/providers/email_channel.rs +++ b/src/openhuman/channels/providers/email_channel.rs @@ -1,553 +1 @@ -#![allow(clippy::uninlined_format_args)] -#![allow(clippy::map_unwrap_or)] -#![allow(clippy::redundant_closure_for_method_calls)] -#![allow(clippy::cast_lossless)] -#![allow(clippy::trim_split_whitespace)] -#![allow(clippy::doc_link_with_quotes)] -#![allow(clippy::doc_markdown)] -#![allow(clippy::too_many_lines)] -#![allow(clippy::unnecessary_map_or)] - -use anyhow::{anyhow, Result}; -use async_imap::extensions::idle::IdleResponse; -use async_imap::types::Fetch; -use async_imap::Session; -use async_trait::async_trait; -use futures::TryStreamExt; -use lettre::message::{header::ContentType, Attachment, MultiPart, SinglePart}; -use lettre::transport::smtp::authentication::Credentials; -use lettre::{Message, SmtpTransport, Transport}; -use mail_parser::{MessageParser, MimeHeaders}; -use rustls::{ClientConfig, RootCertStore}; -use rustls_pki_types::DnsName; -use std::collections::HashSet; -use std::sync::Arc; -use std::time::{Duration, SystemTime, UNIX_EPOCH}; -use tokio::net::TcpStream; -use tokio::sync::{mpsc, Mutex}; -use tokio::time::{sleep, timeout}; -use tokio_rustls::client::TlsStream; -use tokio_rustls::TlsConnector; -use tracing::{debug, error, info, warn}; -use uuid::Uuid; - -use crate::openhuman::channels::traits::{Channel, ChannelMessage, SendMessage}; -pub use crate::openhuman::config::schema::EmailConfig; - -type ImapSession = Session>; - -/// Email channel — IMAP IDLE for instant push notifications, SMTP for outbound -pub struct EmailChannel { - pub config: EmailConfig, - seen_messages: Arc>>, -} - -impl EmailChannel { - pub fn new(config: EmailConfig) -> Self { - Self { - config, - seen_messages: Arc::new(Mutex::new(HashSet::new())), - } - } - - /// Check if a sender email is in the allowlist - pub fn is_sender_allowed(&self, email: &str) -> bool { - if self.config.allowed_senders.is_empty() { - return false; // Empty = deny all - } - if self.config.allowed_senders.iter().any(|a| a == "*") { - return true; // Wildcard = allow all - } - let email_lower = email.to_lowercase(); - self.config.allowed_senders.iter().any(|allowed| { - if allowed.starts_with('@') { - // Domain match with @ prefix: "@example.com" - email_lower.ends_with(&allowed.to_lowercase()) - } else if allowed.contains('@') { - // Full email address match - allowed.eq_ignore_ascii_case(email) - } else { - // Domain match without @ prefix: "example.com" - email_lower.ends_with(&format!("@{}", allowed.to_lowercase())) - } - }) - } - - /// Strip HTML tags from content (basic) - pub fn strip_html(html: &str) -> String { - let mut result = String::new(); - let mut in_tag = false; - for ch in html.chars() { - match ch { - '<' => in_tag = true, - '>' => in_tag = false, - _ if !in_tag => result.push(ch), - _ => {} - } - } - let mut normalized = String::with_capacity(result.len()); - for word in result.split_whitespace() { - if !normalized.is_empty() { - normalized.push(' '); - } - normalized.push_str(word); - } - normalized - } - - /// Extract the sender address from a parsed email - fn extract_sender(parsed: &mail_parser::Message) -> String { - parsed - .from() - .and_then(|addr| addr.first()) - .and_then(|a| a.address()) - .map(|s| s.to_string()) - .unwrap_or_else(|| "unknown".into()) - } - - /// Extract readable text from a parsed email - fn extract_text(parsed: &mail_parser::Message) -> String { - if let Some(text) = parsed.body_text(0) { - return text.to_string(); - } - if let Some(html) = parsed.body_html(0) { - return Self::strip_html(html.as_ref()); - } - for part in parsed.attachments() { - let part: &mail_parser::MessagePart = part; - if let Some(ct) = MimeHeaders::content_type(part) { - if ct.ctype() == "text" { - if let Ok(text) = std::str::from_utf8(part.contents()) { - let name = MimeHeaders::attachment_name(part).unwrap_or("file"); - return format!("[Attachment: {}]\n{}", name, text); - } - } - } - } - "(no readable content)".to_string() - } - - /// Connect to IMAP server with TLS and authenticate - async fn connect_imap(&self) -> Result { - let addr = format!("{}:{}", self.config.imap_host, self.config.imap_port); - debug!("Connecting to IMAP server at {}", addr); - - // Connect TCP - let tcp = TcpStream::connect(&addr).await?; - - // Establish TLS using rustls - let certs = RootCertStore { - roots: webpki_roots::TLS_SERVER_ROOTS.into(), - }; - let config = ClientConfig::builder() - .with_root_certificates(certs) - .with_no_client_auth(); - let tls_stream: TlsConnector = Arc::new(config).into(); - let sni: DnsName = self.config.imap_host.clone().try_into()?; - let stream = tls_stream.connect(sni.into(), tcp).await?; - - // Create IMAP client - let client = async_imap::Client::new(stream); - - // Login - let session = client - .login(&self.config.username, &self.config.password) - .await - .map_err(|(e, _)| anyhow!("IMAP login failed: {}", e))?; - - debug!("IMAP login successful"); - Ok(session) - } - - /// Fetch and process unseen messages from the selected mailbox - async fn fetch_unseen(&self, session: &mut ImapSession) -> Result> { - // Search for unseen messages - let uids = session.uid_search("UNSEEN").await?; - if uids.is_empty() { - return Ok(Vec::new()); - } - - debug!("Found {} unseen messages", uids.len()); - - let mut results = Vec::new(); - let uid_set: String = uids - .iter() - .map(|u| u.to_string()) - .collect::>() - .join(","); - - // Fetch message bodies - let messages = session.uid_fetch(&uid_set, "RFC822").await?; - let messages: Vec = messages.try_collect().await?; - - for msg in messages { - let uid = msg.uid.unwrap_or(0); - if let Some(body) = msg.body() { - if let Some(parsed) = MessageParser::default().parse(body) { - let sender = Self::extract_sender(&parsed); - let subject = parsed.subject().unwrap_or("(no subject)").to_string(); - let body_text = Self::extract_text(&parsed); - let content = format!("Subject: {}\n\n{}", subject, body_text); - let msg_id = parsed - .message_id() - .map(|s| s.to_string()) - .unwrap_or_else(|| format!("gen-{}", Uuid::new_v4())); - - #[allow(clippy::cast_sign_loss)] - let ts = parsed - .date() - .map(|d| { - let naive = chrono::NaiveDate::from_ymd_opt( - d.year as i32, - u32::from(d.month), - u32::from(d.day), - ) - .and_then(|date| { - date.and_hms_opt( - u32::from(d.hour), - u32::from(d.minute), - u32::from(d.second), - ) - }); - naive.map_or(0, |n| n.and_utc().timestamp() as u64) - }) - .unwrap_or_else(|| { - SystemTime::now() - .duration_since(UNIX_EPOCH) - .map(|d| d.as_secs()) - .unwrap_or(0) - }); - - results.push(ParsedEmail { - _uid: uid, - msg_id, - sender, - content, - timestamp: ts, - }); - } - } - } - - // Mark fetched messages as seen - if !results.is_empty() { - let _ = session - .uid_store(&uid_set, "+FLAGS (\\Seen)") - .await? - .try_collect::>() - .await; - } - - Ok(results) - } - - /// Run the IDLE loop, returning when a new message arrives or timeout - /// Note: IDLE consumes the session and returns it via done() - async fn wait_for_changes( - &self, - session: ImapSession, - ) -> Result<(IdleWaitResult, ImapSession)> { - let idle_timeout = Duration::from_secs(self.config.idle_timeout_secs); - - // Start IDLE mode - this consumes the session - let mut idle = session.idle(); - idle.init().await?; - - debug!("Entering IMAP IDLE mode"); - - // wait() returns (future, stop_source) - we only need the future - let (wait_future, _stop_source) = idle.wait(); - - // Wait for server notification or timeout - let result = timeout(idle_timeout, wait_future).await; - - match result { - Ok(Ok(response)) => { - debug!("IDLE response: {:?}", response); - // Done with IDLE, return session to normal mode - let session = idle.done().await?; - let wait_result = match response { - IdleResponse::NewData(_) => IdleWaitResult::NewMail, - IdleResponse::Timeout => IdleWaitResult::Timeout, - IdleResponse::ManualInterrupt => IdleWaitResult::Interrupted, - }; - Ok((wait_result, session)) - } - Ok(Err(e)) => { - // Try to clean up IDLE state - let _ = idle.done().await; - Err(anyhow!("IDLE error: {}", e)) - } - Err(_) => { - // Timeout - RFC 2177 recommends restarting IDLE every 29 minutes - debug!("IDLE timeout reached, will re-establish"); - let session = idle.done().await?; - Ok((IdleWaitResult::Timeout, session)) - } - } - } - - /// Main IDLE-based listen loop with automatic reconnection - async fn listen_with_idle(&self, tx: mpsc::Sender) -> Result<()> { - let mut backoff = Duration::from_secs(1); - let max_backoff = Duration::from_secs(60); - - loop { - match self.run_idle_session(&tx).await { - Ok(()) => { - // Clean exit (channel closed) - return Ok(()); - } - Err(e) => { - error!( - "IMAP session error: {}. Reconnecting in {:?}...", - e, backoff - ); - sleep(backoff).await; - // Exponential backoff with cap - backoff = std::cmp::min(backoff * 2, max_backoff); - } - } - } - } - - /// Run a single IDLE session until error or clean shutdown - async fn run_idle_session(&self, tx: &mpsc::Sender) -> Result<()> { - // Connect and authenticate - let mut session = self.connect_imap().await?; - - // Select the mailbox - session.select(&self.config.imap_folder).await?; - info!( - "Email IDLE listening on {} (instant push enabled)", - self.config.imap_folder - ); - - // Check for existing unseen messages first - self.process_unseen(&mut session, tx).await?; - - loop { - // Enter IDLE and wait for changes (consumes session, returns it via result) - match self.wait_for_changes(session).await { - Ok((IdleWaitResult::NewMail, returned_session)) => { - debug!("New mail notification received"); - session = returned_session; - self.process_unseen(&mut session, tx).await?; - } - Ok((IdleWaitResult::Timeout, returned_session)) => { - // Re-check for mail after IDLE timeout (defensive) - session = returned_session; - self.process_unseen(&mut session, tx).await?; - } - Ok((IdleWaitResult::Interrupted, _)) => { - info!("IDLE interrupted, exiting"); - return Ok(()); - } - Err(e) => { - // Connection likely broken, need to reconnect - return Err(e); - } - } - } - } - - /// Fetch unseen messages and send to channel - async fn process_unseen( - &self, - session: &mut ImapSession, - tx: &mpsc::Sender, - ) -> Result<()> { - let messages = self.fetch_unseen(session).await?; - - for email in messages { - // Check allowlist - if !self.is_sender_allowed(&email.sender) { - warn!("Blocked email from {}", email.sender); - continue; - } - - let is_new = { - let mut seen = self.seen_messages.lock().await; - seen.insert(email.msg_id.clone()) - }; - if !is_new { - continue; - } - - let msg = ChannelMessage { - id: email.msg_id, - reply_target: email.sender.clone(), - sender: email.sender, - content: email.content, - channel: "email".to_string(), - timestamp: email.timestamp, - thread_ts: None, - }; - - if tx.send(msg).await.is_err() { - // Channel closed, exit cleanly - return Ok(()); - } - } - - Ok(()) - } - - fn create_smtp_transport(&self) -> Result { - let creds = Credentials::new(self.config.username.clone(), self.config.password.clone()); - let transport = if self.config.smtp_tls { - SmtpTransport::relay(&self.config.smtp_host)? - .port(self.config.smtp_port) - .credentials(creds) - .build() - } else { - SmtpTransport::builder_dangerous(&self.config.smtp_host) - .port(self.config.smtp_port) - .credentials(creds) - .build() - }; - Ok(transport) - } - - pub fn send_message(&self, email: Message) -> Result<()> { - let transport = self.create_smtp_transport()?; - transport.send(&email)?; - info!("Email sent"); - Ok(()) - } - - pub fn build_plain_message( - &self, - recipient: &str, - subject: &str, - body: &str, - ) -> Result { - Message::builder() - .from(self.config.from_address.parse()?) - .to(recipient.parse()?) - .subject(subject) - .singlepart(SinglePart::plain(body.to_string())) - .map_err(Into::into) - } - - pub fn build_message_with_attachment( - &self, - recipient: &str, - subject: &str, - body: &str, - attachment_name: &str, - content_type: ContentType, - attachment_bytes: Vec, - ) -> Result { - let attachment = - Attachment::new(attachment_name.to_string()).body(attachment_bytes, content_type); - Message::builder() - .from(self.config.from_address.parse()?) - .to(recipient.parse()?) - .subject(subject) - .multipart( - MultiPart::mixed() - .singlepart(SinglePart::plain(body.to_string())) - .singlepart(attachment), - ) - .map_err(Into::into) - } -} - -/// Internal struct for parsed email data -struct ParsedEmail { - _uid: u32, - msg_id: String, - sender: String, - content: String, - timestamp: u64, -} - -/// Result from waiting on IDLE -enum IdleWaitResult { - NewMail, - Timeout, - Interrupted, -} - -#[async_trait] -impl Channel for EmailChannel { - fn name(&self) -> &str { - "email" - } - - async fn send(&self, message: &SendMessage) -> Result<()> { - // Use explicit subject if provided, otherwise fall back to legacy parsing or default - let (subject, body) = if let Some(ref subj) = message.subject { - (subj.as_str(), message.content.as_str()) - } else if message.content.starts_with("Subject: ") { - if let Some(pos) = message.content.find('\n') { - (&message.content[9..pos], message.content[pos + 1..].trim()) - } else { - ("OpenHuman Message", message.content.as_str()) - } - } else { - ("OpenHuman Message", message.content.as_str()) - }; - - let email = self.build_plain_message(message.recipient.as_str(), subject, body)?; - self.send_message(email)?; - info!("Email sent to {}", message.recipient); - Ok(()) - } - - async fn listen(&self, tx: mpsc::Sender) -> Result<()> { - info!( - "Starting email channel with IDLE support on {}", - self.config.imap_folder - ); - self.listen_with_idle(tx).await - } - - async fn health_check(&self) -> bool { - // Fully async health check - attempt IMAP connection - match timeout(Duration::from_secs(10), self.connect_imap()).await { - Ok(Ok(mut session)) => { - // Try to logout cleanly - let _ = session.logout().await; - true - } - Ok(Err(e)) => { - debug!("Health check failed: {}", e); - false - } - Err(_) => { - debug!("Health check timed out"); - false - } - } - } -} - -#[cfg(test)] -#[path = "email_channel_tests.rs"] -mod tests; - -#[cfg(any(test, debug_assertions))] -pub mod test_support { - //! Debug-build helpers for raw integration tests. They exercise the email - //! parser without opening IMAP or SMTP sockets. - - use super::*; - - #[derive(Debug, Clone, PartialEq, Eq)] - pub struct ParsedEmailFixture { - pub sender: String, - pub text: String, - pub subject: Option, - } - - pub fn parse_email_fixture(raw: &[u8]) -> Option { - let parsed = MessageParser::default().parse(raw)?; - Some(ParsedEmailFixture { - sender: EmailChannel::extract_sender(&parsed), - text: EmailChannel::extract_text(&parsed), - subject: parsed.subject().map(str::to_string), - }) - } -} +pub use tinychannels::providers::email_channel::*; diff --git a/src/openhuman/channels/providers/email_channel_tests.rs b/src/openhuman/channels/providers/email_channel_tests.rs deleted file mode 100644 index f971e82a12..0000000000 --- a/src/openhuman/channels/providers/email_channel_tests.rs +++ /dev/null @@ -1,535 +0,0 @@ -use super::*; - -#[test] -fn default_smtp_port_uses_tls_port() { - assert_eq!(EmailConfig::default().smtp_port, 465); -} - -#[test] -fn email_config_default_uses_tls_smtp_defaults() { - let config = EmailConfig::default(); - assert_eq!(config.smtp_port, 465); - assert!(config.smtp_tls); -} - -#[test] -fn default_idle_timeout_is_29_minutes() { - assert_eq!(EmailConfig::default().idle_timeout_secs, 1740); -} - -#[tokio::test] -async fn seen_messages_starts_empty() { - let channel = EmailChannel::new(EmailConfig::default()); - let seen = channel.seen_messages.lock().await; - assert!(seen.is_empty()); -} - -#[tokio::test] -async fn seen_messages_tracks_unique_ids() { - let channel = EmailChannel::new(EmailConfig::default()); - let mut seen = channel.seen_messages.lock().await; - - assert!(seen.insert("first-id".to_string())); - assert!(!seen.insert("first-id".to_string())); - assert!(seen.insert("second-id".to_string())); - assert_eq!(seen.len(), 2); -} - -// EmailConfig tests - -#[test] -fn email_config_default() { - let config = EmailConfig::default(); - assert_eq!(config.imap_host, ""); - assert_eq!(config.imap_port, 993); - assert_eq!(config.imap_folder, "INBOX"); - assert_eq!(config.smtp_host, ""); - assert_eq!(config.smtp_port, 465); - assert!(config.smtp_tls); - assert_eq!(config.username, ""); - assert_eq!(config.password, ""); - assert_eq!(config.from_address, ""); - assert_eq!(config.idle_timeout_secs, 1740); - assert!(config.allowed_senders.is_empty()); -} - -#[test] -fn email_config_custom() { - let config = EmailConfig { - imap_host: "imap.example.com".to_string(), - imap_port: 993, - imap_folder: "Archive".to_string(), - smtp_host: "smtp.example.com".to_string(), - smtp_port: 465, - smtp_tls: true, - username: "user@example.com".to_string(), - password: "pass123".to_string(), - from_address: "bot@example.com".to_string(), - idle_timeout_secs: 1200, - allowed_senders: vec!["allowed@example.com".to_string()], - }; - assert_eq!(config.imap_host, "imap.example.com"); - assert_eq!(config.imap_folder, "Archive"); - assert_eq!(config.idle_timeout_secs, 1200); -} - -#[test] -fn email_config_clone() { - let config = EmailConfig { - imap_host: "imap.test.com".to_string(), - imap_port: 993, - imap_folder: "INBOX".to_string(), - smtp_host: "smtp.test.com".to_string(), - smtp_port: 587, - smtp_tls: true, - username: "user@test.com".to_string(), - password: "secret".to_string(), - from_address: "bot@test.com".to_string(), - idle_timeout_secs: 1740, - allowed_senders: vec!["*".to_string()], - }; - let cloned = config.clone(); - assert_eq!(cloned.imap_host, config.imap_host); - assert_eq!(cloned.smtp_port, config.smtp_port); - assert_eq!(cloned.allowed_senders, config.allowed_senders); -} - -// EmailChannel tests - -#[tokio::test] -async fn email_channel_new() { - let config = EmailConfig::default(); - let channel = EmailChannel::new(config.clone()); - assert_eq!(channel.config.imap_host, config.imap_host); - - let seen_guard = channel.seen_messages.lock().await; - assert_eq!(seen_guard.len(), 0); -} - -#[test] -fn email_channel_name() { - let channel = EmailChannel::new(EmailConfig::default()); - assert_eq!(channel.name(), "email"); -} - -#[test] -fn build_plain_message_uses_subject_and_body() { - let channel = EmailChannel::new(EmailConfig { - from_address: "bot@example.com".to_string(), - ..Default::default() - }); - let message = channel - .build_plain_message("listener@example.com", "Podcast", "Hello there") - .expect("plain message"); - let wire = String::from_utf8_lossy(&message.formatted()).to_string(); - assert!(wire.contains("Subject: Podcast")); - assert!(wire.contains("Hello there")); -} - -#[test] -fn build_message_with_attachment_adds_audio_part() { - let channel = EmailChannel::new(EmailConfig { - from_address: "bot@example.com".to_string(), - ..Default::default() - }); - let message = channel - .build_message_with_attachment( - "listener@example.com", - "Weekly briefing", - "Attached.", - "briefing.mp3", - "audio/mpeg".parse().expect("content type"), - vec![1, 2, 3, 4], - ) - .expect("attachment message"); - let wire = String::from_utf8_lossy(&message.formatted()).to_string(); - assert!(wire.contains("Subject: Weekly briefing")); - assert!(wire.contains("filename=\"briefing.mp3\"")); - assert!(wire.contains("Content-Type: audio/mpeg")); -} - -// is_sender_allowed tests - -#[test] -fn is_sender_allowed_empty_list_denies_all() { - let config = EmailConfig { - allowed_senders: vec![], - ..Default::default() - }; - let channel = EmailChannel::new(config); - assert!(!channel.is_sender_allowed("anyone@example.com")); - assert!(!channel.is_sender_allowed("user@test.com")); -} - -#[test] -fn is_sender_allowed_wildcard_allows_all() { - let config = EmailConfig { - allowed_senders: vec!["*".to_string()], - ..Default::default() - }; - let channel = EmailChannel::new(config); - assert!(channel.is_sender_allowed("anyone@example.com")); - assert!(channel.is_sender_allowed("user@test.com")); - assert!(channel.is_sender_allowed("random@domain.org")); -} - -#[test] -fn is_sender_allowed_specific_email() { - let config = EmailConfig { - allowed_senders: vec!["allowed@example.com".to_string()], - ..Default::default() - }; - let channel = EmailChannel::new(config); - assert!(channel.is_sender_allowed("allowed@example.com")); - assert!(!channel.is_sender_allowed("other@example.com")); - assert!(!channel.is_sender_allowed("allowed@other.com")); -} - -#[test] -fn is_sender_allowed_domain_with_at_prefix() { - let config = EmailConfig { - allowed_senders: vec!["@example.com".to_string()], - ..Default::default() - }; - let channel = EmailChannel::new(config); - assert!(channel.is_sender_allowed("user@example.com")); - assert!(channel.is_sender_allowed("admin@example.com")); - assert!(!channel.is_sender_allowed("user@other.com")); -} - -#[test] -fn is_sender_allowed_domain_without_at_prefix() { - let config = EmailConfig { - allowed_senders: vec!["example.com".to_string()], - ..Default::default() - }; - let channel = EmailChannel::new(config); - assert!(channel.is_sender_allowed("user@example.com")); - assert!(channel.is_sender_allowed("admin@example.com")); - assert!(!channel.is_sender_allowed("user@other.com")); -} - -#[test] -fn is_sender_allowed_case_insensitive() { - let config = EmailConfig { - allowed_senders: vec!["Allowed@Example.COM".to_string()], - ..Default::default() - }; - let channel = EmailChannel::new(config); - assert!(channel.is_sender_allowed("allowed@example.com")); - assert!(channel.is_sender_allowed("ALLOWED@EXAMPLE.COM")); - assert!(channel.is_sender_allowed("AlLoWeD@eXaMpLe.cOm")); -} - -#[test] -fn is_sender_allowed_multiple_senders() { - let config = EmailConfig { - allowed_senders: vec![ - "user1@example.com".to_string(), - "user2@test.com".to_string(), - "@allowed.com".to_string(), - ], - ..Default::default() - }; - let channel = EmailChannel::new(config); - assert!(channel.is_sender_allowed("user1@example.com")); - assert!(channel.is_sender_allowed("user2@test.com")); - assert!(channel.is_sender_allowed("anyone@allowed.com")); - assert!(!channel.is_sender_allowed("user3@example.com")); -} - -#[test] -fn is_sender_allowed_wildcard_with_specific() { - let config = EmailConfig { - allowed_senders: vec!["*".to_string(), "specific@example.com".to_string()], - ..Default::default() - }; - let channel = EmailChannel::new(config); - assert!(channel.is_sender_allowed("anyone@example.com")); - assert!(channel.is_sender_allowed("specific@example.com")); -} - -#[test] -fn is_sender_allowed_empty_sender() { - let config = EmailConfig { - allowed_senders: vec!["@example.com".to_string()], - ..Default::default() - }; - let channel = EmailChannel::new(config); - assert!(!channel.is_sender_allowed("")); - // "@example.com" ends with "@example.com" so it's allowed - assert!(channel.is_sender_allowed("@example.com")); -} - -// strip_html tests - -#[test] -fn strip_html_basic() { - assert_eq!(EmailChannel::strip_html("

Hello

"), "Hello"); - assert_eq!(EmailChannel::strip_html("
World
"), "World"); -} - -#[test] -fn strip_html_nested_tags() { - assert_eq!( - EmailChannel::strip_html("

Hello World

"), - "Hello World" - ); -} - -#[test] -fn strip_html_multiple_lines() { - let html = "
\n

Line 1

\n

Line 2

\n
"; - assert_eq!(EmailChannel::strip_html(html), "Line 1 Line 2"); -} - -#[test] -fn strip_html_preserves_text() { - assert_eq!(EmailChannel::strip_html("No tags here"), "No tags here"); - assert_eq!(EmailChannel::strip_html(""), ""); -} - -#[test] -fn strip_html_handles_malformed() { - assert_eq!(EmailChannel::strip_html("

Unclosed"), "Unclosed"); - // The function removes everything between < and >, so "Text>with>brackets" becomes "Textwithbrackets" - assert_eq!( - EmailChannel::strip_html("Text>with>brackets"), - "Textwithbrackets" - ); -} - -#[test] -fn strip_html_self_closing_tags() { - // Self-closing tags are removed but don't add spaces - assert_eq!(EmailChannel::strip_html("Hello
World"), "HelloWorld"); - assert_eq!(EmailChannel::strip_html("Text

More"), "TextMore"); -} - -#[test] -fn strip_html_attributes_preserved() { - assert_eq!( - EmailChannel::strip_html("Link"), - "Link" - ); -} - -#[test] -fn strip_html_multiple_spaces_collapsed() { - assert_eq!( - EmailChannel::strip_html("

Word

Word

"), - "Word Word" - ); -} - -#[test] -fn strip_html_special_characters() { - assert_eq!( - EmailChannel::strip_html("<tag>"), - "<tag>" - ); -} - -// Default function tests - -#[test] -fn default_imap_port_returns_993() { - assert_eq!(EmailConfig::default().imap_port, 993); -} - -#[test] -fn default_smtp_port_returns_465() { - assert_eq!(EmailConfig::default().smtp_port, 465); -} - -#[test] -fn default_imap_folder_returns_inbox() { - assert_eq!(EmailConfig::default().imap_folder, "INBOX"); -} - -#[test] -fn default_true_returns_true() { - assert!(EmailConfig::default().smtp_tls); -} - -// EmailConfig serialization tests - -#[test] -fn email_config_serialize_deserialize() { - let config = EmailConfig { - imap_host: "imap.example.com".to_string(), - imap_port: 993, - imap_folder: "INBOX".to_string(), - smtp_host: "smtp.example.com".to_string(), - smtp_port: 587, - smtp_tls: true, - username: "user@example.com".to_string(), - password: "password123".to_string(), - from_address: "bot@example.com".to_string(), - idle_timeout_secs: 1740, - allowed_senders: vec!["allowed@example.com".to_string()], - }; - - let json = serde_json::to_string(&config).unwrap(); - let deserialized: EmailConfig = serde_json::from_str(&json).unwrap(); - - assert_eq!(deserialized.imap_host, config.imap_host); - assert_eq!(deserialized.smtp_port, config.smtp_port); - assert_eq!(deserialized.allowed_senders, config.allowed_senders); -} - -#[test] -fn email_config_deserialize_with_defaults() { - let json = r#"{ - "imap_host": "imap.test.com", - "smtp_host": "smtp.test.com", - "username": "user", - "password": "pass", - "from_address": "bot@test.com" - }"#; - - let config: EmailConfig = serde_json::from_str(json).unwrap(); - assert_eq!(config.imap_port, 993); // default - assert_eq!(config.smtp_port, 465); // default - assert!(config.smtp_tls); // default - assert_eq!(config.idle_timeout_secs, 1740); // default -} - -#[test] -fn idle_timeout_deserializes_explicit_value() { - let json = r#"{ - "imap_host": "imap.test.com", - "smtp_host": "smtp.test.com", - "username": "user", - "password": "pass", - "from_address": "bot@test.com", - "idle_timeout_secs": 900 - }"#; - let config: EmailConfig = serde_json::from_str(json).unwrap(); - assert_eq!(config.idle_timeout_secs, 900); -} - -#[test] -fn idle_timeout_deserializes_legacy_poll_interval_alias() { - let json = r#"{ - "imap_host": "imap.test.com", - "smtp_host": "smtp.test.com", - "username": "user", - "password": "pass", - "from_address": "bot@test.com", - "poll_interval_secs": 120 - }"#; - let config: EmailConfig = serde_json::from_str(json).unwrap(); - assert_eq!(config.idle_timeout_secs, 120); -} - -#[test] -fn idle_timeout_propagates_to_channel() { - let config = EmailConfig { - idle_timeout_secs: 600, - ..Default::default() - }; - let channel = EmailChannel::new(config); - assert_eq!(channel.config.idle_timeout_secs, 600); -} - -#[test] -fn email_config_debug_output() { - let config = EmailConfig { - imap_host: "imap.debug.com".to_string(), - ..Default::default() - }; - let debug_str = format!("{:?}", config); - assert!(debug_str.contains("imap.debug.com")); -} - -// ── is_sender_allowed comprehensive matrix ───────────────────── - -fn channel_with_allowlist(allowlist: Vec) -> EmailChannel { - let cfg = EmailConfig { - imap_host: "imap.x".into(), - imap_port: 993, - imap_folder: "INBOX".into(), - smtp_host: "smtp.x".into(), - smtp_port: 465, - smtp_tls: true, - username: "u".into(), - password: "p".into(), - from_address: "me@x".into(), - idle_timeout_secs: 300, - allowed_senders: allowlist, - }; - EmailChannel::new(cfg) -} - -#[test] -fn is_sender_allowed_empty_denies_all() { - let ch = channel_with_allowlist(vec![]); - assert!(!ch.is_sender_allowed("anyone@any.com")); -} - -#[test] -fn is_sender_allowed_wildcard_allows_everyone() { - let ch = channel_with_allowlist(vec!["*".into()]); - assert!(ch.is_sender_allowed("anyone@any.com")); - assert!(ch.is_sender_allowed("other@different.com")); -} - -#[test] -fn is_sender_allowed_full_email_exact_match_case_insensitive() { - let ch = channel_with_allowlist(vec!["alice@example.com".into()]); - assert!(ch.is_sender_allowed("alice@example.com")); - assert!(ch.is_sender_allowed("ALICE@EXAMPLE.COM")); - assert!(!ch.is_sender_allowed("bob@example.com")); -} - -#[test] -fn is_sender_allowed_at_prefix_domain_match() { - let ch = channel_with_allowlist(vec!["@trusted.com".into()]); - assert!(ch.is_sender_allowed("user@trusted.com")); - assert!(ch.is_sender_allowed("other@Trusted.com")); - assert!(!ch.is_sender_allowed("user@untrusted.com")); -} - -#[test] -fn is_sender_allowed_bare_domain_match_is_case_insensitive() { - let ch = channel_with_allowlist(vec!["trusted.com".into()]); - assert!(ch.is_sender_allowed("user@trusted.com")); - assert!(ch.is_sender_allowed("USER@TRUSTED.COM")); - assert!(!ch.is_sender_allowed("user@other.com")); -} - -#[test] -fn is_sender_allowed_prevents_subdomain_confusion() { - // "trusted.com" must NOT match "user@malicioustrusted.com" - let ch = channel_with_allowlist(vec!["trusted.com".into()]); - assert!(!ch.is_sender_allowed("user@notmytrusted.com")); - assert!(!ch.is_sender_allowed("user@trusted.com.evil.com")); -} - -// ── strip_html edge cases ────────────────────────────────────── - -#[test] -fn strip_html_empty_string() { - assert_eq!(EmailChannel::strip_html(""), ""); -} - -#[test] -fn strip_html_only_tags() { - assert_eq!(EmailChannel::strip_html("


"), ""); -} - -#[test] -fn strip_html_unclosed_tag_eats_rest_until_gt() { - // A '<' without '>' enters tag mode; anything after until a '>' is - // discarded. This is the implementation's behaviour — lock it in. - assert_eq!(EmailChannel::strip_html("beforehello

\n\n\n

world

"), - "hello world" - ); -} diff --git a/src/openhuman/channels/providers/imessage.rs b/src/openhuman/channels/providers/imessage.rs index b9ae06f58f..50d431139c 100644 --- a/src/openhuman/channels/providers/imessage.rs +++ b/src/openhuman/channels/providers/imessage.rs @@ -1,318 +1 @@ -use crate::openhuman::channels::traits::{Channel, ChannelMessage, SendMessage}; -use async_trait::async_trait; -use directories::UserDirs; -use rusqlite::{Connection, OpenFlags}; -use std::path::Path; -use tokio::sync::mpsc; - -/// iMessage channel using macOS `AppleScript` bridge. -/// Polls the Messages database for new messages and sends replies via `osascript`. -#[derive(Clone)] -pub struct IMessageChannel { - allowed_contacts: Vec, - poll_interval_secs: u64, -} - -impl IMessageChannel { - pub fn new(allowed_contacts: Vec) -> Self { - Self { - allowed_contacts, - poll_interval_secs: 3, - } - } - - fn is_contact_allowed(&self, sender: &str) -> bool { - if self.allowed_contacts.iter().any(|u| u == "*") { - return true; - } - self.allowed_contacts - .iter() - .any(|u| u.eq_ignore_ascii_case(sender)) - } -} - -/// Escape a string for safe interpolation into `AppleScript`. -/// -/// This prevents injection attacks by escaping: -/// - Backslashes (`\` → `\\`) -/// - Double quotes (`"` → `\"`) -/// - Newlines (`\n` → `\\n`, `\r` → `\\r`) to prevent code injection via line breaks -fn escape_applescript(s: &str) -> String { - s.replace('\\', "\\\\") - .replace('"', "\\\"") - .replace('\n', "\\n") - .replace('\r', "\\r") -} - -/// Validate that a target looks like a valid phone number or email address. -/// -/// This is a defense-in-depth measure to reject obviously malicious targets -/// before they reach `AppleScript` interpolation. -/// -/// Valid patterns: -/// - Phone: starts with `+` followed by digits (with optional spaces/dashes) -/// - Email: contains `@` with alphanumeric chars on both sides -fn is_valid_imessage_target(target: &str) -> bool { - let target = target.trim(); - if target.is_empty() { - return false; - } - - // Phone number: +1234567890 or +1 234-567-8900 - if target.starts_with('+') { - let digits_only: String = target.chars().filter(char::is_ascii_digit).collect(); - // Must have at least 7 digits (shortest valid phone numbers) - return digits_only.len() >= 7 && digits_only.len() <= 15; - } - - // Email: simple validation (contains @ with chars on both sides) - if let Some(at_pos) = target.find('@') { - let local = &target[..at_pos]; - let domain = &target[at_pos + 1..]; - - // Local part: non-empty, alphanumeric + common email chars - let local_valid = !local.is_empty() - && local - .chars() - .all(|c| c.is_alphanumeric() || "._+-".contains(c)); - - // Domain: non-empty, contains a dot, alphanumeric + dots/hyphens - let domain_valid = !domain.is_empty() - && domain.contains('.') - && domain - .chars() - .all(|c| c.is_alphanumeric() || ".-".contains(c)); - - return local_valid && domain_valid; - } - - false -} - -#[async_trait] -impl Channel for IMessageChannel { - fn name(&self) -> &str { - "imessage" - } - - async fn send(&self, message: &SendMessage) -> anyhow::Result<()> { - // Defense-in-depth: validate target format before any interpolation - if !is_valid_imessage_target(&message.recipient) { - anyhow::bail!( - "Invalid iMessage target: must be a phone number (+1234567890) or email (user@example.com)" - ); - } - - // SECURITY: Escape both message AND target to prevent AppleScript injection - // See: CWE-78 (OS Command Injection) - let escaped_msg = escape_applescript(&message.content); - let escaped_target = escape_applescript(&message.recipient); - - let script = format!( - r#"tell application "Messages" - set targetService to 1st account whose service type = iMessage - set targetBuddy to participant "{escaped_target}" of targetService - send "{escaped_msg}" to targetBuddy -end tell"# - ); - - let output = tokio::process::Command::new("osascript") - .arg("-e") - .arg(&script) - .output() - .await?; - - if !output.status.success() { - let stderr = String::from_utf8_lossy(&output.stderr); - anyhow::bail!("iMessage send failed: {stderr}"); - } - - Ok(()) - } - - async fn listen(&self, tx: mpsc::Sender) -> anyhow::Result<()> { - tracing::info!("iMessage channel listening (AppleScript bridge)..."); - - // Query the Messages SQLite database for new messages - // The database is at ~/Library/Messages/chat.db - let db_path = UserDirs::new() - .map(|u| u.home_dir().join("Library/Messages/chat.db")) - .ok_or_else(|| anyhow::anyhow!("Cannot find home directory"))?; - - if !db_path.exists() { - anyhow::bail!( - "Messages database not found at {}. Ensure Messages.app is set up and Full Disk Access is granted.", - db_path.display() - ); - } - - // Open a persistent read-only connection instead of creating - // a new one on every 3-second poll cycle. - let path = db_path.to_path_buf(); - let conn = tokio::task::spawn_blocking(move || -> anyhow::Result { - Ok(Connection::open_with_flags( - &path, - OpenFlags::SQLITE_OPEN_READ_ONLY | OpenFlags::SQLITE_OPEN_NO_MUTEX, - )?) - }) - .await??; - - // Track the last ROWID we've seen (shuttle conn in and out) - let (mut conn, initial_rowid) = - tokio::task::spawn_blocking(move || -> anyhow::Result<(Connection, i64)> { - let rowid = { - let mut stmt = - conn.prepare("SELECT MAX(ROWID) FROM message WHERE is_from_me = 0")?; - let rowid: Option = stmt.query_row([], |row| row.get(0))?; - rowid.unwrap_or(0) - }; - Ok((conn, rowid)) - }) - .await??; - let mut last_rowid = initial_rowid; - - loop { - tokio::time::sleep(tokio::time::Duration::from_secs(self.poll_interval_secs)).await; - - let since = last_rowid; - let (returned_conn, poll_result) = tokio::task::spawn_blocking( - move || -> (Connection, anyhow::Result>) { - let result = (|| -> anyhow::Result> { - let mut stmt = conn.prepare( - "SELECT m.ROWID, h.id, m.text \ - FROM message m \ - JOIN handle h ON m.handle_id = h.ROWID \ - WHERE m.ROWID > ?1 \ - AND m.is_from_me = 0 \ - AND m.text IS NOT NULL \ - ORDER BY m.ROWID ASC \ - LIMIT 20", - )?; - let rows = stmt.query_map([since], |row| { - Ok(( - row.get::<_, i64>(0)?, - row.get::<_, String>(1)?, - row.get::<_, String>(2)?, - )) - })?; - let results = rows.collect::, _>>()?; - Ok(results) - })(); - - (conn, result) - }, - ) - .await - .map_err(|e| anyhow::anyhow!("iMessage poll worker join error: {e}"))?; - conn = returned_conn; - - match poll_result { - Ok(messages) => { - for (rowid, sender, text) in messages { - if rowid > last_rowid { - last_rowid = rowid; - } - - if !self.is_contact_allowed(&sender) { - continue; - } - - if text.trim().is_empty() { - continue; - } - - let msg = ChannelMessage { - id: rowid.to_string(), - sender: sender.clone(), - reply_target: sender.clone(), - content: text, - channel: "imessage".to_string(), - timestamp: std::time::SystemTime::now() - .duration_since(std::time::UNIX_EPOCH) - .unwrap_or_default() - .as_secs(), - thread_ts: None, - }; - - if tx.send(msg).await.is_err() { - return Ok(()); - } - } - } - Err(e) => { - tracing::warn!("iMessage poll error: {e}"); - } - } - } - } - - async fn health_check(&self) -> bool { - if std::env::consts::OS != "macos" { - return false; - } - - let db_path = UserDirs::new() - .map(|u| u.home_dir().join("Library/Messages/chat.db")) - .unwrap_or_default(); - - db_path.exists() - } -} - -/// Get the current max ROWID from the messages table. -/// Uses rusqlite with parameterized queries for security (CWE-89 prevention). -async fn get_max_rowid(db_path: &Path) -> anyhow::Result { - let path = db_path.to_path_buf(); - let result = tokio::task::spawn_blocking(move || -> anyhow::Result { - let conn = Connection::open_with_flags( - &path, - OpenFlags::SQLITE_OPEN_READ_ONLY | OpenFlags::SQLITE_OPEN_NO_MUTEX, - )?; - let mut stmt = conn.prepare("SELECT MAX(ROWID) FROM message WHERE is_from_me = 0")?; - let rowid: Option = stmt.query_row([], |row| row.get(0))?; - Ok(rowid.unwrap_or(0)) - }) - .await??; - Ok(result) -} - -/// Fetch messages newer than `since_rowid`. -/// Uses rusqlite with parameterized queries for security (CWE-89 prevention). -/// The `since_rowid` parameter is bound safely, preventing SQL injection. -async fn fetch_new_messages( - db_path: &Path, - since_rowid: i64, -) -> anyhow::Result> { - let path = db_path.to_path_buf(); - let results = - tokio::task::spawn_blocking(move || -> anyhow::Result> { - let conn = Connection::open_with_flags( - &path, - OpenFlags::SQLITE_OPEN_READ_ONLY | OpenFlags::SQLITE_OPEN_NO_MUTEX, - )?; - let mut stmt = conn.prepare( - "SELECT m.ROWID, h.id, m.text \ - FROM message m \ - JOIN handle h ON m.handle_id = h.ROWID \ - WHERE m.ROWID > ?1 \ - AND m.is_from_me = 0 \ - AND m.text IS NOT NULL \ - ORDER BY m.ROWID ASC \ - LIMIT 20", - )?; - let rows = stmt.query_map([since_rowid], |row| { - Ok(( - row.get::<_, i64>(0)?, - row.get::<_, String>(1)?, - row.get::<_, String>(2)?, - )) - })?; - rows.collect::, _>>().map_err(Into::into) - }) - .await??; - Ok(results) -} - -#[cfg(test)] -#[path = "imessage_tests.rs"] -mod tests; +pub use tinychannels::providers::imessage::*; diff --git a/src/openhuman/channels/providers/imessage_tests.rs b/src/openhuman/channels/providers/imessage_tests.rs deleted file mode 100644 index 852dbb07ea..0000000000 --- a/src/openhuman/channels/providers/imessage_tests.rs +++ /dev/null @@ -1,672 +0,0 @@ -use super::*; - -#[test] -fn creates_with_contacts() { - let ch = IMessageChannel::new(vec!["+1234567890".into()]); - assert_eq!(ch.allowed_contacts.len(), 1); - assert_eq!(ch.poll_interval_secs, 3); -} - -#[test] -fn creates_with_empty_contacts() { - let ch = IMessageChannel::new(vec![]); - assert!(ch.allowed_contacts.is_empty()); -} - -#[test] -fn wildcard_allows_anyone() { - let ch = IMessageChannel::new(vec!["*".into()]); - assert!(ch.is_contact_allowed("+1234567890")); - assert!(ch.is_contact_allowed("random@icloud.com")); - assert!(ch.is_contact_allowed("")); -} - -#[test] -fn specific_contact_allowed() { - let ch = IMessageChannel::new(vec!["+1234567890".into(), "user@icloud.com".into()]); - assert!(ch.is_contact_allowed("+1234567890")); - assert!(ch.is_contact_allowed("user@icloud.com")); -} - -#[test] -fn unknown_contact_denied() { - let ch = IMessageChannel::new(vec!["+1234567890".into()]); - assert!(!ch.is_contact_allowed("+9999999999")); - assert!(!ch.is_contact_allowed("hacker@evil.com")); -} - -#[test] -fn contact_case_insensitive() { - let ch = IMessageChannel::new(vec!["User@iCloud.com".into()]); - assert!(ch.is_contact_allowed("user@icloud.com")); - assert!(ch.is_contact_allowed("USER@ICLOUD.COM")); -} - -#[test] -fn empty_allowlist_denies_all() { - let ch = IMessageChannel::new(vec![]); - assert!(!ch.is_contact_allowed("+1234567890")); - assert!(!ch.is_contact_allowed("anyone")); -} - -#[test] -fn name_returns_imessage() { - let ch = IMessageChannel::new(vec![]); - assert_eq!(ch.name(), "imessage"); -} - -#[test] -fn wildcard_among_others_still_allows_all() { - let ch = IMessageChannel::new(vec!["+111".into(), "*".into(), "+222".into()]); - assert!(ch.is_contact_allowed("totally-unknown")); -} - -#[test] -fn contact_with_spaces_exact_match() { - let ch = IMessageChannel::new(vec![" spaced ".into()]); - assert!(ch.is_contact_allowed(" spaced ")); - assert!(!ch.is_contact_allowed("spaced")); -} - -// ══════════════════════════════════════════════════════════ -// AppleScript Escaping Tests (CWE-78 Prevention) -// ══════════════════════════════════════════════════════════ - -#[test] -fn escape_applescript_double_quotes() { - assert_eq!(escape_applescript(r#"hello "world""#), r#"hello \"world\""#); -} - -#[test] -fn escape_applescript_backslashes() { - assert_eq!(escape_applescript(r"path\to\file"), r"path\\to\\file"); -} - -#[test] -fn escape_applescript_mixed() { - assert_eq!( - escape_applescript(r#"say "hello\" world"#), - r#"say \"hello\\\" world"# - ); -} - -#[test] -fn escape_applescript_injection_attempt() { - // This is the exact attack vector from the security report - let malicious = r#"" & do shell script "id" & ""#; - let escaped = escape_applescript(malicious); - // After escaping, the quotes should be escaped and not break out - assert_eq!(escaped, r#"\" & do shell script \"id\" & \""#); - // Verify all quotes are now escaped (preceded by backslash) - // The escaped string should not have any unescaped quotes (quote not preceded by backslash) - let chars: Vec = escaped.chars().collect(); - for (i, &c) in chars.iter().enumerate() { - if c == '"' { - // Every quote must be preceded by a backslash - assert!( - i > 0 && chars[i - 1] == '\\', - "Found unescaped quote at position {i}" - ); - } - } -} - -#[test] -fn escape_applescript_empty_string() { - assert_eq!(escape_applescript(""), ""); -} - -#[test] -fn escape_applescript_no_special_chars() { - assert_eq!(escape_applescript("hello world"), "hello world"); -} - -#[test] -fn escape_applescript_unicode() { - assert_eq!(escape_applescript("hello 🦀 world"), "hello 🦀 world"); -} - -#[test] -fn escape_applescript_newlines_escaped() { - assert_eq!(escape_applescript("line1\nline2"), "line1\\nline2"); - assert_eq!(escape_applescript("line1\rline2"), "line1\\rline2"); - assert_eq!(escape_applescript("line1\r\nline2"), "line1\\r\\nline2"); -} - -// ══════════════════════════════════════════════════════════ -// Target Validation Tests -// ══════════════════════════════════════════════════════════ - -#[test] -fn valid_phone_number_simple() { - assert!(is_valid_imessage_target("+1234567890")); -} - -#[test] -fn valid_phone_number_with_country_code() { - assert!(is_valid_imessage_target("+14155551234")); -} - -#[test] -fn valid_phone_number_with_spaces() { - assert!(is_valid_imessage_target("+1 415 555 1234")); -} - -#[test] -fn valid_phone_number_with_dashes() { - assert!(is_valid_imessage_target("+1-415-555-1234")); -} - -#[test] -fn valid_phone_number_international() { - assert!(is_valid_imessage_target("+447911123456")); // UK - assert!(is_valid_imessage_target("+81312345678")); // Japan -} - -#[test] -fn valid_email_simple() { - assert!(is_valid_imessage_target("user@example.com")); -} - -#[test] -fn valid_email_with_subdomain() { - assert!(is_valid_imessage_target("user@mail.example.com")); -} - -#[test] -fn valid_email_with_plus() { - assert!(is_valid_imessage_target("user+tag@example.com")); -} - -#[test] -fn valid_email_with_dots() { - assert!(is_valid_imessage_target("first.last@example.com")); -} - -#[test] -fn valid_email_icloud() { - assert!(is_valid_imessage_target("user@icloud.com")); - assert!(is_valid_imessage_target("user@me.com")); -} - -#[test] -fn invalid_target_empty() { - assert!(!is_valid_imessage_target("")); - assert!(!is_valid_imessage_target(" ")); -} - -#[test] -fn invalid_target_no_plus_prefix() { - // Phone numbers must start with + - assert!(!is_valid_imessage_target("1234567890")); -} - -#[test] -fn invalid_target_too_short_phone() { - // Less than 7 digits - assert!(!is_valid_imessage_target("+123456")); -} - -#[test] -fn invalid_target_too_long_phone() { - // More than 15 digits - assert!(!is_valid_imessage_target("+1234567890123456")); -} - -#[test] -fn invalid_target_email_no_at() { - assert!(!is_valid_imessage_target("userexample.com")); -} - -#[test] -fn invalid_target_email_no_domain() { - assert!(!is_valid_imessage_target("user@")); -} - -#[test] -fn invalid_target_email_no_local() { - assert!(!is_valid_imessage_target("@example.com")); -} - -#[test] -fn invalid_target_email_no_dot_in_domain() { - assert!(!is_valid_imessage_target("user@localhost")); -} - -#[test] -fn invalid_target_injection_attempt() { - // The exact attack vector from the security report - assert!(!is_valid_imessage_target(r#"" & do shell script "id" & ""#)); -} - -#[test] -fn invalid_target_applescript_injection() { - // Various injection attempts - assert!(!is_valid_imessage_target(r#"test" & quit"#)); - assert!(!is_valid_imessage_target(r"test\ndo shell script")); - assert!(!is_valid_imessage_target("test\"; malicious code; \"")); -} - -#[test] -fn invalid_target_special_chars() { - assert!(!is_valid_imessage_target("user