fix: address Gemini review — transcription timeout, helper extraction, file_id sanitization

- Add 30s tokio::time::timeout around transcription middleware to prevent
  a slow/hanging Whisper API from blocking the message pipeline (DoS)
- Extract shared EmittedMessage→IncomingMessage conversion into
  convert_emitted_to_incoming() helper, eliminating duplication between
  process_emitted_messages and dispatch_emitted_messages
- Sanitize file_id and file_path in Telegram voice download to reject
  curly braces, preventing credential placeholder injection via malicious
  file_id values like "{OPENAI_API_KEY}"

Co-Authored-By: Claude Opus 4.6 <[email protected]>

channels-src/telegram/src/lib.rs

@@ -909,6 +909,13 @@ fn send_pairing_reply(chat_id: i64, code: &str) -> Result<(), String> {
 /// 1. Call getFile to get the file_path.
 /// 2. Download the file bytes from /file/bot{TOKEN}/{file_path}.
 fn download_voice_file(file_id: &str) -> Result<Vec<u8>, String> {
+    // Reject file_id containing curly braces to prevent credential placeholder
+    // injection (e.g., a malicious file_id like "{OPENAI_API_KEY}" would be
+    // interpreted by the host-side credential injector).
+    if file_id.contains('{') || file_id.contains('}') {
+        return Err("invalid file_id: contains forbidden characters".to_string());
+    }
+
     // Step 1: Call getFile to get file_path
     // Double braces `{{...}}` produce a literal `{TELEGRAM_BOT_TOKEN}` placeholder
     // in the URL, which the host-side credential injector replaces with the real token.
@@ -948,6 +955,11 @@ fn download_voice_file(file_id: &str) -> Result<Vec<u8>, String> {
         .file_path
         .ok_or_else(|| "getFile returned no file_path".to_string())?;
 
+    // Sanitize file_path against credential placeholder injection
+    if file_path.contains('{') || file_path.contains('}') {
+        return Err("invalid file_path: contains forbidden characters".to_string());
+    }
+
     // Step 2: Download the actual file bytes
     let download_url = format!(
         "https://api.telegram.org/file/bot{{TELEGRAM_BOT_TOKEN}}/{}",

src/channels/wasm/wrapper.rs

@@ -1533,31 +1533,12 @@ impl WasmChannel {
                 });
             }
 
-            // Convert to IncomingMessage
-            let mut msg = IncomingMessage::new(&self.name, &emitted.user_id, &emitted.content);
-
-            if let Some(name) = emitted.user_name {
-                msg = msg.with_user_name(name);
-            }
-
-            if let Some(thread_id) = emitted.thread_id {
-                msg = msg.with_thread(thread_id);
-            }
-
-            // Parse metadata JSON
-            if let Ok(metadata) = serde_json::from_str(&emitted.metadata_json) {
-                msg = msg.with_metadata(metadata);
-            }
-
-            // Carry attachments through (moves emitted.attachments into msg)
-            if !emitted.attachments.is_empty() {
-                msg = msg.with_attachments(emitted.attachments);
-            }
-
-            // Apply transcription middleware if available (may replace content with transcript)
-            if let Some(ref middleware) = self.transcription_middleware {
-                msg = middleware.process(msg).await;
-            }
+            let msg = Self::convert_emitted_to_incoming(
+                &self.name,
+                emitted,
+                self.transcription_middleware.as_deref(),
+            )
+            .await;
 
             // Send to stream (log post-transcription state intentionally)
             tracing::info!(
@@ -1791,31 +1772,9 @@ impl WasmChannel {
                 });
             }
 
-            // Convert to IncomingMessage
-            let mut msg = IncomingMessage::new(channel_name, &emitted.user_id, &emitted.content);
-
-            if let Some(name) = emitted.user_name {
-                msg = msg.with_user_name(name);
-            }
-
-            if let Some(thread_id) = emitted.thread_id {
-                msg = msg.with_thread(thread_id);
-            }
-
-            // Parse metadata JSON
-            if let Ok(metadata) = serde_json::from_str(&emitted.metadata_json) {
-                msg = msg.with_metadata(metadata);
-            }
-
-            // Carry attachments through (moves emitted.attachments into msg)
-            if !emitted.attachments.is_empty() {
-                msg = msg.with_attachments(emitted.attachments);
-            }
-
-            // Apply transcription middleware if available (may replace content with transcript)
-            if let Some(middleware) = transcription_middleware {
-                msg = middleware.process(msg).await;
-            }
+            let msg =
+                Self::convert_emitted_to_incoming(channel_name, emitted, transcription_middleware)
+                    .await;
 
             // Send to stream (log post-transcription state intentionally)
             tracing::info!(
@@ -1841,6 +1800,65 @@ impl WasmChannel {
 
         Ok(())
     }
+
+    /// Convert an `EmittedMessage` to an `IncomingMessage`, applying transcription
+    /// middleware with a timeout if available.
+    ///
+    /// Shared by both `process_emitted_messages` (HTTP callback path) and
+    /// `dispatch_emitted_messages` (polling path) to avoid duplication.
+    async fn convert_emitted_to_incoming(
+        channel_name: &str,
+        emitted: EmittedMessage,
+        transcription_middleware: Option<&crate::transcription::TranscriptionMiddleware>,
+    ) -> IncomingMessage {
+        // Save user_id before partial moves for potential timeout fallback
+        let user_id = emitted.user_id.clone();
+
+        let mut msg = IncomingMessage::new(channel_name, &emitted.user_id, &emitted.content);
+
+        if let Some(name) = emitted.user_name {
+            msg = msg.with_user_name(name);
+        }
+
+        if let Some(thread_id) = emitted.thread_id {
+            msg = msg.with_thread(thread_id);
+        }
+
+        // Parse metadata JSON
+        if let Ok(metadata) = serde_json::from_str(&emitted.metadata_json) {
+            msg = msg.with_metadata(metadata);
+        }
+
+        // Carry attachments through (moves emitted.attachments into msg)
+        if !emitted.attachments.is_empty() {
+            msg = msg.with_attachments(emitted.attachments);
+        }
+
+        // Apply transcription middleware with a 30-second timeout to prevent
+        // a slow/hanging provider from blocking the message pipeline indefinitely.
+        if let Some(middleware) = transcription_middleware {
+            match tokio::time::timeout(std::time::Duration::from_secs(30), middleware.process(msg))
+                .await
+            {
+                Ok(processed) => return processed,
+                Err(_) => {
+                    tracing::error!(
+                        channel = %channel_name,
+                        "Transcription timed out after 30s, delivering message without transcript"
+                    );
+                    // Timeout: `msg` was moved into the timed-out future, so
+                    // reconstruct a fallback message from the saved user_id.
+                    return IncomingMessage::new(
+                        channel_name,
+                        &user_id,
+                        "[Voice note: transcription timed out]",
+                    );
+                }
+            }
+        }
+
+        msg
+    }
 }
 
 #[async_trait]