1use std::borrow::Cow;
2use std::error::Error as StdError;
3use std::time::Duration;
4
5use chrono::{
6 DateTime,
7 Utc,
8};
9use reqwest::header::HeaderMap;
10use reqwest::Url;
11
12use crate::error::{
13 PluginError,
14 PluginResult,
15};
16
17const MAX_RETRY_AFTER: Duration = Duration::from_secs(3600);
18const EPOCH_THRESHOLD: u64 = 1_000_000_000;
19const BODY_PREVIEW_CHARS: usize = 300;
20const BODY_PREVIEW_BYTES: usize = BODY_PREVIEW_CHARS * 4;
21
22pub async fn read_prefix(mut resp: reqwest::Response, limit: usize) -> reqwest::Result<Vec<u8>> {
23 let mut body = Vec::new();
24 while body.len() < limit {
25 let Some(chunk) = resp.chunk().await? else {
26 break;
27 };
28 let take = chunk.len().min(limit - body.len());
29 body.extend_from_slice(&chunk[..take]);
30 }
31 Ok(body)
32}
33
34pub async fn body_preview(resp: reqwest::Response) -> String {
35 let body = read_prefix(resp, BODY_PREVIEW_BYTES)
36 .await
37 .unwrap_or_default();
38 truncate_chars(String::from_utf8_lossy(&body).trim(), BODY_PREVIEW_CHARS).into_owned()
39}
40
41pub async fn check_response(resp: reqwest::Response) -> PluginResult<reqwest::Response> {
42 let status = resp.status();
43 if status.is_success() || status.is_redirection() {
44 return Ok(resp);
45 }
46
47 let url = redact_url(resp.url());
48 let retry_after = retry_after(resp.headers(), Utc::now());
49 let rate_limit_exhausted = resp
50 .headers()
51 .get("x-ratelimit-remaining")
52 .and_then(|v| v.to_str().ok())
53 .is_some_and(|v| v.trim() == "0");
54 let has_retry_after = resp.headers().contains_key(reqwest::header::RETRY_AFTER);
55 let body = body_preview(resp).await;
56 let message = format!("{} from {}: {}", status.as_u16(), url, body);
57
58 if status.as_u16() == 403 && (rate_limit_exhausted || has_retry_after) {
59 return Err(PluginError::RateLimited {
60 message,
61 retry_after,
62 });
63 }
64
65 Err(from_status(status.as_u16(), message, retry_after))
66}
67
68pub fn from_status(status: u16, message: String, retry_after: Option<Duration>) -> PluginError {
69 match status {
70 401 => PluginError::AuthenticationFailed(message),
71 403 => PluginError::PermissionDenied(message),
72 404 => PluginError::PipelineNotFound(message),
73 429 => PluginError::RateLimited {
74 message,
75 retry_after,
76 },
77 500..=599 => PluginError::ServiceUnavailable {
78 message,
79 retry_after,
80 },
81 _ => PluginError::ApiError(message),
82 }
83}
84
85pub fn retry_after(headers: &HeaderMap, now: DateTime<Utc>) -> Option<Duration> {
86 let header = |name: &str| {
87 headers
88 .get(name)
89 .and_then(|v| v.to_str().ok())
90 .map(str::trim)
91 };
92
93 let parsed = if let Some(value) = header("retry-after") {
94 parse_retry_after(value, now)
95 } else {
96 header("x-ratelimit-reset")
97 .or_else(|| header("ratelimit-reset"))
98 .and_then(|value| parse_reset(value, now))
99 };
100
101 parsed.map(|d| d.min(MAX_RETRY_AFTER))
102}
103
104fn parse_retry_after(value: &str, now: DateTime<Utc>) -> Option<Duration> {
105 if let Ok(secs) = value.parse::<u64>() {
106 return Some(Duration::from_secs(secs));
107 }
108 let date = DateTime::parse_from_rfc2822(value).ok()?;
109 Some(until(date.with_timezone(&Utc), now))
110}
111
112fn parse_reset(value: &str, now: DateTime<Utc>) -> Option<Duration> {
113 let secs = value.parse::<u64>().ok()?;
114 if secs < EPOCH_THRESHOLD {
115 return Some(Duration::from_secs(secs));
116 }
117 let now_secs = u64::try_from(now.timestamp()).unwrap_or(0);
118 Some(Duration::from_secs(secs.saturating_sub(now_secs)))
119}
120
121fn until(target: DateTime<Utc>, now: DateTime<Utc>) -> Duration {
122 (target - now).to_std().unwrap_or(Duration::ZERO)
123}
124
125pub fn map_send_error(err: reqwest::Error, url: &Url) -> PluginError {
126 let tls = is_tls_error(&err);
127 let network = err.is_timeout() || err.is_connect() || err.is_request();
128 let message = format!("{} ({})", err.without_url(), redact_url(url));
129
130 if tls {
131 PluginError::Tls(message)
132 } else if network {
133 PluginError::NetworkError(message)
134 } else {
135 PluginError::ApiError(message)
136 }
137}
138
139fn is_tls_error(err: &(dyn StdError + 'static)) -> bool {
140 let mut current = Some(err);
141 while let Some(e) = current {
142 if e.downcast_ref::<rustls::Error>().is_some() {
143 return true;
144 }
145 if let Some(inner) = e
146 .downcast_ref::<std::io::Error>()
147 .and_then(std::io::Error::get_ref)
148 {
149 if is_tls_error(inner) {
150 return true;
151 }
152 }
153 current = e.source();
154 }
155 false
156}
157
158pub fn truncate_chars(s: &str, max: usize) -> Cow<'_, str> {
159 match s.char_indices().nth(max) {
160 Some((idx, _)) => Cow::Owned(format!("{}…", &s[..idx])),
161 None => Cow::Borrowed(s),
162 }
163}
164
165pub fn redact_url(url: &Url) -> String {
166 let mut clean = url.clone();
167 let _ = clean.set_username("");
168 let _ = clean.set_password(None);
169 clean.set_query(None);
170 clean.set_fragment(None);
171 clean.to_string()
172}
173
174#[cfg(test)]
175mod tests {
176 use std::io::{
177 Read,
178 Write,
179 };
180 use std::net::TcpListener;
181 use std::sync::Arc;
182
183 use reqwest::header::HeaderValue;
184
185 use super::*;
186
187 fn headers(pairs: &[(&'static str, &str)]) -> HeaderMap {
188 let mut map = HeaderMap::new();
189 for (name, value) in pairs {
190 map.insert(*name, HeaderValue::from_str(value).unwrap());
191 }
192 map
193 }
194
195 fn fixed_now() -> DateTime<Utc> {
196 DateTime::parse_from_rfc3339("2026-10-22T12:00:00Z")
197 .unwrap()
198 .with_timezone(&Utc)
199 }
200
201 #[test]
202 fn retry_after_parses_seconds_and_dates() {
203 let now = fixed_now();
204 assert_eq!(
205 retry_after(&headers(&[("retry-after", "120")]), now),
206 Some(Duration::from_secs(120))
207 );
208 assert_eq!(
209 retry_after(
210 &headers(&[("retry-after", "Thu, 22 Oct 2026 12:01:30 GMT")]),
211 now
212 ),
213 Some(Duration::from_secs(90))
214 );
215 assert_eq!(
216 retry_after(
217 &headers(&[("retry-after", "Thu, 22 Oct 2026 11:00:00 GMT")]),
218 now
219 ),
220 Some(Duration::ZERO)
221 );
222 assert_eq!(retry_after(&headers(&[("retry-after", "soon")]), now), None);
223 assert_eq!(retry_after(&HeaderMap::new(), now), None);
224 }
225
226 #[test]
227 fn retry_after_reads_reset_as_epoch_or_delta() {
228 let now = fixed_now();
229 let epoch = (now.timestamp() + 30).to_string();
230 assert_eq!(
231 retry_after(&headers(&[("x-ratelimit-reset", &epoch)]), now),
232 Some(Duration::from_secs(30))
233 );
234 assert_eq!(
235 retry_after(&headers(&[("ratelimit-reset", "45")]), now),
236 Some(Duration::from_secs(45))
237 );
238 assert_eq!(
239 retry_after(&headers(&[("x-ratelimit-reset", "1000000000")]), now),
240 Some(Duration::ZERO)
241 );
242 }
243
244 #[test]
245 fn retry_after_is_capped_and_saturating() {
246 let now = fixed_now();
247 assert_eq!(
248 retry_after(&headers(&[("retry-after", "86400")]), now),
249 Some(MAX_RETRY_AFTER)
250 );
251 assert_eq!(
252 retry_after(&headers(&[("retry-after", &u64::MAX.to_string())]), now),
253 Some(MAX_RETRY_AFTER)
254 );
255 assert_eq!(
256 retry_after(
257 &headers(&[("x-ratelimit-reset", &u64::MAX.to_string())]),
258 now
259 ),
260 Some(MAX_RETRY_AFTER)
261 );
262 }
263
264 #[test]
265 fn retry_after_prefers_retry_after_header() {
266 let now = fixed_now();
267 assert_eq!(
268 retry_after(
269 &headers(&[("retry-after", "5"), ("x-ratelimit-reset", "500")]),
270 now
271 ),
272 Some(Duration::from_secs(5))
273 );
274 }
275
276 #[test]
277 fn truncate_chars_cuts_on_char_boundaries() {
278 assert_eq!(truncate_chars("héllo wörld", 4), "héll…");
279 assert_eq!(truncate_chars("日本語テキスト", 3), "日本語…");
280 assert_eq!(truncate_chars("short", 10), "short");
281 assert!(matches!(truncate_chars("short", 5), Cow::Borrowed(_)));
282 }
283
284 #[test]
285 fn redact_url_drops_userinfo_query_and_fragment() {
286 let url = Url::parse("https://user:secret@example.com/api/v1?token=abc#frag").unwrap();
287 assert_eq!(redact_url(&url), "https://example.com/api/v1");
288 }
289
290 fn serve_once(response: impl Into<String>) -> Url {
291 let response = response.into();
292 let listener = TcpListener::bind("127.0.0.1:0").unwrap();
293 let addr = listener.local_addr().unwrap();
294 std::thread::spawn(move || {
295 if let Ok((mut stream, _)) = listener.accept() {
296 let mut buf = [0u8; 4096];
297 let _ = stream.read(&mut buf);
298 let _ = stream.write_all(response.as_bytes());
299 }
300 });
301 Url::parse(&format!("http://{addr}/path?token=secret")).unwrap()
302 }
303
304 fn serve_held_open(head: &'static str, sent: usize) -> Url {
305 let listener = TcpListener::bind("127.0.0.1:0").unwrap();
306 let addr = listener.local_addr().unwrap();
307 std::thread::spawn(move || {
308 if let Ok((mut stream, _)) = listener.accept() {
309 let mut buf = [0u8; 4096];
310 let _ = stream.read(&mut buf);
311 let _ = stream.write_all(head.as_bytes());
312 let _ = stream.write_all(&vec![b'x'; sent]);
313 let _ = stream.read(&mut buf);
314 }
315 });
316 Url::parse(&format!("http://{addr}/")).unwrap()
317 }
318
319 #[tokio::test]
320 async fn reads_stop_at_the_limit_without_waiting_for_a_huge_body() {
321 let huge = "HTTP/1.1 200 OK\r\nContent-Length: 1073741824\r\n\r\n";
322 let resp = client()
323 .get(serve_held_open(huge, 8_000))
324 .send()
325 .await
326 .unwrap();
327 let body = tokio::time::timeout(Duration::from_secs(5), read_prefix(resp, 5_000))
328 .await
329 .unwrap()
330 .unwrap();
331 assert_eq!(body.len(), 5_000);
332
333 let failed = "HTTP/1.1 500 Internal Server Error\r\nContent-Length: 1073741824\r\n\r\n";
334 let resp = client()
335 .get(serve_held_open(failed, 8_000))
336 .send()
337 .await
338 .unwrap();
339 let err = tokio::time::timeout(Duration::from_secs(5), check_response(resp))
340 .await
341 .unwrap()
342 .unwrap_err();
343 assert!(err
344 .to_string()
345 .contains(&format!("{}…", "x".repeat(BODY_PREVIEW_CHARS))));
346 }
347
348 async fn status_error(response: impl Into<String>) -> PluginError {
349 let url = serve_once(response);
350 let resp = client().get(url).send().await.unwrap();
351 check_response(resp).await.unwrap_err()
352 }
353
354 #[tokio::test]
355 async fn check_response_maps_every_status_row() {
356 let ok = serve_once("HTTP/1.1 204 No Content\r\nContent-Length: 0\r\n\r\n");
357 let resp = client().get(ok).send().await.unwrap();
358 assert!(check_response(resp).await.is_ok());
359
360 let redirect = serve_once("HTTP/1.1 304 Not Modified\r\nContent-Length: 0\r\n\r\n");
361 let resp = client().get(redirect).send().await.unwrap();
362 assert!(check_response(resp).await.is_ok());
363
364 assert!(matches!(
365 status_error("HTTP/1.1 401 Unauthorized\r\nContent-Length: 4\r\n\r\nnope").await,
366 PluginError::AuthenticationFailed(_)
367 ));
368 assert!(matches!(
369 status_error(
370 "HTTP/1.1 403 Forbidden\r\nx-ratelimit-remaining: 0\r\nContent-Length: 0\r\n\r\n"
371 )
372 .await,
373 PluginError::RateLimited { .. }
374 ));
375 assert!(matches!(
376 status_error("HTTP/1.1 403 Forbidden\r\nretry-after: 30\r\nContent-Length: 0\r\n\r\n")
377 .await,
378 PluginError::RateLimited {
379 retry_after: Some(d),
380 ..
381 } if d == Duration::from_secs(30)
382 ));
383 assert!(matches!(
384 status_error("HTTP/1.1 403 Forbidden\r\nContent-Length: 0\r\n\r\n").await,
385 PluginError::PermissionDenied(_)
386 ));
387 assert!(matches!(
388 status_error("HTTP/1.1 404 Not Found\r\nContent-Length: 0\r\n\r\n").await,
389 PluginError::PipelineNotFound(_)
390 ));
391 assert!(matches!(
392 status_error(
393 "HTTP/1.1 429 Too Many Requests\r\nretry-after: 7\r\nContent-Length: 0\r\n\r\n"
394 )
395 .await,
396 PluginError::RateLimited {
397 retry_after: Some(d),
398 ..
399 } if d == Duration::from_secs(7)
400 ));
401 assert!(matches!(
402 status_error("HTTP/1.1 503 Service Unavailable\r\nContent-Length: 0\r\n\r\n").await,
403 PluginError::ServiceUnavailable {
404 retry_after: None,
405 ..
406 }
407 ));
408 assert!(matches!(
409 status_error("HTTP/1.1 418 I'm a teapot\r\nContent-Length: 0\r\n\r\n").await,
410 PluginError::ApiError(_)
411 ));
412 }
413
414 #[tokio::test]
415 async fn check_response_message_is_redacted_and_truncated() {
416 let content = "é".repeat(400);
417 let body = format!(
418 "HTTP/1.1 500 Internal Server Error\r\nContent-Length: {}\r\n\r\n{content}",
419 content.len()
420 );
421 let err = status_error(body).await;
422 let text = err.to_string();
423 assert!(!text.contains("token=secret"));
424 assert!(text.contains("500 from http://127.0.0.1"));
425 assert!(text.ends_with('…'));
426 }
427
428 fn install_crypto_provider() {
429 let _ = rustls::crypto::ring::default_provider().install_default();
430 }
431
432 fn client() -> reqwest::Client {
433 install_crypto_provider();
434 reqwest::Client::new()
435 }
436
437 fn spawn_tls_server() -> Url {
438 install_crypto_provider();
439 let cert = rustls::pki_types::CertificateDer::from(
440 include_bytes!("../tests/fixtures/self_signed_cert.der").to_vec(),
441 );
442 let key = rustls::pki_types::PrivateKeyDer::Pkcs8(
443 include_bytes!("../tests/fixtures/self_signed_key.der")
444 .to_vec()
445 .into(),
446 );
447 let config = Arc::new(
448 rustls::ServerConfig::builder()
449 .with_no_client_auth()
450 .with_single_cert(vec![cert], key)
451 .unwrap(),
452 );
453 let listener = TcpListener::bind("127.0.0.1:0").unwrap();
454 let addr = listener.local_addr().unwrap();
455 std::thread::spawn(move || {
456 if let Ok((mut stream, _)) = listener.accept() {
457 if let Ok(mut conn) = rustls::ServerConnection::new(config) {
458 let _ = conn.complete_io(&mut stream);
459 }
460 }
461 });
462 Url::parse(&format!(
463 "https://localhost:{}/api?token=secret",
464 addr.port()
465 ))
466 .unwrap()
467 }
468
469 #[tokio::test]
470 async fn map_send_error_detects_untrusted_certificate() {
471 let url = spawn_tls_server();
472 let err = client().get(url.clone()).send().await.unwrap_err();
473 let mapped = map_send_error(err, &url);
474 assert!(matches!(mapped, PluginError::Tls(_)), "{mapped:?}");
475 assert!(!mapped.to_string().contains("token=secret"));
476 }
477
478 #[tokio::test]
479 async fn map_send_error_maps_connect_failure_to_network() {
480 let listener = TcpListener::bind("127.0.0.1:0").unwrap();
481 let addr = listener.local_addr().unwrap();
482 drop(listener);
483 let url = Url::parse(&format!("http://{addr}/?token=secret")).unwrap();
484 let err = client().get(url.clone()).send().await.unwrap_err();
485 let mapped = map_send_error(err, &url);
486 assert!(matches!(mapped, PluginError::NetworkError(_)), "{mapped:?}");
487 assert!(!mapped.to_string().contains("token=secret"));
488 }
489}