netget/client/http_fetch/
mod.rs

1//! The HTTP round trip the HTTP-family clients share, on both targets.
2//!
3//! Natively a [`FetchClient`] wraps the `reqwest::Client` each protocol already builds (its own
4//! timeout, the literal-IP resolver bypass, its connection pool), and every call is reqwest's
5//! own: what goes on the wire is exactly what reqwest sends. In the browser build reqwest cannot
6//! serve (its wasm backend is the browser's `fetch`, which is not `Send` and cannot reach the
7//! page's virtual loopback), so the same calls go through [`transport`] — hyper's HTTP/1.1
8//! client over `tokio::net::TcpStream`, which on wasm32 is `crates/netget-tokio-wasm`'s virtual
9//! network. The client logic above the round trip is one copy.
10//!
11//! The API is the subset of reqwest's the clients use — `get`/`post`/…, `header`, `query`,
12//! `json`, `form`, `body`, `basic_auth`, `timeout`, `send`, and on the response `status`, `headers`,
13//! `text`, `json`, `bytes`, `chunk` — so converting a client is a type change, not a rewrite.
14//!
15//! The transport backend is also constructible natively ([`FetchClient::transport`]), which is
16//! how `tests/client/http/fetch_client_test.rs` pins it down without a browser.
17//!
18//! Differences on the transport, stated: no TLS (`https://` is refused with
19//! [`transport::HTTPS_UNSUPPORTED`] — call [`check_url`] where a client takes its target so the
20//! refusal comes at connect rather than on the first request), no redirects, one connection per
21//! request, and the whole response body is read up front against [`FetchClient`]'s body bound
22//! (8 MiB unless [`FetchClient::with_max_body`] says otherwise), so `chunk()` yields it once.
23
24pub mod transport;
25
26use std::time::Duration;
27
28use anyhow::{anyhow, Context, Result};
29use bytes::Bytes;
30use hyper::header::{HeaderMap, CONTENT_TYPE};
31use hyper::{Method, StatusCode};
32use serde::de::DeserializeOwned;
33use serde::Serialize;
34
35/// An HTTP client: reqwest natively, [`transport`] in the browser (or wherever
36/// [`FetchClient::transport`] is asked for).
37#[derive(Clone)]
38pub struct FetchClient {
39    backend: Backend,
40}
41
42#[derive(Clone)]
43enum Backend {
44    #[cfg(not(target_arch = "wasm32"))]
45    Reqwest(reqwest::Client, usize),
46    Transport {
47        timeout: Duration,
48        max_body: usize,
49        user_agent: Option<String>,
50        wire: transport::Wire,
51    },
52}
53
54impl FetchClient {
55    /// Wrap a reqwest client with the shared buffered-response bound.
56    #[cfg(not(target_arch = "wasm32"))]
57    pub fn from_reqwest(client: reqwest::Client) -> Self {
58        Self {
59            backend: Backend::Reqwest(client, transport::MAX_RESPONSE_BODY_BYTES),
60        }
61    }
62
63    /// The hyper transport: one connection per request, each exchange bounded by `timeout`
64    /// and the response body by [`transport::MAX_RESPONSE_BODY_BYTES`].
65    pub fn transport(timeout: Duration) -> Self {
66        Self {
67            backend: Backend::Transport {
68                timeout,
69                max_body: transport::MAX_RESPONSE_BODY_BYTES,
70                user_agent: None,
71                wire: transport::Wire::Http1,
72            },
73        }
74    }
75
76    /// Bound buffered `bytes`, `text` and `json` responses on both backends.
77    /// Native `chunk()` consumers still stream without an aggregate download cap.
78    pub fn with_max_body(mut self, max_body: usize) -> Self {
79        match &mut self.backend {
80            #[cfg(not(target_arch = "wasm32"))]
81            Backend::Reqwest(_, bound) => *bound = max_body,
82            Backend::Transport {
83                max_body: bound, ..
84            } => *bound = max_body,
85        }
86        self
87    }
88
89    /// Send `User-Agent: user_agent` on every transport request that sets none of its own —
90    /// what reqwest's `ClientBuilder::user_agent` does for a reqwest-backed client, whose
91    /// builder already carries it.
92    #[allow(irrefutable_let_patterns)]
93    pub fn with_user_agent(mut self, user_agent: &str) -> Self {
94        if let Backend::Transport { user_agent: ua, .. } = &mut self.backend {
95            *ua = Some(user_agent.to_string());
96        }
97        self
98    }
99
100    /// Speak HTTP/2 with prior knowledge on the transport (cleartext h2c, no upgrade), as
101    /// reqwest's `ClientBuilder::http2_prior_knowledge` does for a reqwest-backed client,
102    /// whose builder already carries it.
103    #[allow(irrefutable_let_patterns)]
104    pub fn http2_prior_knowledge(mut self) -> Self {
105        if let Backend::Transport { wire, .. } = &mut self.backend {
106            *wire = transport::Wire::Http2PriorKnowledge;
107        }
108        self
109    }
110
111    pub fn request(&self, method: Method, url: &str) -> FetchRequest {
112        match &self.backend {
113            #[cfg(not(target_arch = "wasm32"))]
114            Backend::Reqwest(client, max_body) => FetchRequest {
115                inner: RequestInner::Reqwest(client.request(method, url), *max_body),
116            },
117            Backend::Transport {
118                timeout,
119                max_body,
120                user_agent,
121                wire,
122            } => FetchRequest {
123                inner: RequestInner::Transport(TransportRequest {
124                    method,
125                    url: url.to_string(),
126                    headers: Vec::new(),
127                    user_agent: user_agent.clone(),
128                    wire: *wire,
129                    body: None,
130                    timeout: *timeout,
131                    max_body: *max_body,
132                    error: None,
133                }),
134            },
135        }
136    }
137
138    pub fn get(&self, url: &str) -> FetchRequest {
139        self.request(Method::GET, url)
140    }
141
142    pub fn post(&self, url: &str) -> FetchRequest {
143        self.request(Method::POST, url)
144    }
145
146    pub fn put(&self, url: &str) -> FetchRequest {
147        self.request(Method::PUT, url)
148    }
149
150    pub fn delete(&self, url: &str) -> FetchRequest {
151        self.request(Method::DELETE, url)
152    }
153
154    pub fn head(&self, url: &str) -> FetchRequest {
155        self.request(Method::HEAD, url)
156    }
157
158    pub fn patch(&self, url: &str) -> FetchRequest {
159        self.request(Method::PATCH, url)
160    }
161}
162
163/// Refuse a target the browser transport cannot reach, with the reason; natively every URL
164/// reqwest accepts is accepted. Call it where a client takes its base URL, so an `https://`
165/// target in the browser fails at connect, naming why, rather than on its first request.
166pub fn check_url(url: &str) -> Result<()> {
167    #[cfg(target_arch = "wasm32")]
168    transport::parse_http_url(url)?;
169    #[cfg(not(target_arch = "wasm32"))]
170    let _ = url;
171    Ok(())
172}
173
174/// `scheme://host[:port]` of `url`, host lowercased and a default port dropped, so two
175/// spellings of one origin compare equal and two origins never do.
176pub fn origin_of(url: &str) -> Result<String> {
177    let parsed = url::Url::parse(url).with_context(|| format!("{url:?} is not a URL"))?;
178    anyhow::ensure!(
179        matches!(parsed.scheme(), "http" | "https") && parsed.host_str().is_some(),
180        "{url:?} is not an http(s) URL with a host"
181    );
182    Ok(parsed.origin().ascii_serialization())
183}
184
185/// Resolve an action's `path` against the client's base URL, refusing any other origin.
186///
187/// The HTTP-family clients put the startup `default_headers` (API keys, cookies) and, for
188/// WebDAV, the `auth` credential as `Authorization: Basic` on every request, and the model
189/// that chooses `path` reads the peer's responses. So a client is bound to one origin, its
190/// `remote_addr`, and every URL this returns is on it; anything else is refused by name, so
191/// the repair loop can see it and the operator can open a client for that host on purpose.
192/// A `base_url` without a scheme is `http://`.
193///
194/// - An absolute `http(s)://` `path` (any letter case) is accepted when its origin is the
195///   base's — what a model copying a `Location` header produces.
196/// - Any other `path` is appended to the base, with a `/` inserted unless it already starts
197///   with `/`, `?` or `#` or the base ends with `/`. The separator is what keeps the text in
198///   the path: appended bare, `@attacker.example/x` would become the host (the base turned
199///   into userinfo), `.attacker.example/` would extend the host and `:9999/` would set the
200///   port.
201/// - Whatever the route, the final URL is parsed with the WHATWG parser (`url`, the one
202///   reqwest uses), its scheme, host and port compared with the base's, and userinfo the
203///   base did not carry is refused. The returned string is that parse's serialization, so
204///   the request is sent to exactly the URL that was checked.
205pub fn resolve_same_origin(base_url: &str, path: &str) -> Result<String> {
206    let base = if base_url.contains("://") {
207        base_url.to_string()
208    } else {
209        format!("http://{base_url}")
210    };
211    let bound = origin_of(&base)?;
212    let base_parsed = url::Url::parse(&base).with_context(|| format!("{base:?} is not a URL"))?;
213
214    let has_scheme = |scheme: &str| {
215        path.as_bytes()
216            .get(..scheme.len())
217            .is_some_and(|head| head.eq_ignore_ascii_case(scheme.as_bytes()))
218    };
219    let is_absolute = has_scheme("http://") || has_scheme("https://");
220    let candidate = if is_absolute {
221        path.to_string()
222    } else if path.is_empty() || path.starts_with(['/', '?', '#']) || base.ends_with('/') {
223        format!("{base}{path}")
224    } else {
225        format!("{base}/{path}")
226    };
227
228    let resolved = url::Url::parse(&candidate)
229        .with_context(|| format!("path {path:?} does not form a URL"))?;
230    let asked = origin_of(resolved.as_str())?;
231    anyhow::ensure!(
232        bound == asked,
233        "path {path:?} is on {asked}, but this client is bound to {bound}; use a path \
234         relative to that origin, or open a client for {asked}"
235    );
236    anyhow::ensure!(
237        resolved.username() == base_parsed.username()
238            && resolved.password() == base_parsed.password(),
239        "path {path:?} carries credentials in the URL, which this client bound to {bound} \
240         refuses; set an Authorization header instead"
241    );
242    Ok(resolved.into())
243}
244
245/// A redirect policy that follows up to five hops, none of them to another origin.
246///
247/// reqwest's default follows ten and strips `Authorization` and `Cookie` on a host change,
248/// but not a startup `X-Api-Key`, and a redirect to an internal address is the second half
249/// of the SSRF that `resolve_same_origin` closes on the first request. A 3xx to another
250/// origin is returned to the model as the response it is.
251#[cfg(not(target_arch = "wasm32"))]
252pub fn same_origin_redirects() -> reqwest::redirect::Policy {
253    reqwest::redirect::Policy::custom(|attempt| {
254        let first = attempt
255            .previous()
256            .first()
257            .map(|u| u.origin().ascii_serialization());
258        let next = attempt.url().origin().ascii_serialization();
259        if attempt.previous().len() >= 5 || first.as_deref() != Some(next.as_str()) {
260            attempt.stop()
261        } else {
262            attempt.follow()
263        }
264    })
265}
266
267/// One request being built. Errors in a header or a body are held until [`Self::send`], as
268/// reqwest's builder does.
269pub struct FetchRequest {
270    inner: RequestInner,
271}
272
273enum RequestInner {
274    #[cfg(not(target_arch = "wasm32"))]
275    Reqwest(reqwest::RequestBuilder, usize),
276    Transport(TransportRequest),
277}
278
279struct TransportRequest {
280    method: Method,
281    url: String,
282    headers: Vec<(String, String)>,
283    body: Option<Bytes>,
284    user_agent: Option<String>,
285    wire: transport::Wire,
286    timeout: Duration,
287    max_body: usize,
288    error: Option<anyhow::Error>,
289}
290
291impl TransportRequest {
292    fn has_header(&self, name: &str) -> bool {
293        self.headers
294            .iter()
295            .any(|(n, _)| n.eq_ignore_ascii_case(name))
296    }
297
298    fn fail(&mut self, error: anyhow::Error) {
299        if self.error.is_none() {
300            self.error = Some(error);
301        }
302    }
303}
304
305impl FetchRequest {
306    /// Add a header. Adding the same name twice sends both, as reqwest does.
307    pub fn header(self, name: impl AsRef<str>, value: impl AsRef<str>) -> Self {
308        match self.inner {
309            #[cfg(not(target_arch = "wasm32"))]
310            RequestInner::Reqwest(builder, max_body) => Self {
311                inner: RequestInner::Reqwest(
312                    builder.header(name.as_ref(), value.as_ref()),
313                    max_body,
314                ),
315            },
316            RequestInner::Transport(mut req) => {
317                req.headers
318                    .push((name.as_ref().to_string(), value.as_ref().to_string()));
319                Self {
320                    inner: RequestInner::Transport(req),
321                }
322            }
323        }
324    }
325
326    /// Serialize `value` as the JSON body, with `Content-Type: application/json` unless one
327    /// is already set.
328    pub fn json<T: Serialize + ?Sized>(self, value: &T) -> Self {
329        match self.inner {
330            #[cfg(not(target_arch = "wasm32"))]
331            RequestInner::Reqwest(builder, max_body) => Self {
332                inner: RequestInner::Reqwest(builder.json(value), max_body),
333            },
334            RequestInner::Transport(mut req) => {
335                match serde_json::to_vec(value) {
336                    Ok(body) => {
337                        if !req.has_header(CONTENT_TYPE.as_str()) {
338                            req.headers.push((
339                                CONTENT_TYPE.as_str().to_string(),
340                                "application/json".to_string(),
341                            ));
342                        }
343                        req.body = Some(Bytes::from(body));
344                    }
345                    Err(e) => req.fail(anyhow!("serialize the JSON request body: {e}")),
346                }
347                Self {
348                    inner: RequestInner::Transport(req),
349                }
350            }
351        }
352    }
353
354    /// Append `query`, serialized as `application/x-www-form-urlencoded` pairs, to the URL's
355    /// query string.
356    pub fn query<T: Serialize + ?Sized>(self, query: &T) -> Self {
357        match self.inner {
358            #[cfg(not(target_arch = "wasm32"))]
359            RequestInner::Reqwest(builder, max_body) => Self {
360                inner: RequestInner::Reqwest(builder.query(query), max_body),
361            },
362            RequestInner::Transport(mut req) => {
363                match serde_urlencoded::to_string(query) {
364                    Ok(encoded) if encoded.is_empty() => {}
365                    Ok(encoded) => {
366                        let separator = if req.url.contains('?') { '&' } else { '?' };
367                        req.url = format!("{}{separator}{encoded}", req.url);
368                    }
369                    Err(e) => req.fail(anyhow!("serialize the query string: {e}")),
370                }
371                Self {
372                    inner: RequestInner::Transport(req),
373                }
374            }
375        }
376    }
377
378    /// Serialize `value` as an `application/x-www-form-urlencoded` body.
379    pub fn form<T: Serialize + ?Sized>(self, value: &T) -> Self {
380        match self.inner {
381            #[cfg(not(target_arch = "wasm32"))]
382            RequestInner::Reqwest(builder, max_body) => Self {
383                inner: RequestInner::Reqwest(builder.form(value), max_body),
384            },
385            RequestInner::Transport(mut req) => {
386                match serde_urlencoded::to_string(value) {
387                    Ok(body) => {
388                        if !req.has_header(CONTENT_TYPE.as_str()) {
389                            req.headers.push((
390                                CONTENT_TYPE.as_str().to_string(),
391                                "application/x-www-form-urlencoded".to_string(),
392                            ));
393                        }
394                        req.body = Some(Bytes::from(body));
395                    }
396                    Err(e) => req.fail(anyhow!("serialize the form request body: {e}")),
397                }
398                Self {
399                    inner: RequestInner::Transport(req),
400                }
401            }
402        }
403    }
404
405    /// The raw request body.
406    pub fn body(self, body: impl Into<Bytes>) -> Self {
407        let body = body.into();
408        match self.inner {
409            #[cfg(not(target_arch = "wasm32"))]
410            RequestInner::Reqwest(builder, max_body) => Self {
411                inner: RequestInner::Reqwest(builder.body(body), max_body),
412            },
413            RequestInner::Transport(mut req) => {
414                req.body = Some(body);
415                Self {
416                    inner: RequestInner::Transport(req),
417                }
418            }
419        }
420    }
421
422    /// HTTP Basic authentication, as reqwest writes it: `user:password` (the colon kept when
423    /// there is no password), base64, in `Authorization`.
424    pub fn basic_auth(
425        self,
426        user: impl std::fmt::Display,
427        password: Option<impl std::fmt::Display>,
428    ) -> Self {
429        match self.inner {
430            #[cfg(not(target_arch = "wasm32"))]
431            RequestInner::Reqwest(builder, max_body) => Self {
432                inner: RequestInner::Reqwest(builder.basic_auth(user, password), max_body),
433            },
434            RequestInner::Transport(mut req) => {
435                use base64::Engine;
436                let credentials = match password {
437                    Some(password) => format!("{user}:{password}"),
438                    None => format!("{user}:"),
439                };
440                let encoded = base64::engine::general_purpose::STANDARD.encode(credentials);
441                req.headers.push((
442                    hyper::header::AUTHORIZATION.as_str().to_string(),
443                    format!("Basic {encoded}"),
444                ));
445                Self {
446                    inner: RequestInner::Transport(req),
447                }
448            }
449        }
450    }
451
452    /// Bound this request at `timeout` instead of the client's.
453    pub fn timeout(self, timeout: Duration) -> Self {
454        match self.inner {
455            #[cfg(not(target_arch = "wasm32"))]
456            RequestInner::Reqwest(builder, max_body) => Self {
457                inner: RequestInner::Reqwest(builder.timeout(timeout), max_body),
458            },
459            RequestInner::Transport(mut req) => {
460                req.timeout = timeout;
461                Self {
462                    inner: RequestInner::Transport(req),
463                }
464            }
465        }
466    }
467
468    /// Send the request and read the response head (natively) or the whole response (on the
469    /// transport). A status that is not 2xx is a response, not an error.
470    pub async fn send(self) -> Result<FetchResponse> {
471        match self.inner {
472            #[cfg(not(target_arch = "wasm32"))]
473            RequestInner::Reqwest(builder, max_body) => Ok(FetchResponse {
474                inner: ResponseInner::Reqwest(builder.send().await?, max_body),
475            }),
476            RequestInner::Transport(mut req) => {
477                if let Some(error) = req.error.take() {
478                    return Err(error);
479                }
480                if let Some(user_agent) = req.user_agent.take() {
481                    if !req.has_header(hyper::header::USER_AGENT.as_str()) {
482                        req.headers
483                            .push((hyper::header::USER_AGENT.as_str().to_string(), user_agent));
484                    }
485                }
486                let response = transport::fetch_response_on(
487                    req.wire,
488                    req.method.as_str(),
489                    &req.url,
490                    &req.headers,
491                    req.body,
492                    req.timeout,
493                    req.max_body,
494                )
495                .await?;
496                let (parts, body) = response.into_parts();
497                Ok(FetchResponse {
498                    inner: ResponseInner::Transport {
499                        status: parts.status,
500                        version: parts.version,
501                        headers: parts.headers,
502                        body: Some(body),
503                    },
504                })
505            }
506        }
507    }
508}
509
510/// A response. Natively reqwest's, streamed as reqwest streams it; on the transport already
511/// read in full (bounded), so `chunk` yields the whole body once.
512pub struct FetchResponse {
513    inner: ResponseInner,
514}
515
516enum ResponseInner {
517    #[cfg(not(target_arch = "wasm32"))]
518    Reqwest(reqwest::Response, usize),
519    Transport {
520        status: StatusCode,
521        version: hyper::Version,
522        headers: HeaderMap,
523        body: Option<Bytes>,
524    },
525}
526
527impl FetchResponse {
528    pub fn status(&self) -> StatusCode {
529        match &self.inner {
530            #[cfg(not(target_arch = "wasm32"))]
531            ResponseInner::Reqwest(response, _) => response.status(),
532            ResponseInner::Transport { status, .. } => *status,
533        }
534    }
535
536    /// The HTTP version the response came in.
537    pub fn version(&self) -> hyper::Version {
538        match &self.inner {
539            #[cfg(not(target_arch = "wasm32"))]
540            ResponseInner::Reqwest(response, _) => response.version(),
541            ResponseInner::Transport { version, .. } => *version,
542        }
543    }
544
545    pub fn headers(&self) -> &HeaderMap {
546        match &self.inner {
547            #[cfg(not(target_arch = "wasm32"))]
548            ResponseInner::Reqwest(response, _) => response.headers(),
549            ResponseInner::Transport { headers, .. } => headers,
550        }
551    }
552
553    /// The body's length as the response declared it, where it did.
554    pub fn content_length(&self) -> Option<u64> {
555        match &self.inner {
556            #[cfg(not(target_arch = "wasm32"))]
557            ResponseInner::Reqwest(response, _) => response.content_length(),
558            ResponseInner::Transport { body, headers, .. } => {
559                body.as_ref().map(|b| b.len() as u64).or_else(|| {
560                    headers
561                        .get(hyper::header::CONTENT_LENGTH)
562                        .and_then(|v| v.to_str().ok())
563                        .and_then(|v| v.parse().ok())
564                })
565            }
566        }
567    }
568
569    /// The next piece of the body, `None` at its end.
570    pub async fn chunk(&mut self) -> Result<Option<Bytes>> {
571        match &mut self.inner {
572            #[cfg(not(target_arch = "wasm32"))]
573            ResponseInner::Reqwest(response, _) => Ok(response.chunk().await?),
574            ResponseInner::Transport { body, .. } => Ok(body.take().filter(|b| !b.is_empty())),
575        }
576    }
577
578    /// The whole body.
579    pub async fn bytes(self) -> Result<Bytes> {
580        match self.inner {
581            #[cfg(not(target_arch = "wasm32"))]
582            ResponseInner::Reqwest(response, limit) => read_response_bytes(response, limit).await,
583            ResponseInner::Transport { body, .. } => Ok(body.unwrap_or_default()),
584        }
585    }
586
587    /// The whole body as text. reqwest decodes by the declared charset; the transport reads
588    /// UTF-8, lossily.
589    pub async fn text(self) -> Result<String> {
590        match self.inner {
591            #[cfg(not(target_arch = "wasm32"))]
592            ResponseInner::Reqwest(response, limit) => read_response_text(response, limit).await,
593            ResponseInner::Transport { body, .. } => {
594                Ok(String::from_utf8_lossy(&body.unwrap_or_default()).into_owned())
595            }
596        }
597    }
598
599    /// The whole body parsed as JSON.
600    pub async fn json<T: DeserializeOwned>(self) -> Result<T> {
601        match self.inner {
602            #[cfg(not(target_arch = "wasm32"))]
603            ResponseInner::Reqwest(response, limit) => read_response_json(response, limit).await,
604            ResponseInner::Transport { body, .. } => {
605                serde_json::from_slice(&body.unwrap_or_default())
606                    .context("decode the response body as JSON")
607            }
608        }
609    }
610}
611
612/// A [`FetchClient`] failure as a `std::error::Error`, for crates whose HTTP hook requires one
613/// (oauth2's and openidconnect's `request_async` / `discover_async`).
614#[derive(Debug)]
615pub struct FetchError(pub anyhow::Error);
616
617impl std::fmt::Display for FetchError {
618    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
619        write!(f, "{:#}", self.0)
620    }
621}
622
623impl std::error::Error for FetchError {}
624
625/// One request given as raw parts, for a crate's own HTTP hook to call: the method, the URL,
626/// the headers as (name, value bytes) and the body, back as the status, the headers and the
627/// body. Header values that are not visible ASCII are sent lossily; the hooks that call this
628/// (OAuth2 and OpenID Connect token, discovery and key-set requests) send none.
629pub async fn round_trip_parts(
630    client: &FetchClient,
631    method: &str,
632    url: &str,
633    headers: Vec<(String, Vec<u8>)>,
634    body: Vec<u8>,
635) -> std::result::Result<(u16, Vec<(String, Vec<u8>)>, Bytes), FetchError> {
636    let method: Method = method
637        .parse()
638        .map_err(|e| FetchError(anyhow!("invalid HTTP method {method:?}: {e}")))?;
639    let mut request = client.request(method, url);
640    for (name, value) in &headers {
641        request = request.header(name, String::from_utf8_lossy(value));
642    }
643    if !body.is_empty() {
644        request = request.body(body);
645    }
646    let response = request.send().await.map_err(FetchError)?;
647    let status = response.status().as_u16();
648    let headers = response
649        .headers()
650        .iter()
651        .map(|(name, value)| (name.as_str().to_string(), value.as_bytes().to_vec()))
652        .collect();
653    let body = response.bytes().await.map_err(FetchError)?;
654    Ok((status, headers, body))
655}
656
657/// Read a native buffered response without trusting Content-Length. Streaming callers
658/// use `chunk()` directly; buffered model events have a byte cap and whole-read deadline.
659#[cfg(not(target_arch = "wasm32"))]
660pub async fn read_response_bytes(mut response: reqwest::Response, limit: usize) -> Result<Bytes> {
661    if response
662        .content_length()
663        .is_some_and(|length| length > limit as u64)
664    {
665        anyhow::bail!("HTTP response body exceeds the {limit}-byte cap");
666    }
667    tokio::time::timeout(transport::REQUEST_TIMEOUT, async move {
668        let mut body = bytes::BytesMut::new();
669        while let Some(chunk) = response.chunk().await? {
670            if chunk.len() > limit.saturating_sub(body.len()) {
671                anyhow::bail!("HTTP response body exceeds the {limit}-byte cap");
672            }
673            body.extend_from_slice(&chunk);
674        }
675        Ok(body.freeze())
676    })
677    .await
678    .context("HTTP response body deadline exceeded")?
679}
680
681/// Read bounded text while preserving reqwest's Content-Type charset/BOM decoding.
682#[cfg(not(target_arch = "wasm32"))]
683pub async fn read_response_text(response: reqwest::Response, limit: usize) -> Result<String> {
684    let headers = response.headers().clone();
685    let status = response.status();
686    let body = read_response_bytes(response, limit).await?;
687    let mut buffered = hyper::Response::new(body);
688    *buffered.headers_mut() = headers;
689    *buffered.status_mut() = status;
690    Ok(reqwest::Response::from(buffered).text().await?)
691}
692
693/// Parse JSON only after the complete response meets the byte and time bounds.
694#[cfg(not(target_arch = "wasm32"))]
695pub async fn read_response_json<T: DeserializeOwned>(
696    response: reqwest::Response,
697    limit: usize,
698) -> Result<T> {
699    serde_json::from_slice(&read_response_bytes(response, limit).await?)
700        .context("decode the response body as JSON")
701}
netget/client/http_fetch/mod.rs

netget/client/http_fetch/
mod.rs
