diff --git a/README.md b/README.md index 9c00ecf..e88b048 100644 --- a/README.md +++ b/README.md @@ -349,6 +349,10 @@ Skill 是包含 `SKILL.md` 的目录。加载优先级从高到低： | `channels.feishu.live_updates` | `false` | | `channels.feishu.live_update_interval_ms` | `500` | | `channels.feishu.require_mention` | `true` | +| `channels.feishu.max_image_bytes` | `10485760` | +| `channels.feishu.max_file_bytes` | `26214400` | +| `channels.feishu.media_dir_max_bytes` | `536870912` | +| `channels.feishu.request_timeout_secs` | `30` | 更完整的配置字段说明见 [resources/skills/about-picobot/references/config.md](resources/skills/about-picobot/references/config.md)。 diff --git a/docs/ARCHITECTURE.md b/docs/ARCHITECTURE.md index 95cc01f..9ecf749 100644 --- a/docs/ARCHITECTURE.md +++ b/docs/ARCHITECTURE.md @@ -141,6 +141,7 @@ sequenceDiagram - `cli_chat` 将同一 `turn_updated` 快照发给 TUI 和 WebUI。客户端只保留当前 session 中 revision 更新的 `active_turn`，终态随后由持久化历史校准。 - 飞书对每个 DATA 帧先在 2 秒硬期限内 ACK，再进行有界分片重组，并把完整事件交给容量 32 的连接内处理队列；媒体下载和引用查询不占用正常的 WebSocket 读循环。队列饱和时当前事件在连接任务中同步处理而不丢弃。连接异常采用有上限的指数退避持续重连，不因累计故障永久停止。 - 飞书在协议解析阶段按 `allow_from` 拒绝未授权用户；群聊默认必须明确 @ 运行时解析出的机器人身份，身份解析失败时安全地忽略群消息。飞书把当前消息 ID 作为 `reply_to`，并在私有 metadata 中携带 root/thread 信息；Sink 使用原生 reply API 及 `reply_in_thread` 保持客户端引用和话题位置。飞书默认 `FinalOnly`；开启 `live_updates` 后，第一个可见快照创建卡片，后续编辑同一卡片，终态编辑失败则发送完整结果兜底。reaction 清理在 finish、abort 和 Gateway shutdown 中幂等执行。 +- 飞书入站响应体按类型流式执行字节上限和超时检查，写盘前校验媒体目录总容量，客户端文件名先收敛为安全 basename；出站上传也先检查本地文件大小。消息发送和资源下载对网络错误、429、5xx 和 401 进行有界重试，401 或飞书失效 token 业务码会使租户 token 缓存失效后重新获取。 - DeliveryCoordinator 与 OutboundDispatcher 共享 `(channel, chat_id)` 写锁，避免活动 Turn 终态与独立消息并发写入同一目标。 ### 出站投递 diff --git a/resources/skills/about-picobot/assets/config.example.json b/resources/skills/about-picobot/assets/config.example.json index 1084f6a..51c0947 100644 --- a/resources/skills/about-picobot/assets/config.example.json +++ b/resources/skills/about-picobot/assets/config.example.json @@ -63,7 +63,13 @@ "require_mention": true, "agent": "default", "media_dir": "~/.picobot/media/feishu", - "reaction_emoji": "Typing" + "reaction_emoji": "Typing", + "live_updates": false, + "live_update_interval_ms": 500, + "max_image_bytes": 10485760, + "max_file_bytes": 26214400, + "media_dir_max_bytes": 536870912, + "request_timeout_secs": 30 } }, "memory": { diff --git a/resources/skills/about-picobot/references/config.md b/resources/skills/about-picobot/references/config.md index 2f45ca3..cbe7919 100644 --- a/resources/skills/about-picobot/references/config.md +++ b/resources/skills/about-picobot/references/config.md @@ -100,6 +100,10 @@ Gateway WebUI 的“配置”页可以编辑实际加载的配置文件。读取 | `reaction_emoji` | string | "Typing" | 回复意向表达的表情 | | `live_updates` | bool | false | 是否用单张卡片实时编辑活动 Turn；关闭时只发送终态 | | `live_update_interval_ms` | int | 500 | 卡片更新最小间隔，运行时限制在 250–5000ms | +| `max_image_bytes` | int | 10485760 | 单个入站/出站图片的最大字节数 | +| `max_file_bytes` | int | 26214400 | 单个入站/出站文件、音频或视频的最大字节数 | +| `media_dir_max_bytes` | int | 536870912 | 飞书媒体目录容量上限；达到上限后拒绝新下载，不自动删除旧文件 | +| `request_timeout_secs` | int | 30 | 单次飞书 HTTP 请求及响应体读取的硬超时，运行时限制在 5–120 秒 | 飞书属于外部渠道：无论是否开启实时卡片，都不会接收模型 reasoning；工具只显示紧凑状态。配置修改需重启 Gateway 生效。 diff --git a/resources/templates/config.example.json b/resources/templates/config.example.json index 4b744e9..5103d6c 100644 --- a/resources/templates/config.example.json +++ b/resources/templates/config.example.json @@ -73,7 +73,11 @@ "media_dir": "~/.picobot/media/feishu", "reaction_emoji": "Typing", "live_updates": false, - "live_update_interval_ms": 500 + "live_update_interval_ms": 500, + "max_image_bytes": 10485760, + "max_file_bytes": 26214400, + "media_dir_max_bytes": 536870912, + "request_timeout_secs": 30 } }, "memory": { diff --git a/src/channels/feishu.rs b/src/channels/feishu.rs index d43871b..8e0f6ac 100644 --- a/src/channels/feishu.rs +++ b/src/channels/feishu.rs @@ -39,6 +39,8 @@ const WS_FRAGMENT_TTL: Duration = Duration::from_secs(5 * 60); const MAX_WS_FRAGMENTS: usize = 256; const MAX_WS_EVENT_BYTES: usize = 8 * 1024 * 1024; const STABLE_CONNECTION_WINDOW: Duration = Duration::from_secs(30); +const FEISHU_API_ATTEMPTS: usize = 3; +const FEISHU_INVALID_TOKEN_CODE: i32 = 99_991_663; // ───────────────────────────────────────────────────────────────────────────── // Protobuf types for Feishu WebSocket protocol (pbbp2.proto) @@ -214,6 +216,18 @@ struct FeishuSendTarget { reply_in_thread: bool, } +#[derive(Deserialize)] +struct FeishuMessageResponse { + code: i32, + msg: String, + data: Option, +} + +#[derive(Deserialize)] +struct FeishuMessageData { + message_id: String, +} + impl FeishuSendTarget { fn from_message( chat_id: String, @@ -283,9 +297,17 @@ impl FeishuChannel { let media_dir = workspace_dir.join("media").join("feishu"); config.media_dir = media_dir.to_string_lossy().to_string(); + let request_timeout = Duration::from_secs(config.request_timeout_secs.clamp(5, 120)); + let http_client = reqwest::Client::builder() + .timeout(request_timeout) + .build() + .map_err(|error| { + ChannelError::ConfigError(format!("Failed to build Feishu HTTP client: {error}")) + })?; + Ok(Self { config, - http_client: reqwest::Client::new(), + http_client, running: Arc::new(RwLock::new(false)), shutdown: Arc::new(RwLock::new(None)), run_task: Arc::new(Mutex::new(None)), @@ -296,17 +318,68 @@ impl FeishuChannel { }) } + async fn invalidate_tenant_token(&self) { + *self.tenant_token.write().await = None; + } + + fn request_timeout(&self) -> Duration { + Duration::from_secs(self.config.request_timeout_secs.clamp(5, 120)) + } + + async fn send_authenticated( + &self, + operation: &str, + build: F, + ) -> Result + where + F: Fn(&reqwest::Client, &str) -> reqwest::RequestBuilder, + { + for attempt in 0..FEISHU_API_ATTEMPTS { + let token = self.get_tenant_access_token().await?; + match build(&self.http_client, &token).send().await { + Ok(response) => { + let status = response.status(); + if status == reqwest::StatusCode::UNAUTHORIZED { + self.invalidate_tenant_token().await; + } + if status == reqwest::StatusCode::TOO_MANY_REQUESTS + || status.is_server_error() + || status == reqwest::StatusCode::UNAUTHORIZED + { + if attempt + 1 < FEISHU_API_ATTEMPTS { + let delay = retry_after_delay(response.headers(), attempt); + tokio::time::sleep(delay).await; + continue; + } + return Err(ChannelError::ConnectionError(format!( + "{operation} failed after {} attempts: HTTP {status}", + attempt + 1 + ))); + } + return Ok(response); + } + Err(error) => { + if attempt + 1 == FEISHU_API_ATTEMPTS { + return Err(ChannelError::ConnectionError(format!( + "{operation} failed after {} attempts: {error}", + attempt + 1 + ))); + } + tokio::time::sleep(api_retry_delay(attempt)).await; + } + } + } + unreachable!("Feishu API attempt loop always returns") + } + async fn refresh_bot_open_id(&self) -> Result { - let token = self.get_tenant_access_token().await?; let response = self - .http_client - .get(format!("{}/bot/v3/info", FEISHU_API_BASE)) - .bearer_auth(token) - .send() - .await - .map_err(|error| { - ChannelError::ConnectionError(format!("Bot info HTTP error: {error}")) - })?; + .send_authenticated("fetch Feishu bot info", |client, token| { + client + .get(format!("{}/bot/v3/info", FEISHU_API_BASE)) + .bearer_auth(token) + }) + .await?; let status = response.status(); let body: serde_json::Value = response.json().await.map_err(|error| { ChannelError::Other(format!("Failed to parse bot info response: {error}")) @@ -493,8 +566,6 @@ impl FeishuChannel { .and_then(|v| v.as_str()) .ok_or_else(|| ChannelError::Other("No image_key in message".to_string()))?; - let token = self.get_tenant_access_token().await?; - // Use message resource API for downloading message images let url = format!( "{}/im/v1/messages/{}/resources/{}?type=image", @@ -505,24 +576,24 @@ impl FeishuChannel { tracing::debug!(url = %url, image_key = %image_key, message_id = %message_id, "Downloading image from Feishu via message resource API"); let resp = self - .http_client - .get(&url) - .header("Authorization", format!("Bearer {}", token)) - .send() - .await - .map_err(|e| { - ChannelError::ConnectionError(format!("Download image HTTP error: {}", e)) - })?; + .send_authenticated("download Feishu image", |client, token| { + client.get(&url).bearer_auth(token) + }) + .await?; let status = resp.status(); #[cfg(debug_assertions)] tracing::debug!(status = %status, "Image download response status"); if !status.is_success() { - let error_text = resp.text().await.unwrap_or_default(); + let error_body = + read_response_limited(resp, 64 * 1024, self.request_timeout(), "error response") + .await + .unwrap_or_default(); return Err(ChannelError::Other(format!( "Image download failed {}: {}", - status, error_text + status, + String::from_utf8_lossy(&error_body) ))); } @@ -535,11 +606,13 @@ impl FeishuChannel { let ext = resolve_image_ext(&content_type); - let data = resp - .bytes() - .await - .map_err(|e| ChannelError::Other(format!("Failed to read image data: {}", e)))? - .to_vec(); + let data = read_response_limited( + resp, + self.config.max_image_bytes, + self.request_timeout(), + "image", + ) + .await?; #[cfg(debug_assertions)] tracing::debug!(data_len = %data.len(), content_type = %content_type, "Downloaded image data"); @@ -552,6 +625,13 @@ impl FeishuChannel { ); let file_path = resolve_unique_path(media_dir, &filename).await; + ensure_media_capacity( + media_dir, + data.len() as u64, + self.config.media_dir_max_bytes, + ) + .await?; + tokio::fs::write(&file_path, &data) .await .map_err(|e| ChannelError::Other(format!("Failed to write image: {}", e)))?; @@ -576,8 +656,6 @@ impl FeishuChannel { .and_then(|v| v.as_str()) .ok_or_else(|| ChannelError::Other("No file_key in message".to_string()))?; - let token = self.get_tenant_access_token().await?; - // Use message resource API for downloading message files let url = format!( "{}/im/v1/messages/{}/resources/{}?type=file", @@ -588,34 +666,36 @@ impl FeishuChannel { tracing::debug!(url = %url, file_key = %file_key, message_id = %message_id, "Downloading file from Feishu via message resource API"); let resp = self - .http_client - .get(&url) - .header("Authorization", format!("Bearer {}", token)) - .send() - .await - .map_err(|e| { - ChannelError::ConnectionError(format!("Download file HTTP error: {}", e)) - })?; + .send_authenticated("download Feishu file", |client, token| { + client.get(&url).bearer_auth(token) + }) + .await?; let status = resp.status(); if !status.is_success() { - let error_text = resp.text().await.unwrap_or_default(); + let error_body = + read_response_limited(resp, 64 * 1024, self.request_timeout(), "error response") + .await + .unwrap_or_default(); return Err(ChannelError::Other(format!( "File download failed {}: {}", - status, error_text + status, + String::from_utf8_lossy(&error_body) ))); } - let data = resp - .bytes() - .await - .map_err(|e| ChannelError::Other(format!("Failed to read file data: {}", e)))? - .to_vec(); + let data = read_response_limited( + resp, + self.config.max_file_bytes, + self.request_timeout(), + file_type, + ) + .await?; let filename = content_json .get("file_name") .and_then(|v| v.as_str()) - .map(|s| s.to_string()) + .map(sanitize_filename) .unwrap_or_else(|| { let ext = resolve_file_ext(content_json); if ext.is_empty() { @@ -631,6 +711,13 @@ impl FeishuChannel { }); let file_path = resolve_unique_path(media_dir, &filename).await; + ensure_media_capacity( + media_dir, + data.len() as u64, + self.config.media_dir_max_bytes, + ) + .await?; + tokio::fs::write(&file_path, &data) .await .map_err(|e| ChannelError::Other(format!("Failed to write file: {}", e)))?; @@ -644,6 +731,7 @@ impl FeishuChannel { /// Upload image to Feishu and return the image_key async fn upload_image(&self, file_path: &str) -> Result { + ensure_local_media_size(file_path, self.config.max_image_bytes, "image").await?; let token = self.get_tenant_access_token().await?; let mime = mime_guess::from_path(file_path) @@ -720,6 +808,7 @@ impl FeishuChannel { /// Upload file to Feishu and return the file_key async fn upload_file(&self, file_path: &str) -> Result { + ensure_local_media_size(file_path, self.config.max_file_bytes, "file").await?; let token = self.get_tenant_access_token().await?; let file_name = std::path::Path::new(file_path) @@ -812,23 +901,19 @@ impl FeishuChannel { if emoji.is_empty() { return Ok(None); } - let token = self.get_tenant_access_token().await?; - let resp = self - .http_client - .post(format!( - "{}/im/v1/messages/{}/reactions", - FEISHU_API_BASE, message_id - )) - .header("Authorization", format!("Bearer {}", token)) - .json(&serde_json::json!({ - "reaction_type": { "emoji_type": emoji } - })) - .send() - .await - .map_err(|e| { - ChannelError::ConnectionError(format!("Add reaction HTTP error: {}", e)) - })?; + .send_authenticated("add Feishu reaction", |client, token| { + client + .post(format!( + "{}/im/v1/messages/{}/reactions", + FEISHU_API_BASE, message_id + )) + .bearer_auth(token) + .json(&serde_json::json!({ + "reaction_type": { "emoji_type": emoji } + })) + }) + .await?; #[derive(Deserialize)] struct ReactionResp { @@ -884,20 +969,16 @@ impl FeishuChannel { message_id: &str, reaction_id: &str, ) -> Result<(), ChannelError> { - let token = self.get_tenant_access_token().await?; - let resp = self - .http_client - .delete(format!( - "{}/im/v1/messages/{}/reactions/{}", - FEISHU_API_BASE, message_id, reaction_id - )) - .header("Authorization", format!("Bearer {}", token)) - .send() - .await - .map_err(|e| { - ChannelError::ConnectionError(format!("Remove reaction HTTP error: {}", e)) - })?; + .send_authenticated("remove Feishu reaction", |client, token| { + client + .delete(format!( + "{}/im/v1/messages/{}/reactions/{}", + FEISHU_API_BASE, message_id, reaction_id + )) + .bearer_auth(token) + }) + .await?; #[derive(Deserialize)] struct ReactionResp { @@ -926,19 +1007,12 @@ impl FeishuChannel { /// Fetch the text content of a Feishu message by ID. /// Returns a "[Reply to: ...]" context string, or None on failure. async fn get_message_content(&self, message_id: &str) -> Option { - let token = match self.get_tenant_access_token().await { - Ok(t) => t, - Err(e) => { - tracing::debug!(error = %e, message_id = %message_id, "Feishu: failed to get token for fetching parent message"); - return None; - } - }; - let resp = self - .http_client - .get(format!("{}/im/v1/messages/{}", FEISHU_API_BASE, message_id)) - .header("Authorization", format!("Bearer {}", token)) - .send() + .send_authenticated("fetch Feishu parent message", |client, token| { + client + .get(format!("{}/im/v1/messages/{}", FEISHU_API_BASE, message_id)) + .bearer_auth(token) + }) .await .ok()?; @@ -1019,83 +1093,43 @@ impl FeishuChannel { msg_type: &str, content: &str, ) -> Result { - let token = self.get_tenant_access_token().await?; - let (url, body) = target.request(msg_type, content); - - let resp = self - .http_client - .post(url) - .header("Content-Type", "application/json") - .header("Authorization", format!("Bearer {}", token)) - .json(&body) - .send() - .await - .map_err(|e| { - ChannelError::ConnectionError(format!("Send message HTTP error: {}", e)) - })?; - - #[derive(Deserialize)] - struct SendResp { - code: i32, - msg: String, - data: Option, - } - #[derive(Deserialize)] - struct SendData { - message_id: String, - } - - let send_resp: SendResp = resp - .json() - .await - .map_err(|e| ChannelError::Other(format!("Parse send response error: {}", e)))?; - - if matches!(send_resp.code, 230011 | 231003) - && target.reply_to.is_some() - && !target.reply_in_thread - { - tracing::warn!( - code = send_resp.code, - "Feishu reply target is unavailable; falling back to a new chat message" - ); - let mut create_target = target.clone(); - create_target.reply_to = None; - let (url, body) = create_target.request(msg_type, content); + let mut active_target = target.clone(); + let mut token_refreshes = 0; + loop { + let (url, body) = active_target.request(msg_type, content); let response = self - .http_client - .post(url) - .bearer_auth(self.get_tenant_access_token().await?) - .json(&body) - .send() - .await - .map_err(|error| { - ChannelError::ConnectionError(format!( - "Fallback send message HTTP error: {error}" - )) - })?; - let fallback: SendResp = response.json().await.map_err(|error| { - ChannelError::Other(format!("Parse fallback send response error: {error}")) + .send_authenticated("send Feishu message", |client, token| { + client.post(url.clone()).bearer_auth(token).json(&body) + }) + .await?; + let result: FeishuMessageResponse = response.json().await.map_err(|error| { + ChannelError::Other(format!("Parse send response error: {error}")) })?; - if fallback.code != 0 { + + if result.code == FEISHU_INVALID_TOKEN_CODE && token_refreshes == 0 { + token_refreshes += 1; + self.invalidate_tenant_token().await; + continue; + } + if matches!(result.code, 230011 | 231003) + && active_target.reply_to.is_some() + && !active_target.reply_in_thread + { + tracing::warn!( + code = result.code, + "Feishu reply target is unavailable; falling back to a new chat message" + ); + active_target.reply_to = None; + continue; + } + if result.code != 0 { return Err(ChannelError::Other(format!( - "Fallback send failed: code={} msg={}", - fallback.code, fallback.msg + "Send message failed: code={} msg={}", + result.code, result.msg ))); } - return Ok(fallback - .data - .map_or_else(String::new, |data| data.message_id)); + return Ok(result.data.map_or_else(String::new, |data| data.message_id)); } - if send_resp.code != 0 { - return Err(ChannelError::Other(format!( - "Send message failed: code={} msg={}", - send_resp.code, send_resp.msg - ))); - } - - Ok(send_resp - .data - .map_or_else(String::new, |data| data.message_id)) } /// Extract service_id from WebSocket URL query params @@ -1719,6 +1753,120 @@ fn reconnect_delay(attempt: u32) -> Duration { Duration::from_secs(1_u64 << exponent) } +fn api_retry_delay(attempt: usize) -> Duration { + Duration::from_millis(250_u64.saturating_mul(1_u64 << attempt.min(4))) +} + +fn retry_after_delay(headers: &reqwest::header::HeaderMap, attempt: usize) -> Duration { + headers + .get(reqwest::header::RETRY_AFTER) + .and_then(|value| value.to_str().ok()) + .and_then(|value| value.parse::().ok()) + .map(|seconds| Duration::from_secs(seconds.min(5))) + .unwrap_or_else(|| api_retry_delay(attempt)) +} + +async fn read_response_limited( + mut response: reqwest::Response, + max_bytes: u64, + timeout: Duration, + media_type: &str, +) -> Result, ChannelError> { + if response + .content_length() + .is_some_and(|length| length > max_bytes) + { + return Err(ChannelError::Other(format!( + "Feishu {media_type} exceeds {max_bytes} byte limit" + ))); + } + tokio::time::timeout(timeout, async move { + let mut data = Vec::new(); + while let Some(chunk) = response.chunk().await.map_err(|error| { + ChannelError::ConnectionError(format!("Failed to read {media_type} data: {error}")) + })? { + if (data.len() as u64).saturating_add(chunk.len() as u64) > max_bytes { + return Err(ChannelError::Other(format!( + "Feishu {media_type} exceeds {max_bytes} byte limit" + ))); + } + data.extend_from_slice(&chunk); + } + Ok(data) + }) + .await + .map_err(|_| ChannelError::ConnectionError(format!("Feishu {media_type} download timed out")))? +} + +async fn ensure_media_capacity( + media_dir: &Path, + incoming_bytes: u64, + max_bytes: u64, +) -> Result<(), ChannelError> { + let mut total = 0_u64; + let mut entries = tokio::fs::read_dir(media_dir) + .await + .map_err(|error| ChannelError::Other(format!("Failed to inspect media dir: {error}")))?; + while let Some(entry) = entries + .next_entry() + .await + .map_err(|error| ChannelError::Other(format!("Failed to inspect media dir: {error}")))? + { + if let Ok(metadata) = entry.metadata().await + && metadata.is_file() + { + total = total.saturating_add(metadata.len()); + } + } + if total.saturating_add(incoming_bytes) > max_bytes { + return Err(ChannelError::Other(format!( + "Feishu media directory would exceed {max_bytes} byte limit" + ))); + } + Ok(()) +} + +async fn ensure_local_media_size( + file_path: &str, + max_bytes: u64, + media_type: &str, +) -> Result<(), ChannelError> { + let size = tokio::fs::metadata(file_path) + .await + .map_err(|error| ChannelError::Other(format!("Failed to inspect {media_type}: {error}")))? + .len(); + if size > max_bytes { + return Err(ChannelError::Other(format!( + "Outbound {media_type} is {size} bytes, exceeding {max_bytes} byte limit" + ))); + } + Ok(()) +} + +fn sanitize_filename(filename: &str) -> String { + let basename = Path::new(filename) + .file_name() + .and_then(|value| value.to_str()) + .unwrap_or("file.bin"); + let sanitized: String = basename + .chars() + .map(|character| { + if character.is_control() || matches!(character, '/' | '\\' | ':' | '\0') { + '_' + } else { + character + } + }) + .collect(); + let sanitized = sanitized.trim_matches(['.', ' ']); + if sanitized.is_empty() { + "file.bin".to_string() + } else { + let boundary = sanitized.floor_char_boundary(sanitized.len().min(255)); + sanitized[..boundary].to_string() + } +} + fn parse_post_content(content: &str) -> String { /// Extract text from a single post element (text, link, at-mention). fn extract_element(el: &serde_json::Value, out: &mut Vec) { @@ -2296,20 +2444,16 @@ impl FeishuChannel { message_id: &str, card_content: &str, ) -> Result<(), ChannelError> { - let token = self.get_tenant_access_token().await?; let card: serde_json::Value = serde_json::from_str(card_content) .map_err(|error| ChannelError::Other(format!("Invalid card JSON: {error}")))?; let response = self - .http_client - .patch(format!("{}/im/v1/messages/{}", FEISHU_API_BASE, message_id)) - .header("Content-Type", "application/json") - .header("Authorization", format!("Bearer {}", token)) - .json(&serde_json::json!({ "card": card })) - .send() - .await - .map_err(|error| { - ChannelError::ConnectionError(format!("Update card HTTP error: {error}")) - })?; + .send_authenticated("update Feishu card", |client, token| { + client + .patch(format!("{}/im/v1/messages/{}", FEISHU_API_BASE, message_id)) + .bearer_auth(token) + .json(&serde_json::json!({ "card": card })) + }) + .await?; #[derive(Deserialize)] struct UpdateResp { @@ -2895,6 +3039,10 @@ mod tests { reaction_emoji: "THUMBSUP".to_string(), live_updates: false, live_update_interval_ms: 500, + max_image_bytes: 10 * 1024 * 1024, + max_file_bytes: 25 * 1024 * 1024, + media_dir_max_bytes: 512 * 1024 * 1024, + request_timeout_secs: 30, }, Path::new("/tmp"), ) @@ -3159,6 +3307,72 @@ mod tests { assert_eq!(body["receive_id"], "omt_thread"); } + #[test] + fn inbound_filename_is_reduced_to_a_safe_basename() { + assert_eq!(sanitize_filename("../../secret.txt"), "secret.txt"); + assert_eq!(sanitize_filename(".."), "file.bin"); + assert_eq!(sanitize_filename("report:\0.txt"), "report__.txt"); + } + + #[tokio::test] + async fn media_capacity_and_file_size_are_checked_before_writing_or_uploading() { + let dir = tempfile::tempdir().unwrap(); + let existing = dir.path().join("existing.bin"); + tokio::fs::write(&existing, vec![0_u8; 8]).await.unwrap(); + + ensure_media_capacity(dir.path(), 2, 10).await.unwrap(); + assert!(ensure_media_capacity(dir.path(), 3, 10).await.is_err()); + ensure_local_media_size(existing.to_str().unwrap(), 8, "file") + .await + .unwrap(); + assert!( + ensure_local_media_size(existing.to_str().unwrap(), 7, "file") + .await + .is_err() + ); + } + + #[tokio::test] + async fn authenticated_requests_retry_transient_http_failures() { + use std::sync::atomic::{AtomicUsize, Ordering}; + + let attempts = Arc::new(AtomicUsize::new(0)); + let route_attempts = attempts.clone(); + let app = axum::Router::new().route( + "/test", + axum::routing::get(move || { + let route_attempts = route_attempts.clone(); + async move { + if route_attempts.fetch_add(1, Ordering::SeqCst) < 2 { + axum::http::StatusCode::SERVICE_UNAVAILABLE + } else { + axum::http::StatusCode::OK + } + } + }), + ); + let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); + let address = listener.local_addr().unwrap(); + let server = tokio::spawn(async move { axum::serve(listener, app).await.unwrap() }); + let channel = test_channel(); + *channel.tenant_token.write().await = Some(CachedTenantToken { + value: "test-token".to_string(), + refresh_after: Instant::now() + Duration::from_secs(60), + }); + let url = format!("http://{address}/test"); + + let response = channel + .send_authenticated("test request", |client, token| { + client.get(&url).bearer_auth(token) + }) + .await + .unwrap(); + + assert_eq!(response.status(), reqwest::StatusCode::OK); + assert_eq!(attempts.load(Ordering::SeqCst), 3); + server.abort(); + } + #[tokio::test] async fn turn_sink_creates_once_updates_same_card_and_cleans_up_at_finish() { let state = Arc::new(Mutex::new(MockTurnState::default())); diff --git a/src/config/mod.rs b/src/config/mod.rs index 4397ab4..e8fea34 100644 --- a/src/config/mod.rs +++ b/src/config/mod.rs @@ -86,6 +86,14 @@ pub struct FeishuChannelConfig { pub live_updates: bool, #[serde(default = "default_feishu_live_update_interval_ms")] pub live_update_interval_ms: u64, + #[serde(default = "default_feishu_max_image_bytes")] + pub max_image_bytes: u64, + #[serde(default = "default_feishu_max_file_bytes")] + pub max_file_bytes: u64, + #[serde(default = "default_feishu_media_dir_max_bytes")] + pub media_dir_max_bytes: u64, + #[serde(default = "default_feishu_request_timeout_secs")] + pub request_timeout_secs: u64, } fn default_allow_from() -> Vec { @@ -107,6 +115,22 @@ fn default_feishu_live_update_interval_ms() -> u64 { 500 } +fn default_feishu_max_image_bytes() -> u64 { + 10 * 1024 * 1024 +} + +fn default_feishu_max_file_bytes() -> u64 { + 25 * 1024 * 1024 +} + +fn default_feishu_media_dir_max_bytes() -> u64 { + 512 * 1024 * 1024 +} + +fn default_feishu_request_timeout_secs() -> u64 { + 30 +} + #[derive(Debug, Clone, Deserialize, Serialize)] pub struct ProviderConfig { #[serde(rename = "type")]