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