diff --git a/Cargo.lock b/Cargo.lock index b44a50c403..362c45de22 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -4424,7 +4424,6 @@ dependencies = [ "schemars", "sentry", "serde", - "serde-big-array", "serde_json", "serde_repr", "serde_yaml", @@ -4466,13 +4465,9 @@ dependencies = [ "url", "urlencoding", "uuid 1.23.1", - "wacore", "wait-timeout", "walkdir", "webpki-roots 1.0.7", - "whatsapp-rust", - "whatsapp-rust-tokio-transport", - "whatsapp-rust-ureq-http-client", "whisper-rs", "windows-sys 0.61.2", "wiremock", @@ -6830,20 +6825,44 @@ 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-big-array", "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", + "wacore", + "webpki-roots 1.0.7", + "whatsapp-rust", + "whatsapp-rust-tokio-transport", + "whatsapp-rust-ureq-http-client", ] [[package]] diff --git a/Cargo.toml b/Cargo.toml index 70240e8696..c4a095fafe 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -246,16 +246,10 @@ coins-bip39 = "0.8" curve25519-dalek = { version = "4", default-features = false, features = ["alloc"] } fantoccini = { version = "0.22.0", optional = true, default-features = false, features = ["rustls-tls"] } -serde-big-array = { version = "0.5", optional = true } pdf-extract = "0.10" -# WhatsApp Web — upstream `whatsapp-rust` 0.5. Its Diesel-backed sqlite-storage -# feature links sqlite3 separately from rusqlite 0.40, so the TinyAgents 1.5 -# baseline compiles this provider against wacore's in-memory Backend until a -# rusqlite-backed durable store lands. -whatsapp-rust = { version = "0.5", optional = true, default-features = false, features = ["tokio-runtime"] } -whatsapp-rust-tokio-transport = { version = "0.5", optional = true, default-features = false } -whatsapp-rust-ureq-http-client = { version = "0.5", optional = true } -wacore = { version = "0.5", optional = true, default-features = false } +# The WhatsApp Web provider (and its `whatsapp-rust` / `wacore` / `serde-big-array` +# stack) now lives in the tinychannels crate; the `whatsapp-web` feature forwards +# to `tinychannels/whatsapp-web`. ppt-rs = "0.2.14" [target.'cfg(windows)'.dependencies] @@ -335,7 +329,8 @@ peripheral-rpi = ["dep:rppal"] browser-native = ["dep:fantoccini"] fantoccini = ["browser-native"] landlock = ["sandbox-landlock"] -whatsapp-web = ["dep:whatsapp-rust", "dep:whatsapp-rust-tokio-transport", "dep:whatsapp-rust-ureq-http-client", "dep:wacore", "serde-big-array"] +# The WhatsApp Web provider now lives in tinychannels; forward to its feature. +whatsapp-web = ["tinychannels/whatsapp-web"] # Exposes the destructive `openhuman.test_reset` RPC. Off by default; the E2E # build (app/scripts/e2e-build.sh) flips it on. Shipped binaries never have # this feature so the wipe RPC isn't even registered, let alone reachable. 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/agent/task_dispatcher/executor.rs b/src/openhuman/agent/task_dispatcher/executor.rs index bc125a1a7b..97bca3d072 100644 --- a/src/openhuman/agent/task_dispatcher/executor.rs +++ b/src/openhuman/agent/task_dispatcher/executor.rs @@ -227,7 +227,7 @@ pub(super) async fn run_autonomous( if let Some(thread_id) = session_thread_id.as_deref() { match &result { Ok(response) => { - crate::openhuman::channels::providers::presentation::deliver_response( + crate::openhuman::channels::providers::web::presentation::deliver_response( "system", thread_id, run_id, 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/host/adapters.rs b/src/openhuman/channels/host/adapters.rs new file mode 100644 index 0000000000..bcf1f88e54 --- /dev/null +++ b/src/openhuman/channels/host/adapters.rs @@ -0,0 +1,377 @@ +//! OpenHuman-side implementations of the portable `tinychannels::host` +//! capability traits. +//! +//! Each adapter wraps existing OpenHuman internals (voice factory, inference +//! ops, approval gate, conversation store, shutdown registry, web event bus) +//! and exposes them through the portable, `Config`-free trait surface a ported +//! channel provider consumes. See [`super::build_channel_host`]. + +use std::sync::Arc; + +use async_trait::async_trait; +use parking_lot::Mutex; +use tinychannels::host::{ + AllowlistStore, ApprovalDecision, ApprovalGate, ConversationMessage, ConversationStore, + EventSink, LifecycleRegistry, ReactionDecision, ReactionGate, ReactionQuery, ShutdownHook, + SpeechRequest, SpeechResult, SpeechSynthesizer, Transcriber, TranscriptionRequest, + TranscriptionResult, +}; + +use crate::openhuman::config::Config; + +const LOG_PREFIX: &str = "[channels_host]"; + +// --------------------------------------------------------------------------- +// LifecycleRegistry → core::shutdown +// --------------------------------------------------------------------------- + +/// Bridges [`LifecycleRegistry`] onto the process-global async shutdown hook +/// registry. Providers register teardown that runs once on shutdown. +pub struct CoreShutdownRegistry; + +impl LifecycleRegistry for CoreShutdownRegistry { + fn register_shutdown(&self, name: &str, hook: ShutdownHook) { + let name = name.to_string(); + // `core::shutdown::register` takes a re-callable `Fn`, but our hook is a + // one-shot `FnOnce`; guard it behind a take-once slot so a (theoretical) + // second invocation is a no-op. + let slot = Arc::new(Mutex::new(Some(hook))); + crate::core::shutdown::register(move || { + let slot = Arc::clone(&slot); + let name = name.clone(); + async move { + let taken = slot.lock().take(); + if let Some(hook) = taken { + tracing::debug!("{LOG_PREFIX} running shutdown hook: {name}"); + hook().await; + } + } + }); + } +} + +// --------------------------------------------------------------------------- +// Transcriber → voice STT factory +// --------------------------------------------------------------------------- + +/// Speech-to-text backed by the OpenHuman voice provider factory. +pub struct VoiceTranscriber { + pub config: Arc, +} + +#[async_trait] +impl Transcriber for VoiceTranscriber { + fn name(&self) -> &str { + "openhuman-voice" + } + + async fn transcribe( + &self, + request: TranscriptionRequest, + ) -> anyhow::Result { + let provider = crate::openhuman::voice::effective_stt_provider(&self.config); + tracing::debug!( + "{LOG_PREFIX} transcribe provider={provider} bytes_b64={}", + request.audio_base64.len() + ); + // Empty model → factory substitutes DEFAULT_WHISPER_MODEL. + let stt = crate::openhuman::voice::create_stt_provider(&provider, "", &self.config)?; + let outcome = stt + .transcribe( + &self.config, + &request.audio_base64, + request.mime_type.as_deref(), + request.file_name.as_deref(), + request.language.as_deref(), + ) + .await + .map_err(|e| anyhow::anyhow!(e))?; + Ok(TranscriptionResult { + text: outcome.value.text, + language: request.language, + duration_secs: None, + }) + } +} + +// --------------------------------------------------------------------------- +// SpeechSynthesizer → voice reply_speech +// --------------------------------------------------------------------------- + +/// Text-to-speech backed by the hosted reply-speech synthesizer. +pub struct VoiceSynthesizer { + pub config: Arc, +} + +#[async_trait] +impl SpeechSynthesizer for VoiceSynthesizer { + async fn synthesize(&self, request: SpeechRequest) -> anyhow::Result { + tracing::debug!( + "{LOG_PREFIX} synthesize chars={} voice={:?}", + request.text.len(), + request.voice + ); + let opts = crate::openhuman::voice::reply_speech::ReplySpeechOptions { + voice_id: request.voice, + model_id: None, + output_format: request.format, + voice_settings: None, + }; + let outcome = crate::openhuman::voice::reply_speech::synthesize_reply( + &self.config, + &request.text, + &opts, + ) + .await + .map_err(|e| anyhow::anyhow!(e))?; + let value = outcome.value; + let visemes = serde_json::to_value(&value.visemes).ok(); + Ok(SpeechResult { + audio_base64: value.audio_base64, + mime_type: value.audio_mime, + visemes, + }) + } +} + +// --------------------------------------------------------------------------- +// ReactionGate → inference should_react +// --------------------------------------------------------------------------- + +/// Inference-driven reaction gate backed by the local-AI should-react op. +pub struct InferenceReactionGate { + pub config: Arc, +} + +#[async_trait] +impl ReactionGate for InferenceReactionGate { + async fn should_react(&self, query: ReactionQuery) -> anyhow::Result { + // Honour the runtime gate: when the local model runtime is disabled we + // never react (matches presentation's prior inline guard). + if !self.config.local_ai.runtime_enabled { + tracing::debug!("{LOG_PREFIX} should_react skipped (local runtime disabled)"); + return Ok(ReactionDecision::default()); + } + let outcome = crate::openhuman::inference::ops::inference_should_react( + &self.config, + &query.message, + &query.channel_type, + ) + .await + .map_err(|e| anyhow::anyhow!(e))?; + Ok(ReactionDecision { + should_react: outcome.value.should_react, + emoji: outcome.value.emoji, + reason: None, + }) + } +} + +// --------------------------------------------------------------------------- +// ApprovalGate → approval reply parsing +// --------------------------------------------------------------------------- + +/// Parses inbound approval replies via the OpenHuman approval gate. Raising +/// interactive approvals stays host-internal (the tool gate), so only +/// [`ApprovalGate::parse_reply`] is implemented. +pub struct CoreApprovalGate; + +impl ApprovalGate for CoreApprovalGate { + fn parse_reply(&self, message: &str) -> Option { + crate::openhuman::approval::parse_approval_reply(message).map(|decision| { + use crate::openhuman::approval::ApprovalDecision as Core; + match decision { + Core::ApproveOnce => ApprovalDecision::Approve, + Core::ApproveAlwaysForTool => { + ApprovalDecision::Choice("approve_always_for_tool".to_string()) + } + Core::Deny => ApprovalDecision::Deny, + } + }) + } +} + +// --------------------------------------------------------------------------- +// ConversationStore → memory_conversations +// --------------------------------------------------------------------------- + +/// Durable conversation history backed by the OpenHuman conversation store. +pub struct ConversationHistoryStore { + pub workspace_dir: std::path::PathBuf, +} + +#[async_trait] +impl ConversationStore for ConversationHistoryStore { + async fn history( + &self, + session_key: &str, + limit: usize, + ) -> anyhow::Result> { + let messages = crate::openhuman::memory_conversations::get_messages( + self.workspace_dir.clone(), + session_key, + ) + .map_err(|e| anyhow::anyhow!(e))?; + let start = messages.len().saturating_sub(limit); + Ok(messages[start..] + .iter() + .map(|m| ConversationMessage { + role: m.message_type.clone(), + content: m.content.clone(), + timestamp: None, + }) + .collect()) + } + + async fn append(&self, session_key: &str, message: ConversationMessage) -> anyhow::Result<()> { + let now = chrono::Utc::now().to_rfc3339(); + // `append_message` requires the thread to exist; create-or-noop first. + crate::openhuman::memory_conversations::ensure_thread( + self.workspace_dir.clone(), + crate::openhuman::memory_conversations::CreateConversationThread { + id: session_key.to_string(), + title: session_key.to_string(), + created_at: now.clone(), + parent_thread_id: None, + labels: None, + personality_id: None, + }, + ) + .map_err(|e| anyhow::anyhow!(e))?; + let stored = crate::openhuman::memory_conversations::ConversationMessage { + id: uuid::Uuid::new_v4().to_string(), + content: message.content, + message_type: message.role.clone(), + extra_metadata: serde_json::Value::Null, + sender: message.role, + created_at: now, + }; + crate::openhuman::memory_conversations::append_message( + self.workspace_dir.clone(), + session_key, + stored, + ) + .map_err(|e| anyhow::anyhow!(e))?; + Ok(()) + } +} + +// --------------------------------------------------------------------------- +// AllowlistStore → config.toml channel allowlist +// --------------------------------------------------------------------------- + +/// Persists newly-authorized identities into the on-disk channel allowlist, +/// replicating Telegram's former `persist_allowed_identity` (load +/// `~/.openhuman/config.toml`, append to the channel's `allowed_users`, save). +pub struct ConfigAllowlistStore; + +#[async_trait] +impl AllowlistStore for ConfigAllowlistStore { + async fn persist_allowed_identity(&self, channel: &str, identity: &str) -> anyhow::Result<()> { + use anyhow::Context; + let normalized = identity.trim().trim_start_matches('@').to_string(); + if normalized.is_empty() { + anyhow::bail!("cannot persist empty identity"); + } + + let home = directories::UserDirs::new() + .map(|u| u.home_dir().to_path_buf()) + .context("could not find home directory")?; + let openhuman_dir = home.join(".openhuman"); + let config_path = openhuman_dir.join("config.toml"); + let contents = tokio::fs::read_to_string(&config_path) + .await + .with_context(|| format!("failed to read config file: {}", config_path.display()))?; + let mut config: Config = + toml::from_str(&contents).context("failed to parse config.toml for allowlist")?; + config.config_path = config_path; + config.workspace_dir = openhuman_dir.join("workspace"); + + match channel { + "telegram" => { + let Some(telegram) = config.channels_config.telegram.as_mut() else { + anyhow::bail!("telegram channel config is missing in config.toml"); + }; + if !telegram.allowed_users.iter().any(|u| u == &normalized) { + telegram.allowed_users.push(normalized); + config + .save() + .await + .context("failed to persist allowlist to config.toml")?; + } + } + other => anyhow::bail!("allowlist persist unsupported for channel '{other}'"), + } + tracing::debug!("{LOG_PREFIX} persisted allowed identity for channel={channel}"); + Ok(()) + } +} + +// --------------------------------------------------------------------------- +// EventSink → routes provider events to the right OpenHuman bus +// --------------------------------------------------------------------------- + +/// Routes provider events by `domain`: +/// - `"web"` → the web channel's `WebChannelEvent` broadcast bus (payload +/// must deserialize into a `WebChannelEvent`; presentation builds that shape). +/// - `"channel"` → the global `DomainEvent` bus (telegram reaction fan-out). +/// +/// One capability, two backends — providers don't know which bus they hit. +pub struct OpenHumanEventSink; + +fn json_str(payload: &serde_json::Value, key: &str) -> String { + payload + .get(key) + .and_then(|v| v.as_str()) + .unwrap_or_default() + .to_string() +} + +#[async_trait] +impl EventSink for OpenHumanEventSink { + async fn publish( + &self, + domain: &str, + kind: &str, + payload: serde_json::Value, + ) -> anyhow::Result<()> { + match domain { + "web" => { + let event: crate::core::socketio::WebChannelEvent = serde_json::from_value(payload) + .map_err(|e| { + anyhow::anyhow!( + "{LOG_PREFIX} web event payload not a WebChannelEvent ({kind}): {e}" + ) + })?; + crate::openhuman::channels::providers::web::publish_web_channel_event(event); + } + "channel" => { + use crate::core::event_bus::{publish_global, DomainEvent}; + let event = match kind { + "reaction_received" => DomainEvent::ChannelReactionReceived { + channel: json_str(&payload, "channel"), + sender: json_str(&payload, "sender"), + target_message_id: json_str(&payload, "target_message_id"), + emoji: json_str(&payload, "emoji"), + }, + "reaction_sent" => DomainEvent::ChannelReactionSent { + channel: json_str(&payload, "channel"), + target_message_id: json_str(&payload, "target_message_id"), + emoji: json_str(&payload, "emoji"), + success: payload + .get("success") + .and_then(|v| v.as_bool()) + .unwrap_or(false), + }, + other => { + tracing::warn!("{LOG_PREFIX} unmapped channel event kind: {other}"); + return Ok(()); + } + }; + publish_global(event); + } + other => tracing::warn!("{LOG_PREFIX} unmapped event domain: {other}"), + } + Ok(()) + } +} diff --git a/src/openhuman/channels/host/mod.rs b/src/openhuman/channels/host/mod.rs new file mode 100644 index 0000000000..e8f8d8c8e6 --- /dev/null +++ b/src/openhuman/channels/host/mod.rs @@ -0,0 +1,65 @@ +//! OpenHuman implementation of the `tinychannels::host` capability boundary. +//! +//! [`build_channel_host`] assembles the concrete [`tinychannels::ChannelHost`] +//! from OpenHuman internals (voice, inference, approvals, conversation store, +//! shutdown registry, web event bus). [`build_provider_context`] wraps it into +//! the [`tinychannels::host::ProviderContext`] handed to channel providers. +//! +//! Ported providers reach host capabilities through this context instead of +//! calling OpenHuman internals directly — the inversion that lets them live in +//! the standalone `tinychannels` crate. Lean providers ignore the host. + +mod adapters; + +pub use adapters::{ + ConfigAllowlistStore, ConversationHistoryStore, CoreApprovalGate, CoreShutdownRegistry, + InferenceReactionGate, OpenHumanEventSink, VoiceSynthesizer, VoiceTranscriber, +}; + +use std::sync::Arc; + +use tinychannels::host::{ChannelHostBuilder, ProviderContext}; +use tinychannels::ChannelHost; + +use crate::openhuman::config::Config; + +/// Assemble the full OpenHuman [`ChannelHost`] from a config snapshot. +/// +/// Wires every capability OpenHuman can back today: lifecycle (shutdown), +/// STT, TTS, reaction gate, approval-reply parsing, conversation history, and +/// the web-channel event sink. Capabilities OpenHuman cannot yet express +/// portably (turn dispatch, run ledger, pairing) are simply left unset — a +/// provider that needs one degrades gracefully. +pub fn build_channel_host(config: Arc) -> Arc { + ChannelHostBuilder::new() + .lifecycle(Arc::new(CoreShutdownRegistry)) + .transcriber(Arc::new(VoiceTranscriber { + config: Arc::clone(&config), + })) + .synthesizer(Arc::new(VoiceSynthesizer { + config: Arc::clone(&config), + })) + .reactions(Arc::new(InferenceReactionGate { + config: Arc::clone(&config), + })) + .approvals(Arc::new(CoreApprovalGate)) + .conversations(Arc::new(ConversationHistoryStore { + workspace_dir: config.workspace_dir.clone(), + })) + .events(Arc::new(OpenHumanEventSink)) + .allowlist(Arc::new(ConfigAllowlistStore)) + .build() +} + +/// Build the [`ProviderContext`] handed to a channel provider at construction: +/// the assembled host + the channels config + a pre-built HTTP client. +pub fn build_provider_context(config: &Config, http_client: reqwest::Client) -> ProviderContext { + ProviderContext::new( + build_channel_host(Arc::new(config.clone())), + config.channels_config.clone(), + http_client, + ) +} + +#[cfg(test)] +mod tests; diff --git a/src/openhuman/channels/host/tests.rs b/src/openhuman/channels/host/tests.rs new file mode 100644 index 0000000000..6f379799b3 --- /dev/null +++ b/src/openhuman/channels/host/tests.rs @@ -0,0 +1,182 @@ +//! Unit tests for the OpenHuman ChannelHost capability adapters. + +use super::*; +use std::sync::Arc; +use tinychannels::host::{ + ApprovalDecision, ApprovalGate, ConversationMessage, ConversationStore, EventSink, + ReactionGate, ReactionQuery, Transcriber, +}; + +use crate::openhuman::config::Config; + +// --- ApprovalGate::parse_reply ------------------------------------------- + +#[test] +fn approval_gate_maps_core_replies() { + let gate = CoreApprovalGate; + assert_eq!(gate.parse_reply("yes"), Some(ApprovalDecision::Approve)); + assert_eq!(gate.parse_reply("APPROVE"), Some(ApprovalDecision::Approve)); + assert_eq!(gate.parse_reply("no"), Some(ApprovalDecision::Deny)); + assert_eq!(gate.parse_reply("deny"), Some(ApprovalDecision::Deny)); + assert_eq!(gate.parse_reply("banana"), None); +} + +// --- ReactionGate (runtime-disabled short-circuit) ------------------------ + +#[tokio::test] +async fn reaction_gate_returns_default_when_runtime_disabled() { + let mut config = Config::default(); + config.local_ai.runtime_enabled = false; + let gate = InferenceReactionGate { + config: Arc::new(config), + }; + let decision = gate + .should_react(ReactionQuery { + message: "hello".into(), + channel_type: "web".into(), + }) + .await + .expect("should_react ok"); + assert!(!decision.should_react); + assert!(decision.emoji.is_none()); +} + +// --- OpenHumanEventSink -------------------------------------------------- + +#[tokio::test] +async fn event_sink_accepts_web_channel_event_shape() { + let sink = OpenHumanEventSink; + let ok = sink + .publish( + "web", + "chat_done", + serde_json::json!({ + "event": "chat_done", + "client_id": "c1", + "thread_id": "t1", + "request_id": "r1", + "full_response": "hi" + }), + ) + .await; + assert!(ok.is_ok()); +} + +#[tokio::test] +async fn event_sink_rejects_non_object_payload() { + let sink = OpenHumanEventSink; + let err = sink + .publish("web", "bad", serde_json::json!([1, 2, 3])) + .await; + assert!(err.is_err()); +} + +#[tokio::test] +async fn event_sink_routes_channel_reactions_to_domain_bus() { + let sink = OpenHumanEventSink; + assert!(sink + .publish( + "channel", + "reaction_received", + serde_json::json!({ + "channel": "telegram", "sender": "u1", + "target_message_id": "telegram_1_2", "emoji": "👍" + }), + ) + .await + .is_ok()); + assert!(sink + .publish( + "channel", + "reaction_sent", + serde_json::json!({ + "channel": "telegram", "target_message_id": "telegram_1_2", + "emoji": "👍", "success": true + }), + ) + .await + .is_ok()); + // Unknown domain/kind is a benign no-op. + assert!(sink + .publish("mystery", "x", serde_json::json!({})) + .await + .is_ok()); + assert!(sink + .publish("channel", "unknown", serde_json::json!({})) + .await + .is_ok()); +} + +// --- ConversationHistoryStore -------------------------------------------- + +#[tokio::test] +async fn conversation_store_append_then_history_roundtrips() { + let dir = tempfile::tempdir().expect("tempdir"); + let store = ConversationHistoryStore { + workspace_dir: dir.path().to_path_buf(), + }; + let thread = "thread-1"; + + store + .append( + thread, + ConversationMessage { + role: "user".into(), + content: "first".into(), + timestamp: None, + }, + ) + .await + .expect("append user"); + store + .append( + thread, + ConversationMessage { + role: "assistant".into(), + content: "second".into(), + timestamp: None, + }, + ) + .await + .expect("append assistant"); + + let history = store.history(thread, 10).await.expect("history"); + assert_eq!(history.len(), 2); + assert_eq!(history[0].content, "first"); + assert_eq!(history[0].role, "user"); + assert_eq!(history[1].content, "second"); + assert_eq!(history[1].role, "assistant"); + + // `limit` keeps the most recent messages. + let recent = store.history(thread, 1).await.expect("history limited"); + assert_eq!(recent.len(), 1); + assert_eq!(recent[0].content, "second"); +} + +// --- build_channel_host --------------------------------------------------- + +#[test] +fn build_channel_host_advertises_expected_capabilities() { + let host = build_channel_host(Arc::new(Config::default())); + let caps = host.capabilities(); + assert!(caps.lifecycle); + assert!(caps.stt); + assert!(caps.tts); + assert!(caps.reaction_gate); + assert!(caps.approvals); + assert!(caps.conversation_store); + assert!(caps.event_sink); + assert!(caps.allowlist_store); + // Not yet backed portably: + assert!(!caps.turn_dispatch); + assert!(!caps.run_ledger); + assert!(!caps.memory_recall); +} + +#[test] +fn transcriber_reports_stable_name() { + let stt = VoiceTranscriber { + config: Arc::new(Config::default()), + }; + assert_eq!(stt.name(), "openhuman-voice"); +} diff --git a/src/openhuman/channels/mod.rs b/src/openhuman/channels/mod.rs index 057e15f13b..e869a3876a 100644 --- a/src/openhuman/channels/mod.rs +++ b/src/openhuman/channels/mod.rs @@ -3,6 +3,7 @@ pub mod bus; pub mod cli; pub mod controllers; +pub mod host; pub mod proactive; pub mod providers; pub(crate) mod relay_runtime; 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