diff --git a/crates/goose/src/gateway/telegram.rs b/crates/goose/src/gateway/telegram.rs index 00f51185a..9548c2e73 100644 --- a/crates/goose/src/gateway/telegram.rs +++ b/crates/goose/src/gateway/telegram.rs @@ -2,7 +2,8 @@ use super::{ Gateway, GatewayConfig, GatewayHandler, IncomingMessage, OutgoingMessage, PlatformUser, }; use async_trait::async_trait; -use reqwest::Client; +use reqwest::{Client, RequestBuilder, Response}; +use serde::de::DeserializeOwned; use serde::{Deserialize, Serialize}; use tokio_util::sync::CancellationToken; @@ -129,6 +130,22 @@ impl TelegramGateway { format!("{}/bot{}/{}", self.api_base, self.bot_token, method) } + async fn send_request(request: RequestBuilder) -> reqwest::Result { + request.send().await.map_err(reqwest::Error::without_url) + } + + async fn response_json(response: Response) -> reqwest::Result { + response.json().await.map_err(reqwest::Error::without_url) + } + + async fn response_bytes(response: Response) -> reqwest::Result> { + response + .bytes() + .await + .map(Vec::from) + .map_err(reqwest::Error::without_url) + } + async fn get_updates(&self, offset: Option) -> anyhow::Result> { let mut params = serde_json::json!({ "timeout": POLL_TIMEOUT_SECS, @@ -138,15 +155,14 @@ impl TelegramGateway { params["offset"] = serde_json::json!(offset); } - let resp: TelegramResponse> = self - .client - .post(self.api_url("getUpdates")) - .json(¶ms) - .timeout(std::time::Duration::from_secs(POLL_TIMEOUT_SECS + 10)) - .send() - .await? - .json() - .await?; + let response = Self::send_request( + self.client + .post(self.api_url("getUpdates")) + .json(¶ms) + .timeout(std::time::Duration::from_secs(POLL_TIMEOUT_SECS + 10)), + ) + .await?; + let resp: TelegramResponse> = Self::response_json(response).await?; resp.result.ok_or_else(|| { anyhow::anyhow!( @@ -159,34 +175,32 @@ impl TelegramGateway { async fn send_text(&self, chat_id: i64, text: &str) -> anyhow::Result<()> { let chunks = split_message(text, MAX_MESSAGE_LENGTH); for (index, chunk) in chunks.iter().enumerate() { - let resp = self - .client - .post(self.api_url("sendRichMessage")) - .json(&SendRichMessageRequest { + let resp = Self::send_request(self.client.post(self.api_url("sendRichMessage")).json( + &SendRichMessageRequest { chat_id, rich_message: InputRichMessage { markdown: chunk }, - }) - .send() - .await?; + }, + )) + .await?; - if let Ok(body) = resp.json::>().await { + if let Ok(body) = Self::response_json::>(resp).await + { if !body.ok { tracing::warn!( error = body.description.as_deref().unwrap_or("unknown"), "Telegram rejected rich markdown, falling back to plain text" ); for plain_chunk in &chunks[index..] { - let plain_resp = self - .client - .post(self.api_url("sendMessage")) - .json(&serde_json::json!({ - "chat_id": chat_id, - "text": plain_chunk, - })) - .send() - .await? - .json::>() + let plain_response = + Self::send_request(self.client.post(self.api_url("sendMessage")).json( + &serde_json::json!({ + "chat_id": chat_id, + "text": plain_chunk, + }), + )) .await?; + let plain_resp: TelegramResponse = + Self::response_json(plain_response).await?; if !plain_resp.ok { anyhow::bail!( "Telegram sendMessage failed: {}", @@ -202,14 +216,13 @@ impl TelegramGateway { } async fn send_chat_action(&self, chat_id: i64, action: &str) -> anyhow::Result<()> { - self.client - .post(self.api_url("sendChatAction")) - .json(&serde_json::json!({ + Self::send_request(self.client.post(self.api_url("sendChatAction")).json( + &serde_json::json!({ "chat_id": chat_id, "action": action, - })) - .send() - .await?; + }), + )) + .await?; Ok(()) } @@ -220,14 +233,13 @@ impl TelegramGateway { /// 2. Fetch the raw bytes from `https://api.telegram.org/file/bot/`. async fn download_file(&self, file_id: &str) -> anyhow::Result> { // Step 1 – resolve file_id → file_path - let resp: TelegramResponse = self - .client - .post(self.api_url("getFile")) - .json(&serde_json::json!({ "file_id": file_id })) - .send() - .await? - .json() - .await?; + let response = Self::send_request( + self.client + .post(self.api_url("getFile")) + .json(&serde_json::json!({ "file_id": file_id })), + ) + .await?; + let resp: TelegramResponse = Self::response_json(response).await?; let tg_file = resp.result.ok_or_else(|| { anyhow::anyhow!( @@ -245,8 +257,8 @@ impl TelegramGateway { "{}/file/bot{}/{}", TELEGRAM_API_BASE, self.bot_token, file_path ); - let bytes = self.client.get(&download_url).send().await?.bytes().await?; - Ok(bytes.to_vec()) + let response = Self::send_request(self.client.get(&download_url)).await?; + Ok(Self::response_bytes(response).await?) } /// Save voice bytes to a temporary file and return the path. @@ -516,13 +528,8 @@ impl Gateway for TelegramGateway { } async fn validate_config(&self) -> anyhow::Result<()> { - let resp: TelegramResponse = self - .client - .get(self.api_url("getMe")) - .send() - .await? - .json() - .await?; + let response = Self::send_request(self.client.get(self.api_url("getMe"))).await?; + let resp: TelegramResponse = Self::response_json(response).await?; if !resp.ok { anyhow::bail!( @@ -607,9 +614,13 @@ fn split_message(text: &str, max_len: usize) -> Vec { #[cfg(test)] mod tests { use super::*; + use tokio::io::{AsyncReadExt, AsyncWriteExt}; + use tokio::net::TcpListener; use wiremock::matchers::{body_json, method, path}; use wiremock::{Mock, MockServer, ResponseTemplate}; + const SECRET_BOT_TOKEN: &str = "123456789:AASecret_Telegram_Token"; + fn test_gateway(api_base: String) -> TelegramGateway { TelegramGateway { bot_token: "test-token".to_string(), @@ -618,6 +629,139 @@ mod tests { } } + fn secret_gateway(api_base: String) -> TelegramGateway { + TelegramGateway { + bot_token: SECRET_BOT_TOKEN.to_string(), + client: Client::builder().no_proxy().build().unwrap(), + api_base, + } + } + + fn assert_log_fields_are_redacted( + error: &(impl std::fmt::Display + std::fmt::Debug), + diagnostic: &str, + ) { + let display_log_field = format!("{error}"); + let debug_log_field = format!("{error:?}"); + + for rendered in [&display_log_field, &debug_log_field] { + assert!(!rendered.contains(SECRET_BOT_TOKEN), "{rendered}"); + } + assert!( + display_log_field.contains(diagnostic), + "{display_log_field}" + ); + } + + #[tokio::test] + async fn request_errors_remove_token_url_from_display_and_debug() { + let server = MockServer::start().await; + let request_path = format!("/bot{SECRET_BOT_TOKEN}/getMe"); + let redirect_url = format!("{}{request_path}", server.uri()); + + Mock::given(method("GET")) + .and(path(request_path)) + .respond_with( + ResponseTemplate::new(302).append_header("Location", redirect_url.as_str()), + ) + .mount(&server) + .await; + + let error = secret_gateway(server.uri()) + .validate_config() + .await + .unwrap_err(); + + assert_log_fields_are_redacted(&error, "redirect"); + assert!(error + .downcast_ref::() + .unwrap() + .is_redirect()); + assert!(error + .downcast_ref::() + .unwrap() + .url() + .is_none()); + } + + #[tokio::test] + async fn response_errors_remove_token_url_from_display_and_debug() { + let server = MockServer::start().await; + let request_path = format!("/bot{SECRET_BOT_TOKEN}/getUpdates"); + + Mock::given(method("POST")) + .and(path(request_path)) + .respond_with(ResponseTemplate::new(200).set_body_raw("{", "application/json")) + .mount(&server) + .await; + + let error = secret_gateway(server.uri()) + .get_updates(None) + .await + .unwrap_err(); + + assert_log_fields_are_redacted(&error, "decoding response body"); + assert!(error.downcast_ref::().unwrap().is_decode()); + assert!(error + .downcast_ref::() + .unwrap() + .url() + .is_none()); + } + + #[tokio::test] + async fn timeout_errors_remove_token_url_from_display_and_debug() { + let listener = TcpListener::bind("127.0.0.1:0").await.unwrap(); + let addr = listener.local_addr().unwrap(); + tokio::spawn(async move { + let _connection = listener.accept().await.unwrap(); + std::future::pending::<()>().await; + }); + + let gateway = secret_gateway(format!("http://{addr}")); + let url = gateway.api_url("getMe"); + let error = TelegramGateway::send_request( + gateway + .client + .get(url) + .timeout(std::time::Duration::from_millis(20)), + ) + .await + .unwrap_err(); + + assert_log_fields_are_redacted(&error, "error sending request"); + assert!(error.is_timeout()); + assert!(error.url().is_none()); + } + + #[tokio::test] + async fn body_errors_remove_token_url_from_display_and_debug() { + let listener = TcpListener::bind("127.0.0.1:0").await.unwrap(); + let addr = listener.local_addr().unwrap(); + tokio::spawn(async move { + let (mut socket, _) = listener.accept().await.unwrap(); + let mut request = [0; 2048]; + let _ = socket.read(&mut request).await; + socket + .write_all( + b"HTTP/1.1 200 OK\r\ncontent-length: 32\r\nconnection: close\r\n\r\nshort", + ) + .await + .unwrap(); + }); + + let gateway = secret_gateway(format!("http://{addr}")); + let url = format!("http://{addr}/file/bot{SECRET_BOT_TOKEN}/voice.ogg"); + let response = TelegramGateway::send_request(gateway.client.get(url)) + .await + .unwrap(); + let error = TelegramGateway::response_bytes(response).await.unwrap_err(); + + assert_log_fields_are_redacted(&error, "error decoding response body"); + assert!(error.is_decode()); + assert!(error.url().is_none()); + } + #[tokio::test] async fn send_text_uses_rich_markdown() { let server = MockServer::start().await;