netget/client/http_fetch/
transport.rs

1//! One HTTP/1.1 request and its response, over a stream NetGet opens itself.
2//!
3//! This is the transport every HTTP-family client uses in the browser build, through
4//! [`super::FetchClient`] (the `http` client calls [`fetch`] directly). `reqwest` cannot serve
5//! there: its wasm32 backend is the browser's `fetch`, whose futures are not `Send` (the `Client`
6//! trait requires `Send`) and which cannot reach the page's virtual loopback anyway — NetGet's
7//! servers in the page listen on `crates/netget-tokio-wasm`'s in-memory network, not on anything
8//! `fetch` can dial. So the request is written with hyper 1's `client::conn::http1` over whatever
9//! [`tokio::net::TcpStream`] is on this target: the kernel's natively, the virtual loopback's in
10//! the browser. The client role of hyper's HTTP/1 dispatcher never touches the clock (only the
11//! server role maintains a `Date` header cache, which `vendor/hyper` patches for the browser), so
12//! this path runs there unchanged.
13//!
14//! It compiles on both targets so it can be tested natively (`tests/client/http/transport_test.rs`
15//! drives NetGet's own HTTP and TCP servers through it). The native clients still use reqwest,
16//! which brings TLS, HTTP/2 and connection pooling this module deliberately does not have.
17//!
18//! HTTP/2 is spoken with prior knowledge only ([`Wire::Http2PriorKnowledge`], cleartext h2c
19//! with no upgrade and no ALPN — what `reqwest`'s `http2_prior_knowledge()` does), through
20//! hyper's `client::conn::http2`. Its connection spawns its own tasks through
21//! [`SpawnExecutor`], which is `tokio::spawn` as this crate names it: tokio's natively, the
22//! browser shim's (the JS event loop) on wasm32.
23//!
24//! What it does not do, stated rather than implied: no TLS — an `https://` URL is refused with
25//! the reason (see [`HTTPS_UNSUPPORTED`]); no redirects (a 3xx is reported to the model like any
26//! other status, which is what the model is for); no connection reuse (one connection per
27//! request, closed after the response); no compression negotiation.
28
29use std::time::Duration;
30
31use anyhow::{anyhow, bail, Context, Result};
32use bytes::Bytes;
33use http_body_util::{BodyExt, Full, Limited};
34use hyper_util::rt::TokioIo;
35use tokio::io::{AsyncRead, AsyncWrite};
36
37/// One completed HTTP exchange, the body read as text (lossily, where it is not UTF-8).
38///
39/// What the `http` client reports to the model; [`fetch_response`] keeps the body as bytes for
40/// the clients whose bodies are binary.
41#[derive(Debug, Clone)]
42pub struct HttpExchange {
43    pub status_code: u16,
44    pub status_text: String,
45    pub headers: serde_json::Map<String, serde_json::Value>,
46    pub body: String,
47}
48
49impl HttpExchange {
50    fn from_response(response: hyper::Response<Bytes>) -> Self {
51        let status = response.status();
52        let mut headers = serde_json::Map::new();
53        for (name, value) in response.headers() {
54            if let Ok(text) = value.to_str() {
55                headers.insert(name.to_string(), serde_json::json!(text));
56            }
57        }
58        HttpExchange {
59            status_code: status.as_u16(),
60            status_text: status.to_string(),
61            headers,
62            body: String::from_utf8_lossy(response.body()).into_owned(),
63        }
64    }
65}
66
67/// The largest response body this client reads before refusing the response. The HTTP/3 client
68/// uses the same number; it is far above anything a model can usefully read in one event.
69pub const MAX_RESPONSE_BODY_BYTES: usize = 8 * 1024 * 1024;
70
71/// How long one exchange — connect, request, response head and body — may take in total.
72/// The same 30 seconds the native reqwest client is built with.
73pub const REQUEST_TIMEOUT: Duration = Duration::from_secs(30);
74
75/// Why an `https://` URL is refused on this transport.
76pub const HTTPS_UNSUPPORTED: &str =
77    "https:// is not available on this transport: it has no TLS stack, and in the browser build \
78     no server on the page's virtual network holds a certificate a client could verify. Use \
79     http:// (NetGet's own servers in the page speak plain HTTP)";
80
81/// The connection driver must end even when a caller drops or times out the exchange.
82/// A plain JoinHandle detaches on drop, leaving an unanswered request's socket alive.
83struct ConnectionDriver(tokio::task::JoinHandle<()>);
84
85impl Drop for ConnectionDriver {
86    fn drop(&mut self) {
87        self.0.abort();
88    }
89}
90
91fn parse_method(method: &str) -> Result<hyper::Method> {
92    // Retain case-insensitive convenience for the previously supported methods.
93    // Extension methods are case-sensitive and must reach the peer unchanged.
94    const STANDARD_METHODS: [&str; 7] =
95        ["GET", "POST", "PUT", "DELETE", "HEAD", "PATCH", "OPTIONS"];
96    let method = STANDARD_METHODS
97        .into_iter()
98        .find(|standard| method.eq_ignore_ascii_case(standard))
99        .unwrap_or(method);
100    hyper::Method::from_bytes(method.as_bytes()).context("invalid HTTP method")
101}
102
103/// Where an `http://` URL points: the host and port to dial, the `Host` header, and the
104/// origin-form request target.
105#[derive(Debug, Clone, PartialEq, Eq)]
106pub struct HttpTarget {
107    pub host: String,
108    pub port: u16,
109    pub authority: String,
110    pub path_and_query: String,
111}
112
113/// Parse an absolute `http://` URL. `https://` is refused with [`HTTPS_UNSUPPORTED`]; any other
114/// scheme, or no host, is an error naming the URL.
115pub fn parse_http_url(url: &str) -> Result<HttpTarget> {
116    if url
117        .bytes()
118        .any(|byte| byte.is_ascii_control() || byte.is_ascii_whitespace())
119    {
120        bail!("HTTP URL contains unescaped whitespace or control characters");
121    }
122    let parsed = url::Url::parse(url).context("invalid HTTP URL")?;
123    if parsed.host_str().is_none() {
124        bail!("HTTP URL requires a host");
125    }
126    let uri: hyper::Uri = url.parse().with_context(|| format!("not a URL: {url}"))?;
127    match uri.scheme_str() {
128        Some("http") => {}
129        Some("https") => bail!("{HTTPS_UNSUPPORTED} (asked for {url})"),
130        Some(other) => bail!("unsupported URL scheme {other:?} in {url}; only http:// is"),
131        None => bail!("no scheme in {url}; expected http://host:port/path"),
132    }
133    let authority = uri
134        .authority()
135        .with_context(|| format!("no host in {url}"))?;
136    let host = authority
137        .host()
138        .trim_start_matches('[')
139        .trim_end_matches(']')
140        .to_string();
141    if authority.as_str().contains('@') {
142        bail!("HTTP URL userinfo is unsupported; use an explicit Authorization header");
143    }
144    if host.is_empty() {
145        bail!("HTTP URL requires a nonempty host");
146    }
147    let port = match authority.port() {
148        Some(port) => port
149            .as_str()
150            .parse::<u16>()
151            .context("invalid HTTP URL port")?,
152        None if authority.as_str().ends_with(':') => bail!("HTTP URL port is empty"),
153        None => 80,
154    };
155    let path_and_query = uri
156        .path_and_query()
157        .map(|p| p.as_str().to_string())
158        .filter(|p| !p.is_empty())
159        .unwrap_or_else(|| "/".to_string());
160    Ok(HttpTarget {
161        host,
162        port,
163        authority: authority.as_str().to_string(),
164        path_and_query,
165    })
166}
167
168/// Dial `url`'s host and exchange one request, bounded by `timeout` as a whole and the body by
169/// `max_body` bytes. `headers` are applied in order after `Host`; a `Host` among them replaces
170/// the one derived from the URL.
171pub async fn fetch(
172    method: &str,
173    url: &str,
174    headers: &[(String, String)],
175    body: Option<String>,
176    timeout: Duration,
177    max_body: usize,
178) -> Result<HttpExchange> {
179    fetch_response(
180        method,
181        url,
182        headers,
183        body.map(Bytes::from),
184        timeout,
185        max_body,
186    )
187    .await
188    .map(HttpExchange::from_response)
189}
190
191/// Which HTTP a request is written in.
192#[derive(Debug, Clone, Copy, PartialEq, Eq)]
193pub enum Wire {
194    /// HTTP/1.1.
195    Http1,
196    /// HTTP/2 over cleartext TCP with prior knowledge (h2c, no upgrade, no ALPN).
197    Http2PriorKnowledge,
198}
199
200/// Runs the tasks hyper's HTTP/2 connection spawns (its connection and per-stream body
201/// pipes) with `tokio::spawn` as this crate resolves it: tokio's natively, the browser shim's
202/// on wasm32. hyper-util's `TokioExecutor` would call the real tokio's `spawn` in the browser,
203/// where no tokio runtime exists.
204#[derive(Clone, Copy, Debug, Default)]
205pub struct SpawnExecutor;
206
207impl<F> hyper::rt::Executor<F> for SpawnExecutor
208where
209    F: std::future::Future + Send + 'static,
210    F::Output: Send + 'static,
211{
212    fn execute(&self, future: F) {
213        // Each task ends with the stream or the connection it serves; the exchange aborts the
214        // connection driver on every exit, which ends the rest.
215        drop(tokio::spawn(future));
216    }
217}
218
219/// As [`fetch`], with the request and response bodies as bytes.
220pub async fn fetch_response(
221    method: &str,
222    url: &str,
223    headers: &[(String, String)],
224    body: Option<Bytes>,
225    timeout: Duration,
226    max_body: usize,
227) -> Result<hyper::Response<Bytes>> {
228    fetch_response_on(Wire::Http1, method, url, headers, body, timeout, max_body).await
229}
230
231/// As [`fetch_response`], written in `wire`'s HTTP.
232#[allow(clippy::too_many_arguments)]
233pub async fn fetch_response_on(
234    wire: Wire,
235    method: &str,
236    url: &str,
237    headers: &[(String, String)],
238    body: Option<Bytes>,
239    timeout: Duration,
240    max_body: usize,
241) -> Result<hyper::Response<Bytes>> {
242    let target = parse_http_url(url)?;
243    let exchange = async {
244        let stream = tokio::net::TcpStream::connect((target.host.clone(), target.port))
245            .await
246            .with_context(|| format!("connect to {}:{}", target.host, target.port))?;
247        match wire {
248            Wire::Http1 => {
249                exchange_response(stream, method, &target, headers, body, max_body).await
250            }
251            Wire::Http2PriorKnowledge => {
252                exchange_response_h2(stream, method, &target, headers, body, max_body).await
253            }
254        }
255    };
256    tokio::time::timeout(timeout, exchange)
257        .await
258        .map_err(|_| anyhow!("no complete response from {url} within {timeout:?}"))?
259}
260
261/// Write one request on `io` and read its response, the body bounded by `max_body` bytes.
262pub async fn exchange<S>(
263    io: S,
264    method: &str,
265    target: &HttpTarget,
266    headers: &[(String, String)],
267    body: Option<String>,
268    max_body: usize,
269) -> Result<HttpExchange>
270where
271    S: AsyncRead + AsyncWrite + Send + Unpin + 'static,
272{
273    exchange_response(io, method, target, headers, body.map(Bytes::from), max_body)
274        .await
275        .map(HttpExchange::from_response)
276}
277
278/// As [`exchange`], with the request and response bodies as bytes.
279pub async fn exchange_response<S>(
280    io: S,
281    method: &str,
282    target: &HttpTarget,
283    headers: &[(String, String)],
284    body: Option<Bytes>,
285    max_body: usize,
286) -> Result<hyper::Response<Bytes>>
287where
288    S: AsyncRead + AsyncWrite + Send + Unpin + 'static,
289{
290    let method = parse_method(method)?;
291
292    let mut builder = hyper::Request::builder()
293        .method(method.as_str())
294        .uri(target.path_and_query.as_str())
295        .header(hyper::header::HOST, target.authority.as_str());
296    for (name, value) in headers {
297        let name = hyper::header::HeaderName::from_bytes(name.as_bytes())
298            .with_context(|| format!("invalid header name {name:?}"))?;
299        let value = hyper::header::HeaderValue::from_str(value)
300            .with_context(|| format!("invalid value for header {name}"))?;
301        let map = builder
302            .headers_mut()
303            .context("request builder rejected an earlier part")?;
304        if name == hyper::header::HOST {
305            map.insert(name, value);
306        } else {
307            map.append(name, value);
308        }
309    }
310    let request = builder
311        .body(Full::new(body.unwrap_or_default()))
312        .context("build HTTP request")?;
313
314    let (mut sender, connection) = hyper::client::conn::http1::handshake(TokioIo::new(io))
315        .await
316        .context("HTTP/1.1 handshake")?;
317    // The connection future drives the socket; it ends when the response is read and the
318    // sender dropped. Aborted on every exit so an abandoned exchange cannot keep the socket.
319    let _driver = ConnectionDriver(tokio::spawn(async move {
320        let _ = connection.await;
321    }));
322    let result = async {
323        let response = sender
324            .send_request(request)
325            .await
326            .context("send HTTP request")?;
327        let (parts, body) = response.into_parts();
328        let collected = Limited::new(body, max_body).collect().await.map_err(|e| {
329            if e.downcast_ref::<http_body_util::LengthLimitError>()
330                .is_some()
331            {
332                anyhow!("response body exceeds the {max_body}-byte limit; refused")
333            } else {
334                anyhow!("read HTTP response body: {e}")
335            }
336        })?;
337        Ok(hyper::Response::from_parts(parts, collected.to_bytes()))
338    }
339    .await;
340    result
341}
342
343/// As [`exchange_response`], in HTTP/2 with prior knowledge: the connection preface goes out
344/// first, with no HTTP/1.1 upgrade. The request carries its scheme and authority in the URI,
345/// which hyper writes as the `:scheme` and `:authority` pseudo-headers; no `Host` header is
346/// added (a `Host` among `headers` is sent as given).
347pub async fn exchange_response_h2<S>(
348    io: S,
349    method: &str,
350    target: &HttpTarget,
351    headers: &[(String, String)],
352    body: Option<Bytes>,
353    max_body: usize,
354) -> Result<hyper::Response<Bytes>>
355where
356    S: AsyncRead + AsyncWrite + Send + Unpin + 'static,
357{
358    let method = parse_method(method)?;
359
360    let uri = format!("http://{}{}", target.authority, target.path_and_query);
361    let mut builder = hyper::Request::builder()
362        .method(method.as_str())
363        .uri(uri.as_str())
364        .version(hyper::Version::HTTP_2);
365    for (name, value) in headers {
366        let name = hyper::header::HeaderName::from_bytes(name.as_bytes())
367            .with_context(|| format!("invalid header name {name:?}"))?;
368        let value = hyper::header::HeaderValue::from_str(value)
369            .with_context(|| format!("invalid value for header {name}"))?;
370        builder
371            .headers_mut()
372            .context("request builder rejected an earlier part")?
373            .append(name, value);
374    }
375    let request = builder
376        .body(Full::new(body.unwrap_or_default()))
377        .context("build HTTP/2 request")?;
378
379    let (mut sender, connection) =
380        hyper::client::conn::http2::handshake(SpawnExecutor, TokioIo::new(io))
381            .await
382            .context("HTTP/2 handshake (prior knowledge)")?;
383    let _driver = ConnectionDriver(tokio::spawn(async move {
384        let _ = connection.await;
385    }));
386    let result = async {
387        let response = sender
388            .send_request(request)
389            .await
390            .context("send HTTP/2 request")?;
391        let (parts, body) = response.into_parts();
392        let collected = Limited::new(body, max_body).collect().await.map_err(|e| {
393            if e.downcast_ref::<http_body_util::LengthLimitError>()
394                .is_some()
395            {
396                anyhow!("response body exceeds the {max_body}-byte limit; refused")
397            } else {
398                anyhow!("read HTTP/2 response body: {e}")
399            }
400        })?;
401        Ok(hyper::Response::from_parts(parts, collected.to_bytes()))
402    }
403    .await;
404    result
405}
netget/client/http_fetch/transport.rs

netget/client/http_fetch/
transport.rs
