fix: redact Telegram tokens from request errors (#11343)
Signed-off-by: Jasper Hugo <jasper@spiral.xyz>
This commit is contained in:
Jasper
@@ -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<Response> {
|
||||
request.send().await.map_err(reqwest::Error::without_url)
|
||||
}
|
||||
|
||||
async fn response_json<T: DeserializeOwned>(response: Response) -> reqwest::Result<T> {
|
||||
response.json().await.map_err(reqwest::Error::without_url)
|
||||
}
|
||||
|
||||
async fn response_bytes(response: Response) -> reqwest::Result<Vec<u8>> {
|
||||
response
|
||||
.bytes()
|
||||
.await
|
||||
.map(Vec::from)
|
||||
.map_err(reqwest::Error::without_url)
|
||||
}
|
||||
|
||||
async fn get_updates(&self, offset: Option<i64>) -> anyhow::Result<Vec<TelegramUpdate>> {
|
||||
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<Vec<TelegramUpdate>> = 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<Vec<TelegramUpdate>> = 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::<TelegramResponse<serde_json::Value>>().await {
|
||||
if let Ok(body) = Self::response_json::<TelegramResponse<serde_json::Value>>(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::<TelegramResponse<serde_json::Value>>()
|
||||
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<serde_json::Value> =
|
||||
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<TOKEN>/<file_path>`.
|
||||
async fn download_file(&self, file_id: &str) -> anyhow::Result<Vec<u8>> {
|
||||
// Step 1 – resolve file_id → file_path
|
||||
let resp: TelegramResponse<TelegramFile> = 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<TelegramFile> = 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<serde_json::Value> = 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<serde_json::Value> = Self::response_json(response).await?;
|
||||
|
||||
if !resp.ok {
|
||||
anyhow::bail!(
|
||||
@@ -607,9 +614,13 @@ fn split_message(text: &str, max_len: usize) -> Vec<String> {
|
||||
#[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::<reqwest::Error>()
|
||||
.unwrap()
|
||||
.is_redirect());
|
||||
assert!(error
|
||||
.downcast_ref::<reqwest::Error>()
|
||||
.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::<reqwest::Error>().unwrap().is_decode());
|
||||
assert!(error
|
||||
.downcast_ref::<reqwest::Error>()
|
||||
.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;
|
||||
|
||||
Reference in New Issue
Block a user